feat(gateway): support Hub display alongside another inverter provider

apps/predbat/components.py

@@ -726,6 +726,7 @@ def load_component_class(component_info):
             "mqtt_port": {"required": False, "config": "gateway_mqtt_port", "default": 8883},
             "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_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},
@@ -1038,11 +1039,11 @@ def inverter_source_active(self):
         question rather than a per-inverter one: a component serves whichever inverters it
         discovered, and Inverter only needs to know whether to keep waiting.
         """
-        return any(COMPONENT_LIST.get(name, {}).get("inverter", False) and component for name, component in self.components.items())
+        return any(COMPONENT_LIST.get(name, {}).get("inverter", False) and component and getattr(component, "gateway_display_only", False) is not True for name, component in self.components.items())
 
     def inverter_source_names(self):
         """The display names of every active inverter component, for user-facing messages."""
-        return [COMPONENT_LIST[name]["name"] for name, component in self.components.items() if component and COMPONENT_LIST.get(name, {}).get("inverter", False)]
+        return [COMPONENT_LIST[name]["name"] for name, component in self.components.items() if component and COMPONENT_LIST.get(name, {}).get("inverter", False) and getattr(component, "gateway_display_only", False) is not True]
 
     def inverter_source_status(self):
         """
@@ -1075,6 +1076,8 @@ def inverter_source_status(self):
                 continue
             load_error = self.load_error(name)
             component = self.components.get(name, None)
+            if component and getattr(component, "gateway_display_only", False) is True:
+                continue
             if load_error is not None:
                 state = "failed to start: {}".format(load_error)
             elif not component:

apps/predbat/gateway.py

@@ -231,6 +231,7 @@ class GatewayMQTT(ComponentBase):
     gateway_shared_ct = True
     gateway_integrate_power = False
     gateway_inverter_serial = ()
+    gateway_display_only = False
     _integrated_energy = None
     _unavailable_inverters = ((), ())
     _discovery = {}  # replaced, never changed in place, so this default is not shared state
@@ -244,11 +245,26 @@ class GatewayMQTT(ComponentBase):
     _reserve_before_zero = {}  # replaced, never changed in place
     gateway_device_id = None
 
-    def initialize(self, gateway_device_id=None, mqtt_host=None, mqtt_port=8883, mqtt_token=None, gateway_inverter_serial=None, gateway_evc_automatic=False, gateway_evc_control=False, gateway_shared_ct=True, gateway_integrate_power=False, **kwargs):
+    def initialize(
+        self,
+        gateway_device_id=None,
+        mqtt_host=None,
+        mqtt_port=8883,
+        mqtt_token=None,
+        gateway_inverter_serial=None,
+        gateway_evc_automatic=False,
+        gateway_evc_control=False,
+        gateway_shared_ct=True,
+        gateway_integrate_power=False,
+        gateway_display_only=False,
+        **kwargs,
+    ):
         """Initialize gateway configuration and build MQTT topic strings.
 
         Args:
             gateway_device_id: The gateway's device ID (e.g. "pbgw_abc123").
+            gateway_display_only: Publish display data without registering or controlling any inverter
+                or EV charger. The hub must already be parked by its standby controller.
             mqtt_host: MQTT broker hostname.
             mqtt_port: MQTT broker port (default 8883 for TLS).
             mqtt_token: JWT access token for MQTT authentication.
@@ -270,6 +286,14 @@ def initialize(self, gateway_device_id=None, mqtt_host=None, mqtt_port=8883, mqt
             **kwargs: Additional keyword arguments (ignored).
         """
         self.gateway_device_id = gateway_device_id
+        if isinstance(gateway_display_only, str):
+            value = gateway_display_only.strip().lower()
+            if value not in ("true", "false", "1", "0", "yes", "no", "on", "off"):
+                raise ValueError("gateway_display_only must be a boolean")
+            gateway_display_only = value in ("true", "1", "yes", "on")
+        if not isinstance(gateway_display_only, bool):
+            raise ValueError("gateway_display_only must be a boolean")
+        self.gateway_display_only = gateway_display_only
         self.gateway_evc_automatic = bool(gateway_evc_automatic)
         self.gateway_evc_control = bool(gateway_evc_control)
         self.gateway_shared_ct = bool(gateway_shared_ct)
@@ -415,6 +439,8 @@ def _on_plan_executed(self, charge_windows=None, charge_limits=None, export_wind
         the optimizer's charge/export windows into ExecutionPlan entries and
         queues them for MQTT publishing on the next run() cycle.
         """
+        if self.gateway_display_only:
+            return
         plan_entries = []
 
         # Reserve as a percentage — used to detect freeze charge (charge limit == reserve)
@@ -631,6 +657,11 @@ async def run(self, seconds, first):
             ticks = await self._startup_wait(0, lambda: self._first_connection_attempted)
             if not self._first_connection_attempted and not self.api_stop:
                 self.log("Warn: GatewayMQTT: First connection attempt not yet complete, continuing startup")
+            if self.gateway_display_only:
+                await self._startup_wait(ticks, lambda: self.api_started)
+                if not self.api_started and not self.api_stop:
+                    self.log("Warn: GatewayMQTT: No display telemetry received, continuing startup")
+                return True
             # The auto-config wait runs even when the first attempt failed: the MQTT loop
             # retries with backoff in the background, so a transient broker failure at
             # startup usually still yields telemetry inside the window. Returning early
@@ -651,6 +682,14 @@ async def run(self, seconds, first):
                 self.log("Info: GatewayMQTT: Restarting MQTT listener task")
                 self._mqtt_task = asyncio.ensure_future(self._mqtt_loop())
 
+            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()
+                await self._check_token_refresh()
+                return True
+
             # Publish any queued plan from on_plan_executed hook
             if self._pending_plan:
                 plan_entries, tz = self._pending_plan
@@ -859,6 +898,15 @@ def _process_telemetry(self, data):
 
         self._debug_dump("RX telemetry", status, raw=data)
 
+        if self.gateway_display_only:
+            # A desk display need not discover an inverter. Do not inject the hub's
+            # inverter entities or register hardware against the selected provider.
+            self._last_status = status
+            self._last_telemetry_time = time.time()
+            self.update_success_timestamp()
+            self.api_started = True
+            return
+
         if len(status.inverters) == 0:
             if not (self.gateway_inverter_serial and self._discovery):
                 return
@@ -1807,6 +1855,8 @@ def automatic_config(self):
         (see ``_needs_reconfigure``). Maps proto fields to PredBat args so the plan
         engine has data to work with.
         """
+        if self.gateway_display_only:
+            return
         if not self._last_status:
             self.log("Error: GatewayMQTT: automatic_config called with no status data")
             return
@@ -2469,6 +2519,8 @@ async def publish_plan(self, plan_entries, timezone_str):
             plan_entries: List of plan entry dicts.
             timezone_str: IANA timezone string (e.g. "Europe/London").
         """
+        if self.gateway_display_only:
+            return
         if not self._plan_changed(plan_entries):
             return  # No change, skip publish
 
@@ -2556,6 +2608,9 @@ async def _publish_raw(self, topic, payload, retain=False):
             payload: Bytes to publish.
             retain: Whether to set the retain flag.
         """
+        if self.gateway_display_only and topic != f"{self._topic_base}/predbat_data":
+            self.log(f"Warn: GatewayMQTT: Display mode blocked publish to {topic}")
+            return
         if not (self._mqtt_client and self._mqtt_connected):
             return
 

apps/predbat/tests/test_gateway.py

@@ -7404,8 +7404,10 @@ def test_counters_restart_at_local_midnight(self):
 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
 
     test_classes = [
+        TestGatewayDisplayOnly,
         TestProtobufDecode,
         TestPlanSerialization,
         TestCommandFormat,

apps/predbat/tests/test_gateway_display.py

@@ -0,0 +1,154 @@
+"""Display-only gateway isolation and lifecycle regressions."""
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import gateway_status_pb2 as pb
+from gateway import GatewayMQTT
+
+
+class TestGatewayDisplayOnly:
+    """A display cannot become an inverter source or publish control messages."""
+
+    def _make_gateway(self, display=True):
+        """Build the real gateway state with an in-memory MQTT client."""
+        gw = GatewayMQTT.__new__(GatewayMQTT)
+        gw.base = MagicMock()
+        gw.args = {}
+        gw.log = MagicMock()
+        gw.initialize(gateway_device_id="pbgw_test", mqtt_host="mqtt.test", gateway_display_only=display)
+        gw._mqtt_connected = True
+        gw._mqtt_client = SimpleNamespace(publish=AsyncMock())
+        gw.api_started = False
+        gw.api_stop = False
+        gw.update_success_timestamp = MagicMock()
+        return gw
+
+    def test_zero_inverters_is_a_ready_display(self):
+        """A status with no inverter completes display startup without changing control."""
+        gw = self._make_gateway()
+        gw.automatic_config = MagicMock()
+        gw._inject_entities = MagicMock()
+        status = pb.GatewayStatus(device_id="pbgw_test")
+        gw._process_telemetry(status.SerializeToString())
+        assert gw.api_started
+        assert not gw._auto_configured
+        gw.automatic_config.assert_not_called()
+        gw._inject_entities.assert_not_called()
+        gw.update_success_timestamp.assert_called_once()
+
+    def test_discovered_inverter_cannot_replace_cloud(self):
+        """Even full telemetry never registers inverters or EVs in display mode."""
+        gw = self._make_gateway()
+        gw.set_arg = MagicMock()
+        status = pb.GatewayStatus(device_id="pbgw_test")
+        status.inverters.add(serial="CE12345678", connected=True)
+        status.ev_chargers.add(charge_point_id="charger")
+        gw._process_telemetry(status.SerializeToString())
+        gw.automatic_config()
+        gw.set_arg.assert_not_called()
+        assert not gw._auto_configured
+        assert not gw._configured_ev_chargers
+
+    def test_invalid_telemetry_does_not_start_display(self):
+        """A corrupt message must not satisfy the readiness condition."""
+        gw = self._make_gateway()
+        gw._process_telemetry(b"\xff")
+        assert not gw.api_started
+        assert gw._last_status is None
+
+    def test_display_only_allows_display_topic(self):
+        """The final MQTT boundary blocks all command and schedule paths."""
+        gw = self._make_gateway()
+
+        async def exercise():
+            for suffix in ("command", "ev/command", "schedule", "unknown"):
+                await gw._publish_raw(f"{gw._topic_base}/{suffix}", b"control", retain=True)
+            await gw._publish_raw(f"{gw._topic_base}/predbat_data", b"display", retain=True)
+
+        asyncio.run(exercise())
+        gw._mqtt_client.publish.assert_awaited_once_with(f"{gw._topic_base}/predbat_data", b"display", qos=1, retain=True)
+
+    def test_public_control_entry_points_cannot_send(self):
+        """Direct callers cannot bypass display mode through control helpers."""
+        gw = self._make_gateway()
+
+        async def exercise():
+            await gw.publish_command("set_read_only", enable=False)
+            await gw._send_ev_command({"action": "RemoteStartTransaction"})
+            await gw.publish_plan([], "Europe/London")
+
+        asyncio.run(exercise())
+        gw._mqtt_client.publish.assert_not_awaited()
+        assert gw._last_plan_data is None
+
+    def test_plan_hook_does_not_queue_execution(self):
+        """Display plans travel through JSON, never the autonomous schedule hook."""
+        gw = self._make_gateway()
+        gw._on_plan_executed()
+        assert gw._pending_plan is None
+
+    def test_display_housekeeping_never_releases_standby(self):
+        """An online display publishes JSON and refreshes auth, without control housekeeping."""
+        gw = self._make_gateway()
+        gw._last_status = pb.GatewayStatus(device_id="pbgw_test")
+        gw._publish_predbat_data = AsyncMock()
+        gw._check_token_refresh = AsyncMock()
+        gw._check_read_only_state = AsyncMock()
+        gw._check_inverter_resets = AsyncMock()
+        gw._check_rate_caps = AsyncMock()
+        gw._republish_plan_if_stale = AsyncMock()
+        with patch("gateway.HAS_AIOMQTT", True):
+            assert asyncio.run(gw.run(60, False))
+        gw._publish_predbat_data.assert_awaited_once()
+        gw._check_token_refresh.assert_awaited_once()
+        for method in (gw._check_read_only_state, gw._check_inverter_resets, gw._check_rate_caps, gw._republish_plan_if_stale):
+            method.assert_not_awaited()
+
+    def test_display_startup_does_not_wait_for_auto_config(self):
+        """Display startup accepts zero-inverter telemetry, using the same bounded budget."""
+        gw = self._make_gateway()
+        gw.api_started = True
+        gw._first_connection_attempted = True
+        gw._startup_wait = AsyncMock(side_effect=lambda ticks, ready: ticks if ready() else 999)
+        gw._mqtt_loop = AsyncMock()
+        with patch("gateway.HAS_AIOMQTT", True):
+            assert asyncio.run(gw.run(0, True))
+        assert all(call.args[1]() for call in gw._startup_wait.call_args_list)
+        assert not gw._auto_configured
+
+    def test_display_is_not_an_inverter_source(self):
+        """A display cannot make the engine wait for battery readings it will never supply."""
+        from components import Components
+
+        comps = Components.__new__(Components)
+        comps.components = {"gateway": self._make_gateway()}
+        comps.load_error = lambda name: None
+        comps.is_alive = lambda name: True
+        assert not comps.inverter_source_active()
+        assert comps.inverter_source_names() == []
+        assert comps.inverter_source_status() == []
+        comps.components["gateway"].gateway_display_only = False
+        assert comps.inverter_source_active()
+        assert comps.inverter_source_names() == ["PredBat Gateway"]
+        assert comps.inverter_source_status() == ["PredBat Gateway (OK)"]
+
+    def test_false_string_keeps_existing_control_mode(self):
+        """String config values are parsed rather than treated as truthy strings."""
+        assert not self._make_gateway("false").gateway_display_only
+        assert self._make_gateway("true").gateway_display_only
+
+    def test_invalid_display_flag_fails_closed(self):
+        """An invalid role value must not silently enable inverter control."""
+        try:
+            self._make_gateway("typo")
+        except ValueError:
+            return
+        raise AssertionError("Invalid display flag was accepted")
+
+    def test_control_mode_can_still_publish(self):
+        """The opt-in mode does not block the existing control transport."""
+        gw = self._make_gateway(False)
+        asyncio.run(gw._publish_raw(gw.topic_command, b"control"))
+        gw._mqtt_client.publish.assert_awaited_once()

docs/apps-yaml.md

@@ -615,6 +615,12 @@ See also **ge_cloud_automatic_split_ct** which takes priority over this setting
 Use this to override automatic shared-CT detection if Predbat incorrectly identifies your system as sharing a CT clamp (e.g. when duplicate meter serials are reported by the cloud API but the inverters actually have separate CT clamps).
 This setting takes priority over **ge_cloud_automatic_shared_ct** if both are set.
 
+- **gateway_display_only** - Optional, defaults to false. Set to `true` to send prices, costs and the plan timeline to a paired Hub while another integration controls the battery.
+The Hub does not need a discovered inverter. This mode does not configure inverter or EV charger entities, change the selected inverter integration, or publish commands or executable schedules.
+Keep the Hub in enforced standby (including passive mode on supported firmware) before enabling this option or handing control to another integration. The option only prevents this Predbat component from sending writes; it cannot stop an old schedule or autonomous fallback already running on the Hub.
+The existing firmware still needs a local inverter connection for live power flow and battery readings; cloud-fed live readings and pairing without inverter discovery are separate changes.
+This is a display option, not automatic backup switching.
+
 - **gateway_shared_ct** - Optional, defaults to true. The equivalent of **ge_cloud_automatic_shared_ct** for inverters controlled through the Predbat Gateway.
 When `true`, Predbat treats multiple inverters as sharing a single physical CT clamp for grid and load measurement,
 so only the first inverter's grid and load power readings are used (the rest are zeroed out), preventing double-counting of grid import/export and house load.