admin/goodwe-addon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
PKG-18-cut-0.12.0
goodwe-addon/goodwe_controller/test_freshness.py
T
adminandClaude Opus 5 8a6acd9fa2
checks / unit (pull_request) Successful in 8m0s
Details
SAFETY-56: the gate cannot tell a dead register from a pack that cannot comply, so stop saying it can 
Text only. No behaviour change - same conjunction, same limits, same MISSING,
same 0 W, same NOT READY. `tools/safety56_behaviour_trace.py` says so: 2390
capture reads, a 66-cell swept grid, and six driven controller worlds produce
the byte-identical document `6030175e...` on origin/release/1.0 and on this
branch, while the sentences differ.

THE DEFECT. SAFETY-37's conjunction is satisfied by two disjoint worlds - the
register has stopped reporting reality, or the SOURCE cannot comply with the
command - and nothing `freshness.py` is handed separates them. Three sentences
asserted the first. On the live 2026-09-21 episode it was the second: the pack
read 11 % for the whole 850 s hold, every command in that hold was a discharge,
and the register answered the first CHARGE command three seconds later. The
readiness row was worse still, because it asserted a false fact about a NAMED
entity that was present, numeric and updating - so the operator opened
Developer Tools, found it working, and had every reason to conclude the
diagnostics were broken.

THE FIX. Each message states what was OBSERVED and names BOTH explanations.
Three wordings, no shared helper: two are a log line about a behaviour, the
third is a banner about an entity, and a helper serving both would drop
whichever half the other caller does not need. SoC stays STATED, NOT WIRED -
part 11b drives the two worlds at 50 % and 11 % and the sentence does not move
one byte. The row does not branch on `batt_fresh.stale` either: `stale_note()`
already rides `liveness_notes()` a few rows below, so the discriminator is "is
that row there?", and part 11d asserts it is present in one world and absent in
the other.

THE CHECKS. test_freshness.py part 11 builds the pair of worlds with ground
truth written down rather than derived from the published stream, asserts they
are really two (40 of 40 reads differ) while the published stream is
bit-identical, and only then grades the sentences. Eleven mutations, all red:
each of the three sentences reverted (3, 2 and 2 named reds), the two
explanations transposed at ZERO bytes (1), the second explanation deleted (1),
the admission of ambiguity replaced by a ranking (1), all three reverted at
once (5), and two aimed at this card's own check - collapsing the world pair
(2) and emptying the accusation list, which a positive control against the
three sentences as 0.10.1 shipped them catches (1).

AND TWO OF THE ELEVEN FOUND DEFECTS IN THE CHECK RATHER THAN IN THE CODE.
Transposing `label` and `entity` in the row made it unfindable and the run died
on an IndexError - a red with no name and every later check ungraded, FCST-06's
shape exactly; `_s56_row()` returns a named absence now, 0 named reds before, 3
after. And the first invariance run compared a tree with itself, because it
archived a branch whose edits were not committed yet; the tool now REFUSES to
report "behaviour unchanged" unless the sentences halves differ, and both of
its refusals are driven.

SAFETY-57's part 10h check was the handoff and is inverted rather than deleted:
it required the accusation to still be present so that removing it had to be a
deliberate act. This is that act. Its escape clause is untouched.

Suite: 30/30 in the CI image, checks.yml's exact command, purged tree, :ro.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BwRLw7eKYj2byVdwB6kbjU
2026-09-23 21:02:40 +02:00

1978 lines
104 KiB
Python
Raw Permalink Blame History

"""Runnable check for SAFETY-37's freshness policy. `python3 test_freshness.py`
THE SUBJECT OF THIS FILE IS THE FIXTURE, NOT THE CODE. A pack legitimately
resting at 40 % and a dead sensor reading 40 % ARE THE SAME NUMBER. A check that
cannot separate "fresh and unchanged" from "stale" cannot see the failure it
names, and this repository has recorded that defect - "a check whose DATA cannot
reach the failure it names" - eleven times. So part 1 builds the separation and
ASSERTS THE VALUE STREAMS ARE BIT-IDENTICAL across the pair; everything after it
is downstream of that pair being real.
⚠️ AND THE CARD'S WORDING HAD TO BE SHARPENED TO BUILD IT, WHICH IS SAID HERE
RATHER THAN QUIETLY WORKED AROUND. The card says "only arrival separates them".
For a signal this add-on polls out of Home Assistant's state cache, ARRIVAL
CANNOT SEPARATE THEM: `read_inputs()` runs about once a second and Home
Assistant answers 200 with a number whether or not the publisher behind the
entity is alive, so both members of the pair arrive, identically, forever.
Part 1a demonstrates exactly that - an arrival-age gate is graded and FAILS to
tell the pair apart - and it is the reason the shipped verdict is a conjunction
with an OBLIGATION term rather than an age threshold. Every clock in that
verdict is still an arrival clock: nothing reads a Home Assistant timestamp,
which is the half of "arrival-stamped" that TEL-06 is blocked on and that this
module does honour.
1. THE PAIR. Fresh-and-unchanged vs stale, on identical value streams, on
the policy object itself. Plus 1a, the arrival-age gate that cannot.
2. THE WIRING. The same pair driven through the REAL `Controller.read_inputs`
at a roster of TWO, with the two units carrying DIFFERENT power values.
3. THE CONSUMERS REFUSE, and the counterfactual is run beside them: the
per-unit sign check and the circulating-power sensor, with the gate and
with the gate removed, through real `cycle()` calls.
4. THE LATCH, and the ~1 Hz oscillation it exists to prevent.
5. N=1. Nothing changes on a healthy install - asserted on the shipped read
path, not assumed - and the one thing that DOES change is named.
6. THE BOUNDS, graded against the shipped `config.yaml` at both ends.
7. THE POLICY TABLE tells the truth about what is wired.
8. THE MEASUREMENT the bound rests on, replayed from the committed capture.
9. KNOWN SURVIVORS, graded so they are residuals rather than surprises.
10. SAFETY-57's EXIT from the verdict, against the committed live capture.
11. SAFETY-56: THE SENTENCE THIS GATE CAN HONESTLY SAY. The conjunction is
satisfied by two disjoint worlds - a register that has stopped reporting,
and a source that cannot comply - and nothing here separates them. Part
11 builds both, shows the verdict is byte-identical across them, and
grades the three operator-facing sentences against that ambiguity.
⚠️ INVERTER B DOES NOT PHYSICALLY EXIST (~2026-09-12; goodwe/NOTES.md §1.1b).
Every two-unit case below is two copies of inverter A's behaviour with
DIFFERENT numbers passed in, and no number here is a measured parameter for a
second device.
⚠️ 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. Read MUTATION.md
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
from pathlib import Path
from app import freshness as F
from app.circulation import circulating_w
from app.freshness import BOUNDS, BATT_OBLIGED_S, BATT_UNCHANGED_S, MOVE_W, Freshness
from app.inverter import Inverter
from app.main import Controller
from app.store import DEFAULTS
from app.watchdog import SignWatchdog, SignWatchdogConfig, disagrees
fails = []
HERE = Path(__file__).resolve().parent
def check(name, cond):
print(f" {'ok ' if cond else 'FAIL'} {name}")
if not cond:
fails.append(name)
def run(coro):
return asyncio.new_event_loop().run_until_complete(coro)
# --------------------------------------------------------------------------- #
# fixtures
# --------------------------------------------------------------------------- #
class MemStore:
def __init__(self, data=None):
self.data = dict(DEFAULTS, **(data or {}))
def set(self, k, v):
self.data[k] = v
def get_time(self, *a):
return None
def set_time(self, *a):
pass
class Pub:
enabled = False
def publish(self, *a, **k):
pass
def close(self):
pass
class Hass:
"""A Home Assistant that answers per ENTITY and RECORDS what it served.
⚠️ PER ENTITY, AND THE RECORD IS LOAD-BEARING. A stub that answered one
number whatever it was asked could not tell "B was read from B's sensor"
from "B was handed A's reading" - and the served-value log is what lets
part 1 assert that the fresh run and the stale run saw the SAME numbers.
"""
def __init__(self, numbers=None, limits=None):
self.numbers = dict(numbers or {})
self.limits_by = dict(limits or {})
self.served = [] # (entity, value) in call order
self.writes = []
async def number(self, entity, invert=False):
v = self.numbers.get(entity)
self.served.append((entity, v))
return None if v is None else (-v if invert else v)
async def limits(self, entity):
return self.limits_by.get(entity, (None, None, None))
async def set_number(self, entity, value):
self.writes.append((entity, value))
return True
async def call(self, domain, service, data):
return True
A_RATING = 5000.0
# ⚠️ ILLUSTRATIVE, AND NOT B'S DATASHEET FIGURES. SPIKE-04a established B as a
# GW5048D-ES rated 4600 W with a 10 % floor (goodwe/NOTES.md §1.1b), and this
# fixture deliberately does not use them: 5000/4600 is a 400 W gap, narrower
# than a slew step, and two equal-ish numbers cannot show a check that read the
# wrong unit's. Same reasoning as test_inverter.py's `B_RATING_HYPOTHESIS`.
B_RATING_SIM = 3000.0
A_FLOOR, B_FLOOR_SIM = 10, 15
ROSTER = [
{"name": "inverter-a", "setpoint_entity": "number.sp_a",
"soc_entity": "sensor.soc_a", "batt_entity": "sensor.batt_a",
"max_w": A_RATING, "bms_soc_floor": A_FLOOR},
{"name": "inverter-b", "setpoint_entity": "number.sp_b",
"soc_entity": "sensor.soc_b", "batt_entity": "sensor.batt_b",
"max_w": B_RATING_SIM, "bms_soc_floor": B_FLOOR_SIM},
]
# ⚠️ A AND B CARRY DIFFERENT POWER VALUES IN EVERY FIXTURE BELOW, and the card
# names this as the trap it expects: "if both read the same, the test cannot
# distinguish a frozen reading from a live one and will pass against the broken
# code". SAFETY-30's fix round walked into exactly that with two watchdog holds
# both set to 1.
A_FROZEN_W = -1200.0 # inverter A's register, stuck on a charge figure
B_LIVE_W = 900.0 # inverter B's register, moving, and NOT A's number
BASE = dict(meter_source="off", meter_entity="sensor.grid", setpoint_entity="",
soc_entity="", batt_entity="", max_w=A_RATING, gain=0.6, slew_w=0,
deadband_w=15, target_grid_w=0, step_w=10, heartbeat_s=10,
stale_input_s=15, inverters=ROSTER)
def site(opts=None, numbers=None):
"""A real two-inverter Controller off the shipped `build_roster()`."""
hass = Hass(dict({"sensor.grid": 0.0,
"sensor.soc_a": 50.0, "sensor.batt_a": A_FROZEN_W,
"sensor.soc_b": 55.0, "sensor.batt_b": B_LIVE_W},
**(numbers or {})),
{"number.sp_a": (-A_RATING, A_RATING, 1.0),
"number.sp_b": (-B_RATING_SIM, B_RATING_SIM, 1.0)})
c = Controller(dict(BASE, **(opts or {})), hass, MemStore(), Pub(),
stop=asyncio.Event())
c.inverters[0].dev_min, c.inverters[0].dev_max = -A_RATING, A_RATING
c.inverters[1].dev_min, c.inverters[1].dev_max = -B_RATING_SIM, B_RATING_SIM
return c, hass
# ==============================================================================
print()
print("1. THE PAIR: 'fresh and unchanged' vs 'stale', on IDENTICAL value streams")
# The two runs below feed the policy object the SAME forty readings of the same
# number at the same instants. The ONLY difference is what the controller was
# asking the inverter to do while those readings arrived.
#
# FRESH the site is converged. The command sits where it is, the register
# answers the same watt, and nothing is wrong. This is the case the
# 30-day capture in part 8 measures at up to 263.2 s.
# STALE the site reversed. The command travelled 2500 W and the register did
# not move by one watt in 40 s, which is 2.4x the fastest it could
# possibly be obliged to answer in.
STREAM = [(float(t), A_FROZEN_W) for t in range(0, 40)]
def drive(bound, stream, standing_of):
"""Feed a Freshness one arrival per (t, value). Returns (obj, served, verdicts)."""
f = Freshness(bound)
served, verdicts = [], []
for t, v in stream:
f.submit(v, standing_of(t), t)
served.append(v)
verdicts.append(f.reading())
return f, served, verdicts
_fresh, _fresh_served, _fresh_read = drive(
BOUNDS["batt"], STREAM, lambda t: A_FROZEN_W)
# The command travels from -1200 W to +1300 W at t=1 and stays there: a site
# turnaround, which is the ordinary stimulus SAFETY-31's own campaign sweeps.
_stale, _stale_served, _stale_read = drive(
BOUNDS["batt"], STREAM, lambda t: A_FROZEN_W if t < 1 else 1300.0)
print(f" served (both runs): {_fresh_served[:3]} ... x{len(_fresh_served)}")
# ⚠️ `key=str`, AND IT IS NOT COSMETIC. Without it this line raises TypeError
# comparing None with a float the moment a mutation makes the FRESH run stale -
# which is what M2 (dropping the command-travel term) did: the file died on a
# DIAGNOSTIC PRINT before check one, so the harness reported "red" with no named
# check behind it. MUTATION.md's reporting rule is "which suite went red, and on
# WHICH NAMED CHECK"; a print that can crash first destroys exactly that.
print(f" fresh verdicts: unique={sorted(set(_fresh_read), key=str)}")
print(f" stale verdicts: first None at t="
f"{next((i for i, v in enumerate(_stale_read) if v is None), None)}")
check("⚠️ THE FIXTURE'S OWN PRECONDITION: the two runs saw a BIT-IDENTICAL "
"value stream. If this ever stops being true the pair below is comparing "
"two different experiments and proves nothing about freshness",
_fresh_served == _stale_served
and len({v for v in _fresh_served}) == 1)
check("FRESH AND UNCHANGED is not stale. Forty identical readings over 39 s "
"with the command standing still - a converged site - and the reading "
"reaches its consumers untouched",
not _fresh.stale and _fresh_read == [A_FROZEN_W] * len(STREAM))
check("⚠️ STALE IS. The same forty readings, the same instants, the same "
"numbers - and the command moved 2500 W underneath them. The reading "
"becomes None to its consumers, which is the whole card",
_stale.stale and _stale_read[-1] is None)
check("...and it becomes None at the BOUND, not before: 35.0 s of unchanged "
"arrivals, so the last fresh verdict is at t=34 and the first None at "
"t=35",
_stale_read[34] == A_FROZEN_W and _stale_read[35] is None
and BATT_UNCHANGED_S == 35.0)
check("...and NEVER a held number and never a zero - `reading()` has exactly "
"two shapes, the arrived value or None",
set(type(v) for v in _stale_read) == {float, type(None)}
and 0.0 not in [v for v in _stale_read if v is not None])
check("...and the reason names the three terms an operator has to check, with "
"the numbers that fired it",
"STALE_READING" in _stale.reason and "2500 W" in _stale.reason
and "unchanged for 35 s" in _stale.reason)
# --- 1a. the gate the card asked for first, graded and found unable ----------
# An arrival-age gate is what "arrival-stamped" reads as on first hearing. It is
# built here and run against the SAME pair, because a criterion that cannot be
# satisfied is worth more said out loud than approximated in silence.
_arrival_ages = []
_af = Freshness(BOUNDS["batt"])
for t, v in STREAM:
_af.submit(v, A_FROZEN_W, t)
# Every read produced a number, so on both members of the pair the age
# since the last ARRIVAL is 0.0 at every instant. There is nothing to
# threshold.
_arrival_ages.append(0.0 if _af.last is not None else None)
check("⚠️ AN ARRIVAL-AGE GATE CANNOT TELL THE PAIR APART, and that is why the "
"shipped verdict is not one. Home Assistant answered every read with a "
"number in BOTH runs, so the age since arrival is 0.0 s at every instant "
"of both - a threshold on it is a threshold that can never fire, which "
"is TEL-06's defect rebuilt in this repository",
set(_arrival_ages) == {0.0} and _af.arrivals == len(STREAM))
check("...while the arrival record IS what the shipped verdict is measured on: "
"`unchanged_s` is the gap between arrivals that carried different "
"numbers, on this process's own monotonic clock, and no Home Assistant "
"timestamp is read anywhere in the module",
_stale.unchanged_s(39.0) == 39.0
and "last_updated" not in io.open(HERE / "app" / "freshness.py",
encoding="utf-8").read().split(
'"""', 2)[2])
check("...and a read that produced NOTHING is still an immediate None, exactly "
"as it is today - a failed read is not an arrival and is not an age",
Freshness(BOUNDS["batt"]).reading() is None
and (lambda f: (f.submit(5.0, 0.0, 0.0), f.submit(None, 0.0, 1.0),
f.reading() is None and f.held == 5.0)[-1])(
Freshness(BOUNDS["batt"])))
# ==============================================================================
print()
print("2. THE WIRING: the same pair through the REAL read path, at a roster of 2")
async def _reads(c, hass, plan):
"""Drive real `read_inputs()` calls. `plan` is (t, standing_a, batt_a)."""
out = []
for t, standing_a, batt_a in plan:
hass.numbers["sensor.batt_a"] = batt_a
# `last_write` is what the write path sets and what `standing_w()`
# reads; setting it here is standing in for a write, not for the gate.
c.inverters[0].last_write = standing_a
await c.read_inputs(now_mono=t)
out.append((t, c.inverters[0].batt, c.inverters[1].batt))
return out
_PLAN_FRESH = [(float(t), A_FROZEN_W, A_FROZEN_W) for t in range(0, 40)]
_PLAN_STALE = [(float(t), A_FROZEN_W if t < 1 else 1300.0, A_FROZEN_W)
for t in range(0, 40)]
_cf, _hf = site()
_of = run(_reads(_cf, _hf, _PLAN_FRESH))
_cs, _hs = site()
_os = run(_reads(_cs, _hs, _PLAN_STALE))
print(f" fresh run: A={sorted({r[1] for r in _of})} B={sorted({r[2] for r in _of})}")
print(f" stale run: A={sorted({r[1] for r in _os}, key=str)} "
f"B={sorted({r[2] for r in _os})}")
check("PRECONDITION: both runs were served the same per-entity numbers, so the "
"difference below is the wiring and not the stimulus",
[x for x in _hf.served if x[0] == "sensor.batt_a"]
== [x for x in _hs.served if x[0] == "sensor.batt_a"])
check("⚠️ THE TWO UNITS CARRY DIFFERENT POWER VALUES - A at -1200 W, B at "
"+900 W. A fixture whose units read the same number cannot show a gate "
"that answered for the wrong unit, which is the trap the card names",
A_FROZEN_W != B_LIVE_W and all(r[2] == B_LIVE_W for r in _of))
check("THROUGH THE REAL PATH: a converged site keeps its reading",
all(r[1] == A_FROZEN_W for r in _of))
check("⚠️ THROUGH THE REAL PATH: a frozen register becomes `Inverter.batt is "
"None`. Not a held number, not a zero - the same None a missing entity "
"produces, which is the state §11.1 rule 1 already answers with 0 W",
_os[34][1] == A_FROZEN_W and _os[35][1] is None
and all(r[1] is None for r in _os[35:]))
check("...and ONLY the frozen unit's. B's own reading is untouched, which is "
"the per-unit half: a freshness verdict about A is not evidence about B",
all(r[2] == B_LIVE_W for r in _os) and _cs.inverters[1].batt == B_LIVE_W
and not _cs.inverters[1].batt_fresh.stale)
check("...and `inputs_ok` goes False, so the site stops rather than regulating "
"against a number nobody measured",
_cs.inputs_ok is False and _cf.inputs_ok is True)
check("...and the operator gets a line saying which unit and why, on the same "
"status list the liveness notes use",
any("STALE_READING" in n and "inverter-a" in n
for n in _cs.liveness_notes())
and _cf.liveness_notes() == [])
check("...and the raw number is still recorded for diagnostics, so the "
"operator can see WHAT it was stuck on",
_cs.inverters[0].batt_fresh.last == A_FROZEN_W)
# ==============================================================================
print()
print("3. THE CONSUMERS REFUSE - and the counterfactual runs beside them")
# Both consumers already do the right thing with a None. Neither could ever
# receive one before this card, which is the entire exposure: the gate is the
# thing that was missing BETWEEN the shipped read and the shipped check.
_cfg3 = SignWatchdogConfig(min_w=250.0, hold_s=85.0)
check("PRECONDITION on the shipped predicate: a None measurement does not "
"disagree with anything, so it CLEARS the sign window",
disagrees(1300.0, None, True, _cfg3) is False
and disagrees(1300.0, A_FROZEN_W, True, _cfg3) is True)
check("PRECONDITION on the shipped sensor: a roster with a missing term is "
"None, and None clears the circulation window rather than accumulating",
circulating_w([1300.0, None]) is None
and circulating_w([A_FROZEN_W, B_LIVE_W]) == 2 * min(
abs(A_FROZEN_W), abs(B_LIVE_W)))
async def _cycles(gated, ticks=25, step_mono=12.0, wall=0.20):
"""Real cycles at N=2 with A's register frozen and the LOOP driving.
⚠️ THE COMMAND TRAVEL IS PRODUCED BY THE SHIPPED CONTROL LAW, NOT BY THE
FIXTURE. The meter is pinned at +3000 W of import, so `compute()` ramps the
site into discharge at its own slew rate through the real allocator and the
real write path; nothing here assigns `last_write`. A fixture that set the
standing command by hand would be feeding the gate its own arming condition,
which is the shape of check this repository keeps finding.
`gated=False` is THE COUNTERFACTUAL - the shipped behaviour before this
card, rebuilt by handing the consumers the raw held number the gate
withheld. Same cycles, same stimulus, so the difference between the two
columns is the gate and nothing else.
⚠️ TWO CLOCKS, DELIBERATELY, AND THE WALL-CLOCK ONE IS THE FLAKE RISK.
The freshness verdict runs on the INJECTED monotonic clock (12 s a tick, so
the 35 s bound is crossed on tick 3, deterministically, however loaded the
machine is). The sign watchdog runs on the WALL clock inside `cycle()`,
which cannot be injected - so `unit_sign_disagree_s` is set to 2 s and the
ticks are 0.20 s apart, putting the gate's close at ~0.6 s against a 2 s
window. That ORDER is the field's - 35 s < 75 s < 85 s by construction, part
6 - and the gate is given no head start beyond it.
⚠️ AND THE MARGIN IS MEASURED AND ASSERTED RATHER THAN ASSUMED (CI-06).
`test_peakguard.py` answered two of its own checks with the wall clock, flaked
at ~20 %, and cost a wrong merge diagnosis - a flaky check does not waste a
re-run, it manufactures evidence for whichever hypothesis the reader already
holds. So this returns the wall clock at which the gate closed, and the check
below fails with THAT NUMBER if the box was too slow, rather than reporting a
trip that looks like a code defect.
"""
c, hass = site({"unit_sign_disagree_w": 250, "unit_sign_disagree_s": 2,
"sign_disagree_w": 250, "sign_disagree_s": 2,
"site_sign_disagree_w": 250, "site_sign_disagree_s": 2,
"auto_start": True},
{"sensor.grid": 3000.0})
c.auto = True
circ, tripped_at, standing = [], None, []
t_start, t_gate = time.monotonic(), None
for k in range(ticks):
# A's register is FROZEN on the charge figure it stopped at; B's is
# live and is a DIFFERENT number. Neither is touched by the loop.
hass.numbers["sensor.batt_a"] = A_FROZEN_W
hass.numbers["sensor.batt_b"] = B_LIVE_W
await c.read_inputs(now_mono=k * step_mono)
if not gated:
for inv in c.inverters:
inv.batt = inv.batt_fresh.last
if t_gate is None and c.inverters[0].batt_fresh.stale:
t_gate = time.monotonic() - t_start
circ.append(c.site_circulating_w)
standing.append(c.inverters[0].standing_w())
await c.cycle()
if c.inverters[0].sign_wd.tripped and tripped_at is None:
tripped_at = k
await asyncio.sleep(wall)
return c, circ, tripped_at, standing, t_gate
_cg, _circg, _tripg, _standg, _tgate = run(_cycles(True))
_cu, _circu, _tripu, _standu, _ = run(_cycles(False))
_WD_S = 2.0 # `unit_sign_disagree_s` as this fixture sets it, above
print(f" standing on A (gated): {[round(v) for v in _standg]}")
print(f" gated : sign trip tick={_tripg} circ={_circg[:6]}...")
print(f" gate closed after {_tgate:.2f} s of WALL clock, against the "
f"{_WD_S:.0f} s watchdog window ({_WD_S / _tgate:.1f}x margin)")
print(f" counterfact: sign trip tick={_tripu} circ={_circu[:6]}...")
check("⚠️ PRECONDITION ON THE WALL CLOCK, AND IT IS A CHECK RATHER THAN AN "
"ASSUMPTION (CI-06). The gate closes on the INJECTED clock, the watchdog "
"on the real one, so this fixture only means something while the gate "
f"closes well inside the window: it closed at {_tgate:.2f} s against "
f"{_WD_S:.0f} s. If this fails the machine was slow, not the code - and "
"saying so here is what stops that being read as a defect",
_tgate is not None and _tgate < 0.5 * _WD_S)
check("PRECONDITION: the SHIPPED control law really did travel the command - "
"the gate's arming term is produced by `compute()` and the allocator, "
"not written into the fixture",
max(abs(v) for v in _standu) > MOVE_W
and abs(_standu[3] - _standu[0]) > MOVE_W)
check("⚠️ COUNTERFACTUAL FIRST - the stimulus is REAL. Without the gate, the "
"frozen register false-trips SAFETY-30's per-unit sign check and parks a "
"healthy inverter. That is the fault this card was filed for",
_tripu is not None and _cu.inverters[0].sign_wd.tripped
and "SIGN_DISAGREEMENT" in _cu.inverters[0].sign_wd.reason)
check("⚠️ AND WITH THE GATE IT DOES NOT TRIP. The reading is withheld before "
"the watchdog's window can accumulate, so the check refuses to conclude "
"rather than concluding wrongly - which is what a stale reading deserves",
_tripg is None and not _cg.inverters[0].sign_wd.tripped)
check("...and inverter B was not parked either way - the verdict is per unit",
not _cg.inverters[1].sign_wd.tripped
and not _cu.inverters[1].sign_wd.tripped)
check("⚠️ COUNTERFACTUAL: without the gate the circulating-power sensor "
"reports a loop that is not happening, computed from one frozen term and "
"one live one",
all(v == 2 * min(abs(A_FROZEN_W), abs(B_LIVE_W)) for v in _circu)
and _circu[0] == 1800.0)
check("⚠️ AND WITH THE GATE IT REPORTS `None` - an unknown, which CLEARS the "
"circulation window instead of accumulating it. Both consumers gated, "
"from one place, by one mechanism",
_circg[0] == 1800.0 and _circg[-1] is None
and all(v is None for v in _circg[_circg.index(None):]))
check("...and the circulation watchdog did not latch on the stale evidence",
not _cg.circ_wd.tripped and _cg.circ_wd.over_since is None)
# ==============================================================================
print()
print("4. THE LATCH, and the ~1 Hz oscillation it exists to prevent")
# Declaring the reading stale makes inputs_ok False, which commands 0 W, which
# drops the standing command below MOVE_W, which un-arms the conjunction. Without
# a latch the held number would come straight back to the consumers, and the
# whole thing would flip once per read for as long as the register stayed dead.
_lat = Freshness(BOUNDS["batt"])
for t in range(0, 40):
_lat.submit(A_FROZEN_W, A_FROZEN_W if t < 1 else 1300.0, float(t))
_was = _lat.stale
_after_zero = []
for t in range(40, 50):
_lat.submit(A_FROZEN_W, 0.0, float(t)) # the site has been stopped
_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, 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.
# It matters: a stale verdict stops the site, the site is then commanded 0 W, and
# a single unreachable-Home-Assistant read would drop the latch into a state
# where the conjunction is no longer armed - so the held -1200 W would come back
# to the consumers on the very next successful read.
_lat_fail = Freshness(BOUNDS["batt"])
for t in range(0, 40):
_lat_fail.submit(A_FROZEN_W, A_FROZEN_W if t < 1 else 1300.0, float(t))
_lat_fail.submit(None, 0.0, 40.0) # Home Assistant did not answer
_after_fail = _lat_fail.reading()
_lat_fail.submit(A_FROZEN_W, 0.0, 41.0) # it answers again, same number,
# site now resting at 0 W
check("⚠️ A FAILED READ DOES NOT CLEAR THE LATCH - found by a SURVIVING "
"mutation, not by design. Our inability to fetch a value is not evidence "
"that the value moved, and a stale verdict that a single unreachable read "
"can drop is one a stopped site would drop every time",
_after_fail is None and _lat_fail.stale
and _lat_fail.reading() is None)
# ⚠️ 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
# command that has TRAVELLED 250 W has also ENDED UP above 250 W. The case that
# separates them is a site being brought to REST.
_rest = Freshness(BOUNDS["batt"])
# The register reaches +5 W on the same read at which the old +1300 W command is
# still standing - the write path is serialised, so the register can catch up
# 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_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_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 "
"movement'",
not _rest.stale and _rest.reading() == 5.0)
check("...and the hole that term leaves CANNOT MISLEAD EITHER CONSUMER, which "
"is why it is the right trade: a unit commanded under 250 W is one "
"`watchdog.disagrees()` is False for whatever the measurement says, and a "
"frozen term beside a live unit that is itself resting nets to zero "
"circulating power rather than to a false alarm",
disagrees(0.0, A_FROZEN_W, True, _cfg3) is False
and circulating_w([A_FROZEN_W, 0.0]) == 0.0)
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",
_mov.held == A_FROZEN_W + 1.0 and not _mov.stale)
# ==============================================================================
print()
print("5. N=1: nothing changes on a healthy install, and what DOES is named")
_ONE = dict(BASE, inverters=[], setpoint_entity="number.sp_a",
soc_entity="sensor.soc_a", batt_entity="sensor.batt_a",
bms_soc_floor=A_FLOOR)
def one_site():
hass = Hass({"sensor.grid": 0.0, "sensor.soc_a": 50.0,
"sensor.batt_a": A_FROZEN_W},
{"number.sp_a": (-A_RATING, A_RATING, 1.0)})
c = Controller(dict(_ONE), hass, MemStore(), Pub(), stop=asyncio.Event())
c.inverters[0].dev_min, c.inverters[0].dev_max = -A_RATING, A_RATING
return c, hass
async def _one_healthy():
c, hass = one_site()
seen = []
for t in range(0, 60):
# A healthy register under a moving command: it dithers, so it is never
# the same integer watt twice running.
hass.numbers["sensor.batt_a"] = A_FROZEN_W + (t % 3)
c.inverters[0].last_write = 1300.0 if t % 2 else -1200.0
await c.read_inputs(now_mono=float(t))
seen.append(c.inverters[0].batt)
return c, seen
_c1, _seen1 = run(_one_healthy())
check("⚠️ AC: AT N=1 NOTHING CHANGES ON A HEALTHY INSTALL, and it is asserted "
"rather than assumed. Sixty reads of a dithering register under a command "
"reversing every second: every value reaches the consumers unmodified, "
"the gate never arms, and the status page says nothing new",
_seen1 == [A_FROZEN_W + (t % 3) for t in range(60)]
and not _c1.inverters[0].batt_fresh.stale
and _c1.liveness_notes() == [] and _c1.inputs_ok is True)
check("...and the shipped roster really is one unit, so this is the install "
"that is running in a house today and not a fixture shaped like it",
len(_c1.inverters) == 1 and _c1.inverters[0].name == "inverter-a")
async def _one_frozen():
c, hass = one_site()
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))
return c
_c1f = run(_one_frozen())
check("⚠️ AND WHAT DOES CHANGE AT N=1 IS NAMED RATHER THAN HIDDEN: a FROZEN "
"register now stops the site, where before it read as a healthy "
"measurement. That is a behaviour change on the shipped roster and it is "
"the point of the card - HW-01 was a one-inverter site, and the pack "
"drained to its floor with every detector reading healthy for 6 h 25 m",
_c1f.inverters[0].batt is None and _c1f.inputs_ok is False)
check("...and the direction it fails in is the safe one: a false verdict "
"commands 0 W and keeps commanding (§11.1 rule 1), behind the existing "
"`stale_input_s` hold, and clears itself on the register's next move",
float(_c1f.o.get("stale_input_s")) > 0
and _c1f.inverters[0].batt_fresh.stale)
# ==============================================================================
print()
print("6. THE BOUNDS, graded against the SHIPPED config.yaml at BOTH ends")
def option_default(key):
"""`key`'s default out of config.yaml's `options:` block, at two spaces."""
body = io.open(HERE / "config.yaml", encoding="utf-8").read()
body = body[body.index("\noptions:"):body.index("\nschema:")]
for line in body.splitlines():
if line.startswith(f" {key}:") and not line[2:3].isspace():
return float(line.split(":", 1)[1].strip())
return None
_circ_hold = option_default("circulation_hold_s")
_unit_hold = option_default("unit_sign_disagree_s")
_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})")
check("PRECONDITION: both consumer windows were actually found in config.yaml, "
"so a rename cannot turn this part into a check of two Nones",
_circ_hold == 75.0 and _unit_hold == 85.0 and _unit_min_w == 250.0)
check("⚠️ THE BOUND IS AT LEAST 2x THE PHYSICAL FLOOR. Below 17.0 s a healthy "
"register has not yet been OBLIGED to answer - 6.0 s of plant "
"(NOTES §5.9), one 10 s ESPHome publish (firmware/goodwe-master.yaml:181) "
"and one 1 s add-on read - so a shorter bound trips on a working inverter",
BATT_UNCHANGED_S >= 2 * BATT_OBLIGED_S and BATT_OBLIGED_S == 17.0)
check("⚠️ AND AT MOST HALF THE TIGHTER CONSUMER WINDOW. A gate slower than its "
"consumer hands over the frozen number and explains afterwards. THIS IS "
"THE CHECK THAT BITES IF AN OPERATOR'S DEFAULT MOVES: drop "
"`circulation_hold_s` to 60 in config.yaml and this goes red",
BATT_UNCHANGED_S <= 0.5 * _ceiling)
# ⚠️ THE FIRST DRAFT OF THIS CHECK CLAIMED THE BAND CONTAINED EXACTLY ONE
# INTEGER AND IT DOES NOT - it contains 34, 35, 36 and 37, and the check went
# red on its first run and is written the honest way round here. The band is
# what is derived; the value inside it is a choice, and the only defensible
# claims about the choice are that it IS inside and that it is not a number
# borrowed from somewhere else.
check("...and the admissible band computed from config.yaml is [34.0, 37.5], "
"which contains four integers. 35 is a CHOICE inside a DERIVED band, not "
"a derived number, and it is deliberately not 30 - which is "
"FIRMWARE_INGEST_VETO_S and would read as a borrowed constant - or 60, "
"which is round",
[n for n in range(0, 200)
if 2 * BATT_OBLIGED_S <= n <= 0.5 * _ceiling] == [34, 35, 36, 37]
and BATT_UNCHANGED_S not in (30.0, 60.0))
check("⚠️ AND `MOVE_W` IS PINNED TO config.yaml TOO. It is the largest value "
"that still arms the gate everywhere the per-unit sign check can speak; "
"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",
"update_interval: 10s" in io.open(
HERE.parent / "firmware" / "goodwe-master.yaml",
encoding="utf-8").read()
and F.PUBLISH_PERIOD_S == 10.0 and F.READ_PERIOD_S == 1.0
and F.PLANT_SETTLE_S == 6.0)
# ==============================================================================
print()
print("7. THE POLICY TABLE tells the truth about what is WIRED")
def bounds_users(key):
"""Modules that SUBSCRIPT `BOUNDS[key]` in CODE - parsed, not grepped.
⚠️ THE FIRST DRAFT GREPPED FOR THE STRING AND ANSWERED WRONG, because both
`inverter.py` and `main.py` NAME `BOUNDS["soc"]` in a comment explaining why
it is not wired. A text search cannot tell a mention from a use, and this
part's whole job is to tell them apart - so it walks the AST instead.
"""
out = []
for path in sorted((HERE / "app").glob("*.py")):
if path.name == "freshness.py":
continue
tree = ast.parse(io.open(path, encoding="utf-8").read())
for node in ast.walk(tree):
if (isinstance(node, ast.Subscript)
and isinstance(node.slice, ast.Constant)
and node.slice.value == key
and ((isinstance(node.value, ast.Name)
and node.value.id == "BOUNDS")
or (isinstance(node.value, ast.Attribute)
and node.value.attr == "BOUNDS"))):
out.append(path.name)
break
return out
_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} 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'}")
check("⚠️ `wired=True` MEANS SOMETHING CONSUMES IT. A policy table nobody "
"checks is a comment; this is the row `batt` claims and the module that "
"claims it",
BOUNDS["batt"].wired is True and _uses_batt == ["inverter.py"])
check("⚠️ AND `wired=False` MEANS NOTHING DOES. SoC is TEL-08's and it is "
"STATED here rather than wired, because its obligation term is energy "
"through the pack and the constant that needs has never been measured on "
"this site. A guessed constant would fire on a pack resting at its floor",
BOUNDS["soc"].wired is False and _uses_soc == [])
check("...and the SoC row says what TEL-08 owes, in the row itself, so the "
"next reader does not have to reconstruct why it is unwired",
"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 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))
# ==============================================================================
print()
print("8. THE MEASUREMENT the bound rests on, replayed from the committed data")
# ⚠️ THIS IS THE PART THAT STOPS THE DERIVATION BECOMING A STORY. The docstring
# in freshness.py quotes three numbers off this capture; they are recomputed
# here from the committed files, so a figure that drifts from its evidence goes
# red instead of being re-quoted forever.
#
# ⚠️ AND IT IS THE METER'S REGISTER, NOT 0x0518. No 30-day capture of the
# inverter's own power exists. What this establishes is the only thing it is
# used for: that the legitimate unchanged duration of a 1 W power entity in this
# house is MINUTES, so an unchanged threshold alone has no admissible value.
_cap = sorted(glob.glob(str(HERE / "scenarios" / "p1-net-30d-2026-08"
/ "p1-net-*.jsonl.gz")))
_runs, _prev = [], None
_t0 = time.monotonic()
for _f in _cap:
with gzip.open(_f, "rt") as _fh:
for _line in _fh:
_r = json.loads(_line)
_t = datetime.fromisoformat(_r["t"]).timestamp()
if _prev is not None:
_runs.append((_t - _prev, _pw))
_prev, _pw = _t, float(_r["w"])
_worst = max(_runs)[0] if _runs else 0.0
_over_bound = sum(1 for d, _ in _runs if d >= BATT_UNCHANGED_S)
print(f" {len(_cap)} days, {len(_runs) + 1} samples, "
f"{time.monotonic() - _t0:.1f}s to replay")
print(f" longest unchanged run {_worst:.1f} s; "
f"{_over_bound} runs >= {BATT_UNCHANGED_S:.0f} s")
check("PRECONDITION: the committed capture is all thirty days, so this is not "
"a claim about whatever files happened to be present",
len(_cap) == 30 and len(_runs) + 1 == 485841)
check("⚠️ A REAL 1 W POWER ENTITY HELD ONE VALUE FOR 263.2 s WITH NOTHING "
"WRONG. That single measurement is why the bound is a conjunction and "
"not a threshold: no unchanged threshold under 263 s is free of false "
"positives, and 263 s is 3.5x too slow to protect a 75 s consumer",
abs(_worst - 263.2) < 0.1 and _worst > _ceiling)
check("...and the two figures freshness.py's docstring quotes off this capture "
"are the ones it produces today",
_over_bound == 10
and sum(1 for d, _ in _runs if d >= BATT_OBLIGED_S) == 386)
check("⚠️ AND THE LONG RUNS ARE ALL UNDER AN UNMOVING LOAD, which is what the "
"conjunction keys on. The worst one is 263 s of a steady -899 W export - "
"a house doing one thing - so a gate that also requires the COMMAND to "
"have travelled 250 W is not armed by any of them",
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")
# ⚠️ THE REPO'S RECURRING DEFECT IS A CHECK WHOSE DATA CANNOT REACH THE FAILURE
# IT NAMES. The honest end of that habit is to build the mutations this file
# would NOT see and record them. Each one below is a real hole in the shipped
# gate, demonstrated rather than described.
# S1. A register pinned by the INVERTER's own limit, not by a fault. If the
# device is derating and cannot follow a command that ASKS FOR MORE, the
# register can legitimately hold while the command travels.
_s1 = Freshness(BOUNDS["batt"])
for t in range(0, 40):
_s1.submit(2503.0, 2650.0 if t < 1 else 4600.0, float(t))
check("⚠️ SURVIVOR 1, FALSE POSITIVE: a saturated or derating inverter pinned "
"at 2503 W while the command climbs to 4600 W is declared stale. The "
"register is telling the truth and the gate calls it dead",
_s1.stale)
check("...and its SIZE is bounded and its direction is the safe one: the "
"verdict is a site stop, behind the `stale_input_s` hold, and the moment "
"the command comes back down the register moves and the latch clears. "
"Cost = one spurious stop; it cannot drive the battery anywhere",
(lambda f: (f.submit(2400.0, 0.0, 60.0), not f.stale)[-1])(_s1))
# S2. A unit whose command never travels 250 W. HW-01's own shape, at a
# smaller scale, and the gate is deliberately not armed for it.
_s2 = Freshness(BOUNDS["batt"])
for t in range(0, 400):
_s2.submit(A_FROZEN_W, -1200.0 if t < 1 else -1400.0, float(t))
check("⚠️ SURVIVOR 2, MISS: a register frozen under a command that only ever "
"moves 200 W is NEVER detected, for 400 s or forever. This is scoped, "
"not overlooked - such a unit is one no per-unit sign check can speak "
"about (`unit_sign_disagree_w` is 250 W) and one contributing at most "
"250 W to a loop whose reporting floor is 700 W",
not _s2.stale and _s2.reading() == A_FROZEN_W)
# S3. A register that dithers on dead data. A publisher that keeps republishing
# a stale WINDOW of values rather than one value defeats this entirely.
_s3 = Freshness(BOUNDS["batt"])
for t in range(0, 400):
_s3.submit(A_FROZEN_W + (t % 2), 1300.0, float(t))
check("⚠️ SURVIVOR 3, MISS, AND IT IS THE LARGEST ONE: a dead publisher whose "
"last value happens to ALTERNATE - or a cache replaying two old samples - "
"never triggers the unchanged clock at all. This gate detects a value "
"that STOPPED; it cannot detect a value that is WRONG. Nothing in this "
"add-on can, and TEL-06's arrival stamp inside the firmware is the only "
"layer that could",
not _s3.stale and _s3.reading() is not None)
# S4. The gate is per unit, so a whole-site publisher failure that freezes both
# registers is caught twice rather than missed - recorded because a reader
# might expect the opposite.
_s4a, _s4b = Freshness(BOUNDS["batt"]), Freshness(BOUNDS["batt"])
for t in range(0, 40):
_s4a.submit(A_FROZEN_W, A_FROZEN_W if t < 1 else 1300.0, float(t))
_s4b.submit(B_LIVE_W, B_LIVE_W if t < 1 else -1300.0, float(t))
check("NOT A SURVIVOR, recorded because it looks like one: a publisher failure "
"that freezes BOTH registers is caught on both units independently, on "
"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 IS SAFETY-56'S AND HAS NOW BEEN REWRITTEN, "
"WHICH IS WHAT THIS CHECK WAS FOR. It used to require 'battery-power "
"register is not following its own command' to still be present, so "
"that removing it had to be a deliberate act rather than a side effect "
"of editing the escape clause beside it. SAFETY-56 is that act, and the "
"requirement is now the inverse: the accusation is GONE from this line "
"and part 11 grades what replaced it. The escape clause above is "
"SAFETY-57's and is untouched by it",
"battery-power register is not following its own command"
not in _stale57[0]
and "did not move under a command that did" 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()
print("11. SAFETY-56: the sentence this gate can honestly say")
# ⚠️ THE SUBJECT OF THIS PART IS A CLAIM, NOT A STRING, AND NOTHING IN IT
# CHANGES A DECISION. SAFETY-37's conjunction is correct and is untouched;
# what was wrong is what two sentences CONCLUDED from it. The conjunction is
# satisfied by two disjoint worlds -
#
# DEAD the register has stopped reporting reality. The pack really is
# doing what it was told; the published number does not say so.
# REFUSED the register is honest and the SOURCE cannot comply. The BMS is
# refusing at the pack floor, so the pack does nothing whatever the
# command asks, and an unchanged reading is the TRUTH.
#
# - and nothing this module is handed separates them. Three sentences named
# DEAD as fact: `freshness._grade()`'s "the reading is not following",
# `main.py`'s log line "battery-power register is not following its own
# command", and the NOT READY row's "<entity> is missing or not numeric". On
# the live 2026-09-21 episode all three were the wrong one of the two, and the
# third was the worst because it asserts a false fact about a NAMED entity: the
# operator opens Developer Tools, finds it working, and concludes the
# diagnostics are broken.
#
# So the property graded below is not "the new words are present". It is that
# the two worlds are REAL AND DIFFERENT (11a on the committed capture, 11b in a
# driven pair), that the gate CANNOT separate them (11b), and therefore that a
# clause asserting either one is false half the time - which is what makes 11c
# and 11d a claim rather than decoration. Habit 4: without 11a and 11b those
# two would be substring tests over a sentence this file also wrote.
# --- 11a. the old diagnosis was FALSE on the site it was printed on ----------
# ⚠️ OFF THE COMMITTED CAPTURE, NOT OFF A FIXTURE. Part 10a has already graded
# this file's bytes against the manifest and its decoder's sign convention
# against the frames, so the three series below are evidence rather than a
# story. Every boundary is derived in part 10b from the register's own longest
# hold; nothing here is typed.
_S56_CMDS_HELD = sorted({_D.held_at(_sp, t)
for t in range(int(_HOLD_FROM), int(_PARK))})
_S56_SOC_HELD = sorted({_D.held_at(_soc, t)
for t in range(int(_HOLD_FROM), int(_HUMAN))})
_S56_CHARGE_T, _S56_CHARGE_W = next((t, v) for t, v in _sp
if t > _PARK and v < 0.0)
_S56_AFTER = _D.held_at(_bt, _REG_MOVES + 5.0)
print(f" during the 850 s hold: commands {_S56_CMDS_HELD} W, "
f"SoC {_S56_SOC_HELD} %, register {_HOLD_W:+.0f} W")
print(f" first CHARGE command {hms(_S56_CHARGE_T)} at "
f"{_S56_CHARGE_W:+.0f} W -> register moves {hms(_REG_MOVES)} "
f"({_REG_MOVES - _S56_CHARGE_T:.0f} s) to {_S56_AFTER:+.0f} W")
check("11a. ⚠️ THE OLD DIAGNOSIS WAS FALSE ON THE ONE SITE IT WAS EVER "
"PRINTED ON. Through the whole hold the site asked for DISCHARGE only "
"(+260 to +1060 W) out of a pack reading 11 %, and the register answered "
"-3 W. Then the first CHARGE command arrives and it moves within ONE "
"17.0 s obligation window. A register that answers a reversed command in "
"one window had not stopped reporting; what would not move was the PACK",
all(v > 0.0 for v in _S56_CMDS_HELD)
and min(_S56_CMDS_HELD) == 260.0 and max(_S56_CMDS_HELD) == 1060.0
and _S56_SOC_HELD == [11.0] and _HOLD_W == -3.0
and _S56_CHARGE_W < 0.0
and 0.0 < _REG_MOVES - _S56_CHARGE_T <= F.BATT_OBLIGED_S
and _S56_AFTER != _HOLD_W)
check("...and the gate was RIGHT to fire anyway, which is why this card touches "
"no term of the conjunction. The three limits are still the derived ones "
"and the same capture still arms at 10:52:00 with the verdict standing: "
"a register genuinely stuck against a real command is exactly what "
"SAFETY-37 should catch, and a pack that cannot comply is a state the "
"site must stop in too. What was wrong was never the firing",
BOUNDS["batt"].unchanged_s == BATT_UNCHANGED_S
and BOUNDS["batt"].move_w == MOVE_W
and BOUNDS["batt"].release_s == F.BATT_OBLIGED_S
and hms(_TRIP) == "10:52:00" and _f_after.releases == 1
and next(r for r in _rows_after if r[0] == _TRIP)[1] is True)
# --- 11b. two worlds, one published stream, one verdict ----------------------
# ⚠️ GROUND TRUTH IS WRITTEN DOWN, NOT DERIVED FROM THE PUBLISHED STREAM
# (habit 2). A check that computes "what the pack was really doing" out of the
# number the register published is comparing a thing with itself, which is the
# defect MUTATION.md has recorded fifteen times.
_BATT_ENT = "sensor.batt_a"
_S56_N = 40
_S56_CMD = [800.0 if t < 1 else 1600.0 for t in range(_S56_N)]
_S56_PUB = [-3.0] * _S56_N # what the ENTITY publishes, both
_S56_TRUE_DEAD = list(_S56_CMD) # the pack obeyed; the register lied
_S56_TRUE_REFUSED = [-3.0] * _S56_N # the pack refused; the register told
async def _s56_world(soc_pct):
"""The shipped read path at N=1. The ground-truth stream above is never
handed to the controller - it is what the WORLD is, not an input."""
c, hass = one_site()
hass.numbers["sensor.soc_a"] = soc_pct
for t in range(_S56_N):
hass.numbers[_BATT_ENT] = _S56_PUB[t]
c.inverters[0].last_write = _S56_CMD[t]
await c.read_inputs(now_mono=float(t))
return c, hass
_S56_LOG0 = len(_LOGREC)
# ⚠️ THE TWO WORLDS ARE DRIVEN AT DIFFERENT SoC ON PURPOSE - 50 % against the
# 11 % the live pack sat at - so that "the sentence did not move" below is a
# measurement of this module's `soc` bound being STATED, NOT WIRED, and not an
# artefact of both runs being handed the same percentage. Wiring it is TEL-08's.
_c56_dead, _h56_dead = run(_s56_world(50.0))
_c56_ref, _h56_ref = run(_s56_world(11.0))
_pub_dead = [v for e, v in _h56_dead.served if e == _BATT_ENT]
_pub_ref = [v for e, v in _h56_ref.served if e == _BATT_ENT]
_S56_DIFFER = sum(1 for a, b in zip(_S56_TRUE_DEAD, _S56_TRUE_REFUSED) if a != b)
print(f" DEAD truth {_S56_TRUE_DEAD[:2]}... published {_pub_dead[:2]}...")
print(f" REFUSED truth {_S56_TRUE_REFUSED[:2]}... published {_pub_ref[:2]}...")
check(f"11b. PRECONDITION: the two worlds are REALLY TWO - their ground truth "
f"differs on {_S56_DIFFER} of {_S56_N} reads - and the PUBLISHED stream "
f"is bit-identical in both, so nothing downstream of {_BATT_ENT} can "
f"tell them apart. Without this the rest of part 11 is one world twice",
_S56_DIFFER == _S56_N and _pub_dead == _pub_ref == _S56_PUB
and len(_pub_dead) == _S56_N)
check("...and in REFUSED the register is RIGHT on every one of the 40 reads "
"(published == truth) while in DEAD it is wrong on all 40. That is the "
"fact the old sentence got backwards: it named the register in a world "
"where the register is the only thing telling the truth",
_S56_PUB == _S56_TRUE_REFUSED
and not any(p == t for p, t in zip(_S56_PUB, _S56_TRUE_DEAD)))
_r56_dead = _c56_dead.inverters[0].batt_fresh.reason
_r56_ref = _c56_ref.inverters[0].batt_fresh.reason
check("⚠️ AND THE GATE FIRES IDENTICALLY IN BOTH, WHICH IS THE WHOLE REASON "
"ITS SENTENCE MAY NOT NAME ONE OF THEM: same verdict, same MISSING "
"reading, same site stop - and the sentence is BYTE-IDENTICAL across two "
"worlds that share no ground truth at all",
_c56_dead.inverters[0].batt_fresh.stale is True
and _c56_ref.inverters[0].batt_fresh.stale is True
and _c56_dead.inverters[0].batt is None
and _c56_ref.inverters[0].batt is None
and _c56_dead.inputs_ok is False and _c56_ref.inputs_ok is False
and _r56_dead == _r56_ref and _r56_dead != "")
check("...and the two runs were driven at 50 % and 11 % SoC and the sentence "
"did not move one byte, because `BOUNDS['soc']` is STATED, NOT WIRED. A "
"message honest about its own ambiguity needs no new input, and adding "
"one here would make this sentence depend on a second measurement being "
"right - TEL-08's card, not this one",
_c56_dead.inverters[0].soc == 50.0 and _c56_ref.inverters[0].soc == 11.0
and _r56_dead == _r56_ref)
# --- 11c. both explanations, and no accusation, on every surface -------------
# The two explanations, each tagged with the world it is true OF. The emitted
# sentence is the same in both worlds (11b), so a sentence naming only one is
# false half the time - which is why BOTH are required, as an ORDERED PAIR
# (habit 14): two clauses in one sentence can always be transposed at zero
# bytes, and "the register may be unable to comply / the source may have
# stopped reporting reality" is a different and wrong claim.
_S56_EXPLANATIONS = [
("DEAD", "the register may have stopped reporting reality"),
("REFUSED", "the source may be unable to comply with the command"),
]
# ⚠️ WHAT THE THREE SENTENCES USED TO ASSERT. Each is a claim about the
# REGISTER or about the ENTITY, and 11a measured the register to be alive on
# the live episode while 11d measures the entity to be serving a float. They
# are listed with the file they came from so a reviewer can see the set is the
# whole set rather than the ones that were easy to find.
_S56_ACCUSATIONS = [
("app/freshness.py _grade()", "the reading is not following"),
("app/main.py _read_unit()", "register is not following its own command"),
("app/main.py checks()", f"{_BATT_ENT} is missing or not numeric"),
]
_log56 = [r.getMessage() for r in _LOGREC[_S56_LOG0:] if "SAFETY-37" in r.getMessage()]
def _s56_row(c):
"""The battery-power readiness row, or a NAMED absence - never a raise.
⚠️ IT MUST NOT RAISE, AND THAT IS FCST-06'S RULE RATHER THAN TIDINESS. The
first draft of this part was `[... ][0]`, and the mutation that transposes
`label` and `entity` in the row - zero bytes - makes the row unfindable. The
run then died on an IndexError with every check after this line UNGRADED:
a red with no name attached, which reads like broken infrastructure and gets
retried rather than read. Measured: 0 named reds before, 2 after.
"""
hits = [r for r in c.checks() if r["text"].startswith("battery power")]
if len(hits) != 1:
return {"ok": None, "warn": None,
"text": f"<NO SINGLE `battery power` ROW: {len(hits)} found>"}
return hits[0]
_row56_ref = _s56_row(_c56_ref)
_S56_SURFACES = {
"reason (status page + action log)": _r56_ref,
"log line (SAFETY-37 ERROR)": _log56[-1] if _log56 else "",
"stale_note() (NOT READY row)": _c56_ref.inverters[0].stale_note(),
"checks() battery-power row": _row56_ref["text"],
}
for _n, _s in _S56_SURFACES.items():
print(f" {_n}: {_s[:96]}...")
check("PRECONDITION: the readiness row is FINDABLE BY ITS LABEL in both worlds "
"- exactly one `battery power` row each - so every row graded below is "
"that row, and a mutation that moves `label` out of the first slot is a "
"NAMED red here rather than an IndexError three checks later (FCST-06). "
f"REFUSED: {_row56_ref['text'][:44]!r}",
_row56_ref["ok"] is not None)
check("PRECONDITION: all four operator-facing surfaces were really produced by "
"this run and are non-empty, so the two quantifiers below are not over "
"an empty set - and the log line is the one the SHIPPED read path "
"emitted, not one this file formatted. Both worlds emitted one and the "
"two are the same string, which is 11b again on the log surface",
len(_log56) == 2 and _log56[0] == _log56[1]
and all(len(s) > 30 for s in _S56_SURFACES.values())
and _r56_ref in _log56[-1])
_idx56 = [_r56_ref.find(t) for _w, t in _S56_EXPLANATIONS]
check(f"11c. BOTH EXPLANATIONS ARE NAMED, AND IN THAT ORDER: "
f"{_S56_EXPLANATIONS[0][1]!r} at {_idx56[0]} then "
f"{_S56_EXPLANATIONS[1][1]!r} at {_idx56[1]}. Pinned as an ordered pair "
f"because transposing 'register' and 'source' between the two clauses is "
f"a zero-byte edit that leaves both words in the sentence (habit 14)",
all(i >= 0 for i in _idx56) and _idx56 == sorted(_idx56)
and len(set(_idx56)) == len(_idx56))
check("...and the sentence says in its own words that it CANNOT choose, rather "
"than leaving two explanations side by side for a reader to rank - an "
"operator who reads a list picks the first item",
"CANNOT TELL THEM APART" in _r56_ref)
# ⚠️ THE POSITIVE CONTROL FOR THE `not any(...)` BELOW, AND IT IS NOT OPTIONAL:
# a quantifier over an empty list is True, and so is one over three literals
# with a typo in each. These are the three sentences as 0.10.1 SHIPPED them,
# quoted from the release rather than rebuilt, and the same predicate is
# required to FIND an accusation in every one of them (habit 7: vary one thing
# from an ACCEPTED baseline).
_S56_SHIPPED_010_1 = {
"reason": "STALE_READING: -3 unchanged for 35 s while the standing command "
"moved 800 W to +1600 W - the reading is not following, so it is "
"reported as MISSING rather than as a measurement (limits: "
"unchanged over 35 s, command over 250 W and moved by over 250 W)",
"log": "SAFETY-37: inverter-a battery-power register is not following its "
"own command - STALE_READING: ... Treating it as MISSING, which "
"commands 0 W.",
"row": f"battery power: {_BATT_ENT} is missing or not numeric",
}
check(f"POSITIVE CONTROL: the very same predicate, run against the three "
f"sentences as 0.10.1 SHIPPED them, finds an accusation in EVERY one - "
f"and every one of the {len(_S56_ACCUSATIONS)} literals matches at least "
f"one of them, so neither an empty list nor a mistyped literal can make "
f"the check below pass by vacuity",
len(_S56_ACCUSATIONS) == 3 and len(_S56_EXPLANATIONS) == 2
and all(any(t in old for _w, t in _S56_ACCUSATIONS)
for old in _S56_SHIPPED_010_1.values())
and all(any(t in old for old in _S56_SHIPPED_010_1.values())
for _w, t in _S56_ACCUSATIONS))
check("⚠️ AND NOT ONE OF THE THREE ACCUSATIONS SURVIVES, ON ANY OF THE FOUR "
"SURFACES: " + "; ".join(f"{w} -> {t!r}" for w, t in _S56_ACCUSATIONS)
+ ". This is the check that goes red if any of the three sentences is "
"reverted, and it is stated over the SURFACES rather than over the "
"source files because two of the three are assembled from two modules",
not any(t in s for _w, t in _S56_ACCUSATIONS
for s in _S56_SURFACES.values()))
# --- 11d. the NOT READY row that named a working entity ----------------------
async def _s56_absent():
"""The other half of the row's ambiguity: the entity really is gone."""
c, hass = one_site()
for t in range(3):
hass.numbers[_BATT_ENT] = None # Home Assistant has no state
c.inverters[0].last_write = 0.0
await c.read_inputs(now_mono=float(t))
return c, hass
_c56_abs, _h56_abs = run(_s56_absent())
_last_ref = [v for e, v in _h56_ref.served if e == _BATT_ENT][-1]
_last_abs = [v for e, v in _h56_abs.served if e == _BATT_ENT][-1]
_row56_abs = _s56_row(_c56_abs)
print(f" WITHHELD: {_BATT_ENT} served {_last_ref!r}, stale="
f"{_c56_ref.inverters[0].batt_fresh.stale} ABSENT: served "
f"{_last_abs!r}, stale={_c56_abs.inverters[0].batt_fresh.stale}")
check("11d. PRECONDITION: the row's two causes are both REACHED. In REFUSED "
f"{_BATT_ENT} served a real float on the last read and the gate is "
f"holding the value back; in ABSENT Home Assistant served None and the "
f"gate is not armed at all. Two different causes, one `inv.batt is None`",
_last_ref == -3.0 and _last_abs is None
and _c56_ref.inverters[0].batt_fresh.stale is True
and _c56_abs.inverters[0].batt_fresh.stale is False
and _c56_ref.inverters[0].batt is None
and _c56_abs.inverters[0].batt is None)
check("⚠️ AND THE VERDICT IS UNCHANGED BY THIS CARD - both rows are still "
"ok=False, warn=False, i.e. still NOT READY. The defect was never the "
"pass/fail, it was the row asserting a fact about a named entity that "
"was at that moment present, numeric and updating",
_row56_ref["ok"] is False and _row56_ref["warn"] is False
and _row56_abs["ok"] is False and _row56_abs["warn"] is False
and _row56_ref["text"] == _row56_abs["text"])
check("...and what it says instead is the observation plus BOTH causes, with "
f"{_BATT_ENT} in its own slot - `label` and `entity` are two "
f"interpolations in one sentence and swapping them costs zero bytes, so "
f"the literal around each is pinned (habit 14)",
f"battery power: no usable value this cycle from {_BATT_ENT} - and this "
f"row cannot tell which of two things that is" in _row56_ref["text"]
and "the entity is missing or not numeric" in _row56_ref["text"]
and "a number that this controller is withholding" in _row56_ref["text"])
check("⚠️ AND THE DISCRIMINATOR IT POINTS AT EXISTS AND IS SELECTIVE, which is "
"why the row does not need to ask `batt_fresh.stale` itself: `checks()` "
"carries a STALE_READING row in REFUSED and NONE in ABSENT, so 'is that "
"row below?' answers the question this one declines to answer. A row "
"that pointed at a surface present in both states would be worse than "
"the accusation it replaced",
any("STALE_READING" in r["text"] for r in _c56_ref.checks())
and not any("STALE_READING" in r["text"] for r in _c56_abs.checks()))
check("...and the row's wording is NOT shared with the log line's, which is a "
"design choice rather than an omission: the two sentences have no clause "
"in common longer than the word 'entity', because one explains a "
"BEHAVIOUR to a log reader and the other tells a page reader whether to "
"trust a NAME. A shared helper would have to drop whichever half the "
"other caller does not need",
_S56_EXPLANATIONS[0][1] not in _row56_ref["text"]
and _S56_EXPLANATIONS[1][1] not in _row56_ref["text"]
and "no usable value this cycle" not in _r56_ref)
# --- 11e. what part 11 does NOT grade, with its shape (habit 13) -------------
# * It does not show the live episode was a BMS refusal beyond doubt. 11a
# measures that the register answered a reversed command inside one
# obligation window, which excludes a register that had simply stopped; it
# does not exclude one that was intermittently wrong. That is the point: the
# sentence no longer claims to know either.
# * It does not grade ENGLISH. A wording that named both explanations and then
# said something useless about them would pass 11c.
# * The `no entity configured` branch beside the rewritten row is untouched and
# ungraded here - it has no second explanation, because a blank entity really
# is the only thing it can be.
# * The grid-power and battery-SoC rows go through the same rewritten line and
# have no freshness gate today, so for them the second cause is a hypothesis
# about this controller rather than a live possibility. That is weaker than
# the battery-power case and is the price of not branching per row; a gate on
# either of those signals (TEL-08 would be one) makes it exact.
print()
if fails:
print(f"{len(fails)} FAILED: {', '.join(fails)}")
sys.exit(1)
print("all checks passed")
Reference in New Issue View Git Blame Copy Permalink