diff --git a/goodwe_controller/app/ctl_tasks.py b/goodwe_controller/app/ctl_tasks.py new file mode 100644 index 0000000..d1801de --- /dev/null +++ b/goodwe_controller/app/ctl_tasks.py @@ -0,0 +1,242 @@ +"""Controller methods that run as background tasks. + +Moved unchanged from app/main.py (MAINT-21). DECISIONS/MAIN-orchestration.md.""" + +import asyncio +import contextlib +import logging +import time +from datetime import datetime, timedelta, timezone + +from . import messages +from .inverter import UNIT_A +from .price import brussels, price_option +from .static_tariff import resolve_tariff +from .validate import refuse, validate_startup_config, validate_unit_ranges +from .ctl_inputs import TARIFF_MAX_AGE_S, TARIFF_REFRESH_S + +_LOG = logging.getLogger("goodwe") + + +class TasksMixin: + def _grade_store_migration(self, store) -> None: + """DEV-02. Did SAFETY-02's latch actually survive the key rename? + Graded, not performed: a latch that silently fails to move is a + DISARMED SAFETY GUARD on an install that believes it is protected. The + pre-DEV-02 flat value is compared with what the roster came up with; a + disagreement re-arms the latch by hand (the fail-safe direction is ON). + It runs on EVERY start, not only the migrating one: gating on `rep.ran` + swallowed store.py's rule-2 round trip (upgrade, downgrade with a trip + in the window, upgrade; performed on this site 2026-09-02). + DECISIONS/MAIN-orchestration.md.""" + rep = self.store_migration + if rep.ran and not rep.ok: + _LOG.error("The saved safety stops did not copy over cleanly, so " + "they are read again from the older saved state: %s.", + rep.explain()) + data = getattr(store, "data", {}) + legacy_tripped = bool(data.get("sign_tripped", False)) + if not legacy_tripped or not self.inverters: + return + # The question is "is any namespaced latch set THAT A MACHINE READS" + # (SAFETY-44), asked of the STORE's keys, not the roster's objects: + # re-arming A off a flat key that B's trip mirrored would latch a + # healthy inverter on every restart. Three cases: on the roster, stop; + # off-roster named `UNIT_A` (the broken migration's fingerprint), + # repair onto `inverters[0]`; off-roster, other name: stop unless a + # stranded `UNIT_A` latch is present. This guard is the ONLY repair for + # sites upgraded before the fix. Residual: a first unit really named + # `inverter-a`, latched and removed, is re-armed on `inverters[0]` + # (conservative). DECISIONS/MAIN-orchestration.md. + on_roster = {inv.name for inv in self.inverters} + stranded, on_roster_latched, other_off_roster = [], False, False + for k, v in data.items(): + if not (k.startswith("unit/") and k.endswith("/sign_tripped")): + continue + if not bool(v): + continue + # `unit_key()` builds `unit//`, so the unit is field 1: + # split through the writer's own spelling. + unit = k.split("/")[1] + if unit in on_roster: + on_roster_latched = True + elif unit == UNIT_A: + stranded.append(k) + else: + other_off_roster = True + if on_roster_latched or any(inv.sign_wd.tripped + for inv in self.inverters): + return + if other_off_roster and not stranded: + # A removed unit's own latch and no misdirected one: the flat key + # is that unit's mirror, so re-arming `inverters[0]` from it would + # latch a healthy pack. + return + if stranded: + _LOG.error( + "A saved safety stop belongs to %s, which is not in the " + "inverter list %s, so it is set again on the first inverter.", + ", ".join(sorted(stranded)), sorted(on_roster)) + first = self.inverters[0] + # The one state this method exists for. + reason = str(data.get("sign_reason", "") or "") + first.sign_wd.tripped = True + first.sign_wd.reason = reason or ( + "SIGN_DISAGREEMENT: latched before this version; the reason was " + "not carried across the per-inverter store migration") + first.store_set("sign_reason", first.sign_wd.reason) + first.store_set("sign_tripped", True) + _LOG.error( + "%s is set to 0 W again because an older saved safety stop was " + "found, so find out why before you start control: %s.", + first.name, messages.reason_code(first.sign_wd.reason)) + _LOG.debug("Evidence for %s: %s.", first.name, first.sign_wd.reason) + _LOG.debug( + "SAFETY-02 latch guard: the pre-DEV-02 flat key says the " + "sign-disagreement failsafe is latched and no inverter came up " + "latched. Either the per-inverter migration lost it, or an older " + "add-on wrote it during a rollback. Both would be a disarmed " + "guard, so the latch is re-armed on %s.", first.name) + self.log_event("FAILSAFE re-armed from the pre-DEV-02 store keys - " + + first.sign_wd.reason, + hint=messages.hint("stop_restored", unit=first.name)) + + # -- tasks --------------------------------------------------------------- + # -- OPT-04: the background tariff refresh ------------------------------- + def fresh_tariff_plan(self, now): + """The stored plan if it is still current, else None. NEVER a stale + one. THIS IS THE ONE PLACE THE AGE IS JUDGED, and it is public for + `loop_is_following()`'s reason: a check has to be able to assert on the + shipped expression rather than on a transcription that drifts. + (DECISIONS/MAIN-orchestration-archive.md)""" + if self.tariff_plan is None or self.tariff_at is None: + return None + if (now - self.tariff_at).total_seconds() > TARIFF_MAX_AGE_S: + return None + return self.tariff_plan + + async def run_tariff_refresh(self) -> None: + """Resolve the tariff off the control path, forever. NEVER on the tick. + ⚠️ IT CATCHES EVERYTHING AND KEEPS GOING: raising out of here would + freeze the plan, looking current for ever (`fresh_tariff_plan()`'s age + bound is the backstop). ⚠️ A FAILURE DOES NOT CLEAR THE PLAN + (TARIFF-12): a degraded plan replaces a standing observation only once + it has aged out; a §11.4 fallback never replaces a fresh provider plan. + ⚠️ `tariff_at` DOES NOT MOVE ON A KEPT PLAN. + (DECISIONS/MAIN-orchestration-archive.md)""" + horizon = float(price_option(self.o, "price_horizon_h")) + while not self.stopping: + now = datetime.now(timezone.utc) + try: + plan = await resolve_tariff( + self.o, now, now + timedelta(hours=horizon), now=now, + history=self.price_history, watch=self.price_watch) + standing = self.fresh_tariff_plan(now) + # Three terms, and each of them is the only false one in some + # check: the feed answered (replace), nothing current stands + # (take the degradation - this is what makes the age bound + # govern), or what stands is itself a degradation (replace). + keep = (not plan.from_feed + and standing is not None and standing.from_feed) + if keep: + _LOG.warning( + "The price update was incomplete, so the plan from %s " + "is kept until it is %s old. The update held %s.", + messages.when(brussels(self.tariff_at)), + messages.duration(TARIFF_MAX_AGE_S), plan.words) + else: + self.tariff_plan, self.tariff_at = plan, now + _LOG.info("tariff refreshed: %s", plan.words) + except Exception as err: # noqa: BLE001 + # ⚠️ NOT THE OUTAGE PATH - see the docstring: `load_prices()` + # returns on every failure a site can have, so what reaches + # here is unexpected, and the plan is kept on the same terms. + # (DECISIONS/MAIN-orchestration-archive.md) + _LOG.warning( + "The price update failed (%s), so the plan from %s is kept " + "until it is %s old, and trading stops then.", + err, messages.when(brussels(self.tariff_at)), + messages.duration(TARIFF_MAX_AGE_S)) + with contextlib.suppress(asyncio.TimeoutError): + await asyncio.wait_for(self.stop_event.wait(), + timeout=TARIFF_REFRESH_S) + + async def run_control(self) -> None: + # ⚠️ EACH INVERTER'S RATING FROM ITS OWN ENTITY, WHICH IS AC 4 (one + # read of one `setpoint_entity` left a second inverter's clamp, freeze + # rail and `achievable_w` on the first one's numbers). Concurrently: a + # slow ESPHome on one unit must not delay the start on the others. + # (DECISIONS/MAIN-orchestration-archive.md) + async def _limits(inv): + if inv.driver.configured(): + inv.dev_min, inv.dev_max = await self._driver(inv).limits() + await asyncio.gather(*(_limits(inv) for inv in self.inverters)) + entity = self.o.get("setpoint_entity", "") + # ⚠️ THIS USED TO WARN AND CARRY ON, the bug CFG-01 exists to remove: a + # max_w above the device's rating was silently clamped. amain() ran + # this gate already; it runs AGAIN because ESPHome is often not up at + # boot (limits None) and this is the first moment the real rating is + # known; a conflict refuses through the normal shutdown (0 W). ⚠️ THE + # `if entity:` THAT WRAPPED THIS BLOCK FAILED PERMISSIVE (CFG-04's + # hazard): a per-inverter config leaves the flat `setpoint_entity` + # blank, so the re-check was skipped; the per-unit rule now runs + # unconditionally. (DECISIONS/MAIN-orchestration-archive.md) + problems = [] + if entity: + _first = self.inverters[0] + problems += validate_startup_config(self.o, _first.dev_min, + _first.dev_max) + problems += validate_unit_ranges(self.inverters, skip_entity=entity) + if problems: + refuse(problems) + self.refused = problems + self.log_event("REFUSED: " + problems[0], + hint=messages.hint("refused")) + # Signal and return WITHOUT entering the loop: amain() runs the + # wind-down (0 W, or a hand-back: SMA-10) and exits with code 78. + self.stop_event.set() + return + + last_grid = object() + heartbeat = float(self.o.get("heartbeat_s", 10)) + # ⚠️ SAFETY-12. The floor is a DEADLINE ON THE GAP BETWEEN CYCLES, not + # a second scheduler: every cycle resets it, so on a healthy meter it + # never elapses and the per-cycle gain is unchanged. It exists because + # a meter that repeats a value would run NO cycles, and the `inputs + # missing -> 0 W` claim lives inside cycle(). NOT a freeze detector. ⚠️ + # THE 15 s DEFAULT HAS A CEILING IT DOES NOT OWN: it sits under + # validate.FIRMWARE_INGEST_VETO_S (30 s) and `meter_max_age_s` and + # moves with the firmware's veto; the schema minimum of 5 s is the + # reference meter's interval and deserves a second look. + # (DECISIONS/MAIN-orchestration-archive.md) + floor = float(self.o.get("cycle_floor_s", 15)) + last_cycle = None + while not self.stopping: + await self.read_inputs() + now_mono = asyncio.get_running_loop().time() + # A cycle is one meter update, exactly as on the reference install - + # or the floor elapsing, whichever comes first. + overdue = last_cycle is None or (now_mono - last_cycle) >= floor + if self.grid != last_grid or overdue: + last_grid = self.grid + last_cycle = now_mono + # ⚠️ The FULL cycle, not a degraded one: in P1 mode past + # `meter_max_age_s` (30 s) p1.net_w is None ("inputs missing"). + # ⚠️ TRANSPORT-CONDITIONAL: with `meter_source: off`, + # read_inputs() has NO age gate, so a quiet entity reads as + # current; the floor does not make it fresh. The integrator + # advances per CYCLE, so bounding the gap bounds the + # accumulation. (DECISIONS/MAIN-orchestration-archive.md) + await self.cycle() + # ⚠️ The heartbeat is not an optimisation: the ESP32 zeroes the + # inverter after ~30 s of silence, and refreshing the SAME value + # proves we are alive. ⚠️ ONE DEADLINE PER INVERTER, refreshed + # concurrently, each bounded by WRITE_BUDGET_S (budget plus the + # largest `heartbeat_s` stays under layer 1's veto: graded); each + # unit refreshes its own `inv.target`, never the site target. + # (DECISIONS/MAIN-orchestration-archive.md) + await self.send_heartbeats(asyncio.get_running_loop().time(), + heartbeat) + self.publish() + self.loop_pass_mono = time.monotonic() + await asyncio.sleep(1) diff --git a/goodwe_controller/app/main.py b/goodwe_controller/app/main.py index 9af64f2..77d7000 100644 --- a/goodwe_controller/app/main.py +++ b/goodwe_controller/app/main.py @@ -103,6 +103,7 @@ from .watchdog import ( ) from . import setuponly from . import web +from .ctl_tasks import TasksMixin from .ctl_status import StatusMixin from .ctl_alarms import ( AlarmsMixin, HANDBACK_CYCLES, REANNOUNCE_S, ROLLING_BOUND_CYCLES, @@ -174,7 +175,7 @@ def _load_latch(data, now=None): class Controller(InputsMixin, WritesMixin, TradeMixin, CycleMixin, AlarmsMixin, - StatusMixin): + StatusMixin, TasksMixin): def __init__(self, opts: dict, hass: HomeAssistant, store, mqtt_pub, *, stop, peak_db_path: str = "/data/tariff.db", price_history_path: str = "/data/price_history.json", @@ -793,88 +794,6 @@ class Controller(InputsMixin, WritesMixin, TradeMixin, CycleMixin, AlarmsMixin, if hint is not None: self.hint = f"{datetime.now().astimezone():%H:%M} {hint}" - def _grade_store_migration(self, store) -> None: - """DEV-02. Did SAFETY-02's latch actually survive the key rename? - Graded, not performed: a latch that silently fails to move is a - DISARMED SAFETY GUARD on an install that believes it is protected. The - pre-DEV-02 flat value is compared with what the roster came up with; a - disagreement re-arms the latch by hand (the fail-safe direction is ON). - It runs on EVERY start, not only the migrating one: gating on `rep.ran` - swallowed store.py's rule-2 round trip (upgrade, downgrade with a trip - in the window, upgrade; performed on this site 2026-09-02). - DECISIONS/MAIN-orchestration.md.""" - rep = self.store_migration - if rep.ran and not rep.ok: - _LOG.error("The saved safety stops did not copy over cleanly, so " - "they are read again from the older saved state: %s.", - rep.explain()) - data = getattr(store, "data", {}) - legacy_tripped = bool(data.get("sign_tripped", False)) - if not legacy_tripped or not self.inverters: - return - # The question is "is any namespaced latch set THAT A MACHINE READS" - # (SAFETY-44), asked of the STORE's keys, not the roster's objects: - # re-arming A off a flat key that B's trip mirrored would latch a - # healthy inverter on every restart. Three cases: on the roster, stop; - # off-roster named `UNIT_A` (the broken migration's fingerprint), - # repair onto `inverters[0]`; off-roster, other name: stop unless a - # stranded `UNIT_A` latch is present. This guard is the ONLY repair for - # sites upgraded before the fix. Residual: a first unit really named - # `inverter-a`, latched and removed, is re-armed on `inverters[0]` - # (conservative). DECISIONS/MAIN-orchestration.md. - on_roster = {inv.name for inv in self.inverters} - stranded, on_roster_latched, other_off_roster = [], False, False - for k, v in data.items(): - if not (k.startswith("unit/") and k.endswith("/sign_tripped")): - continue - if not bool(v): - continue - # `unit_key()` builds `unit//`, so the unit is field 1: - # split through the writer's own spelling. - unit = k.split("/")[1] - if unit in on_roster: - on_roster_latched = True - elif unit == UNIT_A: - stranded.append(k) - else: - other_off_roster = True - if on_roster_latched or any(inv.sign_wd.tripped - for inv in self.inverters): - return - if other_off_roster and not stranded: - # A removed unit's own latch and no misdirected one: the flat key - # is that unit's mirror, so re-arming `inverters[0]` from it would - # latch a healthy pack. - return - if stranded: - _LOG.error( - "A saved safety stop belongs to %s, which is not in the " - "inverter list %s, so it is set again on the first inverter.", - ", ".join(sorted(stranded)), sorted(on_roster)) - first = self.inverters[0] - # The one state this method exists for. - reason = str(data.get("sign_reason", "") or "") - first.sign_wd.tripped = True - first.sign_wd.reason = reason or ( - "SIGN_DISAGREEMENT: latched before this version; the reason was " - "not carried across the per-inverter store migration") - first.store_set("sign_reason", first.sign_wd.reason) - first.store_set("sign_tripped", True) - _LOG.error( - "%s is set to 0 W again because an older saved safety stop was " - "found, so find out why before you start control: %s.", - first.name, messages.reason_code(first.sign_wd.reason)) - _LOG.debug("Evidence for %s: %s.", first.name, first.sign_wd.reason) - _LOG.debug( - "SAFETY-02 latch guard: the pre-DEV-02 flat key says the " - "sign-disagreement failsafe is latched and no inverter came up " - "latched. Either the per-inverter migration lost it, or an older " - "add-on wrote it during a rollback. Both would be a disarmed " - "guard, so the latch is re-armed on %s.", first.name) - self.log_event("FAILSAFE re-armed from the pre-DEV-02 store keys - " - + first.sign_wd.reason, - hint=messages.hint("stop_restored", unit=first.name)) - # the roster, and the N=1 views of it @property def units(self) -> tuple: @@ -1108,146 +1027,6 @@ class Controller(InputsMixin, WritesMixin, TradeMixin, CycleMixin, AlarmsMixin, return [peakguard.Pack(inv.name, inv.soc, inv.bms_soc_floor) for inv in self.inverters] - # -- tasks --------------------------------------------------------------- - # -- OPT-04: the background tariff refresh ------------------------------- - def fresh_tariff_plan(self, now): - """The stored plan if it is still current, else None. NEVER a stale - one. THIS IS THE ONE PLACE THE AGE IS JUDGED, and it is public for - `loop_is_following()`'s reason: a check has to be able to assert on the - shipped expression rather than on a transcription that drifts. - (DECISIONS/MAIN-orchestration-archive.md)""" - if self.tariff_plan is None or self.tariff_at is None: - return None - if (now - self.tariff_at).total_seconds() > TARIFF_MAX_AGE_S: - return None - return self.tariff_plan - - async def run_tariff_refresh(self) -> None: - """Resolve the tariff off the control path, forever. NEVER on the tick. - ⚠️ IT CATCHES EVERYTHING AND KEEPS GOING: raising out of here would - freeze the plan, looking current for ever (`fresh_tariff_plan()`'s age - bound is the backstop). ⚠️ A FAILURE DOES NOT CLEAR THE PLAN - (TARIFF-12): a degraded plan replaces a standing observation only once - it has aged out; a §11.4 fallback never replaces a fresh provider plan. - ⚠️ `tariff_at` DOES NOT MOVE ON A KEPT PLAN. - (DECISIONS/MAIN-orchestration-archive.md)""" - horizon = float(price_option(self.o, "price_horizon_h")) - while not self.stopping: - now = datetime.now(timezone.utc) - try: - plan = await resolve_tariff( - self.o, now, now + timedelta(hours=horizon), now=now, - history=self.price_history, watch=self.price_watch) - standing = self.fresh_tariff_plan(now) - # Three terms, and each of them is the only false one in some - # check: the feed answered (replace), nothing current stands - # (take the degradation - this is what makes the age bound - # govern), or what stands is itself a degradation (replace). - keep = (not plan.from_feed - and standing is not None and standing.from_feed) - if keep: - _LOG.warning( - "The price update was incomplete, so the plan from %s " - "is kept until it is %s old. The update held %s.", - messages.when(brussels(self.tariff_at)), - messages.duration(TARIFF_MAX_AGE_S), plan.words) - else: - self.tariff_plan, self.tariff_at = plan, now - _LOG.info("tariff refreshed: %s", plan.words) - except Exception as err: # noqa: BLE001 - # ⚠️ NOT THE OUTAGE PATH - see the docstring: `load_prices()` - # returns on every failure a site can have, so what reaches - # here is unexpected, and the plan is kept on the same terms. - # (DECISIONS/MAIN-orchestration-archive.md) - _LOG.warning( - "The price update failed (%s), so the plan from %s is kept " - "until it is %s old, and trading stops then.", - err, messages.when(brussels(self.tariff_at)), - messages.duration(TARIFF_MAX_AGE_S)) - with contextlib.suppress(asyncio.TimeoutError): - await asyncio.wait_for(self.stop_event.wait(), - timeout=TARIFF_REFRESH_S) - - async def run_control(self) -> None: - # ⚠️ EACH INVERTER'S RATING FROM ITS OWN ENTITY, WHICH IS AC 4 (one - # read of one `setpoint_entity` left a second inverter's clamp, freeze - # rail and `achievable_w` on the first one's numbers). Concurrently: a - # slow ESPHome on one unit must not delay the start on the others. - # (DECISIONS/MAIN-orchestration-archive.md) - async def _limits(inv): - if inv.driver.configured(): - inv.dev_min, inv.dev_max = await self._driver(inv).limits() - await asyncio.gather(*(_limits(inv) for inv in self.inverters)) - entity = self.o.get("setpoint_entity", "") - # ⚠️ THIS USED TO WARN AND CARRY ON, the bug CFG-01 exists to remove: a - # max_w above the device's rating was silently clamped. amain() ran - # this gate already; it runs AGAIN because ESPHome is often not up at - # boot (limits None) and this is the first moment the real rating is - # known; a conflict refuses through the normal shutdown (0 W). ⚠️ THE - # `if entity:` THAT WRAPPED THIS BLOCK FAILED PERMISSIVE (CFG-04's - # hazard): a per-inverter config leaves the flat `setpoint_entity` - # blank, so the re-check was skipped; the per-unit rule now runs - # unconditionally. (DECISIONS/MAIN-orchestration-archive.md) - problems = [] - if entity: - _first = self.inverters[0] - problems += validate_startup_config(self.o, _first.dev_min, - _first.dev_max) - problems += validate_unit_ranges(self.inverters, skip_entity=entity) - if problems: - refuse(problems) - self.refused = problems - self.log_event("REFUSED: " + problems[0], - hint=messages.hint("refused")) - # Signal and return WITHOUT entering the loop: amain() runs the - # wind-down (0 W, or a hand-back: SMA-10) and exits with code 78. - self.stop_event.set() - return - - last_grid = object() - heartbeat = float(self.o.get("heartbeat_s", 10)) - # ⚠️ SAFETY-12. The floor is a DEADLINE ON THE GAP BETWEEN CYCLES, not - # a second scheduler: every cycle resets it, so on a healthy meter it - # never elapses and the per-cycle gain is unchanged. It exists because - # a meter that repeats a value would run NO cycles, and the `inputs - # missing -> 0 W` claim lives inside cycle(). NOT a freeze detector. ⚠️ - # THE 15 s DEFAULT HAS A CEILING IT DOES NOT OWN: it sits under - # validate.FIRMWARE_INGEST_VETO_S (30 s) and `meter_max_age_s` and - # moves with the firmware's veto; the schema minimum of 5 s is the - # reference meter's interval and deserves a second look. - # (DECISIONS/MAIN-orchestration-archive.md) - floor = float(self.o.get("cycle_floor_s", 15)) - last_cycle = None - while not self.stopping: - await self.read_inputs() - now_mono = asyncio.get_running_loop().time() - # A cycle is one meter update, exactly as on the reference install - - # or the floor elapsing, whichever comes first. - overdue = last_cycle is None or (now_mono - last_cycle) >= floor - if self.grid != last_grid or overdue: - last_grid = self.grid - last_cycle = now_mono - # ⚠️ The FULL cycle, not a degraded one: in P1 mode past - # `meter_max_age_s` (30 s) p1.net_w is None ("inputs missing"). - # ⚠️ TRANSPORT-CONDITIONAL: with `meter_source: off`, - # read_inputs() has NO age gate, so a quiet entity reads as - # current; the floor does not make it fresh. The integrator - # advances per CYCLE, so bounding the gap bounds the - # accumulation. (DECISIONS/MAIN-orchestration-archive.md) - await self.cycle() - # ⚠️ The heartbeat is not an optimisation: the ESP32 zeroes the - # inverter after ~30 s of silence, and refreshing the SAME value - # proves we are alive. ⚠️ ONE DEADLINE PER INVERTER, refreshed - # concurrently, each bounded by WRITE_BUDGET_S (budget plus the - # largest `heartbeat_s` stays under layer 1's veto: graded); each - # unit refreshes its own `inv.target`, never the site target. - # (DECISIONS/MAIN-orchestration-archive.md) - await self.send_heartbeats(asyncio.get_running_loop().time(), - heartbeat) - self.publish() - self.loop_pass_mono = time.monotonic() - await asyncio.sleep(1) - async def shutdown(self) -> None: """Deterministic wind-down. Do not skip this.""" self.stopping = True