evcc-io
diff --git a/‎meter/ecoflow_mqtt.go‎
Lines changed: 348 additions & 0 deletions b/‎meter/ecoflow_mqtt.go‎
Lines changed: 348 additions & 0 deletions
@@ -0,0 +1,348 @@
+package meter
+
+import (
+	"context"
+	"crypto/hmac"
+	"crypto/sha256"
+	"crypto/tls"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"io"
+	"maps"
+	"net/http"
+	"os"
+	"strconv"
+	"sync"
+	"time"
+
+	paho "github.com/eclipse/paho.mqtt.golang"
+	"github.com/evcc-io/evcc/util"
+	"github.com/evcc-io/evcc/util/request"
+)
+
+// ecoflowMqttClient maintains a shared MQTT subscription to EcoFlow's public
+// IoT broker. The REST `quota/all` endpoint returns only a static subset of
+// parameters for Stream devices; live per-string PV power, per-second state
+// etc. are delivered exclusively via MQTT. Each configured EcoFlow meter
+// registers its device serial here and reads from the in-memory cache.
+type ecoflowMqttClient struct {
+	log *util.Logger
+
+	accessKey string
+	secretKey string
+	uri       string // REST base, e.g. https://api-e.ecoflow.com
+	http      *http.Client
+
+	startMu  sync.Mutex // serializes the initial certification/connect
+	mu       sync.Mutex
+	state    map[string]map[string]any // sn -> key -> latest value
+	devices  map[string]struct{}       // subscribed serials
+	client   paho.Client
+	username string // certificateAccount, used in quota topic
+	started  bool
+}
+
+// ecoflowMqttClients caches one client per (accessKey, uri) pair so we don't
+// exceed EcoFlow's 10 unique client-ID/day limit and to share the MQTT
+// connection across multiple ecoflow meters for the same account.
+var (
+	ecoflowMqttClientsMu sync.Mutex
+	ecoflowMqttClients   = map[string]*ecoflowMqttClient{}
+)
+
+func ecoflowMqttClientFor(accessKey, secretKey, uri string) *ecoflowMqttClient {
+	ecoflowMqttClientsMu.Lock()
+	defer ecoflowMqttClientsMu.Unlock()
+
+	key := accessKey + "|" + uri
+	if c, ok := ecoflowMqttClients[key]; ok {
+		return c
+	}
+	log := util.NewLogger("ecoflow-mqtt")
+	c := &ecoflowMqttClient{
+		log:       log,
+		accessKey: accessKey,
+		secretKey: secretKey,
+		uri:       uri,
+		http:      request.NewClient(log),
+		state:     make(map[string]map[string]any),
+		devices:   make(map[string]struct{}),
+	}
+	ecoflowMqttClients[key] = c
+	return c
+}
+
+// ecoflowCertResponse is the payload from GET /iot-open/sign/certification.
+type ecoflowCertResponse struct {
+	Code    string `json:"code"`
+	Message string `json:"message"`
+	Data    struct {
+		CertificateAccount  string `json:"certificateAccount"`
+		CertificatePassword string `json:"certificatePassword"`
+		URL                 string `json:"url"`
+		Port                string `json:"port"`
+		Protocol            string `json:"protocol"`
+	} `json:"data"`
+}
+
+// sign builds the HMAC-SHA256 signature for a signed EcoFlow public API
+// request as documented in EcoFlow's developer portal.
+func ecoflowSign(accessKey, secretKey, nonce, timestamp, query string) string {
+	target := "accessKey=" + accessKey + "&nonce=" + nonce + "&timestamp=" + timestamp
+	if query != "" {
+		target = query + "&" + target
+	}
+	h := hmac.New(sha256.New, []byte(secretKey))
+	h.Write([]byte(target))
+	return hex.EncodeToString(h.Sum(nil))
+}
+
+// certify fetches MQTT credentials from the public API.
+func (c *ecoflowMqttClient) certify(ctx context.Context) (*ecoflowCertResponse, error) {
+	nonce := strconv.Itoa(int(time.Now().UnixNano() % 1_000_000))
+	timestamp := strconv.FormatInt(time.Now().UnixMilli(), 10)
+	sign := ecoflowSign(c.accessKey, c.secretKey, nonce, timestamp, "")
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.uri+"/iot-open/sign/certification", nil)
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("accessKey", c.accessKey)
+	req.Header.Set("nonce", nonce)
+	req.Header.Set("timestamp", timestamp)
+	req.Header.Set("sign", sign)
+
+	resp, err := c.http.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("certification: http %d: %s", resp.StatusCode, string(body))
+	}
+
+	var cert ecoflowCertResponse
+	if err := json.Unmarshal(body, &cert); err != nil {
+		return nil, fmt.Errorf("certification: %w (body=%s)", err, string(body))
+	}
+	if cert.Code != "0" {
+		return nil, fmt.Errorf("certification: code=%s message=%s", cert.Code, cert.Message)
+	}
+	return &cert, nil
+}
+
+// ensureStarted lazily establishes the MQTT connection on the first device
+// registration. Subsequent calls are no-ops.
+func (c *ecoflowMqttClient) ensureStarted(ctx context.Context) error {
+	c.startMu.Lock()
+	defer c.startMu.Unlock()
+
+	c.mu.Lock()
+	started := c.started
+	c.mu.Unlock()
+	if started {
+		return nil
+	}
+
+	cert, err := c.certify(ctx)
+	if err != nil {
+		return err
+	}
+
+	port, err := strconv.Atoi(cert.Data.Port)
+	if err != nil {
+		return fmt.Errorf("invalid mqtt port %q: %w", cert.Data.Port, err)
+	}
+
+	// Stable client id that is also unique per host. EcoFlow's broker
+	// permits only a single concurrent connection per client id and imposes
+	// a 10 unique-ID/day limit per account. Hashing (accessKey + hostname)
+	// keeps the id stable across restarts while allowing multiple evcc
+	// instances (e.g. prod + dev) to coexist on the same access key.
+	hostname, _ := os.Hostname()
+	h := sha256.Sum256([]byte(c.accessKey + "|" + hostname))
+	clientID := "evcc-" + hex.EncodeToString(h[:6])
+
+	scheme := cert.Data.Protocol
+	if scheme == "" {
+		scheme = "ssl"
+	}
+	broker := fmt.Sprintf("%s://%s:%d", scheme, cert.Data.URL, port)
+
+	opts := paho.NewClientOptions()
+	opts.AddBroker(broker)
+	opts.SetUsername(cert.Data.CertificateAccount)
+	opts.SetPassword(cert.Data.CertificatePassword)
+	opts.SetClientID(clientID)
+	opts.SetCleanSession(true)
+	opts.SetAutoReconnect(true)
+	opts.SetMaxReconnectInterval(time.Minute)
+	opts.SetConnectTimeout(request.Timeout)
+	opts.SetKeepAlive(30 * time.Second)
+	opts.SetTLSConfig(&tls.Config{MinVersion: tls.VersionTLS12})
+	opts.SetOnConnectHandler(c.onConnect)
+	opts.SetConnectionLostHandler(func(_ paho.Client, err error) {
+		c.log.WARN.Printf("%s connection lost: %v", broker, err)
+	})
+
+	client := paho.NewClient(opts)
+
+	// Publish username before Connect() so the OnConnect handler (which may
+	// fire on a separate goroutine as soon as CONNACK arrives) can build the
+	// correct topic. The client handle is published after connect succeeds.
+	c.mu.Lock()
+	c.username = cert.Data.CertificateAccount
+	c.mu.Unlock()
+
+	c.log.INFO.Printf("connecting %s at %s as %s", clientID, broker, cert.Data.CertificateAccount)
+	tok := client.Connect()
+	if !tok.WaitTimeout(request.Timeout) {
+		return fmt.Errorf("mqtt connect: timeout")
+	}
+	if err := tok.Error(); err != nil {
+		return fmt.Errorf("mqtt connect: %w", err)
+	}
+
+	c.mu.Lock()
+	c.client = client
+	c.started = true
+	c.mu.Unlock()
+
+	return nil
+}
+
+// Register subscribes the client to live updates for the given device serial.
+// Safe to call multiple times; extra calls are no-ops.
+//
+// The serial is added to the device set *before* the connection is
+// established so that the OnConnect callback picks it up and subscribes.
+// This avoids a race where a duplicate manual Subscribe() would overlap with
+// the one fired from OnConnect on initial connect.
+func (c *ecoflowMqttClient) Register(ctx context.Context, sn string) error {
+	c.mu.Lock()
+	_, existed := c.devices[sn]
+	c.devices[sn] = struct{}{}
+	started := c.started
+	client := c.client
+	username := c.username
+	c.mu.Unlock()
+
+	if existed {
+		return nil
+	}
+
+	if !started {
+		// First caller bootstraps the connection; OnConnect will subscribe
+		// to every registered device once the link is up.
+		return c.ensureStarted(ctx)
+	}
+
+	// Connection already established by a previous Register call;
+	// subscribe just for this new device.
+	return c.subscribe(client, username, sn)
+}
+
+func (c *ecoflowMqttClient) subscribe(client paho.Client, username, sn string) error {
+	topic := fmt.Sprintf("/open/%s/%s/quota", username, sn)
+	c.log.DEBUG.Printf("subscribe %s", topic)
+
+	token := client.Subscribe(topic, 1, func(_ paho.Client, msg paho.Message) {
+		c.handleMessage(sn, msg.Payload())
+	})
+	if !token.WaitTimeout(request.Timeout) {
+		return fmt.Errorf("subscribe %s: timeout", topic)
+	}
+	return token.Error()
+}
+
+// onConnect is invoked after a (re)connection and re-subscribes to every
+// registered device.
+func (c *ecoflowMqttClient) onConnect(client paho.Client) {
+	c.mu.Lock()
+	username := c.username
+	sns := make([]string, 0, len(c.devices))
+	for sn := range c.devices {
+		sns = append(sns, sn)
+	}
+	c.mu.Unlock()
+
+	c.log.DEBUG.Printf("connected, resubscribing %d device(s)", len(sns))
+	for _, sn := range sns {
+		if err := c.subscribe(client, username, sn); err != nil {
+			c.log.ERROR.Printf("resubscribe %s: %v", sn, err)
+		}
+	}
+}
+
+// handleMessage parses a quota payload and updates the cache. Payload shapes
+// observed on EcoFlow public MQTT:
+//
+//  1. Stream devices emit a flat JSON object with parameters at the top
+//     level, e.g. {"powGetPv2":214.14,"gridConnectionPower":283.99}.
+//  2. Other devices wrap updates in a {"params": {...}} (or "param") envelope
+//     and may include metadata like timestamp/typeCode at the top level.
+//
+// We accept both shapes: if "params"/"param" is present we take it, otherwise
+// we merge the top-level object as-is (ignoring a few well-known metadata
+// fields that are not quota parameters).
+func (c *ecoflowMqttClient) handleMessage(sn string, payload []byte) {
+	c.log.TRACE.Printf("recv %s: %s", sn, string(payload))
+
+	var raw map[string]any
+	if err := json.Unmarshal(payload, &raw); err != nil {
+		c.log.TRACE.Printf("ignoring non-JSON payload on %s: %v", sn, err)
+		return
+	}
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	dev := c.state[sn]
+	if dev == nil {
+		dev = make(map[string]any)
+		c.state[sn] = dev
+	}
+
+	var merged bool
+	if p, ok := raw["params"].(map[string]any); ok {
+		maps.Copy(dev, p)
+		merged = true
+	}
+	if p, ok := raw["param"].(map[string]any); ok {
+		maps.Copy(dev, p)
+		merged = true
+	}
+
+	if !merged {
+		// flat payload: treat top-level keys as parameters
+		for k, v := range raw {
+			switch k {
+			case "id", "version", "timestamp", "typeCode", "cmdFunc", "cmdId":
+				// skip message envelope fields
+				continue
+			}
+			dev[k] = v
+		}
+	}
+}
+
+// Lookup returns the most recently cached MQTT value for the given serial and
+// parameter name. The boolean is false if either the device has not been seen
+// yet or the key has not been reported.
+func (c *ecoflowMqttClient) Lookup(sn, key string) (any, bool) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	dev, ok := c.state[sn]
+	if !ok {
+		return nil, false
+	}
+	v, ok := dev[key]
+	return v, ok
+}