goodwe-addon/goodwe_controller/app/maintenance.py

"""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.

ONE PACK AT A TIME (MAINT-05). There is now one `Maintenance` per roster entry,
each with its own schedule, its own phase and its own history under its own
Store keys - and therefore, for the first time, two of them able to move power
on ONE grid connection at once. Read `arbitrate()` before touching any of it:
the site holds exactly ONE maintenance slot, a pack that is not IDLE holds it,
and every other pack is held at IDLE under the loop until it is released. That
is not a scheduling nicety. A drain of pack A against a charge of pack B is two
power-moving phases pulling opposite ways across the meter - the circulating
power SAFETY-31 exists to DETECT, arriving here as something this module can
simply not do.
"""

import logging
import math
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone, tzinfo

from .store import MAINT_UNIT_KEYS, unit_key

_LOG = logging.getLogger("goodwe.maint")

IDLE, DRAIN, CHARGE, HOLD = "idle", "drain", "charge", "hold"

# ⚠️ THE PHASES THAT MOVE POWER, WHICH IS NOT THE SAME LIST AS "owns the
# setpoint". The table above is about WHO WRITES the number; this is about
# whether the pack is being pushed. DRAIN and HOLD own the setpoint. CHARGE does
# NOT - the loop owns it - and CHARGE is in here anyway, because the charge-only
# clamp plus the cheap-window floor is what pulls a second pack full while the
# first is being emptied. Leaving CHARGE out is precisely the interlock hole
# this card is about: pack A charging from the grid at 2000 W while pack B
# drains 2000 W to export is 4 kW going round in a circle with the meter at
# zero, and both packs believing they are behaving.
#
# HOLD commands 0 W and moves nothing, and it is in here for the other half of
# the sentence: it OWNS the actuator - the pack is parked at the top for the BMS
# and is deliberately not available to the loop - so a second cycle starting
# under it would still be two recalibrations at once, which is the thing whose
# value is a clean uninterrupted coulomb count.
POWER_MOVING = (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
# MAINT-13: minutes the pack was ALLOWED to charge (the price gate open), not
# wall-clock minutes since the charge began. See CHARGE_WALL_TIMEOUT_MIN.
CHARGE_TIMEOUT_MIN = 2160
# After an abort, do not immediately re-trigger in the same start hour.
RESTART_COOLDOWN_MIN = 90

# MAINT-13 (DECISIONS/MAINT-13-full-pack-is-complete.md). Constants, not options.
# A pack that reaches target - 3 and then takes no more charge for this many
# minutes of allowed charging counts as full.
CHARGE_NEAR_FULL_PCT = 3
CHARGE_STALL_MIN = 60
# Wall-clock backstop for a charge whose gate never opens (no cheap hours, no sun).
CHARGE_WALL_TIMEOUT_MIN = 10080
# A failed cycle (stuck drain or charge) is not retried before this many days.
FAIL_BACKOFF_DAYS = 7
# One tick counts at most this many minutes: a gap means the add-on was down.
CHARGE_TICK_MAX_MIN = 5
# The charge accounting is written to the state file at most this often.
CHARGE_SAVE_MIN = 5

# ⚠️ 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, <literal>)`
# 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,
    # MAINT-10: a LOCAL hour (site_zone() below), 19:00 so the drain runs
    # into the night rather than through the solar day.
    "maintenance_start_hour": 19,
    # ⚠️ 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, <literal>)`
        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")),
        )


# --------------------------------------------------------------------------- #
# MAINT-10: the start hour is local time. See DECISIONS/MAINT-10-local-start-hour.md.
# --------------------------------------------------------------------------- #
def _load_zone(name):
    """A tzinfo for an IANA name, or `(None, why)`."""
    if not name or not isinstance(name, str):
        return None, "none given"
    try:
        from zoneinfo import ZoneInfo
        return ZoneInfo(name.strip()), ""
    except Exception as err:                      # noqa: BLE001 - any failure
        return None, f"{name!r} is not a known zone ({type(err).__name__})"


def site_zone(ha_zone, ha_why: str, env_tz) -> tuple:
    """`(tzinfo, name, source, why)`: Home Assistant's zone (source "ha"), else
    the container's TZ ("tz"), else UTC ("utc"). `why` says what was missing."""
    tz, why = _load_zone(ha_zone)
    if tz is not None:
        return tz, ha_zone, "ha", ""
    ha_why = ha_why or why
    tz, env_why = _load_zone(env_tz)
    if tz is not None:
        return tz, env_tz, "tz", ha_why
    return timezone.utc, "UTC", "utc", f"Home Assistant: {ha_why}, TZ: {env_why}"


def in_start_slot(now: datetime, start_hour: int, tz: tzinfo) -> bool:
    """Is `now` inside today's one-hour start slot, read on the local clock?

    Normally the slot is the local hour `start_hour`. On the day the clocks go
    forward and that hour does not exist, it is the first hour after the jump.
    On the day they go back and it happens twice, only the first one counts.
    Worked from UTC instants only (`astimezone`), so it does not depend on how
    a tzinfo resolves a wall time that is missing or doubled.
    """
    if now.tzinfo is None:
        now = now.replace(tzinfo=timezone.utc)
    here = now.astimezone(tz).replace(tzinfo=None)
    before = (now - timedelta(hours=1)).astimezone(tz).replace(tzinfo=None)
    if here.hour == start_hour:
        # the second pass through a repeated hour: an hour ago read the same
        return not (before.date() == here.date() and before.hour == start_hour)
    slot = here.replace(hour=start_hour, minute=0, second=0, microsecond=0)
    # the clock jumped over the start of the slot within the last hour
    return before < slot and here >= slot + timedelta(hours=1)


def next_slot_start(now: datetime, start_hour: int, tz: tzinfo):
    """The start of the slot `now` is in, or else of the next one; None if
    there is none within 50 h (no real zone does that)."""
    step = timedelta(minutes=15)
    t = now.replace(minute=now.minute - now.minute % 15, second=0,
                    microsecond=0) - timedelta(hours=1)
    for _ in range(4 * 50):
        if (in_start_slot(t, start_hour, tz)
                and not in_start_slot(t - step, start_hour, tz)
                and t + timedelta(hours=1) > now):
            return t
        t += step
    return None


@dataclass
class MaintResult:
    phase: str
    owns_setpoint: bool
    setpoint_w: float | None = None
    charge_only: bool = False
    events: list = field(default_factory=list)
    # ⚠️ WHY A DEFERRAL IS A FIELD AND NOT JUST A LOG LINE. A pack that is due
    # and is not starting looks, from every surface an operator has, exactly
    # like a pack whose schedule has quietly stopped firing - §12's opening
    # sentence, the failure this whole module exists because of. It is carried
    # so the status page can SAY "waiting for the other pack" rather than
    # showing an idle machine and a due date in the past. Empty on every path
    # that is not deferring, so a roster of one never sees it.
    deferred: str = ""
    # MAINT-13: "drain" or "charge" on the one tick where a cycle FAILED, so
    # the Controller can tell the operator once. Empty on every other tick.
    failed: str = ""


class UnitHistory:
    """One pack's maintenance history in the Store, under one spelling.

    ⚠️ THE NAMESPACE IS A CONSTRUCTOR ARGUMENT AND `None` IS A REAL VALUE, NOT
    AN OMISSION. `unit=None` means the FLAT keys - `phase`, `phase_started`,
    `last_completed`, `last_start_attempt` - which is what every install has
    written since the add-on shipped and what a one-inverter site keeps writing
    after MAINT-05. §6.2's identity: with one pack there is nothing for a
    namespace to distinguish it FROM, and adding one would change the layout of
    a live state file to say something it already said. `arbiter.actuator_for()`
    makes the same move for the same reason and its docstring argues it.

    ⚠️ `mirror_flat` IS THE ROLLBACK VIEW AND IT IS ONLY EVER THE FIRST PACK'S.
    At N>=2 pack A's history has been migrated into its namespace and is written
    there; the flat keys are ALSO kept in step, so an operator who rolls the
    add-on back to a single-FSM version finds a CURRENT history rather than one
    dated the day the second inverter was declared. store.migrate_maint_keys()
    carries the argument, and it is DEV-02's rule 2 applied to history instead
    of to a latch: "left in the file" is not the same promise as "still true".
    Pack B has no flat spelling to mirror to - the flat keys mean pack A on
    every install that has them - so it is constructed with mirroring off, and
    a mirror from B would OVERWRITE A's rollback view with B's phase.
    """

    def __init__(self, store, unit: str | None = None, mirror_flat: bool = False):
        self.store = store
        self.unit = unit
        self.mirror_flat = bool(mirror_flat)

    def key(self, name: str) -> str:
        """The Store key this pack's `name` lives under."""
        return name if self.unit is None else unit_key(self.unit, name)

    def get(self, name: str, default=None):
        return getattr(self.store, "data", {}).get(self.key(name), default)

    def set(self, name: str, value) -> None:
        self.store.set(self.key(name), value)
        if self.mirror_flat:
            self.store.set(name, value)

    def get_time(self, name: str):
        return self.store.get_time(self.key(name))

    def set_time(self, name: str, when) -> None:
        self.store.set_time(self.key(name), when)
        if self.mirror_flat:
            self.store.set_time(name, when)


class Maintenance:
    """ONE pack's recalibration cycle: its schedule, its phase, its history.

    ⚠️ ONE PER ROSTER ENTRY SINCE MAINT-05, AND THE SLOT IS NOT ITS OWN. This
    object decides WHEN ITS PACK would like to start; `arbitrate()` decides
    which of them may, because the constraint is a property of the grid
    connection and no single FSM can see it. Never call `tick()` in a loop
    without passing the `blocked_by` that `arbitrate()` returned for it: that is
    the two-packs-at-once bug with the interlock present but not wired.
    """

    def __init__(self, cfg: MaintConfig, store, unit: str | None = None,
                 name: str = "", mirror_flat: bool = False,
                 tz: tzinfo | None = None):
        self.cfg = cfg
        # MAINT-10: the zone `cfg.start_hour` is read in - main.py resolves it
        # once at startup (`site_zone()`). UTC when nobody says, as before.
        self.tz = tz or timezone.utc
        # ⚠️ `self.store` IS THE VIEW, NOT THE Store. Everything in this class
        # names an UNQUALIFIED key - "phase", "last_completed" - and the view is
        # the one place that turns it into the key this pack owns. A raw
        # `store.set("phase", ...)` anywhere below would write pack A's history
        # from pack B's state machine, which is the shared-state bug this card
        # is about, and there would be nothing to see at a roster of one.
        self.store = UnitHistory(store, unit, mirror_flat)
        self.unit = unit
        # What an event or a refusal calls this pack. At a roster of one it is
        # never rendered - a single machine cannot be deferred - so the shipped
        # one-inverter reason strings are untouched.
        self.name = str(name or unit or "the pack")
        # Edge-memo for the deferral note. The action log holds 40 lines and a
        # deferral lasts as long as the other pack's cycle - hours - so a line
        # per cycle would flush the very context that explains it. Same pattern
        # as `Controller._arb_action` and `_maint_shape_suspended`.
        self._deferred = ""
        # MAINT-13: the charge accounting (see _charge_account), in memory and
        # written to the state file every CHARGE_SAVE_MIN minutes.
        self._trk = None
        self._trk_saved = None

    # -- state ---------------------------------------------------------------
    @property
    def phase(self) -> str:
        return self.store.get("phase", IDLE)

    def _enter(self, phase: str, now: datetime, events: list, msg: str) -> None:
        self._trk = None
        if self.store.get("charge_track") is not None:
            self.store.set("charge_track", 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)

    def waited_days(self, now: datetime) -> float:
        """Days since this pack last COMPLETED a cycle; `inf` if it never has.

        ⚠️ THE ARBITRATION KEY, AND IT IS DELIBERATELY NOT `due()`. `due()` is a
        yes/no and two packs that are both due are indistinguishable by it -
        which is how a roster-order tie-break gets written and how the first
        entry then wins every contested month forever while the second drifts.
        This is an ORDERING: the pack whose coulomb counter has gone longest
        without a recalibration goes first. `inf` for never-recalibrated is the
        same statement `due()` makes about an absent `last_completed`, kept as
        one so a fresh pack cannot lose to one that ran last month.
        """
        last = self.store.get_time("last_completed")
        return math.inf if last is None else (now - last).total_seconds() / 86400.0

    def _cooled(self, now: datetime) -> bool:
        """Is the post-abort cooldown over?"""
        last_try = self.store.get_time("last_start_attempt")
        return (last_try is None
                or (now - last_try) >= timedelta(minutes=RESTART_COOLDOWN_MIN))

    def backoff_until(self, now: datetime):
        """MAINT-13: when a failed cycle may be tried again, or None.

        None when nothing failed, when the back-off is over, or when a cycle
        completed after the failure (a forced start that worked ends it).
        """
        failed = self.store.get_time("last_failed")
        if failed is None:
            return None
        done = self.store.get_time("last_completed")
        if done is not None and done >= failed:
            return None
        until = failed + timedelta(days=FAIL_BACKOFF_DAYS)
        return until if now < until else None

    def wants_start(self, now: datetime) -> bool:
        """Would this pack begin a cycle on this tick if the site let it?

        ⚠️ THE SAME PREDICATE `tick()` USES, NOT A COPY OF IT. `arbitrate()` has
        to know who is asking before anybody moves, and a second transcription
        of "enabled, in the start hour, due and cooled down" is a second opinion
        about who was asking - so the IDLE branch of `tick()` calls this one.
        """
        return (self.phase == IDLE and self.cfg.enabled
                and in_start_slot(now, self.cfg.start_hour, self.tz)
                and self.due(now)
                and self._cooled(now)
                and self.backoff_until(now) is None)

    # -- operations ----------------------------------------------------------
    def force_start(self, now: datetime, blocked_by: str = "") -> 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.

        ⚠️ REGARDLESS OF THE CALENDAR, NOT REGARDLESS OF THE INTERLOCK. The
        calendar is a preference and an operator may overrule it; "one pack
        moves power at a time" is a property of the grid connection and an
        operator pressing a button does not add a second one. A forced start
        against a busy site is REFUSED, by name, and the phase machine does not
        move - the refusal is returned as an event so the operator reads it in
        the action log rather than watching nothing happen.
        """
        if blocked_by:
            msg = (f"maintenance: {self.name} cannot start now, because "
                   f"{blocked_by}")
            _LOG.warning("%s", msg)
            return [msg]
        events = []
        self.store.set_time("last_start_attempt", now)
        self._enter(DRAIN, now, events, "maintenance: forced start -> drain")
        return events

    def _charge_account(self, now: datetime, soc: float, gate_open: bool) -> dict:
        """MAINT-13: what this charge has used so far, as minutes of ALLOWED time.

        `open_min` counts only ticks with the price gate open; `flat_min` counts
        gate-open minutes since the SoC last beat its best; a closed gate pauses
        both. One tick adds at most CHARGE_TICK_MAX_MIN, so a gap (add-on down)
        counts for nothing. A state file from before MAINT-13, or from another
        charge, starts from the wall-clock minutes so far: the old behaviour.
        """
        since = self.store.get("phase_started")
        trk = self._trk
        if trk is None:
            raw = self.store.get("charge_track")
            if isinstance(raw, dict) and raw.get("since") == since:
                trk = dict(raw)
            else:
                trk = {"since": since, "open_min": self._elapsed_min(now),
                       "flat_min": 0.0, "best": float(soc), "at": None}
            self._trk = trk
        at = trk.get("at")
        try:
            last = datetime.fromisoformat(at) if at else None
        except (TypeError, ValueError):
            last = None
        dt = 0.0 if last is None else min(
            CHARGE_TICK_MAX_MIN, max(0.0, (now - last).total_seconds() / 60.0))
        trk["at"] = now.isoformat()
        if soc > trk["best"]:
            trk["best"], trk["flat_min"] = float(soc), 0.0
        elif gate_open:
            trk["flat_min"] += dt
        if gate_open:
            trk["open_min"] += dt
        saved = self._trk_saved
        if saved is None or (now - saved).total_seconds() >= CHARGE_SAVE_MIN * 60:
            self.store.set("charge_track", dict(trk))
            self._trk_saved = now
        return trk

    def _fail(self, now: datetime, kind: str, why: str) -> MaintResult:
        """MAINT-13: stand down, remember the failure, and tell the Controller."""
        events = self.abort(now, why)
        self.store.set_time("last_failed", now)
        msg = (f"maintenance: the cycle did not finish. It waits "
               f"{FAIL_BACKOFF_DAYS} days before it tries again.")
        _LOG.warning("%s", msg)
        events.append(msg)
        return MaintResult(IDLE, True, 0.0, False, events, failed=kind)

    def abort(self, now: datetime, why: str) -> list:
        events = []
        self._enter(IDLE, now, events, f"maintenance: ABORT ({why})")
        return events

    def _defer_note(self, why: str, events: list) -> list:
        """Say once, not every second, that this pack is waiting its turn.

        ⚠️ THE WORDING IS DELIBERATELY NOT AN ABORT AND NOT A TIMEOUT. §12's
        notes are explicit that a 12 h phase timeout means "the battery is not
        following commands - a FAULT, not a slow day", and the one thing that
        must not happen here is an operator reading a deferral as that. This
        line says the pack is WAITING, names the pack it is waiting for, and
        says it will run next - none of which any failure path in this module
        says.
        """
        if self._deferred != why:
            self._deferred = why
            msg = (f"maintenance: {self.name} is due and waits its turn, "
                   f"because {why}. It starts at the next start hour once the "
                   f"other battery is done. This is not a stalled phase.")
            _LOG.info("%s", msg)
            events.append(msg)
        return events

    # -- the machine ---------------------------------------------------------
    def tick(self, now: datetime, soc: float | None,
             blocked_by: str = "", gate_open: bool = True) -> MaintResult:
        """One cycle of this pack's state machine.

        `blocked_by` is `arbitrate()`'s answer for THIS pack: empty when the
        site's one maintenance slot is available to it, otherwise the reason it
        is not. It gates two things and nothing else - a start, and a phase that
        is already moving power on a pack that does not hold the slot.

        `gate_open` (MAINT-13) is whether the pack may charge now (the price
        gate); only those minutes count towards the charge timeout.
        """
        events: list = []
        phase = self.phase

        # ⚠️ THE SLOT IS CHECKED BEFORE THE PHASE, AND THIS BRANCH IS A RECOVERY
        # RATHER THAN AN EXPECTED PATH. `arbitrate()` grants the slot to the
        # pack that is already mid-cycle, so a pack that is BOTH moving power
        # and blocked means two histories on disk say "drain" - a hand-edited
        # state file, or a store carried across a version that had no interlock.
        # The fail-safe direction is unambiguous: two packs pulling against each
        # other on one connection is the fault SAFETY-31 latches the whole site
        # for. Stand this one down to 0 W and say why; the holder keeps its
        # cycle, which is the outcome that loses no coulomb count at all.
        if blocked_by and phase in POWER_MOVING:
            events += self.abort(now, f"interlock - {blocked_by}")
            return MaintResult(IDLE, True, 0.0, False, events)

        if phase == IDLE:
            if self.wants_start(now):
                if blocked_by:
                    # ⚠️ `last_start_attempt` IS NOT STAMPED HERE, AND THAT IS
                    # THE STARVATION FIX RATHER THAN AN OMISSION. That key is
                    # the post-ABORT cooldown: it exists so a cycle that failed
                    # does not re-trigger inside the same start hour. A deferral
                    # is not an attempt - nothing was tried and nothing failed -
                    # and stamping it would put this pack under a 90 minute
                    # cooldown for a start hour that is 60 minutes long, so the
                    # pack that lost the arbitration would still be locked out
                    # at the moment the winner released the slot.
                    #
                    # ⚠️ AND THE PHASE DOES NOT MOVE, WHICH IS THE OTHER HALF.
                    # `phase_started` is untouched, so the 12 h DRAIN and 36 h
                    # CHARGE timeouts - which mean "the battery is not following
                    # commands, a fault" - cannot be reached by waiting. A
                    # deferred pack is IDLE under the loop, not a stuck one.
                    return MaintResult(IDLE, False, None, False,
                                       self._defer_note(blocked_by, events),
                                       blocked_by)
                self._deferred = ""
                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)
            self._deferred = ""
            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:
                return self._fail(now, "drain", f"drain stuck at {soc:.0f} % "
                                                f"after {elapsed:.0f} min")
            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)
            trk = self._charge_account(now, soc, gate_open)
            near = soc >= self.cfg.soc_target - CHARGE_NEAR_FULL_PCT
            # MAINT-13 rule 1: close to full and no longer taking charge.
            if near and trk["flat_min"] >= CHARGE_STALL_MIN:
                self._enter(HOLD, now, events,
                            f"maintenance: the battery takes no more charge at "
                            f"{soc:.0f} %, so the charge counts as done -> holding "
                            f"{self.cfg.hold_min} min to balance")
                return MaintResult(HOLD, True, 0.0, False, events)
            if (trk["open_min"] > CHARGE_TIMEOUT_MIN
                    or elapsed > CHARGE_WALL_TIMEOUT_MIN):
                # MAINT-13 rule 2: out of time close to full is complete.
                if near:
                    self._enter(HOLD, now, events,
                                f"maintenance: charge time is up at {soc:.0f} %, "
                                f"close enough to full, so the charge counts as "
                                f"done -> holding {self.cfg.hold_min} min to balance")
                    return MaintResult(HOLD, True, 0.0, False, events)
                return self._fail(now, "charge",
                                  f"charge stuck at {soc:.0f} % after "
                                  f"{trk['open_min']:.0f} min of charging time")
            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)


def _days(n: float) -> str:
    return "never recalibrated" if n == math.inf else f"{n:.0f} d since its last cycle"


def arbitrate(machines, now: datetime) -> list:
    """Grant the site's ONE maintenance slot. Returns a list parallel to
    `machines`: "" for the pack that may move power, otherwise the reason the
    pack may not.

    ⚠️ THE SLOT IS A PROPERTY OF THE GRID CONNECTION, WHICH IS WHY IT IS NOT ON
    `Maintenance`. Each FSM can see its own pack's schedule and nothing else; the
    quantity being rationed - "how many phases are moving power across one
    meter" - is a sum over the roster, exactly like `sum(|B_i|) - |sum(B_i)|` in
    SAFETY-31 and for the same reason. A per-pack answer to a site-level
    question is the shape of defect this repository keeps finding.

    ⚠️ IT IS DERIVED, NOT PERSISTED, AND THAT IS DELIBERATE. The holder is
    "whichever pack's stored phase is not IDLE" - a fact already on disk, in the
    per-pack history the packs themselves write. A separate `maint_slot_holder`
    key would be a second opinion about the same thing, and the two would
    disagree the first time the add-on was killed between writing the phase and
    writing the holder. Restart-safe by construction: a controller that comes
    back up mid-drain reads the drain and re-derives the holder from it.

    TWO CLAUSES, IN ORDER:

    1. A PACK ALREADY MOVING POWER KEEPS THE SLOT. Nothing preempts a cycle in
       flight, because a recalibration interrupted halfway is worth less than
       nothing - the coulomb counter is graded on ONE uninterrupted drain and
       refill, so a cycle stopped at 40 % has spent the energy and bought no
       calibration. If more than one pack somehow comes up moving power, the
       FIRST in roster order holds and `tick()` stands the others down; see the
       recovery branch there.

    2. OTHERWISE THE LONGEST-WAITING CANDIDATE STARTS. `wants_start()` is asked
       of every pack, and among those that say yes the slot goes to the one with
       the largest `waited_days()`. NOT roster order: two packs on one monthly
       schedule are due in the same hour every month, and a roster-order tie
       would hand the slot to `inverters[0]` every single time while the second
       pack's coulomb counter drifted forever - which is, in a different
       costume, the exact defect this card was opened for. Roster order breaks a
       genuine numeric tie and nothing else.

    ⚠️ AT A ROSTER OF ONE THIS IS THE IDENTITY FUNCTION. One machine is never
    another machine's blocker: clause 1 finds at most itself and clause 2 has at
    most one candidate, so it returns `[""]` on every tick and no reason string
    it can build is ever rendered. That is what makes the one-inverter behaviour
    byte-identical rather than merely similar.
    """
    blocks = [""] * len(machines)
    busy = [(i, m) for i, m in enumerate(machines) if m.phase in POWER_MOVING]
    if busy:
        held, holder = busy[0]
        for i, m in enumerate(machines):
            if i != held:
                blocks[i] = (f"{holder.name} is mid-cycle ({holder.phase}) and "
                             f"holds the site's one maintenance slot")
        return blocks

    wants = [(i, m) for i, m in enumerate(machines) if m.wants_start(now)]
    if len(wants) <= 1:
        return blocks
    # Largest waited_days wins; `-index` makes a numeric tie fall to roster
    # order rather than to whatever `max()` happens to see first.
    won, winner = max(wants, key=lambda p: (p[1].waited_days(now), -p[0]))
    for i, m in wants:
        if i != won:
            blocks[i] = (f"{winner.name} has waited longer "
                         f"({_days(winner.waited_days(now))} against "
                         f"{_days(m.waited_days(now))}) and takes the site's "
                         f"one maintenance slot first")
    return blocks