From 431bd778bd41f72536800986ba32360c82799c6b Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Tue, 13 Oct 2026 08:09:28 +0100 Subject: [PATCH 1/5] fix(gateway): never enable discharge/export after a failed window write For a direct Hub battery, adjust_force_export wrote the discharge start and end as two separate set_discharge_slot commands, then enabled the scheduled discharge and switched to export mode regardless of whether the endpoint writes were applied. schedule_write_ok only withheld the later commit record. When the end write was refused after the start was accepted, the Hub was left enabled with a stale wrapping window (e.g. 02:55-00:01) and actively exporting. Send both endpoints in one full-serial set_discharge_slot for direct Hub batteries and verify both endpoint read-backs before continuing. Any refusal, staged or partial ACK, timeout, publication error or swallowed async-bridge exception returns before the enable write, export mode and commit, leaving the commit pending for the next cycle to retry. Both endpoints are tracked in the control ledger and share one backoff key. Read-only, entity ownership and serial checks refuse before publishing. Coordinators keep their staged separate-endpoint sequence; other providers, disable and midnight windows are unchanged. Mirrors the charge-window fix in #5531. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_018BaJjLpuMEtQYfvmTRcFJf --- apps/predbat/gateway.py | 38 +++ apps/predbat/inverter.py | 84 ++++++- apps/predbat/tests/test_gateway.py | 2 + .../tests/test_gateway_discharge_window.py | 234 ++++++++++++++++++ 4 files changed, 356 insertions(+), 2 deletions(-) create mode 100644 apps/predbat/tests/test_gateway_discharge_window.py diff --git a/apps/predbat/gateway.py b/apps/predbat/gateway.py index d561501d4..8083a4c7a 100644 --- a/apps/predbat/gateway.py +++ b/apps/predbat/gateway.py @@ -2654,6 +2654,44 @@ async def _update_charge_slot(self, entity_id, hhmm, serial): await self._send_control(entity_id, "set_charge_slot", schedule_json=json.dumps(schedule), serial=serial) self.log(f"Info: GatewayMQTT: Charge slot update: {schedule}") + def discharge_window_uses_staging(self, start_entity, end_entity): + """Identify a coordinator synchronously, before crossing an async bridge.""" + if not isinstance(start_entity, str) or not start_entity.endswith("_discharge_slot1_start"): + return False + if end_entity != start_entity.removesuffix("_start") + "_end": + return False + serial = self._serial_from_entity_id(start_entity) + record = getattr(self, "_discovery", {}).get((serial or "").upper(), {}) + return bool(serial) and record.get("type") in GATEWAY_COORDINATOR_TYPES + + async def update_discharge_window(self, start_entity, end_entity, start, end): + """Send an engine discharge window as one full-serial command, not two partial slots. + + True means publication was attempted, not that the inverter applied it. The + caller must verify both endpoint entities, which only an applied ACK or + physical telemetry updates. Coordinators use their existing separate path: + those devices can stage a disabled slot and apply it only on enable. + Manual single-endpoint selects keep their path. + """ + if getattr(self.base, "set_read_only", False): + return False + if not isinstance(start_entity, str) or not start_entity.endswith("_discharge_slot1_start"): + return False + if end_entity != start_entity.removesuffix("_start") + "_end": + return False + serial = self._serial_from_entity_id(start_entity) + if not serial or self._serial_from_entity_id(end_entity) != serial: + return False + if self.discharge_window_uses_staging(start_entity, end_entity): + return False + try: + times = [datetime.datetime.strptime(value, "%H:%M:%S") for value in (start, end)] + except (ValueError, TypeError): + return False + schedule = {field: value.hour * 100 + value.minute for field, value in zip(("start", "end"), times)} + await self._send_control(start_entity, "set_discharge_slot", schedule_json=json.dumps(schedule), serial=serial) + return True + async def _update_discharge_slot(self, entity_id, hhmm, serial): """Send set_discharge_slot command with updated start or end time.""" if "_start" in entity_id: diff --git a/apps/predbat/inverter.py b/apps/predbat/inverter.py index 604ab2317..7a7688f7c 100644 --- a/apps/predbat/inverter.py +++ b/apps/predbat/inverter.py @@ -2927,6 +2927,74 @@ def adjust_idle_time(self, charge_start=None, charge_end=None, discharge_start=N self.write_and_poll_option("idle_end_time", idle_end_time_id, idle_end) self.idle_end_minutes = idle_end_minutes + def write_and_poll_gateway_discharge_window(self, new_start, new_end): + """Verify a complete Hub slot together, or return None for another provider. + + A start equal to the old end cannot be sent alone while discharging is enabled: + it temporarily disables the slot and the Hub correctly refuses it. Sending + both bounds also handles swapping the bounds of a midnight-spanning slot. + """ + from ha import run_async + + entities = [self.base.get_arg(name, indirect=False, index=self.id) for name in ("discharge_start_time", "discharge_end_time")] + owners = [getattr(self.base, "dashboard_index_app", {}).get(entity) for entity in entities] + if "gateway" not in owners: + return None + components = getattr(self.base, "components", None) + gateway = components.get_component("gateway") if components else None + names = ("discharge_start_time", "discharge_end_time") + target = (new_start, new_end) + if owners != ["gateway", "gateway"] or gateway is None: + self.base.record_status(f"Warn: Inverter {self.id} Hub discharge window has inconsistent entity ownership", had_errors=True) + return False + if not all(self.check_write_entity("write_and_poll_gateway_discharge_window", name, entity, value) for name, entity, value in zip(names, entities, target)): + return False + if gateway.discharge_window_uses_staging(*entities): + return None # Coordinator slot staging must precede its enable write. + + def read_pair(): + """Read both endpoint entities without substituting desired values.""" + return tuple(self.base.get_state_wrapper(entity, refresh=True) for entity in entities) + + before = read_pair() + ledger = self.base.control_ledger + if ledger is not None: + for name, entity, value, wanted in zip(names, entities, before, target): + self._ledger_observe(ledger, name, entity, value) + if value != wanted: + ledger.note_write_attempt(entity) + key = "|".join(entities) + verified = before == target + for _retry in range(0 if verified else self._write_attempts("discharge_window", key, target)): + try: + published = run_async(gateway.update_discharge_window(*entities, *target)) + except Exception as error: + self.base.log(f"Warn: Inverter {self.id} Hub discharge window publication failed: {error}") + break + if published is not True: + break + self._poll_after_write(entities[0], lambda state: state == new_start and self.base.get_state_wrapper(entities[1], refresh=True) == new_end) + verified = read_pair() == target + if verified: + self.count_register_writes += 1 + self.registers_moved += sum(old != wanted for old, wanted in zip(before, target)) + break + self._write_backoff_result("discharge_window", key, target, verified) + if ledger is not None: + for name, entity, value in zip(names, entities, target): + if verified: + if before == target: + ledger.record_ownership_from_read(entity, name, value, now=time.time(), generation=self._ledger_generation(entity)) + else: + ledger.record_write(entity, name, value, now=time.time(), generation=self._ledger_generation(entity)) + else: + ledger.clear(entity) + if not verified: + message = f"Warn: Inverter {self.id} write to Hub discharge window failed: wanted {target}, got {read_pair()}" + self.base.log(message) + self.base.record_status(message, had_errors=True) + return verified + def adjust_force_export(self, force_export, new_start_time=None, new_end_time=None): """ Adjust force export on/off and set the time window correctly @@ -3024,8 +3092,20 @@ def adjust_force_export(self, force_export, new_start_time=None, new_end_time=No else: self.adjust_idle_time(discharge_start="00:00:00", discharge_end="00:00:00") + # Direct Hub batteries accept the complete slot as one physical write. Do not + # enable or enter export mode after a refused, staged or incomplete read-back. + # Coordinators retain their existing staged-endpoint then enable sequence. + gateway_window = None + if (new_start and new_start != old_start) or (new_end and new_end != old_end): + gateway_window = self.write_and_poll_gateway_discharge_window(new_start or old_start, new_end or old_end) + if gateway_window is False: + self.record_commit("export_window", (new_start, new_end, force_export), False) + return + if gateway_window is True: + changed_start_end = True + # Change start time - if new_start and (new_start != old_start or is_hm_format): + if gateway_window is None and new_start and (new_start != old_start or is_hm_format): self.base.log("Inverter {} set new export start time to {}".format(self.id, new_start)) if "discharge_start_time" in self.base.args: # Always write to this as it is the GE default @@ -3047,7 +3127,7 @@ def adjust_force_export(self, force_export, new_start_time=None, new_end_time=No self.log("Warn: Inverter {} unable write export start time as neither REST or discharge_start_time are set".format(self.id)) # Change end time - if new_end and (new_end != old_end or is_hm_format): + if gateway_window is None and new_end and (new_end != old_end or is_hm_format): self.base.log("Inverter {} Set new export end time to {} was {}".format(self.id, new_end, old_end)) if "discharge_end_time" in self.base.args: # Always write to this as it is the GE default diff --git a/apps/predbat/tests/test_gateway.py b/apps/predbat/tests/test_gateway.py index f83d2b08b..7df90464c 100644 --- a/apps/predbat/tests/test_gateway.py +++ b/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_discharge_window import TestGatewayDischargeWindow test_classes = [ + TestGatewayDischargeWindow, TestProtobufDecode, TestPlanSerialization, TestCommandFormat, diff --git a/apps/predbat/tests/test_gateway_discharge_window.py b/apps/predbat/tests/test_gateway_discharge_window.py new file mode 100644 index 000000000..2416ce0d6 --- /dev/null +++ b/apps/predbat/tests/test_gateway_discharge_window.py @@ -0,0 +1,234 @@ +"""Joined engine discharge window, command serialization and ACK regressions.""" +# cspell:ignore excepthook + +import asyncio +import datetime +import json +import types +import unittest +from unittest.mock import MagicMock, patch + +from gateway import GatewayMQTT, GATEWAY_COORDINATOR_TYPES +from inverter import Inverter +from tests.test_gateway import TestCommandAck + + +class DischargeBase: + """Synthetic service boundary backed by the production gateway ACK cache.""" + + START = "select.predbat_gateway_456789_discharge_slot1_start" + END = "select.predbat_gateway_456789_discharge_slot1_end" + ENABLE = "switch.predbat_gateway_456789_discharge_enabled" + + def __init__(self, gateway): + """Configure one direct battery and retain errors and service calls.""" + self.gateway = gateway + self.args = {"discharge_start_time": self.START, "discharge_end_time": self.END, "scheduled_discharge_enable": self.ENABLE} + self.dashboard_index_app = {entity: "gateway" for entity in self.args.values()} + self.components = types.SimpleNamespace(get_component=lambda name: gateway if name == "gateway" else None) + self.control_ledger = None + self.set_read_only = self.set_inverter_notify = False + self.inverter_clock_skew_discharge_start = self.inverter_clock_skew_discharge_end = 0 + self.logs, self.errors, self.services = [], [], [] + + def get_arg(self, name, default=None, index=None, indirect=True, **kwargs): + """Resolve configured entities or their observed values.""" + entity = self.args.get(name) + return self.gateway.cache.get(entity, default) if indirect else entity + + def get_state_wrapper(self, entity_id, **kwargs): + """Read the production ACK cache.""" + return self.gateway.cache.get(entity_id) + + def log(self, message): + """Retain execution messages.""" + self.logs.append(message) + + def record_status(self, message, **kwargs): + """Retain failed verification reports.""" + self.errors.append(message) + + def call_service_wrapper(self, service, entity_id, **kwargs): + """Route ordinary entity writes to their actual gateway handlers.""" + self.services.append((service, entity_id, kwargs)) + if service.endswith("/select_option"): + asyncio.run(self.gateway.select_event(entity_id, kwargs["option"])) + else: + asyncio.run(self.gateway.switch_event(entity_id, service.split("/")[1])) + + +class TestGatewayDischargeWindow(unittest.TestCase): + """Exercise actual inverter writes, gateway commands and ACK handling.""" + + def fixture(self, outcome="applied", start="00:00:00", end="00:01:00", enabled="off"): + """Use synthetic transport with applied, failed, staged or missing ACKs.""" + gw = TestCommandAck()._make_gateway() + base = DischargeBase(gw) + gw.base = base + gw.topic_command = "predbat/devices/pbgw_test/command" + gw.publish_command = types.MethodType(GatewayMQTT.publish_command, gw) + gw.cache.update({base.START: start, base.END: end, base.ENABLE: enabled}) + wire, mode = [], {"outcome": outcome} + + async def transport(topic, payload, retain=False): + """Apply actual ACK updates while controlling transport outcomes.""" + command = json.loads(payload) + wire.append(command) + self.assertFalse(retain) + if mode["outcome"] == "exception": + raise OSError("synthetic MQTT failure") + if mode["outcome"] == "timeout": + return + refused = False + if command["command"] == "set_discharge_slot": + part = json.loads(command["schedule_json"]) + # Both the legacy single endpoint and paired paths reach this sink. + hhmm = lambda value: int(value[:2]) * 100 + int(value[3:5]) + applied = {"start": hhmm(gw.cache[base.START]), "end": hhmm(gw.cache[base.END]), **part} + refused = mode["outcome"] == "refused" and "end" in part + if mode["outcome"] == "staged": + applied["staged"] = True + elif mode["outcome"] == "partial": + applied["end"] = 1 + else: + applied = {"enable": command["enable"]} + ack = {"command_id": command["command_id"], "command": command["command"], "ok": not refused, "applied": applied} + if refused: + ack["error"] = "modbus_write_failed" + gw._process_ack(json.dumps(ack)) + + gw._publish_raw = transport + inv = Inverter.__new__(Inverter) + inv.base, inv.log, inv.id = base, base.log, 0 + inv.count_register_writes = inv.registers_moved = 0 + inv.inv_write_backoff = False + inv.write_backoff = {} + inv.inv_write_max_retry, inv.inv_write_and_poll_sleep = 3, 0.01 + inv.inv_has_discharge_enable_time = inv.inv_has_charge_enable_time = True + inv.inv_has_ge_inverter_mode = inv.inv_time_button_press = False + inv.inv_charge_time_format = "H:M" + inv.reserve_percent = 4 + inv.commit_pending, inv.last_committed = {}, {} + inv.sleep = lambda seconds: None + inv.adjust_idle_time = MagicMock() + inv.adjust_inverter_mode = MagicMock() + return inv, gw, base, wire, mode + + def apply(self, inv, force=True, start="02:55:00", end="03:01:00"): + """Run the actual discharge scheduling method.""" + parse = lambda value: datetime.datetime.fromisoformat("2026-10-13T" + value) if value else None + inv.adjust_force_export(force, parse(start), parse(end)) + + def test_failed_endpoint_never_enables_stale_window(self): + """A failed discharge pair must not enable its stale midnight window.""" + inv, gw, base, wire, mode = self.fixture(outcome="refused") + self.apply(inv) + self.assertTrue(base.errors) + self.assertFalse(any(command["command"] == "set_discharge_enable" for command in wire)) + inv.adjust_inverter_mode.assert_not_called() + self.assertEqual(gw.cache[base.ENABLE], "off") + self.assertTrue(inv.commit_pending.get("export_window")) + + def test_pair_applied_before_enable_and_full_serial(self): + """Only both applied endpoint values allow the subsequent enable command.""" + inv, gw, base, wire, mode = self.fixture() + self.apply(inv) + self.assertEqual([command["command"] for command in wire], ["set_discharge_slot", "set_discharge_enable"]) + self.assertEqual(json.loads(wire[0]["schedule_json"]), {"start": 255, "end": 301}) + self.assertEqual(wire[0]["serial"], "CE123456789") + self.assertEqual((gw.cache[base.START], gw.cache[base.END], gw.cache[base.ENABLE]), ("02:55:00", "03:01:00", "on")) + self.assertFalse(base.errors) + + def test_unverified_pair_clears_ledger_and_cannot_activate(self): + """Refusal, incomplete or staged ACKs, timeout and exceptions fail closed.""" + for outcome in ("refused", "partial", "staged", "timeout", "exception"): + inv, gw, base, wire, mode = self.fixture(outcome=outcome) + base.control_ledger = MagicMock() + self.apply(inv) + self.assertTrue(base.errors, outcome) + self.assertFalse(any(command["command"] == "set_discharge_enable" for command in wire), outcome) + self.assertEqual(base.control_ledger.clear.call_count, 2, outcome) + base.control_ledger.record_write.assert_not_called() + inv.adjust_inverter_mode.assert_not_called() + self.assertTrue(inv.commit_pending.get("export_window")) + self.assertNotIn("export_window", inv.last_committed) + + def test_retry_after_refusal_can_apply_and_enable(self): + """A later verified retry recovers without treating a failed pair as committed.""" + inv, gw, base, wire, mode = self.fixture(outcome="refused") + self.apply(inv) + mode["outcome"] = "applied" + with patch("gateway._monotonic", return_value=10**12): + self.apply(inv) + self.assertEqual(gw.cache[base.END], "03:01:00") + self.assertEqual(gw.cache[base.ENABLE], "on") + self.assertFalse(inv.commit_pending["export_window"]) + self.assertEqual(json.loads(wire[-2]["schedule_json"]), {"start": 255, "end": 301}) + + def test_read_only_owner_and_serial_refuse(self): + """Do not publish across read-only or mismatched endpoint boundaries.""" + for reason in ("read_only", "owner", "serial"): + inv, gw, base, wire, mode = self.fixture() + if reason == "read_only": + base.set_read_only = True + elif reason == "owner": + base.dashboard_index_app[base.END] = "gecloud" + else: + entity = base.END.replace("456789", "654321") + base.args["discharge_end_time"] = entity + base.dashboard_index_app[entity] = "gateway" + self.apply(inv) + self.assertFalse(wire, reason) + self.assertTrue(base.errors, reason) + inv.adjust_inverter_mode.assert_not_called() + + def test_async_exception_does_not_fall_back(self): + """A swallowed async bridge exception remains failure, never coordinator fallback.""" + inv, gw, base, wire, mode = self.fixture(outcome="exception") + + async def timer(): + """Match a timer invoking synchronous engine execution on its event loop.""" + self.apply(inv) + + with patch("threading.excepthook") as error: + asyncio.run(timer()) + error.assert_called_once() + self.assertEqual(len(wire), 1) + self.assertFalse(base.services) + inv.adjust_inverter_mode.assert_not_called() + + def test_other_provider_and_coordinator_keep_individual_sequence(self): + """Only direct Hub inverters opt into the new pair and verification gate.""" + for kind in (None, *GATEWAY_COORDINATOR_TYPES): + inv, gw, base, wire, mode = self.fixture() + if kind is None: + base.dashboard_index_app = {entity: "gecloud" for entity in base.args.values()} + else: + gw._discovery = {"CE123456789": {"type": kind}} + self.apply(inv) + self.assertEqual([json.loads(command["schedule_json"]) for command in wire if command["command"] == "set_discharge_slot"], [{"start": 255}, {"end": 301}]) + self.assertEqual(wire[-1]["command"], "set_discharge_enable") + + def test_disable_and_midnight_zero_window_remain_valid(self): + """A disable request never activates export and may program midnight endpoints.""" + inv, gw, base, wire, mode = self.fixture(enabled="on") + self.apply(inv, force=False, start="00:00:00", end="00:00:00") + self.assertFalse(any(command.get("enable") is True for command in wire)) + self.assertEqual(gw.cache[base.ENABLE], "off") + self.assertEqual((gw.cache[base.START], gw.cache[base.END]), ("00:00:00", "00:00:00")) + + def test_success_ledger_and_backoff(self): + """Paired writes own both endpoints and use a stable pair retry key.""" + inv, gw, base, wire, mode = self.fixture(outcome="refused") + base.control_ledger = MagicMock() + inv.inv_write_backoff = True + self.apply(inv) + key = base.START + "|" + base.END + self.assertEqual(inv.write_backoff[key]["failures"], 1) + mode["outcome"] = "applied" + with patch("gateway._monotonic", return_value=10**12): + self.apply(inv) + self.assertNotIn(key, inv.write_backoff) + writes = base.control_ledger.record_write.call_args_list + self.assertTrue(any(call.args[:3] == (base.START, "discharge_start_time", "02:55:00") for call in writes)) + self.assertTrue(any(call.args[:3] == (base.END, "discharge_end_time", "03:01:00") for call in writes)) From 529e6f89136a0a1a7d5436bc3fe7381e8f5b2681 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Tue, 13 Oct 2026 08:18:11 +0100 Subject: [PATCH 2/5] fix(gateway): detect coordinator staging before discharge ownership checks A coordinator whose discharge end entity was not (yet) owned by the gateway was refused with an ownership error instead of keeping its staged separate-endpoint sequence. Identify coordinators first, as that decision only needs the entity names and discovery record. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_018BaJjLpuMEtQYfvmTRcFJf --- apps/predbat/inverter.py | 4 ++-- apps/predbat/tests/test_gateway_discharge_window.py | 11 +++++++++++ 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/apps/predbat/inverter.py b/apps/predbat/inverter.py index 7a7688f7c..4303b3756 100644 --- a/apps/predbat/inverter.py +++ b/apps/predbat/inverter.py @@ -2944,13 +2944,13 @@ def write_and_poll_gateway_discharge_window(self, new_start, new_end): gateway = components.get_component("gateway") if components else None names = ("discharge_start_time", "discharge_end_time") target = (new_start, new_end) + if gateway is not None and gateway.discharge_window_uses_staging(*entities): + return None # Coordinator slot staging must precede its enable write. if owners != ["gateway", "gateway"] or gateway is None: self.base.record_status(f"Warn: Inverter {self.id} Hub discharge window has inconsistent entity ownership", had_errors=True) return False if not all(self.check_write_entity("write_and_poll_gateway_discharge_window", name, entity, value) for name, entity, value in zip(names, entities, target)): return False - if gateway.discharge_window_uses_staging(*entities): - return None # Coordinator slot staging must precede its enable write. def read_pair(): """Read both endpoint entities without substituting desired values.""" diff --git a/apps/predbat/tests/test_gateway_discharge_window.py b/apps/predbat/tests/test_gateway_discharge_window.py index 2416ce0d6..c24768c82 100644 --- a/apps/predbat/tests/test_gateway_discharge_window.py +++ b/apps/predbat/tests/test_gateway_discharge_window.py @@ -1,4 +1,5 @@ """Joined engine discharge window, command serialization and ACK regressions.""" + # cspell:ignore excepthook import asyncio @@ -209,6 +210,16 @@ def test_other_provider_and_coordinator_keep_individual_sequence(self): self.assertEqual([json.loads(command["schedule_json"]) for command in wire if command["command"] == "set_discharge_slot"], [{"start": 255}, {"end": 301}]) self.assertEqual(wire[-1]["command"], "set_discharge_enable") + def test_coordinator_with_partial_ownership_keeps_staged_sequence(self): + """A coordinator is identified before ownership checks, so it is never refused by them.""" + for kind in GATEWAY_COORDINATOR_TYPES: + inv, gw, base, wire, mode = self.fixture() + gw._discovery = {"CE123456789": {"type": kind}} + base.dashboard_index_app[base.END] = None + self.apply(inv) + self.assertFalse(any("inconsistent entity ownership" in error for error in base.errors), kind) + self.assertEqual([json.loads(command["schedule_json"]) for command in wire if command["command"] == "set_discharge_slot"], [{"start": 255}, {"end": 301}]) + def test_disable_and_midnight_zero_window_remain_valid(self): """A disable request never activates export and may program midnight endpoints.""" inv, gw, base, wire, mode = self.fixture(enabled="on") From 57b7b4aaa1f85121f0ada3be4c29208a4fe76818 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Tue, 13 Oct 2026 08:21:30 +0100 Subject: [PATCH 3/5] fix(gateway): stop an armed export when its window move is refused; no re-poll of a refusal - When the paired window write fails while scheduled export is already enabled, turn the schedule off and return to Eco instead of leaving the previous window exporting on a plan that has moved away from it (skipped in read-only mode). - A refusal of the exact paired command ends the attempt immediately: the poll stops when it arrives, and a retry inside the ack window is not published (_send_control would suppress it anyway), so a refused write no longer holds the engine for the full poll timeout per retry. - Drop the unreachable end-serial comparison (the end entity name is derived from the start, so it carries the same serial) and name the test case after what it exercises. - Document that an exception inside a running event loop reaches the caller as None via run_async's thread, not as an exception. - Test the write/moved counters for the paired path. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_018BaJjLpuMEtQYfvmTRcFJf --- apps/predbat/gateway.py | 42 +++++++++++++++---- apps/predbat/inverter.py | 15 ++++++- .../tests/test_gateway_discharge_window.py | 35 +++++++++++++++- 3 files changed, 80 insertions(+), 12 deletions(-) diff --git a/apps/predbat/gateway.py b/apps/predbat/gateway.py index 8083a4c7a..1d377c106 100644 --- a/apps/predbat/gateway.py +++ b/apps/predbat/gateway.py @@ -2675,22 +2675,46 @@ async def update_discharge_window(self, start_entity, end_entity, start, end): """ if getattr(self.base, "set_read_only", False): return False - if not isinstance(start_entity, str) or not start_entity.endswith("_discharge_slot1_start"): - return False - if end_entity != start_entity.removesuffix("_start") + "_end": + if self.discharge_window_uses_staging(start_entity, end_entity): return False - serial = self._serial_from_entity_id(start_entity) - if not serial or self._serial_from_entity_id(end_entity) != serial: + kwargs = self._discharge_window_kwargs(start_entity, end_entity, start, end) + if kwargs is None: return False - if self.discharge_window_uses_staging(start_entity, end_entity): + if self.discharge_window_refused(start_entity, end_entity, start, end): + # _send_control would not re-send it inside the ack window, so polling again + # would only wait out the timeout for an answer that has already arrived + self.log(f"Info: GatewayMQTT: discharge window {start}-{end} for {kwargs['serial']} was just refused, not retrying yet") return False + await self._send_control(start_entity, "set_discharge_slot", **kwargs) + return True + + def _discharge_window_kwargs(self, start_entity, end_entity, start, end): + """The paired set_discharge_slot fields for a slot1 start/end entity pair, or None.""" + # The end entity name is derived from the start, so both carry the same serial + if not isinstance(start_entity, str) or not start_entity.endswith("_discharge_slot1_start"): + return None + if end_entity != start_entity.removesuffix("_start") + "_end": + return None + serial = self._serial_from_entity_id(start_entity) + if not serial: + return None try: times = [datetime.datetime.strptime(value, "%H:%M:%S") for value in (start, end)] except (ValueError, TypeError): - return False + return None schedule = {field: value.hour * 100 + value.minute for field, value in zip(("start", "end"), times)} - await self._send_control(start_entity, "set_discharge_slot", schedule_json=json.dumps(schedule), serial=serial) - return True + return {"schedule_json": json.dumps(schedule), "serial": serial} + + def discharge_window_refused(self, start_entity, end_entity, start, end): + """Whether the hub refused this exact paired window within the ack window.""" + kwargs = self._discharge_window_kwargs(start_entity, end_entity, start, end) + if kwargs is None or not (getattr(self, "_ack_subscribed", False) and self._mqtt_connected and self._acks_seen): + return False + with self._command_lock: + entry = self._pending_commands.get(start_entity) + if entry is None or entry["command"] != "set_discharge_slot" or entry["kwargs"] != kwargs: + return False + return entry["state"] == "refused" and _monotonic() - entry["sent_at"] < _COMMAND_ACK_WINDOW async def _update_discharge_slot(self, entity_id, hhmm, serial): """Send set_discharge_slot command with updated start or end time.""" diff --git a/apps/predbat/inverter.py b/apps/predbat/inverter.py index 4303b3756..88ec9851a 100644 --- a/apps/predbat/inverter.py +++ b/apps/predbat/inverter.py @@ -2969,12 +2969,18 @@ def read_pair(): try: published = run_async(gateway.update_discharge_window(*entities, *target)) except Exception as error: + # Only reached without a running event loop: inside one, run_async's thread hands + # the exception to the thread exception hook and returns None, handled just below self.base.log(f"Warn: Inverter {self.id} Hub discharge window publication failed: {error}") break if published is not True: break - self._poll_after_write(entities[0], lambda state: state == new_start and self.base.get_state_wrapper(entities[1], refresh=True) == new_end) + # A refusal is final for this cycle (the identical command is not re-sent within the + # ack window), so stop waiting as soon as it arrives rather than polling to the timeout + self._poll_after_write(entities[0], lambda state: (state == new_start and self.base.get_state_wrapper(entities[1], refresh=True) == new_end) or gateway.discharge_window_refused(*entities, *target)) verified = read_pair() == target + if not verified and gateway.discharge_window_refused(*entities, *target): + break if verified: self.count_register_writes += 1 self.registers_moved += sum(old != wanted for old, wanted in zip(before, target)) @@ -3099,6 +3105,13 @@ def adjust_force_export(self, force_export, new_start_time=None, new_end_time=No if (new_start and new_start != old_start) or (new_end and new_end != old_end): gateway_window = self.write_and_poll_gateway_discharge_window(new_start or old_start, new_end or old_end) if gateway_window is False: + if force_export and old_discharge_enable and not getattr(self.base, "set_read_only", False): + # The previous window is still armed: leaving it would keep exporting on a + # schedule the plan has moved away from, so stop exporting until a write lands. + # (Without force_export the schedule was already turned off above.) + self.log("Warn: Inverter {} Hub discharge window write failed, turning off scheduled export".format(self.id)) + self.write_and_poll_switch("scheduled_discharge_enable", self.base.get_arg("scheduled_discharge_enable", indirect=False, index=self.id), False) + self.adjust_inverter_mode(False) self.record_commit("export_window", (new_start, new_end, force_export), False) return if gateway_window is True: diff --git a/apps/predbat/tests/test_gateway_discharge_window.py b/apps/predbat/tests/test_gateway_discharge_window.py index c24768c82..ecda37374 100644 --- a/apps/predbat/tests/test_gateway_discharge_window.py +++ b/apps/predbat/tests/test_gateway_discharge_window.py @@ -140,6 +140,36 @@ def test_pair_applied_before_enable_and_full_serial(self): self.assertEqual((gw.cache[base.START], gw.cache[base.END], gw.cache[base.ENABLE]), ("02:55:00", "03:01:00", "on")) self.assertFalse(base.errors) + def test_paired_write_counts_one_write_and_both_moved_registers(self): + """The pair is one physical write that moves both endpoints.""" + inv, gw, base, wire, mode = self.fixture(enabled="on") + self.apply(inv) + self.assertEqual([command["command"] for command in wire], ["set_discharge_slot"]) + self.assertEqual((inv.count_register_writes, inv.registers_moved), (1, 2)) + + def test_refused_window_disables_an_already_exporting_schedule(self): + """A refused move must not leave the previous enabled window exporting.""" + inv, gw, base, wire, mode = self.fixture(outcome="refused", start="02:00:00", end="04:00:00", enabled="on") + self.apply(inv, start="02:30:00", end="03:00:00") + self.assertEqual(json.loads(wire[0]["schedule_json"]), {"start": 230, "end": 300}) + self.assertEqual((gw.cache[base.START], gw.cache[base.END]), ("02:00:00", "04:00:00")) + self.assertEqual(gw.cache[base.ENABLE], "off") + self.assertEqual([command.get("enable") for command in wire if command["command"] == "set_discharge_enable"], [False]) + inv.adjust_inverter_mode.assert_called_once_with(False) + self.assertTrue(inv.commit_pending["export_window"]) + + def test_refusal_is_not_polled_to_timeout_or_resent(self): + """A refusal ends the attempt at once; the identical command is not re-sent or re-polled.""" + inv, gw, base, wire, mode = self.fixture(outcome="refused") + inv.inv_write_and_poll_sleep = 3600 # any poll to timeout would hang the test + polls = [] + original = inv._poll_after_write + inv._poll_after_write = lambda *args, **kwargs: polls.append(args) or original(*args, **kwargs) + self.apply(inv) + self.assertEqual(len(polls), 1) + self.assertEqual([command["command"] for command in wire], ["set_discharge_slot"]) + self.assertTrue(base.errors) + def test_unverified_pair_clears_ledger_and_cannot_activate(self): """Refusal, incomplete or staged ACKs, timeout and exceptions fail closed.""" for outcome in ("refused", "partial", "staged", "timeout", "exception"): @@ -166,15 +196,16 @@ def test_retry_after_refusal_can_apply_and_enable(self): self.assertFalse(inv.commit_pending["export_window"]) self.assertEqual(json.loads(wire[-2]["schedule_json"]), {"start": 255, "end": 301}) - def test_read_only_owner_and_serial_refuse(self): + def test_read_only_owner_and_mismatched_pair_refuse(self): """Do not publish across read-only or mismatched endpoint boundaries.""" - for reason in ("read_only", "owner", "serial"): + for reason in ("read_only", "owner", "mismatched_pair"): inv, gw, base, wire, mode = self.fixture() if reason == "read_only": base.set_read_only = True elif reason == "owner": base.dashboard_index_app[base.END] = "gecloud" else: + # An end slot of another serial is not this start's pair entity = base.END.replace("456789", "654321") base.args["discharge_end_time"] = entity base.dashboard_index_app[entity] = "gateway" From ed83eec35c01edae80337d09e75817d4681ebef1 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Tue, 13 Oct 2026 08:22:25 +0100 Subject: [PATCH 4/5] test(gateway): a successful move of an exporting schedule stays enabled The fail-safe disable applies only to a failed window write. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_018BaJjLpuMEtQYfvmTRcFJf --- apps/predbat/tests/test_gateway_discharge_window.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/apps/predbat/tests/test_gateway_discharge_window.py b/apps/predbat/tests/test_gateway_discharge_window.py index ecda37374..e76a77dde 100644 --- a/apps/predbat/tests/test_gateway_discharge_window.py +++ b/apps/predbat/tests/test_gateway_discharge_window.py @@ -158,6 +158,16 @@ def test_refused_window_disables_an_already_exporting_schedule(self): inv.adjust_inverter_mode.assert_called_once_with(False) self.assertTrue(inv.commit_pending["export_window"]) + def test_applied_move_of_an_exporting_schedule_keeps_it_exporting(self): + """The fail-safe disable is only for a failed write: a successful move stays enabled.""" + inv, gw, base, wire, mode = self.fixture(start="02:00:00", end="04:00:00", enabled="on") + self.apply(inv, start="02:30:00", end="03:00:00") + self.assertEqual([command["command"] for command in wire], ["set_discharge_slot"]) + self.assertEqual((gw.cache[base.START], gw.cache[base.END], gw.cache[base.ENABLE]), ("02:30:00", "03:00:00", "on")) + inv.adjust_inverter_mode.assert_called_once_with(True, changed_start_end=True) + self.assertFalse(inv.commit_pending["export_window"]) + self.assertFalse(base.errors) + def test_refusal_is_not_polled_to_timeout_or_resent(self): """A refusal ends the attempt at once; the identical command is not re-sent or re-polled.""" inv, gw, base, wire, mode = self.fixture(outcome="refused") From a40e6a3f9ba4595f9e38778c8a0c501d96e1d8d3 Mon Sep 17 00:00:00 2001 From: Mark Gascoyne Date: Tue, 13 Oct 2026 08:25:59 +0100 Subject: [PATCH 5/5] fix(gateway): only an explicit Hub refusal turns off an exporting schedule The fail-safe disable fired on any unverified window write, including a slow but successful one (ACK outside the poll, older firmware without an applied block, telemetry not yet landed). That stopped a correctly re-windowed export for a cycle and toggled enable/mode. Take the disable path only when the Hub refused the exact paired command; an unconfirmed write leaves the running export alone and keeps the commit pending for the next cycle. Document that execute.py still reports "Exporting" for a cycle whose window was refused. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_018BaJjLpuMEtQYfvmTRcFJf --- apps/predbat/inverter.py | 23 +++++++++++++++---- .../tests/test_gateway_discharge_window.py | 11 +++++++++ 2 files changed, 29 insertions(+), 5 deletions(-) diff --git a/apps/predbat/inverter.py b/apps/predbat/inverter.py index 88ec9851a..769b6f593 100644 --- a/apps/predbat/inverter.py +++ b/apps/predbat/inverter.py @@ -3001,6 +3001,13 @@ def read_pair(): self.base.record_status(message, had_errors=True) return verified + def gateway_discharge_window_refused(self, new_start, new_end): + """Whether the Hub explicitly refused this exact paired discharge window just now.""" + entities = [self.base.get_arg(name, indirect=False, index=self.id) for name in ("discharge_start_time", "discharge_end_time")] + components = getattr(self.base, "components", None) + gateway = components.get_component("gateway") if components else None + return gateway is not None and gateway.discharge_window_refused(*entities, new_start, new_end) + def adjust_force_export(self, force_export, new_start_time=None, new_end_time=None): """ Adjust force export on/off and set the time window correctly @@ -3105,11 +3112,17 @@ def adjust_force_export(self, force_export, new_start_time=None, new_end_time=No if (new_start and new_start != old_start) or (new_end and new_end != old_end): gateway_window = self.write_and_poll_gateway_discharge_window(new_start or old_start, new_end or old_end) if gateway_window is False: - if force_export and old_discharge_enable and not getattr(self.base, "set_read_only", False): - # The previous window is still armed: leaving it would keep exporting on a - # schedule the plan has moved away from, so stop exporting until a write lands. - # (Without force_export the schedule was already turned off above.) - self.log("Warn: Inverter {} Hub discharge window write failed, turning off scheduled export".format(self.id)) + refused = self.gateway_discharge_window_refused(new_start or old_start, new_end or old_end) + if force_export and old_discharge_enable and refused and not getattr(self.base, "set_read_only", False): + # The Hub refused the move, so the previous window is still armed: leaving it + # would keep exporting on a schedule the plan has moved away from, so stop + # exporting until a write lands. Only an explicit refusal does this - a write + # that is merely not confirmed yet (slow ACK, telemetry lag) may well have + # applied, and the plan still wants to export, so it is left running and the + # pending commit retries it. (Without force_export the schedule was already + # turned off above.) The caller is not told: execute.py still reports + # "Exporting" for this cycle, and the next cycle retries the window. + self.log("Warn: Inverter {} Hub refused the discharge window, turning off scheduled export".format(self.id)) self.write_and_poll_switch("scheduled_discharge_enable", self.base.get_arg("scheduled_discharge_enable", indirect=False, index=self.id), False) self.adjust_inverter_mode(False) self.record_commit("export_window", (new_start, new_end, force_export), False) diff --git a/apps/predbat/tests/test_gateway_discharge_window.py b/apps/predbat/tests/test_gateway_discharge_window.py index e76a77dde..104e04e23 100644 --- a/apps/predbat/tests/test_gateway_discharge_window.py +++ b/apps/predbat/tests/test_gateway_discharge_window.py @@ -158,6 +158,17 @@ def test_refused_window_disables_an_already_exporting_schedule(self): inv.adjust_inverter_mode.assert_called_once_with(False) self.assertTrue(inv.commit_pending["export_window"]) + def test_unconfirmed_move_of_an_exporting_schedule_is_left_running(self): + """Only a refusal disables: an unanswered or incomplete write may have applied.""" + for outcome in ("timeout", "partial", "staged"): + inv, gw, base, wire, mode = self.fixture(outcome=outcome, start="02:00:00", end="04:00:00", enabled="on") + self.apply(inv, start="02:30:00", end="03:00:00") + self.assertEqual([command["command"] for command in wire], ["set_discharge_slot"], outcome) + self.assertEqual(gw.cache[base.ENABLE], "on", outcome) + inv.adjust_inverter_mode.assert_not_called() + self.assertTrue(inv.commit_pending["export_window"], outcome) + self.assertTrue(base.errors, outcome) + def test_applied_move_of_an_exporting_schedule_keeps_it_exporting(self): """The fail-safe disable is only for a failed write: a successful move stays enabled.""" inv, gw, base, wire, mode = self.fixture(start="02:00:00", end="04:00:00", enabled="on")