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

SAFETY-119 round 3: rolling 15-minute budget for the hidden integrator bound
checks / unit (pull_request) Successful in 42m33s
Details
infra-pin / fresh (pull_request) Successful in 2s
Details

Browse Source 
At most 8 hidden bound cycles in any 900 s across all hand-backs, replacing
the per-run budget. Chosen from a sweep (N 3 to 10, 1 to 15 s meters, both
cases); DECISIONS has the table. Tests: never case at 5, 10, 15 s, quarter
edge at 2, 10, 15 s, the rolling budget at its boundaries.

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
glenn schrooyen
2026-10-20 17:27:43 +02:00
co-authored by Claude
parent 85542e443b
commit 103db2fcda
3 changed files with 213 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files
goodwe_controller/test_safety119.py
+151 -103
View File
@@ -9,10 +9,10 @@ It crossed its rail for one or two cycles, the loop's ask beat the one-sided
limit, so the cycle counted as the loop's own and raised
`goodwe_integrator_bound`. The battery gave what it was told. Now the loop's
bound does not count until the plant's settle (SHORT_CYCLES cycles and
SHORT_SPAN_S) has passed since the hand-back. Round 2: one such window per
run of hand-backs, hiding at most RUN_BOUND_CYCLES bound cycles and
SHORT_SPAN_S in all, so a bound that stays warns when the guard and the
charge keep taking turns. DECISIONS/SAFETY-119-bound-after-hand-back.md.
SHORT_SPAN_S) has passed since the hand-back. Round 3: across all
hand-backs the bound is hidden on at most ROLLING_BOUND_CYCLES cycles in any
ROLLING_S seconds, so a bound that stays warns when the guard and the charge
keep taking turns. DECISIONS/SAFETY-119-bound-after-hand-back.md.
1. The rig's shape (1 s meter that lags the battery, 1 s battery lag, house
800 W, rating 2500 W, limit 2500 W, margin 50, charge 2000 W, no
@@ -23,10 +23,12 @@ charge keep taking turns. DECISIONS/SAFETY-119-bound-after-hand-back.md.
hand-back (after the settle).
3. With a usable forecast: the same commands and the same alarms as without
the fix.
4. Repeated hand-backs (N=2, 5 s meter, the guard and the charge take turns):
it warns at least once per quarter; a start or a blind cycle is not a
4. Repeated hand-backs (N=2, the guard and the charge take turns), 5, 10 and
15 s meters: it warns (every quarter at 5 s and 10 s); the quarter-edge
case stays silent at 10 s and 15 s too; a start or a blind cycle is not a
hand-back; a slow meter delays by cycles, not seconds.
5. The rule at its boundaries (`Controller._settling`).
5. The rule at its boundaries (`Controller._settling`): the settle, and the
rolling budget.
6. test_safety84's and test_safety87's re-pinned digests: the same commands
and claims; only OUTPUT_CLAMP log rows differ.
@@ -248,14 +250,14 @@ _first = notes(_hb[1])[:1]
print(f" hand-back to a loop at its rail at 600 s: first notification "
f"{_first}, without the fix {notes(_hb_old[1])[:1]}; command at the "
f"end {_hb[0][-1][1]}")
check("⭐ the loop at its rail after a hand-back still warns, once the "
"settle has passed (12 cycles, 22 s): delay, not loss",
_first and 600.0 + app_main.SHORT_SPAN_S <= _first[0] <= 640.0
MB = 8
check(f"⭐ the loop at its rail after a hand-back still warns, once "
f"{MB} of its bound cycles are hidden (1 s meter: {MB} s)",
_first and 600.0 + MB <= _first[0] <= 600.0 + MB + 4.0
and _hb[0][-1][1] == -2500.0
and _hb[1].clamp_wd.bound > 100)
check("...without the fix it warned at once (non-vacuous)",
notes(_hb_old[1])[:1] and notes(_hb_old[1])[0] < 600.0
+ app_main.SHORT_SPAN_S)
notes(_hb_old[1])[:1] and notes(_hb_old[1])[0] < 600.0 + MB)
# The same charge with no forecast; at 1795 s, under the guard's step-in at
# the quarter's end, the house goes to 4000 W, over the 2500 W rating.
_up = run(house(1, step=(1795.0, 4000.0)), dur=1900.0)
@@ -309,32 +311,79 @@ def round1(self, owns, now):
return True
# N=2 (2 x 2500 W), a 5 s meter, a cheap-window charge, no usable forecast,
# the house 50 W over the 5000 W rating from 300 s: the loop never held the
# wire for more than 10 cycles. Round 1: no notification in 2 h.
rig.TICK = 5.0
_nv = run(house(1, step=(300.0, 5050.0)), n=2)
_nv_old = no_fix(lambda: run(house(1, step=(300.0, 5050.0)), n=2))
_nv_r1 = patched(Controller, "_settling", round1)(
lambda: run(house(1, step=(300.0, 5050.0)), n=2))
rig.TICK = 1.0
_per_q = [sum(1 for t in notes(_nv[1]) if int(t // QUARTER) == q)
for q in range(8)]
print(f" N=2, 5 s meter, house 5050 W: notifications {len(notes(_nv[1]))} "
f"(per quarter {_per_q}); round 1's rule {len(notes(_nv_r1[1]))}; "
f"without the fix {len(notes(_nv_old[1]))}; highest quarter "
f"{max(quarters(_nv[1], 5.0)):.0f} W")
check("⭐ it warns at least once in every quarter after the first warning "
"(the guard and the charge keep taking turns)",
len(notes(_nv[1])) >= 7 and all(_per_q[q] >= 1 for q in range(1, 8)))
# N=2 (2 x 2500 W), a cheap-window charge, no usable forecast, the house 50 W
# over the 5000 W rating from 300 s: the loop never held the wire for more
# than 10 cycles. Round 1: no notification in 2 h; round 2: none at 10 s or
# 15 s meters.
def never(tick, hw=5050.0, fix=True):
rig.TICK = tick
try:
f = lambda: run(house(1, step=(300.0, hw)), n=2)
return f() if fix else no_fix(f)
finally:
rig.TICK = 1.0
def per_quarter(c):
return [sum(1 for t in notes(c) if int(t // QUARTER) == q)
for q in range(8)]
_nv = {t: never(t) for t in (5.0, 10.0, 15.0)}
_nv_old = {t: never(t, fix=False) for t in (5.0, 10.0, 15.0)}
_nv_r1 = patched(Controller, "_settling", round1)(lambda: never(5.0))
for _t in _nv:
print(f" N=2, {_t:.0f} s meter, house 5050 W: notifications "
f"{len(notes(_nv[_t][1]))} (per quarter "
f"{per_quarter(_nv[_t][1])}), without the fix "
f"{len(notes(_nv_old[_t][1]))}, highest quarter "
f"{max(quarters(_nv[_t][1], _t)):.0f} W")
print(f" round 1's rule at 5 s: {len(notes(_nv_r1[1]))}")
check("⭐ 5 s meter: it warns in every quarter after the first",
all(per_quarter(_nv[5.0][1])[q] >= 1 for q in range(1, 8)))
check("⭐ 10 s meter: it warns in every quarter after the first (round 2 "
"gave none)",
all(per_quarter(_nv[10.0][1])[q] >= 1 for q in range(1, 8)))
check("⭐ 15 s meter: it warns (round 2 gave none), in at least one quarter "
"of the seven after the first",
sum(1 for n in per_quarter(_nv[15.0][1])[1:] if n) >= 1)
check("non-vacuous: round 1's rule (a window per hand-back) never warned "
"here, and the fix still hides some of what was raised without it",
notes(_nv_r1[1]) == []
and _nv[1].clamp_wd.bound < _nv_old[1].clamp_wd.bound)
check("...the cap held and every command is the same as without the fix",
len(quarters(_nv[1], 5.0)) == 8
and max(quarters(_nv[1], 5.0)) < 5000.0
and commands(_nv[0]) == commands(_nv_old[0]))
notes(_nv_r1[1]) == [] and all(
_nv[t][1].clamp_wd.bound < _nv_old[t][1].clamp_wd.bound
for t in _nv))
check("...every command is the same as without the fix, at 5, 10 and 15 s",
all(commands(_nv[t][0]) == commands(_nv_old[t][0]) for t in _nv))
check("...and at 5 s the cap held",
len(quarters(_nv[5.0][1], 5.0)) == 8
and max(quarters(_nv[5.0][1], 5.0)) < 5000.0)
# The quarter-edge case (part 1) at the slower meters: still no warning.
# The meter's own lag limits the cap there, the same with and without the fix.
_qe = {}
for _t, _seeds in ((10.0, ((1, 1), (2, 1), (3, 2))),
(15.0, ((1, 1), (2, 1), (3, 2))),
(2.0, ((8, 1), (9, 2)))): # a budget of 7 warns here
rig.TICK = _t
for _seed, _dl in _seeds:
_qe[_t, _seed] = (run(house(_seed), delay=_dl),
no_fix(lambda: run(house(_seed), delay=_dl)))
rig.TICK = 1.0
print(f" {_t:.0f} s meter, quarter edge: notifications "
f"{[len(notes(r[0][1])) for k, r in _qe.items() if k[0] == _t]} "
f"(without the fix "
f"{[len(notes(r[1][1])) for k, r in _qe.items() if k[0] == _t]})")
check("⭐ the quarter-edge case at 10 s and 15 s meters, and at 2 s on two "
"more seeds: no notification",
all(notes(r[0][1]) == [] and r[0][1].clamp_wd.bound == 0
for r in _qe.values()))
check("...every command the same as without the fix; without it the false "
"alarm is there (non-vacuous)",
all(commands(r[0][0]) == commands(r[1][0]) for r in _qe.values())
and all(len(notes(r[1][1])) >= 1 for r in _qe.values()))
check("the budget is the one the sweep chose (8). Changing it means "
"re-running the sweep in DECISIONS/SAFETY-119-bound-after-hand-back.md",
app_main.ROLLING_BOUND_CYCLES == 8 and app_main.ROLLING_S == 900.0)
# A start, or a blind cycle, is not a hand-back: the loop at its rail after
# 20 s of not running warns at once (R2: a restart counted as a hand-back).
@@ -348,8 +397,8 @@ check("⭐ a start after a blind spell is not a hand-back: it warns at the "
notes(_bl[1])[:1] and 20.0 <= notes(_bl[1])[0] <= 26.0
and notes(_bl[1]) == notes(_bl_old[1]))
# The settle is SHORT_CYCLES cycles, not only SHORT_SPAN_S (R4): at a 5 s
# meter the loop at its rail after a hand-back warns 11 cycles (55 s) later.
# The budget is cycles, not seconds (R4): at a 5 s meter the loop at its
# rail after a hand-back warns 8 cycles (40 s) later, not 8 s.
rig.TICK = 5.0
_sl = run(house(1, step=(600.0, -6000.0)), dur=900.0,
extra={"peak_cap_w": 20000})
@@ -357,9 +406,9 @@ rig.TICK = 1.0
_sl1 = [t for t in notes(_sl[1]) if t >= 600.0][:1]
print(f" 5 s meter, hand-back to a loop at its rail at 600 s: first "
f"notification {_sl1}")
check("⭐ at a 5 s meter the settle is 11 cycles: the first warning comes "
"55 s after the hand-back, not 22 s",
_sl1 and 600.0 + 11 * 5.0 <= _sl1[0] <= 600.0 + 11 * 5.0 + 10.0)
check(f"⭐ at a 5 s meter the budget is cycles, not seconds: the first "
f"warning comes {MB} cycles ({MB * 5} s) after the hand-back",
_sl1 and 600.0 + MB * 5.0 <= _sl1[0] <= 600.0 + MB * 5.0 + 10.0)
# ============================================================================ #
print("\n5. the rule at its boundaries")
@@ -369,7 +418,7 @@ class _C:
_settling = Controller._settling
def __init__(self):
self._wire_lost, self._handback, self._away = False, None, None
self._wire_lost, self._handback, self._hidden = False, None, ()
self.loop_clamped = (False, "", False)
@@ -383,6 +432,7 @@ def seq(c, steps):
N, SPAN = app_main.SHORT_CYCLES, app_main.SHORT_SPAN_S
M, ROLL = 8, 900.0 # the chosen budget, written out on purpose
_c = _C()
check("from the start, owned every cycle: never settling (no hand-back)",
seq(_c, [(True, float(i)) for i in range(30)]) == [False] * 30)
@@ -393,37 +443,36 @@ check(f"after another owner: settling for {N - 1} owned cycles, then not "
_out == [False] + [True] * (N - 1) + [False] * 3)
_c = _C()
_out = seq(_c, [(False, 0.0)] + [(True, 0.5 * i) for i in range(1, 60)])
_end = _out.index(False, 1)
_end = _out.index(False, 1) if False in _out[1:] else -1
check(f"at a 0.5 s meter the span decides: settling until {SPAN:.0f} s",
_end == 1 + int(SPAN / 0.5) and all(_out[1:_end]))
_c = _C()
_out = seq(_c, [(False, 0.0), (True, 2.0), (True, 4.0), (False, 6.0),
(True, 8.0), (True, 10.0)])
check("losing the wire again restarts the settle (but not the run)",
check("losing the wire again restarts the settle",
_out == [False, True, True, False, True, True]
and _c._handback == (8.0, 2, 0, 0))
and _c._handback == (8.0, 2))
_c = _C()
check("no monotonic clock: the cycle count alone decides",
seq(_c, [(False, None)] + [(True, None)] * (N + 1))
== [False] + [True] * (N - 1) + [False] * 2)
# One run of hand-backs hides RUN_BOUND_CYCLES bound cycles and SHORT_SPAN_S,
# however often the wire changes hands. T = the loop's bound that cycle.
M = app_main.RUN_BOUND_CYCLES
# The rolling budget: the loop's bound is hidden on at most M cycles in any
# ROLL seconds, across all hand-backs. T = the loop's bound that cycle.
assert M < N - 1, "the budget is meant to run out inside one settle"
def turns(c, tick, owned, lost, periods, first=0.0):
"""A hand-back, then `periods` x (`owned` bound cycles, `lost` cycles
away). Returns (time, hidden) of each owned bound cycle."""
def turns(c, tick, lost, periods, first=0.0, bound=True):
"""A hand-back, then `periods` x (1 owned cycle with the loop's bound,
`lost` cycles away). Returns [(time, hidden)] of the owned cycles."""
t, out = first, []
for _ in range(lost):
c._settling(False, t)
t += tick
for _ in range(periods):
for _ in range(owned):
c.loop_clamped = (True, "", False)
out.append((t, c._settling(True, t)))
t += tick
c.loop_clamped = (bound, "", False)
out.append((t, c._settling(True, t)))
t += tick
for _ in range(lost):
c._settling(False, t)
t += tick
@@ -431,69 +480,68 @@ def turns(c, tick, owned, lost, periods, first=0.0):
_c = _C()
_o, _ = turns(_c, 5.0, 1, 3, 20)
_hid = [h for _t, h in _o]
check(f"at 5 s, one bound cycle per turn: {M} are hidden, the {M + 1}th and "
"every later one counts",
_hid == [True] * M + [False] * (20 - M))
_o = seq(_c, [(False, 0.0)] + [(True, 5.0 * i, True) for i in range(1, N)])
check(f"one hand-back at 5 s, the bound on every cycle: {M} are hidden, "
f"then it counts for the rest of the settle ({N - 1} cycles)",
_o == [False] + [True] * M + [False] * (N - 1 - M))
_c = _C()
_o, _ = turns(_c, 1.0, 2, 1, 12)
_first, _o0 = first(t for t, h in _o if not h), nth(_o, 0)[0]
check(f"at 1 s the span holds too: not counted before {SPAN:.0f} s after "
f"the first hidden one (counted at {_first:.0f} s)",
present(_first, _o0) and _o0 + SPAN <= _first <= _o0 + SPAN + 3.0
and sum(h for _t, h in _o) > M)
_o, _ = turns(_c, 5.0, 3, 60)
_hid = [t for t, h in _o if h]
check(f"across hand-backs (one bound cycle per turn, 5 s): the first {M} "
"are hidden, then every one counts",
[h for _t, h in _o[:M + 5]] == [True] * M + [False] * 5)
check(f"...never more than {M} hidden in any {ROLL:.0f} s, and they come "
f"back as the oldest ages out (exactly {M} in the first {ROLL:.0f} s "
"after the first)",
all(sum(1 for x in _hid if t <= x < t + ROLL) <= M for t in _hid)
and sum(1 for x in _hid if x < _hid[0] + ROLL) == M
and any(x >= _hid[0] + ROLL for x in _hid))
_c = _C()
_t = 0.0
for _ in range(20): # an over-rating: always counts
for _ in range(3):
_c._settling(False, _t)
_t += 5.0
_c.loop_clamped = (True, "", True)
_c._settling(True, _t)
_t += 5.0
for _ in range(3):
_c._settling(False, _t)
_t += 5.0
_c._hidden = (0.0,) * M
_c._wire_lost, _c._handback = False, (ROLL - 30.0, 0)
_c.loop_clamped = (True, "", False)
check("an over-rating does not use up what the run may hide",
_c._settling(True, _t) is True)
_a = _c._settling(True, ROLL - 1.0)
_c.loop_clamped = (True, "", False)
_b = _c._settling(True, ROLL)
check(f"the budget refills {ROLL:.0f} s after the oldest hidden cycle: "
f"counted at {ROLL - 1:.0f} s, hidden again at {ROLL:.0f} s",
_a is False and _b is True)
_c = _C()
_o1, _t = turns(_c, 2.0, 1, 3, 20)
_o1, _t = turns(_c, 2.0, 3, 20)
for _ in range(N): # the wire away a whole settle
_c._settling(False, _t)
_t += 2.0
_o2 = []
for _ in range(20):
_c.loop_clamped = (True, "", False)
_o2.append(_c._settling(True, _t))
_c._settling(False, _t + 2.0)
_t += 4.0
check("a settle with the wire away ends the run: the next hand-back hides "
f"{M} again",
[h for _t2, h in _o1] == [True] * M + [False] * (20 - M)
and _o2 == [True] * M + [False] * (20 - M))
_o2, _ = turns(_c, 2.0, 3, 5, first=_t)
check("the budget is not per hand-back or per run: after a whole settle "
"with the wire away, a new hand-back within the same "
f"{ROLL:.0f} s hides nothing more",
[h for _t2, h in _o1[:M]] == [True] * M
and not any(h for _t2, h in _o2))
_c = _C()
_o1, _t = turns(_c, 2.0, 1, 3, 20)
for _ in range(N - 1 - 3): # 3 already: N - 1 away
_c._settling(False, _t)
_t += 2.0
_c.loop_clamped = (True, "", False)
check("...but not a shorter time away: the same run, nothing more hidden",
_c._settling(True, _t) is False)
_c = _C()
_o1, _t = turns(_c, 2.0, 1, 3, 20)
_o1, _t = turns(_c, 2.0, 3, 20)
_ = [_c._settling(True, _t + 2.0 * i) for i in range(N + 1)]
_c._settling(False, _t + 40.0)
_c.loop_clamped = (True, "", False)
check("the loop holding the wire a whole settle ends the run too: the next "
"hand-back hides again",
_c._settling(True, _t + 42.0) is True)
check("the loop holding the wire a whole settle ends the window: the next "
"hand-back hides nothing more while the budget is spent",
_c._settling(True, _t + 42.0) is False)
_c = _C()
_c._wire_lost = True
_t = 0.0
for _ in range(M + 4): # an over-rating: always counts
_c.loop_clamped = (True, "", True)
_c._settling(True, _t)
_c._wire_lost = True
_t += 5.0
_c._settling(False, _t)
_c.loop_clamped = (True, "", False)
check("an over-rating does not use up the budget",
_c._settling(True, _t + 5.0) is True and len(_c._hidden) == 1)
# An over-rating (a command over `max_w`) is a bug, never settled away.
_ctl = _hb[1]
_ctl.loop_clamped = (True, "x", True)
_ctl._wire_lost, _ctl._handback = True, None
_ctl._wire_lost, _ctl._handback, _ctl._hidden = True, None, ()
class _Res: