"""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, see # DECISIONS/SMA-10-two-idle-states.md) 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)