admin/goodwe-addon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
43b274035f88ee008ac815627193aafd7a92728c
goodwe-addon/goodwe_controller/test_clamp.py
T
glenn schrooyenandClaude Opus 5.5 770988c902
checks / unit (pull_request) Successful in 8m44s
Details
TEST-26: a missing value in a check is a named FAIL, not a traceback 
Checks and prints that indexed an empty list, exhausted next(), or formatted
None killed the rest of their file (FCST-06). Read them through the new
named_absence.py (nth/first/field/given/present) and conjoin a presence term.

- every site in the brief, plus the family the new scan and a sweep found
- test_named_absence.py: scans every test_*.py for unguarded filtered-list
  indexing, next() without a default and numeric formats on maybe-None
  names; proven to bite; prints what it cannot see, counted
- test_price.py 16a: the `_T17_CAPS is _T17_CAPS` tautology now grades the
  two assemble() calls' arguments
- test_control.py: coverage table re-measured on runs graded to the end

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BwRLw7eKYj2byVdwB6kbjU
2026-09-25 00:06:58 +02:00

1866 lines
98 KiB
Python
Raw Blame History

"""Runnable check for SAFETY-03, the output clamp and its winding-loop alarm.
python3 test_clamp.py
No framework, asserts only, same house style as test_freeze.py.
Seven parts, and parts 3 and 4 are the ones that carry SAFETY-03's own
acceptance criteria; part 6 is TEST-11's and part 7 is SAFETY-28's:
1. THE CLAMP ARITHMETIC, on the shipped `clamp_to_rating()` - boundaries,
symmetry, the device second stage, and the non-finite cases. Table driven.
2. THE ALARM'S CADENCE - rising edge, re-assert, hysteresis - on the shipped
`ClampWatchdog`. Every engagement counted, notifications not flooded.
3. AC 1 + AC 3, WIRED. A computed setpoint that exceeds the rating, driven
through the REAL `Controller.cycle()`: nothing over the rating reaches the
device, and the alarm fires. Includes the historical runaway from
specs.md 10.3 replayed against an unclamped transcription, so the check
says what the clamp is actually worth in watts rather than asserting a
constant.
4. AC 2, AND IT IS THE LOAD-BEARING HALF. A loop that SATURATES WITHOUT
CLAMPING - the inverter refusing to follow a perfectly legal command - must
raise NO clamp alarm. If it does, the bug signal is inside the noise again
and the whole ticket has bought nothing.
5. NO PATH BYPASSES IT. write_setpoint() is the only route to the wire, so it
is called directly with an out-of-range value the way the heartbeat and the
operator's buttons call it.
6. TEST-11. THE GUARD READS THE WRITE THAT JUST HAPPENED. `self.clamp_last`
is assigned by `write_setpoint()` and by nothing else, so a
`check_output_clamp()` moved above the write keeps its loop-side evidence
and silently loses its write-side half. Graded as a CONSEQUENCE - what the
watchdog could see when it was consulted - so that a rename or a helper
does not trip it, and with the write-path half exercised on a stimulus of
its own, because parts 3-5 never produce one through cycle().
7. SAFETY-28. THE WRITE PATH IS GATED ON THE ALLOCATION RECORD - a value the
single allocator did not place on THIS tick does not reach the wire, a
record from a previous tick cannot authorise one, and 0 W is authorised
unconditionally so the failsafe can never deadlock behind the gate. This is
the check that replaced a config-shape proxy: `ACTUATOR_WRITE_KEYS` graded
how many option KEYS exist, which stays 1 while a per-inverter config adds
a second real writer.
------------------------------------------------------------------------------
WHY PART 4 ALSO ASSERTS ON `Decision`'s BOOLEANS AND NOT ON `reason`
`Decision.reason` is one string and the stages after the clamp overwrite it. The
check below drives a cycle that clamps the integrator AND is then slew limited,
and asserts that `reason` does NOT say so while `i_clamped` does. That is not a
style preference - an alarm keyed on the string is silent on exactly the cycles a
winding loop produces, and this is the check that would go red if someone
"simplified" main.py back onto it.
------------------------------------------------------------------------------
WHAT THIS CHECK DOES NOT COVER
- The firmware-side magnitude clamp (the card's "two clamps, not one"). This
add-on's `write_setpoint()` is the add-on half; the ESP32 half is DRIVER-07's
and is not in this repo's Python.
- A rig run. Nothing here needs one: the clamp is arithmetic on the write path
and part 3 drives the shipped `Controller.cycle()` against a stub Home
Assistant, which is the same surface test_freeze.py part 1-2 use. The wire-
level invariant the card asks for (`|emitted| <= rating` on every frame) is
asserted here over every value that reached `set_number()`, which is the last
thing this process controls; proving it at the inverter needs the witness log
and belongs with DRIVER-07.
⚠️ 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, and a size-preserving
edit reverted inside one second is invisible to CPython's staleness test. That
is TEST-15, and it has already turned one live mutation into a false survivor.
Read MUTATION.md before you report a survivor.
"""
import asyncio
import io
import math
import re
import sys
from datetime import datetime, timedelta, timezone
from named_absence import given, nth, present
from app.clamp import (
CLEAR_CYCLES, REASON_CODE, REASSERT_S, SUSTAINED_CYCLES, ClampWatchdog,
clamp_to_rating,
is_commandable,
)
from app.allocator import Allocation, UnitPlan, plan
from app.arbiter import P_LOOP, Claim, UnitBounds, allocate, resolve
from app.control import Tuning, compute
from app.inverter import Inverter
from app.main import Controller
from app import shadow
fails = []
T0 = datetime(2026, 8, 30, 9, 0, 0, tzinfo=timezone.utc)
# The live install's numbers. NOTES.md and freeze.py both say max_w: 5000 there,
# which is also the device rating, so this is the bound the alarm is written at.
LIVE_MAX_W = 5000.0
CLAMP_ID = "goodwe_output_clamp"
def check(name, cond):
print(f" {'ok ' if cond else 'FAIL'} {name}")
if not cond:
fails.append(name)
# ==============================================================================
print("1. the clamp arithmetic, on the shipped expression")
# (requested, limit, expected applied, expected engaged, expected bound)
TABLE = [
(5001.0, 5000.0, 5000.0, True, "max_w"), # +limit+1
(-5001.0, 5000.0, -5000.0, True, "max_w"), # -limit-1, the export side
(5000.0, 5000.0, 5000.0, False, ""), # exactly +limit: NOT engaged
(-5000.0, 5000.0, -5000.0, False, ""), # exactly -limit: NOT engaged
(0.0, 5000.0, 0.0, False, ""), # the failsafe value
(1234.0, 5000.0, 1234.0, False, ""),
(14768.0, 5000.0, 5000.0, True, "max_w"), # 10.3's reported runaway
]
for requested, limit, want_applied, want_engaged, want_bound in TABLE:
r = clamp_to_rating(requested, limit)
check(f"{requested:+.0f} W against a {limit:.0f} W rating -> "
f"{r.applied_w:+.0f} W, engaged={r.engaged}, bound={r.bound!r}",
r.applied_w == want_applied and r.engaged is want_engaged
and r.bound == want_bound)
# It is a MAGNITUDE clamp (3.2), so import and export must be bound identically.
# Derived over a sweep rather than asserted on two hand-picked values.
asym = [m for m in range(0, 12001, 37)
if clamp_to_rating(m, LIVE_MAX_W).applied_w
!= -clamp_to_rating(-m, LIVE_MAX_W).applied_w]
check(f"symmetric over {len(range(0, 12001, 37))} magnitudes "
f"({len(asym)} asymmetric)", not asym)
# The engagement flag has to mean "the RAW value differed", which is the words
# AC 2 uses. Re-derived from the result's own fields, not from the table.
mismatch = [m for m in range(0, 12001, 37)
for r in [clamp_to_rating(m, LIVE_MAX_W)]
if r.engaged != (r.requested_w != r.applied_w)]
check("engaged is exactly 'raw differs from applied', over the same sweep",
not mismatch)
# The device second stage. Only reachable when max_w is ABOVE the device range,
# i.e. when validate_startup_config() did not run or no longer holds.
r = clamp_to_rating(3000.0, 5000.0, dev_min=-2000.0, dev_max=2000.0)
check(f"a device narrower than max_w still bites, and says which bound "
f"({r.applied_w:+.0f} W, {r.bound!r})",
r.applied_w == 2000.0 and r.engaged and r.bound == "device")
r = clamp_to_rating(3000.0, 5000.0, dev_min=-5000.0, dev_max=5000.0)
check("...and is a no-op once the startup validation holds",
r.applied_w == 3000.0 and not r.engaged)
# ⚠️ REGRESSION, REVIEW-FOUND. THE FIRST STAGE USED TO `return` AND THE DEVICE
# STAGE WAS THEN NEVER REACHED, so the harder the overshoot the less the guard
# did: at max_w 5000 against a +-4000 device, +9999 emitted +5000 - OUTSIDE THE
# DEVICE RANGE - while a mere +4500 was correctly held to +4000. That inverts
# the guard and it was a regression against the pre-PR `write_setpoint()`, whose
# comment read "never send a value the device will reject outright".
#
# The property is asserted over a sweep, not on the two witness values, because
# the defect was invisible to every single-value check that happened to sit
# inside one stage.
DEV_LO, DEV_HI = -4000.0, 4000.0
escapes = [(m, clamp_to_rating(m, 5000.0, DEV_LO, DEV_HI).applied_w)
for m in range(-15000, 15001, 61)
for a in [clamp_to_rating(m, 5000.0, DEV_LO, DEV_HI).applied_w]
if not (DEV_LO <= a <= DEV_HI)]
big = clamp_to_rating(9999.0, 5000.0, DEV_LO, DEV_HI)
small = clamp_to_rating(4500.0, 5000.0, DEV_LO, DEV_HI)
print(f" max_w 5000 vs a device of {DEV_LO:+.0f}..{DEV_HI:+.0f}: "
f"+9999 -> {big.applied_w:+.0f} W ({big.bound}), "
f"+4500 -> {small.applied_w:+.0f} W ({small.bound})")
check(f"BOTH stages always run - {big.requested_w:+.0f} W emits "
f"{big.applied_w:+.0f} W, inside the device's {DEV_HI:+.0f} W, and names "
f"both bounds ({big.bound!r})",
big.applied_w == DEV_HI and big.bound == "max_w+device")
check(f"...and it holds over the whole range, so no overshoot escapes the "
f"device ({len(escapes)} escapes in {len(range(-15000, 15001, 61))})",
not escapes)
check("...a big overshoot is not clamped LESS than a small one",
abs(big.applied_w) <= abs(small.applied_w) + 0.0)
print(" non-finite input is loss of input, not an out-of-range value")
for bad in (None, float("nan"), float("inf"), float("-inf"), "abc", object()):
check(f"is_commandable({bad!r:.30}) is False", is_commandable(bad) is False)
for good in (0, -5000.0, 5000, 12.5):
check(f"is_commandable({good!r}) is True", is_commandable(good) is True)
# ⚠️ WHY THAT GUARD EXISTS, AND THE ANSWER IS WORSE THAN "it passes a nan
# through". MEASURED HERE, not recalled: every comparison against a nan is
# False, so `min(limit, nan)` keeps its FIRST argument and a naive clamp turns a
# nan into a command at the FULL RATING - and then reports engaged=True, because
# limit != nan, so the winding-loop counter gets a transport bug filed in it.
# The number below is computed by the same expression the guard replaces.
nan = float("nan")
naive = max(-LIVE_MAX_W, min(LIVE_MAX_W, nan))
check(f"a naive clamp turns a nan into {naive:+,.0f} W and calls it "
f"engaged={naive != nan} - hence the guard",
naive == LIVE_MAX_W and (naive != nan) is True)
# ==============================================================================
print()
print("2. the alarm's cadence: every engagement counted, no flood")
wd = ClampWatchdog()
t = T0
fired = [wd.note(True, "winding", t + timedelta(seconds=i)) for i in range(10)]
check(f"the rising edge fires ({fired[0]})", fired[0] is True)
check(f"the next 9 one-second cycles fire {sum(fired[1:])} more times "
f"(re-assert is {REASSERT_S:.0f} s)", sum(fired[1:]) == 0)
check(f"but every one of them is counted: engagements={wd.engagements}",
wd.engagements == 10)
check(f"and it is one episode, not ten: episodes={wd.episodes}",
wd.episodes == 1)
# --- THE RE-ASSERT CADENCE, PINNED IN BOTH DIRECTIONS ------------------------
# ⚠️ REVIEW-FOUND. This used to read `wd.note(True, ..., T0 + REASSERT_S + 9)`,
# which is green for ANY value of the constant - REASSERT_S = 3600 passed, and an
# hour of silence is precisely what the module promises not to do. A check whose
# stimulus is written in terms of the constant it is grading cannot grade it.
#
# The requirements are stated in wall-clock seconds and derived from what the
# alarm is FOR, so they do not move when the constant does:
#
# floor - a loop pinned on the rail must speak at least MIN_PER_QUARTER times
# inside the 900 s quarter that bills the capacity tariff. That window
# is the same one freeze.py and DOCS.md reason about.
# ceiling - it must not speak on more than MAX_FRACTION of cycles, or it is
# the per-cycle flood the split exists to remove.
QUARTER_S = 900.0
MIN_PER_QUARTER = 5
MAX_FRACTION = 0.20
pinned = ClampWatchdog()
alarms = sum(1 for i in range(int(QUARTER_S))
if pinned.note(True, "pinned", T0 + timedelta(seconds=i)))
print(f" pinned on the rail for {QUARTER_S:.0f} s at 1 s cycles: "
f"{pinned.engagements} engagements, {alarms} alarm(s) "
f"({alarms / QUARTER_S:.1%} of cycles)")
check(f"a loop pinned for a whole billing quarter speaks at least "
f"{MIN_PER_QUARTER} times (it spoke {alarms})",
alarms >= MIN_PER_QUARTER)
# ⚠️ REVIEW-FOUND, SAFETY-41, and it is the SAME DEFECT as the one recorded
# twenty lines above, one notch weaker. The sweep above runs i = 0..899 and the
# rising edge at i=0 is FREE, so it measures `1 + floor(899 / REASSERT_S)` - one
# alarm more than a real quarter earns. Measured on this very check:
# REASSERT_S = 180 -> 5 alarms, green (and 180 is the true edge: on the worst
# REASSERT_S = 224 -> 5 alarms, GREEN sliding 900 s window 180 gives 5 and
# REASSERT_S = 225 -> 4 alarms, red 181 gives 4)
# So the check as written permits 224 s while the REQUIREMENT permits 180 s, and
# anyone raising the constant to 200 s would get a green suite and conclude the
# floor was met when a real quarter would earn 4 alarms. Grade the cadence
# DIRECTLY, so the check enforces the ceiling its own comment claims for it.
check(f"...and the CADENCE itself honours the floor, independent of where the "
f"sweep happens to start: floor({QUARTER_S:.0f}/{REASSERT_S:.0f}) = "
f"{int(QUARTER_S // REASSERT_S)} >= {MIN_PER_QUARTER}",
QUARTER_S // REASSERT_S >= MIN_PER_QUARTER)
check(f"...and on no more than {MAX_FRACTION:.0%} of cycles (it was "
f"{alarms / QUARTER_S:.1%})", alarms <= MAX_FRACTION * QUARTER_S)
check("...while every single cycle is still counted",
pinned.engagements == int(QUARTER_S))
# --- THE SCATTERED REGIME, WHICH IS THE ONE THE OWNER ACTUALLY HIT -----------
# ⚠️ REVIEW-FOUND, SAFETY-41, and it falsified the first version of this card.
# The rising edge used to fire UNCONDITIONALLY, so REASSERT_S governed only the
# still-engaged path. A condition that engages and clears repeatedly therefore
# raised one notification PER EPISODE at any value of the constant.
#
# That is what the live install produced: 48 engagements in ~25 minutes,
# scattered about one cycle in six. At the measured ~5 s cycle the gaps are ~25 s
# of clean cycles, past CLEAR_CYCLES x 5 s = 15 s, so the alarm de-asserted
# between almost every engagement. Raising REASSERT_S 30 -> 150 removed NONE of
# them. The interval now bounds the notification RATE across episodes.
#
# The stimulus is written in wall-clock seconds and in the shape the field
# actually produced, so it does not move when the constant does.
SCATTER_CYCLE_S = 5.0
SCATTER_ONE_IN = 6
SCATTER_CYCLES = 305
scat = ClampWatchdog()
scat_alarms = sum(1 for i in range(SCATTER_CYCLES)
if scat.note(i % SCATTER_ONE_IN == 0, "scattered",
T0 + timedelta(seconds=SCATTER_CYCLE_S * i)))
_scat_span = SCATTER_CYCLE_S * SCATTER_CYCLES
print(f" scattered 1-in-{SCATTER_ONE_IN} over {_scat_span:.0f} s at "
f"{SCATTER_CYCLE_S:.0f} s cycles: {scat.episodes} episodes, "
f"{scat.engagements} engagements, {scat_alarms} alarm(s)")
check(f"a condition that de-asserts between engagements is rate-limited ACROSS "
f"episodes, not once per episode ({scat_alarms} alarms for "
f"{scat.episodes} episodes)",
scat_alarms < scat.episodes)
check(f"...and the rate honours the same interval the pinned case does: at most "
f"ceil({_scat_span:.0f}/{REASSERT_S:.0f}) + 1 = "
f"{int(_scat_span // REASSERT_S) + 1}",
scat_alarms <= int(_scat_span // REASSERT_S) + 1)
check("...while EVERY episode and EVERY engagement is still counted - the "
"budget is on notifications, never on evidence",
scat.episodes == scat.engagements == SCATTER_CYCLES // SCATTER_ONE_IN + 1)
# ⚠️ THE COUNTERFACTUAL, AND THE FIRST VERSION OF IT COULD NOT FAIL FOR THE
# DEFECT IT NAMED. It read `ClampWatchdog().note(True, ...) is True` - a FRESH
# object, so it only ever exercised the `last_alarm is None` branch, which is the
# trivially-true one. A fresh watchdog has no earlier episode to be held back BY.
# That is the third instance in this file of the family recorded at line 216 and
# at the 224 s sweep, and it was guarding the most safety-critical property in
# the change. Review found it.
#
# The stimulus must therefore SPEND THE BUDGET FIRST: a benign one-cycle clamp,
# a gap long enough to de-assert, then a continuous runaway. Without the
# escalation clause the runaway's first notification is held for up to
# REASSERT_S - measured at 140 s before it was added.
_esc = ClampWatchdog()
_esc.note(True, "benign", T0) # spends the budget
for _i in range(1, 1 + CLEAR_CYCLES + 1): # de-assert
_esc.note(False, "", T0 + timedelta(seconds=5.0 * _i))
_RUNAWAY_AT = 5.0 * (CLEAR_CYCLES + 2)
_esc_fires = [5.0 * (CLEAR_CYCLES + 2 + _j)
for _j in range(200)
if _esc.note(True, "runaway",
T0 + timedelta(seconds=5.0 * (CLEAR_CYCLES + 2 + _j)))]
_esc_delay = _esc_fires[0] - _RUNAWAY_AT if _esc_fires else None
print(f" budget spent by a benign episode, then a continuous runaway from "
f"t={_RUNAWAY_AT:.0f}s: first alarm at t={nth(_esc_fires, 0):.0f}s "
f"(delay {given(_esc_delay):.0f}s)")
check(f"⚠️ a SUSTAINED runaway speaks even when an earlier, unrelated episode "
f"spent the budget - it is not held for REASSERT_S ({given(_esc_delay):.0f}s "
f"delay, escalation is {SUSTAINED_CYCLES} cycles)",
_esc_delay is not None and _esc_delay < REASSERT_S / 2)
check("...and it speaks at the escalation boundary rather than immediately, "
"which is what keeps the scattered flood suppressed",
_esc_delay is not None and _esc_delay >= 5.0 * (SUSTAINED_CYCLES - 1))
# ⚠️ `last_alarm` SURVIVING THE DE-ASSERT IS NOW LOAD-BEARING, AND NOTHING
# PINNED IT. Review built the mutation: if the de-assert path ALSO refreshed
# `last_alarm = now`, the chatter case collapses from 50 notifications to 1 - the
# alarm never fires again. That is the "alarm that never fires" class, which is a
# different bug from one that fires later, and the suite was green on it.
_starve = ClampWatchdog()
_starve_fires = sum(1 for _i in range(1600)
if _starve.note(_i % 4 == 0, "chatter",
T0 + timedelta(seconds=5.0 * _i)))
print(f" chatter 1-engaged/{CLEAR_CYCLES}-clean over 8000 s: "
f"{_starve.episodes} episodes, {_starve_fires} notification(s)")
check(f"a de-assert does not RESET the notification budget, so repeated chatter "
f"cannot starve the alarm to silence ({_starve_fires} notifications over "
f"8000 s, not 1)",
_starve_fires > 10)
# --- THE HYSTERESIS, PINNED AGAINST THE CHATTER DIRECTION --------------------
# ⚠️ REVIEW-FOUND. This used to issue `range(CLEAR_CYCLES - 1)` clean cycles,
# which at CLEAR_CYCLES = 1 issues ZERO of them and passes while printing
# "0 clean cycles do NOT clear it". Green for every value >= 1 - including 1,
# which IS the chatter the constant exists to prevent. The stimulus is a literal
# now: ONE clean cycle must not clear the alarm, whatever the constant says.
wd2 = ClampWatchdog()
wd2.note(True, "winding", T0)
wd2.note(False, "", T0 + timedelta(seconds=1))
check(f"ONE clean cycle does NOT clear the alarm (active={wd2.active}) - "
f"a command dithering across the rail must not chatter",
wd2.active is True)
survived = 1
while wd2.active and survived < 50:
wd2.note(False, "", T0 + timedelta(seconds=1 + survived))
survived += 1
print(f" it took {survived} consecutive clean cycles to de-assert "
f"(CLEAR_CYCLES = {CLEAR_CYCLES})")
check(f"...and it does de-assert, after {survived} of them, rather than "
f"latching", wd2.active is False and survived == CLEAR_CYCLES)
check("...and the engagement count survives the clear, so the evidence does",
wd2.engagements == 1)
check("a clean cycle on an idle watchdog says nothing",
wd2.note(False, "", T0 + timedelta(seconds=300)) is False)
# ==============================================================================
print()
print("3. AC 1 + AC 3, wired: over the rating both clamps AND alarms")
class _Store:
data = {}
def set(self, *a):
pass
def get_time(self, *a):
return None
class _Pub:
enabled = False
def publish(self, *a, **k):
pass
def close(self):
pass
class _Hass:
"""Stub HA. `track` False models an inverter that is switched off entirely,
which is the 10.3 runaway's own setup."""
def __init__(self, grid, batt=0.0, track=True):
self.grid = grid
self.batt = batt
self.track = track
self.calls = []
self.writes = []
async def limits(self, entity):
return (-LIVE_MAX_W, LIVE_MAX_W, 1.0)
async def number(self, entity, invert=False):
return {"sensor.grid": self.grid, "sensor.soc": 50.0,
"sensor.batt": self.batt}.get(entity)
async def set_number(self, entity, value):
self.writes.append(float(value))
if self.track:
self.batt = float(value)
return True
async def call(self, domain, service, data):
self.calls.append((domain, service, data))
return True
def wired(grid, cycles=40, batt=0.0, track=True, **over):
async def go():
hass = _Hass(grid, batt=batt, track=track)
c = Controller(dict(
{"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
"soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
"meter_source": "off", "max_w": LIVE_MAX_W, "gain": 0.6,
"slew_w": 1000, "deadband_w": 15, "target_grid_w": 0,
"step_w": 10, "heartbeat_s": 10, "stale_input_s": 15},
**over), hass, _Store(), _Pub(), stop=asyncio.Event())
c.auto = True
c.dev_min, c.dev_max = -LIVE_MAX_W, LIVE_MAX_W
# The red status row exists only WHILE the clamp is engaged, and the
# hysteresis clears it a few cycles later - so it is captured live
# rather than looked for after the run, when it is correctly gone.
c.red_snapshot = []
for _ in range(cycles):
await c.read_inputs()
await c.cycle()
if c.clamp_wd.active and not c.red_snapshot:
c.red_snapshot = c.checks()
return hass, c
return asyncio.run(go())
def clamp_notes(hass):
return [d for dom, svc, d in hass.calls
if (dom, svc) == ("persistent_notification", "create")
and d.get("notification_id") == CLAMP_ID]
# THE 10.3 RUNAWAY, replayed. "With the inverter switched off entirely, an
# unclamped integrator climbed ~130 W per 4 s to over 10 kW and, pinned at the
# 5 kW rail, reported 14 768 W." `track=False` is the inverter switched off: it
# reports 0 W however hard it is commanded, so the error never closes.
hass_r, ctl_r = wired(6000.0, cycles=60, track=False)
emitted = [abs(w) for w in hass_r.writes]
# What the SAME error would have done to an accumulator with no bound - the
# transcription is deliberately the unclamped velocity form 10.3 describes, so
# the figure below is computed here and not quoted from the spec.
runaway, err = 0.0, 6000.0
for _ in range(60):
runaway += 0.6 * err
print(f" unclamped, the same 60 cycles reach {runaway:,.0f} W; "
f"the largest value that reached the device is {max(emitted):,.0f} W")
check(f"an unbounded accumulator really does run away here "
f"({runaway:,.0f} W >> {LIVE_MAX_W:.0f} W)", runaway > 3 * LIVE_MAX_W)
check(f"AC 1: NOT ONE of the {len(emitted)} values that reached the device "
f"exceeds the {LIVE_MAX_W:.0f} W rating",
emitted and max(emitted) <= LIVE_MAX_W)
# ⚠️ AND THE CLAMP IS NOT THE ONLY THING HOLDING THAT LINE HERE, which is worth
# saying out loud rather than letting the number above imply otherwise. Against
# an inverter reporting 0 W the saturation freeze engages after
# `saturation_cycles` and pins the command well BELOW the rail - that is why the
# largest emitted value is what it is and not 5000 W. The clamp is what makes it
# bounded; the freeze is what makes it small. A check that read
# `max(emitted) == max_w` would have been asserting the freeze, not the clamp.
check(f"...and the saturation freeze is what pins it at {max(emitted):,.0f} W "
f"rather than the rail", ctl_r.sat_count >= ctl_r.tuning.saturation_cycles)
notes_r = clamp_notes(hass_r)
print(f" the loop clamped on {ctl_r.clamp_wd.engagements} of {60} cycles, "
f"in {ctl_r.clamp_wd.episodes} episode(s), raising {len(notes_r)} "
f"notification(s)")
check("AC 3: and it ALARMED - silently clipping this is the defect",
ctl_r.clamp_wd.engagements > 0 and len(notes_r) >= 1)
check("...with its own notification_id, not shared with any other guard",
all(d["notification_id"] == CLAMP_ID for d in notes_r))
check("...one episode, re-asserted rather than one per cycle",
ctl_r.clamp_wd.episodes == 1 and len(notes_r) < ctl_r.clamp_wd.engagements)
check("...and it names itself a bug signal, not a limit in use",
bool(notes_r) and "out of range" in notes_r[0]["message"])
rows_r = [e for e in ctl_r.events if REASON_CODE in e]
check(f"the action log records every engagement, coalesced into one row per "
f"episode ({len(rows_r)} row(s), {ctl_r.clamp_wd.engagements} "
f"engagement(s))",
len(rows_r) == ctl_r.clamp_wd.episodes
and f"{ctl_r.clamp_wd.engagements} engagement(s)" in rows_r[0])
# ⚠️ REGRESSION, AND IT WAS A REAL BUG BEFORE IT WAS A CHECK. `log_event()` keeps
# 40 rows. A row per engagement flushes that ring inside 40 cycles of a winding
# loop, and the first thing it evicted was SAFETY-18's own "METER FROZEN?" row -
# i.e. this guard silencing another guard's log. Driven here with a plant that
# TRACKS the command, so the saturation freeze never engages and the integrator
# clamps on essentially every cycle.
hass_l, ctl_l = wired(9000.0, cycles=80, track=True)
rows_l = [e for e in ctl_l.events if REASON_CODE in e]
other_l = [e for e in ctl_l.events if REASON_CODE not in e]
print(f" 80 winding cycles: {ctl_l.clamp_wd.engagements} engagement(s) -> "
f"{len(rows_l)} log row(s), leaving room for {len(other_l)} other event(s)")
check(f"a long winding episode does NOT flood the 40-row event log "
f"({len(rows_l)} rows for {ctl_l.clamp_wd.engagements} engagements)",
ctl_l.clamp_wd.engagements > 40 and len(rows_l) <= ctl_l.clamp_wd.episodes)
check("...and the count in the row is the counter's, so nothing is lost",
f"{ctl_l.clamp_wd.engagements} engagement(s)" in nth(rows_l, -1))
# The RED row exists only while engaged, and the episode above is short, so it
# is captured mid-run rather than asserted after the hysteresis has cleared it.
check("the status page goes red WHILE it is engaged",
any(not c["ok"] and not c["warn"] and "OUTPUT CLAMP ENGAGED" in c["text"]
for c in ctl_r.red_snapshot))
check("...and once clear, a warn row still says how many times it engaged",
any(c["warn"] and f"{ctl_r.clamp_wd.engagements} time(s)" in c["text"]
for c in ctl_r.checks()))
# ⚠️ NOT A TRIP. 11.2's action is "clamp and alarm"; the loop keeps running and
# keeps writing the clamped value. If a later cut changes that it must come here.
check(f"it is NOT a trip - the command stands at {ctl_r.target:+.0f} W, not 0 W",
ctl_r.target != 0.0 and ctl_r.target == hass_r.writes[-1])
check("...and the controller still reports itself running, not failsafe",
ctl_r.auto and not ctl_r.sign_wd.tripped)
# The export side, because the clamp is a magnitude clamp and a sign bug in it
# would be invisible on the import case alone.
hass_x, ctl_x = wired(-6000.0, cycles=60, track=False)
check(f"the export side clamps and alarms too "
f"({min(hass_x.writes):+,.0f} W, {len(clamp_notes(hass_x))} note(s), "
f"{ctl_x.clamp_wd.engagements} engagement(s))",
min(hass_x.writes) < 0 and abs(min(hass_x.writes)) <= LIVE_MAX_W
and clamp_notes(hass_x) and ctl_x.clamp_wd.engagements > 0)
# ==============================================================================
print()
print("4. AC 2: saturation WITHOUT clamping must NOT raise a clamp alarm")
# A legal command the inverter will not follow. The meter sits just outside the
# deadband so the loop moves a little and then rests; `batt` is pinned far from
# whatever is commanded, which is the saturation condition (|cmd - actual| >
# saturation_w) and nothing else. The integrator never approaches the rating.
hass_s, ctl_s = wired(60.0, cycles=40, batt=2500.0, track=False)
peak = max(abs(w) for w in hass_s.writes)
print(f" saturated for {ctl_s.sat_count} consecutive cycles; the command "
f"peaked at {peak:.0f} W, {peak / LIVE_MAX_W:.1%} of the rating")
# ⚠️ THE POSITIVE HALF FIRST. A negative check that saw nothing happen is not an
# observation - if the saturation freeze never engaged, "no clamp alarm" is
# vacuous and this check is worthless.
check(f"the saturation freeze REALLY DID engage (sat_count="
f"{ctl_s.sat_count} >= {ctl_s.tuning.saturation_cycles})",
ctl_s.sat_count >= ctl_s.tuning.saturation_cycles)
check("...and the event log says so, in the saturation's own words",
any("saturation freeze" in e for e in ctl_s.events))
check(f"...while the command stayed well inside the rating "
f"({peak:.0f} W < {LIVE_MAX_W:.0f} W)", peak < LIVE_MAX_W)
# Now the negative, which is only worth anything because of the three above.
check("AC 2: NO clamp notification was raised", not clamp_notes(hass_s))
check("AC 2: the clamp counted ZERO engagements", ctl_s.clamp_wd.engagements == 0)
check("AC 2: no OUTPUT_CLAMP row in the action log",
not [e for e in ctl_s.events if REASON_CODE in e])
check("AC 2: and no clamp row on the status page",
not [c for c in ctl_s.checks() if "output clamp" in c["text"].lower()])
print(" ...and the two alarms do not share a notification_id")
ids_clamp = {d["notification_id"] for d in clamp_notes(hass_r)}
ids_other = {d.get("notification_id") for dom, svc, d in hass_r.calls
if svc == "create"} - ids_clamp
check(f"clamp ids {sorted(ids_clamp)} are disjoint from every other alarm's "
f"{sorted(x for x in ids_other if x)}", not (ids_clamp & ids_other))
print(" ...and the alarm cannot be keyed on Decision.reason")
# A cycle that clamps the integrator AND is then slew limited. `reason` reports
# the LAST stage, so a string-keyed alarm is silent on it; the boolean is not.
t = Tuning(max_w=2000.0, slew_w=100.0, gain=0.6)
d = compute(prev_w=1000.0, grid_w=6000.0, actual_w=1000.0, tuning=t, i_w=1900.0)
print(f" reason={d.reason!r} i_clamped={d.i_clamped} clamped={d.clamped}")
check("the integrator DID clamp on that cycle", d.i_clamped is True)
check("...but `reason` says 'slew-limited' - the clamp was overwritten "
"downstream, so a string-keyed alarm is SILENT here",
d.reason == "slew-limited" and "clamp" not in d.reason)
# And the same for the OUTPUT clamp, on a config that lets the integrator run
# above the rail - then both flags fire and `reason` still reports neither.
t2 = Tuning(max_w=2000.0, integrator_max_w=3000.0, slew_w=100.0, gain=0.6)
d2 = compute(prev_w=1000.0, grid_w=6000.0, actual_w=1000.0, tuning=t2, i_w=2900.0)
print(f" reason={d2.reason!r} i_clamped={d2.i_clamped} clamped={d2.clamped}")
check("both clamps engaged, and `reason` reports neither",
d2.i_clamped and d2.clamped and "clamp" not in d2.reason)
# ⚠️ HOW THE FREEZE AND THE CLAMP INTERACT, MEASURED RATHER THAN ASSUMED, AND
# THE MEASUREMENT REFUSED TWO GUESSES BEFORE THIS ONE. Written first as "a
# frozen cycle never clamps" (false), then as "only the i_w == 0 reseed
# exception clamps" (also false - at i_w == 0 with a matching actual the cycle
# is not saturated at all, so it is never frozen). What the sweep actually
# shows: a frozen cycle clamps exactly when the integrator REVERSES. control.py
# permits that on purpose - "it may fall, cross zero, or reverse outright" -
# because the alternative deadlocks the loop, and a reversal is a full swing
# that can clear the bound on the far side in one step.
#
# That matters for this ticket rather than being trivia: it is the one shape
# where the saturation freeze and a clamp engagement coincide, so it is exactly
# where an alarm that shared a channel with saturation would lose the clamp.
grid_sweep = (-9000.0, -3000.0, 3000.0, 9000.0)
grid = [(pp, g) for pp in (-1900.0, -500.0, 0.0, 500.0, 1900.0)
for g in grid_sweep]
runs = {(pp, g): compute(pp, g, 0.0, t, sat_count=5, i_w=pp) for pp, g in grid}
frozen = [k for k, d in runs.items() if d.frozen]
frozen_clamps = [k for k in frozen if runs[k].i_clamped]
# The property, stated independently of the control law's own branches: the
# correction pushes AGAINST the standing integrator, and the resulting swing
# clears the bound.
reversals = [(pp, g) for pp, g in frozen
if pp * g < 0 and abs(pp + t.gain * g) > t.max_w]
print(f" {len(frozen)} of {len(grid)} sweep cells are frozen; "
f"those that also clamp: {frozen_clamps}")
check(f"a frozen cycle CAN clamp, and only by reversing "
f"({len(frozen_clamps)} of {len(frozen)} frozen cells)",
frozen_clamps and frozen_clamps == reversals)
check("...and a frozen cycle pushing the way it already leans never clamps - "
"that is the freeze doing its job",
not [k for k in frozen if runs[k].i_clamped and k[0] * k[1] > 0])
check("a healthy cycle sets neither flag",
not any([compute(0.0, 100.0, 0.0, t).i_clamped,
compute(0.0, 100.0, 0.0, t).clamped]))
# ==============================================================================
print()
print("5. no path reaches the wire around the clamp")
def arm(c, value):
"""Give `c` a real allocation record authorising `value` on the CURRENT tick.
⚠️ SAFETY-28 MADE THIS NECESSARY, AND THE FACT THAT IT IS NECESSARY IS THE
TICKET. `write_setpoint()` now refuses any value the single allocator did
not place on this tick, so the probes below - which call it directly, the
way the heartbeat and the operator's buttons do - have to say where their
number came from. Before this they did not, and that was precisely the hole:
a second writer's number reached the wire on nothing but the fact that
somebody called the method.
⚠️ AND IT IS BUILT THROUGH THE SHIPPED `resolve()`, NOT BY HAND. A helper
that assembled an `Allocation` itself would be a second allocator living in
the test file, and every check below would then be grading the fixture's
idea of a record rather than the one main.py produces. The tick is bumped
the same way `cycle()` bumps it, for the same reason.
"""
c.alloc_tick += 1
res = resolve([Claim.set("loop", P_LOOP, value, "test stimulus")],
units=c.units, live=c.units, tick=c.alloc_tick)
c.allocation = res.allocation
return res.allocation
async def _direct():
hass = _Hass(0.0)
# ⚠️ SAFETY-10: `shadow_mode: "off"` EXPLICITLY, BECAUSE THIS FIXTURE
# NEVER PRESSES START. The shipped default is `auto`, which is SHADOW
# MODE while control is stopped - so without this line the probe below
# would be graded against a controller that is correctly refusing to
# command anything, and the clamp it is here to exercise would never be
# reached. The shipped write path is unaffected: a stopped controller
# only ever writes 0 W in production, which shadow mode also writes.
c = Controller({"setpoint_entity": "number.sp", "max_w": 2000,
"meter_source": "off", "shadow_mode": "off"},
hass, _Store(), _Pub(), stop=asyncio.Event())
c.dev_min, c.dev_max = -LIVE_MAX_W, LIVE_MAX_W
# The shape the heartbeat, shutdown() and handle_action() all use - each one
# authorised by this tick's allocation, which is what those callers have
# too: the heartbeat re-affirms `self.target`, and the buttons write 0 W,
# which needs no allocator at all (see part 7).
for wanted in (9999.0, -9999.0):
arm(c, wanted)
await c.write_setpoint(wanted, force=True)
# A nan is loss of input, not an unauthorised write: rule 1 turns it into
# 0 W ABOVE the gate, and 0 W is always authorised. No arming, on purpose -
# if the gate ever moved above the coercion this line would go red.
await c.write_setpoint(float("nan"), force=True)
arm(c, 1500.0)
await c.write_setpoint(1500.0, force=True)
return hass, c
hass_d, ctl_d = asyncio.run(_direct())
print(f" asked for 9999 / -9999 / nan / 1500 W; the device saw "
f"{hass_d.writes}")
check("a direct out-of-range write is clamped, not passed through",
hass_d.writes[:2] == [2000.0, -2000.0])
check("a nan is refused and becomes 0 W, per 11.1 rule 1",
hass_d.writes[2] == 0.0)
check("...and is NOT counted as a clamp engagement - it is a transport bug",
sum(1 for e in ctl_d.events if REASON_CODE in e) == 2)
check("an in-range value is untouched", hass_d.writes[3] == 1500.0)
check("every emitted frame satisfies |emitted| <= the rating",
all(abs(w) <= 2000.0 for w in hass_d.writes))
# --- 5b. ALLOC-01: AND THE ALLOCATOR DID NOT ADD ONE ------------------------
# ⚠️ "NO PATH REACHES THE WIRE AROUND THE CLAMP" IS A CLAIM ABOUT THE WHOLE
# ADD-ON, so it has to be re-made every time a module learns to produce a number
# that looks like a setpoint. ALLOC-01 added `app/allocator.py`, whose output is
# a dict of per-inverter watts - numbers that are exactly the shape of a thing
# somebody would write - and the temptation it creates is real: a second writer
# for a second inverter is DEV-02's card, and the wrong way to build it is a
# `set_number` call next to the allocation.
#
# Two checks, and neither is about the allocator's arithmetic:
# * it CANNOT write, and that is established from its imports rather than
# from its behaviour, because a behavioural check only sees the paths a
# fixture happens to take;
# * at the shipped roster the number that reaches the wire is still the
# arbiter's `target_w`, unclamped by the allocator, so SAFETY-03's clamp is
# still the binding one and not a second opinion after a first clamp.
import app.allocator as _alloc_mod # noqa: E402
_alloc_src = io.open(_alloc_mod.__file__, encoding="utf-8").read()
_alloc_imports = sorted(
n.split(".")[0]
for line in _alloc_src.splitlines() if line.startswith(("import ", "from "))
for n in [line.split()[1]])
print(f" app/allocator.py imports: {_alloc_imports}")
check("the allocator imports NOTHING that can reach the wire - no `hass`, no "
"`aiohttp`, not even `main`. It is a pure function of intervals and a "
"number, so 'it cannot write' is a property of the file rather than of "
"the paths a fixture happened to take",
_alloc_imports == ["dataclasses", "math"])
check("...and it never mentions the setpoint entity, `set_number`, or the "
"clamp - the three names any bypass would have to spell",
not any(n in _alloc_src for n in ("set_number", "setpoint_entity",
"clamp_to_rating")))
check("...and it must not: it has no measurement either. The allocator reading "
"P1 would be a SECOND integrator on the site measurement, which is the "
"same two-writers failure one level up",
not any(n in _alloc_src for n in ("p1", "P1Ingest", "grid_w")))
check("at the shipped roster of ONE the allocator is a pass-through, so the "
"number handed to write_setpoint() is the arbiter's own target and "
"SAFETY-03's clamp is still the FIRST and only clamp it meets. An "
"allocator that pre-clamped here would leave this file grading a value "
"that was already in range, and the clamp untested on the live path",
allocate(9999.0, {"inverter-a": UnitBounds()}) == ({"inverter-a": 9999.0},
True))
check("...and that unclamped 9999 W is exactly what part 5 above just proved "
"the write path stops at the rating. The two halves meet here: the "
"allocator does not clamp, the wire does",
hass_d.writes[0] == 2000.0)
# ==============================================================================
print()
print("6. TEST-11: the guard's evidence is THIS cycle's write, not the last one")
# WHY THIS PART EXISTS. `check_output_clamp()` sits one line below
# `await self.write_setpoint(self.target)` in cycle(), and moving it above the
# write left this file, test_freeze.py and test_loop.py ALL GREEN. That is not a
# missing assertion about line numbers - it is a missing assertion about
# EVIDENCE. The guard has two sources:
#
# * `self.loop_clamped`, set from the Decision well ABOVE the write. Still
# seen wherever the call is put, which is why the move is nearly invisible:
# on the §10.3 replay above, every engagement the alarm reports is this one.
# * `self.clamp_last`, which ONLY `write_setpoint()` assigns (main.py) and
# which is None until it has run. This half is what loses its evidence.
#
# So the check below does not pin the call's POSITION IN THE SOURCE, and must
# not: the comment at that line warns against tidying SAFETY-03's ordering and
# SAFETY-18's into one rule, and a check a rename breaks is worse than no check.
# It pins the CONSEQUENCE - that when the guard is consulted, `self.clamp_last`
# is the result of the write of THIS cycle - and part 6b runs the identical
# assertions through a deliberately refactored controller to prove it.
class _SpyWatchdog(ClampWatchdog):
"""The shipped watchdog, plus a record of what the guard could SEE.
⚠️ IT HOOKS THE CONSUMER, NOT THE GUARD'S NAME. Wrapping
`check_output_clamp` by name would make a rename - AC 2's legitimate
refactor - look identical to the defect, and this ticket is explicitly not
allowed to buy its check with a nuisance. `clamp_wd.note()` is the one call
the guard exists to make, so it is hooked instead: it is reached from the
guard under any name, from any helper, as long as that helper runs after the
write.
"""
def __init__(self):
super().__init__()
self.owner = None
self.seen = [] # clamp_last.requested_w at each consultation
self.seen_write = [] # was the WRITE-path half engaged, per cycle
def note(self, engaged, reason, now):
cl = self.owner.clamp_last
self.seen.append(None if cl is None else cl.requested_w)
self.seen_write.append(bool(cl and cl.engaged))
return super().note(engaged, reason, now)
def probed(cls=Controller, grid=9000.0, cycles=5, **over):
"""`wired()`, with the watchdog replaced by the spy and the per-cycle
command returned. A RAMP is used on purpose: the slew limiter makes every
cycle's command a different number, which is what makes "this cycle's
evidence" and "last cycle's evidence" distinguishable at all."""
async def go():
# `track=True` - the plant follows, so the saturation freeze never
# engages and the slew limiter keeps the command MOVING; against a
# dead plant the freeze pins it flat after three cycles and the ramp
# stops being a ramp.
hass = _Hass(grid, track=True)
c = cls(dict(
{"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
"soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
"meter_source": "off", "max_w": LIVE_MAX_W, "gain": 0.6,
"slew_w": 1000, "deadband_w": 15, "target_grid_w": 0,
"step_w": 10, "heartbeat_s": 10, "stale_input_s": 15},
**over), hass, _Store(), _Pub(), stop=asyncio.Event())
c.auto = True
c.dev_min, c.dev_max = -LIVE_MAX_W, LIVE_MAX_W
c.clamp_wd = _SpyWatchdog()
c.clamp_wd.owner = c
asked = []
for _ in range(cycles):
await c.read_inputs()
await c.cycle()
asked.append(c.target)
return hass, c, asked
return asyncio.run(go())
def evidence_checks(label, asked, wd):
"""The same four claims, run against the shipped controller (6a) and against
a refactored one (6b). Sharing them is the point: if a rename or a helper
changed the answer, this function would say so in the same words."""
seen = wd.seen
print(f" {label}: asked "
f"{[f'{w:+.0f}' for w in asked]}")
print(f" {label}: the guard saw "
f"{['None' if s is None else f'{s:+.0f}' for s in seen]}")
check(f"{label}: PRECONDITION - the command MOVES on every cycle, so this "
f"cycle's evidence and last cycle's are different numbers",
len(set(asked)) == len(asked) and len(asked) >= 4)
check(f"{label}: the guard was consulted exactly once per cycle",
len(seen) == len(asked))
check(f"{label}: AC 1 - on EVERY cycle self.clamp_last held the result of "
f"the write THAT CYCLE. write_setpoint() is the only line that "
f"assigns self.clamp_last, so a check_output_clamp() moved above the "
f"write grades the PREVIOUS cycle's write instead",
seen == asked)
check(f"{label}: ...and on the FIRST cycle self.clamp_last was already set "
f"- before the run's first write it is None, i.e. the write-path half "
f"of the guard has no evidence at all",
bool(seen) and seen[0] is not None)
hass_e, ctl_e, asked_e = probed()
evidence_checks("shipped", asked_e, ctl_e.clamp_wd)
# ⚠️ AND HERE IS WHY AC 3 NEEDS A STIMULUS OF ITS OWN, MEASURED RATHER THAN
# ASSERTED. Through the real cycle() the write-path half never engages: the
# arbiter files the device range and max_w as P_SAFETY LIMIT claims, so
# `resolution.target_w` is already inside both bounds before write_setpoint()
# re-applies them (main.py calls that duplication defence in depth, and it is).
# Every engagement the alarm reports on this path is therefore the loop-side
# one - which is exactly what makes the move nearly invisible, and exactly why
# 6c below drives the write path directly instead of hoping cycle() will.
check(f"MEASURED: the write-path half engaged 0 times in {len(asked_e)} real "
f"cycles ({sum(ctl_e.clamp_wd.seen_write)}), so no cycle()-driven check "
f"can be watching it", not any(ctl_e.clamp_wd.seen_write))
class _RefactoredController(Controller):
"""AC 2, as a fixture rather than as a promise: the two refactors a P3 tidy
would actually make, applied at once.
* the guard is RENAMED to `_clamp_guard()`, the old name left as the thin
delegate such a rename leaves behind;
* the write MOVES INTO A HELPER, `_emit_setpoint()`.
Neither changes when the guard runs relative to the write, so every check in
`evidence_checks()` must stay green. If one of them goes red here, the check
is grading the text of main.py and has to be thrown away.
⚠️ THE SIGNATURE GAINED `unit=` IN DEV-02 AND THIS FIXTURE FOLLOWED IT, which
is the honest update and not a weakening: `write_setpoint()` now addresses ONE
inverter (or, with `unit=None`, commands the whole site to 0 W), because a
controller with a roster cannot have one write path that means "the
inverter". A delegate that dropped the argument would silently re-broadcast
at N>=2, so it is passed through here exactly as a real refactor would have
to pass it through.
"""
async def _emit_setpoint(self, value, *, force=False, unit=None):
await Controller.write_setpoint(self, value, force=force, unit=unit)
async def write_setpoint(self, value, *, force=False, unit=None):
await self._emit_setpoint(value, force=force, unit=unit)
async def _clamp_guard(self):
await Controller.check_output_clamp(self)
async def check_output_clamp(self):
await self._clamp_guard()
# ⚠️ VERIFIED AGAINST main.py ITSELF, NOT ONLY AGAINST THIS SUBCLASS. Renaming
# `check_output_clamp` throughout main.py AND moving both the write and the guard
# into a new `_emit_and_guard()` helper called from cycle() leaves every check in
# this part green; moving the guard above the write turns the two AC 1 lines
# below red. The one thing a repo-wide rename does cost is the direct call in 6c,
# which is a mechanical find/replace and fails as an AttributeError naming the
# new method - not as an assertion claiming the ordering is wrong, which is the
# false signal this part was written to avoid.
hass_f, ctl_f, asked_f = probed(cls=_RefactoredController)
evidence_checks("renamed + helper", asked_f, ctl_f.clamp_wd)
check("AC 2: the refactored controller behaves identically to the shipped one "
"- same commands, same evidence, so the check above is not pinning "
"either name",
asked_f == asked_e and ctl_f.clamp_wd.seen == ctl_e.clamp_wd.seen)
# --------------------------------------------------------------------------- #
# 6c. AC 3 - the WRITE-PATH half, exercised
# --------------------------------------------------------------------------- #
# The stimulus is SAFETY-20's shape and that is not a coincidence: `max_w` 5000
# on an inverter that reports +-2500 is the one configuration where the value
# put on the wire DIFFERS from the value the loop asked for while every loop-side
# flag stays clean. That difference is the whole content of `clamp_last`, and a
# DEVICE-bound engagement is the only kind `loop_clamped` cannot also see.
#
# It is driven through `write_setpoint()` and the guard directly rather than
# through cycle(), because the measurement above says cycle() cannot produce it
# today. That is stated as a limitation, not hidden: what is being graded here is
# the guard's own contract - given a write that clamped, alarm - and the ordering
# in cycle() is what 6a grades.
async def _write_path(moved=False):
"""`moved` runs the two calls in the order the ticket asks about."""
hass = _Hass(0.0)
c = Controller({"setpoint_entity": "number.sp", "meter_source": "off",
"max_w": LIVE_MAX_W},
hass, _Store(), _Pub(), stop=asyncio.Event())
c.auto = True
# The inverter is SMALLER than the configured ceiling - SAFETY-20's install.
c.dev_min, c.dev_max = -2500.0, 2500.0
c.target = 3000.0
# SAFETY-28: the write path only carries numbers the allocator placed, so
# this stimulus is authorised the same way cycle()'s is. It has to be: the
# clamp is what this part grades, and a gate refusal would substitute 0 W
# and there would be no clamp to see.
arm(c, c.target)
if moved:
await c.check_output_clamp()
await c.write_setpoint(c.target)
else:
await c.write_setpoint(c.target)
await c.check_output_clamp()
return hass, c
hass_w, ctl_w = asyncio.run(_write_path())
notes_w = clamp_notes(hass_w)
print(f" max_w {LIVE_MAX_W:.0f} on a +-2500 W device, asked for "
f"{ctl_w.target:+.0f} W: the device saw {hass_w.writes[-1]:+.0f} W, "
f"bound={ctl_w.clamp_last.bound!r}, loop_clamped={ctl_w.loop_clamped[0]}")
check("PRECONDITION: the write really did clamp, on the DEVICE bound",
ctl_w.clamp_last is not None and ctl_w.clamp_last.engaged
and ctl_w.clamp_last.bound == "device"
and hass_w.writes[-1] == 2500.0)
check("...and the loop-side half is SILENT here, so self.clamp_last is the "
"only evidence that this happened at all",
ctl_w.loop_clamped == (False, ""))
check("AC 3: the guard alarmed on write-path evidence alone",
ctl_w.clamp_wd.engagements == 1 and len(notes_w) == 1)
check("...and the operator is told WHICH bound bit, so a device-rating clamp "
"is not reported as a max_w one",
bool(notes_w) and "device bound" in notes_w[0]["message"])
check("...and the action log carries the row, written by write_setpoint()",
any(REASON_CODE in e and "device bound" in e for e in ctl_w.events))
# The same stimulus, the same clamp, the two calls swapped. This is the ticket
# in one pair of runs.
hass_m, ctl_m = asyncio.run(_write_path(moved=True))
print(f" the same stimulus with the guard consulted BEFORE the write: "
f"{ctl_m.clamp_wd.engagements} engagement(s), "
f"{len(clamp_notes(hass_m))} notification(s), and the clamp still "
f"happened (engaged={ctl_m.clamp_last.engaged})")
check("THE DEFECT, DEMONSTRATED: consulted before the write the guard is "
"SILENT - self.clamp_last is still None at that point - while the wire "
"carries a clamped value either way",
not clamp_notes(hass_m) and ctl_m.clamp_wd.engagements == 0
and ctl_m.clamp_last.engaged and hass_m.writes[-1] == 2500.0)
check("...and main.py:856's consumer is the line that goes quietly false",
bool(ctl_w.clamp_last and ctl_w.clamp_last.engaged))
# --------------------------------------------------------------------------- #
# 6d. SAFETY-38 round 4 - THE ORDERED PAIR INSIDE `ClampResult.describe()`
# --------------------------------------------------------------------------- #
# ⚠️ THIS IS THE ALARM THE OWNER HAS BEEN COMPLAINING ABOUT ALL DAY, AND UNTIL
# THIS BLOCK NO test_*.py MENTIONED `describe()` AT ALL. `clamp.py:253` builds
# the whole §15 action-log row and the whole ClampWatchdog notification reason
# out of its own attributes:
#
# f"{REASON_CODE}: the computed setpoint {self.requested_w:+.0f} W was
# clamped to {self.applied_w:+.0f} W by the {self.bound} bound
# (rating {self.limit_w:.0f} W) - something upstream is winding"
#
# Transpose `requested_w` and `applied_w` and the guard reports clamping a
# setpoint UPWARD - here, to +9999 W by a bound whose rating the SAME SENTENCE
# gives as 2000 W. MEASURED on this tree: a zero-byte edit (22 507 bytes before,
# 22 507 after, sha256 0153133741497fa5 -> 47e4e371043e3201), the emitted
# sentence changes, and the full 21-file suite stayed GREEN. Nothing decides on
# the order, so nothing behavioural can ever see it - which is why the check
# below has to be on the ORDERED PAIR and on the CONTIGUOUS LITERAL. A substring
# test for either number passes under the swap; both numbers are still there.
#
# ⚠️ AND THE OPERANDS ARE READ OFF THE PLANT, NOT OFF `clamp_last`. `ctl.target`
# is what the loop asked for and `hass.writes[-1]` is what the device was
# actually sent; neither is a `ClampResult` field. Grading the sentence against
# `result.requested_w` would compare the swap against itself - DEV-02's defect,
# and the same one that made the first cut of this card's own render probe emit
# thirteen empty diffs.
async def _sentence(max_w, dev, wanted):
"""One real write over `max_w`, then the guard, on a real Controller."""
hass = _Hass(0.0)
c = Controller({"setpoint_entity": "number.sp", "meter_source": "off",
"max_w": max_w},
hass, _Store(), _Pub(), stop=asyncio.Event())
c.auto = True
c.dev_min, c.dev_max = -dev, dev
c.target = wanted
arm(c, wanted)
await c.write_setpoint(wanted)
await c.check_output_clamp()
return hass, c
hass_x, ctl_x = _sentence_run = asyncio.run(_sentence(2000.0, 5000.0, 9999.0))
notes_x = clamp_notes(hass_x)
_asked_x = ctl_x.target # what the loop asked for
_sent_x = hass_x.writes[-1] # what the device was actually sent
_rating_x = ctl_x.inverters[0].max_w # the bound the sentence names
_rows_x = [e for e in ctl_x.events if REASON_CODE in e]
print(f" max_w {_rating_x:.0f} on a +-5000 W device, asked for "
f"{_asked_x:+.0f} W: the device saw {_sent_x:+.0f} W")
print(f" action log: {_rows_x[0] if _rows_x else '(none)'}")
check("PRECONDITION: the write really clamped on the MAX_W bound, and the two "
"operands are DIFFERENT NUMBERS that RENDER differently - equal ones "
"would make the transposed sentence identical and every check below "
"vacuous. The asked-for magnitude is also strictly ABOVE the rating "
"while the sent one is not, which is what makes the swapped sentence "
"self-contradictory rather than merely wrong",
ctl_x.clamp_last is not None and ctl_x.clamp_last.engaged
and ctl_x.clamp_last.bound == "max_w"
and _asked_x != _sent_x
and f"{_asked_x:+.0f}" != f"{_sent_x:+.0f}"
and abs(_sent_x) <= _rating_x < abs(_asked_x))
check("⚠️ THE ORDERED PAIR IN THE §15 ACTION-LOG ROW (main.py:1366). The "
"contiguous literal names the setpoint the loop COMPUTED first and the "
"one that REACHED THE WIRE second. Under the transposition this row says "
"the loop asked for +2000 W and the guard clamped it UP to +9999 W - "
"past the 2000 W rating printed four words later - which is the alarm "
"reading backwards on the one line an operator is woken by",
any(f"the computed setpoint {_asked_x:+.0f} W was clamped to "
f"{_sent_x:+.0f} W" in e for e in _rows_x))
check("⚠️ ...AND IN THE ClampWatchdog NOTIFICATION REASON (main.py:2452), "
"which is a SECOND surface off the same sentence and has to be pinned "
"separately - SAFETY-38 round 3's whole lesson was that pinning one call "
"site leaves the other shipping",
bool(notes_x)
and f"the computed setpoint {_asked_x:+.0f} W was clamped to "
f"{_sent_x:+.0f} W" in notes_x[0]["message"])
check("...and the TRANSPOSED sentence appears on NEITHER surface, so the two "
"checks above discriminate rather than decorate",
not any(f"the computed setpoint {_sent_x:+.0f} W was clamped to "
f"{_asked_x:+.0f} W" in e for e in _rows_x)
and f"the computed setpoint {_sent_x:+.0f} W was clamped to "
f"{_asked_x:+.0f} W" not in nth(notes_x, 0)["message"]
and present(nth(notes_x, 0)))
# The DEVICE bound, off part 6c's controller: same sentence, other stage. The
# unit of coverage is the call site, and `bound` is the only thing that differs
# between the two rows an operator can receive.
_asked_w, _sent_w = ctl_w.target, hass_w.writes[-1]
check("...and the same ordered pair holds on the DEVICE-bound row, where the "
"loop asked for +3000 W and the device took +2500 W. PRECONDITION: those "
"two also render differently",
f"{_asked_w:+.0f}" != f"{_sent_w:+.0f}"
and any(f"the computed setpoint {_asked_w:+.0f} W was clamped to "
f"{_sent_w:+.0f} W" in e for e in ctl_w.events)
and f"the computed setpoint {_asked_w:+.0f} W was clamped to "
f"{_sent_w:+.0f} W" in nth(notes_w, 0)["message"])
# ==============================================================================
print()
print("7. SAFETY-28: the write path is gated on the allocation record")
# WHY THIS PART IS IN THIS FILE. Part 5 already claims "no path reaches the wire
# around the clamp", and the clamp is the last thing between a bug and the
# hardware. SAFETY-28 adds a second thing on the same path and asks a different
# question of it: not "is this number in range" but "WHO CHOSE THIS NUMBER".
#
# THE DEFECT IT REPLACES. specs §1.5 makes two writers on one actuator a
# non-goal, and until this ticket that was enforced by
# `len(validate.ACTUATOR_WRITE_KEYS) == 1` - one option key, so a second writer
# is not expressible. True, and a statement about the OPTIONS SCHEMA rather than
# about the writes: add CFG-04's per-inverter config and there are two real
# writers with that length still 1. test_validate.py part 2 grades the config
# half; this part grades the write half, which is the one that does not care
# what the options look like.
A_UNIT, B_HYPO = "inverter-a", "inverter-b"
def fresh(**over):
"""A Controller with a stub HA, no cycle run, and therefore NO record.
⚠️ SAFETY-10: `shadow_mode: "off"` for `_direct()`'s reason - this fixture
never presses Start, and the shipped default is shadow mode while control
is stopped. Part 7 grades who is allowed to reach the WIRE; a controller
that is correctly commanding nothing cannot answer that question.
"""
hass = _Hass(0.0)
c = Controller(dict({"setpoint_entity": "number.sp", "max_w": 2000,
"meter_source": "off", "shadow_mode": "off"}, **over),
hass, _Store(), _Pub(), stop=asyncio.Event())
c.dev_min, c.dev_max = -LIVE_MAX_W, LIVE_MAX_W
return hass, c
# --------------------------------------------------------------------------- #
# 7a. AC 1 - an unauthorised value is REFUSED AT THE WRITE PATH
# --------------------------------------------------------------------------- #
# The second writer, modelled as what it would actually look like in code: some
# other strategy with arithmetic of its own calling the one method that reaches
# the wire. It has a perfectly reasonable number - in range, finite, nothing the
# clamp would object to - and the only thing wrong with it is that nothing
# allocated it.
async def _second_writer():
hass, c = fresh()
arm(c, 1500.0) # the allocator placed 1500 W this tick
await c.write_setpoint(1500.0, force=True) # the loop's own write
await c.write_setpoint(1200.0, force=True) # ...and the second writer's
return hass, c
hass_2w, ctl_2w = asyncio.run(_second_writer())
print(f" allocator placed +1500 W; a second writer asked for +1200 W; "
f"the device saw {hass_2w.writes}")
check("AC 1: a value the allocation record does not contain never reaches the "
"device. 1200 W is in range and finite - the clamp has no opinion about "
"it - so the ONLY thing that stopped it is that no allocator placed it",
1200.0 not in hass_2w.writes)
check("...and it is REFUSED, not merely logged: 0 W goes on the wire in its "
"place. Returning instead would leave the inverter latched on the last "
"command with the ESP32 heartbeat still being fed - holding an "
"unauthorised setpoint quietly, which is the outcome this gate exists "
"to prevent",
hass_2w.writes == [1500.0, 0.0])
check("...and the action log says so in words an operator can act on",
any("WRITE REFUSED" in e for e in ctl_2w.events))
check("...while the authorised write of the SAME shape went through "
"untouched, so the gate is not simply refusing everything",
hass_2w.writes[0] == 1500.0)
# --------------------------------------------------------------------------- #
# 7b. AC 2 - a record from a previous tick cannot authorise a write in this one
# --------------------------------------------------------------------------- #
# ⚠️ THIS IS THE CACHED-ENVELOPE FAULT WEARING A DIFFERENT HAT. allocator.py's
# rule is that a unit's interval is read per tick and never cached, because a
# pack at its floor or a stale link moves it between one cycle and the next.
# A record is that envelope written down - so code that keeps one and replays it
# is caching the envelope through the record, and the allocator's own purity
# cannot stop it. The tick stamp is what stops it.
async def _replay():
hass, c = fresh()
stale = arm(c, 1500.0) # tick N
await c.write_setpoint(1500.0, force=True)
arm(c, 800.0) # tick N+1: the envelopes have moved on
c.allocation = stale # ...and somebody puts the old one back
await c.write_setpoint(1500.0, force=True)
return hass, c, stale
hass_rp, ctl_rp, stale_rec = asyncio.run(_replay())
print(f" a tick-{stale_rec.tick} record replayed on tick "
f"{ctl_rp.alloc_tick}: the device saw {hass_rp.writes}")
check("AC 2: replaying the previous tick's record does NOT authorise its "
"value again - the write is refused and 0 W goes out instead",
hass_rp.writes == [1500.0, 0.0])
check("...and the refusal names both ticks, so the reason is legible rather "
"than 'computer says no'",
"tick" in (ctl_rp.write_refusal(1500.0) or "")
and str(stale_rec.tick) in (ctl_rp.write_refusal(1500.0) or "")
and str(ctl_rp.alloc_tick) in (ctl_rp.write_refusal(1500.0) or ""))
check("...and an UNSTAMPED record - one built by a caller that passed no tick "
"at all - is refused too, because 'I cannot say when I was made' must "
"fail closed and not read as tick zero",
ctl_rp.allocation is not None
and (lambda c: (setattr(c, "allocation",
plan(1500.0, {A_UNIT: UnitBounds()}))
or c.write_refusal(1500.0) is not None))(ctl_rp))
# ⚠️ AND THE TICK HAS TO ADVANCE ON THE LIVE PATH, WHICH IS A DIFFERENT CLAIM
# FROM ANY OF THE ABOVE. Everything so far drives `arm()`, which bumps the
# counter itself - so all of it would stay green with `self.alloc_tick += 1`
# deleted from cycle(), and the replay protection would be gone on the only path
# that matters. This drives REAL cycles and asks the same question of them.
async def _live_staleness():
hass = _Hass(6000.0)
c = Controller({"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
"soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
"meter_source": "off", "max_w": LIVE_MAX_W, "gain": 0.6,
"slew_w": 1000, "step_w": 10, "stale_input_s": 15},
hass, _Store(), _Pub(), stop=asyncio.Event())
c.auto = True
c.dev_min, c.dev_max = -LIVE_MAX_W, LIVE_MAX_W
await c.read_inputs()
await c.cycle()
first, first_v = c.allocation, c.target
await c.read_inputs()
await c.cycle()
# ...and now the HOLD: one missed poll, which must NOT age the record, or a
# single dropped meter reading would drop the site to 0 W.
held_before = c.alloc_tick
c.grid = None
c.inputs_bad_since = datetime.now(timezone.utc)
await c.cycle()
hass.writes.clear()
await c.write_setpoint(c.target, force=True) # the heartbeat's own call
return c, first, first_v, held_before, hass
ctl_ls, rec1, val1, held_before, hass_ls = asyncio.run(_live_staleness())
print(f" two real cycles: tick {rec1.tick} -> {ctl_ls.allocation.tick}; "
f"a held cycle left it at {ctl_ls.alloc_tick}")
check("PRECONDITION: two real cycles produced two records with two different "
"ticks and a value worth replaying",
rec1.tick != ctl_ls.allocation.tick and val1 != 0.0)
check("the tick advances on the LIVE path, so cycle 1's record does not "
"authorise a write on cycle 2. Everything above this line drives the "
"test's own `arm()` and would stay green with the increment deleted from "
"cycle() - this is the check that would not",
(lambda c: (setattr(c, "allocation", rec1)
or c.write_refusal(val1) is not None))(ctl_ls))
check("⚠️ AND A CYCLE THAT PRODUCED NO RESOLUTION DOES NOT AGE IT. The "
"'inputs missing, not yet stale' hold returns before the allocator runs, "
"precisely so one dropped meter reading does not zero the inverter, and "
"the heartbeat re-affirms self.target through this same gate while it "
"does. A tick incremented per LOOP ITERATION rather than per RESOLUTION "
"would refuse that heartbeat and turn the hold into the outage it exists "
"to prevent",
ctl_ls.alloc_tick == held_before and hass_ls.writes == [ctl_ls.target]
and ctl_ls.target != 0.0)
# --------------------------------------------------------------------------- #
# 7c. THE FAILSAFE CANNOT DEADLOCK BEHIND THE GATE
# --------------------------------------------------------------------------- #
# ⚠️ THE HAZARD THIS PART EXISTS FOR. `write_setpoint()` is also how layer 2
# commands 0 W: cycle()'s two refusal branches substitute a 0 W failsafe and
# write it like any other cycle, and shutdown() and the operator's stop button
# do the same. Every one of those runs while the allocator has REFUSED - `ok` is
# False and there are no shares - so a gate that asked the allocator's
# permission to command zero would block the safety write behind the very
# component whose failure the safety write answers. A gate that can do that is
# worse than the hole it closes.
#
# The resolution: 0 W is authorised unconditionally, and it is authorised on the
# VALUE rather than on a `failsafe=True` flag the caller asserts. The argument is
# at Controller.write_refusal(): the hazard §1.5 names is N-times loop gain from
# broadcasting one target to N inverters, and at S = 0 the broadcast and the
# correct allocation COINCIDE - a_i = 1 for every unit gives every unit 0 W,
# which is exactly its share. There is no over-command to make at zero.
_hostile = []
_h, _c = fresh()
_hostile.append(("no record at all", _c))
_h, _c = fresh()
_c.alloc_tick = 1
_c.allocation = plan(1500.0, {A_UNIT: UnitBounds()}, scope_errors=("mis-routed",),
tick=1)
_hostile.append(("the allocator REFUSED this tick", _c))
_h, _c = fresh()
_c.allocation = arm(_c, 1500.0)
_c.alloc_tick += 7 # ...and then seven ticks went by
_hostile.append(("only a stale record", _c))
for _why, _c in _hostile:
check(f"0 W is authorised with {_why}", _c.write_refusal(0.0) is None)
check(f"...and a NON-zero write is still refused with {_why}, so the zero "
f"exit is a floor and not an amnesty",
_c.write_refusal(1500.0) is not None)
async def _failsafe_wire():
"""The refusal path END TO END, through the real cycle().
⚠️ THE ROSTER HERE IS A HYPOTHESIS AND NOTHING IS MEASURED FROM IT.
INVERTER B DOES NOT EXIST. It is named only so `live_units()` can return
nothing and drive the allocator into REFUSE_EMPTY, which is the state
main.py's first refusal branch is written for and the one this check needs:
the allocator itself has failed, and layer 2 must still be able to command
zero.
"""
hass = _Hass(0.0)
class _Blind(Controller):
def live_units(self):
return ()
c = _Blind({"setpoint_entity": "number.sp", "meter_source": "off",
"max_w": LIVE_MAX_W, "soc_entity": "sensor.soc",
"batt_entity": "sensor.batt", "meter_entity": "sensor.grid"},
hass, _Store(), _Pub(), stop=asyncio.Event())
# ⚠️ DEV-02: THE ROSTER IS GROWN BY ADDING AN `Inverter`, NOT BY ASSIGNING
# `units`. `Controller.units` is now DERIVED from `Controller.inverters` and
# has no setter, because a settable roster is a site that claims two
# actuators while holding one writer - a half-driven site with nothing to
# say so. The object carries the writer, which is what makes this fixture a
# roster of two rather than a tuple of two names.
c.dev_min, c.dev_max = -LIVE_MAX_W, LIVE_MAX_W # while the roster is one
_b = Inverter(B_HYPO, setpoint_entity="number.sp_b",
soc_entity="sensor.soc", batt_entity="sensor.batt",
max_w=LIVE_MAX_W, store=c.store)
_b.dev_min, _b.dev_max = -LIVE_MAX_W, LIVE_MAX_W
_b.soc, _b.batt = 50.0, 0.0
c.inverters.append(_b)
c.auto = True
for _ in range(3):
await c.read_inputs()
await c.cycle()
return hass, c
hass_fs, ctl_fs = asyncio.run(_failsafe_wire())
print(f" allocator refused every tick "
f"({ctl_fs.allocation.refused!r}); the device saw {hass_fs.writes}")
check("the allocator having failed, the 0 W failsafe still REACHES THE WIRE - "
"this is the deadlock the gate must not have, driven through the real "
"cycle() and its real refusal branch rather than asserted",
not ctl_fs.allocation.ok and hass_fs.writes == [0.0, 0.0]
and ctl_fs.target == 0.0)
check("...and it was the ALLOCATOR's refusal that produced it, not the gate's "
"- if the gate were the thing zeroing this cycle the action log would "
"say WRITE REFUSED and the distinction between the two failures would "
"be lost to whoever has to debug it",
any("ARBITER: REFUSED" in e for e in ctl_fs.events)
and not any("WRITE REFUSED" in e for e in ctl_fs.events))
# --------------------------------------------------------------------------- #
# 7d. `unmet_w` IS NOT `ok`, AND THE GATE MUST NOT FUSE THEM
# --------------------------------------------------------------------------- #
# allocator.py says it plainly: an allocation that could not place every watt is
# still SAFE TO WRITE and strictly better than commanding zero. Fusing the two
# would let ONE SATURATED INVERTER PARK THE WHOLE SITE - the site target goes
# unmet by 3800 W, every share is legal, and a gate reading `unmet_w` would
# answer that by commanding 0 W and meeting even less of it.
_tight = plan(4000.0, {A_UNIT: UnitBounds(lo=-100.0, hi=100.0),
B_HYPO: UnitBounds(lo=-100.0, hi=100.0)}, tick=1)
_h, _c = fresh()
_c.alloc_tick, _c.allocation = 1, _tight
print(f" saturated site: {_tight.explain()}")
check("PRECONDITION: this record is ok AND has watts it could not place",
_tight.ok and _tight.unmet_w == 3800 and _tight.shares[A_UNIT] == 100)
check("the gate authorises this unit's share anyway. `unmet_w` is not read - "
"it is the site target that went unmet, not a broken allocation",
_c.write_refusal(100.0) is None)
check("...and it still refuses a value that is not the share, so 'do not read "
"unmet_w' has not been implemented as 'read nothing'",
_c.write_refusal(4000.0) is not None)
# --------------------------------------------------------------------------- #
# 7e. A RECORD THAT CONTRADICTS ITSELF AUTHORISES NOTHING
# --------------------------------------------------------------------------- #
# `UnitPlan.explain()` renders a share outside its own interval as
# "ENVELOPE VIOLATED, DO NOT WRITE". This is the DO NOT WRITE. The shape is the
# half-open clamp defect ALLOC-01 fixed: a plan stating the floor and a share
# below it in the same breath, and agreeing with itself.
#
# ⚠️ UNREACHABLE FROM THE ALLOCATOR AS IT STANDS, AND GRADED ANYWAY. At the
# shipped roster of one the pass-through's interval is (-inf, +inf) and every
# number is inside it. PR #40's reviewer proved what an unreachable ungraded
# branch is worth by replacing one with `if False:` and watching the whole suite
# stay green, so this one is exercised on a constructed record rather than
# trusted.
_liar = Allocation(target_w=500.0, sum_w=500, shares={A_UNIT: 500},
plans=(UnitPlan(A_UNIT, 500, 2000.0, math.inf, 1.0,
"weighted share"),), tick=1)
_h, _c = fresh()
_c.alloc_tick, _c.allocation = 1, _liar
print(f" {_liar.plans[0].explain()}")
check("PRECONDITION: the record is `ok`, and its share IS what the gate would "
"otherwise match on - so nothing but the envelope question stops it",
_liar.ok and _liar.shares[A_UNIT] == 500)
check("a share outside its own plan's interval is refused, and the refusal "
"quotes the plan's own DO NOT WRITE line",
"DO NOT WRITE" in (_c.write_refusal(500.0) or ""))
# --------------------------------------------------------------------------- #
# 7f. AC 5 - AT N=1 NOTHING CHANGES
# --------------------------------------------------------------------------- #
# The strongest form of this is not a new assertion, it is the 17 other files in
# this suite passing unedited on real cycles. What is added here is the
# statement that the live path is not being quietly zeroed: over part 3's
# sixty-cycle replay - the 10.3 runaway, an inverter that never follows - the
# gate refused nothing at all.
check("AC 5: across part 3's 60 real cycles the gate refused ZERO writes, so "
"the shipped one-inverter path reaches the wire exactly as it did "
"before. A gate that silently substituted 0 W here would leave every "
"clamp check above green and the install dead",
not [e for e in ctl_r.events if "WRITE REFUSED" in e])
check("...and the numbers those cycles put on the wire are the arbiter's own "
"target, unrounded and unaltered by the record it now has to appear in",
hass_r.writes and all(abs(w) <= LIVE_MAX_W for w in hass_r.writes))
# --------------------------------------------------------------------------- #
# 8. SAFETY-10 - SHADOW MODE: THE DECISION PATH RUNS AND NOTHING REACHES THE
# INVERTER
# --------------------------------------------------------------------------- #
# This part belongs in THIS file and not in a new one: shadow mode is a
# substitution inside `_write_unit()`, one line below SAFETY-03's clamp and two
# below SAFETY-28's gate, and the property it claims - "nothing reaches the
# inverter" - is the same property parts 5 and 7 already grade from the other
# side. The fake `set_number()` in this file IS the wire.
#
# ⚠️ EVERY CHECK BELOW IS GRADED ON THE WRITES OR ON THE RENDERED SURFACE, NOT
# ON `Controller.shadow`. A check that asserted the flag would pass on a build
# in which the flag is set and the substitution is deleted, which is precisely
# the build this section exists to refuse.
#
# ⚠️ AND EVERY ONE OF THEM RUNS THE SAME STIMULUS IN BOTH MODES. MUTATION.md's
# rule: a per-thing check that has never seen the things DISAGREE is a total.
# `_s10_live` and `_s10_shadow` are the same 6000 W import through the same
# `Controller.cycle()`; the only difference is one option.
print("\n8. SAFETY-10 - shadow mode\n")
def _s10(shadow_mode, *, auto=True, cycles=25, grid=6000.0, dev=None, **over):
"""The same wired run as part 3, with `shadow_mode` and the start button.
⚠️ `auto` IS A PARAMETER BECAUSE `shadow_mode: auto` READS IT. The three
values and the two button positions are six states and they are not all
the same state; the pure-function table over all six is in
`test_validate.py`, and 8f says why it is not here.
"""
async def go():
hass = _Hass(grid, track=True)
c = Controller(dict(
{"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
"soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
"meter_source": "off", "max_w": LIVE_MAX_W, "gain": 0.6,
"slew_w": 1000, "deadband_w": 15, "target_grid_w": 0,
"step_w": 10, "heartbeat_s": 10, "stale_input_s": 15,
"shadow_mode": shadow_mode},
**over), hass, _Store(), _Pub(), stop=asyncio.Event())
c.auto = auto
# `dev` narrows what the DEVICE says it will accept, which is the
# clamp's second stage - the only way to make the clamp engage on a
# target the loop has already bounded at `max_w`.
c.dev_min, c.dev_max = ((-LIVE_MAX_W, LIVE_MAX_W) if dev is None
else (-dev, dev))
# ⚠️ CAPTURED PER CYCLE, BECAUSE THE HEARTBEAT OVERWRITES IT.
# `clamp_last` is assigned by every write, and the heartbeat below
# writes `inv.target` - 0 W in shadow mode - so the clamp's verdict on
# the DECISION is gone by the end of the run. Looking for it afterwards
# would grade the refresh instead of the cycle.
c.clamp_seen = []
for n in range(cycles):
await c.read_inputs()
await c.cycle()
if c.clamp_last is not None:
c.clamp_seen.append(c.clamp_last)
# ⚠️ THE HEARTBEAT IS DRIVEN HERE AND NOT LEFT OUT, BECAUSE IT IS
# THE HALF THAT PROVES SHADOW MODE IS NOT SILENCE. `_write_unit()`
# skips an unchanged value unless `force` is set, so a cycles-only
# fixture sees exactly ONE 0 W write and could not tell "holds the
# inverter at 0 W" from "wrote 0 W once and went quiet" - and the
# second of those leaves a real install's ESP32 in a tripped
# stale-input watchdog for the whole run. `heartbeat_s=0` makes
# every tick due, which is the shipped path with its clock wound
# forward rather than a second implementation of it.
await c.send_heartbeats(float(n), 0.0)
return hass, c
return asyncio.run(go())
_live_h, _live_c = _s10("off")
_shad_h, _shad_c = _s10("on")
# --- 8a. THE PRECONDITION, WHICH IS THE HALF THAT MAKES THE REST MEAN ANYTHING
# MUTATION.md habit 4: a check whose DATA cannot reach the failure it names is
# decoration. "No non-zero value was written" is trivially true of a stimulus
# that would not have produced one, so the same stimulus is first shown to
# produce a large one with the option off.
check("PRECONDITION: with shadow_mode off this stimulus really does drive the "
f"wire - {len(_live_h.writes)} writes, "
f"{len([w for w in _live_h.writes if w != 0.0])} of them non-zero, "
f"largest {max((abs(w) for w in _live_h.writes), default=0):.0f} W",
len([w for w in _live_h.writes if w != 0.0]) >= 10
and max((abs(w) for w in _live_h.writes), default=0) > 1000.0
and 0.0 not in _live_h.writes)
# --- 8b. NOTHING THE DECISION PATH PRODUCED REACHES THE INVERTER -------------
check("AC 1: in shadow mode EVERY value that reached set_number() is 0 W. "
f"Distinct values written: {sorted(set(_shad_h.writes))}",
_shad_h.writes and set(_shad_h.writes) == {0.0})
# ⚠️ AND IT IS NOT SILENCE. The inverter holds its last command forever and the
# ESP32's stale-input watchdog needs a refresh; a shadow mode that stopped
# writing would leave a standing command up under a page saying nothing is
# commanded. A check that only asserted "no non-zero write" would pass on that
# build too - this is the half that separates them.
check("...and it WROTE that 0 W rather than going silent, which is what keeps "
f"the inverter from holding its last command ({len(_shad_h.writes)} "
"writes)",
len(_shad_h.writes) >= 2)
# --- 8c. THE DECISION PATH ACTUALLY RAN ------------------------------------
# "Nothing reached the inverter" is also true of a controller that decided
# nothing. Both halves, or the feature is an off switch with a long name.
check("the full decision path still ran: the loop reached a target and the "
f"arbiter resolved one ({_shad_c.loop_w!r} W, "
f"would have commanded {_shad_c.shadow_would_w!r} W)",
_shad_c.loop_w is not None and abs(_shad_c.loop_w) > 500.0
and _shad_c.shadow_would_w is not None
and abs(_shad_c.shadow_would_w) > 500.0)
check("...and the decision it recorded is a number a live controller would "
"have written - it is inside the rail and it is not the 0 W that went "
"on the wire",
_shad_c.shadow_would_w != 0.0
and abs(_shad_c.shadow_would_w) <= LIVE_MAX_W)
# --- 8d. THE CARRIED COMMAND TRACKS THE WIRE, WHICH IS WHAT MAKES THE EXIT
# SAFE
check("the carried site command is what is STANDING ON THE WIRE, not the "
f"decision ({_shad_c.target!r} W standing, "
f"{_shad_c.shadow_would_w!r} W decided)",
_shad_c.target == 0.0 and _shad_c.shadow_would_w != 0.0)
check("...and every unit's share is too, so SAFETY-42's delivery watchdog is "
"not handed a share nothing was told to produce",
all(i.target == 0.0 for i in _shad_c.inverters))
def _s10_exit(shadow_first):
"""Run twelve cycles, force shadow mode off, take ONE more cycle.
Answers with `inv.last_write` either side of that boundary - WHAT STANDS ON
THE WIRE, updated only on an accepted write, which is the number the
inverter is actually latched on. Reading `hass.writes` instead would be
blind to the converged case, where the controller correctly re-sends
nothing at all.
⚠️ THE COMPARISON IS AGAINST THE SAME RUN THAT NEVER ENTERED SHADOW MODE,
not against a constant. `slew_w` bounds a step from the standing command,
so "did it resume from the wire or from a fiction" is a question about a
DIFFERENCE, and a hard-coded expectation would be the number the code
produces compared with itself.
"""
async def go():
hass = _Hass(6000.0, track=True)
c = Controller(
{"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
"soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
"meter_source": "off", "max_w": LIVE_MAX_W, "gain": 0.6,
"slew_w": 1000, "deadband_w": 15, "target_grid_w": 0,
"step_w": 10, "heartbeat_s": 10, "stale_input_s": 15,
"shadow_mode": "on" if shadow_first else "off"},
hass, _Store(), _Pub(), stop=asyncio.Event())
c.auto = True
c.dev_min, c.dev_max = -LIVE_MAX_W, LIVE_MAX_W
for _ in range(12):
await c.read_inputs()
await c.cycle()
before = c.inverters[0].last_write
c.o["shadow_mode"] = "off"
await c.read_inputs()
await c.cycle()
# ⚠️ NOT `float()`. `last_write` is None until something is ACCEPTED,
# and a build in which shadow mode writes nothing at all would turn
# this fixture into a TypeError - which MUTATION.md's FCST-06 section
# is about: a traceback grades none of the checks after it AND reads
# like a survivor. It is handed back as None and named below instead.
return before, c.inverters[0].last_write, c
return asyncio.run(go())
_exit_before, _exit_after, _exit_c = _s10_exit(True)
_never_before, _never_after, _never_c = _s10_exit(False)
print(f" shadow run: standing {_exit_before!r} W -> {_exit_after!r} W on "
"the first live cycle")
print(f" never in shadow: standing {_never_before!r} W -> "
f"{_never_after!r} W on the same cycle")
check("SOMETHING REACHED THE WIRE IN SHADOW MODE. `last_write` is set only on "
"an ACCEPTED write, so None here is the build that stops writing "
"altogether and leaves the inverter latched on its last command - the "
"failure shadow mode must not be a new way to reach",
_exit_before is not None and _exit_after is not None
and _never_before is not None and _never_after is not None)
_exit_before = -1.0 if _exit_before is None else float(_exit_before)
_exit_after = -1.0 if _exit_after is None else float(_exit_after)
_never_before = 0.0 if _never_before is None else float(_never_before)
_never_after = 0.0 if _never_after is None else float(_never_after)
check("PRECONDITION: the two arrangements DISAGREE - the run that was never in "
"shadow mode is standing somewhere else entirely at that boundary, so "
"the comparison below is not a number against itself",
_never_before != _exit_before and abs(_never_before) > 1000.0)
check("PRECONDITION: and the shadow run really was standing at 0 W, so the "
"step below is measured from the wire and not from a coincidence",
_exit_before == 0.0)
check("AC 2, leaving: the first live command after shadow mode is one slew "
"step from the 0 W that was actually standing on the wire - NOT one step "
"from the decision shadow mode reached, which never existed at the "
f"battery. |{_exit_after:.0f}| <= slew 1000 W",
abs(_exit_after) <= 1000.0)
check("...and the run that was never in shadow mode is NOT inside that bound "
"at the same cycle, which is what says the bound is a real constraint "
"here and not a property of the stimulus",
abs(_never_after) > 1000.0)
def _s10_enter():
"""Live first, then shadow mode, on a battery standing somewhere else."""
async def go():
hass = _Hass(6000.0, track=True)
c = Controller(
{"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
"soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
"meter_source": "off", "max_w": LIVE_MAX_W, "gain": 0.6,
"slew_w": 1000, "deadband_w": 15, "target_grid_w": 0,
"step_w": 10, "heartbeat_s": 10, "stale_input_s": 15,
"shadow_mode": "off"}, hass, _Store(), _Pub(),
stop=asyncio.Event())
c.auto = True
c.dev_min, c.dev_max = -LIVE_MAX_W, LIVE_MAX_W
for _ in range(12):
await c.read_inputs()
await c.cycle()
standing = hass.writes[-1]
c.o["shadow_mode"] = "on"
await c.read_inputs()
await c.cycle()
# ⚠️ `inverters[0].last_write`, NOT `hass.writes[-1]`. The question is
# WHAT THE INVERTER IS LATCHED ON after the first shadow cycle, and a
# build that writes NOTHING leaves it latched on the live value while
# `hass.writes[-1]` still reads that same number - the two look
# identical from the write list and are opposite at the battery.
return standing, float(c.inverters[0].last_write), c
return asyncio.run(go())
_stood, _first_shadow, _enter_c = _s10_enter()
print(f" standing when shadow mode came on: {_stood:.0f} W")
check("PRECONDITION: the battery was standing at a real, non-zero command when "
"shadow mode was switched on - otherwise 'it reached 0 W' says nothing",
abs(_stood) > 500.0)
check("AC 2, entering: after ONE shadow cycle the inverter is latched on 0 W, "
"not on what it was doing a cycle earlier. The inverter holds its last "
"command FOREVER, so a shadow mode that merely stopped writing would "
"leave that command standing under a page saying nothing is commanded "
f"({_stood:+.0f} W -> {_first_shadow:+.0f} W)",
_first_shadow == 0.0)
# --- 8e. THE OTHER GUARDS ON THE WRITE PATH STILL OBSERVE --------------------
# The substitution is the LAST thing before the wire, below SAFETY-28's gate and
# below SAFETY-03's clamp, so both still grade the number the decision path
# produced. A shadow run that could not say "it would have hit the rating here"
# is answering a different question from the one it is for.
_cl_h, _cl_c = _s10("on", cycles=12, grid=9000.0, dev=400.0)
_cl_req = [r.requested_w for r in _cl_c.clamp_seen]
print(f" decision {_cl_c.shadow_would_w!r} W; what the clamp was asked "
f"about at each cycle boundary: {sorted(set(_cl_req))}")
check("PRECONDITION: the decision is a real, non-zero number and the wire "
"carried only 0 W on this run, so the two are distinguishable",
_cl_c.shadow_would_w is not None and _cl_c.shadow_would_w != 0.0
and set(_cl_h.writes) == {0.0})
check("SAFETY-03 still OBSERVES in shadow mode: the clamp was asked about the "
"number THE DECISION PATH PRODUCED, not about the 0 W that went on the "
"wire. That is what puts the substitution BELOW the clamp, and it is "
"what lets a shadow run report that the rating would have been hit",
_cl_req and all(r == _cl_c.shadow_would_w for r in _cl_req))
check("...and the clamp's own alarm still ran on it - the winding-loop "
"engagement reached the action log on a run where nothing was commanded",
any("clamp" in e.lower() or "CLAMP" in e for e in _cl_c.events))
check("SAFETY-28's gate still OBSERVES too - it refused nothing on this run, "
"which is the same answer it gives live, so shadow mode has not been "
"implemented by making every write unauthorised",
not [e for e in _cl_c.events if "WRITE REFUSED" in e])
# ⚠️ AND SAFETY-18 IS GRADED ON WHAT IT WAS HANDED, WHICH IS AN ORDERING
# CLAIM. `check_meter_frozen()` takes `self.target` - the arbiter's RESOLVED
# INTENT - and `_shadow_settle()` re-seeds that to 0 W at the very end of the
# cycle, AFTER it. Re-seed one line earlier and the freeze detector grades a
# 0 W that no decision produced: its conjunction needs a command at the rail,
# so it goes permanently blind and the whole suite stays green. The detector's
# own `evaluate` is wrapped here because the arguments are the only place that
# ordering is observable.
_fz_seen = []
_fz_real = _shad_c.freeze_wd.evaluate
def _fz_spy(unchanged, target, *a, **k):
_fz_seen.append(target)
return _fz_real(unchanged, target, *a, **k)
_shad_c.freeze_wd.evaluate = _fz_spy
async def _one_more():
await _shad_c.read_inputs()
await _shad_c.cycle()
asyncio.run(_one_more())
_shad_c.freeze_wd.evaluate = _fz_real
print(f" SAFETY-18 was handed: {_fz_seen}")
check("PRECONDITION: the freeze detector was consulted at all on that cycle",
len(_fz_seen) == 1)
check("SAFETY-18 still OBSERVES the DECISION: it is handed the arbiter's "
"resolved intent, not the 0 W the wire gate substituted. Re-seeding one "
"line earlier would hand it a command no decision produced and its "
"conjunction - a command at the rail against a meter that has stopped "
"moving - could never be satisfied again",
_fz_seen and _fz_seen[0] == _shad_c.shadow_would_w
and _fz_seen[0] != shadow.WIRE_W)
# --- 8f. THE MODE TABLE LIVES IN test_validate.py ---------------------------
# `shadow.engaged()` is a pure function of the option and the start button, and
# the six states it answers for are graded beside the rule that validates the
# option - not here, where a defeat of the DEFAULT stops part 3's wired run
# reaching the wire at all and takes this whole file down before part 8 runs.
# ⚠️ THAT IS MEASURED, NOT ANTICIPATED: the mutation that makes `auto` resolve
# to shadow unconditionally aborts part 3 with an IndexError, so its red count
# here is a LOWER BOUND and none of part 8 is graded (MUTATION.md, FCST-06).
# --- 8g. THE OPERATOR STOP, WHICH IS THE ONE FAILSAFE THIS CARD CHANGES ------
_stop_live_h, _stop_live_c = _s10("off", auto=False, cycles=6)
_stop_shad_h, _stop_shad_c = _s10("auto", auto=False, cycles=6)
print(f" stopped, shadow off: {_stop_live_c.reason!r}")
print(f" stopped, shadow on : {_stop_shad_c.reason!r}")
check("PRECONDITION: with shadow mode off, a stopped controller files the "
"P_SAFETY stop claim and the arbiter resolves to THAT source, at 0 W",
_stop_live_c.reason.startswith("safety")
and _stop_live_c.target == 0.0)
check("in shadow mode that claim is NOT filed, so the arbiter resolves to the "
"REAL winner and the recorded decision is the real one. A P_SAFETY claim "
"out-ranks everything - left in, every shadow run would answer the wrong "
f"question ({_stop_shad_c.reason!r})",
_stop_shad_c.reason.startswith("loop")
and _stop_shad_c.shadow_would_w is not None
and abs(_stop_shad_c.shadow_would_w) > 0.0)
check("...and the actuator ends at exactly the same place either way, because "
"the substitution that replaced the claim is STRICTLY STRONGER: a claim "
"can be out-ranked and a wire gate cannot",
set(_stop_shad_h.writes) == {0.0} == set(_stop_live_h.writes))
# --- 8h. THE RENDERED SURFACE (TARIFF-11's pattern) -------------------------
# Graded on `status_payload()`, which IS the status page: web.py's `PAGE` is a
# static shell whose banner and rows are filled from this JSON. Named slots, not
# substring tests, so a transposition moves a slot.
_ROW = "Shadow mode"
_shad_pay = _shad_c.status_payload()
_live_pay = _live_c.status_payload()
_shad_rows = dict(_shad_pay["rows"])
_live_rows = dict(_live_pay["rows"])
print(f" banner: {_shad_pay['banner']!r}")
print(f" row : {_shad_rows.get(_ROW)!r}")
check("the status page has a Shadow mode row in BOTH states, so its ABSENCE is "
"never how an operator learns which one they are in",
_ROW in _shad_rows and _ROW in _live_rows)
check("AC 3: the banner states plainly that the battery is not being "
"controlled",
"SHADOW MODE" in _shad_pay["banner"]
and "NOT being controlled" in _shad_pay["banner"])
check("...and it is not green. A page that reads `ok` while nothing is being "
"commanded is the 'looks healthy but is inert' failure itself",
_shad_pay["level"] == "warn")
check("...and it is FIRST in the cascade: the same controller has other "
"conditions to report and none of them displaces this sentence",
_live_pay["banner"] != _shad_pay["banner"]
and "SHADOW MODE" not in _live_pay["banner"])
_slot = re.search(r"WOULD HAVE COMMANDED ([-+][\d ]+) W, inverter held at "
r"([-\d]+) W", _shad_rows[_ROW])
check("the row answers 'what would it have done' with a NUMBER in its own "
"named slot, and says where the inverter actually is in a second one - "
"two slots in one sentence, so a transposition moves them",
_slot is not None)
if _slot is not None:
_would = float(_slot.group(1).replace(" ", ""))
check("...and that number is the decision the path reached, not the 0 W "
f"standing on the wire ({_would:.0f} vs {_shad_c.target:.0f})",
_would == round(float(_shad_c.shadow_would_w))
and _would != _shad_c.target)
check("...and the second slot is the STANDING command, so the two cannot "
"be read as one number",
float(_slot.group(2)) == 0.0)
check("the row answers 'and why' by carrying the arbiter's own sentence, "
"rather than a re-derivation of it",
_shad_c.reason in _shad_rows[_ROW])
check("with shadow mode off the row says so, and says what follows from it - "
"which is the arrangement that makes the one above a claim about this "
"controller and not about the string",
"NOT in shadow mode" in _live_rows[_ROW]
and "DOES reach the inverter" in _live_rows[_ROW]
and "WOULD HAVE COMMANDED" not in _live_rows[_ROW])
# ⚠️ A STANDING LATCH IS NAMED RATHER THAN HIDDEN. The shadow banner sits above
# the three `bad` branches, so it has to carry what they would have said - a
# latch that is up while the controller is inert is up the moment it stops
# being inert.
_shad_c.inverters[0].sign_wd.tripped = True
_latched = _shad_c.status_payload()
print(f" banner with a latch up: {_latched['banner']!r}")
check("a failsafe latch that is standing is NAMED in the shadow banner and "
"turns it red, so nobody turns shadow mode off into a battery that will "
"not move and a page that never mentioned why",
_latched["level"] == "bad"
and "LATCH IS STILL UP" in _latched["banner"]
and _shad_c.inverters[0].name in _latched["banner"])
check("...and it still says the battery is not being controlled, so the latch "
"has not displaced the sentence that may never be displaced",
"NOT being controlled" in _latched["banner"])
_shad_c.inverters[0].sign_wd.tripped = False
# --- 8i. THE RECORD A LATER ANALYSIS READS ----------------------------------
# A shadow run is only worth running if it answers "what would it have done,
# and why" AFTERWARDS, from the 40-row action log, and not only on a page
# somebody happened to be looking at.
_shadow_lines = [e for e in _shad_c.events if "SHADOW" in e]
print(f" action log, shadow lines: {len(_shadow_lines)} of "
f"{len(_shad_c.events)}")
check("the action log records that shadow mode came on, in words",
any("SHADOW MODE ON" in e for e in _shadow_lines))
check("...and what it would have commanded, with the arbiter's reason beside "
"it",
any("would have commanded" in e and "loop" in e for e in _shadow_lines))
check("...EDGE-logged, not level-logged: 25 cycles did not flush the 40-row "
"log that has to carry the context of what the run found",
len(_shadow_lines) < 25 and len(_shad_c.events) <= 40)
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