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

SAFETY-24: an energy floor under the peak guard's forced discharge
checks / unit (pull_request) Successful in 3m29s
Details

Browse Source 
SAFETY-07 can command a discharge to hold a billed quarter down, and
nothing below it consulted the state of charge. On the 2026-08-20 replay
it asks for 878 Wh in a day. Relying on the BMS is not absurd - it is a
real floor and it works - but it is the pack's protection rather than the
operator's, and a month of aggressive quarters leaves the battery
repeatedly at that cut-out with no EMS-side record of why. The first
symptom is a pack never available when wanted, which is indistinguishable
from the SoC drift MAINT-01 exists to correct.

decide() now takes soc and soc_floor, both required and neither
defaulted, and discharge_permitted() is the one place the question is
answered. Above the floor nothing changes. At or below it the guard files
a 0 W floor instead of a sized one - one-directional, so charging stays
blocked and no discharge is capped - and reports floor_limited with the
reason it bound. It is not a trip: every stop in 11.2's table still
outranks the claim, and the floor is never negative, so 9.1's asymmetry
survives as a property of the object (0 negative floors over 1296 swept
combinations, SoC included).

The SoC is graded only after `wanted > 0` is established, so it is never
evaluated where it governs nothing. An unreadable SoC is floor-limited,
not a full battery - a full battery is exactly the value that would
disarm the floor, and this project shipped that defaulting once already.

Failing to shave is not silent: one notification per episode
(goodwe_peak_soc_floor), distinct from the saturation alarm because the
two ask for opposite operator actions, plus its own status row and an
action-log line.

Verified by replay. The 2026-08-20 capture is driven against a modelled
pack sized from the day's own measured appetite, so it genuinely
exhausts: without the floor it reaches the BMS cut-out and holds there
for 1227 guard cycles; with it the pack stops at 20.0 % and never touches
the BMS. The energy withheld equals the energy between the two floors to
within one cycle at the rail. The honest half is reported too - the worst
quarter goes 4338 W -> 5030 W, because a floor-limited guard shaves less
on purpose.

Two new options, mirrored once in peakguard.PEAK_OPTION_DEFAULTS
(TARIFF-02's pattern), imported by validate.py and main.py rather than
typed as literals and pinned to config.yaml by a check that reads the
file - so CFG-03 still tracks three places. validate.py rule 8 refuses
peak_soc_floor <= bms_soc_floor at startup, in CFG-01/MAINT-04's manner.
maintenance_soc_floor is deliberately not reused and not renamed: it is a
drain target, not a spend limit, and the rename is recorded as a decision
to revisit if a third floor appears.

Eleven mutations of the shipped source were run against the checks; all
eleven go red and there are no survivors.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015VXjA3LWccN1rJsk3okyPC
This commit is contained in:
glenn schrooyen
2026-08-31 00:29:36 +02:00
co-authored by Claude Opus 5
parent e7c95bfd8a
commit e709554857
7 changed files with 1177 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
goodwe_controller/test_peakguard.py
+673 -41
View File
@@ -2,7 +2,7 @@
python3 test_peakguard.py
Eight parts. What each is for, and - where it matters - what it is deliberately
Ten parts. What each is for, and - where it matters - what it is deliberately
NOT allowed to be tested at:
1. THE CONTROL LAW, `excess_w`, graded AWAY from the two points where it is
@@ -36,8 +36,18 @@ NOT allowed to be tested at:
6. THE REPLAY. 2026-08-20's 5029.8 W billed quarter, closed-loop, and the
healthy day's baseline. Every figure re-derived from the committed capture.
7. THE WIRING, through the real `Controller.cycle()` and its real status page.
8. MUTATION. Six one-line defeats of this guard, each run against the checks
above to show they are not vacuous.
8. SAFETY-24: THE ENERGY FLOOR under the forced discharge, graded only where
a forced discharge is actually being asked for - the SoC is a degenerate
stimulus on every cycle where `wanted <= 0`, and grading it there is how
six bugs hid in this repository this week. Includes the sweep that shows
the floor is never negative, which is the property the whole `lo`-limit
shape rests on.
9. THE EXHAUSTING REPLAY (AC 4) AND THE STARTUP REFUSAL. The same 2026-08-20
capture, against a MODELLED PACK small enough that the guard's own 878 Wh
day would take it to the BMS cut-out - so the two worlds, before and after
this card, are run against each other rather than reasoned about.
10. MUTATION. Eleven one-line defeats of these two guards, each run against
the checks above to show they are not vacuous.
------------------------------------------------------------------------------
WHAT THE REPLAY IS AND IS NOT
@@ -66,7 +76,9 @@ support: the same quarters, with and without the guard.
import asyncio
import ast
import dataclasses
import gzip
import inspect
import json
import re
import shutil
@@ -89,9 +101,11 @@ from app.p1 import (
)
from app.peakguard import (
BLIND_COVERAGE, BLIND_FROZEN, BLIND_NO_BLOCK, CEILING_CAP,
CEILING_MONTH_PEAK, PeakDecision, blindness, decide, excess_w,
quarter_ceiling_w, uncovered_s,
CEILING_MONTH_PEAK, FLOOR_AT_SOC_FLOOR, FLOOR_SOC_UNREADABLE,
PEAK_OPTION_DEFAULTS, PeakDecision, blindness, decide, discharge_permitted,
excess_w, quarter_ceiling_w, uncovered_s,
)
from app.validate import validate_startup_config
from app.tariff import (
NOT_OBSERVED, NO_RECORD, OBSERVED, PeakReading, PeakStore, PeakTracker,
month_key,
@@ -166,13 +180,29 @@ CYCLE_S = read_option("cycle_floor_s")
# ceiling and the GW5000S-BP's rating, which is why the two agree - so the
# device figure is taken from the schema rather than typed.
LIVE_MAX_W = read_schema_max("max_w")
# SAFETY-24's two options, out of the shipped file for the same reason.
SOC_FLOOR = read_option("peak_soc_floor")
BMS_FLOOR = read_option("bms_soc_floor")
MAINT_FLOOR = read_option("maintenance_soc_floor")
# A state of charge that comfortably permits a forced discharge, DERIVED from
# the floor rather than typed: every part of this file that is not about
# SAFETY-24 has to grade SAFETY-07's behaviour at a SoC where the new floor is
# not the thing deciding the answer, or it would be measuring the wrong guard.
SOC_OK = SOC_FLOOR + 30.0
print("SAFETY-07 - the quarter-peak projection guard")
print("SAFETY-24 - and the energy floor under its forced discharge")
print(f" read from config.yaml: peak_cap_w {CAP_W:.0f} W, "
f"meter_max_age_s {MAX_HOLD_S:.0f} s, meter_freeze_s {FREEZE_S:.0f} s, "
f"cycle_floor_s {CYCLE_S:.0f} s, max_w schema ceiling {LIVE_MAX_W:.0f} W")
print(f" ...and peak_soc_floor {SOC_FLOOR:.0f} %, bms_soc_floor "
f"{BMS_FLOOR:.0f} %, maintenance_soc_floor {MAINT_FLOOR:.0f} %")
check("every threshold this check grades against came out of config.yaml",
None not in (CAP_W, MAX_HOLD_S, FREEZE_S, CYCLE_S, LIVE_MAX_W))
None not in (CAP_W, MAX_HOLD_S, FREEZE_S, CYCLE_S, LIVE_MAX_W,
SOC_FLOOR, BMS_FLOOR, MAINT_FLOOR))
check("...and the SoC these checks use as 'plenty of battery' is comfortably "
"clear of the floor, so nothing outside part 9 is silently grading "
"SAFETY-24 instead of SAFETY-07", SOC_OK > SOC_FLOOR + 10.0)
B = datetime(2026, 8, 30, 9, 0, 0, tzinfo=timezone.utc) # a quarter boundary
@@ -239,7 +269,7 @@ BATT_NOW = 400.0
a_edge = averager_at(CEIL, 450.0)
p_edge = PeakTracker.project(a_edge, B + timedelta(seconds=450))
d_edge = decide(p_edge, batt_w=BATT_NOW, cap_w=CAP_W, reading=None,
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S)
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR)
e_edge = excess_w(p_edge, CEIL)
print(f" held exactly AT the ceiling for 450 s: projected "
f"{p_edge.projected_w:.2f} W, excess {e_edge:.3f} W, "
@@ -254,7 +284,7 @@ check("at the engagement boundary the excess is zero, so the floor is exactly "
# released as soon as `excess <= 0` hands the actuator back at the exact instant
# it is working, and the load springs straight back.
released = decide(p_edge, batt_w=BATT_NOW, cap_w=CAP_W, reading=None,
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S)
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR)
would_release_on_excess = excess_w(p_edge, CEIL) <= 0.0
print(f" at the boundary: excess {e_edge:.3f} W (an `excess > 0` test would "
f"RELEASE: {would_release_on_excess}), floor {released.floor_w:.0f} W "
@@ -266,7 +296,7 @@ check("...and the claim is held there, not released. `excess` falls to zero as "
d_gone = decide(PeakTracker.project(averager_at(500.0, 450.0),
B + timedelta(seconds=450)),
batt_w=BATT_NOW, cap_w=CAP_W, reading=None,
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S)
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR)
check("...and it DOES release once the load falls under what the quarter can "
"afford, so 'threshold the floor' is not 'never let go'",
not d_gone.engaged and d_gone.floor_w is None)
@@ -275,7 +305,7 @@ check("...and it DOES release once the load falls under what the quarter can "
a_over = averager_at(CEIL + 200.0, 450.0)
p_over = PeakTracker.project(a_over, B + timedelta(seconds=450))
d_over = decide(p_over, batt_w=BATT_NOW, cap_w=CAP_W, reading=None,
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S)
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR)
print(f" 200 W over for 450 s: {d_over.explain()}")
check("a quarter projected over the ceiling engages, and the floor is the "
"MEASURED battery power plus the derived excess",
@@ -284,7 +314,7 @@ check("a quarter projected over the ceiling engages, and the floor is the "
check("...and the floor moves with the battery measurement, so it is a claim "
"about delivered power rather than a fixed setpoint",
not near(decide(p_over, batt_w=BATT_NOW + 250.0, cap_w=CAP_W, reading=None,
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S).floor_w,
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR).floor_w,
d_over.floor_w, 1.0))
# THE WIND-UP ARGUMENT, made concrete. A floor built on INTENT re-adds its own
@@ -295,7 +325,7 @@ delivered_floor = []
intent_floor = []
for _ in range(5):
d_del = decide(p_over, batt_w=stuck_batt, cap_w=CAP_W, reading=None,
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S)
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR)
delivered_floor.append(d_del.floor_w)
intent = intent + excess_w(p_over, CEIL) # what the intent form would do
intent_floor.append(intent)
@@ -378,7 +408,7 @@ absurd = averager_at(LIVE_MAX_W * 4, 870.0)
p_absurd = PeakTracker.project(absurd, B + timedelta(seconds=870))
DEV_MAX = LIVE_MAX_W - 500.0 # a device rating tighter than max_w
d_absurd = decide(p_absurd, batt_w=1000.0, cap_w=CAP_W, reading=None,
achievable_w=min(LIVE_MAX_W, DEV_MAX), max_hold_s=MAX_HOLD_S)
achievable_w=min(LIVE_MAX_W, DEV_MAX), max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR)
r6 = resolve([Claim.limit("device", P_SAFETY, -DEV_MAX, DEV_MAX, "rating"),
Claim.limit("supervised", P_SAFETY, -LIVE_MAX_W, LIVE_MAX_W, "max_w"),
Claim.set("loop", P_LOOP, -200.0, "grid"),
@@ -461,7 +491,7 @@ print(f" a real PeakStore on a missing volume: state={rd_broken.state}, "
check("the real store answers NOT_OBSERVED rather than a peak of 0 W",
rd_broken.state == NOT_OBSERVED and rd_broken.peak_w is None)
d_broken = decide(p_over, batt_w=BATT_NOW, cap_w=CAP_W, reading=rd_broken,
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S)
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR)
print(f" {d_broken.explain()}")
check("AND THE GUARD STILL ENGAGES ON IT. An unreadable peak must not disarm "
"the cap - 'no peak recorded' is precisely the reading that would, and it "
@@ -489,7 +519,7 @@ print("4. blindness: three ways a projection is not evidence")
fresh = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
p_fresh = PeakTracker.project(fresh, B)
d_fresh = decide(p_fresh, batt_w=0.0, cap_w=CAP_W, reading=None,
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S)
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR)
print(f" fresh averager: projected {p_fresh.projected_w:.0f} W, "
f"block_start={p_fresh.block_start}; {d_fresh.explain()}")
check("a fresh averager's projection is 0 W, i.e. comfortably under any "
@@ -517,7 +547,7 @@ RAILED = LIVE_MAX_W # the loop has wound to the rail
is_frozen = looks_frozen(FREEZE_S + 1.0, -RAILED, LIVE_MAX_W,
FreezeWatchdogConfig(hold_s=FREEZE_S))
d_frozen = decide(p_frozen, batt_w=-RAILED, cap_w=CAP_W, reading=None,
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S,
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR,
frozen=is_frozen)
print(f" meter stuck at -4000 W with the command at {-RAILED:+.0f} W: "
f"projected {p_frozen.projected_w:.0f} W, coverage "
@@ -543,7 +573,7 @@ check("...and the same command NOT frozen is left alone, so the blind branch is
peakguard.claim(decide(
p_frozen, batt_w=-RAILED, cap_w=CAP_W,
reading=None, achievable_w=LIVE_MAX_W,
max_hold_s=MAX_HOLD_S, frozen=False))
max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR, frozen=False))
or Claim.limit("noop", P_PEAK)]).target_w,
-RAILED, 1e-9))
@@ -554,7 +584,7 @@ p_gap = PeakTracker.project(gapped, B + timedelta(seconds=600))
clean = averager_at(4200.0, 600.0)
p_clean = PeakTracker.project(clean, B + timedelta(seconds=600))
d_gap = decide(p_gap, batt_w=0.0, cap_w=CAP_W, reading=None,
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S)
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR)
print(f" a {GAP:.0f} s hole at 4200 W: uncovered {uncovered_s(p_gap):.0f} s, "
f"projected {p_gap.projected_w:.0f} W vs {p_clean.projected_w:.0f} W with "
f"no hole")
@@ -618,7 +648,7 @@ OLD = (p_held.partial_ws if hasattr(p_held, "partial_ws")
else HELD * 300.0) # acc, which the old form would still have had
old_projected = (HELD * 300.0 + 0.0 * (QUARTER_S - 300.0)) / QUARTER_S
d_true = decide(p_held, batt_w=0.0, cap_w=CAP_W, reading=None,
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S)
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR)
print(f" same fixture: shipped projection {p_held.projected_w:.0f} W "
f"(engaged={d_true.engaged}), the defaulted form's "
f"{old_projected:.0f} W (would engage={old_projected > CAP_W})")
@@ -636,24 +666,55 @@ REPLAY = sorted((HERE / "scenarios").glob("p1-net-*.jsonl.gz"))
check("the committed replay fixtures are present", len(REPLAY) >= 2)
def replay(path, *, guarded: bool, cap_w=CAP_W, max_w=LIVE_MAX_W):
def replay(path, *, guarded: bool, cap_w=CAP_W, max_w=LIVE_MAX_W, pack=None,
soc_floor=None):
"""Drive one capture through the real ingest, tracker and guard.
See the module docstring for the model and why its baseline is pessimistic.
⚠️ SAFETY-24's `pack` IS A SECOND MODEL, AND IT IS STATED HERE RATHER THAN
HIDDEN. The captures carry meter power only, so the pack is simulated:
pack = (usable_wh, start_soc_pct, bms_floor_pct)
* SoC falls by the DELIVERED discharge, integrated over the real telegram
timestamps - the same integral that already produces `wh`, so the two
cannot disagree;
* the BMS is modelled as a HARD CUT at `bms_floor_pct`: at or below it the
modelled battery delivers 0 W whatever the guard asks for. Without that,
the "before this card" run would be unphysical and its comparison
worthless - the BMS really does stop the pack, and the finding this card
answers is that it is then the ONLY thing that does;
* nothing recharges the pack. One day, one direction. That is pessimistic
in the same deliberate way the zero-baseline discharge is: it isolates
what the peak guard alone spends.
`pack=None` pins the SoC at SOC_OK and reproduces SAFETY-07's numbers
exactly, which is what parts 6's existing checks grade.
"""
ing = P1Ingest(phases=1, max_age_s=MAX_HOLD_S)
store = fresh_store(f"{path.stem}-{'g' if guarded else 'u'}")
store = fresh_store(f"{path.stem}-{'g' if guarded else 'u'}"
f"{'' if pack is None else f'-p{soc_floor:.0f}'}")
tracker = PeakTracker(store, cap_w=cap_w)
batt = 0.0 # delivered discharge, W (positive)
next_cycle = None
graded_engaged = blind_engaged = saturated = cycles = 0
limited = at_bms = neg_floor = 0
wh = 0.0
prev_t = None
usable_wh, soc, bms = (None, SOC_OK, None) if pack is None else (
float(pack[0]), float(pack[1]), float(pack[2]))
floor_pct = SOC_FLOOR if soc_floor is None else float(soc_floor)
min_soc = soc
for line in gzip.open(path, "rt", encoding="utf-8"):
row = json.loads(line)
t = datetime.fromisoformat(row["t"])
if prev_t is not None:
wh += batt * (t - prev_t).total_seconds() / 3600.0
dwh = batt * (t - prev_t).total_seconds() / 3600.0
wh += dwh
if usable_wh:
soc = max(0.0, soc - 100.0 * dwh / usable_wh)
min_soc = min(min_soc, soc)
prev_t = t
ing.submit(sample(t, float(row["w"]) - batt))
tracker.sync(ing)
@@ -663,7 +724,10 @@ def replay(path, *, guarded: bool, cap_w=CAP_W, max_w=LIVE_MAX_W):
cycles += 1
d = decide(PeakTracker.project(ing.averager, t),
batt_w=batt, cap_w=cap_w, reading=tracker.peak(t),
achievable_w=max_w, max_hold_s=MAX_HOLD_S, frozen=False)
achievable_w=max_w, max_hold_s=MAX_HOLD_S, soc=soc,
soc_floor=floor_pct, frozen=False)
if d.floor_w is not None and d.floor_w < 0.0:
neg_floor += 1
if d.engaged:
(globals(), None)
if d.blind:
@@ -671,15 +735,25 @@ def replay(path, *, guarded: bool, cap_w=CAP_W, max_w=LIVE_MAX_W):
else:
graded_engaged += 1
saturated += 1 if d.saturated else 0
limited += 1 if d.floor_limited else 0
if usable_wh and soc <= bms:
at_bms += 1
if not guarded:
continue
batt = max(0.0, d.floor_w) if (d.engaged and d.floor_w is not None) else 0.0
# THE BMS, MODELLED. It is a real floor and it works - that is the whole
# point of stating the risk fairly - so the "before" world is not a pack
# discharged to nothing, it is a pack pinned at the BMS limit.
if usable_wh and soc <= bms:
batt = 0.0
month = month_key(datetime.fromisoformat(json.loads(
gzip.open(path, "rt", encoding="utf-8").readline())["t"]))
return {
"blocks": dict(store.blocks_in(month)), "cycles": cycles,
"graded": graded_engaged, "blind": blind_engaged,
"saturated": saturated, "wh": wh, "month": month,
"limited": limited, "at_bms": at_bms, "neg_floor": neg_floor,
"end_soc": soc, "min_soc": min_soc,
}
@@ -872,7 +946,7 @@ check("...and the status page says NOT COVERED rather than 'idle' or a green "
# Controller rather than left to flake. It is bounded at three attempts: a fourth
# would mean something is wrong that is not a boundary.
def wired_over_limit(db, want_floor_w, opts=None, store=None, batt_w=0.0,
standing_w=None):
standing_w=None, soc=55.0):
"""A real Controller driven one real cycle over a real over-limit quarter.
⚠️ THE LOAD IS DERIVED FROM WHERE IN THE QUARTER THE WALL CLOCK LANDS, and
@@ -886,7 +960,7 @@ def wired_over_limit(db, want_floor_w, opts=None, store=None, batt_w=0.0,
for attempt in range(4):
ctl, hs = built(db if attempt == 0 else Path(f"{db}.{attempt}"), opts,
store=_Store(store) if store is not None else None)
ctl.auto, ctl.soc, ctl.batt = True, 55.0, batt_w
ctl.auto, ctl.soc, ctl.batt = True, soc, batt_w
ctl.dev_min, ctl.dev_max = -LIVE_MAX_W, LIVE_MAX_W
if standing_w is not None:
# The command STANDING ON THE WIRE, which is what B3 below needs to
@@ -911,6 +985,9 @@ def wired_over_limit(db, want_floor_w, opts=None, store=None, batt_w=0.0,
ctl.cycle())
g = ctl.peak_guard
if g is not None and g.engaged and not g.blind:
# ⚠️ A FLOOR-LIMITED DECISION IS A LANDING, NOT A MISS. It is
# `engaged` and not `blind`, which is what this loop is waiting for;
# SAFETY-24's own fixtures below arrive here on purpose.
return ctl, hs, net_w, remaining
# Only a quarter boundary crossed between the last telegram and
# `cycle()`'s own clock gets here. Bounded at four: a fifth would mean
@@ -1094,7 +1171,8 @@ check("B3: main.py sizes the floor from the MEASURED battery power, not from "
# judged at a single point where they need not differ.
_proj3 = PeakTracker.project(b3.p1.averager, datetime.now(timezone.utc))
_kw = dict(cap_w=CAP_W, reading=b3.peak.peak(), achievable_w=LIVE_MAX_W,
max_hold_s=b3.p1.averager.max_hold_s)
max_hold_s=b3.p1.averager.max_hold_s, soc=SOC_OK,
soc_floor=SOC_FLOOR)
_intent = _shipped_decide(_proj3, batt_w=STANDING, **_kw)
check("one cycle of the intent form is exactly the delivered form plus the "
"command standing on the wire - which is the whole defect in one line: "
@@ -1131,10 +1209,41 @@ _call = next(n for n in ast.walk(ast.parse(_mainsrc))
if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute)
and n.func.attr == "decide")
_batt_src = ast.unparse(next(k.value for k in _call.keywords if k.arg == "batt_w"))
print(f" main.py's decide(batt_w=...) is `{_batt_src}`")
_soc_src = ast.unparse(next(k.value for k in _call.keywords if k.arg == "soc"))
print(f" main.py's decide(batt_w=...) is `{_batt_src}`, decide(soc=...) is "
f"`{_soc_src}`")
check("...and the call site says so in the source: the argument is the measured "
"battery attribute, not the standing command and not the resolved target",
_batt_src == "self.batt")
# ⚠️ SAFETY-24, AND ON THE SAME SINGLE CALL SITE. This grader takes the FIRST
# `.decide(` in main.py and reads a NAME rather than a value, so a second call
# site - or a decoy - would defeat it; this card deliberately adds neither, and
# extends the existing grader instead of writing a second one.
check("...and `soc` is handed over as the MEASURED reading, bare. Not "
"`self.soc or 100`, not a `float(..., 100)`, not the maintenance floor: "
"a full battery is exactly the value that disarms SAFETY-24's floor, and "
"NOTES.md 4.4b records this project shipping that exact defaulting once",
_soc_src == "self.soc")
check("...and main.py still has exactly ONE `.decide(` call site, so grading "
"the first one grades the shipped one",
sum(1 for n in ast.walk(ast.parse(_mainsrc))
if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute)
and n.func.attr == "decide") == 1)
# ⚠️ AND WHICH KEY THE FLOOR COMES FROM, GRADED ON THE SOURCE. A mutation that
# swapped `peak_soc_floor` for `maintenance_soc_floor` SURVIVED every
# behavioural check in this file, because the two defaults are close enough that
# no fixture separated them - which is exactly the coupling the card forbids and
# exactly the kind of survivor a value-based check cannot see. This grades the
# NAME, and part 9 below now also separates them behaviourally.
_floor_src = ast.unparse(next(k.value for k in _call.keywords
if k.arg == "soc_floor")).replace("'", '"')
print(f" main.py's decide(soc_floor=...) is `{_floor_src}`")
check("...and the floor comes from `peak_soc_floor`, defaulted from the ONE "
"mirror - not from `maintenance_soc_floor` (a drain target, not a spend "
"limit) and not from a literal, which would be CFG-03's fourth site",
'"peak_soc_floor"' in _floor_src
and "maintenance_soc_floor" not in _floor_src
and 'PEAK_OPTION_DEFAULTS["peak_soc_floor"]' in _floor_src)
# SAFETY-24 must be pointed at from the module that needs it, not only from a
# review thread - a ticket nobody can find from the code is a ticket nobody
@@ -1147,10 +1256,20 @@ check("peakguard.py names SAFETY-24, the missing SoC backstop on the forced "
# CFG-03: no fourth defaults site.
cfg = (HERE / "config.yaml").read_text(encoding="utf-8")
val = (HERE / "app" / "validate.py").read_text(encoding="utf-8")
check("no new option was added: config.yaml names no peak-guard key, and "
"§14's kW/W trap `capacity_peak_limit_kw` stays declined",
"peak_guard" not in cfg and "capacity_peak_limit_kw" not in cfg
and "peak_shave" not in cfg)
# ⚠️ REWRITTEN BY SAFETY-24, BECAUSE IT IS NO LONGER TRUE AS WRITTEN. SAFETY-07
# added no option and this check said so. This card adds TWO and says which -
# leaving the old sentence in place while the file it reads has grown two keys
# would be a green line asserting the opposite of the diff. What is still
# refused is a THIRD: §14's kW/W trap stays declined, and no key beyond the two
# named ones exists.
_peak_keys = sorted(k for k in ("peak_guard", "peak_shave",
"capacity_peak_limit_kw")
if k in cfg)
check("SAFETY-24 adds exactly the two keys it names and no more: §14's kW/W "
"trap `capacity_peak_limit_kw` stays declined, and no speculative "
"`peak_guard`/`peak_shave` block appeared beside it",
_peak_keys == []
and sorted(PEAK_OPTION_DEFAULTS) == ["bms_soc_floor", "peak_soc_floor"])
# Graded on the parsed call syntax, not on the text: the module docstring names
# `peak_cap_w` while arguing that it does not read it, so a grep would be
# satisfied by deleting the explanation and would fail on writing one.
@@ -1161,11 +1280,14 @@ _getters = [n for n in ast.walk(ast.parse(src))
if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute)
and n.func.attr == "get"]
print(f" peakguard.py makes {len(_getters)} options-style `.get()` calls")
check("peakguard.py reads no option of its own: it makes no `.get()` call at "
"all and takes `cap_w` as an argument. CFG-03's three defaults sites are "
"config.yaml's `options:`, config.yaml's `schema:` and main.py's literal "
"fallbacks - this card adds no fourth",
not _getters and "cap_w" in _DECIDE_ARGS)
check("peakguard.py still READS no option: it makes no `.get()` call at all "
"and takes `cap_w`, `soc` and `soc_floor` as arguments. CFG-03's three "
"defaults sites are config.yaml's `options:`, config.yaml's `schema:` "
"and main.py's fallbacks; PEAK_OPTION_DEFAULTS is a MIRROR of the first, "
"TARIFF-02-style, not a fourth site - it is what main.py and validate.py "
"read INSTEAD of typing literals",
not _getters and "cap_w" in _DECIDE_ARGS and "soc" in _DECIDE_ARGS
and "soc_floor" in _DECIDE_ARGS)
# And the fallback in main.py must equal the shipped default, for the reason
# test_validate.py's CFG-03 block spells out at length: the two diverging is a
# live defect in this repo for `maintenance_charge_w`, and it is not to be
@@ -1182,7 +1304,456 @@ check("...and every literal fallback for peak_cap_w in main.py equals the "
# ==============================================================================
print()
print("8. mutation: what defeats this guard, and which check bites")
print("8. SAFETY-24: the energy floor under the forced discharge")
# ⚠️ THE STIMULUS FIRST, AND IT IS NOT DEGENERATE. Everything below is graded on
# a quarter that GENUINELY WANTS A FORCED DISCHARGE - `wanted > 0` - because the
# state of charge governs nothing on a cycle where the guard was never going to
# discharge. A check placed on a quiet quarter would pass with the gate removed,
# with the gate inverted, and with the gate reading the wrong sensor.
a_hungry = averager_at(CEIL + 900.0, 600.0)
p_hungry = PeakTracker.project(a_hungry, B + timedelta(seconds=600))
WANTED = BATT_NOW + excess_w(p_hungry, CEIL)
print(f" the stimulus: a quarter projecting {p_hungry.projected_w:.0f} W "
f"against a {CEIL:.0f} W ceiling with "
f"{QUARTER_S - p_hungry.elapsed_s:.0f} s left; a forced discharge of "
f"{WANTED:+.0f} W is what would hold it")
check("the stimulus really does want a forced discharge, so the SoC is a "
"quantity that decides something here", WANTED > 0.0)
# --- the predicate, which is the ONE place the question is answered ----------
PERMITS = discharge_permitted(SOC_OK, SOC_FLOOR)
AT = discharge_permitted(SOC_FLOOR, SOC_FLOOR)
UNDER = discharge_permitted(SOC_FLOOR - 1.0, SOC_FLOOR)
UNREAD = discharge_permitted(None, SOC_FLOOR)
NOFLOOR = discharge_permitted(SOC_OK, None)
print(f" discharge_permitted: {SOC_OK:.0f} % -> {PERMITS!r}")
print(f" {SOC_FLOOR:.0f} % (exactly at) -> {AT!r}")
print(f" None (unreadable) -> {UNREAD!r}")
check("a pack above the floor may be discharged", PERMITS is None)
check("...a pack AT the floor may not - `<=`, not `<`. Permitting one more "
"cycle at the floor would make the floor mean 'one cycle below the "
"floor', which is the whole quantity this option names",
AT == FLOOR_AT_SOC_FLOOR and UNDER == FLOOR_AT_SOC_FLOOR)
check("...an UNREADABLE SoC is refused, and refused with its OWN reason. A "
"failed read is not an observation, and a full battery is exactly the "
"value that would disarm this floor",
UNREAD == FLOOR_SOC_UNREADABLE and UNREAD != AT)
check("...and no configured floor is not permission either", NOFLOOR is not None)
# THE DISARMING ALTERNATIVE, COMPUTED RATHER THAN DESCRIBED. `float(soc or 100)`
# is the shape NOTES.md 4.4b records the stage-5 loop actually shipping.
check("...and the defaulting form that was ACTUALLY shipped once in this "
"project - reading a missing SoC as a full battery - would have permitted "
"it, so this is a live direction and not a hypothetical",
discharge_permitted(float(None or 100), SOC_FLOOR) is None)
# --- the decision -----------------------------------------------------------
d_ok = decide(p_hungry, batt_w=BATT_NOW, cap_w=CAP_W, reading=None,
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=SOC_OK,
soc_floor=SOC_FLOOR)
d_lim = decide(p_hungry, batt_w=BATT_NOW, cap_w=CAP_W, reading=None,
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S,
soc=SOC_FLOOR, soc_floor=SOC_FLOOR)
d_unread = decide(p_hungry, batt_w=BATT_NOW, cap_w=CAP_W, reading=None,
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=None,
soc_floor=SOC_FLOOR)
print(f" with battery: {d_ok.explain()}")
print(f" at the floor: {d_lim.explain()}")
check("with battery to spare the guard sizes the forced discharge exactly as "
"SAFETY-07 always did - this card changes nothing above the floor",
d_ok.engaged and not d_ok.floor_limited and near(d_ok.floor_w, WANTED, 1e-6))
check("at the floor it STOPS, and the floor it files is 0 W - not the discharge "
"it wanted, and not a negative number",
d_lim.engaged and d_lim.floor_limited and near(d_lim.floor_w, 0.0))
check("...and it REPORTS being floor-limited rather than silently under-"
"shaving: the decision carries the flag, the reason it bound, the SoC it "
"read and the floor it read",
d_lim.floor_limited and d_lim.soc_floor_reason == FLOOR_AT_SOC_FLOOR
and near(d_lim.soc, SOC_FLOOR) and near(d_lim.soc_floor, SOC_FLOOR))
check("...and it still says what it WOULD have taken, so the under-shave is a "
"number rather than a silence",
near(d_lim.wanted_w, WANTED, 1e-6) and d_lim.wanted_w > d_lim.floor_w)
check("...and its one-line explanation tells the operator the consequence in "
"the words that matter - the quarter is going over the ceiling",
"FLOOR-LIMITED" in d_lim.explain()
and "WILL GO OVER THE CEILING" in d_lim.explain()
and f"{CEIL:.0f} W" in d_lim.explain())
check("an unreadable SoC lands in the same branch, with the OTHER reason - so "
"an operator reading one line can tell a flat battery from a dead sensor",
d_unread.floor_limited and near(d_unread.floor_w, 0.0)
and d_unread.soc_floor_reason == FLOOR_SOC_UNREADABLE
and d_unread.soc is None)
check("floor-limited and saturated are MUTUALLY EXCLUSIVE, by construction: the "
"branch returns before the achievable_w clamp that sets `saturated`, so "
"the two notifications can never both fire on one cycle",
not d_lim.saturated and not d_unread.saturated)
# --- and NOT graded where it is degenerate ----------------------------------
# The same empty pack on a quarter that is UNDER the ceiling. A gate placed one
# line earlier would report FLOOR-LIMITED here - on a quiet afternoon, over a
# guard that was never going to discharge - and an operator who is told the peak
# will be exceeded on a quarter that is nowhere near it stops reading the
# notification that matters.
d_quiet = decide(PeakTracker.project(averager_at(500.0, 450.0),
B + timedelta(seconds=450)),
batt_w=BATT_NOW, cap_w=CAP_W, reading=None,
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S,
soc=SOC_FLOOR - 5.0, soc_floor=SOC_FLOOR)
print(f" an EMPTY pack on a quiet quarter: {d_quiet.explain()}")
check("an empty pack on a quarter that does not need shaving is IDLE, not "
"floor-limited. The SoC is only graded where a forced discharge is "
"actually being asked for - grading it earlier would raise the alarm on "
"every quiet cycle of the day",
not d_quiet.engaged and not d_quiet.floor_limited
and d_quiet.floor_w is None)
# And the BLIND branch is untouched: it never forces a discharge, so there is
# nothing for an energy floor to bound there.
d_blind_low = decide(p_frozen, batt_w=-RAILED, cap_w=CAP_W, reading=None,
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S,
soc=SOC_FLOOR - 5.0, soc_floor=SOC_FLOOR, frozen=True)
check("...and an empty pack does not disturb the BLIND branch either: SAFETY-"
"18's cut is still a 0 W floor, still blocks charging, and is still not "
"reported as an energy problem",
d_blind_low.engaged and near(d_blind_low.floor_w, 0.0)
and d_blind_low.blind == BLIND_FROZEN and not d_blind_low.floor_limited)
# --- the claim, through the real arbiter ------------------------------------
lc = peakguard.claim(d_lim)
print(f" floor-limited claim: {lc.source} {lc.kind} lo={lc.lo:.0f} "
f"hi={lc.hi} reason={lc.reason!r}")
r_charge = resolve(SAFE + [Claim.set("loop", P_LOOP, -1500.0, "grid"), lc])
r_disch = resolve(SAFE + [Claim.set("loop", P_LOOP, 1800.0, "grid"), lc])
r_stop = resolve(SAFE + [Claim.set("safety", P_SAFETY, 0.0, "stopped"), lc])
print(f" loop wants to charge 1500 W: {r_charge.explain()}")
print(f" loop wants to discharge 1800 W: {r_disch.explain()}")
check("a floor-limited guard STILL blocks charging - 0 W is one-directional, "
"it can only ever lower delivered offtake, and adding import in the "
"quarter that sets the bill is the one thing it must not permit",
near(r_charge.target_w, 0.0) and not r_charge.contradiction)
check("...and it does NOT pull the loop off a discharge it is already doing. A "
"`lo` limit is structurally incapable of capping discharge, which is why "
"the action here is 0 W and never a negative floor",
near(r_disch.target_w, 1800.0, 1e-9))
check("...and every P_SAFETY stop still outranks it. This is an ENERGY bound, "
"not a trip: 11.2's table is unchanged by this card",
near(r_stop.target_w, 0.0) and not r_stop.contradiction)
check("...and its arbiter reason names the SoC floor rather than reusing "
"'blind' or 'quarter peak', so the one line an operator reads tells the "
"three 0 W states apart", lc.reason == "SoC floor"
and peakguard.claim(d_ok).reason == "quarter peak")
# --- THE PROPERTY THE WHOLE SHAPE RESTS ON: no negative floor, ever ----------
# A `lo` limit is incapable of capping discharge or export only while it is
# non-negative. If this card could produce `lo < 0` it would have broken 9.1 as
# a property of the object, and the guard would be capping the very thing it is
# forbidden to cap. Swept rather than argued.
SWEEP_NEG = SWEEP_BOTH = SWEEP_N = 0
SWEEP_HI = True
for load in (0.0, CEIL - 500.0, CEIL, CEIL + 300.0, CEIL + 3000.0,
LIVE_MAX_W * 3):
for elapsed in (30.0, 450.0, 870.0):
_pr = PeakTracker.project(averager_at(load, elapsed),
B + timedelta(seconds=elapsed))
for bw in (-2000.0, 0.0, 400.0, 4000.0):
for sc in (None, 0.0, BMS_FLOOR, SOC_FLOOR, SOC_FLOOR + 0.5, 100.0):
for fl in (SOC_FLOOR, BMS_FLOOR + 1.0, 55.0):
_d = decide(_pr, batt_w=bw, cap_w=CAP_W, reading=None,
achievable_w=LIVE_MAX_W,
max_hold_s=MAX_HOLD_S, soc=sc, soc_floor=fl)
SWEEP_N += 1
if _d.floor_w is not None and _d.floor_w < 0.0:
SWEEP_NEG += 1
if _d.floor_limited and _d.saturated:
SWEEP_BOTH += 1
_c = peakguard.claim(_d)
if _c is not None and not (isinf(_c.hi) and _c.hi > 0):
SWEEP_HI = False
print(f" swept {SWEEP_N} (load x elapsed x battery x SoC x floor) "
f"combinations, SoC included: {SWEEP_NEG} negative floors, "
f"{SWEEP_BOTH} decisions both floor-limited and saturated")
check(f"NOT ONE of the {SWEEP_N} combinations produces a negative floor. That "
f"is 9.1's 'discharge and export are never capped by the peak guard' as a "
f"property of the object, and SAFETY-24 preserves it - the only two "
f"values floor_w takes are 0.0 and a positive min(wanted, achievable)",
SWEEP_NEG == 0 and SWEEP_N > 600)
check("...and `hi` stays +inf on every claim any of them files",
SWEEP_HI)
check("...and no combination is both floor-limited and saturated, so the two "
"notifications are exclusive on measured inputs and not merely by reading "
"the code", SWEEP_BOTH == 0)
# --- TEST-14: a required field must not quietly gain a default --------------
# ⚠️ PORTED AS A SCAN, NOT HAND-WRITTEN PER FIELD. TEST-14 is open because an
# outcome type's required field gained a default in three separate files and no
# check noticed. The pinned sets below go red if ANY field or argument moves
# between required and defaulted - including a future one nobody has thought of
# - which is the whole difference between a scan and a case.
_sig = inspect.signature(decide)
_required_kw = {n for n, prm in _sig.parameters.items()
if prm.kind is prm.KEYWORD_ONLY and prm.default is prm.empty}
print(f" decide() required keyword-only arguments: {sorted(_required_kw)}")
check("`soc` and `soc_floor` are REQUIRED keyword-only arguments of decide() "
"with no default. A default on either is TEST-14's exact shape and would "
"be the quiet fallback this card exists to remove - and a default on "
"`soc_floor` would additionally be CFG-03's fourth defaults site",
_required_kw == {"batt_w", "cap_w", "reading", "achievable_w", "soc",
"soc_floor"})
_req_fields = tuple(f.name for f in dataclasses.fields(PeakDecision)
if f.default is dataclasses.MISSING
and f.default_factory is dataclasses.MISSING)
print(f" PeakDecision fields with no default: {list(_req_fields)}")
check("...and PeakDecision's required fields are unchanged by this card: the "
"four SAFETY-07 pinned, and the SAFETY-24 additions are genuinely "
"optional reporting rather than required fields handed a default",
_req_fields == ("engaged", "floor_w", "ceiling_w", "ceiling_from"))
# ==============================================================================
print()
print("9. the exhausting replay (AC 4), and the startup refusal")
# ⚠️ THE PACK IS SIZED FROM THE MEASUREMENT, NOT PICKED. `WH_07` is what the
# guard actually asks for on 2026-08-20 - re-derived from the committed capture
# on every run, never quoted - and the modelled pack is then chosen so the energy
# between the start and the EMS floor is ONE THIRD of it. That makes "a quarter
# that would exhaust the pack" true by construction of the measurement rather
# than by a number typed into this file: the day wants three times what the pack
# has to give above the floor, so it must run out, and without an EMS floor it
# must run through the BMS one too.
BAD = [pth for pth in REPLAY if "2026-08-20" in pth.name][0]
WH_BAD = replay(BAD, guarded=True)["wh"]
START_SOC = SOC_FLOOR + (SOC_FLOOR - BMS_FLOOR)
USABLE_WH = 100.0 * (WH_BAD / 3.0) / (START_SOC - SOC_FLOOR)
PACK = (USABLE_WH, START_SOC, BMS_FLOOR)
# How far past the floor one cycle at the rail can carry the pack before the
# next decision sees it. DERIVED from the cycle and the modelled pack, because a
# tolerance typed as a constant is a tolerance that stops meaning anything.
OVERSHOOT_PP = 100.0 * (LIVE_MAX_W * CYCLE_S / 3600.0) / USABLE_WH
print(f" the guard's measured appetite on {BAD.name}: {WH_BAD:.0f} Wh")
print(f" modelled pack: {USABLE_WH:.0f} Wh usable, starting at "
f"{START_SOC:.0f} %, BMS cut at {BMS_FLOOR:.0f} %, EMS floor at "
f"{SOC_FLOOR:.0f} %")
print(f" -> {(START_SOC - SOC_FLOOR) / 100.0 * USABLE_WH:.0f} Wh above the "
f"EMS floor, {(START_SOC - BMS_FLOOR) / 100.0 * USABLE_WH:.0f} Wh above "
f"the BMS one - against {WH_BAD:.0f} Wh of appetite; one cycle at the "
f"rail is {OVERSHOOT_PP:.2f} pp of this pack")
check("the fixture is a pack this day genuinely exhausts, and that is derived "
"from the day's own measured appetite rather than asserted: the energy "
"above the BMS cut-out is less than the guard wants",
(START_SOC - BMS_FLOOR) / 100.0 * USABLE_WH < WH_BAD)
# TWO WORLDS, SAME PACK, SAME CAPTURE.
# BEFORE: the EMS floor set to the BMS cut-out - i.e. the BMS is the sole
# energy backstop, which is precisely what SAFETY-07 shipped.
# AFTER: the EMS floor as config.yaml ships it.
BEFORE = replay(BAD, guarded=True, pack=PACK, soc_floor=BMS_FLOOR)
AFTER = replay(BAD, guarded=True, pack=PACK, soc_floor=SOC_FLOOR)
print(f" BEFORE (BMS is the only backstop): ends at {BEFORE['end_soc']:.1f} %"
f", low-water {BEFORE['min_soc']:.1f} %, {BEFORE['wh']:.0f} Wh out, "
f"{BEFORE['at_bms']} guard cycles spent AT the BMS limit, "
f"{BEFORE['limited']} floor-limited")
print(f" AFTER (peak_soc_floor {SOC_FLOOR:.0f} %): ends at "
f"{AFTER['end_soc']:.1f} %, low-water {AFTER['min_soc']:.1f} %, "
f"{AFTER['wh']:.0f} Wh out, {AFTER['at_bms']} cycles at the BMS limit, "
f"{AFTER['limited']} floor-limited")
check("BEFORE: the day really does take the pack to the BMS cut-out and hold it "
"there. That is the finding this card answers - not a pack destroyed, a "
"pack repeatedly parked at its own protection with no EMS-side record of "
"why", BEFORE["min_soc"] <= BMS_FLOOR + 1e-9 and BEFORE["at_bms"] > 0)
check("AFTER: the forced discharge STOPS AT THE CONFIGURED FLOOR - to within "
"the one cycle of overshoot the 15 s tick allows, derived above and not "
"typed - and the pack never reaches the BMS limit at all. That is AC 1, "
"replayed rather than reasoned about",
AFTER["min_soc"] >= SOC_FLOOR - OVERSHOOT_PP and AFTER["at_bms"] == 0
and AFTER["min_soc"] > BMS_FLOOR + 1.0)
check("...and it does not merely stop, it SAYS SO: floor-limited cycles are "
"counted off the decisions themselves, so the under-shave is reported "
"rather than silent", AFTER["limited"] > 0)
# ⚠️ WHAT "BEFORE" IS AND IS NOT. It is the shipped code with the EMS floor set
# equal to the BMS cut-out, which reproduces the pre-card ENERGY trajectory
# exactly - the pack drains to the BMS and the BMS is the only thing that stops
# it. What it does not reproduce is the silence: pre-card there was no gate to
# report at all, whereas this run reports one at the BMS limit, by which point
# the pack is already there. The trajectory is the thing being compared.
check("...and the two worlds differ in the trajectory, not merely in a flag: "
"before, every floor-limited cycle happens with the pack ALREADY at the "
"BMS limit; after, none of them do",
BEFORE["at_bms"] > 0 and AFTER["at_bms"] == 0)
# THE DERIVED VERDICT: the energy the card withholds is exactly the energy
# between the two floors. Not "less", not "a bit" - the difference in Wh equals
# the difference in percentage points times the pack, to within one cycle.
GAP_WH = (SOC_FLOOR - BMS_FLOOR) / 100.0 * USABLE_WH
TOL = 2.0 * LIVE_MAX_W * CYCLE_S / 3600.0 # one cycle of overshoot per run
print(f" energy withheld: {BEFORE['wh'] - AFTER['wh']:.1f} Wh, against "
f"{GAP_WH:.1f} Wh between the two floors (tolerance {TOL:.1f} Wh = one "
f"cycle of overshoot at the rail in each of the two runs)")
check("the energy this card withholds IS the energy between the two floors, to "
"within one cycle at the rail - so the floor is doing the arithmetic it "
"claims and is not merely correlated with doing less",
abs((BEFORE["wh"] - AFTER["wh"]) - GAP_WH) <= TOL)
check("...and no cycle in either world files a negative floor, on real "
"captured data and not only on the synthetic sweep",
BEFORE["neg_floor"] == 0 and AFTER["neg_floor"] == 0)
# THE HONEST HALF. Stopping costs money, and the check says how much rather than
# looking away from it.
_ub = max(BEFORE["blocks"].values())
_ab = max(AFTER["blocks"].values())
print(f" the price of the floor: worst quarter {_ub:.0f} W before, "
f"{_ab:.0f} W after - a floor-limited guard shaves LESS, on purpose, and "
f"the operator is told so")
check("stopping at the floor costs peak shaving, and this file says so out "
"loud rather than reporting only the SoC it saved: the worst quarter is "
"no better with the floor in place", _ab >= _ub - 1e-6)
# --- the wiring: one real cycle on an empty pack -----------------------------
# ⚠️ `maintenance_soc_floor` IS SET, AND SET TO A VALUE THAT WOULD FLIP THE
# ANSWER. The pack sits one point under `peak_soc_floor` and well ABOVE the
# maintenance floor, so a controller that read the maintenance key would come
# out NOT limited here. Without this the two keys are indistinguishable on every
# fixture in this file - a mutation swapping them survived until it was added.
MAINT_LOW = 5
c24, hass24, NET24, _ = wired_over_limit(
TMP / "soc24.db", LIVE_MAX_W * 0.3, soc=SOC_FLOOR - 1.0,
opts={"maintenance_soc_floor": MAINT_LOW})
check(f"the fixture separates the two floors: the pack at {SOC_FLOOR - 1:.0f} % "
f"is under peak_soc_floor {SOC_FLOOR:.0f} % and over "
f"maintenance_soc_floor {MAINT_LOW} %, so reading the wrong key gives the "
f"wrong answer here", MAINT_LOW < SOC_FLOOR - 1.0 < SOC_FLOOR)
print(f" a real cycle over a {NET24:.0f} W quarter with the pack at "
f"{SOC_FLOOR - 1.0:.0f} %: {c24.peak_guard.explain()}")
check("the REAL Controller reaches the floor-limited branch, so main.py hands "
"decide() a state of charge at all", c24.peak_guard.floor_limited)
check("...and the number written to the inverter is not a forced discharge - "
"the pack is not spent for money below the floor",
hass24.writes and hass24.writes[-1] >= -1e-6
and hass24.writes[-1] <= c24.peak_guard.wanted_w)
check("...and the operator is TOLD THE PEAK WILL BE EXCEEDED, exactly once, "
"with its own notification id - AC 2, and the reason failing to shave "
"must not be silent",
hass24.notifications.count("goodwe_peak_soc_floor") == 1
and "goodwe_peak_saturated" not in hass24.notifications)
asyncio.get_event_loop_policy().new_event_loop().run_until_complete(c24.cycle())
check("...once per episode, not once per cycle - a quarter of an hour of "
"identical cards is how an operator learns to dismiss the one that "
"mattered", hass24.notifications.count("goodwe_peak_soc_floor") == 1)
rows24 = [r for r in c24.checks() if "peak guard" in r["text"].lower()]
print(" status row: " + (rows24[0]["text"] if rows24 else "(none)"))
check("the status page carries exactly one peak-guard row and it is NOT ok - "
"the money is going to be spent and a human is the only actuator left",
len(rows24) == 1 and not rows24[0]["ok"]
and "FLOOR-LIMITED" in rows24[0]["text"])
check("...and it reaches the action log too", any("FLOOR-LIMITED" in e
for e in c24.events))
# A missing SoC sensor, through the same real cycle. `inputs_ok` stops the loop
# at 0 W in that state, so the guard's own answer is what is graded here.
d_nosoc = decide(p_hungry, batt_w=0.0, cap_w=CAP_W, reading=None,
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S,
soc=None, soc_floor=SOC_FLOOR)
check("and a Controller whose SoC entity does not answer hands decide() None, "
"which is floor-limited rather than a full battery - the whole "
"'a failed read is not an observation' rule, on the one input where the "
"disarming value is the plausible one",
d_nosoc.floor_limited and not d_nosoc.saturated)
# --- the startup refusal, in CFG-01 / MAINT-04's manner ----------------------
def opts24(**kw):
"""Otherwise-startable options, so the ONLY thing under test is rule 8."""
base = {"setpoint_entity": "number.sp", "meter_source": "off",
"meter_entity": "sensor.grid", "max_w": 2000,
"meter_phases": 1, "meter_max_age_s": 30, "sign_disagree_w": 300,
"maintenance_charge_w": 2000,
"peak_soc_floor": PEAK_OPTION_DEFAULTS["peak_soc_floor"],
"bms_soc_floor": PEAK_OPTION_DEFAULTS["bms_soc_floor"]}
base.update(kw)
return base
BASE_OK = validate_startup_config(opts24())
check("the SHIPPED pair starts - a validator that refuses config.yaml's own "
"defaults is a validator nobody keeps", BASE_OK == [])
def rule8(problems):
return [x for x in problems if "bms_soc_floor" in x]
below = rule8(validate_startup_config(opts24(peak_soc_floor=BMS_FLOOR - 5)))
equal = rule8(validate_startup_config(opts24(peak_soc_floor=BMS_FLOOR)))
above = rule8(validate_startup_config(opts24(peak_soc_floor=BMS_FLOOR + 1)))
junk = validate_startup_config(opts24(peak_soc_floor="soon"))
print(f" refusal at {BMS_FLOOR - 5:.0f} %: {below[0][:96] if below else '(none)'}...")
check("a floor BELOW the pack's own cut-out is refused at startup, in CFG-01 / "
"MAINT-04's manner - the operator states a bound, the BMS enforces a "
"different one, and nothing would say which won", len(below) == 1)
check("...and so is a floor EXACTLY AT it, which is the interesting case: equal "
"buys zero margin, so it can never be the thing that stops the discharge "
"and the status page would report a backstop that cannot bind. Rule 6 "
"permits equality because equality there leaves one working point; here "
"it leaves none", len(equal) == 1)
check("...while one point above is accepted, so this refuses the disarmed case "
"only and not 'close to the limit'", above == [])
check("...and the refusal NAMES BOTH NUMBERS and the shipped default, so the "
"operator can act on it without reading the source",
f"{BMS_FLOOR:.0f} %" in below[0]
and f"{BMS_FLOOR - 5:.0f} %" in below[0]
and f"{PEAK_OPTION_DEFAULTS['peak_soc_floor']:.0f} %" in below[0])
check("...and a non-numeric floor is refused rather than defaulted, for the "
"reason every read in this file is: a value that cannot be parsed is not "
"an observation of anything",
any("must both be numbers" in x for x in junk))
# ⚠️ AND IT IS NOT SCOPED TO ANYTHING. Rule 6's comment gives the argument: a
# contradictory pair is contradictory whether or not the feature is switched on,
# and scoping it leaves the trap armed for the one operator action that cannot
# be expected to change what a floor means.
check("...and the rule is not scoped to meter_source or to any enable flag - "
"the pair is wrong on a P1 install too, and 'turn P1 on' is not an action "
"that should change what a SoC floor means",
len(rule8(validate_startup_config(opts24(
peak_soc_floor=BMS_FLOOR - 5, meter_source="ha_signed",
p1_net_entity="sensor.net")))) == 1)
# --- CFG-03: the mirror, pinned to config.yaml -------------------------------
# TARIFF-02's pattern exactly - and pinned by READING the shipped file, not by
# transcribing it, which is the only version of this check that can catch drift.
_found24 = {k: read_option(k) for k in PEAK_OPTION_DEFAULTS}
print(f" config.yaml ships {_found24}, peakguard mirrors "
f"{dict(PEAK_OPTION_DEFAULTS)}")
check("every SAFETY-24 option is shipped in config.yaml's options: block",
all(v is not None for v in _found24.values()))
check("...and each shipped default equals PEAK_OPTION_DEFAULTS, the ONE in-code "
"mirror - TARIFF-02's move, so this card adds no fourth defaults site",
all(float(_found24[k]) == float(v)
for k, v in PEAK_OPTION_DEFAULTS.items()))
_schema24 = [k for k in PEAK_OPTION_DEFAULTS if read_schema_max(k) is not None]
check("...and both are in the schema: block too, or the Supervisor drops the "
"option and the add-on silently runs on the mirror instead of the "
"operator's value", sorted(_schema24) == sorted(PEAK_OPTION_DEFAULTS))
_val_src = (HERE / "app" / "validate.py").read_text(encoding="utf-8")
check("...and validate.py takes its fallbacks from that same mirror rather than "
"typing literals - which is the divergence its own rule 6 documents "
"(maintenance_charge_w 2000 there, 2500 in main.py), refused a second "
"outing here",
"PEAK_OPTION_DEFAULTS" in _val_src
and not re.search(r'"peak_soc_floor"\s*,\s*[0-9]', _val_src)
and not re.search(r'"peak_soc_floor"\s*,\s*[0-9]', _mainsrc))
check("...and 20 % is not a number typed into config.yaml on a hunch: it is 10 "
"points of margin over the pack's own cut-out, which on the 13.8 kWh "
"reference pack is 1.57x the guard's worst measured day",
PEAK_OPTION_DEFAULTS["peak_soc_floor"]
- PEAK_OPTION_DEFAULTS["bms_soc_floor"] == 10
and 0.10 * 13800.0 / WH_BAD > 1.5)
check("...and it is a SEPARATE KEY from maintenance_soc_floor, whose SHIPPED "
f"VALUE ({MAINT_FLOOR:.0f} %) is not this one ({SOC_FLOOR:.0f} %) - which "
"is the argument for not sharing them, measured off config.yaml rather "
"than asserted: they are a drain TARGET and a spend LIMIT, and the site "
"that wants one does not thereby want the other",
MAINT_FLOOR != SOC_FLOOR
and "maintenance_soc_floor" not in PEAK_OPTION_DEFAULTS)
# ==============================================================================
print()
print("10. mutation: what defeats these guards, and which check bites")
# Each entry re-implements ONE defeat and reports whether a check above would
# have gone red. The last row is the honest one: it does NOT bite, and saying so
@@ -1199,7 +1770,7 @@ MUT.append(("blindness() returns None - the degenerate 0 W projection is graded"
# 2. the frozen term dropped from the guard's inputs.
MUT.append(("frozen=False - SAFETY-18's term dropped",
not decide(p_frozen, batt_w=-RAILED, cap_w=CAP_W, reading=None,
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S,
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR,
frozen=False).engaged and d_frozen.engaged,
"part 4: 'THE BACKSTOP: SAFETY-18's runaway ... cut to 0 W'"))
@@ -1247,12 +1818,68 @@ MUT.append(("NOT_OBSERVED silently replaced by a peak of 0.0 W",
"NOTHING - and by construction: the ceiling is max(cap, peak), so a "
"zero peak is the TIGHTEST possible input to it"))
# ---- SAFETY-24's own defeats ------------------------------------------------
# 8. the SoC gate removed altogether - i.e. the guard exactly as SAFETY-07
# shipped it. Re-derived on the same modelled pack, not described.
MUT.append(("the SoC gate removed - SAFETY-07's unbounded forced discharge",
BEFORE["min_soc"] <= BMS_FLOOR and AFTER["min_soc"] > BMS_FLOOR,
"part 9: 'the forced discharge STOPS AT THE CONFIGURED FLOOR'"))
# 9. an unreadable SoC read as a full battery. THE DISARMING DIRECTION, and the
# one this project has actually shipped once (NOTES.md 4.4b).
MUT.append(("`soc or 100` - an unreadable SoC read as a full battery",
discharge_permitted(float(None or 100), SOC_FLOOR) is None
and d_unread.floor_limited,
"part 8: 'an unreadable SoC is refused, and refused with its OWN "
"reason'"))
# 10. the gate moved ABOVE the `wanted > 0` test, i.e. graded where it is
# degenerate. Counted on the healthy day rather than argued: every quiet
# cycle would report the peak was about to be exceeded.
_early = sum(1 for _t in (500.0, 1000.0, 2000.0)
if discharge_permitted(SOC_FLOOR - 1.0, SOC_FLOOR) is not None
and not decide(PeakTracker.project(
averager_at(_t, 450.0), B + timedelta(seconds=450)),
batt_w=BATT_NOW, cap_w=CAP_W, reading=None,
achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S,
soc=SOC_FLOOR - 1.0, soc_floor=SOC_FLOOR).engaged)
MUT.append((f"the SoC gate moved above the `wanted > 0` test ({_early} quiet "
f"quarters would have alarmed)",
_early > 0 and not d_quiet.floor_limited,
"part 8: 'an empty pack on a quarter that does not need shaving is "
"IDLE, not floor-limited'"))
# 11. rule 8 written `<` instead of `<=`, so a floor EQUAL to the BMS cut-out
# is accepted and buys zero margin.
MUT.append(("validate rule 8 written `<` instead of `<=`",
len(rule8(validate_startup_config(
opts24(peak_soc_floor=BMS_FLOOR)))) == 1,
"part 9: 'so is a floor EXACTLY AT it, which is the interesting "
"case'"))
# 12. the floor-limited branch filing `wanted` anyway - a report with no action,
# which is the shape a "just log it" review comment produces.
MUT.append(("floor-limited reported but the floor still filed at `wanted`",
near(d_lim.floor_w, 0.0) and d_lim.wanted_w > 0.0,
"part 8: 'at the floor it STOPS, and the floor it files is 0 W'"))
# 13. THE SECOND ONE THAT DOES NOT BITE, and it is a real limitation rather than
# a defect. Stated, because the alternative is a reader believing this card
# bounds more than it does.
_loop_unbounded = near(r_disch.target_w, 1800.0, 1e-9)
MUT.append(("the grid-following loop discharging past this floor",
not _loop_unbounded,
"NOTHING, ON PURPOSE - this card bounds THE PEAK GUARD's forced "
"discharge, the thing SAFETY-07 introduced. SAFETY-18's finding "
"that the loop itself has no SoC guard is still open, and a `lo` "
"limit is structurally incapable of closing it"))
for label, caught, where in MUT:
print(f" {'caught' if caught else 'NOT CAUGHT':>10} {label}")
print(f" {where}")
check("the six mutations that CAN disarm this guard are each caught by a check "
"above - a guard whose checks survive its own defeat is not a guard",
all(m[1] for m in MUT[:6]))
check("the eleven mutations that CAN disarm these guards are each caught by a "
"check above - a guard whose checks survive its own defeat is not a "
"guard", all(m[1] for m in MUT[:6]) and all(m[1] for m in MUT[7:12]))
check("...and the seventh is honestly reported as not caught, because it cannot "
"disarm anything: `max(cap, peak)` makes a zero peak the strictest "
"ceiling there is, so the disarming direction is closed by the shape of "
@@ -1261,6 +1888,11 @@ check("...and the seventh is honestly reported as not caught, because it cannot
"same zero into FULL freedom, which part 3 shows is why headroom_w() is "
"not consumed here", not MUT[6][1]
and quarter_ceiling_w(CAP_W, _zero_peak)[0] == CAP_W)
check("...and so is the last one, which is not a defeat at all but the EDGE OF "
"THIS CARD: the grid-following loop can still take the pack past this "
"floor, because the floor bounds the peak guard's own forced discharge "
"and a `lo` limit cannot cap a discharge. Saying so here is worth more "
"than a green line that would be untrue", not MUT[12][1])
# ==============================================================================