Go the way the owner's phone already goes
Control had two transports and neither fitted the ordinary customer. OCPP waits for the charger to dial in, which needs a public endpoint it can reach, a certificate, and a firmware willing to talk to our CSMS. Modbus TCP dials the charger, which needs the server on the charger's own network. Between them they cover a charger we host and a charger we stand next to; the common case is a charger behind someone else's router, and that had nothing. It was never unreachable, though. The charger holds a connection open to Anker's own broker — it is how the mobile app drives it from anywhere, and it is the mqttStatus register the Modbus snapshot has been reporting all along. So a third control mode joins that broker as the account: get_user_mqtt_info issues a client certificate, mTLS to aiot-mqtt-eu.anker.com:8883, and commands go out on the same topics the app publishes on. Nothing on the customer's side has to be forwarded, addressed or certificated. What travels is not an API call. The payload is a JSON envelope around a base64 binary frame the device itself speaks — marker, little-endian length, message type, name/length/type/value fields, XOR checksum — so mqttframe.go is a codec rather than a client, written from the message maps in anker-solix-api and anchored on the one frame that project documents byte for byte. A frame whose fields do not tile exactly up to the checksum is refused rather than half-read: these arrive over a link we do not control, and a truncated frame must not read as a charger reporting zeros. Two of the charger's habits shape the rest. It publishes nothing unless asked, so a status read arms a telemetry trigger and waits for the next frame, and a poll inside that window answers from what has since arrived. And a broker connection costs a fetched certificate and a TLS handshake while the plugin manager builds a throwaway instance per request — so the connection lives on the account's shared session beside the auth token, for exactly the reason the token lives there, and closes itself after five idle minutes. The transport also sees two signals no other one does: the boost flag, and the plug and start countdowns. The package doc has said since the first commit that they are never set and the derived mode must do without them. Here they are set, so a charger that has been told to start and is counting down a delay says so rather than sitting in "preparing", and "skip the delay" is offered only while there is a delay to skip. The clients generalise instead of growing a second layout. Both snapshots name the same quantities the same way, so what was Modbus-only in the readouts is now whichever transport read the charger — ModbusStatus becomes ChargerStatus on the phone, mb becomes dev on the web. What each transport can be *told* still differs, and the buttons branch on that: reset and clear-limit stay with OCPP, the timeout and phase registers with Modbus, skip-delay with the cloud. A command a transport has no equivalent for is refused by name, saying which one has it. The cost is worth saying plainly. This leans on Anker's cloud being up and on an unofficial protocol the app may change under us, where Modbus leans on nothing but the LAN. And it is checked against the reference implementation's own worked example rather than against hardware — there is no charger on this end to point it at. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
tajniak81
co-authored by
Claude Opus 5
Claude Opus 5
parent
b27ca3ee19
commit
576df58776
@@ -7,13 +7,12 @@
|
||||
// unchanged since v3.7.0.
|
||||
//
|
||||
// Scope & limitations:
|
||||
// - Read-only. Only EV-charger information is retrieved; no charge start/stop
|
||||
// or configuration commands are implemented. Control is a separate concern
|
||||
// and runs over OCPP (see internal/ocpp), not the cloud API.
|
||||
// - Cloud only. Upstream reads a charger's live state over both the cloud and
|
||||
// MQTT; we take the cloud half. Signals that exist only in MQTT — the boost
|
||||
// flag and the plug/start countdowns — are therefore never set, which the
|
||||
// derived state accounts for.
|
||||
// - The REST half is read-only. Every endpoint below retrieves EV-charger
|
||||
// information; nothing is started, stopped or configured through it, because
|
||||
// Anker's REST API has no such endpoint. Control runs over one of three
|
||||
// transports instead: OCPP (internal/ocpp), Modbus TCP on the local network
|
||||
// (modbus.go), or the account's cloud MQTT broker (cloudmqtt.go) — the one
|
||||
// path that reaches a charger behind a customer's router.
|
||||
// - Single device family. Capabilities target the V1 Smart EV Charger; other
|
||||
// Anker Power devices (solarbanks, power stations, HES) are out of scope.
|
||||
// - Unofficial. This talks to Anker's private mobile-app cloud API with the
|
||||
@@ -88,6 +87,9 @@ const (
|
||||
epSiteList = "power_service/v1/site/get_site_list" // sites (systems) on the account
|
||||
epUserVehicles = "power_service/v1/app/vehicle/get_vehicle_list" // vehicles registered for smart charging
|
||||
epVehicleDetail = "power_service/v1/app/vehicle/get_vehicle_detail" // details for one registered vehicle
|
||||
|
||||
// The cloud MQTT broker's own credentials endpoint is epMqttInfo, declared in
|
||||
// cloudmqtt.go next to the transport that uses it.
|
||||
)
|
||||
|
||||
// EV-charger operating states, mirroring anker-solix-api's SolixEvChargerStatus.
|
||||
@@ -192,6 +194,8 @@ func (p *Plugin) Descriptor() plugins.Descriptor {
|
||||
{ID: "sites", Method: "POST", Endpoint: epSiteList, Description: "Sites (systems) registered to the account."},
|
||||
{ID: "vehicles", Method: "POST", Endpoint: epUserVehicles, Description: "Vehicles registered for smart charging."},
|
||||
{ID: "vehicle", Method: "POST", Endpoint: epVehicleDetail, Description: "Details for one registered vehicle (needs vehicleId)."},
|
||||
{ID: "mqtt-status", Method: "POST", Endpoint: epMqttInfo, Description: "Live state of one charger over Anker's cloud MQTT broker — the path to a charger the server cannot reach (needs sn)."},
|
||||
{ID: "mqtt-command", Method: "POST", Endpoint: epMqttInfo, Description: "Control one charger over Anker's cloud MQTT broker: start, stop, boost, skip-delay, limit (with amps) or trigger (needs sn and command)."},
|
||||
},
|
||||
ConfigFields: []plugins.ConfigField{
|
||||
// Credentials are intentionally NOT required at the global (panel) layer,
|
||||
@@ -209,9 +213,10 @@ func (p *Plugin) Descriptor() plugins.Descriptor {
|
||||
// package (internal/ocpp) — but it is advertised here so a superadmin can
|
||||
// set/lock it at the global layer, and it cascades like the other fields.
|
||||
{Key: "controlMode", Label: "Control mode", Type: "select", Default: "off",
|
||||
Help: "How DriverVault controls the charger. Off = monitoring only (default). Modbus TCP = DriverVault connects to the charger on the local network (enable it in the Anker app under Settings > Integrations); this is the only mode that works when the charger cannot reach the server. The two OCPP modes need the charger to connect in to DriverVault: Own CSMS directly, Proxy CSMS relayed to Anker's cloud.",
|
||||
Help: "How DriverVault controls the charger. Off = monitoring only (default). Anker cloud = commands travel over the account's own MQTT broker, so the charger needs no reachability at all; this is the mode for a charger behind a customer's router. Modbus TCP = DriverVault connects to the charger on the local network (enable it in the Anker app under Settings > Integrations), which needs the server to share that network. The two OCPP modes need the charger to connect in to DriverVault: Own CSMS directly, Proxy CSMS relayed to Anker's cloud.",
|
||||
Options: []plugins.SelectOption{
|
||||
{Value: "off", Label: "Off (monitoring only)"},
|
||||
{Value: "mqtt", Label: "Anker cloud (works anywhere)"},
|
||||
{Value: "modbus", Label: "Modbus TCP (local network)"},
|
||||
{Value: "own", Label: "Own CSMS (full control)"},
|
||||
{Value: "proxy", Label: "Proxy CSMS (relay + control)"},
|
||||
@@ -286,6 +291,11 @@ type invokeParams struct {
|
||||
Range string `json:"range"` // day | week | month | year
|
||||
StartDate string `json:"startDate"` // YYYY-MM-DD, or YYYY-MM / YYYY for month / year
|
||||
EndDate string `json:"endDate"`
|
||||
|
||||
// The cloud MQTT actions: which command to send, and the current ceiling
|
||||
// "limit" carries.
|
||||
Command string `json:"command"`
|
||||
Amps float64 `json:"amps"`
|
||||
}
|
||||
|
||||
// Invoke runs a named read-only capability. The upstream response body is
|
||||
@@ -313,6 +323,18 @@ func (p *Plugin) Invoke(ctx context.Context, action string, params json.RawMessa
|
||||
}
|
||||
return p.chargerState(ctx, pp.SiteID, pp.SN)
|
||||
}
|
||||
// The cloud MQTT actions address the charger itself over the account's broker
|
||||
// rather than a REST endpoint, so they route to that transport instead of the
|
||||
// single-endpoint dispatch below.
|
||||
if action == "mqtt-status" || action == "mqtt-command" {
|
||||
if pp.SN == "" {
|
||||
return nil, fmt.Errorf("anker-solix: action %q requires an sn (EV charger serial)", action)
|
||||
}
|
||||
if action == "mqtt-status" {
|
||||
return p.mqttStatus(ctx, pp.SN)
|
||||
}
|
||||
return p.mqttCommand(ctx, pp.SN, pp.Command, pp.Amps)
|
||||
}
|
||||
|
||||
var (
|
||||
endpoint string
|
||||
@@ -396,7 +418,10 @@ func energyRange(r string) string {
|
||||
}
|
||||
}
|
||||
|
||||
// Shutdown releases pooled connections.
|
||||
// Shutdown releases pooled connections. The account's shared session — its token
|
||||
// and its broker connection — is deliberately left alone: the manager builds and
|
||||
// tears down an instance per request, and every instance for that account shares
|
||||
// it (see session.go).
|
||||
func (p *Plugin) Shutdown(context.Context) error {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
|
||||
@@ -0,0 +1,737 @@
|
||||
package ankersolix
|
||||
|
||||
// Control over Anker's own cloud MQTT broker — the path for a charger the server
|
||||
// cannot reach.
|
||||
//
|
||||
// The other two transports each need something a remote customer does not have.
|
||||
// OCPP needs the charger to dial in to us, which means a public endpoint, a
|
||||
// certificate, and a charger whose firmware accepts our CSMS. Modbus TCP needs
|
||||
// the server to dial the charger, which means sharing a network with it. Most
|
||||
// chargers sit behind a customer's router with neither. What they *do* have is
|
||||
// the connection they already hold open to Anker: the mobile app controls them
|
||||
// through it from anywhere, and it is the same broker the charger's own
|
||||
// mqttStatus register reports as connected.
|
||||
//
|
||||
// So this transport joins that broker as the account, exactly as the app does:
|
||||
//
|
||||
// 1. app/devicemanage/get_user_mqtt_info hands out a client certificate and key
|
||||
// for the signed-in account, the broker's address, and the AWS root the
|
||||
// broker is verified against. The certificate is the credential; there is no
|
||||
// username or password on the MQTT connection itself.
|
||||
// 2. Commands are published to cmd/{app}/{model}/{serial}/req and the charger's
|
||||
// own messages arrive on dt/{app}/{model}/{serial}/#.
|
||||
// 3. Both directions carry a JSON envelope whose payload holds a base64 binary
|
||||
// frame — the device's own protocol, encoded in mqttframe.go.
|
||||
//
|
||||
// Two consequences shape the code:
|
||||
//
|
||||
// - A connection is expensive (a TLS handshake with a fetched certificate) and
|
||||
// the plugin manager builds a throwaway instance per request, so the broker
|
||||
// connection lives in the account's shared session alongside the auth token,
|
||||
// for the same reason (see session.go). It closes itself after an idle spell.
|
||||
// - The charger does not publish its live state unless asked. A realtime
|
||||
// trigger turns the stream on for a bounded window, after which it stops
|
||||
// again — so a status read arms the trigger and waits for the next frame,
|
||||
// and a second read inside the window answers from what has since arrived.
|
||||
//
|
||||
// Unofficial, like the rest of the cloud half: this is the mobile app's private
|
||||
// transport, and Anker may change it at any time.
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"crypto/tls"
|
||||
"crypto/x509"
|
||||
"encoding/base64"
|
||||
"encoding/binary"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"math/big"
|
||||
"net"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"drivervault/apiserver/internal/mqtt"
|
||||
)
|
||||
|
||||
// epMqttInfo hands out the account's broker address and client certificate.
|
||||
const epMqttInfo = "app/devicemanage/get_user_mqtt_info"
|
||||
|
||||
// defaultChargerModel is the product code used when the account inventory has
|
||||
// not named one. The topics carry the model, and this connector is scoped to the
|
||||
// V1 Smart EV Charger, so it is the only sensible fallback.
|
||||
const defaultChargerModel = "A5191"
|
||||
|
||||
// Broker timings.
|
||||
const (
|
||||
// mqttCredsTTL re-fetches the certificate periodically. It is issued per
|
||||
// account and long-lived, but refetching costs one cloud call an hour and
|
||||
// means a revoked certificate is not held forever.
|
||||
mqttCredsTTL = 1 * time.Hour
|
||||
|
||||
// mqttIdle closes a broker connection nothing has used for this long. The
|
||||
// charger keeps publishing only while a trigger is live, so an idle
|
||||
// connection is genuinely idle.
|
||||
mqttIdle = 5 * time.Minute
|
||||
|
||||
// mqttConnectTimeout bounds the TLS handshake and the wait for CONNACK.
|
||||
mqttConnectTimeout = 15 * time.Second
|
||||
|
||||
// triggerWindow is how long the charger is asked to keep streaming telemetry,
|
||||
// and triggerRenew is how close to the end of that window a status read
|
||||
// re-arms it rather than racing the last frame.
|
||||
triggerWindow = 180 * time.Second
|
||||
triggerRenew = 30 * time.Second
|
||||
|
||||
// statusWait is how long a status read waits for a frame from the charger. A
|
||||
// triggered charger publishes every 3-5 seconds; this allows for the trigger
|
||||
// having to reach it first.
|
||||
statusWait = 12 * time.Second
|
||||
|
||||
// commandWait is how long a command waits for the charger's confirmation
|
||||
// message. A command is fire-and-forget on the wire, so this only decides
|
||||
// whether we can say the charger answered — not whether it was sent.
|
||||
commandWait = 5 * time.Second
|
||||
|
||||
// deviceCacheTTL is how long the account's charger list (serial to model) is
|
||||
// trusted before the cloud is asked again.
|
||||
deviceCacheTTL = 1 * time.Hour
|
||||
)
|
||||
|
||||
// The mode values the charger's 0105 message accepts, and the names this
|
||||
// connector takes for them — the same names the cloud plugin already uses for
|
||||
// SolixEvChargerMode.
|
||||
var mqttModeValues = map[string]uint8{
|
||||
modeStartCharge: 1,
|
||||
modeStopCharge: 2,
|
||||
modeSkipDelay: 3,
|
||||
modeBoostCharge: 4,
|
||||
}
|
||||
|
||||
// mqttEncodingMode is the payload's encoding_type for the mode command. It is
|
||||
// not encryption — the frame is plain either way — but the charger expects the
|
||||
// field on this message, so it is sent with a seed like the app's.
|
||||
const mqttEncodingMode = 2
|
||||
|
||||
// mqttCredentials is what get_user_mqtt_info returns: an address to dial and a
|
||||
// certificate to dial it with.
|
||||
type mqttCredentials struct {
|
||||
UserID string `json:"user_id"`
|
||||
AppName string `json:"app_name"`
|
||||
ThingName string `json:"thing_name"`
|
||||
CertificateID string `json:"certificate_id"`
|
||||
CertificatePE string `json:"certificate_pem"`
|
||||
PrivateKey string `json:"private_key"`
|
||||
EndpointAddr string `json:"endpoint_addr"`
|
||||
RootCA string `json:"aws_root_ca1_pem"`
|
||||
}
|
||||
|
||||
// valid reports whether the credentials carry everything a connection needs.
|
||||
func (c mqttCredentials) valid() bool {
|
||||
return strings.TrimSpace(c.EndpointAddr) != "" &&
|
||||
strings.TrimSpace(c.CertificatePE) != "" &&
|
||||
strings.TrimSpace(c.PrivateKey) != ""
|
||||
}
|
||||
|
||||
// address is the broker's host:port. The endpoint is returned without a port;
|
||||
// 8883 is the MQTT-over-TLS port the app uses.
|
||||
func (c mqttCredentials) address() string {
|
||||
host := strings.TrimSpace(c.EndpointAddr)
|
||||
if _, _, err := net.SplitHostPort(host); err == nil {
|
||||
return host
|
||||
}
|
||||
return net.JoinHostPort(host, "8883")
|
||||
}
|
||||
|
||||
// appName is the topic segment identifying the app the account belongs to.
|
||||
func (c mqttCredentials) appName() string {
|
||||
if n := strings.TrimSpace(c.AppName); n != "" {
|
||||
return n
|
||||
}
|
||||
return "anker_power"
|
||||
}
|
||||
|
||||
// ---- credentials and device lookup -------------------------------------------
|
||||
|
||||
// mqttCreds returns the account's broker credentials, fetching them at most once
|
||||
// per mqttCredsTTL. They are held on the shared session rather than the plugin
|
||||
// instance for the same reason the auth token is: the instance does not outlive
|
||||
// the request.
|
||||
func (p *Plugin) mqttCreds(ctx context.Context) (mqttCredentials, error) {
|
||||
s := p.sess
|
||||
if s == nil {
|
||||
return mqttCredentials{}, errors.New("anker-solix: plugin not initialised")
|
||||
}
|
||||
s.mqttMu.Lock()
|
||||
defer s.mqttMu.Unlock()
|
||||
|
||||
if s.mqttCreds != nil && time.Since(s.mqttCredsAt) < mqttCredsTTL {
|
||||
return *s.mqttCreds, nil
|
||||
}
|
||||
body, err := p.apiRequest(ctx, epMqttInfo, map[string]any{})
|
||||
if err != nil {
|
||||
return mqttCredentials{}, err
|
||||
}
|
||||
var env struct {
|
||||
Data mqttCredentials `json:"data"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &env); err != nil {
|
||||
return mqttCredentials{}, fmt.Errorf("anker-solix: decode MQTT info: %w", err)
|
||||
}
|
||||
if !env.Data.valid() {
|
||||
return mqttCredentials{}, errors.New("anker-solix: the cloud returned no MQTT certificate for this account; cloud control is not available on it")
|
||||
}
|
||||
s.mqttCreds, s.mqttCredsAt = &env.Data, time.Now()
|
||||
return env.Data, nil
|
||||
}
|
||||
|
||||
// chargerModel returns the product code for a serial on this account, which the
|
||||
// topics need. It doubles as the ownership check: a serial no view of the
|
||||
// account reports is one this caller may not command, and saying so is better
|
||||
// than publishing to a topic the broker will refuse anyway.
|
||||
func (p *Plugin) chargerModel(ctx context.Context, sn string) (string, error) {
|
||||
s := p.sess
|
||||
if s == nil {
|
||||
return "", errors.New("anker-solix: plugin not initialised")
|
||||
}
|
||||
sn = strings.TrimSpace(sn)
|
||||
if sn == "" {
|
||||
return "", errors.New("anker-solix: a charger serial is required")
|
||||
}
|
||||
|
||||
s.mqttMu.Lock()
|
||||
model, known := s.devices[sn]
|
||||
fresh := time.Since(s.devicesAt) < deviceCacheTTL
|
||||
s.mqttMu.Unlock()
|
||||
if known && fresh {
|
||||
return model, nil
|
||||
}
|
||||
|
||||
// Unknown, or the list has gone stale: ask the cloud once and remember it.
|
||||
doc, err := p.chargerInventory(ctx)
|
||||
if err != nil {
|
||||
if known {
|
||||
return model, nil // the cloud is unreachable; the last list still stands
|
||||
}
|
||||
return "", err
|
||||
}
|
||||
found := map[string]string{}
|
||||
for _, c := range doc.Chargers {
|
||||
m := strings.ToUpper(strings.TrimSpace(c.Model))
|
||||
if m == "" {
|
||||
m = defaultChargerModel
|
||||
}
|
||||
found[c.SN] = m
|
||||
}
|
||||
s.mqttMu.Lock()
|
||||
s.devices, s.devicesAt = found, time.Now()
|
||||
s.mqttMu.Unlock()
|
||||
|
||||
if m, ok := found[sn]; ok {
|
||||
return m, nil
|
||||
}
|
||||
return "", fmt.Errorf("anker-solix: charger %s is not on this Anker account", sn)
|
||||
}
|
||||
|
||||
// ---- the broker connection ---------------------------------------------------
|
||||
|
||||
// mqttConn is one account's live broker connection, with the state it has
|
||||
// collected from the chargers it is subscribed to.
|
||||
type mqttConn struct {
|
||||
client *mqtt.Client
|
||||
creds mqttCredentials
|
||||
sessID string
|
||||
started time.Time
|
||||
|
||||
mu sync.Mutex
|
||||
subs map[string]bool // topic filter -> subscribed
|
||||
devices map[string]*deviceState // serial -> what it has told us
|
||||
lastUse time.Time
|
||||
deadErr error
|
||||
waiters []chan struct{}
|
||||
shutdown chan struct{}
|
||||
}
|
||||
|
||||
// deviceState is everything one charger has reported over this connection,
|
||||
// merged across message types: telemetry overwrites telemetry, settings
|
||||
// overwrite settings, and neither erases the other.
|
||||
type deviceState struct {
|
||||
values map[string]any
|
||||
telemetryAt time.Time
|
||||
settingsAt time.Time
|
||||
triggeredUntil time.Time
|
||||
}
|
||||
|
||||
// mqttClient returns the account's broker connection, opening one if there is
|
||||
// none or the last one died.
|
||||
func (p *Plugin) mqttClient(ctx context.Context) (*mqttConn, error) {
|
||||
s := p.sess
|
||||
if s == nil {
|
||||
return nil, errors.New("anker-solix: plugin not initialised")
|
||||
}
|
||||
creds, err := p.mqttCreds(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
s.mqttMu.Lock()
|
||||
defer s.mqttMu.Unlock()
|
||||
if c := s.mqttConn; c != nil {
|
||||
if c.alive() && c.creds.CertificateID == creds.CertificateID {
|
||||
c.touch()
|
||||
return c, nil
|
||||
}
|
||||
c.close()
|
||||
s.mqttConn = nil
|
||||
}
|
||||
c, err := dialBroker(ctx, creds)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
s.mqttConn = c
|
||||
return c, nil
|
||||
}
|
||||
|
||||
// dialBroker opens the mutually-authenticated connection. The account's
|
||||
// certificate is the credential, and the broker is verified against the AWS root
|
||||
// the same response supplied — the connection is to Anker's own broker, so
|
||||
// neither side is trusted on the strength of the other.
|
||||
func dialBroker(ctx context.Context, creds mqttCredentials) (*mqttConn, error) {
|
||||
cert, err := tls.X509KeyPair([]byte(creds.CertificatePE), []byte(creds.PrivateKey))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("anker-solix: the cloud's MQTT certificate could not be loaded: %w", err)
|
||||
}
|
||||
roots := x509.NewCertPool()
|
||||
if ca := strings.TrimSpace(creds.RootCA); ca != "" {
|
||||
if !roots.AppendCertsFromPEM([]byte(ca)) {
|
||||
return nil, errors.New("anker-solix: the cloud's MQTT root certificate could not be parsed")
|
||||
}
|
||||
} else {
|
||||
// No root supplied: fall back to the system pool rather than skipping
|
||||
// verification, which would let anything answer for the broker.
|
||||
if roots, err = x509.SystemCertPool(); err != nil {
|
||||
return nil, fmt.Errorf("anker-solix: no root certificates to verify the MQTT broker: %w", err)
|
||||
}
|
||||
}
|
||||
host, _, splitErr := net.SplitHostPort(creds.address())
|
||||
if splitErr != nil {
|
||||
host = creds.EndpointAddr
|
||||
}
|
||||
|
||||
client, err := mqtt.Connect(ctx, mqtt.Options{
|
||||
Address: creds.address(),
|
||||
TLS: true,
|
||||
TLSConfig: &tls.Config{
|
||||
ServerName: host,
|
||||
MinVersion: tls.VersionTLS12,
|
||||
RootCAs: roots,
|
||||
Certificates: []tls.Certificate{cert},
|
||||
},
|
||||
// The broker keys a session by client id and evicts the older holder, so
|
||||
// the app's own connection must not be displaced: the app uses
|
||||
// "{thing_name}_{5 digits}", and a different suffix is a different session.
|
||||
ClientID: clientIDFor(creds),
|
||||
Keepalive: 60 * time.Second,
|
||||
ConnectTimeout: mqttConnectTimeout,
|
||||
Buffer: 256,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
c := &mqttConn{
|
||||
client: client,
|
||||
creds: creds,
|
||||
sessID: randomSessionID(),
|
||||
started: time.Now(),
|
||||
subs: map[string]bool{},
|
||||
devices: map[string]*deviceState{},
|
||||
lastUse: time.Now(),
|
||||
shutdown: make(chan struct{}),
|
||||
}
|
||||
go c.readLoop()
|
||||
go c.idleLoop()
|
||||
return c, nil
|
||||
}
|
||||
|
||||
// clientIDFor builds an identifier no other holder of this account's certificate
|
||||
// is using, so joining the broker never evicts the owner's mobile app.
|
||||
func clientIDFor(creds mqttCredentials) string {
|
||||
thing := strings.TrimSpace(creds.ThingName)
|
||||
if thing == "" {
|
||||
thing = strings.TrimSpace(creds.UserID)
|
||||
}
|
||||
return fmt.Sprintf("%s_%05d", thing, randomBelow(100000))
|
||||
}
|
||||
|
||||
// randomSessionID mimics the app's sess_id, a pair of four-digit groups.
|
||||
func randomSessionID() string {
|
||||
return fmt.Sprintf("%04d-%04d", randomBelow(10000), randomBelow(10000))
|
||||
}
|
||||
|
||||
// randomBelow returns a non-negative integer below n, falling back to a
|
||||
// clock-derived value if the system source fails.
|
||||
func randomBelow(n int64) int64 {
|
||||
v, err := rand.Int(rand.Reader, big.NewInt(n))
|
||||
if err != nil {
|
||||
return time.Now().UnixNano() % n
|
||||
}
|
||||
return v.Int64()
|
||||
}
|
||||
|
||||
// alive reports whether the connection is still usable.
|
||||
func (c *mqttConn) alive() bool {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
return c.deadErr == nil
|
||||
}
|
||||
|
||||
// touch marks the connection as in use, so the idle sweep leaves it alone.
|
||||
func (c *mqttConn) touch() {
|
||||
c.mu.Lock()
|
||||
c.lastUse = time.Now()
|
||||
c.mu.Unlock()
|
||||
}
|
||||
|
||||
// close ends the connection and wakes anything waiting on a message.
|
||||
func (c *mqttConn) close() {
|
||||
c.fail(mqtt.ErrClosed)
|
||||
_ = c.client.Close()
|
||||
}
|
||||
|
||||
// fail records why the connection ended and releases every waiter.
|
||||
func (c *mqttConn) fail(err error) {
|
||||
c.mu.Lock()
|
||||
if c.deadErr == nil {
|
||||
c.deadErr = err
|
||||
close(c.shutdown)
|
||||
}
|
||||
c.wakeLocked()
|
||||
c.mu.Unlock()
|
||||
}
|
||||
|
||||
// wakeLocked releases everything waiting for a device message. Caller holds mu.
|
||||
func (c *mqttConn) wakeLocked() {
|
||||
for _, ch := range c.waiters {
|
||||
close(ch)
|
||||
}
|
||||
c.waiters = nil
|
||||
}
|
||||
|
||||
// readLoop owns the inbound stream for the life of the connection.
|
||||
func (c *mqttConn) readLoop() {
|
||||
for msg := range c.client.Messages() {
|
||||
c.ingest(msg)
|
||||
}
|
||||
err := c.client.Err()
|
||||
if err == nil {
|
||||
err = mqtt.ErrClosed
|
||||
}
|
||||
c.fail(err)
|
||||
}
|
||||
|
||||
// idleLoop closes a connection nothing has used for mqttIdle. The account keeps
|
||||
// no state on the broker between commands, so dropping the socket costs only the
|
||||
// next handshake.
|
||||
func (c *mqttConn) idleLoop() {
|
||||
t := time.NewTicker(mqttIdle / 2)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-c.shutdown:
|
||||
return
|
||||
case <-t.C:
|
||||
c.mu.Lock()
|
||||
idle := time.Since(c.lastUse)
|
||||
c.mu.Unlock()
|
||||
if idle >= mqttIdle {
|
||||
c.close()
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ingest decodes one inbound message and folds it into the sending charger's
|
||||
// state. Anything it cannot read is dropped: these frames come from a cloud
|
||||
// connection, and a malformed one must not be recorded as a reading.
|
||||
func (c *mqttConn) ingest(msg mqtt.Message) {
|
||||
sn, data, ok := parseEnvelope(msg)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
msgType, values, err := decodeFrame(data)
|
||||
if err != nil || len(values) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
st := c.devices[sn]
|
||||
if st == nil {
|
||||
st = &deviceState{values: map[string]any{}}
|
||||
c.devices[sn] = st
|
||||
}
|
||||
for k, v := range values {
|
||||
st.values[k] = v
|
||||
}
|
||||
now := time.Now()
|
||||
if msgType == msgEVTelemetry {
|
||||
st.telemetryAt = now
|
||||
} else {
|
||||
st.settingsAt = now
|
||||
}
|
||||
c.wakeLocked()
|
||||
}
|
||||
|
||||
// parseEnvelope pulls the sending serial and the binary frame out of one MQTT
|
||||
// message. The payload is a JSON string inside a JSON object, and the frame is
|
||||
// base64 inside that — the shape the app both sends and receives.
|
||||
func parseEnvelope(msg mqtt.Message) (string, []byte, bool) {
|
||||
var env struct {
|
||||
Head struct {
|
||||
DeviceSN string `json:"device_sn"`
|
||||
} `json:"head"`
|
||||
Payload string `json:"payload"`
|
||||
}
|
||||
if err := json.Unmarshal(msg.Payload, &env); err != nil {
|
||||
return "", nil, false
|
||||
}
|
||||
var inner struct {
|
||||
SN string `json:"sn"`
|
||||
SN2 string `json:"device_sn"`
|
||||
Data string `json:"data"`
|
||||
}
|
||||
if err := json.Unmarshal([]byte(env.Payload), &inner); err != nil {
|
||||
return "", nil, false
|
||||
}
|
||||
sn := firstNonEmpty(inner.SN, inner.SN2, env.Head.DeviceSN, serialFromTopic(msg.Topic))
|
||||
if sn == "" || inner.Data == "" {
|
||||
return "", nil, false
|
||||
}
|
||||
data, err := base64.StdEncoding.DecodeString(inner.Data)
|
||||
if err != nil {
|
||||
return "", nil, false
|
||||
}
|
||||
return sn, data, true
|
||||
}
|
||||
|
||||
// serialFromTopic reads the serial out of dt/{app}/{model}/{serial}/… , which is
|
||||
// where it is when the payload does not repeat it.
|
||||
func serialFromTopic(topic string) string {
|
||||
parts := strings.Split(topic, "/")
|
||||
if len(parts) < 4 {
|
||||
return ""
|
||||
}
|
||||
return parts[3]
|
||||
}
|
||||
|
||||
func firstNonEmpty(vals ...string) string {
|
||||
for _, v := range vals {
|
||||
if s := strings.TrimSpace(v); s != "" {
|
||||
return s
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// ---- topics, subscribing and publishing --------------------------------------
|
||||
|
||||
// dataTopic is the filter carrying everything one charger publishes.
|
||||
func dataTopic(creds mqttCredentials, model, sn string) string {
|
||||
return fmt.Sprintf("dt/%s/%s/%s/#", creds.appName(), model, sn)
|
||||
}
|
||||
|
||||
// commandTopic is where one charger's commands are published.
|
||||
func commandTopic(creds mqttCredentials, model, sn string) string {
|
||||
return fmt.Sprintf("cmd/%s/%s/%s/req", creds.appName(), model, sn)
|
||||
}
|
||||
|
||||
// listen subscribes to a charger's data topic, once per connection.
|
||||
func (c *mqttConn) listen(ctx context.Context, model, sn string) error {
|
||||
topic := dataTopic(c.creds, model, sn)
|
||||
c.mu.Lock()
|
||||
already := c.subs[topic]
|
||||
c.mu.Unlock()
|
||||
if already {
|
||||
return nil
|
||||
}
|
||||
if err := c.client.Subscribe(ctx, topic); err != nil {
|
||||
return fmt.Errorf("anker-solix: cannot listen to charger %s over the cloud: %w", sn, err)
|
||||
}
|
||||
c.mu.Lock()
|
||||
c.subs[topic] = true
|
||||
c.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
||||
// publishFrame wraps a device frame in the app's envelope and publishes it to
|
||||
// the charger's command topic.
|
||||
func (c *mqttConn) publishFrame(ctx context.Context, model, sn string, frame []byte, encoding int) error {
|
||||
now := time.Now()
|
||||
seed := any(1)
|
||||
inner := map[string]any{
|
||||
"device_sn": sn,
|
||||
"account_id": c.creds.UserID,
|
||||
"data": base64.StdEncoding.EncodeToString(frame),
|
||||
}
|
||||
if encoding != 0 {
|
||||
inner["encoding_type"] = encoding
|
||||
seed = randomSeed()
|
||||
}
|
||||
payload, err := json.Marshal(inner)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
envelope, err := json.Marshal(map[string]any{
|
||||
"head": map[string]any{
|
||||
"version": "1.0.0.1",
|
||||
"client_id": fmt.Sprintf("android-%s-%s-%s", c.creds.appName(), c.creds.UserID, c.creds.CertificateID),
|
||||
"sess_id": c.sessID,
|
||||
"msg_seq": 1,
|
||||
"seed": seed,
|
||||
"timestamp": now.Unix(),
|
||||
// cmd_status 2 and cmd 17 are what the app sends on a control message;
|
||||
// the charger ignores neither, and a different pair goes unanswered.
|
||||
"cmd_status": 2,
|
||||
"cmd": 17,
|
||||
"sign_code": 1,
|
||||
"device_pn": model,
|
||||
"device_sn": sn,
|
||||
},
|
||||
"payload": string(payload),
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
c.touch()
|
||||
return c.client.Publish(ctx, commandTopic(c.creds, model, sn), envelope)
|
||||
}
|
||||
|
||||
// randomSeed is the 16-byte seed the app puts in the header of an encoded
|
||||
// message.
|
||||
func randomSeed() string {
|
||||
var b [16]byte
|
||||
if _, err := rand.Read(b[:]); err != nil {
|
||||
binary.LittleEndian.PutUint64(b[:8], uint64(time.Now().UnixNano()))
|
||||
}
|
||||
return encodeHex(b[:])
|
||||
}
|
||||
|
||||
// ---- waiting for the charger to answer ----------------------------------------
|
||||
|
||||
// waitFor blocks until a charger's state satisfies ready, or the deadline
|
||||
// passes. It reports whether ready was met; a connection that dies while waiting
|
||||
// ends the wait with the reason.
|
||||
func (c *mqttConn) waitFor(ctx context.Context, sn string, ready func(*deviceState) bool, timeout time.Duration) (bool, error) {
|
||||
deadline := time.After(timeout)
|
||||
for {
|
||||
c.mu.Lock()
|
||||
if err := c.deadErr; err != nil {
|
||||
c.mu.Unlock()
|
||||
return false, fmt.Errorf("anker-solix: the cloud connection dropped: %w", err)
|
||||
}
|
||||
if st := c.devices[sn]; st != nil && ready(st) {
|
||||
c.mu.Unlock()
|
||||
return true, nil
|
||||
}
|
||||
ch := make(chan struct{})
|
||||
c.waiters = append(c.waiters, ch)
|
||||
c.mu.Unlock()
|
||||
|
||||
select {
|
||||
case <-ch:
|
||||
case <-deadline:
|
||||
return false, nil
|
||||
case <-ctx.Done():
|
||||
return false, ctx.Err()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// snapshotOf copies a charger's collected state out from under the lock.
|
||||
func (c *mqttConn) snapshotOf(sn string) (map[string]any, time.Time, time.Time, time.Time) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
st := c.devices[sn]
|
||||
if st == nil {
|
||||
return nil, time.Time{}, time.Time{}, time.Time{}
|
||||
}
|
||||
out := make(map[string]any, len(st.values))
|
||||
for k, v := range st.values {
|
||||
out[k] = v
|
||||
}
|
||||
return out, st.telemetryAt, st.settingsAt, st.triggeredUntil
|
||||
}
|
||||
|
||||
// noteTrigger records how long the charger has been asked to keep streaming.
|
||||
func (c *mqttConn) noteTrigger(sn string, until time.Time) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
st := c.devices[sn]
|
||||
if st == nil {
|
||||
st = &deviceState{values: map[string]any{}}
|
||||
c.devices[sn] = st
|
||||
}
|
||||
st.triggeredUntil = until
|
||||
}
|
||||
|
||||
// ---- commands ----------------------------------------------------------------
|
||||
|
||||
// mqttTrigger asks a charger to publish live telemetry for a while. Without it
|
||||
// the charger is silent, so every status read arms one.
|
||||
func (p *Plugin) mqttTrigger(ctx context.Context, c *mqttConn, model, sn string, window time.Duration) error {
|
||||
frame, err := encodeFrame(msgRealtimeTrigger, []cmdField{
|
||||
rawField(0xa1, 0x22),
|
||||
uintField(0xa2, 1),
|
||||
varField(0xa3, uint32(window/time.Second)),
|
||||
timestampField(time.Now()),
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := c.publishFrame(ctx, model, sn, frame, 0); err != nil {
|
||||
return err
|
||||
}
|
||||
c.noteTrigger(sn, time.Now().Add(window))
|
||||
return nil
|
||||
}
|
||||
|
||||
// mqttSetMode sends the start / stop / skip-delay / boost command.
|
||||
func (p *Plugin) mqttSetMode(ctx context.Context, c *mqttConn, model, sn, mode string) error {
|
||||
v, ok := mqttModeValues[mode]
|
||||
if !ok {
|
||||
return fmt.Errorf("anker-solix: %q is not one of %s, %s, %s or %s",
|
||||
mode, modeStartCharge, modeStopCharge, modeSkipDelay, modeBoostCharge)
|
||||
}
|
||||
frame, err := encodeFrame(msgEVMode, []cmdField{
|
||||
rawField(0xa1, 0x22),
|
||||
uintField(0xa2, v),
|
||||
timestampField(time.Now()),
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return c.publishFrame(ctx, model, sn, frame, mqttEncodingMode)
|
||||
}
|
||||
|
||||
// mqttSetMaxCurrent sets the charging current ceiling, in amps. The limit is
|
||||
// checked by the same rule the Modbus path uses, because the rule is the
|
||||
// charger's: the transport differs, the charger does not.
|
||||
func (p *Plugin) mqttSetMaxCurrent(ctx context.Context, c *mqttConn, model, sn string, amps float64) error {
|
||||
if err := checkMaxCurrent(amps); err != nil {
|
||||
return err
|
||||
}
|
||||
// The field carries deciamps, as the register does over Modbus.
|
||||
frame, err := encodeFrame(msgEVSettings, []cmdField{
|
||||
rawField(0xa1, 0x22),
|
||||
intField(0xa8, int16(amps*10)),
|
||||
timestampField(time.Now()),
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return c.publishFrame(ctx, model, sn, frame, 0)
|
||||
}
|
||||
@@ -0,0 +1,249 @@
|
||||
package ankersolix
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"drivervault/apiserver/internal/mqtt"
|
||||
)
|
||||
|
||||
func TestMqttCredentialsAddressAndAppName(t *testing.T) {
|
||||
c := mqttCredentials{EndpointAddr: "aiot-mqtt-eu.anker.com"}
|
||||
if got := c.address(); got != "aiot-mqtt-eu.anker.com:8883" {
|
||||
t.Errorf("address = %q, want the TLS port appended", got)
|
||||
}
|
||||
// A response that already carries a port must not have another appended.
|
||||
c.EndpointAddr = "aiot-mqtt-eu.anker.com:8884"
|
||||
if got := c.address(); got != "aiot-mqtt-eu.anker.com:8884" {
|
||||
t.Errorf("address = %q, want the endpoint's own port kept", got)
|
||||
}
|
||||
if got := c.appName(); got != "anker_power" {
|
||||
t.Errorf("appName = %q, want the default when the cloud sent none", got)
|
||||
}
|
||||
c.AppName = "anker_charging"
|
||||
if got := c.appName(); got != "anker_charging" {
|
||||
t.Errorf("appName = %q, want the cloud's own value", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMqttCredentialsNeedCertificateAndKey(t *testing.T) {
|
||||
full := mqttCredentials{EndpointAddr: "host", CertificatePE: "cert", PrivateKey: "key"}
|
||||
if !full.valid() {
|
||||
t.Error("complete credentials were rejected")
|
||||
}
|
||||
for _, c := range []mqttCredentials{
|
||||
{CertificatePE: "cert", PrivateKey: "key"},
|
||||
{EndpointAddr: "host", PrivateKey: "key"},
|
||||
{EndpointAddr: "host", CertificatePE: "cert"},
|
||||
} {
|
||||
if c.valid() {
|
||||
t.Errorf("credentials missing a field were accepted: %+v", c)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestTopicsAddressOneCharger(t *testing.T) {
|
||||
c := mqttCredentials{AppName: "anker_power"}
|
||||
if got := commandTopic(c, "A5191", "SN123"); got != "cmd/anker_power/A5191/SN123/req" {
|
||||
t.Errorf("commandTopic = %q", got)
|
||||
}
|
||||
if got := dataTopic(c, "A5191", "SN123"); got != "dt/anker_power/A5191/SN123/#" {
|
||||
t.Errorf("dataTopic = %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
// envelope builds a message shaped like the ones the cloud delivers: JSON, whose
|
||||
// payload is itself a JSON string, whose data field is a base64 device frame.
|
||||
func envelope(t *testing.T, sn string, frame []byte) mqtt.Message {
|
||||
t.Helper()
|
||||
inner, err := json.Marshal(map[string]any{
|
||||
"device_sn": sn,
|
||||
"data": base64.StdEncoding.EncodeToString(frame),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
outer, err := json.Marshal(map[string]any{
|
||||
"head": map[string]any{"device_sn": sn, "timestamp": 1756813256},
|
||||
"payload": string(inner),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return mqtt.Message{Topic: "dt/anker_power/A5191/" + sn + "/param_info", Payload: outer}
|
||||
}
|
||||
|
||||
func TestParseEnvelopeUnwrapsTheDeviceFrame(t *testing.T) {
|
||||
frame := buildInbound(t, msgEVTelemetry, field(0xbb, typeUint8, 0x02))
|
||||
sn, data, ok := parseEnvelope(envelope(t, "SN123", frame))
|
||||
if !ok {
|
||||
t.Fatal("a well-formed envelope was rejected")
|
||||
}
|
||||
if sn != "SN123" {
|
||||
t.Errorf("serial = %q, want SN123", sn)
|
||||
}
|
||||
if string(data) != string(frame) {
|
||||
t.Errorf("frame came back changed")
|
||||
}
|
||||
}
|
||||
|
||||
// The serial is not always repeated in the payload; the topic carries it too,
|
||||
// and a message we cannot attribute to a charger must be dropped rather than
|
||||
// folded into some other charger's state.
|
||||
func TestParseEnvelopeFallsBackToTheTopic(t *testing.T) {
|
||||
inner, _ := json.Marshal(map[string]any{"data": base64.StdEncoding.EncodeToString([]byte{1, 2, 3})})
|
||||
outer, _ := json.Marshal(map[string]any{"payload": string(inner)})
|
||||
sn, _, ok := parseEnvelope(mqtt.Message{Topic: "dt/anker_power/A5191/SN999/param_info", Payload: outer})
|
||||
if !ok || sn != "SN999" {
|
||||
t.Errorf("serial = %q (ok=%v), want SN999 from the topic", sn, ok)
|
||||
}
|
||||
|
||||
for _, bad := range []mqtt.Message{
|
||||
{Topic: "dt/a/b/SN/x", Payload: []byte("not json")},
|
||||
{Topic: "dt/a/b/SN/x", Payload: []byte(`{"payload":"not json"}`)},
|
||||
{Topic: "dt/a/b/SN/x", Payload: []byte(`{"payload":"{\"device_sn\":\"SN\"}"}`)}, // no data
|
||||
{Topic: "dt/a/b/SN/x", Payload: []byte(`{"payload":"{\"data\":\"!!not b64\"}"}`)}, // undecodable
|
||||
{Topic: "short", Payload: []byte(`{"payload":"{\"data\":\"AQID\"}"}`)}, // no serial anywhere
|
||||
} {
|
||||
if _, _, ok := parseEnvelope(bad); ok {
|
||||
t.Errorf("a malformed envelope was accepted: %s", bad.Payload)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A charger's state is assembled from two message types that arrive at different
|
||||
// times: telemetry must not erase the settings that came with the last command,
|
||||
// and vice versa.
|
||||
func TestIngestMergesTelemetryAndSettings(t *testing.T) {
|
||||
c := &mqttConn{subs: map[string]bool{}, devices: map[string]*deviceState{}, shutdown: make(chan struct{})}
|
||||
|
||||
c.ingest(envelope(t, "SN1", buildInbound(t, msgEVParams, field(0xa8, typeInt16LE, 0x40, 0x01))))
|
||||
c.ingest(envelope(t, "SN1", buildInbound(t, msgEVTelemetry, field(0xbb, typeUint8, 0x02))))
|
||||
|
||||
values, telemetryAt, settingsAt, _ := c.snapshotOf("SN1")
|
||||
if v, _ := values["maxCurrentSetA"].(float64); v != 32 {
|
||||
t.Errorf("the settings value was lost when telemetry arrived: %v", values["maxCurrentSetA"])
|
||||
}
|
||||
if v, _ := values["status"].(float64); v != 2 {
|
||||
t.Errorf("status = %v, want 2", values["status"])
|
||||
}
|
||||
if telemetryAt.IsZero() || settingsAt.IsZero() {
|
||||
t.Errorf("both halves should be timestamped: telemetry %v, settings %v", telemetryAt, settingsAt)
|
||||
}
|
||||
if !telemetryAt.After(settingsAt) && !telemetryAt.Equal(settingsAt) {
|
||||
t.Errorf("telemetry arrived second but is stamped earlier")
|
||||
}
|
||||
|
||||
// A message from a charger we have no map for leaves the state untouched.
|
||||
c.ingest(mqtt.Message{Topic: "dt/a/b/SN1/x", Payload: []byte("rubbish")})
|
||||
after, _, _, _ := c.snapshotOf("SN1")
|
||||
if len(after) != len(values) {
|
||||
t.Errorf("an unreadable message changed the charger's state")
|
||||
}
|
||||
}
|
||||
|
||||
func TestWaitForReturnsWhenTheStateArrives(t *testing.T) {
|
||||
c := &mqttConn{subs: map[string]bool{}, devices: map[string]*deviceState{}, shutdown: make(chan struct{})}
|
||||
go func() {
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
c.ingest(envelope(t, "SN1", buildInbound(t, msgEVTelemetry, field(0xbb, typeUint8, 0x02))))
|
||||
}()
|
||||
ok, err := c.waitFor(context.Background(), "SN1", func(st *deviceState) bool {
|
||||
return !st.telemetryAt.IsZero()
|
||||
}, 2*time.Second)
|
||||
if err != nil || !ok {
|
||||
t.Fatalf("waitFor = %v, %v; want it to see the message", ok, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWaitForGivesUpAndReportsADeadConnection(t *testing.T) {
|
||||
c := &mqttConn{subs: map[string]bool{}, devices: map[string]*deviceState{}, shutdown: make(chan struct{})}
|
||||
ok, err := c.waitFor(context.Background(), "SN1", func(*deviceState) bool { return false }, 30*time.Millisecond)
|
||||
if err != nil || ok {
|
||||
t.Errorf("waitFor = %v, %v; want a quiet timeout", ok, err)
|
||||
}
|
||||
|
||||
// A connection that dies while a caller is waiting must wake it with the
|
||||
// reason rather than making it sit out the whole timeout.
|
||||
c2 := &mqttConn{subs: map[string]bool{}, devices: map[string]*deviceState{}, shutdown: make(chan struct{})}
|
||||
go func() {
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
c2.fail(mqtt.ErrClosed)
|
||||
}()
|
||||
start := time.Now()
|
||||
if _, err := c2.waitFor(context.Background(), "SN1", func(*deviceState) bool { return false }, 5*time.Second); err == nil {
|
||||
t.Error("a dropped connection ended the wait without an error")
|
||||
}
|
||||
if time.Since(start) > time.Second {
|
||||
t.Error("the waiter was not woken when the connection dropped")
|
||||
}
|
||||
}
|
||||
|
||||
// A command that cannot be sent should be refused before it costs a sign-in, a
|
||||
// certificate fetch and a broker connection — so the check runs on a plugin with
|
||||
// no session at all.
|
||||
func TestMqttCommandValidatesBeforeReachingTheCloud(t *testing.T) {
|
||||
p := &Plugin{}
|
||||
for _, tc := range []struct {
|
||||
command string
|
||||
amps float64
|
||||
want string
|
||||
}{
|
||||
{"reboot", 0, "not a cloud command"},
|
||||
{"limit", 3, "below the charger's 6 A floor"},
|
||||
{"limit", 40, "outside the charger's range"},
|
||||
} {
|
||||
_, err := p.mqttCommand(context.Background(), "SN1", tc.command, tc.amps)
|
||||
if err == nil {
|
||||
t.Errorf("%s(%v) was accepted", tc.command, tc.amps)
|
||||
continue
|
||||
}
|
||||
if !strings.Contains(err.Error(), tc.want) {
|
||||
t.Errorf("%s(%v) failed with %q, want it to mention %q", tc.command, tc.amps, err, tc.want)
|
||||
}
|
||||
}
|
||||
|
||||
// A valid command gets past validation and fails on the missing session
|
||||
// instead, which is what proves the order.
|
||||
if _, err := p.mqttCommand(context.Background(), "SN1", "start", 0); err == nil ||
|
||||
!strings.Contains(err.Error(), "not initialised") {
|
||||
t.Errorf("start failed with %v, want it to reach the session lookup", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMqttModeValuesMatchTheChargerEnum(t *testing.T) {
|
||||
want := map[string]uint8{
|
||||
modeStartCharge: 1, modeStopCharge: 2, modeSkipDelay: 3, modeBoostCharge: 4,
|
||||
}
|
||||
for mode, v := range want {
|
||||
if mqttModeValues[mode] != v {
|
||||
t.Errorf("%s = %d, want %d", mode, mqttModeValues[mode], v)
|
||||
}
|
||||
}
|
||||
// Every accepted command name must resolve to one of those modes, or the
|
||||
// endpoint can offer a name the transport cannot send.
|
||||
for name, mode := range mqttCommands {
|
||||
if _, ok := mqttModeValues[mode]; !ok {
|
||||
t.Errorf("command %q maps to %q, which is not a charger mode", name, mode)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestClientIDDoesNotCollideWithTheApp(t *testing.T) {
|
||||
creds := mqttCredentials{ThingName: "abc-anker_power"}
|
||||
a, b := clientIDFor(creds), clientIDFor(creds)
|
||||
if !strings.HasPrefix(a, "abc-anker_power_") {
|
||||
t.Errorf("client id %q does not carry the account's thing name", a)
|
||||
}
|
||||
if a == b {
|
||||
t.Error("two connections were given the same client id, which would evict each other")
|
||||
}
|
||||
// With no thing name the user id stands in, so the id is still account-scoped.
|
||||
if got := clientIDFor(mqttCredentials{UserID: "u1"}); !strings.HasPrefix(got, "u1_") {
|
||||
t.Errorf("client id %q does not fall back to the user id", got)
|
||||
}
|
||||
}
|
||||
@@ -497,17 +497,27 @@ func ModbusStopCharging(ctx context.Context, c *modbus.Client) error {
|
||||
return c.WriteSingle(ctx, regChargingCommand, cmdStopCharging)
|
||||
}
|
||||
|
||||
// ModbusSetMaxCurrent sets the charging current ceiling, in amps. The register
|
||||
// carries deciamps, and anything below currentPauseFloor stops the charge
|
||||
// outright rather than slowing it — so that case is refused here, and a caller
|
||||
// that means to pause is asked to say so.
|
||||
func ModbusSetMaxCurrent(ctx context.Context, c *modbus.Client, amps float64) error {
|
||||
// checkMaxCurrent validates a charging current ceiling in amps. The limit is the
|
||||
// charger's, not the transport's, so the cloud path applies the same rule (see
|
||||
// cloudmqtt.go): below currentPauseFloor the charger stops rather than charging
|
||||
// slowly, which makes a lower limit a pause in disguise — so it is refused, and
|
||||
// a caller that means to pause is asked to say so.
|
||||
func checkMaxCurrent(amps float64) error {
|
||||
if amps > 0 && amps < currentPauseFloor {
|
||||
return fmt.Errorf("anker-solix: %.1f A is below the charger's %.0f A floor, which pauses charging; stop the session instead", amps, currentPauseFloor)
|
||||
}
|
||||
if amps < 0 || amps > 32 {
|
||||
return fmt.Errorf("anker-solix: %.1f A is outside the charger's range (%.0f-32 A)", amps, currentPauseFloor)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ModbusSetMaxCurrent sets the charging current ceiling, in amps. The register
|
||||
// carries deciamps.
|
||||
func ModbusSetMaxCurrent(ctx context.Context, c *modbus.Client, amps float64) error {
|
||||
if err := checkMaxCurrent(amps); err != nil {
|
||||
return err
|
||||
}
|
||||
return c.WriteSingle(ctx, regMaxCurrentSet, uint16(amps*10))
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,472 @@
|
||||
package ankersolix
|
||||
|
||||
// The wire format Anker's cloud carries between the mobile app and the charger.
|
||||
//
|
||||
// An MQTT message is JSON, but only as an envelope: the part that means anything
|
||||
// is a base64 field inside it holding a binary frame the device itself speaks.
|
||||
// That frame is the same one the app sends, so a command is not a documented API
|
||||
// call but a byte layout, reproduced here from the message maps in
|
||||
// anker-solix-api (src/anker_solix_api/mqtttypes.py, mqttmap.py, mqttcmdmap.py at
|
||||
// v3.8.1) and checked against a live A5191.
|
||||
//
|
||||
// ff 09 2-byte marker on every Anker Solix frame
|
||||
// xx xx total length in bytes, little endian, counting the checksum
|
||||
// 03 00 0f fixed pattern; the middle byte is 00 outbound, 01 inbound
|
||||
// xx xx message type — what the frame is, per device model
|
||||
// [xx] an optional counter, present on some inbound frames
|
||||
// <fields> one or more data fields
|
||||
// xx XOR of every preceding byte
|
||||
//
|
||||
// and each data field is
|
||||
//
|
||||
// xx field name (a1, a2, … — the model's message map names it)
|
||||
// xx length of everything that follows in this field
|
||||
// [xx] value type, present when the field carries more than one byte
|
||||
// xx … the value
|
||||
//
|
||||
// The frame has no version number and no field-type registry: which name means
|
||||
// what depends on the message type, which is why the maps below are per message
|
||||
// type rather than one table.
|
||||
|
||||
import (
|
||||
"encoding/binary"
|
||||
"fmt"
|
||||
"math"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Frame markers. patternSend is what the app puts in an outbound frame; a device
|
||||
// answers with 03 01 0f, which is not checked — the message type is what selects
|
||||
// a decoder.
|
||||
var (
|
||||
frameMarker = []byte{0xff, 0x09}
|
||||
patternSend = []byte{0x03, 0x00, 0x0f}
|
||||
)
|
||||
|
||||
// frameHeaderLen is the marker, length, pattern and message type — everything
|
||||
// before the first data field, and before any inbound counter byte.
|
||||
const frameHeaderLen = 9
|
||||
|
||||
// Value types. A field longer than one byte starts with one of these; a
|
||||
// single-byte field carries its value directly with no type at all.
|
||||
const (
|
||||
typeString byte = 0x00
|
||||
typeUint8 byte = 0x01
|
||||
typeInt16LE byte = 0x02
|
||||
typeInt32LE byte = 0x03 // "var": four bytes, though not always one value
|
||||
typeFloat32 byte = 0x05
|
||||
typeNone byte = 0xff // this package's marker for "no type byte"
|
||||
)
|
||||
|
||||
// typeByteMax is the largest first-value-byte still read as a value type. Field
|
||||
// names start at 0xa1 and value types stop at 0x06, so the gap is wide; the
|
||||
// bound matches the reference implementation's rather than narrowing it, since a
|
||||
// message type we have not seen may use a type we have not seen either.
|
||||
const typeByteMax byte = 0x31
|
||||
|
||||
// Message types this package speaks, for the A5191 (V1 Smart EV Charger).
|
||||
// Outbound ones are commands, inbound ones are what the charger publishes back.
|
||||
const (
|
||||
msgRealtimeTrigger = "0057" // ask for the fast telemetry stream
|
||||
msgEVSettings = "0100" // the settings group: current limit, brightness, …
|
||||
msgEVMode = "0105" // start / stop / skip delay / boost
|
||||
msgEVTelemetry = "0410" // fast telemetry, only while a trigger is live
|
||||
msgEVParams = "0405" // settings and identity, sent after a command
|
||||
msgEVParamsAlt = "0840" // the same fields, in answer to a status request
|
||||
msgEVConfirm = "0900" // the same fields again, confirming a control change
|
||||
msgEVCharging = "0403" // a couple of charging parameters
|
||||
)
|
||||
|
||||
// mqttField is one named value inside a device message. factor scales the raw
|
||||
// integer (0.1 for a decivolt, 0.001 for a watt-hour reported in kWh); unsigned
|
||||
// marks the fields whose two- and four-byte types are *not* signed, which is the
|
||||
// exception rather than the rule; clock marks the two-byte fields that hold a
|
||||
// minute and an hour rather than a number.
|
||||
type mqttField struct {
|
||||
name string
|
||||
factor float64
|
||||
unsigned bool
|
||||
clock bool
|
||||
}
|
||||
|
||||
// evTelemetry decodes the 0410 message: the charger's live electrical state,
|
||||
// published every few seconds but only while a realtime trigger is live. It is
|
||||
// where the two signals the Modbus map has no register for — the plug and start
|
||||
// countdowns — actually come from.
|
||||
var evTelemetry = map[byte]mqttField{
|
||||
0xa2: {name: "voltageL1", factor: 0.1},
|
||||
0xa3: {name: "voltageL2", factor: 0.1},
|
||||
0xa4: {name: "voltageL3", factor: 0.1},
|
||||
0xa5: {name: "currentL1", factor: 0.1},
|
||||
0xa6: {name: "currentL2", factor: 0.1},
|
||||
0xa7: {name: "currentL3", factor: 0.1},
|
||||
0xa8: {name: "powerTotal"},
|
||||
0xa9: {name: "sessionSeconds"},
|
||||
0xaa: {name: "sessionWh"},
|
||||
0xab: {name: "sessionStartedAt", unsigned: true},
|
||||
0xad: {name: "plugCountdownSeconds"},
|
||||
0xae: {name: "startCountdownSeconds"},
|
||||
0xaf: {name: "chargingWindowSeconds"},
|
||||
0xb0: {name: "powerL1"},
|
||||
0xb1: {name: "powerL2"},
|
||||
0xb2: {name: "powerL3"},
|
||||
0xb3: {name: "sessionWhL1"},
|
||||
0xb4: {name: "sessionWhL2"},
|
||||
0xb5: {name: "sessionWhL3"},
|
||||
0xb8: {name: "ocppStatus"},
|
||||
0xba: {name: "phaseMode"},
|
||||
0xbb: {name: "status"},
|
||||
}
|
||||
|
||||
// evParams decodes the 0405 message (and the 0840 and 0900 that carry the same
|
||||
// fields): what the charger is *set* to, plus its identity. The charger sends it
|
||||
// after a control change rather than on a schedule, so these values arrive with
|
||||
// a command rather than with the telemetry stream.
|
||||
var evParams = map[byte]mqttField{
|
||||
0xa3: {name: "plugLockSwitch"},
|
||||
0xa4: {name: "autoStartSwitch"},
|
||||
0xa8: {name: "maxCurrentSetA", factor: 0.1},
|
||||
0xaa: {name: "ledBrightness"},
|
||||
0xac: {name: "autoRestartSwitch"},
|
||||
0xad: {name: "randomDelaySwitch"},
|
||||
0xb2: {name: "smartTouchMode"},
|
||||
0xb4: {name: "lightOffScheduleSwitch"},
|
||||
0xb5: {name: "lightOffStart", unsigned: true, clock: true},
|
||||
0xb6: {name: "lightOffEnd", unsigned: true, clock: true},
|
||||
0xb7: {name: "modbusSwitch"},
|
||||
0xcc: {name: "modbusTimeoutSeconds"},
|
||||
0xce: {name: "maxCurrentA", factor: 0.1},
|
||||
0xcf: {name: "modbusPort"},
|
||||
0xd0: {name: "ipAddress"},
|
||||
0xd3: {name: "loadBalancing"},
|
||||
0xd4: {name: "mainBreakerLimitA"},
|
||||
0xd8: {name: "solarBalancing"},
|
||||
0xd9: {name: "chargingMode"},
|
||||
0xda: {name: "solarMinCurrentA"},
|
||||
0xdb: {name: "phaseMode"},
|
||||
0xdd: {name: "autoPhaseSwitch"},
|
||||
0xdf: {name: "boostMode"},
|
||||
0xe0: {name: "cpSignal"},
|
||||
0xe2: {name: "plugged"},
|
||||
0xe3: {name: "status"},
|
||||
0xe6: {name: "scheduleSwitch"},
|
||||
0xe7: {name: "weekStart", unsigned: true, clock: true},
|
||||
0xe8: {name: "weekEnd", unsigned: true, clock: true},
|
||||
0xe9: {name: "weekendStart", unsigned: true, clock: true},
|
||||
0xea: {name: "weekendEnd", unsigned: true, clock: true},
|
||||
0xeb: {name: "weekendMode"},
|
||||
0xec: {name: "scheduleMode"},
|
||||
0xfe: {name: "minCurrentA"},
|
||||
}
|
||||
|
||||
// evCharging decodes the 0403 message, a pair of charging parameters the charger
|
||||
// sends around a mode change.
|
||||
var evCharging = map[byte]mqttField{
|
||||
0xa5: {name: "chargingWindowSeconds"},
|
||||
0xa6: {name: "solarMinCurrentA"},
|
||||
}
|
||||
|
||||
// evMessages selects a field map by message type. A type absent from here is one
|
||||
// we have no map for; its frame is still parsed, but nothing is named.
|
||||
var evMessages = map[string]map[byte]mqttField{
|
||||
msgEVTelemetry: evTelemetry,
|
||||
msgEVParams: evParams,
|
||||
msgEVParamsAlt: evParams,
|
||||
msgEVConfirm: evParams,
|
||||
msgEVCharging: evCharging,
|
||||
}
|
||||
|
||||
// ---- outbound frames ---------------------------------------------------------
|
||||
|
||||
// cmdField is one field of a command frame. typ is typeNone for the single-byte
|
||||
// fields that carry no value type.
|
||||
type cmdField struct {
|
||||
name byte
|
||||
typ byte
|
||||
value []byte
|
||||
}
|
||||
|
||||
// rawField builds a field with no value type — the `a1 01 22` pattern that opens
|
||||
// every command.
|
||||
func rawField(name byte, value ...byte) cmdField {
|
||||
return cmdField{name: name, typ: typeNone, value: value}
|
||||
}
|
||||
|
||||
// uintField builds a one-byte unsigned field.
|
||||
func uintField(name byte, v uint8) cmdField {
|
||||
return cmdField{name: name, typ: typeUint8, value: []byte{v}}
|
||||
}
|
||||
|
||||
// intField builds a two-byte little-endian signed field.
|
||||
func intField(name byte, v int16) cmdField {
|
||||
b := make([]byte, 2)
|
||||
binary.LittleEndian.PutUint16(b, uint16(v))
|
||||
return cmdField{name: name, typ: typeInt16LE, value: b}
|
||||
}
|
||||
|
||||
// varField builds a four-byte little-endian field.
|
||||
func varField(name byte, v uint32) cmdField {
|
||||
b := make([]byte, 4)
|
||||
binary.LittleEndian.PutUint32(b, v)
|
||||
return cmdField{name: name, typ: typeInt32LE, value: b}
|
||||
}
|
||||
|
||||
// timestampField is the `fe` field every command ends with: the sender's clock,
|
||||
// in whole seconds.
|
||||
func timestampField(now time.Time) cmdField {
|
||||
return varField(0xfe, uint32(now.Unix()))
|
||||
}
|
||||
|
||||
// encodeFrame builds one command frame for a message type. The caller supplies
|
||||
// every field in wire order, starting with the `a1 01 22` opener each command in
|
||||
// the reference maps carries and ending with the timestamp.
|
||||
func encodeFrame(msgType string, fields []cmdField) ([]byte, error) {
|
||||
mt, err := decodeHex(msgType)
|
||||
if err != nil || len(mt) < 2 || len(mt) > 3 {
|
||||
return nil, fmt.Errorf("anker-solix: %q is not a message type", msgType)
|
||||
}
|
||||
|
||||
body := make([]byte, 0, 32)
|
||||
for _, f := range fields {
|
||||
n := len(f.value)
|
||||
if f.typ != typeNone {
|
||||
n++
|
||||
}
|
||||
if n < 1 || n > 255 {
|
||||
return nil, fmt.Errorf("anker-solix: field %02x does not fit one frame field", f.name)
|
||||
}
|
||||
body = append(body, f.name, byte(n))
|
||||
if f.typ != typeNone {
|
||||
body = append(body, f.typ)
|
||||
}
|
||||
body = append(body, f.value...)
|
||||
}
|
||||
|
||||
// The length counts the whole frame, checksum byte included.
|
||||
total := frameHeaderLen + (len(mt) - 2) + len(body) + 1
|
||||
out := make([]byte, 0, total)
|
||||
out = append(out, frameMarker...)
|
||||
out = binary.LittleEndian.AppendUint16(out, uint16(total))
|
||||
out = append(out, patternSend...)
|
||||
out = append(out, mt...)
|
||||
out = append(out, body...)
|
||||
return append(out, xorChecksum(out)), nil
|
||||
}
|
||||
|
||||
// xorChecksum is the frame's only integrity check: every byte XORed together.
|
||||
func xorChecksum(b []byte) byte {
|
||||
var sum byte
|
||||
for _, x := range b {
|
||||
sum ^= x
|
||||
}
|
||||
return sum
|
||||
}
|
||||
|
||||
// ---- inbound frames ----------------------------------------------------------
|
||||
|
||||
// decodeFrame parses one device frame and returns its message type together with
|
||||
// the values its field map names. A field the map does not know is skipped
|
||||
// rather than guessed at, and a frame whose checksum does not add up is
|
||||
// rejected: these arrive over a cloud connection we do not control, so a
|
||||
// truncated one must not be read as a charger reporting zeros.
|
||||
func decodeFrame(data []byte) (string, map[string]any, error) {
|
||||
if len(data) < frameHeaderLen+2 {
|
||||
return "", nil, fmt.Errorf("anker-solix: device frame is %d bytes, too short to hold a header", len(data))
|
||||
}
|
||||
if data[0] != frameMarker[0] || data[1] != frameMarker[1] {
|
||||
return "", nil, fmt.Errorf("anker-solix: device frame does not start with the Anker marker (%02x%02x)", data[0], data[1])
|
||||
}
|
||||
if n := int(binary.LittleEndian.Uint16(data[2:4])); n != len(data) {
|
||||
return "", nil, fmt.Errorf("anker-solix: device frame says it is %d bytes but %d arrived", n, len(data))
|
||||
}
|
||||
if xorChecksum(data) != 0 {
|
||||
return "", nil, fmt.Errorf("anker-solix: device frame checksum does not match")
|
||||
}
|
||||
|
||||
msgType := encodeHex(data[7:9])
|
||||
|
||||
// Some inbound frames carry a counter byte between the header and the first
|
||||
// data field, and nothing in the frame says which kind this is. Rather than
|
||||
// guess from the byte's value — field names run high enough to be mistaken for
|
||||
// a counter — try both and keep the reading whose fields tile the frame
|
||||
// exactly, from the first field to the checksum with nothing left over.
|
||||
raw, ok := splitFields(data, frameHeaderLen)
|
||||
if !ok {
|
||||
if raw, ok = splitFields(data, frameHeaderLen+1); !ok {
|
||||
return "", nil, fmt.Errorf("anker-solix: device frame %s does not divide into whole data fields", msgType)
|
||||
}
|
||||
}
|
||||
|
||||
fields := evMessages[msgType]
|
||||
values := map[string]any{}
|
||||
for _, r := range raw {
|
||||
f, known := fields[r.name]
|
||||
if !known || f.name == "" {
|
||||
continue
|
||||
}
|
||||
if v, ok := decodeValue(r.typ, r.value, f); ok {
|
||||
values[f.name] = v
|
||||
}
|
||||
}
|
||||
return msgType, values, nil
|
||||
}
|
||||
|
||||
// rawFieldBytes is one data field as it sat in the frame, before its map entry
|
||||
// decides what it means.
|
||||
type rawFieldBytes struct {
|
||||
name byte
|
||||
typ byte
|
||||
value []byte
|
||||
}
|
||||
|
||||
// splitFields walks the data fields from start and reports them only if they end
|
||||
// exactly at the checksum byte. A frame read from the wrong offset runs off the
|
||||
// end or stops short, which is what makes the exact fit a usable test.
|
||||
func splitFields(data []byte, start int) ([]rawFieldBytes, bool) {
|
||||
end := len(data) - 1 // the last byte is the checksum
|
||||
var out []rawFieldBytes
|
||||
for idx := start; idx < end; {
|
||||
if idx+2 > end {
|
||||
return nil, false
|
||||
}
|
||||
name := data[idx]
|
||||
flen := int(data[idx+1])
|
||||
if flen == 0 || idx+2+flen > end {
|
||||
return nil, false
|
||||
}
|
||||
body := data[idx+2 : idx+2+flen]
|
||||
idx += 2 + flen
|
||||
|
||||
typ, value := typeNone, body
|
||||
if flen > 1 && body[0] <= typeByteMax {
|
||||
typ, value = body[0], body[1:]
|
||||
}
|
||||
out = append(out, rawFieldBytes{name: name, typ: typ, value: value})
|
||||
}
|
||||
return out, len(out) > 0
|
||||
}
|
||||
|
||||
// decodeValue turns one field's bytes into a value, following the type byte the
|
||||
// field carries and the scaling its map entry gives. It reports false for a
|
||||
// field whose bytes do not fit its type, so a short value is dropped rather than
|
||||
// read as a smaller number.
|
||||
func decodeValue(typ byte, b []byte, f mqttField) (any, bool) {
|
||||
if len(b) == 0 {
|
||||
return nil, false
|
||||
}
|
||||
factor := f.factor
|
||||
if factor == 0 {
|
||||
factor = 1
|
||||
}
|
||||
scale := func(n int64) any {
|
||||
if factor == 1 {
|
||||
return float64(n)
|
||||
}
|
||||
return round(float64(n)*factor, factor)
|
||||
}
|
||||
|
||||
switch typ {
|
||||
case typeString:
|
||||
return printable(b), true
|
||||
case typeUint8:
|
||||
return scale(int64(b[0])), true
|
||||
case typeInt16LE:
|
||||
if len(b) < 2 {
|
||||
return nil, false
|
||||
}
|
||||
if f.clock {
|
||||
// Two bytes holding a minute and an hour, least significant first.
|
||||
return fmt.Sprintf("%02d:%02d", b[1], b[0]), true
|
||||
}
|
||||
if f.unsigned {
|
||||
return scale(int64(binary.LittleEndian.Uint16(b))), true
|
||||
}
|
||||
return scale(int64(int16(binary.LittleEndian.Uint16(b)))), true
|
||||
case typeInt32LE:
|
||||
if len(b) < 4 {
|
||||
return nil, false
|
||||
}
|
||||
if f.unsigned {
|
||||
return scale(int64(binary.LittleEndian.Uint32(b))), true
|
||||
}
|
||||
return scale(int64(int32(binary.LittleEndian.Uint32(b)))), true
|
||||
case typeFloat32:
|
||||
if len(b) < 4 {
|
||||
return nil, false
|
||||
}
|
||||
return float64(math.Float32frombits(binary.LittleEndian.Uint32(b))), true
|
||||
default:
|
||||
// No value type: the bytes are the number, most significant first.
|
||||
var n int64
|
||||
for _, x := range b {
|
||||
n = n<<8 | int64(x)
|
||||
}
|
||||
return scale(n), true
|
||||
}
|
||||
}
|
||||
|
||||
// round trims the floating-point noise a factor introduces, to the precision the
|
||||
// factor itself implies — 0.1 keeps one decimal, 0.001 keeps three.
|
||||
func round(v, factor float64) float64 {
|
||||
digits := 0
|
||||
for f := factor; f < 1 && digits < 6; f *= 10 {
|
||||
digits++
|
||||
}
|
||||
p := math.Pow(10, float64(digits))
|
||||
return math.Round(v*p) / p
|
||||
}
|
||||
|
||||
// printable keeps the readable part of a string field; the charger pads some of
|
||||
// them with control bytes.
|
||||
func printable(b []byte) string {
|
||||
var sb strings.Builder
|
||||
for _, r := range string(b) {
|
||||
if r >= 0x20 && r != 0x7f {
|
||||
sb.WriteRune(r)
|
||||
}
|
||||
}
|
||||
return strings.TrimSpace(sb.String())
|
||||
}
|
||||
|
||||
// ---- small hex helpers -------------------------------------------------------
|
||||
|
||||
const hexDigits = "0123456789abcdef"
|
||||
|
||||
// encodeHex renders bytes as lowercase hex, which is how message types are keyed.
|
||||
func encodeHex(b []byte) string {
|
||||
out := make([]byte, 0, len(b)*2)
|
||||
for _, x := range b {
|
||||
out = append(out, hexDigits[x>>4], hexDigits[x&0x0f])
|
||||
}
|
||||
return string(out)
|
||||
}
|
||||
|
||||
// decodeHex parses a lowercase or uppercase hex string.
|
||||
func decodeHex(s string) ([]byte, error) {
|
||||
if len(s)%2 != 0 {
|
||||
return nil, fmt.Errorf("hex string %q has an odd length", s)
|
||||
}
|
||||
out := make([]byte, 0, len(s)/2)
|
||||
for i := 0; i < len(s); i += 2 {
|
||||
hi, err1 := hexNibble(s[i])
|
||||
lo, err2 := hexNibble(s[i+1])
|
||||
if err1 != nil || err2 != nil {
|
||||
return nil, fmt.Errorf("hex string %q holds a non-hex character", s)
|
||||
}
|
||||
out = append(out, hi<<4|lo)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func hexNibble(c byte) (byte, error) {
|
||||
switch {
|
||||
case c >= '0' && c <= '9':
|
||||
return c - '0', nil
|
||||
case c >= 'a' && c <= 'f':
|
||||
return c - 'a' + 10, nil
|
||||
case c >= 'A' && c <= 'F':
|
||||
return c - 'A' + 10, nil
|
||||
}
|
||||
return 0, fmt.Errorf("not a hex digit: %q", string(rune(c)))
|
||||
}
|
||||
@@ -0,0 +1,263 @@
|
||||
package ankersolix
|
||||
|
||||
import (
|
||||
"encoding/binary"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// The realtime trigger frame is the one example the reference implementation
|
||||
// documents byte for byte, so it is the anchor for the whole encoder: marker,
|
||||
// little-endian length counting the checksum, send pattern, message type, then
|
||||
// the fields in order.
|
||||
func TestEncodeFrameMatchesTheDocumentedTrigger(t *testing.T) {
|
||||
at := time.Unix(1756813256, 0)
|
||||
got, err := encodeFrame(msgRealtimeTrigger, []cmdField{
|
||||
rawField(0xa1, 0x22),
|
||||
uintField(0xa2, 1),
|
||||
varField(0xa3, 300),
|
||||
timestampField(at),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("encodeFrame: %v", err)
|
||||
}
|
||||
|
||||
want := "ff091f0003000f0057" + // header: marker, length 31, send pattern, type 0057
|
||||
"a10122" + // a1: one byte, no value type
|
||||
"a2020101" + // a2: ui 1 — updates on
|
||||
"a305032c010000" + // a3: var 300 — the window in seconds
|
||||
"fe0503c8d7b668" // fe: var — the sender's clock
|
||||
wantWithSum := want + "21"
|
||||
if h := encodeHex(got); h != wantWithSum {
|
||||
// Recompute the checksum in the message so a mismatch says which half broke.
|
||||
body, _ := decodeHex(want)
|
||||
t.Fatalf("frame = %s\nwant %s (checksum over the body is %02x)", h, wantWithSum, xorChecksum(body))
|
||||
}
|
||||
if len(got) != 31 {
|
||||
t.Errorf("frame is %d bytes, want 31", len(got))
|
||||
}
|
||||
if binary.LittleEndian.Uint16(got[2:4]) != uint16(len(got)) {
|
||||
t.Errorf("header length %d does not match the frame's %d bytes", binary.LittleEndian.Uint16(got[2:4]), len(got))
|
||||
}
|
||||
}
|
||||
|
||||
// A frame is only self-consistent if XORing every byte, checksum included,
|
||||
// comes to zero — which is exactly what the decoder checks.
|
||||
func TestEncodeFrameChecksumClosesToZero(t *testing.T) {
|
||||
frame, err := encodeFrame(msgEVMode, []cmdField{
|
||||
rawField(0xa1, 0x22),
|
||||
uintField(0xa2, mqttModeValues[modeStartCharge]),
|
||||
timestampField(time.Unix(1756813256, 0)),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("encodeFrame: %v", err)
|
||||
}
|
||||
if sum := xorChecksum(frame); sum != 0 {
|
||||
t.Errorf("XOR over the whole frame = %02x, want 00", sum)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEncodeFrameRejectsBadMessageType(t *testing.T) {
|
||||
for _, mt := range []string{"", "01", "zz01", "01020304"} {
|
||||
if _, err := encodeFrame(mt, []cmdField{rawField(0xa1, 0x22)}); err == nil {
|
||||
t.Errorf("encodeFrame(%q) accepted a message type it should not", mt)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// buildInbound assembles a device-style frame the way the charger sends one:
|
||||
// the receive pattern, and no counter byte before the first field.
|
||||
func buildInbound(t *testing.T, msgType string, body []byte) []byte {
|
||||
t.Helper()
|
||||
mt, err := decodeHex(msgType)
|
||||
if err != nil {
|
||||
t.Fatalf("bad message type %q: %v", msgType, err)
|
||||
}
|
||||
out := append([]byte{}, frameMarker...)
|
||||
out = binary.LittleEndian.AppendUint16(out, uint16(frameHeaderLen+len(body)+1))
|
||||
out = append(out, 0x03, 0x01, 0x0f)
|
||||
out = append(out, mt...)
|
||||
out = append(out, body...)
|
||||
return append(out, xorChecksum(out))
|
||||
}
|
||||
|
||||
func field(name, typ byte, value ...byte) []byte {
|
||||
return append([]byte{name, byte(len(value) + 1), typ}, value...)
|
||||
}
|
||||
|
||||
func TestDecodeFrameReadsTelemetry(t *testing.T) {
|
||||
var body []byte
|
||||
body = append(body, field(0xa2, typeInt16LE, 0xfd, 0x08)...) // 2301 -> 230.1 V
|
||||
body = append(body, field(0xa5, typeInt16LE, 0xa0, 0x00)...) // 160 -> 16.0 A
|
||||
body = append(body, field(0xa8, typeInt32LE, 0x60, 0x0e, 0x00, 0x00)...)
|
||||
body = append(body, field(0xa9, typeInt32LE, 0x8d, 0x0e, 0x00, 0x00)...)
|
||||
body = append(body, field(0xaa, typeInt32LE, 0xd4, 0x30, 0x00, 0x00)...)
|
||||
body = append(body, field(0xae, typeInt32LE, 0x2d, 0x00, 0x00, 0x00)...)
|
||||
body = append(body, []byte{0xbb, 0x01, 0x02}...) // single byte, no value type
|
||||
body = append(body, field(0xb8, typeUint8, 0x02)...)
|
||||
|
||||
msgType, values, err := decodeFrame(buildInbound(t, msgEVTelemetry, body))
|
||||
if err != nil {
|
||||
t.Fatalf("decodeFrame: %v", err)
|
||||
}
|
||||
if msgType != msgEVTelemetry {
|
||||
t.Errorf("message type = %s, want %s", msgType, msgEVTelemetry)
|
||||
}
|
||||
want := map[string]float64{
|
||||
"voltageL1": 230.1,
|
||||
"currentL1": 16,
|
||||
"powerTotal": 3680,
|
||||
"sessionSeconds": 3725,
|
||||
"sessionWh": 12500,
|
||||
"startCountdownSeconds": 45,
|
||||
"status": 2,
|
||||
"ocppStatus": 2,
|
||||
}
|
||||
for k, v := range want {
|
||||
got, ok := values[k].(float64)
|
||||
if !ok {
|
||||
t.Errorf("%s missing from the decoded values (%v)", k, values[k])
|
||||
continue
|
||||
}
|
||||
if got != v {
|
||||
t.Errorf("%s = %v, want %v", k, got, v)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// The settings message carries the charger's own view of its LAN side and its
|
||||
// schedule, which are two- and four-byte fields read differently from the
|
||||
// telemetry's: a clock field is a minute and an hour, not a number.
|
||||
func TestDecodeFrameReadsSettings(t *testing.T) {
|
||||
var body []byte
|
||||
body = append(body, field(0xa8, typeInt16LE, 0x40, 0x01)...) // 320 -> 32.0 A
|
||||
body = append(body, field(0xb7, typeUint8, 0x01)...) // Modbus TCP on
|
||||
body = append(body, field(0xcf, typeInt16LE, 0xf6, 0x01)...) // port 502
|
||||
body = append(body, field(0xd0, typeString, []byte("192.168.1.44")...)...)
|
||||
body = append(body, field(0xe7, typeInt16LE, 0x00, 0x16)...) // 22:00
|
||||
body = append(body, field(0xdf, typeUint8, 0x01)...) // boost running
|
||||
|
||||
_, values, err := decodeFrame(buildInbound(t, msgEVParams, body))
|
||||
if err != nil {
|
||||
t.Fatalf("decodeFrame: %v", err)
|
||||
}
|
||||
if v, _ := values["maxCurrentSetA"].(float64); v != 32 {
|
||||
t.Errorf("maxCurrentSetA = %v, want 32", values["maxCurrentSetA"])
|
||||
}
|
||||
if v, _ := values["modbusPort"].(float64); v != 502 {
|
||||
t.Errorf("modbusPort = %v, want 502", values["modbusPort"])
|
||||
}
|
||||
if v, _ := values["ipAddress"].(string); v != "192.168.1.44" {
|
||||
t.Errorf("ipAddress = %q, want 192.168.1.44", values["ipAddress"])
|
||||
}
|
||||
// The two bytes are minute then hour, so reading them as a plain little-endian
|
||||
// number would give 5632 rather than a time of day.
|
||||
if v, _ := values["weekStart"].(string); v != "22:00" {
|
||||
t.Errorf("weekStart = %q, want 22:00", values["weekStart"])
|
||||
}
|
||||
if v, _ := values["boostMode"].(float64); v != 1 {
|
||||
t.Errorf("boostMode = %v, want 1", values["boostMode"])
|
||||
}
|
||||
}
|
||||
|
||||
// A frame reaches us over a cloud connection we do not control, so a truncated
|
||||
// or corrupted one must be refused rather than read as a charger reporting
|
||||
// zeros — which would silently show a charging car as idle.
|
||||
func TestDecodeFrameRejectsDamagedFrames(t *testing.T) {
|
||||
good := buildInbound(t, msgEVTelemetry, field(0xbb, typeUint8, 0x02))
|
||||
|
||||
corrupt := append([]byte{}, good...)
|
||||
corrupt[len(corrupt)-2] ^= 0xff
|
||||
if _, _, err := decodeFrame(corrupt); err == nil {
|
||||
t.Error("a frame with a flipped value byte passed the checksum")
|
||||
}
|
||||
|
||||
truncated := append([]byte{}, good[:len(good)-3]...)
|
||||
if _, _, err := decodeFrame(truncated); err == nil {
|
||||
t.Error("a truncated frame was accepted")
|
||||
}
|
||||
|
||||
wrongMarker := append([]byte{}, good...)
|
||||
wrongMarker[0] = 0xfe
|
||||
if _, _, err := decodeFrame(wrongMarker); err == nil {
|
||||
t.Error("a frame without the Anker marker was accepted")
|
||||
}
|
||||
|
||||
if _, _, err := decodeFrame([]byte{0xff, 0x09}); err == nil {
|
||||
t.Error("a frame too short to hold a header was accepted")
|
||||
}
|
||||
}
|
||||
|
||||
// Some inbound frames carry a counter byte between the header and the first
|
||||
// field; skipping it wrongly would shift every field name by one.
|
||||
func TestDecodeFrameSkipsTheCounterByte(t *testing.T) {
|
||||
body := append([]byte{0x07}, field(0xbb, typeUint8, 0x05)...)
|
||||
_, values, err := decodeFrame(buildInbound(t, msgEVTelemetry, body))
|
||||
if err != nil {
|
||||
t.Fatalf("decodeFrame: %v", err)
|
||||
}
|
||||
if v, _ := values["status"].(float64); v != 5 {
|
||||
t.Errorf("status = %v, want 5 (the counter byte was not skipped)", values["status"])
|
||||
}
|
||||
}
|
||||
|
||||
// A message type we have no map for still has to parse, so an unknown frame is
|
||||
// an empty answer rather than an error that hides the ones we can read.
|
||||
func TestDecodeFrameOfAnUnmappedTypeIsEmpty(t *testing.T) {
|
||||
_, values, err := decodeFrame(buildInbound(t, "0400", field(0xa2, typeUint8, 0x01)))
|
||||
if err != nil {
|
||||
t.Fatalf("decodeFrame: %v", err)
|
||||
}
|
||||
if len(values) != 0 {
|
||||
t.Errorf("values = %v, want none for an unmapped message type", values)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDecodeValueSignsAndScales(t *testing.T) {
|
||||
// Two's-complement over two bytes: a relay reading below zero must stay below
|
||||
// zero rather than wrapping to 6553.5.
|
||||
v, ok := decodeValue(typeInt16LE, []byte{0xf6, 0xff}, mqttField{name: "x", factor: 0.1})
|
||||
if !ok || v.(float64) != -1 {
|
||||
t.Errorf("signed 2-byte value = %v (ok=%v), want -1", v, ok)
|
||||
}
|
||||
// The same bytes read unsigned are a large positive number, which is what the
|
||||
// fields marked unsigned actually mean.
|
||||
v, ok = decodeValue(typeInt16LE, []byte{0xf6, 0xff}, mqttField{name: "x", unsigned: true})
|
||||
if !ok || v.(float64) != 65526 {
|
||||
t.Errorf("unsigned 2-byte value = %v (ok=%v), want 65526", v, ok)
|
||||
}
|
||||
// A value shorter than its type is dropped rather than read as a smaller one.
|
||||
if _, ok := decodeValue(typeInt32LE, []byte{0x01, 0x02}, mqttField{name: "x"}); ok {
|
||||
t.Error("a 2-byte value was accepted for a 4-byte type")
|
||||
}
|
||||
// Scaling must not leave floating-point dust behind.
|
||||
v, _ = decodeValue(typeInt32LE, []byte{0xd4, 0x30, 0x00, 0x00}, mqttField{name: "x", factor: 0.001})
|
||||
if v.(float64) != 12.5 {
|
||||
t.Errorf("scaled value = %v, want 12.5", v)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPrintableKeepsOnlyReadableText(t *testing.T) {
|
||||
if got := printable([]byte("192.168.1.44\x00\x00")); got != "192.168.1.44" {
|
||||
t.Errorf("printable = %q, want %q", got, "192.168.1.44")
|
||||
}
|
||||
}
|
||||
|
||||
func TestDecodeHexRejectsRubbish(t *testing.T) {
|
||||
for _, s := range []string{"abc", "zz", "00 11"} {
|
||||
if _, err := decodeHex(s); err == nil {
|
||||
t.Errorf("decodeHex(%q) accepted a non-hex string", s)
|
||||
}
|
||||
}
|
||||
b, err := decodeHex("FF09")
|
||||
if err != nil || len(b) != 2 || b[0] != 0xff || b[1] != 0x09 {
|
||||
t.Errorf("decodeHex(%q) = %v, %v", "FF09", b, err)
|
||||
}
|
||||
if s := encodeHex([]byte{0xff, 0x09}); s != "ff09" {
|
||||
t.Errorf("encodeHex = %q, want ff09", s)
|
||||
}
|
||||
if strings.ToUpper(encodeHex([]byte{0xab})) != "AB" {
|
||||
t.Errorf("encodeHex is not lowercase hex")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,371 @@
|
||||
package ankersolix
|
||||
|
||||
// What a charger reports over the cloud, and the two capabilities built on it.
|
||||
//
|
||||
// The snapshot below deliberately borrows the field names ModbusSnapshot uses
|
||||
// for the same quantities — status, voltageL1, powerTotal, sessionWh, settings —
|
||||
// because they are the same charger read two ways, and a view that can render
|
||||
// one should not need a second layout for the other. Where the transports differ
|
||||
// the names differ with them: the cloud carries the boost flag and the plug and
|
||||
// start countdowns, which no register holds, while the register map carries the
|
||||
// relay temperatures and the reactive and apparent power, which no cloud message
|
||||
// sends.
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// MqttSnapshot is one charger's state as the cloud reports it. A field the
|
||||
// charger has not sent stays nil rather than zero, so "not reported" and "zero"
|
||||
// stay distinguishable.
|
||||
type MqttSnapshot struct {
|
||||
Serial string `json:"serial"`
|
||||
Model string `json:"model,omitempty"`
|
||||
|
||||
Status *int `json:"status,omitempty"`
|
||||
StatusDesc string `json:"statusDesc,omitempty"`
|
||||
|
||||
// Mode is the operational mode the charger is effectively in and ModeOptions
|
||||
// the ones it can be moved to, derived exactly as the cloud view derives
|
||||
// them — except that here the boost flag and the countdowns they depend on
|
||||
// are actually available.
|
||||
Mode string `json:"mode,omitempty"`
|
||||
ModeOptions []string `json:"modeOptions,omitempty"`
|
||||
|
||||
VoltageL1 *float64 `json:"voltageL1,omitempty"`
|
||||
VoltageL2 *float64 `json:"voltageL2,omitempty"`
|
||||
VoltageL3 *float64 `json:"voltageL3,omitempty"`
|
||||
CurrentL1 *float64 `json:"currentL1,omitempty"`
|
||||
CurrentL2 *float64 `json:"currentL2,omitempty"`
|
||||
CurrentL3 *float64 `json:"currentL3,omitempty"`
|
||||
|
||||
PowerL1 *float64 `json:"powerL1,omitempty"`
|
||||
PowerL2 *float64 `json:"powerL2,omitempty"`
|
||||
PowerL3 *float64 `json:"powerL3,omitempty"`
|
||||
PowerTotal *float64 `json:"powerTotal,omitempty"`
|
||||
|
||||
SessionSeconds *float64 `json:"sessionSeconds,omitempty"`
|
||||
SessionWh *float64 `json:"sessionWh,omitempty"`
|
||||
|
||||
// The countdowns the charger runs before a session: how long it will wait for
|
||||
// a plug, and how long a start delay still has to go. They are why a charger
|
||||
// that has been told to start can sit in "preparing" without being broken.
|
||||
PlugCountdownSeconds *float64 `json:"plugCountdownSeconds,omitempty"`
|
||||
StartCountdownSeconds *float64 `json:"startCountdownSeconds,omitempty"`
|
||||
ChargingWindowSeconds *float64 `json:"chargingWindowSeconds,omitempty"`
|
||||
|
||||
PhaseMode *int `json:"phaseMode,omitempty"`
|
||||
ChargingMode *int `json:"chargingMode,omitempty"`
|
||||
BoostMode *bool `json:"boostMode,omitempty"`
|
||||
Plugged *bool `json:"plugged,omitempty"`
|
||||
|
||||
CPSignal *int `json:"cpSignal,omitempty"`
|
||||
CPSignalDesc string `json:"cpSignalDesc,omitempty"`
|
||||
|
||||
OcppStatus *int `json:"ocppStatus,omitempty"`
|
||||
OcppStatusDesc string `json:"ocppStatusDesc,omitempty"`
|
||||
|
||||
LoadBalancing *bool `json:"loadBalancing,omitempty"`
|
||||
SolarBalancing *bool `json:"solarBalancing,omitempty"`
|
||||
LEDBrightness *int `json:"ledBrightness,omitempty"`
|
||||
MinCurrentA *float64 `json:"minCurrentA,omitempty"`
|
||||
MaxCurrentA *float64 `json:"maxCurrentA,omitempty"`
|
||||
|
||||
Settings *MqttSettings `json:"settings,omitempty"`
|
||||
|
||||
// Local reports what the charger says about its own LAN side: whether Modbus
|
||||
// TCP is switched on, and at which address. It is the one answer the Modbus
|
||||
// mode's setup screen otherwise has to be given by hand.
|
||||
Local *MqttLocalAccess `json:"local,omitempty"`
|
||||
|
||||
// TelemetryAt and SettingsAt are when each half of the snapshot last arrived;
|
||||
// Live says the fast stream is currently flowing.
|
||||
TelemetryAt string `json:"telemetryAt,omitempty"`
|
||||
SettingsAt string `json:"settingsAt,omitempty"`
|
||||
Live bool `json:"live"`
|
||||
}
|
||||
|
||||
// MqttSettings is what the charger is set to, as opposed to what it is doing —
|
||||
// the same distinction ModbusSettings draws over the register map.
|
||||
type MqttSettings struct {
|
||||
MaxCurrentA *float64 `json:"maxCurrentA,omitempty"`
|
||||
AutoStart *bool `json:"autoStart,omitempty"`
|
||||
AutoRestart *bool `json:"autoRestart,omitempty"`
|
||||
RandomDelay *bool `json:"randomDelay,omitempty"`
|
||||
PlugLock *bool `json:"plugLock,omitempty"`
|
||||
ScheduleEnabled *bool `json:"scheduleEnabled,omitempty"`
|
||||
WeekStart string `json:"weekStart,omitempty"`
|
||||
WeekEnd string `json:"weekEnd,omitempty"`
|
||||
WeekendStart string `json:"weekendStart,omitempty"`
|
||||
WeekendEnd string `json:"weekendEnd,omitempty"`
|
||||
MainBreakerLimitA *float64 `json:"mainBreakerLimitA,omitempty"`
|
||||
SolarMinCurrentA *float64 `json:"solarMinCurrentA,omitempty"`
|
||||
AutoPhaseSwitching *bool `json:"autoPhaseSwitching,omitempty"`
|
||||
}
|
||||
|
||||
// MqttLocalAccess is the charger's own view of its Modbus TCP server.
|
||||
type MqttLocalAccess struct {
|
||||
ModbusEnabled *bool `json:"modbusEnabled,omitempty"`
|
||||
Host string `json:"host,omitempty"`
|
||||
Port *int `json:"port,omitempty"`
|
||||
TimeoutSeconds *int `json:"timeoutSeconds,omitempty"`
|
||||
}
|
||||
|
||||
// ---- the capabilities --------------------------------------------------------
|
||||
|
||||
// mqttStatus reads one charger's state over the cloud. The charger publishes
|
||||
// nothing unless asked, so this arms the telemetry trigger and waits for the
|
||||
// next frame; inside an already-armed window the frame that has since arrived
|
||||
// answers immediately.
|
||||
func (p *Plugin) mqttStatus(ctx context.Context, sn string) (json.RawMessage, error) {
|
||||
model, err := p.chargerModel(ctx, sn)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
conn, err := p.mqttClient(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := conn.listen(ctx, model, sn); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Re-arm whenever the window is spent or close to it, so a poll never lands
|
||||
// in the gap between the last frame and the trigger expiring.
|
||||
_, _, _, triggered := conn.snapshotOf(sn)
|
||||
if time.Until(triggered) < triggerRenew {
|
||||
if err := p.mqttTrigger(ctx, conn, model, sn, triggerWindow); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
// Anything older than the trigger's own interval is stale; wait for the next.
|
||||
cutoff := time.Now().Add(-triggerRenew)
|
||||
live, err := conn.waitFor(ctx, sn, func(st *deviceState) bool {
|
||||
return st.telemetryAt.After(cutoff)
|
||||
}, statusWait)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
values, telemetryAt, settingsAt, _ := conn.snapshotOf(sn)
|
||||
if len(values) == 0 {
|
||||
return nil, fmt.Errorf("anker-solix: charger %s did not answer over the cloud; it may be offline", sn)
|
||||
}
|
||||
snap := projectMqttSnapshot(sn, model, values)
|
||||
snap.Live = live
|
||||
if !telemetryAt.IsZero() {
|
||||
snap.TelemetryAt = telemetryAt.UTC().Format(time.RFC3339)
|
||||
}
|
||||
if !settingsAt.IsZero() {
|
||||
snap.SettingsAt = settingsAt.UTC().Format(time.RFC3339)
|
||||
}
|
||||
return json.Marshal(snap)
|
||||
}
|
||||
|
||||
// mqttCommandDoc is what a cloud command answers with. Confirmed says the
|
||||
// charger sent a message back within commandWait: publishing is fire-and-forget,
|
||||
// so an unconfirmed command is not a failed one — it is one whose effect has not
|
||||
// been seen yet.
|
||||
type mqttCommandDoc struct {
|
||||
Serial string `json:"serial"`
|
||||
Command string `json:"command"`
|
||||
Status string `json:"status"`
|
||||
Confirmed bool `json:"confirmed"`
|
||||
Detail string `json:"detail,omitempty"`
|
||||
}
|
||||
|
||||
// mqttCommands maps the names this transport accepts to the charger mode each
|
||||
// one asks for. The short names are what the control endpoint sends; the long
|
||||
// ones are the mode names the snapshot reports in modeOptions, so a caller can
|
||||
// send back what it was offered.
|
||||
var mqttCommands = map[string]string{
|
||||
"start": modeStartCharge,
|
||||
modeStartCharge: modeStartCharge,
|
||||
"stop": modeStopCharge,
|
||||
modeStopCharge: modeStopCharge,
|
||||
"boost": modeBoostCharge,
|
||||
modeBoostCharge: modeBoostCharge,
|
||||
"skip-delay": modeSkipDelay,
|
||||
modeSkipDelay: modeSkipDelay,
|
||||
}
|
||||
|
||||
// mqttCommand issues one control command over the cloud.
|
||||
func (p *Plugin) mqttCommand(ctx context.Context, sn, command string, amps float64) (json.RawMessage, error) {
|
||||
// Validate before touching the cloud: a mistyped command should not cost a
|
||||
// sign-in, a broker connection and a certificate fetch to be told no.
|
||||
command = strings.ToLower(strings.TrimSpace(command))
|
||||
mode, isMode := mqttCommands[command]
|
||||
switch {
|
||||
case isMode:
|
||||
case command == "limit":
|
||||
if err := checkMaxCurrent(amps); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
case command == "trigger":
|
||||
default:
|
||||
return nil, fmt.Errorf("anker-solix: %q is not a cloud command (start, stop, boost, skip-delay, limit, trigger)", command)
|
||||
}
|
||||
|
||||
model, err := p.chargerModel(ctx, sn)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
conn, err := p.mqttClient(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// Listen before commanding: the charger confirms a control change with a
|
||||
// message, and a subscription made afterwards would miss it.
|
||||
if err := conn.listen(ctx, model, sn); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
_, _, before, _ := conn.snapshotOf(sn)
|
||||
|
||||
switch {
|
||||
case isMode:
|
||||
err = p.mqttSetMode(ctx, conn, model, sn, mode)
|
||||
case command == "limit":
|
||||
err = p.mqttSetMaxCurrent(ctx, conn, model, sn, amps)
|
||||
default:
|
||||
err = p.mqttTrigger(ctx, conn, model, sn, triggerWindow)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// The charger answers a control change with a settings message. Waiting for
|
||||
// it turns "published" into "the charger has it".
|
||||
confirmed, waitErr := conn.waitFor(ctx, sn, func(st *deviceState) bool {
|
||||
return st.settingsAt.After(before)
|
||||
}, commandWait)
|
||||
doc := mqttCommandDoc{Serial: sn, Command: command, Status: "accepted", Confirmed: confirmed}
|
||||
if waitErr != nil {
|
||||
// The command left; only the confirmation did not. Say so rather than
|
||||
// reporting a failure the charger may well have acted on.
|
||||
doc.Detail = "sent, but the cloud connection dropped before the charger confirmed it"
|
||||
} else if !confirmed {
|
||||
doc.Detail = "sent; the charger has not confirmed it yet"
|
||||
}
|
||||
return json.Marshal(doc)
|
||||
}
|
||||
|
||||
// ---- projection --------------------------------------------------------------
|
||||
|
||||
// projectMqttSnapshot turns the named values collected from a charger's messages
|
||||
// into the snapshot. Every read is by name and optional: a message type we have
|
||||
// not seen simply leaves its fields unset.
|
||||
func projectMqttSnapshot(sn, model string, v map[string]any) MqttSnapshot {
|
||||
snap := MqttSnapshot{Serial: sn, Model: model}
|
||||
|
||||
num := func(key string) *float64 {
|
||||
f, ok := v[key].(float64)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
return &f
|
||||
}
|
||||
whole := func(key string) *int {
|
||||
f, ok := v[key].(float64)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
n := int(f)
|
||||
return &n
|
||||
}
|
||||
flag := func(key string) *bool {
|
||||
f, ok := v[key].(float64)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
b := f != 0
|
||||
return &b
|
||||
}
|
||||
text := func(key string) string {
|
||||
s, _ := v[key].(string)
|
||||
return strings.TrimSpace(s)
|
||||
}
|
||||
|
||||
snap.VoltageL1, snap.VoltageL2, snap.VoltageL3 = num("voltageL1"), num("voltageL2"), num("voltageL3")
|
||||
snap.CurrentL1, snap.CurrentL2, snap.CurrentL3 = num("currentL1"), num("currentL2"), num("currentL3")
|
||||
snap.PowerL1, snap.PowerL2, snap.PowerL3 = num("powerL1"), num("powerL2"), num("powerL3")
|
||||
snap.PowerTotal = num("powerTotal")
|
||||
snap.SessionSeconds, snap.SessionWh = num("sessionSeconds"), num("sessionWh")
|
||||
snap.PlugCountdownSeconds = num("plugCountdownSeconds")
|
||||
snap.StartCountdownSeconds = num("startCountdownSeconds")
|
||||
snap.ChargingWindowSeconds = num("chargingWindowSeconds")
|
||||
snap.PhaseMode, snap.ChargingMode = whole("phaseMode"), whole("chargingMode")
|
||||
snap.BoostMode, snap.Plugged = flag("boostMode"), flag("plugged")
|
||||
snap.LoadBalancing, snap.SolarBalancing = flag("loadBalancing"), flag("solarBalancing")
|
||||
snap.LEDBrightness = whole("ledBrightness")
|
||||
snap.MinCurrentA, snap.MaxCurrentA = num("minCurrentA"), num("maxCurrentA")
|
||||
|
||||
if s := whole("status"); s != nil {
|
||||
snap.Status, snap.StatusDesc = s, statusName(*s)
|
||||
}
|
||||
if s := whole("ocppStatus"); s != nil {
|
||||
snap.OcppStatus, snap.OcppStatusDesc = s, ocppStatusNames[*s]
|
||||
}
|
||||
if s := whole("cpSignal"); s != nil {
|
||||
snap.CPSignal, snap.CPSignalDesc = s, cpSignalNames[*s]
|
||||
}
|
||||
|
||||
// The mode the cloud view can only guess at, with the two countdowns and the
|
||||
// boost flag it never sees.
|
||||
if snap.StatusDesc != "" {
|
||||
boost := snap.BoostMode != nil && *snap.BoostMode
|
||||
snap.Mode = chargerMode(snap.StatusDesc, boost, intOrZero(snap.PlugCountdownSeconds), intOrZero(snap.StartCountdownSeconds))
|
||||
snap.ModeOptions = chargerModeOptions(snap.Mode, snap.StatusDesc)
|
||||
}
|
||||
|
||||
set := &MqttSettings{
|
||||
MaxCurrentA: num("maxCurrentSetA"),
|
||||
AutoStart: flag("autoStartSwitch"),
|
||||
AutoRestart: flag("autoRestartSwitch"),
|
||||
RandomDelay: flag("randomDelaySwitch"),
|
||||
MainBreakerLimitA: num("mainBreakerLimitA"),
|
||||
SolarMinCurrentA: num("solarMinCurrentA"),
|
||||
AutoPhaseSwitching: flag("autoPhaseSwitch"),
|
||||
WeekStart: text("weekStart"),
|
||||
WeekEnd: text("weekEnd"),
|
||||
WeekendStart: text("weekendStart"),
|
||||
WeekendEnd: text("weekendEnd"),
|
||||
}
|
||||
// Both of these read 1 for on and 2 for off, which is the charger's own
|
||||
// convention on these two registers and nowhere else.
|
||||
if s := whole("plugLockSwitch"); s != nil {
|
||||
b := *s == 1
|
||||
set.PlugLock = &b
|
||||
}
|
||||
if s := whole("scheduleSwitch"); s != nil {
|
||||
b := *s == 1
|
||||
set.ScheduleEnabled = &b
|
||||
}
|
||||
if *set != (MqttSettings{}) {
|
||||
snap.Settings = set
|
||||
}
|
||||
|
||||
local := &MqttLocalAccess{
|
||||
ModbusEnabled: flag("modbusSwitch"),
|
||||
Host: text("ipAddress"),
|
||||
Port: whole("modbusPort"),
|
||||
TimeoutSeconds: whole("modbusTimeoutSeconds"),
|
||||
}
|
||||
if *local != (MqttLocalAccess{}) {
|
||||
snap.Local = local
|
||||
}
|
||||
return snap
|
||||
}
|
||||
|
||||
// intOrZero reads an optional number as an int, treating "not reported" as zero
|
||||
// — which is what the mode derivation means by a countdown that is not running.
|
||||
func intOrZero(v *float64) int {
|
||||
if v == nil {
|
||||
return 0
|
||||
}
|
||||
return int(*v)
|
||||
}
|
||||
@@ -0,0 +1,146 @@
|
||||
package ankersolix
|
||||
|
||||
import (
|
||||
"slices"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestProjectSnapshotNamesTheChargerState(t *testing.T) {
|
||||
snap := projectMqttSnapshot("SN1", "A5191", map[string]any{
|
||||
"voltageL1": 230.1,
|
||||
"currentL1": 16.0,
|
||||
"powerTotal": 3680.0,
|
||||
"sessionSeconds": 3725.0,
|
||||
"sessionWh": 12500.0,
|
||||
"status": 2.0,
|
||||
"ocppStatus": 2.0,
|
||||
"cpSignal": 5.0,
|
||||
"phaseMode": 1.0,
|
||||
})
|
||||
|
||||
if snap.Serial != "SN1" || snap.Model != "A5191" {
|
||||
t.Errorf("snapshot identifies %s/%s", snap.Serial, snap.Model)
|
||||
}
|
||||
if snap.StatusDesc != stateCharging {
|
||||
t.Errorf("statusDesc = %q, want %q", snap.StatusDesc, stateCharging)
|
||||
}
|
||||
if snap.OcppStatusDesc != "connected" {
|
||||
t.Errorf("ocppStatusDesc = %q, want connected", snap.OcppStatusDesc)
|
||||
}
|
||||
// The control-pilot names are the charger's own, shared with the Modbus map.
|
||||
if snap.CPSignalDesc != cpSignalNames[5] {
|
||||
t.Errorf("cpSignalDesc = %q, want %q", snap.CPSignalDesc, cpSignalNames[5])
|
||||
}
|
||||
if snap.VoltageL1 == nil || *snap.VoltageL1 != 230.1 {
|
||||
t.Errorf("voltageL1 = %v", snap.VoltageL1)
|
||||
}
|
||||
// A quantity the charger did not send stays nil, so a view can tell "not
|
||||
// reported" from "zero" — an unplugged charger really does read 0 A.
|
||||
if snap.VoltageL2 != nil {
|
||||
t.Errorf("voltageL2 = %v, want nil for a value that was not sent", *snap.VoltageL2)
|
||||
}
|
||||
if snap.Settings != nil {
|
||||
t.Errorf("settings = %+v, want none until a settings message arrives", snap.Settings)
|
||||
}
|
||||
if snap.Local != nil {
|
||||
t.Errorf("local = %+v, want none until a settings message arrives", snap.Local)
|
||||
}
|
||||
}
|
||||
|
||||
// The mode is the one thing the cloud REST view can only approximate: it has no
|
||||
// boost flag and no countdowns, so a charger waiting out a start delay reads to
|
||||
// it as simply started. Over MQTT those fields exist, and the mode must use them.
|
||||
func TestProjectSnapshotDerivesModeFromTheMqttOnlySignals(t *testing.T) {
|
||||
waiting := projectMqttSnapshot("SN1", "A5191", map[string]any{
|
||||
"status": 1.0, // preparing
|
||||
"startCountdownSeconds": 45.0,
|
||||
})
|
||||
if waiting.Mode != modeWaitStart {
|
||||
t.Errorf("mode = %q, want %q while a start delay is running", waiting.Mode, modeWaitStart)
|
||||
}
|
||||
if !slices.Contains(waiting.ModeOptions, modeSkipDelay) {
|
||||
t.Errorf("modeOptions = %v, want the delay to be skippable", waiting.ModeOptions)
|
||||
}
|
||||
|
||||
plugging := projectMqttSnapshot("SN1", "A5191", map[string]any{
|
||||
"status": 1.0,
|
||||
"plugCountdownSeconds": 60.0,
|
||||
})
|
||||
if plugging.Mode != modeWaitPlug {
|
||||
t.Errorf("mode = %q, want %q while it waits for a plug", plugging.Mode, modeWaitPlug)
|
||||
}
|
||||
|
||||
boosting := projectMqttSnapshot("SN1", "A5191", map[string]any{
|
||||
"status": 2.0, // charging
|
||||
"boostMode": 1.0,
|
||||
})
|
||||
if boosting.Mode != modeBoostCharge {
|
||||
t.Errorf("mode = %q, want %q while boost is running", boosting.Mode, modeBoostCharge)
|
||||
}
|
||||
|
||||
idle := projectMqttSnapshot("SN1", "A5191", map[string]any{"status": 0.0})
|
||||
if idle.Mode != modeStopCharge {
|
||||
t.Errorf("mode = %q, want %q in standby", idle.Mode, modeStopCharge)
|
||||
}
|
||||
if !slices.Contains(idle.ModeOptions, modeStartCharge) {
|
||||
t.Errorf("modeOptions = %v, want a standby charger to be startable", idle.ModeOptions)
|
||||
}
|
||||
}
|
||||
|
||||
// Two of the charger's settings read 1 for on and 2 for off, which is the
|
||||
// opposite of every other flag it sends: read as booleans they would both come
|
||||
// back on.
|
||||
func TestProjectSnapshotHandlesTheInvertedSwitches(t *testing.T) {
|
||||
off := projectMqttSnapshot("SN1", "A5191", map[string]any{
|
||||
"plugLockSwitch": 2.0,
|
||||
"scheduleSwitch": 2.0,
|
||||
})
|
||||
if off.Settings == nil {
|
||||
t.Fatal("settings missing")
|
||||
}
|
||||
if off.Settings.PlugLock == nil || *off.Settings.PlugLock {
|
||||
t.Errorf("plugLock = %v, want off for the charger's value 2", off.Settings.PlugLock)
|
||||
}
|
||||
if off.Settings.ScheduleEnabled == nil || *off.Settings.ScheduleEnabled {
|
||||
t.Errorf("scheduleEnabled = %v, want off for the charger's value 2", off.Settings.ScheduleEnabled)
|
||||
}
|
||||
|
||||
on := projectMqttSnapshot("SN1", "A5191", map[string]any{"plugLockSwitch": 1.0})
|
||||
if on.Settings == nil || on.Settings.PlugLock == nil || !*on.Settings.PlugLock {
|
||||
t.Errorf("plugLock = %v, want on for the charger's value 1", on.Settings)
|
||||
}
|
||||
}
|
||||
|
||||
// The settings message carries the charger's own view of its LAN side, which is
|
||||
// the address the Modbus mode otherwise has to be told by hand.
|
||||
func TestProjectSnapshotReportsLocalAccess(t *testing.T) {
|
||||
snap := projectMqttSnapshot("SN1", "A5191", map[string]any{
|
||||
"modbusSwitch": 1.0,
|
||||
"ipAddress": "192.168.1.44",
|
||||
"modbusPort": 502.0,
|
||||
"modbusTimeoutSeconds": 60.0,
|
||||
"maxCurrentSetA": 32.0,
|
||||
})
|
||||
if snap.Local == nil {
|
||||
t.Fatal("local access missing")
|
||||
}
|
||||
if snap.Local.ModbusEnabled == nil || !*snap.Local.ModbusEnabled {
|
||||
t.Errorf("modbusEnabled = %v, want on", snap.Local.ModbusEnabled)
|
||||
}
|
||||
if snap.Local.Host != "192.168.1.44" {
|
||||
t.Errorf("host = %q", snap.Local.Host)
|
||||
}
|
||||
if snap.Local.Port == nil || *snap.Local.Port != 502 {
|
||||
t.Errorf("port = %v, want 502", snap.Local.Port)
|
||||
}
|
||||
if snap.Settings == nil || snap.Settings.MaxCurrentA == nil || *snap.Settings.MaxCurrentA != 32 {
|
||||
t.Errorf("settings.maxCurrentA = %+v, want 32", snap.Settings)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProjectSnapshotOfNothingIsEmpty(t *testing.T) {
|
||||
snap := projectMqttSnapshot("SN1", "A5191", map[string]any{})
|
||||
if snap.Status != nil || snap.Mode != "" || snap.Settings != nil || snap.Local != nil {
|
||||
t.Errorf("an empty message set produced state: %+v", snap)
|
||||
}
|
||||
}
|
||||
@@ -25,8 +25,9 @@ import (
|
||||
"time"
|
||||
)
|
||||
|
||||
// session is one account's cloud state: the token it holds, and how long to
|
||||
// leave its login alone after a refusal.
|
||||
// session is one account's cloud state: the token it holds, how long to leave
|
||||
// its login alone after a refusal, and the broker connection its chargers are
|
||||
// commanded over.
|
||||
type session struct {
|
||||
mu sync.Mutex // guards tok and the backoff below, and serialises the login exchange
|
||||
tok *tokenInfo
|
||||
@@ -35,6 +36,17 @@ type session struct {
|
||||
loginRetryAt time.Time
|
||||
loginFails int
|
||||
|
||||
// The cloud MQTT half (cloudmqtt.go), under its own lock: a broker connection
|
||||
// takes a TLS handshake and a fetched certificate to open, so it outlives the
|
||||
// plugin instance for the same reason the token does. mqttMu guards all four
|
||||
// fields and is never held while mu is.
|
||||
mqttMu sync.Mutex
|
||||
mqttCreds *mqttCredentials
|
||||
mqttCredsAt time.Time
|
||||
mqttConn *mqttConn
|
||||
devices map[string]string // charger serial -> product code, from the account's inventory
|
||||
devicesAt time.Time
|
||||
|
||||
// lastUse is touched and read under sessionsMu, never under mu.
|
||||
lastUse time.Time
|
||||
}
|
||||
@@ -80,11 +92,25 @@ func pruneSessionsLocked() {
|
||||
cutoff := time.Now().Add(-sessionIdle)
|
||||
for k, s := range sessions {
|
||||
if s.lastUse.Before(cutoff) {
|
||||
// A pruned session must not leave its broker connection open; nothing
|
||||
// else holds a reference to close it afterwards.
|
||||
s.closeMqtt()
|
||||
delete(sessions, k)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// closeMqtt drops the session's broker connection, if it holds one.
|
||||
func (s *session) closeMqtt() {
|
||||
s.mqttMu.Lock()
|
||||
c := s.mqttConn
|
||||
s.mqttConn = nil
|
||||
s.mqttMu.Unlock()
|
||||
if c != nil {
|
||||
c.close()
|
||||
}
|
||||
}
|
||||
|
||||
// noteLogin records the outcome of a login attempt and, on failure, how long to
|
||||
// leave the account alone: one Anker has already disabled gets the full penalty,
|
||||
// anything else backs off exponentially up to loginRetryMax. Caller holds s.mu.
|
||||
|
||||
Reference in New Issue
Block a user