glenn schrooyenandClaude Opus 5
c7eb8c6b97
The battery is 12 mutations, one freshly copied tree each, purged, `:ro`,
`python3 -B`, byte deltas measured. First round: 9 red, 3 survivors. All three
are now accounted for and the ledger is in DECISIONS §8.
M5 - `share <= 0` relaxed to `share == 0`, +0 B - WAS A REAL GAP AND THE TWO
OBVIOUS CHECKS DID NOT CLOSE IT. `short_by(-2000, 0) is None` and
`short_by(-2000, -100) is None` both stayed GREEN under the mutant: for a charge
share that is UNDER-delivered `share - measured` is negative and the `gap > 0`
test answers None anyway, so neither input can separate the two spellings. The
one that can is a unit moving MORE than its charge share - -2000 W told,
-3000 W measured, `share - measured` = +1000 W - which the mutant reports as a
shortfall. It is not one; it is something else driving that pack, which is
SAFETY-30's and SAFETY-31's question. That input is now the check, and M5 is RED.
MUTATION.md's third failure mode in miniature: the assertion ran, the sentence
was true, and the data could not reach the failure it named.
M11 - `inv.is_live(now)[0]` replaced by a literal `True`, -15 B - was a WIRING
gap the pure checks structurally could not see. `account(live=False, ...)` was
graded as a FUNCTION; nothing graded that `cycle()` asks the question, because no
fixture had a unit whose writes were being rejected. Such a unit still has its
last ACCEPTED command standing on the wire (`write_setpoint` leaves `last_write`
alone on a rejected write) and still measures nothing, so it looks exactly like a
wedged one to every number this check reads - and reporting it would be the
second, louder alarm about a state the allocator has already handled and logged.
`test_delivery.py` part 6 now builds that unit, with `rejected_since` back-dated
past LINK_STALE_S because the fixture runs in two seconds and the timer is
test_inverter.py's to grade. M11 is RED.
M3 - `len(asked) < 2` relaxed to `< 1`, +0 B - IS STILL A SURVIVOR, and it is an
EQUIVALENT MUTANT. With one asked unit `unaccounted` is a subset of a one-element
set, so it is either empty (`not unaccounted` returns {}) or equal to `asked`
(`len(unaccounted) == len(asked)` returns {}); with zero asked units both
spellings return {} as well. No input separates them, so no check can.
⚠️ AND FINDING IT CORRECTED THE MODULE'S OWN PROSE, WHICH IS THE MORE USEFUL
HALF. `delinquent()`'s docstring claimed `len(asked) < 2` was what makes N=1
inert. It is not - the PROPER-SUBSET rule is, and that line is a fast path and a
signpost. A false attribution of the mechanism in the file's own prose is exactly
what this repository keeps finding, and a mutation that survives is how it
surfaced. The line stays, with the equivalence written down beside it the way
circulation.circulating_w() records its own max(0.0, ...).
Also recorded, rather than left implicit: `account()`'s `parked` branch is graded
as a function but its WIRING is not mutated on its own; part 6's reversed-unit
fixture exercises it end to end, which is weaker, and the ledger says so.
Baseline 23/23 green before and after. app/main.py is byte-identical to the
previous commit - both closures are a test and a comment.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BwRLw7eKYj2byVdwB6kbjU
881 lines
41 KiB
Python
881 lines
41 KiB
Python
"""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 LINK_STALE_S
|
|
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)
|
|
# ⚠️ THE CASE THAT ACTUALLY GRADES THE `share <= 0` GUARD, AND THE MUTATION
|
|
# BATTERY IS WHAT FOUND IT MISSING. Relaxing that guard to `share == 0` left the
|
|
# two rows above GREEN - for a charge share that is UNDER-delivered,
|
|
# `share - measured` is negative and the `gap > 0` test answers None anyway, so
|
|
# the mutant is masked. The one input where the two differ is a unit moving MORE
|
|
# than its charge share: -2000 W told, -3000 W measured, `share - measured` =
|
|
# +1000 W, which the mutant reports as a SHORTFALL. It is not one - it is
|
|
# something else driving that pack, which is SAFETY-30's and SAFETY-31's
|
|
# question - and reading it as "short of its share" would name the wrong fault.
|
|
check("...and a unit moving MORE than its charge share is not short of it "
|
|
"either. This is the input that separates `share <= 0` from `share == 0`; "
|
|
"the two rows above do not, because the `gap > 0` test masks them",
|
|
short_by(-2000, -3000) is None and short_by(-2000, -2600) 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 a unit the ALLOCATOR HAS ALREADY DROPPED is not reported twice")
|
|
# ⚠️ A WIRING CHECK THE PURE ONE IN PART 1 CANNOT MAKE, AND THE MUTATION BATTERY
|
|
# IS WHAT FOUND IT MISSING. `account(live=False, ...)` is graded up there; what
|
|
# is graded HERE is that `cycle()` actually ASKS `inv.is_live(now)`. Replacing
|
|
# that call with a literal `True` left every check green, because no fixture had
|
|
# a unit whose writes were being rejected. A dropped unit still has its last
|
|
# ACCEPTED command standing on the wire (`write_setpoint` leaves `last_write`
|
|
# alone on a rejected write) and still measures nothing, so it looks exactly
|
|
# like a wedged one - and reporting it would be a second, louder alarm about a
|
|
# state the allocator has already handled and logged, which is the
|
|
# SAFETY-15/SAFETY-19 failure this module takes its 30 s floor from.
|
|
|
|
|
|
class _LinkDeadHass(_DeliveryHass):
|
|
"""B is healthy, then its LINK dies: writes rejected and output gone.
|
|
|
|
⚠️ `rejected_since` IS BACK-DATED RATHER THAN WAITED OUT. `is_live()`'s
|
|
window is `LINK_STALE_S` = 30 s of wall clock and this fixture runs in about
|
|
two; waiting it out would make the check's runtime the thing it grades. What
|
|
is graded is the CALL, not the timer - `test_inverter.py` part 3 owns the
|
|
timer.
|
|
"""
|
|
|
|
def __init__(self, dead_from=5, **kw):
|
|
super().__init__(**kw)
|
|
self.dead_from = int(dead_from)
|
|
self.cycle_n = 0
|
|
|
|
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), 0.0 if self.cycle_n >= self.dead_from else w(b)]
|
|
|
|
async def set_number(self, entity, value):
|
|
self.writes.append((entity, float(value)))
|
|
if entity == "number.sp_b" and self.cycle_n >= self.dead_from:
|
|
return False
|
|
return True
|
|
|
|
|
|
async def _link_dead():
|
|
hass = _LinkDeadHass(dead_from=5)
|
|
c = _ctrl(hass)
|
|
b = c.inverters[1]
|
|
seen = []
|
|
for n in range(18):
|
|
hass.cycle_n = n
|
|
if n == 5:
|
|
# the moment the link dies, back-dated past the firmware's window
|
|
b.rejected_since = (datetime.now(timezone.utc)
|
|
- timedelta(seconds=LINK_STALE_S + 10))
|
|
await c.read_inputs()
|
|
await c.cycle()
|
|
seen.append(dict(live=b.is_live(datetime.now(timezone.utc))[0],
|
|
stood=float(b.standing_w()), meas=b.batt,
|
|
named=b.delivery_wd.tripped))
|
|
await asyncio.sleep(0.15)
|
|
return c, seen
|
|
|
|
|
|
_cld, _sld = asyncio.run(_link_dead())
|
|
_bld = _cld.inverters[1]
|
|
check("the fixture REACHES the case: B is not live, is still standing at a "
|
|
"discharge command, and is measuring nothing - i.e. it looks exactly like "
|
|
"a wedged unit to every number this check reads",
|
|
not _sld[-1]["live"] and _sld[-1]["stood"] > _cld.tuning.saturation_w
|
|
and _sld[-1]["meas"] == 0.0)
|
|
check("...and it is NOT reported, because `account()` was asked and answered. "
|
|
"The allocator has already dropped it and said so in the action log; a "
|
|
"second alarm about that is the SAFETY-15/19 failure",
|
|
not _bld.delivery_wd.tripped and _bld.delivery_wd.since is None)
|
|
check("...and the answer is the LINK one specifically, not the SoC one - so a "
|
|
"reader can tell which branch withheld the alarm",
|
|
"link" in (account(live=False, parked=False, soc=_bld.soc,
|
|
bms_floor=_bld.bms_soc_floor) or ""))
|
|
print(f" B: live={_sld[-1]['live']} standing={_sld[-1]['stood']:.0f} W "
|
|
f"measured={_sld[-1]['meas']:.0f} W -> not reported")
|
|
|
|
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")
|