admin/goodwe-addon
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

SAFETY-57: the stale-reading verdict gets an exit, because its own consequence is what stopped it being tested
checks / unit (pull_request) Successful in 7m58s
Details

Browse Source 
SAFETY-37's freshness gate reports a battery-power register as MISSING when it
stops following its own command, and a missing input is "command 0 W and keep
commanding" (§11.1 rule 1). The gate's only exit was "an arrival carrying a
DIFFERENT number" - and at 0 W nothing is asking the register to be anywhere.
The state had no exit reachable from inside it, and the log line promised the
opposite in the same sentence that announced the 0 W.

REPRODUCED FROM THE WIRE, NOT MODELLED. scenarios/safety57-2026-09-21/ is 3840
verbatim frames off the RS-485 pair, read from the passive sniffer, decoded by
tools/decode_bus_capture.py (0 unparsed, 0 bad CRC, 0 unknown shapes; the
mode->sign convention is graded off the capture, not trusted). The box was
running 0.10.0, so the 0 W in the trace IS the add-on's own claim:

  10:51:24.7  0x0518 reaches -3 W. Pack at 11 %, the BMS refuses the discharge,
              the inverter idles. THE REGISTER IS HEALTHY AND TRUTHFUL.
  10:51:29.7  the command reaches +1060 W - 800 W of travel
  10:52:00    the conjunction arms. Inverter.batt becomes None
  10:52:21.7  the site commands 0 W, on the wire
  ...         0x0518 does not change value for 850.0 s. The setpoint does not
              change value for 790.0 s. The house exports 2.4 kW throughout
  ~11:04      a human restarts the add-on. That is what ends it

AND IT HAD BEEN ESCAPING BY LUCK. Replaying the committed 16-day loop capture
through the shipped conjunction finds 93 trips whose wait for the BMS
keep-alive pulse that cleared them has a median of 143.4 s, a worst case of
21440.1 s, and 17 of 93 at or beyond this episode's 710 s. That distribution
belongs to the battery's firmware, not to this code.

THE CARD'S OPTION 2 ALREADY HELD AND IS NOT THE BUG. _grade() has required
|standing| >= MOVE_W since SAFETY-37. The trip was at +1060 W and the stuck
state is at 0 W; they are different moments, and an entry-side rule cannot
reach a state entered legitimately and then persisted. Nothing in the
conjunction is weakened here.

THE RULE, SAFETY-13's SHAPE ONE MODULE OVER - which latches get an exit, not
how long the block lasts:

  the verdict is withdrawn when the obligation that produced it has been gone
  for RELEASE_S, and the register is then re-graded from scratch.

RELEASE_S is BATT_OBLIGED_S, 17.0 s - the obligation window already derived in
the file (6.0 plant + 10.0 publish + 1.0 read), taken at the FLOOR of its band
for the mirror of MOVE_W's reason. Which half is safe is the opposite of
SAFETY-13's: being wrong long is the failure being fixed, being wrong short
costs one bounded re-grading window. Below 17 s the release pre-empts the
cheaper exit - the register answering by itself - and that is graded
behaviourally, not asserted.

The whole parked interval is now stale_input_s + RELEASE_S = 32 s. On the live
episode: telemetry back at 10:52:39 instead of 11:05:35, with 0x0518 still
answering the same -3 W. RED BEFORE (release_s=inf, a graded counterfactual and
not a second implementation), GREEN AFTER.

WHAT IT COSTS, AND IT IS NOT SMALL. A register dead for good is re-graded for
ever: over 1201 reads the held number reaches a consumer on 630 (52.5 %)
against 35 (2.9 %) before. Payable only because one window is at most 35 s and
the two consumers hold for 75 s and 85 s, so neither can reach a verdict inside
one and the re-trip clears both windows. No detection SAFETY-37 had is given
back - a permanent None clears the same windows for ever too - and the command
is derived from the P1 meter, never from batt.

Two clocks, deliberately: changed_mono/unchanged_s() stay facts about the
REGISTER, graded_from/graded_s() are the VERDICT's window. Collapsing them
either lies to the operator or defeats the re-grading.

main.py's stale line had its ESCAPE CLAUSE corrected - it promised a clearance
the state could not produce - and gained a STALE WITHDRAWN edge that says in
its own words that nothing has been learned about the register. The DIAGNOSIS
half is SAFETY-56's and is deliberately untouched, with a named check asserting
it is still there so that rewrite has to be deliberate.

Battery: 22 mutations, 21 red, 1 argued equivalent, 0 unexplained survivors.
Two findings kept in DECISIONS: _release()'s move_w test swapped for
unchanged_s SURVIVED at +5 B because every fixture rested at exactly 0 W (habit
11, closed by a 0/100/300 W fixture); and the code shipped a habit-22 comment
claiming the _release/_grade order was load-bearing, which the battery
falsified - the two act on mutually exclusive predicates, so the order is not
observable. MUTATION.md gains habit 23 for a mutation that ran, fingerprinted
clean, and was a no-op on the fixture because `0.0 or x` is x.

Suite: 30/30 in the CI image, checks.yml's exact command, purged tree, :ro, one
file at a time.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BwRLw7eKYj2byVdwB6kbjU
This commit is contained in:
admin
2026-09-21 12:33:56 +02:00
co-authored by Claude Opus 5
parent 6f5e8e378a
commit 43c144b8d9
10 changed files with 1921 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
goodwe_controller/test_freshness.py
+774 -16
View File
@@ -49,10 +49,14 @@ before you report a survivor.
import ast
import asyncio
import dataclasses
import glob
import gzip
import hashlib
import importlib.util
import io
import json
import logging
import sys
import time
from datetime import datetime, timezone
@@ -500,11 +504,13 @@ for t in range(40, 50):
_after_zero.append(_lat.reading())
check("PRECONDITION: it is stale before the site is stopped",
_was and _lat.reading() is None)
check("⚠️ AND IT STAYS STALE ONCE THE SITE COMMANDS 0 W. The conjunction is no "
"longer armed - |standing| is 0 W, below the 250 W term - so an unlatched "
"verdict would hand the held -1200 W straight back and the site would "
"restart into it, then stop, then restart, once per read",
all(v is None for v in _after_zero))
check("⚠️ AND IT STAYS STALE ONCE THE SITE COMMANDS 0 W, for RELEASE_S. The "
"conjunction is no longer armed - |standing| is 0 W, below the 250 W "
"term - so an unlatched verdict would hand the held -1200 W straight "
"back and the site would restart into it, then stop, then restart, once "
"per read. The ten reads here are inside the 17.0 s SAFETY-57 waits",
all(v is None for v in _after_zero) and len(_after_zero) == 10
and 10 < F.RELEASE_S)
# ⚠️ AND A FAILED READ MUST NOT CLEAR IT EITHER. This check exists because the
# mutation that added `self.stale = False` to `submit()`'s None branch SURVIVED
# the whole sweep - no fixture had ever injected a read failure while latched.
@@ -526,11 +532,26 @@ check("⚠️ A FAILED READ DOES NOT CLEAR THE LATCH - found by a SURVIVING "
_after_fail is None and _lat_fail.stale
and _lat_fail.reading() is None)
check("⚠️ AND ONLY MOVEMENT CLEARS IT - a different number arriving, which is "
"exactly the evidence the verdict was about. Not a timer, not a restart, "
"not the operator's start button",
(lambda f: (f.submit(A_FROZEN_W + 1.0, 0.0, 60.0),
(not f.stale) and f.reading() == A_FROZEN_W + 1.0)[-1])(_lat))
# ⚠️ THIS CHECK USED TO READ "AND ONLY MOVEMENT CLEARS IT ... Not a timer, not
# a restart, not the operator's start button", AND SAFETY-57 IS THE CARD THAT
# FALSIFIED IT. It was driven on `_lat` - already 20 s into a 0 W rest - so
# under the shipped exit it would now pass having never tested movement at all,
# which is habit 9: a check that goes on being true for a reason it does not
# name. It is rebuilt here on a FRESH object that has never rested, so the only
# thing that can clear it is the arrival.
_mov = Freshness(BOUNDS["batt"])
for t in range(0, 40):
_mov.submit(A_FROZEN_W, A_FROZEN_W if t < 1 else 1300.0, float(t))
_mov_was = _mov.stale
_mov.submit(A_FROZEN_W + 1.0, 1300.0, 40.0) # the register moves one watt
check("⚠️ MOVEMENT CLEARS IT WHILE THE OBLIGATION IS STILL STANDING - a "
"different number arriving under a command that has NOT rested, which is "
"exactly the evidence the verdict was about. Not a restart, not the "
"operator's start button. SAFETY-57's timer cannot be what cleared this "
"one: the command was +1300 W on every read, so the release clock never "
"started",
_mov_was and not _mov.stale and _mov.reading() == A_FROZEN_W + 1.0
and _mov.releases == 0 and _mov.rested_since is None)
# ⚠️ THE MAGNITUDE TERM OF THE CONJUNCTION, AND THIS CHECK EXISTS BECAUSE A
# MUTATION SURVIVED WITHOUT IT. Deleting `abs(standing_w) < move_w` from
# `_grade()` left every other check in this file green, because in all of them a
@@ -542,12 +563,12 @@ _rest = Freshness(BOUNDS["batt"])
# BEFORE the command settles - and then the site is commanded to 0 W and the
# register sits at +5 W, correctly, for a minute.
_rest.submit(1290.0, 1300.0, 0.0)
_rest.submit(5.0, 1300.0, 1.0) # standing_at_change captures 1300 W
_rest.submit(5.0, 1300.0, 1.0) # standing_at_grade captures 1300 W
for t in range(2, 60):
_rest.submit(5.0, 0.0, float(t)) # site at rest; register truthfully +5 W
check("⚠️ THE MAGNITUDE TERM IS LOAD-BEARING, and it took a surviving mutation "
"to show it. A register that reached +5 W while the old +1300 W command "
"was still standing has a `standing_at_change` of 1300 W, so the TRAVEL "
"was still standing has a `standing_at_grade` of 1300 W, so the TRAVEL "
"term alone reads 1300 W of unfollowed command for the whole minute the "
"site rests - and calls a perfectly truthful reading dead. Requiring "
"|standing| >= 250 W as well is what says 'a site at rest is not owed "
@@ -564,7 +585,7 @@ check("...and the hole that term leaves CANNOT MISLEAD EITHER CONSUMER, which "
check("...and one watt is enough, because 0x0518 is a 1 W register that "
"dithers +/-1 W while holding (goodwe/NOTES.md §5.11, measured "
"2026-08-22) - a working register clears this on its next 10 s publish",
_lat.held == A_FROZEN_W + 1.0 and not _lat.stale)
_mov.held == A_FROZEN_W + 1.0 and not _mov.stale)
# ==============================================================================
@@ -655,6 +676,7 @@ _unit_min_w = option_default("unit_sign_disagree_w")
_ceiling = min(_circ_hold, _unit_hold)
print(f" floor {BATT_OBLIGED_S:.1f} s = settle 6.0 + publish 10.0 + read 1.0")
print(f" bound {BATT_UNCHANGED_S:.1f} s")
print(f" exit {F.RELEASE_S:.1f} s (SAFETY-57, the same obligation window)")
print(f" ceil {_ceiling:.1f} s = min(circulation_hold_s {_circ_hold:.0f}, "
f"unit_sign_disagree_s {_unit_hold:.0f})")
@@ -690,6 +712,24 @@ check("⚠️ AND `MOVE_W` IS PINNED TO config.yaml TOO. It is the largest value
"raise `unit_sign_disagree_w` and the gate would stop arming in states "
"that check still fires in, which this catches",
MOVE_W == _unit_min_w)
check("⚠️ AND SAFETY-57'S EXIT SITS INSIDE THAT SAME BAND, AT ITS FLOOR - the "
"mirror of MOVE_W taking its own band's ceiling. The question a release "
"answers is 'has the register had its chance to answer the command "
"standing now?', which is the obligation window and nothing else, so the "
"exit is exactly the floor and strictly shorter than the bound whose "
"verdict it withdraws. An exit at or above the bound would mean the site "
"spends longer parked than the evidence that parked it took to gather",
F.RELEASE_S == BATT_OBLIGED_S and F.RELEASE_S < BATT_UNCHANGED_S
and BOUNDS["batt"].release_s == F.RELEASE_S)
check("⚠️ AND THE WHOLE PARKED INTERVAL IS NOW BOUNDED BY TWO SHIPPED NUMBERS "
"AND NOTHING ELSE: `stale_input_s` (15 s, config.yaml) then `release_s` "
"(17.0 s) = 32 s. The schema caps `stale_input_s` at 120, so even the "
"worst accepted configuration parks for 137 s - where the live episode "
"parked for 790 s and had no upper bound at all",
float(option_default("stale_input_s")) + F.RELEASE_S == 32.0
and "stale_input_s: int(5,120)" in io.open(
HERE / "config.yaml", encoding="utf-8").read()
and 120.0 + F.RELEASE_S == 137.0)
check("...and the module's three floor terms are the numbers this repository "
"actually carries, not a transcription of them: the ESPHome publish "
"interval is read back out of the shipped firmware",
@@ -734,7 +774,7 @@ _uses_batt = bounds_users("batt")
_uses_soc = bounds_users("soc")
for _k, _b in BOUNDS.items():
print(f" {_k:5} wired={_b.wired!s:5} unchanged={_b.unchanged_s:>6} "
f"move={_b.move_w:>6} {_b.signal}")
f"move={_b.move_w:>6} release={_b.release_s:>6} {_b.signal}")
print(f" BOUNDS['batt'] subscripted in CODE by: {_uses_batt}")
print(f" BOUNDS['soc'] subscripted in CODE by: {_uses_soc or 'nothing'}")
@@ -752,10 +792,17 @@ check("...and the SoC row says what TEL-08 owes, in the row itself, so the "
"energy" in BOUNDS["soc"].derivation
and "153.6 Wh" in BOUNDS["soc"].derivation
and "TEL-08" in BOUNDS["soc"].derivation)
check("...and every wired bound is a real number - an infinite bound is how a "
"row silently becomes decorative",
check("...and every wired bound is a real number IN ALL THREE COLUMNS - an "
"infinite bound is how a row silently becomes decorative, and SAFETY-57's "
"`release_s` is the newest place that can happen: at `inf` the module is "
"byte-for-byte the one that parked the live site for 790 s",
all(b.unchanged_s < float("inf") and b.move_w < float("inf")
and b.release_s < float("inf")
for b in BOUNDS.values() if b.wired))
check("...and the UNWIRED row is infinite in all three, so nothing about "
"SoC's absent bound can be read as a policy that exists",
all(b.unchanged_s == b.move_w == b.release_s == float("inf")
for b in BOUNDS.values() if not b.wired))
# ==============================================================================
@@ -809,6 +856,96 @@ check("⚠️ AND THE LONG RUNS ARE ALL UNDER AN UNMOVING LOAD, which is what th
max(_runs)[1] == -899.0)
# --- 8b. AND THE EXIT THE BOUND USED TO HAVE, ON THE SAME HOUSE -------------
# ⚠️ THIS IS THE MEASUREMENT SAFETY-57 RESTS ON, and it is about the REGISTER
# rather than about any code: before this card the only way out of the verdict
# was "an arrival carrying a different number", so the length of the state IS
# the time to 0x0518's next change. The committed 16-day loop capture is
# replayed at 1 Hz through the SHIPPED conjunction to find the trips, and each
# trip is then measured against the register's own next move.
#
# ⚠️ WHAT THIS REPLAY IS AND IS NOT. The capture runs 2026-09-02..17 and 0.10.0
# - the first version carrying this gate - was cut on 2026-09-16, so the
# setpoint column is an UNGATED controller's. That makes each trip an honest
# register-versus-command divergence off the real bus, and it makes the
# post-trip command column something a gated box would NOT have written. The
# figure taken from it is therefore only the one that does not depend on the
# command at all: how long this register goes without changing value. If
# anything it UNDER-states the wait, because a site parked at 0 W gives the
# register even less reason to move than the one recorded here.
_LOOPDIR = HERE / "scenarios" / "loop-2026-09-02_2026-09-17"
_loop_days = sorted({Path(f).name.rsplit("-", 1)[0]
for f in glob.glob(str(_LOOPDIR / "*-batt.jsonl.gz"))})
def _loop_series(day, sig):
out = []
with gzip.open(_LOOPDIR / f"{day}-{sig}.jsonl.gz", "rt") as fh:
for line in fh:
row = json.loads(line)
try:
v = float(row["v"])
except ValueError:
continue # 'unavailable' / 'unknown' are not readings
out.append((datetime.fromisoformat(row["t"]).timestamp(), v))
return out
_t8b = time.monotonic()
_waits, _trips8b = [], 0
for _day in _loop_days:
_b, _s = _loop_series(_day, "batt"), _loop_series(_day, "setpoint")
if not _b or not _s:
continue
_moves = [r[0] for i, r in enumerate(_b) if i == 0 or r[1] != _b[i - 1][1]]
_f8b, _prev8b = Freshness(BOUNDS["batt"]), False
_ib = _is = 0
_t = max(_b[0][0], _s[0][0])
_end = min(_b[-1][0], _s[-1][0])
while _t <= _end:
while _ib + 1 < len(_b) and _b[_ib + 1][0] <= _t:
_ib += 1
while _is + 1 < len(_s) and _s[_is + 1][0] <= _t:
_is += 1
_f8b.submit(_b[_ib][1], _s[_is][1], _t)
if _f8b.stale and not _prev8b:
_trips8b += 1
_nxt = next((m for m in _moves if m > _t), None)
if _nxt is not None:
_waits.append(_nxt - _t)
_prev8b = _f8b.stale
_t += 1.0
_waits.sort()
_p50 = _waits[len(_waits) // 2]
print(f" {len(_loop_days)} days replayed at 1 Hz in "
f"{time.monotonic() - _t8b:.1f}s: {_trips8b} trips")
print(f" wait for the register's next change after a trip: "
f"min {_waits[0]:.1f} s median {_p50:.1f} s max {_waits[-1]:.1f} s; "
f"{sum(1 for w in _waits if w >= 710.0)} at or over the live episode's "
f"710 s")
check("PRECONDITION: all sixteen committed days were replayed, not whichever "
"files happened to be present",
len(_loop_days) == 16 and _loop_days[0] == "2026-09-02"
and _loop_days[-1] == "2026-09-17" and len(_waits) == _trips8b)
check("⚠️ THE OLD EXIT WAS NOT RARE AND IT WAS NOT SHORT. Ninety-three trips in "
"sixteen days of this house, and the only thing that could clear any of "
"them - the register carrying a different number - took a MEDIAN of "
"143.4 s and as long as 21440.1 s. Seventeen of the ninety-three are at "
"or beyond the 710 s the live site was parked for on 2026-09-21",
_trips8b == 93 and abs(_p50 - 143.4) < 0.1
and abs(_waits[-1] - 21440.1) < 0.1
and sum(1 for w in _waits if w >= 710.0) == 17)
check("⚠️ AND THAT DISTRIBUTION IS A PROPERTY OF THIS BMS, NOT OF THIS CODE - "
"which is the whole reason the exit could not stay as it was. The short "
"waits are the pack's keep-alive pulse twitching 0x0518 and satisfying "
"the clearing rule BY ACCIDENT: a quarter of them are under 20 s, and "
"nothing in this add-on asked for any of them. After SAFETY-57 the wait "
"is `stale_input_s + release_s` = 32 s whatever the register does",
sum(1 for w in _waits if w < 20.0) >= 20
and float(option_default("stale_input_s")) + BOUNDS["batt"].release_s == 32.0)
# ==============================================================================
print()
print("9. KNOWN SURVIVORS - graded, so they are residuals and not surprises")
@@ -871,6 +1008,627 @@ check("NOT A SURVIVOR, recorded because it looks like one: a publisher failure "
"each unit's own command travel and each unit's own number",
_s4a.stale and _s4b.stale and _s4a.held != _s4b.held)
# S5. SAFETY-57's own, and it is a FALSE POSITIVE that predates this card.
# The unchanged clock keeps running while the site rests, so the FIRST read
# at which a command travels 250 W after a long quiet already satisfies all
# three terms - the register is given no time at all to answer the new
# command. Present on the shipped tree; the live capture in part 10 contains
# one (10:48:14, cleared one second later).
_s5 = Freshness(BOUNDS["batt"])
for t in range(0, 300): # five minutes of a resting site
_s5.submit(-3.0, 0.0, float(t))
_s5_quiet = _s5.stale
_s5.submit(-3.0, 1500.0, 300.0) # the loop asks for 1500 W
check("⚠️ SURVIVOR 5, FALSE POSITIVE, AND IT IS OLDER THAN SAFETY-57: after a "
"long rest the unchanged clock is already past 35 s, so the very FIRST "
"read carrying a travelled command trips the gate - before the 17.0 s in "
"which a healthy register could have answered it. The gate accuses on "
"read one and the register answers on read eleven",
not _s5_quiet and _s5.stale and _s5.graded_s(300.0) == 300.0)
check("...and its SIZE is one `stale_input_s` hold, which is why it is recorded "
"here rather than fixed here: `cycle()` returns at `inputs missing "
"{n}s` for the whole of it, so nothing is commanded differently, and the "
"register's next publish clears it. Fixing it means measuring the "
"unchanged clock from when the OBLIGATION appeared rather than from the "
"register's last move, which re-derives the 35.0 s bound and is a "
"different card",
(lambda f: (f.submit(-4.0, 1500.0, 311.0), not f.stale)[-1])(_s5)
and float(option_default("stale_input_s")) >= 5.0)
# ==============================================================================
print()
print("10. SAFETY-57: THE EXIT. The state the verdict creates, and the way out")
# ⚠️ THE ACCEPTANCE CRITERION OF THIS CARD, IN ONE SENTENCE, AND EVERY CHECK
# BELOW IS GRADED AGAINST IT: a scenario in which the register is held CONSTANT
# and the command is gated must recover WITH NO CHANGE TO THE REGISTER. Parts 1
# to 9 grade whether the gate can be ENTERED. A check that only shows a latch
# latching is decoration for this card.
#
# ⚠️ THE COUNTERFACTUAL IS `release_s = inf`, AND IT IS NOT A TRANSCRIPTION.
# `_release()` with an infinite bound never fires, `_open_window()` is then only
# ever called where the register moves, and `graded_from`/`standing_at_grade`
# are `changed_mono`/the command at that change - which is byte-for-byte the
# pre-SAFETY-57 conjunction. That equivalence is ASSERTED in 10b against part
# 1's own fixtures rather than claimed here, because a counterfactual nobody
# graded is a second implementation.
_BEFORE = dataclasses.replace(BOUNDS["batt"], release_s=float("inf"))
def _decoder():
"""The committed decoder, loaded from `tools/` rather than transcribed."""
spec = importlib.util.spec_from_file_location(
"_s57decode", HERE / "tools" / "decode_bus_capture.py")
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
def hms(t):
return f"{int(t // 3600):02d}:{int(t % 3600 // 60):02d}:{int(t % 60):02d}"
def _raises(fn):
try:
fn()
except ValueError:
return True
return False
# `stale_input_s` is READ from the shipped config.yaml, not typed here: the
# parked half of every period below is that option, and a check that carried its
# own copy would go on being true after an operator moved it.
_S57_HOLD = float(option_default("stale_input_s"))
def _dead_site(bound, reads=1201, dead_w=A_FROZEN_W, hold_s=None):
"""A register frozen for good, with §11.1 rule 1 driving the command.
Returns `(freshness, exposure)` where `exposure[i]` is True when read `i`
handed the consumers a number. The command rule is the one `cycle()` uses -
hold the standing command for `stale_input_s`, then 0 W - and the loop's
own ask when telemetry is available; it is written here rather than driven
through `cycle()` because that hold runs on the WALL clock and the bound
under test runs on the injected one. 10g drives the real path for one
period instead.
"""
hold_s = _S57_HOLD if hold_s is None else hold_s
f = Freshness(bound)
standing, bad_since, exposure = dead_w, None, []
for t in range(0, reads):
f.submit(dead_w, standing, float(t))
live = f.reading() is not None
exposure.append(live)
if live:
bad_since = None
standing = 1300.0 if t >= 1 else dead_w
else:
bad_since = t if bad_since is None else bad_since
standing = standing if t - bad_since < hold_s else 0.0
return f, exposure
def _run_lengths(exposure):
"""The lengths of the unbroken runs of True in `exposure`."""
out, n = [], 0
for live in exposure:
if live:
n += 1
elif n:
out.append(float(n))
n = 0
if n:
out.append(float(n))
return out
def _release_instant(release_s):
"""Seconds after the command rests at which the verdict is withdrawn.
One stimulus, one varied term (habit 7): the same 40 s of unfollowed
+1300 W command, then a resting site, under each candidate bound.
"""
f = Freshness(dataclasses.replace(BOUNDS["batt"], release_s=release_s))
for t in range(0, 40):
f.submit(A_FROZEN_W, A_FROZEN_W if t < 1 else 1300.0, float(t))
for t in range(40, 200):
f.submit(A_FROZEN_W, 0.0, float(t))
if not f.stale:
return float(t) - 40.0
return None
# --- 10a. the capture, and what it is allowed to be evidence of --------------
_D = _decoder()
_CAPDIR = HERE / "scenarios" / "safety57-2026-09-21"
_CAPFILE = _CAPDIR / "bus-a-2026-09-21-1030-1110.log.gz"
_man = json.loads(io.open(_CAPDIR / "manifest.json", encoding="utf-8").read())
_events, _census = _D.decode(_CAPFILE)
_sp = _D.series(_events, "setpoint_w")
_bt = _D.series(_events, "batt_w")
_soc = _D.series(_events, "soc_pct")
print(f" capture {_CAPFILE.name}: {_census}")
print(f" setpoint {len(_sp)} frames batt {len(_bt)} frames "
f"soc {len(_soc)} frames, "
f"{hms(_sp[0][0])}..{hms(_sp[-1][0])} Europe/Brussels")
check("PRECONDITION: the committed capture is the file the manifest names, "
"byte for byte. A scenario whose bytes can drift from the numbers quoted "
"off them is a story",
hashlib.sha256(io.open(_CAPFILE, "rb").read()).hexdigest()
== _man["file"]["sha256_gz"]
and _man["file"]["bytes_gz"] == _CAPFILE.stat().st_size)
check("PRECONDITION: every line of it was understood. 3840 frames, and the "
"decoder's own census reports ZERO unparsed, ZERO bad CRC and ZERO "
"unknown shapes - so nothing in the episode below was dropped on the "
"floor by the instrument",
_census["lines"] == 3840 and _census["unparsed"] == 0
and _census["bad_crc"] == 0 and _census["unknown"] == 0
and _census["writes"] == 1200 and _census["responses"] == 720)
# ⚠️ THE SIGN IS THE ONE THING THIS DECODER COULD GET EXACTLY BACKWARDS, and a
# backwards sign would invert the whole episode - a discharge the BMS refused
# would read as a charge it accepted. So the mode/sign correspondence is GRADED
# OFF THE CAPTURE rather than trusted: every charge command is followed by a
# negative register reading, and there is no reading of the other sign.
_after_charge = [
_D.held_at(_bt, t + F.BATT_OBLIGED_S)
for t, v in _sp if v <= -1000.0]
_after_charge = [v for v in _after_charge if v is not None]
print(f" readings {F.BATT_OBLIGED_S:.0f} s after a <= -1000 W command: "
f"{len(_after_charge)} of them, "
f"max {max(_after_charge):+.0f} W, min {min(_after_charge):+.0f} W")
check("⚠️ THE DECODER'S SIGN CONVENTION IS GRADED, NOT ASSERTED. EMS mode 2 is "
"read as charge and mode 3 as discharge; if that were backwards the "
"episode would read as a refused CHARGE. Every command at or under "
"-1000 W in this capture is answered, one obligation window later, by a "
"NEGATIVE register reading - and not one positive one",
len(_after_charge) > 200 and max(_after_charge) < 0.0
and _D.signed_setpoint_w(2, 2170) == -2170.0
and _D.signed_setpoint_w(3, 1060) == +1060.0)
check("...and an EMS mode this decoder has never seen RAISES rather than "
"defaulting to a sign. A silent default is how a firmware change becomes "
"a mis-signed series nobody re-derives",
_raises(lambda: _D.signed_setpoint_w(1, 500))
and _raises(lambda: _D.signed_setpoint_w(0, 0)))
# --- 10b. THE LIVE EPISODE, RED BEFORE AND GREEN AFTER -----------------------
def _replay_capture(bound):
"""The shipped `_read_unit` loop, at 1 Hz, against the capture.
⚠️ THE COMMAND COLUMN IS NOT A MODEL. The capture was taken off a box
running 0.10.0, i.e. running the defect: the 0 W at 10:52:21 IS the site's
own `inputs missing -> command 0 W`, recorded on the wire. So this replay is
closed-loop for the whole of the parked interval, and the only thing in the
window it cannot account for is the human restart at ~11:04.
"""
f = Freshness(bound)
t0 = int(max(_sp[0][0], _bt[0][0])) + 1
t1 = int(min(_sp[-1][0], _bt[-1][0]))
rows, t = [], float(t0)
while t <= t1:
f.submit(_D.held_at(_bt, t), _D.held_at(_sp, t), t)
rows.append((t, f.stale, f.held, _D.held_at(_sp, t), f.releases))
t += 1.0
return f, rows
def _edges(rows):
return [r for i, r in enumerate(rows) if i == 0 or r[1] != rows[i - 1][1]]
_f_before, _rows_before = _replay_capture(_BEFORE)
_f_after, _rows_after = _replay_capture(BOUNDS["batt"])
# ⚠️ THE EPISODE'S BOUNDARIES ARE READ OFF THE CAPTURE, NOT TYPED. It is
# defined by the thing it is about - the register's LONGEST unbroken hold in
# the window - and every other instant is derived from that one. A fixture that
# named 10:52:00 by hand would go on naming it after the file changed.
# ⚠️ CHANGE POINTS, NOT FRAMES. 0x0518 is POLLED every 10 s and answers every
# poll, so consecutive frames are always 10 s apart whatever the register is
# doing - a gap computed over frames measures the poll period and nothing else,
# which is `freshness.py`'s own first paragraph in miniature.
_bt_moves = [r for i, r in enumerate(_bt) if i == 0 or r[1] != _bt[i - 1][1]]
_holds = [(_bt_moves[i + 1][0] - _bt_moves[i][0], _bt_moves[i][0],
_bt_moves[i + 1][0], _bt_moves[i][1])
for i in range(len(_bt_moves) - 1)]
_HOLD_S, _HOLD_FROM, _REG_MOVES, _HOLD_W = max(_holds)
_TRIP = next(t for t, s, _h, _sw, _r in _rows_after if s and t > _HOLD_FROM)
_PARK = next(t for t, v in _sp if t > _TRIP and v == 0.0)
_HUMAN = next(t for t, v in _sp if t > _PARK and v != 0.0)
_blips = [t for t, s, _h, _sw, _r in _edges(_rows_after)
if s and t < _HOLD_FROM]
print(f" longest register hold {_HOLD_S:.1f} s at {_HOLD_W:+.0f} W, "
f"{hms(_HOLD_FROM)}..{hms(_REG_MOVES)}")
print(f" trip {hms(_TRIP)} site commands 0 W {hms(_PARK)} "
f"human restart's first command {hms(_HUMAN)} "
f"register moves {hms(_REG_MOVES)}")
print(f" earlier one-read trips in this window (survivor 5): "
f"{[hms(t) for t in _blips]}")
def _first_clear_after(rows, t_from):
return next((t for t, s, _h, _sw, _r in rows if t > t_from and not s), None)
_clear_before = _first_clear_after(_rows_before, _TRIP)
_clear_after = _first_clear_after(_rows_after, _TRIP)
print(f" BEFORE (release_s=inf): stale {hms(_TRIP)} -> clear "
f"{hms(_clear_before)} {_clear_before - _TRIP:.0f} s")
print(f" AFTER (release_s={BOUNDS['batt'].release_s:.0f}): stale "
f"{hms(_TRIP)} -> clear {hms(_clear_after)} "
f"{_clear_after - _TRIP:.0f} s")
check("PRECONDITION: `release_s=inf` REPRODUCES the pre-SAFETY-57 module and is "
"not a second implementation of it. Part 1's own fresh/stale pair, driven "
"through both bounds, gives bit-identical verdict streams - the release "
"is the only behaviour the counterfactual removes",
drive(_BEFORE, STREAM, lambda t: A_FROZEN_W)[2] == _fresh_read
and drive(_BEFORE, STREAM,
lambda t: A_FROZEN_W if t < 1 else 1300.0)[2] == _stale_read)
check("PRECONDITION: the episode is in the capture where the manifest says it "
"is, and it is the window's WORST hold rather than one picked out of "
"several. 0x0518 answers -3 W for 850.0 s from 10:51:24, the gate arms "
"at 10:52:00, the site's own 0 W reaches the wire at 10:52:21, and "
"nothing else moves until a human does at 11:05:31",
abs(_HOLD_S - 850.0) < 0.1 and _HOLD_W == -3.0
and hms(_HOLD_FROM) == "10:51:24" and hms(_TRIP) == "10:52:00"
and hms(_PARK) == "10:52:21" and hms(_HUMAN) == "11:05:31"
and hms(_REG_MOVES) == "11:05:34"
and sorted(_holds)[-2][0] < 0.5 * _HOLD_S)
check("⚠️ RED BEFORE. On the shipped-before module the verdict outlives the "
"whole parked interval: from the trip to the first command a HUMAN put "
"on the wire, 811 s later, `reading()` is None on every one of the 811 "
"reads - and the register never changes value in any of them, so there "
"is nothing the module could have cleared on",
all(s for t, s, _h, _sw, _r in _rows_before if _TRIP <= t < _HUMAN)
and len([1 for t, s, _h, _sw, _r in _rows_before
if _TRIP <= t < _HUMAN]) == 812
and {v for t, v in _bt if _TRIP <= t < _HUMAN} <= {-3.0}
and _clear_before > _HUMAN)
check("⚠️ AND WHAT ENDS IT ON THE SHIPPED-BEFORE MODULE IS OUTSIDE THE MODULE. "
"The clear lands three seconds after the register moves, and the register "
"moves because a person restarted the add-on and the restarted loop "
"commanded a CHARGE the pack accepted. Nothing inside this add-on "
"produced that command",
_clear_before > _REG_MOVES and _clear_before - _REG_MOVES <= 2.0
and _HUMAN < _REG_MOVES)
check("⚠️ GREEN AFTER, AND THE REGISTER NEVER MOVES - THIS IS THE CARD. The "
"verdict is withdrawn 17 s after the site's own 0 W reaches the wire, "
"while 0x0518 is still answering the same -3 W it has answered since "
"10:51:24, and `reading()` is a number again with no external change of "
"any kind",
_PARK + BOUNDS["batt"].release_s <= _clear_after
< _PARK + BOUNDS["batt"].release_s + 1.0
and hms(_clear_after) == "10:52:39" and _f_after.releases >= 1
and _D.held_at(_bt, _clear_after) == _D.held_at(_bt, _TRIP) == -3.0
and next(r for r in _rows_after if r[0] == _clear_after)[2] == -3.0)
check("...and the recovery is worth 776 s of this one episode: the site is "
"handed its telemetry back at 10:52:39 instead of at 11:05:35, and 772 "
"of those seconds are BEFORE the human touched anything. That is the "
"13 minutes it spent parked while the house exported 2.4 kW",
_clear_before - _clear_after == 776.0
and abs((_HUMAN - _clear_after) - 772.7) < 0.01
and all(not s for t, s, _h, _sw, _r in _rows_after
if _clear_after <= t < _HUMAN))
check("⚠️ AND THE RELEASE STICKS RATHER THAN CHATTERING. From the withdrawal to "
"the human's first command the verdict is withdrawn on every one of the "
"773 reads, and exactly ONE release was taken in the whole episode - the "
"1 Hz oscillation the latch exists to prevent is still prevented, "
"because the window a release opens is 35 s wide",
len([1 for t, s, _h, _sw, _r in _rows_after
if _clear_after <= t < _HUMAN]) == 773
and _f_after.releases == 1)
# --- 10c. what a release costs, measured -------------------------------------
# ⚠️ THE EXPOSURE IS THE NUMBER THE SAFETY ARGUMENT RESTS ON, so it is counted
# rather than reasoned about: how many reads hand a consumer the held number
# after a release, and how long the longest unbroken run of them is. The whole
# argument is that one run is shorter than either consumer's hold.
_dead, _exp57 = _dead_site(BOUNDS["batt"])
_deadb, _expb = _dead_site(_BEFORE)
_runs57 = _run_lengths(_exp57)
print(f" register dead for good, {len(_exp57)} reads, "
f"stale_input_s={_S57_HOLD:.0f}")
print(f" before : held number reached a consumer on {sum(_expb)} reads "
f"({100 * sum(_expb) / len(_expb):.1f} %), runs={_run_lengths(_expb)}")
print(f" after : held number reached a consumer on {sum(_exp57)} reads "
f"({100 * sum(_exp57) / len(_exp57):.1f} %), releases={_dead.releases}, "
f"longest run={max(_runs57):.0f} s")
check("⚠️ THE EXIT IS NOT FREE AND THE PRICE IS STATED AS A NUMBER. A register "
"that is dead for good is re-graded for ever: over 1201 reads the held "
"number reaches a consumer on 630 of them against 35 before, and 17 "
"releases are taken. This check exists so that cost cannot be quietly "
"forgotten, and so a change that makes it worse has to move a number "
"somebody reads",
sum(_expb) == 35 and sum(_exp57) == 630 and _dead.releases == 17
and _deadb.releases == 0)
check("⚠️ AND IT IS PAYABLE ONLY BECAUSE ONE RUN IS SHORTER THAN EITHER "
"CONSUMER'S HOLD. Every unbroken run of exposure is exactly one 35.0 s "
"grading window, and 35 < 75 (circulation_hold_s) < 85 "
"(unit_sign_disagree_s) - so NEITHER detector can reach a verdict inside "
"one, and the re-trip that ends each run hands them a None, which CLEARS "
"their windows rather than accumulating into them",
set(_runs57) == {BOUNDS["batt"].unchanged_s}
and BOUNDS["batt"].unchanged_s < _circ_hold < _unit_hold)
check("...and nothing SAFETY-37 detects is given back by that, which is the "
"half that makes the trade a trade. A permanent None clears both windows "
"for ever too, so on the shipped-before module the same two detectors are "
"equally unable to speak about this unit. The exposure buys the site its "
"meter-derived control back and costs no detection that existed",
disagrees(1300.0, None, True, _cfg3) is False
and circulating_w([1300.0, None]) is None)
# --- 10d. the bound, graded as a family --------------------------------------
# ⚠️ ASSERTING `release_s == BATT_OBLIGED_S` WOULD BE COMPARING THE SHIPPED
# EXPRESSION AGAINST ITSELF (habit 2, and MAINT-05's whole card). The literal is
# asserted, the three terms it is built from are asserted, and then the BAND is
# graded behaviourally against its neighbours.
print(f" RELEASE_S {F.RELEASE_S:.1f} s = settle {F.PLANT_SETTLE_S:.1f} "
f"+ publish {F.PUBLISH_PERIOD_S:.1f} + read {F.READ_PERIOD_S:.1f}")
check("⚠️ THE BOUND IS 17.0 s AND IT IS THE PHYSICAL OBLIGATION WINDOW, "
"asserted against the literal rather than against the expression that "
"produces it. Same three terms as the floor of `BATT_UNCHANGED_S`'s "
"band, because it is the same question: has the register had its chance "
"to answer the command standing NOW?",
BOUNDS["batt"].release_s == 17.0 and F.RELEASE_S == 17.0
and F.PLANT_SETTLE_S + F.PUBLISH_PERIOD_S + F.READ_PERIOD_S == 17.0)
_family = {r: _release_instant(r) for r in (5.0, 16.0, 17.0, 18.0, 35.0,
float("inf"))}
for _r, _at in sorted(_family.items(), key=lambda kv: kv[0]):
print(f" release_s {_r:>5} -> withdrawn at t+{_at}"
if _at is not None else f" release_s {_r:>5} -> never")
check("⚠️ THE FAMILY IS COMPUTED AND PRINTED, so a bound no fixture separates "
"shows up as a named survivor. Each candidate withdraws the verdict "
"exactly its own number of seconds after the command rests, and they are "
"six DISTINCT answers - a check that could not tell 17 from 35 could not "
"tell this card from a slower version of the bug",
_family == {5.0: 5.0, 16.0: 16.0, 17.0: 17.0, 18.0: 18.0, 35.0: 35.0,
float("inf"): None})
# ⚠️ AND THE FLOOR IS GRADED BEHAVIOURALLY, WHICH IS THE HALF A CONSTANT CHECK
# CANNOT DO. Below 17.0 s the timer fires before a healthy register could have
# published its answer to the 0 W command - so it takes the expensive exit when
# the free one was one read away. The register here answers at the obligation
# window's own edge, which is the case the bound is derived from.
_free_exit = {}
for _r in (5.0, 17.0):
_f = Freshness(dataclasses.replace(BOUNDS["batt"], release_s=_r))
for t in range(0, 40):
_f.submit(A_FROZEN_W, A_FROZEN_W if t < 1 else 1300.0, float(t))
for t in range(40, 40 + int(F.BATT_OBLIGED_S) - 1):
_f.submit(A_FROZEN_W, 0.0, float(t)) # resting, register not yet in
_f.submit(-1150.0, 0.0, float(40 + int(F.BATT_OBLIGED_S) - 1))
_free_exit[_r] = (_f.releases, _f.stale, _f.held)
print(f" register answers the 0 W command at +16 s: {_free_exit}")
check("⚠️ THE FLOOR BITES, AND IT BITES ON BEHAVIOUR. A register that answers "
"the new 0 W command inside its own obligation window clears the latch "
"BY MOVING - the free exit - and at 17.0 s that is what happens, with "
"`releases` still 0. At 5.0 s the timer has already fired and the site "
"has paid for a re-grading window it did not need",
_free_exit[17.0][0] == 0 and _free_exit[5.0][0] == 1
and _free_exit[17.0][2] == _free_exit[5.0][2] == -1150.0)
# --- 10e. the release does not rewrite the register's history ----------------
_hist = Freshness(BOUNDS["batt"])
for t in range(0, 40):
_hist.submit(A_FROZEN_W, A_FROZEN_W if t < 1 else 1300.0, float(t))
for t in range(40, 120):
_hist.submit(A_FROZEN_W, 0.0, float(t))
print(f" at t=119 after one release: unchanged_s={_hist.unchanged_s(119.0)} "
f"graded_s={_hist.graded_s(119.0)} releases={_hist.releases}")
check("⚠️ TWO CLOCKS, AND THE ONE THE OPERATOR READS IS THE REGISTER'S. A "
"release re-opens the VERDICT's window; it does not make the register "
"younger. At t=119 the register has not carried a different number for "
"119 s and `unchanged_s` says 119 s, while the grading window opened at "
"the release at t=57 and says 62 s. A single clock here would either lie "
"to the operator or defeat the re-grading",
_hist.unchanged_s(119.0) == 119.0 and _hist.graded_s(119.0) == 62.0
and _hist.releases == 1 and _hist.held == A_FROZEN_W
and _hist.changed_mono == 0.0 and _hist.graded_from == 57.0)
# --- 10f. the obligation coming back cancels the timer -----------------------
_resume = Freshness(BOUNDS["batt"])
for t in range(0, 40):
_resume.submit(A_FROZEN_W, A_FROZEN_W if t < 1 else 1300.0, float(t))
for t in range(40, 40 + int(BOUNDS["batt"].release_s) - 1):
_resume.submit(A_FROZEN_W, 0.0, float(t)) # 16 s of rest
_resume.submit(A_FROZEN_W, 1300.0, 56.0) # the command comes back
_resume_mid = (_resume.stale, _resume.rested_since)
for t in range(57, 57 + int(BOUNDS["batt"].release_s) - 1):
_resume.submit(A_FROZEN_W, 1300.0, float(t)) # and stays
check("⚠️ A TIMER THAT DOES NOT RESET IS NOT A TIMER. Sixteen seconds of rest, "
"one second short of the bound, and then the site puts the obligation "
"back: the clock is cleared, not carried, so the verdict stands. "
"Carrying it would withdraw a verdict about a command that is standing "
"RIGHT NOW",
_resume_mid == (True, None) and _resume.stale
and _resume.releases == 0 and _resume.reading() is None)
_resume.submit(A_FROZEN_W, 0.0, 80.0)
for t in range(81, 81 + int(BOUNDS["batt"].release_s)):
_resume.submit(A_FROZEN_W, 0.0, float(t))
check("...and it starts again from the SECOND rest, not from the first: the "
"release lands 17.0 s after t=80, which is 41 s after the first rest "
"began. The total time spent resting is not what the bound is about",
_resume.releases == 1 and _resume.released_at == 97.0
and not _resume.stale)
# ⚠️ AND "NO LONGER STANDING" IS `MOVE_W`, NOT SOME OTHER THRESHOLD IN THE SAME
# OBJECT. Found by mutating the SIBLING (habit 11): `_release()`'s test against
# `bound.move_w` swapped for `bound.unchanged_s` SURVIVED the whole file at
# +5 B, because every fixture above rests at exactly 0 W and 0 is below both
# 35 and 250. The band between them is where they separate, and a site can sit
# in it - a peak-guard floor, a shadow substitution or any claim that wins with
# a small non-zero number leaves a command there while the verdict stands.
_band = {}
for _w in (0.0, 100.0, 300.0):
_f = Freshness(BOUNDS["batt"])
for t in range(0, 40):
_f.submit(A_FROZEN_W, A_FROZEN_W if t < 1 else 1300.0, float(t))
for t in range(40, 40 + 2 * int(BOUNDS["batt"].release_s)):
_f.submit(A_FROZEN_W, _w, float(t))
_band[_w] = _f.releases
print(f" releases after 34 s of a standing command of 0 / 100 / 300 W: "
f"{_band}")
check("⚠️ THE OBLIGATION TEST IS THE 250 W ONE, AND THE 100 W CASE IS WHAT "
"SAYS SO. A command of 100 W is below `move_w`, so the register is owed "
"nothing and the verdict is withdrawn exactly as at 0 W; a command of "
"300 W is above it, so the question is still standing and the verdict "
"is too. Every other fixture in this file rests at 0 W, where `move_w` "
"and `unchanged_s` cannot be told apart",
_band == {0.0: 1, 100.0: 1, 300.0: 0}
and BOUNDS["batt"].unchanged_s < 100.0 < BOUNDS["batt"].move_w < 300.0)
# ⚠️ AND THE ONE EQUIVALENT MUTANT THIS CARD FOUND, ARGUED RATHER THAN ROUNDED
# OFF. Swapping `_release()` and `_grade()` in `submit()` is ZERO BYTES and
# survived the whole battery. It is not a missing fixture and no fixture can
# close it: the two act on MUTUALLY EXCLUSIVE predicates - `_grade()` only when
# `|standing| >= move_w`, `_release()` only when `|standing| < move_w` - so at
# most one of them can do anything on any read. This check pins the disjointness
# rather than the order, which is the property that actually holds.
_disjoint = []
for _w in (0.0, 100.0, 249.0, 250.0, 251.0, 1300.0):
_a = Freshness(BOUNDS["batt"])
_b = Freshness(BOUNDS["batt"])
for t in range(0, 40):
for _f in (_a, _b):
_f.submit(A_FROZEN_W, A_FROZEN_W if t < 1 else 1300.0, float(t))
for t in range(40, 80):
_a._release(_w, float(t))
_a._grade(_w, float(t))
_b._grade(_w, float(t))
_b._release(_w, float(t))
_disjoint.append((_w, _a.stale == _b.stale
and _a.releases == _b.releases
and _a.graded_from == _b.graded_from))
print(f" release-then-grade vs grade-then-release, by standing command: "
f"{_disjoint}")
check("⚠️ AN ARGUED EQUIVALENT MUTANT, NOT A SURVIVOR NOBODY LOOKED AT. "
"Swapping the two calls in `submit()` is zero bytes and changes nothing "
"on either side of `move_w`, including AT it - because the two are gated "
"on mutually exclusive tests and cannot both act on one read. The "
"docstring used to claim the order was load-bearing; it is not, and the "
"battery is what said so",
all(same for _w, same in _disjoint) and len(_disjoint) == 6)
# --- 10g. through the REAL read path, and the operator is told ---------------
_LOGREC = []
class _S57Handler(logging.Handler):
def emit(self, record):
_LOGREC.append(record)
logging.getLogger("goodwe").addHandler(_S57Handler())
logging.getLogger("goodwe").setLevel(logging.DEBUG)
async def _wired_release():
"""The whole chain at N=1: real `read_inputs`, real `Inverter`, real log."""
c, hass = one_site()
seen = []
for t in range(0, 40):
hass.numbers["sensor.batt_a"] = A_FROZEN_W
c.inverters[0].last_write = A_FROZEN_W if t < 1 else 1300.0
await c.read_inputs(now_mono=float(t))
seen.append((t, c.inverters[0].batt, c.inputs_ok))
for t in range(40, 40 + int(BOUNDS["batt"].release_s) + 2):
hass.numbers["sensor.batt_a"] = A_FROZEN_W # the register NEVER moves
c.inverters[0].last_write = 0.0 # the site's own 0 W
await c.read_inputs(now_mono=float(t))
seen.append((t, c.inverters[0].batt, c.inputs_ok))
return c, seen
_c57, _seen57 = run(_wired_release())
_trip57 = next(t for t, b, _ok in _seen57 if b is None)
_back57 = next(t for t, b, _ok in _seen57 if t > _trip57 and b is not None)
print(f" real read path: batt None from t={_trip57}, a number again at "
f"t={_back57}, register held {A_FROZEN_W:+.0f} throughout")
check("⚠️ THROUGH THE SHIPPED READ PATH, NOT THE POLICY OBJECT. "
"`Inverter.batt` goes None at the bound, `inputs_ok` goes False - and "
"17.0 s after the site's own 0 W it is a NUMBER again and `inputs_ok` is "
"True, with `sensor.batt_a` serving the identical -1200.0 on every one "
"of those reads",
_trip57 == 35 and _back57 == 57
and all(v == A_FROZEN_W for _e, v in _c57.hass.served
if _e == "sensor.batt_a")
and _seen57[_trip57][2] is False and _seen57[-1][2] is True
and _c57.inverters[0].batt == A_FROZEN_W)
check("...and the site stops holding inputs bad, which is what ends the "
"`inputs missing -> command 0 W` claim §11.1 rule 1 files",
_c57.inputs_bad_since is None and _c57.inputs_ok is True)
_ev57 = [e for e in _c57.events if "STALE" in e]
_msgs57 = [r.getMessage() for r in _LOGREC]
_rel57 = [m for m in _msgs57 if "STALE_WITHDRAWN" in m]
print(f" events: {[e.split(' ', 1)[1][:44] for e in _ev57]}")
check("⚠️ A LATCH THAT LETS GO ON A TIMER AND SAYS NOTHING IS ONE NO OPERATOR "
"CAN AUDIT - and this one is withdrawing a SITE STOP. Both edges reach "
"the event log and the logger, in order, naming the unit",
len(_ev57) == 2 and "STALE inverter-a" in _ev57[1]
and "STALE WITHDRAWN inverter-a" in _ev57[0]
and len(_rel57) == 1 and "inverter-a" in _rel57[0])
check("⚠️ AND THE RELEASE LINE REFUSES TO CLAIM A RECOVERY, because none has "
"happened: it says in its own words that nothing has been learned about "
"the register, and it prints the value the register is STILL holding",
"NOTHING HAS BEEN LEARNED ABOUT THE REGISTER" in _rel57[0]
and f"still holding {A_FROZEN_W:+.0f}" in _rel57[0])
# --- 10h. the two sentences, as ORDERED PAIRS (habit 14) ---------------------
# ⚠️ AN `in` TEST AGAINST A FORMATTED SENTENCE GRADES PRESENCE, NEVER
# ARRANGEMENT. Both of the sentences this card writes carry two numbers that
# could be transposed at zero bytes - 17.0 s and 250 W in the stale line's new
# escape clause, and 250 W and 17.0 s again in the release note - so each is
# asserted with its value embedded in its own literal position.
_stale57 = [m for m in _msgs57 if "SAFETY-37" in m]
print(f" escape clause: ...{_stale57[0][-150:]}")
check("⚠️ THE STALE LINE'S PROMISE IS NOW TRUE, AND IT IS PINNED AS AN ORDERED "
"PAIR. It used to end 'It clears itself the moment the register reports "
"a different value.' - which was FALSE, because the 0 W in the sentence "
"before it is what stops the register being asked. The new clause names "
"17 s in the TIME slot and 250 W in the POWER slot; swapped, this goes "
"red",
f"failing that {BOUNDS['batt'].release_s:.0f} s after the standing "
f"command falls below {BOUNDS['batt'].move_w:.0f} W" in _stale57[0])
check("...and the release note names 250 W in the POWER slot and 17 s in the "
"TIME slot, in that order, which is the reverse arrangement of the same "
"two numbers",
f"sat below {BOUNDS['batt'].move_w:.0f} W for "
f"{BOUNDS['batt'].release_s:.0f} s" in _rel57[0])
check("...and the 35.0 s the register is re-graded over is named in the note "
"too, because it is what the operator is being told to wait out",
f"re-graded over a fresh {BOUNDS['batt'].unchanged_s:.0f} s window"
in _rel57[0])
check("⚠️ AND THE DIAGNOSIS HALF OF THE STALE LINE IS UNTOUCHED, WHICH IS "
"SAFETY-56'S. That card is rewriting 'is not following its own command' "
"- the live episode's register was following perfectly and the pack was "
"simply empty - and two cards editing one sentence is how one of them "
"silently loses. This check is here so that rewrite is a deliberate act",
"battery-power register is not following its own command" in _stale57[0])
check("...and a re-trip's reason names the TWO durations separately, each "
"beside its own value, because after a release the verdict's window and "
"the register's own unchanged time are different facts",
(lambda f: (
[f.submit(A_FROZEN_W, A_FROZEN_W if t < 1 else 1300.0, float(t))
for t in range(0, 40)],
[f.submit(A_FROZEN_W, 0.0, float(t)) for t in range(40, 80)],
[f.submit(A_FROZEN_W, 1300.0, float(t)) for t in range(80, 130)],
f.stale and f.releases == 1
and "unchanged for 92 s" in f.reason
and "the 35 s window this verdict rests on opened when the previous "
"one was withdrawn" in f.reason
and "graded 2 times" in f.reason)[-1])(
Freshness(BOUNDS["batt"])))
print()
if fails: