"""The monthly battery maintenance cycle. ⚠️ THIS IS THE PART THE VENDOR CONTROLLER DID THAT NOTHING ELSE REPLACES. A full low -> full cycle each month is a BMS-health behaviour on a 16S LiFePO4 pack: the full charge is when the BMS gets to BALANCE CELLS, the deep discharge is what RECALIBRATES THE COULOMB COUNTER. Dropping it breaks nothing visibly - it degrades the pack over months, and the first symptom is a state-of-charge reading nobody can trust any more. Highest consequence, lowest visibility. OWNERSHIP, not priority. Exactly one thing writes the setpoint at any moment: phase setpoint owned by behaviour idle the control loop normal grid-to-zero drain THIS module forced discharge, exports the surplus charge the control loop charge-only clamp + cheap-window floor hold THIS module 0 W, parked full so the BMS can balance The loop is never "disabled" - it yields. The heartbeat to the inverter keeps running in every phase, so the hardware watchdog stays fed and a crash here still ends with the inverter at 0 W rather than latched. """ import logging from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone _LOG = logging.getLogger("goodwe.maint") IDLE, DRAIN, CHARGE, HOLD = "idle", "drain", "charge", "hold" # A drain of a 13.8 kWh usable pack (300 Ah x 51.2 V, DoD 90) at the shipped # 2000 W is ~6.9 h even with no house load helping - and less in practice, since # house load drains alongside the export and soc_floor stops short of empty. # 12 h means the battery is not following commands - a fault, not a slow day. The charge phase deliberately waits for sun and for cheap hours, so # it is expected to span a night; 36 h still bounds it. DRAIN_TIMEOUT_MIN = 720 CHARGE_TIMEOUT_MIN = 2160 # After an abort, do not immediately re-trigger in the same start hour. RESTART_COOLDOWN_MIN = 90 # ⚠️ CFG-03. THE ONE IN-CODE MIRROR OF config.yaml FOR EVERY `maintenance_*` # OPTION, and the only place in the add-on where one of these numbers is typed. # # WHAT WENT WRONG WITHOUT IT. Until this card each default was written down # three times - config.yaml's `options:` block, an `opts.get(key, )` # fallback in main.py, and this dataclass - and MAINT-04 proved they had already # drifted: config.yaml said 2000, the other two said 2500. An options dict with # `maintenance_charge_w` ABSENT was therefore ACCEPTED by validate.py's rule 6 # (which read 2000) and then BUILT by main.py at 2500, against a 2000 W ceiling # - the exact `lo > hi` contradiction rule 6 exists to refuse, resolving to 0 W # with `contradiction=True`. Unreachable through the Supervisor, which always # supplies every key; fully reachable from sim/inject.py and from the test # suite, which is where anyone debugging maintenance actually is. # # WHERE THE SINGLE SOURCE ACTUALLY IS: config.yaml's `options:` block. It is # what the Supervisor reads and what the operator edits, so it wins by # definition. This dict is a MIRROR of it and not a second opinion: # test_maintenance.py pins every entry by READING the shipped file, so a drift # goes red in CI instead of shipping. That is TARIFF-02's `PRICE_OPTION_DEFAULTS` # and SAFETY-24's `PEAK_OPTION_DEFAULTS` pattern, now applied to the family that # named the card. A mirror rather than a runtime read because the Dockerfile # copies `app/` and `run.sh` into the image and NOT config.yaml - the add-on # cannot open the file it is configured by, so there is nothing to read at 03:00 # on a client's Pi, and a parser that could fail at import is a new way for the # add-on not to start. # # ⚠️ DEFAULTS ONLY. NO BOUNDS LIVE HERE, and that is a decision, not an # omission. The RANGE of each key (`maintenance_charge_w: int(500,5000)`) lives # in config.yaml's `schema:` block and nowhere else, because the Supervisor is # what enforces it and the Supervisor reads only that file. What this module # would gain by mirroring the bounds too is a fifth place to disagree with. # validate.py's rule 6 is NOT a range and does not belong here either: it is a # relation BETWEEN TWO KEYS (`maintenance_charge_w` against `max_w`) that the # schema grammar cannot express at all, which is the whole reason that rule is # written in Python. # # ⚠️ KEYED BY THE OPTION NAME, not by the MaintConfig field name. The dict is # what `from_options()` indexes while reading the operator's dict, so the two # strings have to be the same string; a `charge_w` key here would reintroduce a # transcription step between config.yaml and the reader. MAINT_OPTION_DEFAULTS = { "maintenance_enabled": False, "maintenance_interval_days": 28, "maintenance_start_hour": 10, # ⚠️ 2000, NOT THE 2500 THIS SHIPPED AS, and config.yaml carries the # argument: both rates must stay at or below `max_w` (2000) or the # cheap-window charge shaping asks the arbiter for `hi = -charge_w` against # a supervised `lo = -max_w` and contradicts itself into 0 W. Raising max_w # is what unlocks a faster cycle - raise it FIRST, then these. "maintenance_discharge_w": 2000, "maintenance_charge_w": 2000, "maintenance_soc_floor": 11, "maintenance_soc_target": 99, "maintenance_hold_min": 120, } @dataclass(frozen=True) class MaintConfig: """The maintenance cycle's settings, already parsed. ⚠️ EVERY DEFAULT IS READ OUT OF `MAINT_OPTION_DEFAULTS`, never typed. A literal in this class is what made `MaintConfig()` a THIRD opinion about what a missing key means - see the block above - so `MaintConfig()` and `MaintConfig.from_options({})` and validate.py's fallback are now arithmetically the same number by construction rather than by vigilance. """ enabled: bool = bool(MAINT_OPTION_DEFAULTS["maintenance_enabled"]) interval_days: int = int(MAINT_OPTION_DEFAULTS["maintenance_interval_days"]) start_hour: int = int(MAINT_OPTION_DEFAULTS["maintenance_start_hour"]) discharge_w: float = float(MAINT_OPTION_DEFAULTS["maintenance_discharge_w"]) charge_w: float = float(MAINT_OPTION_DEFAULTS["maintenance_charge_w"]) soc_floor: float = float(MAINT_OPTION_DEFAULTS["maintenance_soc_floor"]) soc_target: float = float(MAINT_OPTION_DEFAULTS["maintenance_soc_target"]) hold_min: int = int(MAINT_OPTION_DEFAULTS["maintenance_hold_min"]) @classmethod def from_options(cls, opts: dict) -> "MaintConfig": """Build from an add-on options dict. ⚠️ THE ONLY PLACE IN THE ADD-ON THAT READS A `maintenance_*` OPTION. main.py used to do this inline as eight `opts.get(key, )` calls, which is eight opportunities to type a number that config.yaml does not ship - and one of them had already been taken. Moving the read next to the mirror means the fallback cannot be anything but the mirror, because there is no dict literal in sight of it to disagree with. ⚠️ `key in opts`, not `opts.get(key) or ...`. A shipped `false` and a shipped `0` are values an operator can legitimately set, and the truthy form would quietly replace both with the default - which is this card's own bug wearing a different hat. """ def _opt(key): return opts[key] if key in opts else MAINT_OPTION_DEFAULTS[key] return cls( enabled=bool(_opt("maintenance_enabled")), interval_days=int(_opt("maintenance_interval_days")), start_hour=int(_opt("maintenance_start_hour")), discharge_w=float(_opt("maintenance_discharge_w")), charge_w=float(_opt("maintenance_charge_w")), soc_floor=float(_opt("maintenance_soc_floor")), soc_target=float(_opt("maintenance_soc_target")), hold_min=int(_opt("maintenance_hold_min")), ) @dataclass class MaintResult: phase: str owns_setpoint: bool setpoint_w: float | None = None charge_only: bool = False events: list = field(default_factory=list) class Maintenance: def __init__(self, cfg: MaintConfig, store): self.cfg = cfg self.store = store # -- state --------------------------------------------------------------- @property def phase(self) -> str: return self.store.data.get("phase", IDLE) def _enter(self, phase: str, now: datetime, events: list, msg: str) -> None: self.store.set("phase", phase) self.store.set_time("phase_started", now) _LOG.info("%s", msg) events.append(msg) def _elapsed_min(self, now: datetime) -> float: started = self.store.get_time("phase_started") return 0.0 if started is None else (now - started).total_seconds() / 60.0 def due(self, now: datetime) -> bool: last = self.store.get_time("last_completed") if last is None: return True return (now - last) >= timedelta(days=self.cfg.interval_days) def next_due(self, now: datetime): last = self.store.get_time("last_completed") return None if last is None else last + timedelta(days=self.cfg.interval_days) # -- operations ---------------------------------------------------------- def force_start(self, now: datetime) -> list: """Start a cycle regardless of the calendar. Commissioning uses this: a cycle must be proven end to end before a site is signed off, and nobody waits a month to find out the schedule never fires. """ events = [] self.store.set_time("last_start_attempt", now) self._enter(DRAIN, now, events, "maintenance: forced start -> drain") return events def abort(self, now: datetime, why: str) -> list: events = [] self._enter(IDLE, now, events, f"maintenance: ABORT ({why})") return events # -- the machine --------------------------------------------------------- def tick(self, now: datetime, soc: float | None) -> MaintResult: events: list = [] phase = self.phase if phase == IDLE: if self.cfg.enabled and now.hour == self.cfg.start_hour and self.due(now): last_try = self.store.get_time("last_start_attempt") cooled = ( last_try is None or (now - last_try) >= timedelta(minutes=RESTART_COOLDOWN_MIN) ) if cooled: self.store.set_time("last_start_attempt", now) self._enter(DRAIN, now, events, "maintenance: due -> drain (exports the surplus)") return MaintResult(DRAIN, True, self.cfg.discharge_w, False, events) return MaintResult(IDLE, False, None, False, events) # ⚠️ Fail toward inaction. Without a state-of-charge reading none of the # transitions below mean anything, and a stale "100 %" would end the # charge phase having charged nothing. Hold the current phase's command # and wait; the phase timeout is the backstop. if soc is None: _LOG.warning("maintenance: no SoC reading, holding phase %s", phase) if phase == DRAIN: return MaintResult(DRAIN, True, self.cfg.discharge_w, False, events) if phase == HOLD: return MaintResult(HOLD, True, 0.0, False, events) return MaintResult(CHARGE, False, None, True, events) elapsed = self._elapsed_min(now) if phase == DRAIN: if soc <= self.cfg.soc_floor: self._enter(CHARGE, now, events, f"maintenance: drain complete at {soc:.0f} % " f"after {elapsed:.0f} min -> charge") return MaintResult(CHARGE, False, None, True, events) if elapsed > DRAIN_TIMEOUT_MIN: events += self.abort(now, f"drain stuck at {soc:.0f} % after {elapsed:.0f} min") return MaintResult(IDLE, True, 0.0, False, events) return MaintResult(DRAIN, True, self.cfg.discharge_w, False, events) if phase == CHARGE: if soc >= self.cfg.soc_target: self._enter(HOLD, now, events, f"maintenance: charge complete at {soc:.0f} % after " f"{elapsed:.0f} min -> holding {self.cfg.hold_min} min to balance") return MaintResult(HOLD, True, 0.0, False, events) if elapsed > CHARGE_TIMEOUT_MIN: events += self.abort(now, f"charge stuck at {soc:.0f} % after {elapsed:.0f} min") return MaintResult(IDLE, True, 0.0, False, events) return MaintResult(CHARGE, False, None, True, events) if phase == HOLD: if elapsed >= self.cfg.hold_min: self.store.set_time("last_completed", now) self._enter(IDLE, now, events, f"maintenance: cycle COMPLETE ({elapsed:.0f} min hold) " "- handing back to the loop") return MaintResult(IDLE, False, None, False, events) # 0 W, not "do nothing": commanding zero is what stops the loop # pulling the pack straight back down the moment it is full, which # is exactly when the BMS needs it parked at the top. return MaintResult(HOLD, True, 0.0, False, events) _LOG.error("maintenance: unknown phase %r, forcing idle", phase) events += self.abort(now, f"unknown phase {phase!r}") return MaintResult(IDLE, True, 0.0, False, events)