Add Marstek MQTT responder to answer CT002/CT003 polls locally

CHANGELOG.md

@@ -1,6 +1,7 @@
 # Changelog
 
 ## Next
+- Added Marstek MQTT responder inside MQTT Insights: when `[MARSTEK]` credentials are configured, AstraMeter answers the Marstek CT002/CT003 poll protocol on the MQTT broker using the managed cloud MAC, so combined with [hame-relay](https://github.com/tomquist/hame-relay) the emulator's readings appear as a CT002/CT003 in the Marstek mobile app. Enabled by default; set `MARSTEK_MQTT_ENABLED = false` in `[MQTT_INSIGHTS]` to opt out.
 - Added opt-in web-based configuration editor (`WEB_CONFIG_ENABLED = True` in `[GENERAL]`) accessible at `http://<host>:52500/config`; supports editing all config sections and keys with type-appropriate inputs, comment preservation, and a Save & Restart button
 - **Breaking:** Rebrand project from "B2500 Meter" to "AstraMeter" (formerly b2500-meter). Package renamed to `astrameter`, CLI commands are now `astrameter` and `astra-sim`. Docker image moved from `ghcr.io/tomquist/b2500-meter` to `ghcr.io/tomquist/astrameter` (the legacy `ghcr.io/tomquist/b2500-meter` image is still published in parallel for backward compatibility). Home Assistant users must update their app repository URL to `https://github.com/tomquist/astrameter#main`.
 - Added CT002/CT003 emulation for steering multiple Marstek storage devices over the Marstek CT UDP protocol, with opt-in efficiency optimization that concentrates power on fewer batteries at low demand and rotates fairly over time (`MIN_EFFICIENT_POWER`, `EFFICIENCY_ROTATION_INTERVAL`, and related tuning options)

README.md

@@ -822,6 +822,9 @@ HA_DISCOVERY_PREFIX = homeassistant
 | `BASE_TOPIC` | `astrameter` | Root topic for all published messages |
 | `HA_DISCOVERY` | `true` | Enable Home Assistant MQTT Device Discovery |
 | `HA_DISCOVERY_PREFIX` | `homeassistant` | HA discovery topic prefix |
+| `MARSTEK_MQTT_ENABLED` | `true` | Respond to Marstek CT002/CT003 MQTT polls on this broker (requires `[MARSTEK]`) |
+
+**Marstek app visibility**: when `[MARSTEK]` credentials are configured, AstraMeter registers a managed fake CT device in the Marstek cloud. With `MARSTEK_MQTT_ENABLED=true` (default) it also answers the CT's MQTT polls on this broker. Combined with [hame-relay](https://github.com/tomquist/hame-relay) bridging the local broker to the Marstek cloud, the emulator's readings then show up as a CT002/CT003 in the Marstek mobile app. Set `MARSTEK_MQTT_ENABLED=false` to keep MQTT Insights but opt out of this responder.
 
 **Published entities** (per CT002 consumer):
 - Grid power (L1/L2/L3/total), charge target (L1/L2/L3), reported power, saturation

config.ini.example

@@ -365,3 +365,9 @@ THROTTLE_INTERVAL = 0
 #HA_DISCOVERY = true
 ## HA discovery prefix (default: homeassistant)
 #HA_DISCOVERY_PREFIX = homeassistant
+## Respond to Marstek CT002/CT003 MQTT polls on this broker (default: true).
+## Combined with hame-relay (https://github.com/tomquist/hame-relay) the
+## emulator's CT readings surface in the Marstek mobile app. Requires a
+## working [MARSTEK] section: the MAC used in the MQTT topics is the
+## managed fake device AstraMeter registers in the Marstek cloud.
+#MARSTEK_MQTT_ENABLED = true

src/astrameter/config/config_loader.py

@@ -619,5 +619,8 @@ def read_mqtt_insights_config(
                 addon_slug=(
                     config.get(section, "ADDON_SLUG", fallback="").strip() or None
                 ),
+                marstek_mqtt_enabled=config.getboolean(
+                    section, "MARSTEK_MQTT_ENABLED", fallback=True
+                ),
             )
     return None

src/astrameter/main.py

@@ -1,6 +1,7 @@
 import argparse
 import asyncio
 import configparser
+import contextlib
 import os
 import signal
 from collections import OrderedDict
@@ -17,7 +18,11 @@
     MarstekConfig,
     ensure_managed_fake_device,
 )
-from astrameter.mqtt_insights import MqttInsightsService
+from astrameter.mqtt_insights import (
+    MarstekMqttBinding,
+    MqttInsightsService,
+    normalize_mac,
+)
 from astrameter.powermeter import Powermeter
 from astrameter.shelly import Shelly
 from astrameter.version_info import get_git_commit_sha
@@ -73,6 +78,7 @@ async def run_device(
     powermeters: list[tuple[Powermeter, ClientFilter]],
     device_id: str | None = None,
     insights: MqttInsightsService | None = None,
+    marstek_mac: str = "",
 ):
     logger.debug(f"Starting device: {device_type}")
 
@@ -294,11 +300,51 @@ def _shelly_event_listener(dev_id, battery_ip, data):
             device_id or "", device.force_efficiency_rotation
         )
 
+    # Marstek MQTT responder — only wired up when Marstek credentials
+    # yielded a managed MAC (so hame-relay can route the replies back to
+    # the Marstek app) and the feature is enabled.
+    if isinstance(device, CT002) and insights and insights.marstek_mqtt_enabled:
+        if marstek_mac:
+
+            async def _marstek_get_values(
+                _pms: list[tuple[Powermeter, ClientFilter]] = powermeters,
+            ) -> list[float]:
+                chosen: Powermeter | None = next(
+                    (pm for pm, cf in _pms if cf.matches("0.0.0.0")), None
+                )
+                if chosen is None and _pms:
+                    chosen = _pms[0][0]
+                if chosen is None:
+                    return [0.0, 0.0, 0.0]
+                await chosen.wait_for_next_message()
+                vs = await chosen.get_powermeter_watts()
+                return [float(vs[i]) if i < len(vs) else 0.0 for i in range(3)]
+
+            await insights.register_marstek(
+                MarstekMqttBinding(
+                    device_id=device_id or "",
+                    ct_type=device.ct_type,
+                    mac=marstek_mac,
+                    get_values=_marstek_get_values,
+                    wifi_rssi=device.wifi_rssi,
+                )
+            )
+        else:
+            logger.info(
+                "Marstek MQTT responder not wired for %s: no managed MAC "
+                "available. Enable [MARSTEK] with MAILBOX/PASSWORD to use "
+                "this feature, or set MARSTEK_MQTT_ENABLED=false to silence "
+                "this notice.",
+                device_id,
+            )
+
     try:
         await device.wait()
     finally:
         if insights and isinstance(device, CT002):
             insights.unregister_handlers(device_id or "")
+            with contextlib.suppress(Exception):
+                await insights.unregister_marstek(device_id or "")
         try:
             await device.stop()
         except Exception:
@@ -311,7 +357,9 @@ async def async_main(
     device_types: list[str],
     device_ids: list[str],
     skip_test: bool,
+    managed_macs: dict[str, str] | None = None,
 ):
+    managed_macs = managed_macs or {}
     web_server = None
     if cfg.getboolean("GENERAL", "ENABLE_WEB_SERVER", fallback=True):
         logger.info("Starting web server...")
@@ -364,7 +412,15 @@ async def async_main(
 
         await asyncio.gather(
             *(
-                run_device(device_type, cfg, args, powermeters, device_id, insights)
+                run_device(
+                    device_type,
+                    cfg,
+                    args,
+                    powermeters,
+                    device_id,
+                    insights,
+                    managed_macs.get(device_type, ""),
+                )
                 for device_type, device_id in zip(
                     device_types, device_ids, strict=False
                 )
@@ -521,7 +577,11 @@ def main():
 
     _apply_cli_overrides(cfg, args)
 
-    # Optional Marstek cloud registration for managed fake CT devices (sync, before event loop)
+    # Optional Marstek cloud registration for managed fake CT devices (sync, before event loop).
+    # When registration succeeds, the returned MAC is captured per device
+    # type so the Marstek MQTT responder in MQTT Insights uses the same
+    # MAC that hame-relay will route back to the Marstek app.
+    managed_macs: dict[str, str] = {}
     marstek_enabled = cfg.getboolean("MARSTEK", "ENABLE", fallback=False)
     if marstek_enabled:
         mailbox = cfg.get("MARSTEK", "MAILBOX", fallback="")
@@ -545,7 +605,11 @@ def main():
                 for dt in ("ct002", "ct003"):
                     if dt in device_types:
                         any_ct = True
-                        ensure_managed_fake_device(marstek_cfg, dt)
+                        created = ensure_managed_fake_device(marstek_cfg, dt)
+                        if created is not None:
+                            normalized = normalize_mac(str(created.get("mac", "")))
+                            if normalized:
+                                managed_macs[dt] = normalized
                 if any_ct:
                     logger.info(
                         "Managed fake CT registration completed. Fake CT devices appear as offline in the Marstek app CT list (this is expected)."
@@ -593,7 +657,9 @@ def _restart_handler(signum, frame):
     while True:
         restart_requested = False
         try:
-            asyncio.run(async_main(cfg, args, device_types, device_ids, skip_test))
+            asyncio.run(
+                async_main(cfg, args, device_types, device_ids, skip_test, managed_macs)
+            )
             break  # clean exit
         except KeyboardInterrupt:
             if not restart_requested:

src/astrameter/mqtt_insights/__init__.py

@@ -1,5 +1,11 @@
 """MQTT Insights — publish internal state to MQTT with HA Discovery."""
 
+from .marstek_mqtt import MarstekMqttBinding, normalize_mac
 from .service import MqttInsightsConfig, MqttInsightsService
 
-__all__ = ["MqttInsightsConfig", "MqttInsightsService"]
+__all__ = [
+    "MarstekMqttBinding",
+    "MqttInsightsConfig",
+    "MqttInsightsService",
+    "normalize_mac",
+]

src/astrameter/mqtt_insights/marstek_mqtt.py

@@ -0,0 +1,107 @@
+"""Marstek MQTT responder — answer CT002/CT003 poll requests on the local broker.
+
+Pure helpers (topic formatting, payload building, poll detection) plus a
+binding dataclass that the :class:`MqttInsightsService` stores per device.
+
+The wire protocol (topics + CSV key=value payload) matches what the real
+Marstek CT002 uses against the Marstek cloud broker. When hame-relay is
+bridging the local broker to the cloud, responses produced here reach the
+Marstek mobile app as if they came from a real CT002.
+"""
+
+from __future__ import annotations
+
+import re
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+
+APP_TOPIC_TEMPLATES = (
+    "hame_energy/{ct_type}/App/{mac}/ctrl",
+    "marstek_energy/{ct_type}/App/{mac}/ctrl",
+)
+DEVICE_TOPIC_TEMPLATES = (
+    "hame_energy/{ct_type}/device/{mac}/ctrl",
+    "marstek_energy/{ct_type}/device/{mac}/ctrl",
+)
+
+# Matches observed real-device values; included so hm2mqtt-style parsers
+# recognise the message as a well-formed runtime-info frame.
+DEFAULT_VER_V = 148
+
+_APP_TOPIC_RE = re.compile(
+    r"^(?:hame|marstek)_energy/(?P<ct_type>[^/]+)/App/(?P<mac>[^/]+)/ctrl$"
+)
+_MAC_HEX_RE = re.compile(r"^[0-9a-f]{12}$")
+
+
+@dataclass(frozen=True)
+class MarstekMqttBinding:
+    """Per-device registration used by the MQTT Insights service."""
+
+    device_id: str
+    ct_type: str
+    mac: str
+    get_values: Callable[[], Awaitable[list[float]]]
+    wifi_rssi: int
+    ver_v: int = DEFAULT_VER_V
+
+
+def normalize_mac(raw: str) -> str:
+    """Lowercase, strip ``:``/``-``; return ``""`` if not 12 hex chars."""
+    if not raw:
+        return ""
+    cleaned = raw.replace(":", "").replace("-", "").strip().lower()
+    return cleaned if _MAC_HEX_RE.fullmatch(cleaned) else ""
+
+
+def is_poll_payload(body: bytes) -> bool:
+    """Return True iff *body* is a CSV containing ``cd=1``."""
+    if not body:
+        return False
+    try:
+        text = body.decode("utf-8")
+    except UnicodeDecodeError:
+        return False
+    for chunk in text.split(","):
+        key, sep, value = chunk.partition("=")
+        if not sep:
+            continue
+        if key.strip().lower() == "cd" and value.strip() == "1":
+            return True
+    return False
+
+
+def parse_app_topic(topic: str) -> tuple[str, str] | None:
+    """Return ``(ct_type, mac)`` for a Marstek App topic, else ``None``."""
+    match = _APP_TOPIC_RE.match(topic)
+    if not match:
+        return None
+    return match.group("ct_type"), match.group("mac").lower()
+
+
+def app_topics_for(binding: MarstekMqttBinding) -> tuple[str, str]:
+    old, new = APP_TOPIC_TEMPLATES
+    return (
+        old.format(ct_type=binding.ct_type, mac=binding.mac),
+        new.format(ct_type=binding.ct_type, mac=binding.mac),
+    )
+
+
+def device_topics_for(binding: MarstekMqttBinding) -> tuple[str, str]:
+    old, new = DEVICE_TOPIC_TEMPLATES
+    return (
+        old.format(ct_type=binding.ct_type, mac=binding.mac),
+        new.format(ct_type=binding.ct_type, mac=binding.mac),
+    )
+
+
+def build_response(binding: MarstekMqttBinding, watts: list[float]) -> bytes:
+    """Build the CSV ``k=v`` response body for a runtime-info frame."""
+    vs = list(watts) + [0.0] * (3 - len(watts))
+    a, b, c = (round(v) for v in vs[:3])
+    total = a + b + c
+    payload = (
+        f"pwr_a={a},pwr_b={b},pwr_c={c},pwr_t={total},"
+        f"wif_r={binding.wifi_rssi},ver_v={binding.ver_v},wif_s=2"
+    )
+    return payload.encode("utf-8")