goodwe-addon/goodwe_controller/test_delivery.py

"""Runnable check for SAFETY-42 - a unit that stops delivering its share.
`python3 test_delivery.py`

⚠️ THE FIXTURE IS THE CARD, AND IT IS PART 4. What this guard has to do is tell
a WEDGED unit from a COLD PACK AT ITS FLOOR, and those two are IDENTICAL in the
power domain - the same shares, the same measured watts, the same meter, the
same `saturated-freeze`. Measured on admin/goodwe's two-master rig over a
120-cycle run: the two arms agree on every one of 120 cycles in (grid, site
target, per-unit share, per-unit measured power), to the last emitted watt. If
this file's two fixtures ever stop differing, the check has stopped being
graded on the only thing that separates the two cases.

⚠️ MUTATING THE MODULE THIS FILE GRADES? PURGE `__pycache__` FIRST - or run
`tools/mutation_run.sh`, which does. `python3 -B` alone does NOT save you: it
stops bytecode being WRITTEN, not a stale one being READ. That is TEST-15, and
it has already turned one live mutation into a false survivor. Read MUTATION.md
before you report a survivor.
"""

import asyncio
import io
import re
import sys
from datetime import datetime, timedelta, timezone

from app.control import Tuning
from app.delivery import (
    DELIVERY_OPTION_DEFAULTS, REASON_CODE, DeliveryConfig, DeliveryWatchdog,
    UnitDelivery, account, delinquent, short_by,
)
from app.inverter import Inverter
from app.main import Controller
from app.peakguard import discharge_permitted
from app.store import unit_key

fails = []
T0 = datetime(2026, 9, 2, 12, 0, 0, tzinfo=timezone.utc)


def check(name, cond):
    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
    if not cond:
        fails.append(name)


def U(name, share, meas, why=None):
    return UnitDelivery(name, share, meas, why)


# ===========================================================================
print("=" * 72)
print("1. the quantity, and the reduction that makes it readable")
print("=" * 72)

check("short_by is the watts of a DISCHARGE share that did not arrive",
      short_by(2000, 0) == 2000.0 and short_by(2000, 1400) == 600.0)
check("...delivering the share, or more than it, is not a shortfall - both "
      "answer None rather than a number a caller would have to sign-test",
      short_by(2000, 2000) is None and short_by(2000, 2600) is None)
check("...a CHARGE share answers None. There is no configured charge ceiling "
      "in this add-on, so a pack in constant-voltage taper cannot be explained "
      "and arming it would fire on every maintenance charge",
      short_by(-2000, 0) is None and short_by(-2000, -100) is None)
check("...a missing reading is None, never 'the whole share'. Zero would be "
      "this check inventing a fault out of an absence",
      short_by(2000, None) is None and short_by(None, 0) is None)

print()
print("   `account()` - the ONLY thing that separates a wedged unit from a")
print("   cold pack, because nothing in the power domain does")
check("a pack at its own cut-out accounts for its shortfall",
      account(live=True, parked=False, soc=10.0, bms_floor=10.0) is not None)
check("...and a pack above it does not",
      account(live=True, parked=False, soc=10.1, bms_floor=10.0) is None)
check("...the boundary is `<=`, the same one discharge_permitted() draws, "
      "BECAUSE IT IS discharge_permitted() - not a transcription of it",
      (account(live=True, parked=False, soc=10.0, bms_floor=10.0) is None)
      == (discharge_permitted(10.0, 10.0) is None)
      and (account(live=True, parked=False, soc=10.1, bms_floor=10.0) is None)
      == (discharge_permitted(10.1, 10.0) is None))
check("a unit the allocator has already dropped accounts for itself - firing "
      "inside a state already being handled is the SAFETY-15/19 failure",
      account(live=False, parked=False, soc=50.0, bms_floor=10.0) is not None)
check("a unit parked at 0 W by a sign trip accounts for itself: not "
      "delivering is compliance",
      account(live=True, parked=True, soc=50.0, bms_floor=10.0) is not None)
check("EVERY UNKNOWN FAILS TOWARD SILENCE - an unreadable SoC and an unstated "
      "floor both account for the shortfall, which is the OPPOSITE direction "
      "to discharge_permitted()'s own caller and is the right one here: an "
      "alarm nobody can justify is the crying-wolf failure this file exists "
      "to avoid",
      account(live=True, parked=False, soc=None, bms_floor=10.0) is not None
      and account(live=True, parked=False, soc=50.0, bms_floor=None) is not None)
print(f"       e.g. {account(live=True, parked=False, soc=9.0, bms_floor=10.0)!r}")

print()
print("   the PROPER-SUBSET rule - what keeps a site-wide cause out")
_wedged = delinquent([U("a", 2000, 2000), U("b", 2000, 0)], 500)
check("one unit short and unexplained beside a sibling that is not: NAMED",
      list(_wedged) == ["b"])
check("...and the sentence names the unit, the two watt figures and the peer, "
      "because 'B is not delivering' and 'B is not delivering while A is' are "
      "different facts and only the second one is this check's conclusion",
      REASON_CODE in _wedged["b"] and "+2000 W" in _wedged["b"]
      and "+0 W" in _wedged["b"] and "a" in _wedged["b"])
print(f"       {_wedged['b']}")
_cold = delinquent([U("a", 2000, 2000), U("b", 2000, 0, "its pack is empty")], 500)
check("THE CARD: the same numbers, with the shortfall ACCOUNTED FOR, name "
      "nobody. This pair of assertions is the whole guard",
      _cold == {})
_both = delinquent([U("a", 2000, 0), U("b", 2000, 0)], 500)
check("EVERY asked unit short and unexplained: NOBODY is named. That is a "
      "site condition - a derate, a curtailment, a plant that cannot - and "
      "this check has no instrument that tells it from every box being wedged "
      "at once. Reporting it would be the saturation-shaped alarm again",
      _both == {})
_three = delinquent([U("a", 2000, 2000), U("b", 2000, 0), U("c", 2000, 0)], 500)
check("...but a proper subset of THREE names both of its members",
      sorted(_three) == ["b", "c"])
_mixed = delinquent([U("a", 2000, 0, "at its floor"), U("b", 2000, 0)], 500)
check("a wedged unit beside a pack that is legitimately empty still names the "
      "wedged one: `asked` is the denominator, not `short`",
      list(_mixed) == ["b"])
check("a shortfall AT the threshold is not one - the bound is strict, so a "
      "unit exactly on the gap the control law calls ordinary tracking is not "
      "reported",
      delinquent([U("a", 2000, 2000), U("b", 2000, 1500)], 500) == {}
      and list(delinquent([U("a", 2000, 2000), U("b", 2000, 1499)], 500)) == ["b"])
check("a unit given no share is not asked and cannot be named, and it is not "
      "a peer either - with only one unit asked there is nothing to compare",
      delinquent([U("a", 0, 0), U("b", 2000, 0)], 500) == {})

print()
print("   AC 5: AT ONE INVERTER IT IS INERT, AND BY ALGEBRA RATHER THAN BY A")
print("   ROSTER GUARD - `len(asked) < 2` is unreachable-by-construction there")
_n1 = [delinquent([U("a", s, m)], 500)
       for s in (0, 500, 2000, 5000, -2000, None)
       for m in (0, 250, 2000, -2000, None)]
check(f"every one of {len(_n1)} single-unit rosters over the whole "
      f"(share, measured) grid answers {{}}", all(r == {} for r in _n1))
check("...including a single unit that IS short and unexplained, which is the "
      "case a caller might expect to fire. It must not: with no sibling, "
      "'this box cannot' and 'this plant cannot' are the same observation",
      delinquent([U("a", 2000, 0)], 500) == {})


# ===========================================================================
print()
print("=" * 72)
print("2. CFG-03: config.yaml is the source of truth and there is ONE copy")
print("=" * 72)
_CFG_TEXT = io.open("config.yaml", encoding="utf-8").read()


def read_option(key, block="options"):
    """The value of `key` inside one top-level block of config.yaml, as text.

    ⚠️ READ OUT OF THE FILE, NOT TRANSCRIBED. CFG-03's whole complaint is that a
    default lives in more than one place; a check that hard-coded the number
    here would be a THIRD copy, and the one that never goes red.

    ⚠️ ANCHORED AT EXACTLY TWO SPACES, WHICH IS CFG-04's LESSON AND NOT A STYLE
    CHOICE. a leading-whitespace pattern matches a PER-INVERTER `max_w` nested under
    `inverters:` as readily as the site one, first hit wins, and a reader that
    quietly answers inverter #1's number is the defect `test_inverter.py`'s
    config.yaml read-closure audit exists to hunt. This function is in that
    audit's table by signature - first argument `key` - so it is fed a
    two-inverter config and graded on what it answers.
    """
    body = _CFG_TEXT.split(f"\n{block}:", 1)[1]
    if block == "options":
        body = body.split("\nschema:", 1)[0]
    m = re.search(rf"^  {re.escape(key)}:\s*(\S+)\s*$", body, re.M)
    return m.group(1) if m else None


check("`delivery_hold_s` is in config.yaml's options: block",
      read_option("delivery_hold_s") is not None)
check("...and DELIVERY_OPTION_DEFAULTS agrees with it, so the running add-on - "
      "which cannot read config.yaml, the Dockerfile does not ship it - uses "
      "the number an operator sees on the Configuration page",
      int(read_option("delivery_hold_s"))
      == int(DELIVERY_OPTION_DEFAULTS["delivery_hold_s"]))
check("...and it has a schema: entry, so the Supervisor refuses a value "
      "outside the range the derivation answers for",
      read_option("delivery_hold_s", "schema") is not None)
_schema = read_option("delivery_hold_s", "schema")
_lo, _hi = (int(x) for x in re.match(r"int\((\d+),(\d+)\)", _schema).groups())
check(f"the schema floor is inverter.LINK_STALE_S = 30 s ({_schema}): below "
      f"the firmware's own veto window a unit not executing its share is a "
      f"LINK fault the add-on already answers by dropping it from the split",
      _lo == 30)
check("the schema ceiling is the 900 s quarter that bills the capacity tariff "
      "- a site running short of one unit's share must be reported well "
      "inside one", _hi == 900)
check("the shipped default is inside its own schema",
      _lo <= int(DELIVERY_OPTION_DEFAULTS["delivery_hold_s"]) <= _hi)
check("THERE IS NO SECOND KEY. The watts half is `Tuning.saturation_w`, read "
      "at runtime, so it cannot drift from the number the control law itself "
      "calls 'not following' - CFG-03 already tracks three places a default "
      "has to be kept in step and this card does not earn a fourth",
      list(DELIVERY_OPTION_DEFAULTS) == ["delivery_hold_s"]
      and not re.search(r"^\s+delivery_w:", _CFG_TEXT, re.M))
print(f"       delivery_hold_s: {DELIVERY_OPTION_DEFAULTS['delivery_hold_s']} s, "
      f"schema {_schema}; watts = Tuning.saturation_w = {Tuning().saturation_w:.0f} W")


# ===========================================================================
print()
print("=" * 72)
print("3. the window and the latch")
print("=" * 72)


def drive(seconds, cfg=None, sentence="short", step=5):
    wd = DeliveryWatchdog(cfg or DeliveryConfig(hold_s=60.0))
    at = None
    for t in range(0, int(seconds) + 1, step):
        if wd.evaluate(sentence, T0 + timedelta(seconds=t)) and at is None:
            at = t
    return wd, at


_wd, _at = drive(300)
check("a shortfall held past the window reports", _wd.tripped)
check("...at the window, not before", _at == 60)
check("...and the reason carries the sentence AND how long it stood",
      "short" in _wd.reason and "held for" in _wd.reason and "60 s" in _wd.reason)
print(f"       {_wd.reason!r}")
_wd2, _at2 = drive(55)
check("a shortfall shorter than the window reports nothing",
      not _wd2.tripped and _at2 is None)

_wd3 = DeliveryWatchdog(DeliveryConfig(hold_s=60.0))
for _t in range(0, 300, 5):
    _wd3.evaluate("short" if (_t // 5) % 2 == 0 else None,
                  T0 + timedelta(seconds=_t))
check("an INTERMITTENT shortfall never accumulates: any clean cycle clears the "
      "window, so this needs a CONTINUOUS episode the way SAFETY-02 and "
      "SAFETY-31 do", not _wd3.tripped and _wd3.since is None)

_wd4, _ = drive(300)
_before = _wd4.reason
_wd4.evaluate(None, T0 + timedelta(seconds=999))
check("it LATCHES - and not for SAFETY-02's reason. Nothing here removes the "
      "evidence, because nothing here changes what is commanded; it latches so "
      "that a unit which stopped delivering for minutes and then resumed "
      "cannot erase its own record",
      _wd4.tripped and _wd4.reason == _before)
_wd4.reset()
check("...and reset() is the operator's press, which clears everything",
      not _wd4.tripped and _wd4.reason == "" and _wd4.since is None)

_wd5 = DeliveryWatchdog(DeliveryConfig(hold_s=60.0))
_wd5.evaluate("short", T0)
_wd5.evaluate("short", T0 + timedelta(seconds=5))
_since = _wd5.since
_wd5.arm(DeliveryConfig(hold_s=120.0))
check("arm() is CONFIGURATION, NOT STATE: it does not touch the window, the "
      "latch or the reason. A detector that cleared its own window every cycle "
      "when it was told what it was measuring would never accumulate a hold - "
      "that is SignWatchdog.arm()'s lesson, found the expensive way",
      _wd5.since == _since and _wd5.cfg.hold_s == 120.0 and not _wd5.tripped)
check("held_s takes the caller's clock rather than reading one",
      _wd5.held_s(T0 + timedelta(seconds=65)) == 65.0)
_wd6 = DeliveryWatchdog(DeliveryConfig(hold_s=60.0), tripped=True, reason="from disk")
check("a latch RESTORED from the store comes up tripped, with its reason",
      _wd6.tripped and _wd6.reason == "from disk")
check("...and a reason with no latch behind it is dropped, not carried",
      DeliveryWatchdog(tripped=False, reason="stale").reason == "")


# ===========================================================================
print()
print("=" * 72)
print("4. END TO END, AND THIS IS THE CARD: a wedged unit and a cold pack at")
print("   its floor are IDENTICAL in the power domain. Only the SoC differs.")
print("=" * 72)
# ⚠️ THE PLANT IS MODELLED AND SAYS SO. `_DeliveryHass` is two inverters that
# obey INSTANTLY - no ramp, no lag - and one meter that is `load - sum(measured)`.
# That is deliberately not a claim about transients: the worst legitimate
# episode a real ramp and lag can produce is measured on admin/goodwe's
# two-master rig and recorded in DECISIONS/SAFETY-42-unit-delivery.md. What this
# fixture is for is the DISCRIMINATION - which of two identical power traces is
# reported - and for that an instant plant is the honest model, because it
# removes the one mechanism that could make an obedient unit look absent.
#
# ⚠️ AND INVERTER B DOES NOT PHYSICALLY EXIST (NOTES.md §1.1b). Every
# two-inverter figure in this file, and every one in the decision record, is
# TWO COPIES OF INVERTER A - `plantsim.INVERTER_B`, `measured=False`.

_LOAD_W = 3000.0
_FLOOR_PCT = 10


class _DeliveryHass:
    """Two inverters, one meter, and unit B that produces nothing.

    `b_soc` is the ONLY thing that differs between the two arms of the
    experiment: 50 % is a wedged unit with a healthy pack, `_FLOOR_PCT` is a
    pack sitting exactly on its own cut-out. Every watt on every wire is the
    same in both.
    """

    def __init__(self, load_w=_LOAD_W, b_soc=50.0):
        self.load_w = float(load_w)
        self.b_soc = float(b_soc)
        self.writes = []
        self.calls = []
        self.ctrl = None

    async def call(self, domain, service, data):
        self.calls.append((domain, service, data))

    async def limits(self, entity):
        return None

    def measured(self):
        if self.ctrl is None:
            return [0.0, 0.0]
        a = self.ctrl.inverters[0]
        # A obeys. B produces NOTHING, whatever it is told - and whether that is
        # a wedged power stage or an empty pack is not a question its power
        # register can answer.
        return [float((a.last_write if a.last_write is not None else a.target)
                      or 0.0), 0.0]

    async def number(self, entity, invert=False):
        m = self.measured()
        return {"sensor.grid": self.load_w - sum(m),
                "sensor.soc_a": 50.0, "sensor.soc_b": self.b_soc,
                "sensor.batt_a": m[0], "sensor.batt_b": m[1]}.get(entity)

    async def set_number(self, entity, value):
        self.writes.append((entity, float(value)))
        return True


def _roster():
    return [{"name": n, "setpoint_entity": f"number.sp_{t}",
             "soc_entity": f"sensor.soc_{t}", "batt_entity": f"sensor.batt_{t}",
             "max_w": 2000, "bms_soc_floor": _FLOOR_PCT}
            for n, t in (("inverter-a", "a"), ("inverter-b", "b"))]


def _ctrl(hass, store=None, **over):
    """A two-inverter Controller, with the NEIGHBOURING guards given long windows.

    ⚠️ THE LONG `circulation_hold_s` AND SIGN HOLDS ARE ISOLATION, NOT A CLAIM
    THAT THOSE GUARDS ARE SILENT HERE - part 7 asserts what each of them
    actually concludes on this same fixture, which is the honest version of the
    claim. What must not happen is the SITE being stopped by another guard while
    this section is trying to grade an attribution, because then every check
    below would be measuring the wrong thing.
    """
    opts = {"meter_entity": "sensor.grid", "meter_source": "off",
            "inverters": _roster(), "gain": 0.6, "slew_w": 1000,
            # ⚠️ 1 s, NOT THE SHIPPED 95. This window is WALL CLOCK - the
            # detector takes the caller's `now`, and `_drive()` runs fourteen
            # real cycles 0.2 s apart, so a shipped-length hold would make every
            # check below pass for the wrong reason: nothing would have tripped
            # yet and "no alarm" would read the same as "no fault". The SHIPPED
            # value is graded in part 2 against config.yaml, which is where a
            # number belongs; what is graded here is the attribution.
            "delivery_hold_s": 1,
            "unit_sign_disagree_s": 3600, "site_sign_disagree_s": 3600,
            "circulation_hold_s": 3600, "auto_start": True}
    opts.update(over)
    c = Controller(opts, hass, store if store is not None else _MemStore(),
                   _P(), stop=asyncio.Event())
    for inv in c.inverters:
        inv.dev_min, inv.dev_max = -inv.max_w, inv.max_w
    c.auto = True
    hass.ctrl = c
    return c


class _MemStore:
    """A Store that keeps its dict across "restarts", like the real file does.

    Seeded from `store.DEFAULTS` for test_watchdog.py's reason: a fake that
    hand-listed keys answers None where the real store answers a default, and a
    check reading "this latch was never written" then cannot tell an absent key
    from a False one.
    """

    def __init__(self):
        from app.store import DEFAULTS as _SDEF
        self.data = dict(_SDEF)

    def set(self, k, v):
        self.data[k] = v

    def get_time(self, *a):
        return None


class _P:
    enabled = False

    def publish(self, *a, **k):
        pass

    def close(self):
        pass


async def _drive(c, n=14, dt=0.2, watch=None):
    seen = []
    for _ in range(n):
        await c.read_inputs()
        await c.cycle()
        if watch is not None:
            seen.append(watch(c))
        await asyncio.sleep(dt)
    return seen


def _sample(c):
    return dict(
        grid=c.grid, target=round(c.target, 6),
        shares=tuple(round(float(i.last_write if i.last_write is not None
                                 else i.target), 6) for i in c.inverters),
        meas=tuple(None if i.batt is None else round(float(i.batt), 6)
                   for i in c.inverters),
        soc=tuple(i.soc for i in c.inverters),
        named=tuple(sorted(i.name for i in c.inverters
                           if i.delivery_wd.tripped)),
        reason=c.reason)


async def _arm(b_soc):
    hass = _DeliveryHass(b_soc=b_soc)
    c = _ctrl(hass)
    return c, hass, await _drive(c, watch=_sample)


_cw, _hw, _sw = asyncio.run(_arm(50.0))          # WEDGED: healthy pack, no output
_cf, _hf, _sf = asyncio.run(_arm(float(_FLOOR_PCT)))   # COLD: pack at its floor


def _power_domain(rows):
    return [(r["grid"], r["target"], r["shares"], r["meas"]) for r in rows]


_pw, _pf = _power_domain(_sw), _power_domain(_sf)
check(f"THE TWO ARMS ARE BYTE-IDENTICAL IN THE POWER DOMAIN on all "
      f"{len(_pw)} cycles - grid, site target, every share, every measured "
      f"watt. No threshold on any of those numbers, at any N and any X, can "
      f"separate a wedged unit from a pack at its floor, and the card's own "
      f"proposed discriminator reads only those numbers",
      len(_pw) == len(_pf) and _pw == _pf)
check("...and the ONLY thing that differs is the state of charge",
      [r["soc"] for r in _sw] != [r["soc"] for r in _sf])
print(f"       both arms, last cycle: grid {_sw[-1]['grid']:.0f} W, shares "
      f"{tuple(round(v) for v in _sw[-1]['shares'])}, measured "
      f"{tuple(round(v) for v in _sw[-1]['meas'])}, reason {_sw[-1]['reason']!r}")
print(f"       SoC: wedged arm {_sw[-1]['soc']}  cold arm {_sf[-1]['soc']}")

check("AC 1: the WEDGED unit is NAMED, and only it",
      _sw[-1]["named"] == ("inverter-b",))
check("AC 2: the COLD PACK AT ITS FLOOR IS NOT - on the identical trace",
      _sf[-1]["named"] == ())
print(f"       {_cw.inverters[1].delivery_wd.reason}")

# ⚠️ THE VACUOUS-PASS GUARD, AND IT IS NOT DECORATION. MUTATION.md's third
# failure mode is a check whose data cannot reach the failure it names: "the
# cold arm named nobody" reads exactly the same against a fixture where nothing
# was short, nothing was asked, or the loop never ran. So the cold arm is
# asserted to have REACHED the branch and been turned away by the account, not
# to have missed it.
_cold_b = _cf.inverters[1]
check("...and the cold arm REACHED that decision rather than falling short of "
      "it: B was asked for a discharge share, was short of it by more than "
      "saturation_w, and was turned away by `account()` alone",
      float(_cold_b.standing_w()) > 0
      and short_by(_cold_b.standing_w(), _cold_b.batt) > _cf.tuning.saturation_w
      and account(live=True, parked=False, soc=_cold_b.soc,
                  bms_floor=_cold_b.bms_soc_floor) is not None)
check("...and the WEDGED arm's B differs from it in exactly that one place",
      account(live=True, parked=False, soc=_cw.inverters[1].soc,
              bms_floor=_cw.inverters[1].bms_soc_floor) is None)

# ⚠️ WHICH FLOOR, AND THIS IS A WIRING CHECK THAT THE PURE ONES ABOVE CANNOT
# MAKE. `peakguard.pack_floor_pct()` conjoins a pack's own cut-out with the
# SITE's `peak_soc_floor`, and handing THAT to `account()` would look correct
# and would be wrong: `peak_soc_floor` bounds a money-saving discharge only
# (SAFETY-25), ordinary self-consumption discharges straight past it, and a
# wedged unit sitting anywhere between the two floors would be EXPLAINED AWAY.
# So this arm puts B exactly there - 15 %, above its own 10 % cut-out and below
# a 20 % `peak_soc_floor` - and it must still be named.


async def _between_floors():
    hass = _DeliveryHass(b_soc=15.0)
    c = _ctrl(hass, peak_soc_floor=20)
    await _drive(c)
    return c


_cbf = asyncio.run(_between_floors())
check("a wedged unit BETWEEN its own cut-out and the site's peak_soc_floor is "
      "still named: the floor this check asks about is the pack's own, not the "
      "peak guard's, and swapping them would hide the fault in a band an "
      "ordinary install has 10 points of",
      _cbf.inverters[1].delivery_wd.tripped
      and _cbf.inverters[1].soc == 15.0
      and float(_cbf.o.get("peak_soc_floor")) == 20.0)


# ===========================================================================
print()
print("=" * 72)
print("5. IT REPORTS AND DOES NOT ACT - asserted by counterfactual, not by")
print("   reading the code")
print("=" * 72)


async def _never_trips():
    hass = _DeliveryHass(b_soc=50.0)
    c = _ctrl(hass, delivery_hold_s=900)     # the schema ceiling: never, here
    return c, await _drive(c, watch=_sample)


_cn, _sn = asyncio.run(_never_trips())
check("the guard TRIPPED in one run and not in the other, so the comparison "
      "below is between two different states of this detector",
      _sw[-1]["named"] == ("inverter-b",) and _sn[-1]["named"] == ()
      and _cw.inverters[1].delivery_wd.tripped
      and not _cn.inverters[1].delivery_wd.tripped)
check("...and EVERY WATT IS THE SAME. Grid, site target, both shares and both "
      "measured powers are identical cycle for cycle whether this guard "
      "reported or not - which is what 'it does not act' means, stated as a "
      "measurement instead of as an intention",
      _power_domain(_sw) == _power_domain(_sn))
check("...the wedged unit is still being commanded its share after the trip, "
      "not parked: this is NOT SAFETY-02's remedy and must not become it",
      abs(float(_cw.inverters[1].target)) > 0)
check("...and nothing named `delivery` reaches the arbiter - the resolution's "
      "own sentence is unchanged by the trip",
      "deliver" not in _cw.reason.lower() and _cw.reason == _cn.reason)
_pay = _cw.status_payload()
check("...and control is still running, which is the whole difference from "
      "every other latched guard in this add-on",
      _cw.auto and _pay["level"] != "bad")
check("...but the banner does NOT go on saying 'Running - holding grid at "
      "zero'. That sentence is a claim the site cannot make while a unit is "
      "absent, and 'the page reads healthy while the fault runs' is the shape "
      "this whole card exists to remove. It names the unit and stays a WARNING",
      _pay["level"] == "warn" and "NOT DELIVERING" in _pay["banner"]
      and "inverter-b" in _pay["banner"]
      and "holding grid at zero" not in _pay["banner"])
print(f"       banner: {_pay['level']!r} / {_pay['banner']!r}")
_row = [r for r in _cw.checks() if "NOT DELIVERING" in r["text"]]
check("the status page carries it as a CAUTION, not a failure: the site is "
      "still running and short, which a red cross would misdescribe",
      len(_row) == 1 and _row[0]["ok"] is False and _row[0]["warn"] is True)


# ===========================================================================
print()
print("=" * 72)
print("6. what the NEIGHBOURING guards conclude on this same fixture - the")
print("   honest version of 'every existing guard is silent'")
print("=" * 72)
_b = _cw.inverters[1]
check("SAFETY-30 per unit: SILENT. It needs OPPOSITION on both sides of "
      f"{_cw.unit_sign_cfg.min_w:.0f} W, and a unit at 0 W is not opposing - "
      "it is absent, so the gate never opens",
      not _b.sign_wd.tripped and _b.sign_wd.disagree_since is None)
check("SAFETY-30 site: SILENT. The sum of the standing commands still agrees "
      "with the meter in DIRECTION; only the magnitude is short",
      not _cw.site_sign_wd.tripped and _cw.site_sign_wd.disagree_since is None)
check("SAFETY-31: SILENT, and identically so. sum(|Bi|) - |sum(Bi)| is ZERO "
      "when one unit does nothing - circulation needs two units pulling "
      "AGAINST each other",
      _cw.site_circulating_w == 0.0 and not _cw.circ_wd.tripped)
check("the saturation freeze: FIRES - and that is the point. It is the one "
      "signal that does, and clamp.py:93-96 documents it as the alarm an "
      "operator learns to ignore, which is why this card needed a different "
      "signal rather than a louder one",
      _cw.sat_count >= _cw.tuning.saturation_cycles)
print(f"       saturation counter {_cw.sat_count} (freezes at "
      f"{_cw.tuning.saturation_cycles}); site commanded "
      f"{_cw.target:.0f} W against {_cw.site_batt_w:.0f} W measured")
check("...and it fires IDENTICALLY on the benign arm, on the identical trace, "
      "at the same counter. A saturation-shaped alarm cannot tell these two "
      "apart at any threshold and any duration",
      _cf.sat_count == _cw.sat_count
      and _cf.site_batt_w == _cw.site_batt_w
      and _cf.target == _cw.target)

print()
print("   ...and the one case where BOTH have something to say: a unit that is")
print("   REVERSED is short too, and SAFETY-30 gets there first")
# ⚠️ THE OVERLAP IS REAL AND IT RESOLVES ITSELF, WHICH IS WORTH ASSERTING
# RATHER THAN ASSUMING. A unit measured at -2000 W against a +2000 W share is
# OPPOSING (SAFETY-30's fault) and is also short by 4000 W with nothing to
# explain it (this one). SAFETY-30's hold is shorter, it parks the unit, the
# share goes to 0 W - and this check's gate closes on `share > 0` while
# `account()` answers "parked". Two alarms for one fault is how an operator
# learns to skim, and this is the arrangement that prevents it.


class _ReversedHass(_DeliveryHass):
    """Unit B does the OPPOSITE of what it is told - SAFETY-30's fault."""

    def measured(self):
        a, b = self.ctrl.inverters
        w = lambda i: float((i.last_write if i.last_write is not None
                             else i.target) or 0.0)
        return [w(a), -w(b)]


async def _reversed_run():
    hass = _ReversedHass(b_soc=50.0)
    # SAFETY-30's per-unit hold SHORTER than this check's, which is the shipped
    # ordering too: 85 s against 185 s.
    c = _ctrl(hass, unit_sign_disagree_s=1, delivery_hold_s=2)
    seen = await _drive(c, n=24, dt=0.15, watch=lambda c: dict(
        sign=c.inverters[1].sign_wd.tripped,
        deliv=c.inverters[1].delivery_wd.tripped,
        share=float(c.inverters[1].target),
        acct=account(live=True, parked=c.inverters[1].sign_wd.tripped,
                     soc=c.inverters[1].soc,
                     bms_floor=c.inverters[1].bms_soc_floor)))
    return c, seen


_crev, _srev = asyncio.run(_reversed_run())
check("a REVERSED unit is short by more than saturation_w too, so both guards "
      "have evidence - the fixture reaches the overlap rather than dodging it",
      any(r["deliv"] or r["sign"] for r in _srev)
      and any(abs(r["share"]) > _crev.tuning.saturation_w for r in _srev))
check("SAFETY-30 gets there first and PARKS it - its hold is shorter, which is "
      "also the shipped ordering (85 s against 185 s)",
      _crev.inverters[1].sign_wd.tripped)
check("...and once parked, this check goes quiet on its own: the share is 0 W "
      "so it is no longer ASKED, and `account()` answers 'parked' anyway. ONE "
      "alarm for one fault",
      not _crev.inverters[1].delivery_wd.tripped
      and _crev.inverters[1].delivery_wd.since is None
      and float(_crev.inverters[1].target) == 0.0
      and _srev[-1]["acct"] is not None)
print(f"       shipped ordering: unit sign hold "
      f"{_crev.unit_sign_cfg.hold_s:.0f} s in this fixture, delivery hold "
      f"{_crev.delivery_cfg.hold_s:.0f} s; account now {_srev[-1]['acct']!r}")


# ===========================================================================
print()
print("=" * 72)
print("7. AC 5: AT ONE INVERTER NOTHING CHANGES - asserted, not assumed")
print("=" * 72)


class _OneHass(_DeliveryHass):
    """One inverter that produces nothing against a non-zero share."""

    def measured(self):
        return [0.0]

    async def number(self, entity, invert=False):
        return {"sensor.grid": self.load_w, "sensor.soc": 50.0,
                "sensor.batt": 0.0}.get(entity)


async def _one():
    hass = _OneHass()
    opts = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
            "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
            "meter_source": "off", "max_w": 2000, "gain": 0.6,
            # ⚠️ THE SAME 1 s AS THE TWO-INVERTER FIXTURE, AND THAT IS WHAT
            # MAKES THIS SECTION NON-VACUOUS. The run below is 2 s of wall clock
            # against a 1 s window, so a hold is not what stops this firing -
            # the reduction is. With a shipped-length hold "nothing was
            # reported" would be true of a fixture that simply ran out of time.
            "bms_soc_floor": _FLOOR_PCT, "delivery_hold_s": 1,
            "sign_disagree_s": 3600, "auto_start": True}
    store = _MemStore()
    c = Controller(opts, hass, store, _P(), stop=asyncio.Event())
    c.inverters[0].dev_min, c.inverters[0].dev_max = -2000.0, 2000.0
    c.auto = True
    hass.ctrl = c
    await _drive(c, n=20, dt=0.1)
    return c, hass, store


_c1, _h1, _st1 = asyncio.run(_one())
_only = _c1.inverters[0]
check("a roster of ONE, whose only unit is standing at a discharge share and "
      "delivering nothing for the whole run - the exact fault, at N=1",
      len(_c1.inverters) == 1 and float(_only.standing_w()) > 0
      and _only.batt == 0.0
      and short_by(_only.standing_w(), _only.batt) > _c1.tuning.saturation_w)
check("...and NOTHING is reported. With no sibling, 'this box cannot' and "
      "'this plant cannot' are the same observation, and the saturation freeze "
      "is still the whole answer there", not _only.delivery_wd.tripped)
check("...the window was never even opened", _only.delivery_wd.since is None)
check("...and no store key was written, so a one-inverter install's state file "
      "is byte-identical to what it was before this card",
      unit_key(_only.name, "delivery_tripped") not in _st1.data
      and unit_key(_only.name, "delivery_reason") not in _st1.data)
_n1row = [r for r in _c1.checks() if "delivery watchdog" in r["text"]]
check("...and the status page says n/a rather than 'armed', which would be a "
      "claim to be watching something that cannot happen",
      len(_n1row) == 1 and _n1row[0]["ok"] and "n/a at one inverter" in _n1row[0]["text"])
check("...no persistent notification of this kind was raised either",
      not [c for c in _h1.calls
           if "not_delivering" in str(c[2].get("notification_id", ""))])


# ===========================================================================
print()
print("=" * 72)
print("8. SAFETY-34: the report survives a Home Assistant restart, which is")
print("   how the 2026-09-02 outage went unseen for 6 h 25 m")
print("=" * 72)


async def _restart():
    shared = _MemStore()
    h1 = _DeliveryHass(b_soc=50.0)
    c1 = _ctrl(h1, store=shared)
    await _drive(c1)
    # A NEW process, a NEW HomeAssistant object (so the in-memory notification
    # is gone with it), the SAME durable file. That is what a Home Assistant
    # restart is from this add-on's point of view: the add-on goes with it.
    h2 = _DeliveryHass(b_soc=50.0)
    c2 = _ctrl(h2, store=shared)
    await c2.check_not_delivering()
    return c1, h1, c2, h2, shared


_r1, _rh1, _r2, _rh2, _rs = asyncio.run(_restart())
_key = unit_key("inverter-b", "delivery_tripped")
check("the latch reached the durable file, namespaced under the unit it is "
      "about", _rs.data.get(_key) is True)
check("...with its reason, and the reason was written FIRST - a crash between "
      "the two must not persist a latch nobody can explain",
      bool(_rs.data.get(unit_key("inverter-b", "delivery_reason"))))
check("...and there is NO flat mirror of it, where SAFETY-02 has one: the "
      "version a rollback would land on drives exactly one actuator, a roster "
      "at which this check is inert by construction",
      "delivery_tripped" not in _rs.data and "delivery_reason" not in _rs.data)
check("a fresh process comes up with the report STILL STANDING",
      _r2.inverters[1].delivery_wd.tripped)
check("...and RE-RAISES the notification, because a persistent_notification is "
      "in memory and did not survive - that is the whole of SAFETY-34",
      any(c[0] == "persistent_notification"
          and "not_delivering" in str(c[2].get("notification_id", ""))
          for c in _rh2.calls))
_note = [c for c in _rh2.calls
         if "not_delivering" in str(c[2].get("notification_id", ""))][0]
check("the notification NAMES THE UNIT - in its id, its title and its body - "
      "which is card requirement 2 and the whole reason SAFETY-30 split "
      "per-unit from site",
      "inverter-b" in _note[2]["notification_id"].replace("_", "-")
      and "inverter-b" in _note[2]["title"]
      and "inverter-b" in _note[2]["message"])
check("...and it says, in the operator's own sentence, that nothing has been "
      "commanded differently. An alarm that let somebody believe the add-on "
      "had handled it would be worse than no alarm",
      "NOTHING HAS BEEN COMMANDED DIFFERENTLY" in _note[2]["message"])
print(f"       id={_note[2]['notification_id']!r} title={_note[2]['title']!r}")
_before_calls = len(_rh2.calls)
await_again = asyncio.run(_r2.check_not_delivering())
check("...and it is raised ONCE per process, not once per cycle: the alarm "
      "storm this add-on has argued against twice already",
      len(_rh2.calls) == _before_calls)

_r2.auto = False
asyncio.run(_r2.handle_action("auto_toggle"))     # off
asyncio.run(_r2.handle_action("auto_toggle"))     # and on again: the press
check("the operator's START press is the only way out, and it clears the "
      "latch, the reason and the 'a human has seen this' flag together",
      not _r2.inverters[1].delivery_wd.tripped
      and _rs.data.get(_key) is False
      and _rs.data.get(unit_key("inverter-b", "delivery_reason")) == ""
      and _r2._delivery_announced == set())


print()
if fails:
    print(f"{len(fails)} FAILED: {', '.join(fails)}")
    sys.exit(1)
print("all checks passed")