diff --git a/apps/predbat/components.py b/apps/predbat/components.py index 061535ad7..80e7fbcb9 100644 --- a/apps/predbat/components.py +++ b/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}, diff --git a/apps/predbat/gateway.py b/apps/predbat/gateway.py index a7b832cbc..88a6de5a4 100644 --- a/apps/predbat/gateway.py +++ b/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. diff --git a/apps/predbat/gecloud.py b/apps/predbat/gecloud.py index 7ba1d008b..89a2ea06e 100644 --- a/apps/predbat/gecloud.py +++ b/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) diff --git a/apps/predbat/live_display.py b/apps/predbat/live_display.py new file mode 100644 index 000000000..2f6743834 --- /dev/null +++ b/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 diff --git a/apps/predbat/tests/test_gateway.py b/apps/predbat/tests/test_gateway.py index c465d72a7..a74286690 100644 --- a/apps/predbat/tests/test_gateway.py +++ b/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, diff --git a/apps/predbat/tests/test_live_display.py b/apps/predbat/tests/test_live_display.py new file mode 100644 index 000000000..4fb0caad6 --- /dev/null +++ b/apps/predbat/tests/test_live_display.py @@ -0,0 +1,258 @@ +"""Measured provider provenance and real optional MQTT publisher regressions.""" + +import asyncio +import copy +import datetime +import json +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +from gecloud import GECloudDirect +from live_display import gecloud_readings, live_display_payload, selected_serials, source_identity, validated_binding +from tests.test_gateway import TestPublishPredbatData as _PublisherFixture + +NOW = 1800000000 +DEVICE = "pbgw_0123456789ab" + + +def provider(serials=("INV001",), soc=0): + """Make complete synthetic API observations, including valid idle/empty data.""" + stamp = datetime.datetime.fromtimestamp(NOW, datetime.timezone.utc).isoformat() + raw = {"time": stamp, "battery": {"percent": soc, "power": 0}, "solar": {"power": 0}, "grid": {"power": 0}, "consumption": 0} + return SimpleNamespace( + devices_dict={"battery": list(serials), "site_ids": {s.lower(): 7 for s in serials}, "battery_meters": {s.lower(): ["CT001"] for s in serials}}, + live_display_samples={s.lower(): copy.deepcopy(raw) for s in serials}, + info={s.lower(): {"info": {"battery": {"nominal_capacity": 100, "nominal_voltage": 50}}} for s in serials}, + ) + + +def base_for(component, name="gecloud"): + """Wrap a provider in the actual component registry shape.""" + return SimpleNamespace(components=SimpleNamespace(components={name: component})) + + +def test_zero_is_a_complete_actual_reading(): + """Empty battery and idle powers are valid when the source is proven.""" + payload = live_display_payload(base_for(provider()), DEVICE, "INV001", NOW) + assert payload["available"] and payload["soc"] == 0 + assert all(payload[key] == 0 for key in ("battery_w", "solar_w", "grid_w", "load_w")) + assert payload["source_id"] == source_identity("givenergy", ["inv001"]) + + +def test_incomplete_or_invalid_raw_sample_is_unavailable(): + """Sensor defaults and merged historical fields cannot become actuals.""" + for field, value in [("battery", None), ("time", None), ("consumption", None), ("consumption", True), ("consumption", float("nan"))]: + component = provider() + component.live_display_samples["inv001"][field] = value + result = live_display_payload(base_for(component), DEVICE, "INV001", NOW) + assert not result["available"] and "soc" not in result + + +def test_source_time_is_bounded(): + """A newly received stale/future sample is unavailable.""" + for offset in (-301, 31): + component = provider() + component.live_display_samples["inv001"]["time"] = datetime.datetime.fromtimestamp(NOW + offset, datetime.timezone.utc).isoformat() + assert not live_display_payload(base_for(component), DEVICE, "INV001", NOW)["available"] + + +def test_selected_identity_and_topology_must_be_proven(): + """Reject account-wide, unsupported and ambiguous provider selections.""" + component = provider() + for selected in (None, [], ["INV001", "inv001"], "OTHER"): + assert not live_display_payload(base_for(component), DEVICE, selected, NOW)["available"] + for field in ("gateway", "ems", "pv"): + unsupported = provider() + unsupported.devices_dict[field] = ["OTHER"] + assert not live_display_payload(base_for(unsupported), DEVICE, "INV001", NOW)["available"] + assert not live_display_payload(base_for(component, "fox"), DEVICE, "INV001", NOW)["available"] + base = base_for(component) + base.components.components["gateway"] = SimpleNamespace(gateway_display_only=False) + assert not live_display_payload(base, DEVICE, "INV001", NOW)["available"] + base = base_for(component) + base.control_write_gate = SimpleNamespace(provider="foxess") + assert not live_display_payload(base, DEVICE, "INV001", NOW)["available"] + + +def test_explicit_shared_ct_aggregation(): + """Count common site load/grid once, sum battery/PV, capacity-weight actual SoC.""" + component = provider(("INV001", "INV002"), soc=20) + component.info["inv002"]["info"]["battery"]["nominal_capacity"] = 200 + component.live_display_samples["inv002"]["battery"] = {"percent": 80, "power": 300} + for raw in component.live_display_samples.values(): + raw["consumption"] = 400 + raw["grid"]["power"] = -100 + raw["solar"]["power"] = 50 + result = gecloud_readings(component, selected_serials(["INV002", "INV001"]), NOW) + assert result == {"observed_at": NOW, "soc": 60, "capacity_wh": 15000, "grid_w": -100, "load_w": 400, "battery_w": -300, "solar_w": 100} + component.devices_dict["battery_meters"]["inv002"] = ["OTHER"] + assert not live_display_payload(base_for(component), DEVICE, ["INV001", "INV002"], NOW)["available"] + + +def test_raw_capture_does_not_inherit_old_fields(): + """Exercise the actual fetch method: normal control merge stays, display rejects it.""" + component = provider() + raw = copy.deepcopy(component.live_display_samples["inv001"]) + raw["battery"]["percent"] = None + component.async_get_inverter_data_retry = AsyncMock(return_value=raw) + merged = asyncio.run(GECloudDirect.async_get_inverter_status(component, "INV001", {"battery": {"percent": 55}})) + assert merged["battery"]["percent"] == 55 + assert component.live_display_samples["inv001"]["battery"]["percent"] is None + assert not live_display_payload(base_for(component), DEVICE, "INV001", NOW)["available"] + component.async_get_inverter_data_retry.return_value = None + asyncio.run(GECloudDirect.async_get_inverter_status(component, "INV001", merged)) + assert component.live_display_samples["inv001"] is None + + +def published_gateway(enabled=True, parked=True): + """Use the actual publisher with its existing isolated state reader fixture.""" + gw = _PublisherFixture()._make_gateway() + gw.gateway_device_id = DEVICE + gw.gateway_inverter_serial = "INV001" + gw.gateway_display_only = parked + gw.gateway_live_display = enabled + gw.gateway_live_display_binding = {"enabled": True, "device_id": DEVICE, "provider": "givenergy", "source_id": source_identity("givenergy", ["INV001"])} + gw._last_live_display_publish = 0 + gw.base = base_for(provider()) + gw._topic_base = "predbat/devices/" + DEVICE + return gw + + +def test_server_binding_validation_and_metadata_publisher(): + """Only exact whitelist metadata crosses the dedicated read-only path.""" + from gateway import GatewayMQTT + + gw = published_gateway() + binding = gw.gateway_live_display_binding + assert validated_binding(binding, DEVICE, "INV001", True) == binding + assert source_identity("givenergy", ["CE2119G197", "CE2105G024"]) == "3915601d8c60cbe0bb8768f509be81c928e8f5df16cc42dff14efdf0f4bce2f6" + for bad in (None, {**binding, "source_id": "a" * 64}, {**binding, "device_id": "pbgw_aaaaaaaaaaaa"}, {**binding, "provider": "foxess"}, {**binding, "poll_interval_ms": 1}, {**binding, "enabled": 1}): + assert validated_binding(bad, DEVICE, "INV001", True) == {"enabled": False} + gw._mqtt_connected = True + gw._mqtt_client = SimpleNamespace(publish=AsyncMock()) + gw._loop = None + gw._last_live_binding = None + gw._last_live_binding_publish = 0 + with patch("gateway.time.time", return_value=NOW): + asyncio.run(gw._publish_display_binding()) + asyncio.run(gw._publish_display_binding()) + assert gw._mqtt_client.publish.await_count == 1 + asyncio.run(gw._publish_display_binding(force=True)) # reconnect bypasses dedup + assert gw._mqtt_client.publish.await_count == 2 + topic, data = gw._mqtt_client.publish.await_args.args + assert topic == "predbat/devices/" + DEVICE + "/config" + assert json.loads(data) == {"live_display": binding} + assert gw._mqtt_client.publish.await_args.kwargs == {"qos": 1, "retain": False} + with patch("gateway.time.time", return_value=NOW + 61): + asyncio.run(gw._publish_display_binding()) + assert gw._mqtt_client.publish.await_count == 3 + gw.gateway_live_display = False + asyncio.run(gw._publish_display_binding()) + assert json.loads(gw._mqtt_client.publish.await_args.args[1]) == {"live_display": {"enabled": False}} + gw.gateway_display_only = False + gw._mqtt_client.publish.reset_mock() + asyncio.run(gw._publish_display_binding(force=True)) + gw._mqtt_client.publish.assert_not_awaited() + gw.gateway_display_only = True + # General config/control remains prohibited through the original raw path. + asyncio.run(GatewayMQTT._publish_raw(gw, topic, b'{"poll_interval_ms":1}')) + gw._mqtt_client.publish.assert_not_awaited() + + +def test_invalid_binding_fails_closed_and_failed_publish_retries(): + """An unbound feed is unavailable and failed metadata sends are not deduplicated.""" + gw = published_gateway() + gw.gateway_live_display_binding = None + with patch("gateway.time.time", return_value=NOW): + asyncio.run(gw._publish_predbat_data()) + assert not json.loads(gw._published[0][1])["live_display"]["available"] + gw._mqtt_connected = True + gw._mqtt_client = SimpleNamespace(publish=AsyncMock(side_effect=RuntimeError("synthetic failure"))) + gw._loop = None + gw._last_live_binding = None + gw._last_live_binding_publish = 0 + asyncio.run(gw._publish_display_binding()) + assert gw._last_live_binding is None + gw._mqtt_client.publish.side_effect = None + asyncio.run(gw._publish_display_binding()) + assert gw._mqtt_client.publish.await_count == 2 + assert gw._last_live_binding == {"enabled": False} + + +def test_real_publisher_is_opt_in_parked_bounded_and_heartbeats(): + """Cover the real producer path and export its exact synthetic wire contract.""" + for enabled, parked in ((False, True), (True, False)): + gw = published_gateway(enabled, parked) + asyncio.run(gw._publish_predbat_data()) + assert "live_display" not in json.loads(gw._published[0][1]) + gw = published_gateway() + with patch("gateway.time.time", return_value=NOW): + asyncio.run(gw._publish_predbat_data()) + asyncio.run(gw._publish_predbat_data()) + assert len(gw._published) == 1 + with patch("gateway.time.time", return_value=NOW + 61): + asyncio.run(gw._publish_predbat_data()) + assert len(gw._published) == 2 + topic, data, retain = gw._published[0] + assert retain and len(topic.encode()) + len(data) + 16 <= 1400 + assert json.loads(data)["live_display"]["available"] + + +def write_wire_fixture(path, backend_path=None): + """Export actual publisher/config bytes, optionally starting from SaaS output.""" + gw = published_gateway() + if backend_path: + with open(backend_path, encoding="utf-8") as source: + config = json.load(source) + serials = config["gateway_inverter_serial"] + gw.base = base_for(provider(serials)) + gw.args = {} + gw.initialize( + gateway_device_id=config["gateway_device_id"], + mqtt_host=config["gateway_mqtt_host"], + gateway_display_only=config["gateway_display_only"], + gateway_inverter_serial=serials, + gateway_live_display=config["gateway_live_display"], + gateway_live_display_binding=config["gateway_live_display_binding"], + ) + serial = selected_serials(gw.gateway_inverter_serial)[0].lower() + sample = gw.base.components.components["gecloud"].live_display_samples[serial] + sample["battery"]["power"] = 321 + sample["grid"]["power"] = -456 + sample["solar"]["power"] = 78 + sample["consumption"] = 99 + gw._mqtt_connected = True + gw._mqtt_client = SimpleNamespace(publish=AsyncMock()) + gw._loop = None + gw._last_live_binding = None + gw._last_live_binding_publish = 0 + with patch("gateway.time.time", return_value=NOW): + asyncio.run(gw._publish_display_binding(force=True)) + asyncio.run(gw._publish_predbat_data()) + topic, data, _ = gw._published[0] + metadata_topic, metadata_payload = gw._mqtt_client.publish.await_args.args + assert metadata_topic == "predbat/devices/" + DEVICE + "/config" + assert not gw._mqtt_client.publish.await_args.kwargs["retain"] + config = json.loads(metadata_payload)["live_display"] + assert config["enabled"] + with open(path, "w", encoding="utf-8") as output: + json.dump({"topic": topic, "payload": data.decode(), "config": config}, output) + + +def run_live_display_tests(): + """Run the same contract assertions through the existing CLI test registry.""" + test_zero_is_a_complete_actual_reading() + test_incomplete_or_invalid_raw_sample_is_unavailable() + test_source_time_is_bounded() + test_selected_identity_and_topology_must_be_proven() + test_explicit_shared_ct_aggregation() + test_raw_capture_does_not_inherit_old_fields() + test_real_publisher_is_opt_in_parked_bounded_and_heartbeats() + test_server_binding_validation_and_metadata_publisher() + test_invalid_binding_fails_closed_and_failed_publish_retries() + + +if __name__ == "__main__": + import sys + + write_wire_fixture(sys.argv[1], sys.argv[2] if len(sys.argv) > 2 else None) diff --git a/docs/hub-live-display.md b/docs/hub-live-display.md new file mode 100644 index 000000000..363d0980e --- /dev/null +++ b/docs/hub-live-display.md @@ -0,0 +1,99 @@ +# Optional measured Hub display feed + +This is an incremental GivEnergy Cloud adapter for a provider-independent wire +contract. It does not complete SaaS preference wiring or add FoxESS support. +No deployment, enrollment or customer provider change is included. + +Producer opt-in requires both `gateway_display_only: true` and the new boolean +`gateway_live_display: true` (default false), plus `gateway_device_id` and an +explicit `gateway_inverter_serial` string/list. The component registry exposes +the flag, but SaaS must bind the saved preference and exact physical selection +to these arguments plus `gateway_live_display_binding`. The binding must have +exactly enabled/device_id/provider/source_id fields matching the paired Hub and +canonical selection. Missing, invalid or disabled bindings cannot produce usable +actuals. The original raw publisher remains restricted to `predbat_data`. + +A separate metadata-only publisher constructs a fixed device `/config` topic +and whitelist-only `{live_display: binding}` body. It requires strict display +role, valid paired device identity and matching topic base. It publishes on +reconnect and every 60 seconds, sending `{enabled:false}` when live opt-in is +absent/off/invalid. These sends are deliberately **non-retained**: replacing the +shared topic's retained full configuration would erase passive cutover intent, +mesh keys or other desired settings. Only this display metadata bypasses battery +write gating; no bus/control configuration is accepted and generic raw config +publication is still blocked. Failures retry without advancing dedup state. + +`predbat/devices//predbat_data` gains `live_display` only when opted +in. Version 1 contains `v: 1`, `enabled: true`, `device_id`, `provider`, `source_id`, +and `available`. Complete actuals add integer `observed_at` (UTC epoch seconds), +`soc` (0..100), `capacity_wh` (1..1000000), `battery_w` (+charge), `grid_w` +(+export), `solar_w` and `load_w`. Signed powers are bounded to +/-1000000 W; +solar/load are non-negative. Genuine zero is valid. Missing values are omitted, +never filled from sensors or plan `block_soc` forecasts. An unavailable message +clears the receiver's previous actuals. + +Source identity is SHA256 of UTF-8 `provider + "\n" + "\n".join(serials)`, where +serials are the sorted, unique uppercase selected physical inverter serials. +Selections contain 1..16 serials, each `[A-Za-z0-9_-]{1,32}`. Case-insensitive +duplicates are rejected. Current provider is `givenergy`. This binds selection, +not authority: firmware separately requires authorized device `/config` opt-in +with that exact provider/source/device binding. Existing instance and staff MQTT +principals can publish `/config`; it is not service-exclusive. Telemetry alone +cannot enable display or change the binding. Server-issued preference/selection +configuration is emitted by the separate SaaS follow-up; this branch validates +it before publication. If opting out removes the Gateway component altogether, the stopped instance +cannot send a final disable. Matching firmware treats valid non-retained metadata +as a **150-second monotonic lease**, renewed by this 60-second publisher. It +expires the external role independently of actual-data freshness after the last +enable received, applied on the next display-state tick. Telemetry never renews it; retained enable replay is rejected; +explicit disable is immediate when delivered. This preserves shared retained +configuration and ordinary passive/local UI without changing control state. +The customer save-to-engine-config-reload delay is separate, so this is not an +immediate save-time revocation guarantee. The paired SaaS lifecycle verifier +exercises real preference SQL save, config output, Components.start/stop, TLS MQTT +and firmware decoder/projection with deterministic lease time. + +Only one registered inverter provider may be active. Raw GE API samples are +captured before the existing null-preserving merge, so old fields cannot gain a +new timestamp. A sample must be complete, timezone stamped, at most 300 seconds +old, and no more than 30 seconds ahead. The adapter requires the exact GE battery +set at one known site. EMS, Gateway and extra PV compositions are unavailable. +Multiple direct batteries require one explicitly common meter identity; grid +and house load are counted once, battery/PV power summed, actual SoC capacity +weighted. Different or unknown meters, mismatched sites, sample skew above 60 +seconds, absent nominal capacity/voltage or unsupported providers are unavailable. +This deliberately leaves some real systems unsupported until topology semantics +are proven. It does not guess aggregation or use configured fallback capacity. + +Live display publication includes a 60-second heartbeat even when readings are +unchanged. The compact complete MQTT packet budget is 1400 bytes, including UTF-8 +topic and 16 bytes for protocol header/QoS overhead. Optional marginal matrices +are omitted if needed; an oversized packet is not published, so receiver expiry +applies. Ordinary local-controller publishing stays on its existing JSON path. + +Validation: + +```sh +# From coverage/, includes the new assertions through the existing registry: +python ../apps/predbat/unit_test.py --test gateway > /tmp/live-display-registry.log 2>&1 +# From repository root, export actual publisher bytes for firmware host proof: +PYTHONPATH=apps/predbat python apps/predbat/tests/test_live_display.py /tmp/live-display-wire.json +# Optionally supply actual synthetic SaaS-helper output as the second path: +PYTHONPATH=apps/predbat python apps/predbat/tests/test_live_display.py /tmp/live-display-wire.json /tmp/hub-live-display-backend-fixture.json +``` + +The fixture is wholly synthetic, with no network or customer credentials. The +firmware repository's `docs/hub-live-display.md` documents the matching decoder +and actual DisplayState projection proof. Impact reports for initialize, +`async_get_inverter_status`, `_publish_predbat_data`, and the gateway test runner +were LOW in the indexed reference, with the publisher called by Gateway.run. +The initial index was a reference checkout; source call sites were also +inspected. The final exact-worktree refreshed index maps the expected Gateway +run/reconnect/publication paths and GE sample capture. Its aggregate change scan +reports HIGH because Gateway.run participates in eight indexed execution flows, +despite LOW individual pre-edit impacts. Existing control publishing and read-only +handlers are unchanged, and the full gateway CLI regression group passes. This +aggregate warning requires review and is not a customer deployment clearance. +Affected paths are raw GE observation capture, optional display publication and +component argument registration. Controller handoff and boot attachment paths are +unchanged. diff --git a/mkdocs.yml b/mkdocs.yml index 1ab2829c2..a8881012b 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -34,6 +34,7 @@ nav: - random-scenarios.md - rest_api.md - plugins.md + - hub-live-display.md - 'Predheat': - predheat.md markdown_extensions: