Optional measured provider feed for parked Hub displays

apps/predbat/components.py

@@ -728,6 +728,8 @@ def load_component_class(component_info):
             "mqtt_token": {"required": True, "secret": True, "config": "gateway_mqtt_token"},
             "gateway_inverter_serial": {"required": False, "config": "gateway_inverter_serial", "default": None},
             "gateway_display_only": {"required": False, "config": "gateway_display_only", "default": False},
+            "gateway_live_display": {"required": False, "config": "gateway_live_display", "default": False},
+            "gateway_live_display_binding": {"required": False, "config": "gateway_live_display_binding", "default": None},
             "gateway_evc_automatic": {"required": False, "config": "gateway_evc_automatic", "default": False},
             "gateway_evc_control": {"required": False, "config": "gateway_evc_control", "default": False},
             "gateway_shared_ct": {"required": False, "config": "gateway_shared_ct", "default": True},

apps/predbat/gateway.py

@@ -11,6 +11,7 @@
 import json
 import math
 import os
+import re
 import ssl
 import threading
 import time
@@ -22,6 +23,7 @@
 
 from component_base import ComponentBase
 from control_handoff import control_gate_for, gated_mqtt_publish
+from live_display import live_display_payload, validated_binding
 
 try:
     import gateway_status_pb2 as pb
@@ -287,6 +289,11 @@ def initialize(
             **kwargs: Additional keyword arguments (ignored).
         """
         self.gateway_device_id = gateway_device_id
+        self.gateway_live_display = kwargs.get("gateway_live_display", False) is True
+        self.gateway_live_display_binding = kwargs.get("gateway_live_display_binding")
+        self._last_live_display_publish = 0
+        self._last_live_binding = None
+        self._last_live_binding_publish = 0
         if isinstance(gateway_display_only, str):
             value = gateway_display_only.strip().lower()
             if value not in ("true", "false", "1", "0", "yes", "no", "on", "off"):
@@ -688,8 +695,10 @@ async def run(self, seconds, first):
             if self.gateway_display_only:
                 # The JSON display topic is separate from the executable schedule. Never
                 # refresh schedules, release read-only, reset inverters or control EVs here.
-                if self._mqtt_connected and self._last_status is not None:
-                    await self._publish_predbat_data()
+                if self._mqtt_connected:
+                    await self._publish_display_binding()
+                    if self._last_status is not None:
+                        await self._publish_predbat_data()
                 await self._check_token_refresh()
                 self._update_display_health()
                 return True
@@ -762,6 +771,7 @@ async def _mqtt_loop(self):
                     self._first_connection_attempted = True
                     backoff = 5  # Reset backoff on successful connection
                     self.log(f"Info: GatewayMQTT: Connected to {self.mqtt_host}:{self.mqtt_port}")
+                    await self._publish_display_binding(force=True)
 
                     # Subscribe to status and LWT topics
                     await client.subscribe(self.topic_status, qos=1)
@@ -2369,16 +2379,38 @@ async def _publish_predbat_data(self):
                 "export_rate": (rate_anchors or {}).get("export_rate"),
             }
 
-            # Only publish if data changed
-            if payload == self._last_predbat_data:
+            live_enabled = getattr(self, "gateway_display_only", False) and getattr(self, "gateway_live_display", False)
+            if live_enabled:
+                payload["live_display"] = live_display_payload(self.base, self.gateway_device_id, self.gateway_inverter_serial, int(time.time()))
+                if not self._live_display_binding()["enabled"]:
+                    payload["live_display"] = {"v": 1, "enabled": True, "device_id": self.gateway_device_id, "available": False}
+                # Bound existing optional strings/matrices too: predbat_data's
+                # MQTT decoder receives a single chunk, not a reassembled blob.
+                payload["predbat_status"] = str(payload["predbat_status"])[:31]
+                payload["predbat_status_detail"] = str(payload["predbat_status_detail"])[:63]
+                payload["block_state"] = str(payload["block_state"])[:15]
+
+            # Live readings need a heartbeat even at genuine zero/unchanged SOC.
+            if payload == self._last_predbat_data and (not live_enabled or time.time() - self._last_live_display_publish < 60):
                 return
 
-            self._last_predbat_data = dict(payload)  # store copy without timestamp so dedup works next cycle
+            dedup_payload = dict(payload)
             payload["timestamp"] = int(time.time())
-            payload_json = json.dumps(payload)
             topic = f"{self._topic_base}/predbat_data"
+            payload_json = json.dumps(payload, separators=(",", ":")) if live_enabled else json.dumps(payload)
+            # Include UTF-8 topic bytes and a conservative MQTT header budget.
+            if live_enabled and len(payload_json.encode()) + len(topic.encode()) + 16 > 1400:
+                payload["marginal_costs"] = []
+                payload["marginal_time_labels"] = []
+                payload_json = json.dumps(payload, separators=(",", ":"))
+                if len(payload_json.encode()) + len(topic.encode()) + 16 > 1400:
+                    self.log("Warn: Live Hub display packet exceeds 1400-byte limit; not published")
+                    return
 
             await self._publish_raw(topic, payload_json.encode(), retain=True)
+            self._last_predbat_data = dedup_payload  # store after the publish call completes
+            if live_enabled:
+                self._last_live_display_publish = time.time()
             self.log(f"Info: Published predbat_data: price={payload['current_price']}p avg={payload['avg_price']}p cost=£{payload['total_cost']}")
 
         except Exception as e:
@@ -2595,6 +2627,41 @@ async def publish_command(self, command, command_id=None, **kwargs):
         else:
             self.log(f"Warn: GatewayMQTT: Not connected — cannot publish command: {command}")
 
+    def _live_display_binding(self):
+        """Validate only the server-issued binding for this selected display role."""
+        return validated_binding(getattr(self, "gateway_live_display_binding", None), self.gateway_device_id, self.gateway_inverter_serial, getattr(self, "gateway_live_display", False))
+
+    async def _publish_display_binding(self, force=False):
+        """Publish whitelist-only display metadata, never general config or control.
+
+        The existing raw publisher remains restricted to predbat_data in display
+        mode. This private path constructs its topic and every payload field,
+        validates the read-only display role, and cannot carry bus/control config.
+        """
+        if self.gateway_display_only is not True or not self._mqtt_connected or self._mqtt_client is None:
+            return
+        if not re.fullmatch(r"pbgw_[a-f0-9]{12}", self.gateway_device_id) or self._topic_base != f"predbat/devices/{self.gateway_device_id}":
+            return
+        binding = self._live_display_binding()
+        now = time.time()
+        if not force and binding == self._last_live_binding and now - self._last_live_binding_publish < 60:
+            return
+        topic = f"{self._topic_base}/config"
+        payload = json.dumps({"live_display": binding}, separators=(",", ":")).encode()
+        if len(topic.encode()) + len(payload) + 16 > 1400:
+            return
+        try:
+            if self._loop is not None and self._loop is not asyncio.get_running_loop():
+                future = asyncio.run_coroutine_threadsafe(self._mqtt_client.publish(topic, payload, qos=1, retain=False), self._loop)
+                await asyncio.wrap_future(future)
+            else:
+                await self._mqtt_client.publish(topic, payload, qos=1, retain=False)
+        except Exception as error:
+            self.log(f"Warn: GatewayMQTT: Display binding publish failed: {error}")
+            return
+        self._last_live_binding = binding
+        self._last_live_binding_publish = now
+
     async def _publish_raw(self, topic, payload, retain=False):
         """Publish raw bytes to an MQTT topic.
 

apps/predbat/gecloud.py

@@ -613,6 +613,7 @@ def initialize(self, ge_cloud_direct, api_key, automatic, automatic_evc=False, e
         self.register_list = {}
         self.settings = {}
         self.status = {}
+        self.live_display_samples = {}  # unmerged measured samples, never sensor defaults
         self.meter = {}
         self.info = {}
         self.register_entity_map = {}
@@ -3240,6 +3241,12 @@ async def async_get_inverter_status(self, serial, previous=None):
         if previous is None:
             previous = {}
         result = await self.async_get_inverter_data_retry(GE_API_INVERTER_STATUS, serial)
+        # Display provenance must retain neither missing fields nor an old
+        # field under a new timestamp. The ordinary merged control cache stays
+        # unchanged; the optional display adapter consumes only this raw sample.
+        if not hasattr(self, "live_display_samples"):
+            self.live_display_samples = {}
+        self.live_display_samples[serial.lower()] = result
         if result is None:
             return previous
         return merge_non_null(result, previous)

apps/predbat/live_display.py

@@ -0,0 +1,143 @@
+"""Provider adapters for a bounded, measured-only optional Hub display feed."""
+
+import datetime
+import hashlib
+import math
+import re
+
+MAX_AGE = 300
+MAX_POWER = 1000000
+
+
+def selected_serials(value):
+    """Require an explicit, unique physical inverter set instead of all devices."""
+    values = [value] if isinstance(value, str) else value
+    if not isinstance(values, (tuple, list)) or not 1 <= len(values) <= 16:
+        raise ValueError("selected inverter set missing")
+    if any(not isinstance(item, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,32}", item) for item in values):
+        raise ValueError("invalid selected inverter identity")
+    serials = sorted(item.upper() for item in values)
+    if len(set(serials)) != len(serials):
+        raise ValueError("duplicate selected inverter identity")
+    return serials
+
+
+def source_identity(provider, serials):
+    """Bind the provider-independent wire feed to the selected physical set."""
+    return hashlib.sha256((provider + "\n" + "\n".join(selected_serials(serials))).encode()).hexdigest()
+
+
+def validated_binding(value, device_id, serials, enabled):
+    """Whitelist server-issued display metadata or explicitly revoke prior opt-in."""
+    disabled = {"enabled": False}
+    try:
+        if enabled is not True or not isinstance(value, dict) or set(value) != {"enabled", "device_id", "provider", "source_id"}:
+            return disabled
+        if not isinstance(device_id, str) or not re.fullmatch(r"pbgw_[a-f0-9]{12}", device_id):
+            return disabled
+        expected = {"enabled": True, "device_id": device_id, "provider": "givenergy", "source_id": source_identity("givenergy", serials)}
+        if value.get("enabled") is not True or value != expected:
+            return disabled
+        return expected
+    except (TypeError, ValueError):
+        return disabled
+
+
+def number(value, minimum, maximum):
+    """Accept real finite measurements, including zero, without numeric coercion."""
+    if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or not minimum <= value <= maximum:
+        raise ValueError("missing or invalid measurement")
+    return value
+
+
+def gecloud_readings(component, serials, now):
+    """Adapt complete, unmerged GE samples with explicit shared-CT aggregation.
+
+    EMS/Gateway compositions and extra PV devices remain unavailable until their
+    measurement provenance is proven. Multiple direct batteries need one known
+    common grid meter, so grid/load are counted once. Battery/PV power is summed,
+    and actual SoC is weighted by each battery's reported nominal Wh capacity.
+    """
+    devices = component.devices_dict
+    if not isinstance(devices, dict) or devices.get("ems") or devices.get("gateway") or devices.get("pv"):
+        raise ValueError("unsupported site composition")
+    if selected_serials(devices.get("battery")) != serials:
+        raise ValueError("selected inverter set differs from provider site")
+    site_ids = devices.get("site_ids", {})
+    sites = {site_ids.get(serial.lower()) for serial in serials}
+    if None in sites or len(sites) != 1:
+        raise ValueError("ambiguous provider site")
+    if len(serials) > 1:
+        meters = [devices.get("battery_meters", {}).get(serial.lower()) for serial in serials]
+        if any(not isinstance(meter, list) or len(meter) != 1 or meter[0] is None for meter in meters) or any(meter != meters[0] for meter in meters):
+            raise ValueError("ambiguous grid meter aggregation")
+    samples = component.live_display_samples
+    rows = []
+    for serial in serials:
+        raw = samples.get(serial.lower())
+        if not isinstance(raw, dict):
+            raise ValueError("no complete raw provider sample")
+        stamp = datetime.datetime.fromisoformat(raw["time"].replace("Z", "+00:00"))
+        if stamp.tzinfo is None:
+            raise ValueError("source timestamp has no timezone")
+        observed = int(stamp.timestamp())
+        if observed < now - MAX_AGE or observed > now + 30:
+            raise ValueError("source measurement is stale or future dated")
+        battery = raw["battery"]
+        info = component.info[serial.lower()]["info"]["battery"]
+        capacity = number(info["nominal_capacity"], 0.001, 100000) * number(info["nominal_voltage"], 0.001, 10000)
+        number(capacity, 1, 1000000)
+        rows.append(
+            {
+                "observed": observed,
+                "soc": number(battery["percent"], 0, 100),
+                "capacity": capacity,
+                # GE's raw battery sign is +discharge, firmware is +charge.
+                "battery": -number(battery["power"], -MAX_POWER, MAX_POWER),
+                "solar": number(raw["solar"]["power"], 0, MAX_POWER),
+                "grid": number(raw["grid"]["power"], -MAX_POWER, MAX_POWER),
+                "load": number(raw["consumption"], 0, MAX_POWER),
+            }
+        )
+    if max(row["observed"] for row in rows) - min(row["observed"] for row in rows) > 60:
+        raise ValueError("site samples are not contemporaneous")
+    capacity = sum(row["capacity"] for row in rows)
+    number(capacity, 1, 1000000)
+    result = {
+        "observed_at": min(row["observed"] for row in rows),
+        "soc": round(sum(row["soc"] * row["capacity"] for row in rows) / capacity),
+        "capacity_wh": round(capacity),
+        "grid_w": round(rows[0]["grid"]),
+        "load_w": round(rows[0]["load"]),
+        "battery_w": round(sum(row["battery"] for row in rows)),
+        "solar_w": round(sum(row["solar"] for row in rows)),
+    }
+    for key in ("battery_w", "solar_w", "grid_w", "load_w"):
+        number(result[key], -MAX_POWER if key in ("battery_w", "grid_w") else 0, MAX_POWER)
+    return result
+
+
+def live_display_payload(base, device_id, configured_serials, now):
+    """Return unavailable unless one supported, selected provider proves readings."""
+    result = {"v": 1, "enabled": True, "device_id": device_id, "available": False}
+    try:
+        if not isinstance(device_id, str) or not re.fullmatch(r"pbgw_[a-f0-9]{12}", device_id):
+            raise ValueError("invalid paired Hub identity")
+        from components import COMPONENT_LIST
+
+        components = base.components.components
+        sources = [(name, component) for name, component in components.items() if component and COMPONENT_LIST.get(name, {}).get("inverter") and getattr(component, "gateway_display_only", False) is not True]
+        if len(sources) != 1 or sources[0][0] != "gecloud":
+            raise ValueError("unsupported or ambiguous selected provider")
+        provider = "givenergy"
+        serials = selected_serials(configured_serials)
+        result.update(provider=provider, source_id=source_identity(provider, serials))
+        gate = getattr(base, "__dict__", {}).get("control_write_gate")
+        if gate is not None and gate.provider != provider:
+            raise ValueError("selected controller provider differs")
+        result.update(gecloud_readings(sources[0][1], serials, now), available=True)
+    except (AttributeError, KeyError, TypeError, ValueError, OverflowError):
+        # No partial readings or made-up zero. Firmware clears any previous live
+        # values on this message, and also expires them without new messages.
+        pass
+    return result

apps/predbat/tests/test_gateway.py

@@ -7405,6 +7405,17 @@ def run_gateway_tests(my_predbat=None):
     """Run all GatewayMQTT tests. Returns True on failure, False on success."""
     from tests.test_gateway_token_refresh import TestIsAuthFailure, TestApplyRefreshResponse, TestMaybeRefreshOnAuthError
     from tests.test_gateway_display import TestGatewayDisplayOnly
+    from tests.test_live_display import run_live_display_tests
+
+    try:
+        run_live_display_tests()
+        print("  OK: measured live display provider/publisher contract")
+    except Exception as error:
+        import traceback
+
+        print(f"  FAIL: measured live display contract: {error}")
+        traceback.print_exc()
+        return True
 
     test_classes = [
         TestGatewayDisplayOnly,