glenn schrooyenandClaude Opus 5.5
ba31e7c8b1
During a HOLD the holding battery gives what keeps the quarter at the peak aim. N=1: the peak floor is refiled at P_MAINTENANCE, as for a drain (SAFETY-83). N>=2: the hold's pin is raised to the floor minus what the other live batteries may place, so they give first. A site stop still decides alone. The hold ends on its clock even if the level dipped. Below the aim nothing changes (rows identical). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
367 lines
16 KiB
Python
367 lines
16 KiB
Python
"""Runnable check for SAFETY-86: the peak guard outranks the maintenance HOLD.
|
|
`python3 test_safety86.py`
|
|
|
|
During the HOLD the battery rests full at 0 W (P_MAINTENANCE, above the peak
|
|
floor), so the house imported everything and quarters went over the peak
|
|
limit. Now the holding battery gives what keeps the quarter at the aim: at N=1
|
|
the peak floor is refiled at P_MAINTENANCE, at N>=2 the hold's pin is raised
|
|
to what the other batteries cannot give. Drives the REAL Controller
|
|
(tools/safety84_identity.py's harness). DECISIONS/SAFETY-86-peak-over-hold.md.
|
|
|
|
1. N=1, house over the aim: every quarter lands on the aim, the hold goes on,
|
|
no 'Load too high'; without the fix the quarter goes over and it fires.
|
|
2. Under the aim nothing changes, at N=1 and N=2 (rows identical).
|
|
3. N=2, house over the aim: the other battery gives first, the holding one
|
|
only the shortfall; with battery modes on and B Off too.
|
|
4. A month peak banked above the limit: the aim is that peak - margin.
|
|
5. A safety stop still parks everything, and the hold gives again after.
|
|
6. The level dips below the target: the hold still ends on its clock.
|
|
7. The rule at its boundaries, on claim lists.
|
|
|
|
Mutations (MUTATION.md, purged copies) are listed in the pull request.
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
import sys
|
|
import types
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, HERE)
|
|
sys.path.insert(0, os.path.join(HERE, "tools"))
|
|
logging.disable(logging.WARNING)
|
|
|
|
import opt11_identity as rig # noqa: E402
|
|
import safety84_identity as s84 # noqa: E402
|
|
from app import control as _ctl # noqa: E402
|
|
from app import packmode # noqa: E402
|
|
from app.arbiter import (Claim, P_MAINTENANCE, P_PEAK, P_SAFETY, # noqa
|
|
resolve)
|
|
from app.main import Controller # noqa: E402
|
|
from app.maintenance import DRAIN, HOLD, IDLE, MaintResult # noqa: E402
|
|
from app.store import unit_key # noqa: E402
|
|
|
|
fails = []
|
|
A, B = "inverter-a", "inverter-b"
|
|
AIM = 3450.0 # peak_cap_w 3500 - peak_margin_w 50
|
|
TOL = 10.0 # W over the aim a banked quarter may show (SAFETY-81)
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
def store_hold(n, b_off=False):
|
|
"""A store with battery A in its HOLD from the first cycle."""
|
|
s = rig._Store()
|
|
for k in ("phase", "phase_started"):
|
|
s.data[k if n == 1 else unit_key(A, k)] = (
|
|
"hold" if k == "phase" else rig.BASE.isoformat())
|
|
if b_off:
|
|
s.data[unit_key(B, packmode.USE_KEY)] = False
|
|
return s
|
|
|
|
|
|
def full(u, t):
|
|
"""A rests full; B is half full."""
|
|
return 99.0 if u == "a" else 60.0
|
|
|
|
|
|
def run(house, n, *, hold_min=600, b_off=False, extra=None, **kw):
|
|
kw.setdefault("soc", full)
|
|
kw.setdefault("dur", 2700.0)
|
|
return s84.run(n, load=house if callable(house) else lambda t: house,
|
|
cheap=False, store=store_hold(n, b_off),
|
|
extra=dict({"maintenance_hold_min": hold_min},
|
|
**(extra or {})), **kw)
|
|
|
|
|
|
def without_fix(fn):
|
|
real = Controller._peak_over_hold
|
|
Controller._peak_over_hold = lambda self, claims, results: None
|
|
try:
|
|
return fn()
|
|
finally:
|
|
Controller._peak_over_hold = real
|
|
|
|
|
|
def cmd(r):
|
|
return dict(r[2])
|
|
|
|
|
|
def banked(c):
|
|
return [b.offtake_avg_w for b in c.p1.blocks]
|
|
|
|
|
|
def on_aim(c, aim=AIM):
|
|
b = banked(c)
|
|
return len(b) >= 2 and all(aim - TOL <= v <= aim + TOL for v in b)
|
|
|
|
|
|
def alarmed(c):
|
|
"""Did SAFETY-81's 'Load too high' alarm fire on this run?"""
|
|
return any("PEAK_GUARD_SATURATED" in e for e in c.events)
|
|
|
|
|
|
def refused(rows):
|
|
return sum(1 for r in rows if r[7] and "REFUSED" in r[7])
|
|
|
|
|
|
def peak_prio(r):
|
|
return [c[1] for c in r[-1][0] if c[0] == "peak"]
|
|
|
|
|
|
def hold_pin(r):
|
|
return [c[3] for c in r[-1][0] if c[0] == "maintenance" and c[2] == "set"]
|
|
|
|
|
|
# ============================================================================ #
|
|
print("1. N=1: a house over the aim during the HOLD")
|
|
N1 = {h: run(h, 1) for h in (5000.0, 6500.0, 7000.0)}
|
|
N1_OLD = {h: without_fix(lambda: run(h, 1)) for h in N1}
|
|
for h in N1:
|
|
print(f" house {h:.0f}: banked {[round(v) for v in banked(N1[h][1])]} W"
|
|
f" (before the fix {[round(v) for v in banked(N1_OLD[h][1])]} W), "
|
|
f"last {N1[h][0][-1][6]}")
|
|
check("⭐ every quarter lands on the aim (3450 W) for a 5000, 6500 and "
|
|
"7000 W house, and 'Load too high' never fires",
|
|
all(on_aim(c) and not alarmed(c) for _r, c in N1.values()))
|
|
check("...the battery gives, raised for the peak limit, and stays in its "
|
|
"HOLD: the phase is hold and the claim still says so",
|
|
all(c.maints[0].phase == HOLD and cmd(rows[-1])["a"] > 1000.0
|
|
and "limited by peak" in rows[-1][6]
|
|
and c.why_words.endswith("raised for the peak limit")
|
|
for rows, c in N1.values()))
|
|
check("non-vacuous: without the fix the battery sits at 0 W, the quarter "
|
|
"banks the whole house and 'Load too high' fires",
|
|
all(all(cmd(r)["a"] == 0.0 for r in rows) and min(banked(c)) >= h - 1
|
|
and alarmed(c) for h, (rows, c) in N1_OLD.items()))
|
|
|
|
|
|
def hob(t):
|
|
"""A 6000 W house with a 5 s spike to 11000 W every 150 s."""
|
|
return 11000.0 if t % 150 < 5 else 6000.0
|
|
|
|
|
|
_sp_rows, _sp = run(hob, 1)
|
|
check("...and 5 s spikes that saturate the guard for a cycle do not fire it "
|
|
"(SAFETY-81's 10 s hold)",
|
|
any(r[-1][1] and r[-1][1][1] >= 5000.0 for r in _sp_rows)
|
|
and not alarmed(_sp))
|
|
|
|
# ============================================================================ #
|
|
print("\n2. under the aim the HOLD is as before")
|
|
BELOW = {(n, h): (run(h, n, dur=1800.0),
|
|
without_fix(lambda: run(h, n, dur=1800.0)))
|
|
for n, h in ((1, 1200.0), (1, 3000.0), (1, -1500.0),
|
|
(2, 1500.0), (2, 3000.0), (2, 5000.0))}
|
|
check("⭐ N=1 at a 1200 W and 3000 W house and a 1500 W surplus, N=2 at "
|
|
"1500, 3000 and 5000 W (B covers it): battery A is at 0 W on every "
|
|
"cycle and every row equals the run without the fix",
|
|
all(all(cmd(r)["a"] == 0.0 for r in new[0]) and new[0] == old[0]
|
|
for new, old in BELOW.values()))
|
|
check("...and the peak claim stays at P_PEAK on every cycle",
|
|
all(p == P_PEAK for new, _o in BELOW.values() for r in new[0]
|
|
for p in peak_prio(r)))
|
|
|
|
# ============================================================================ #
|
|
print("\n3. N=2: the other battery gives first, the holding one the rest")
|
|
N2 = {h: run(h, 2) for h in (6500.0, 7000.0)}
|
|
N2_OLD = {h: without_fix(lambda: run(h, 2)) for h in N2}
|
|
for h, (rows, c) in N2.items():
|
|
print(f" house {h:.0f}: banked {[round(v) for v in banked(c)]} W "
|
|
f"(before the fix {[round(v) for v in banked(N2_OLD[h][1])]} W), "
|
|
f"last {rows[-1][2]}")
|
|
check("⭐ house 6500 and 7000 W, B at most 2500 W: every quarter lands on "
|
|
"the aim, nothing is refused or split both ways, no 'Load too high'",
|
|
all(on_aim(c) and refused(rows) == 0 and not alarmed(c)
|
|
and all(cmd(r)["a"] * cmd(r)["b"] >= 0.0 for r in rows)
|
|
for rows, c in N2.values()))
|
|
_gives = [r for rows, _c in N2.values() for r in rows if cmd(r)["a"] > 0.0]
|
|
check("⭐ A gives only the shortfall: on every cycle A gives, B is at its "
|
|
"2500 W and A's pin is the guard's floor minus 2500 W",
|
|
len(_gives) > 500
|
|
and all(cmd(r)["b"] == 2500.0 and hold_pin(r)
|
|
and abs(hold_pin(r)[0] - (r[-1][1][1] - 2500.0)) < 1e-6
|
|
for r in _gives))
|
|
check("...A stays in its HOLD, and its last command is the shortfall "
|
|
"(6500 - 3450 - 2500 = 550 W, give or take the meter's 1 W)",
|
|
all(c.maints[0].phase == HOLD for _r, c in N2.values())
|
|
and 540.0 <= cmd(N2[6500.0][0][-1])["a"] <= 560.0)
|
|
check("non-vacuous: without the fix A sits at 0 W, the quarter goes over "
|
|
"the limit and 'Load too high' fires",
|
|
all(all(cmd(r)["a"] == 0.0 for r in rows) and max(banked(c)) > 3900.0
|
|
and alarmed(c) for rows, c in N2_OLD.values()))
|
|
|
|
_off_rows, _off = run(6500.0, 2, b_off=True,
|
|
extra={"battery_modes_enabled": True})
|
|
print(f" B Off, house 6500: banked {[round(v) for v in banked(_off)]} W, "
|
|
f"last {_off_rows[-1][2]}")
|
|
check("⭐ battery modes on, B Off: the Off battery still gives before the "
|
|
"holding one (maintenance outranks a mode), the quarter lands on the "
|
|
"aim, nothing refused",
|
|
on_aim(_off) and cmd(_off_rows[-1])["b"] == 2500.0
|
|
and 540.0 <= cmd(_off_rows[-1])["a"] <= 560.0
|
|
and refused(_off_rows) == 0 and not alarmed(_off))
|
|
|
|
# ============================================================================ #
|
|
print("\n4. a month peak banked above the limit")
|
|
BANK = {n: (run(6500.0, n, month_peak=4000.0),
|
|
without_fix(lambda: run(6500.0, n, month_peak=4000.0)))
|
|
for n in (1, 2)}
|
|
for n, (new, old) in BANK.items():
|
|
print(f" N={n} house 6500, month peak 4000: banked "
|
|
f"{[round(v) for v in banked(new[1])]} W (before the fix "
|
|
f"{[round(v) for v in banked(old[1])]} W)")
|
|
check("⭐ under a month peak of 4000 W the aim is 3950 W (peak - margin), "
|
|
"at N=1 and N=2, and no 'Load too high'",
|
|
all(on_aim(new[1], 3950.0) and not alarmed(new[1])
|
|
for new, _old in BANK.values()))
|
|
check("non-vacuous: before the fix N=1 banks the whole house and N=2 banks "
|
|
"4000 W, over that aim",
|
|
min(banked(BANK[1][1][1])) >= 6499.0
|
|
and min(banked(BANK[2][1][1])) >= 3999.0)
|
|
|
|
# ============================================================================ #
|
|
print("\n5. a safety stop still wins")
|
|
|
|
|
|
def blind_at(t0, t1):
|
|
"""A's power reading is missing from `t0` to `t1`: a site safety SET."""
|
|
def before(c, t):
|
|
s84.feed(False)(c, t)
|
|
if not getattr(c.hass, "_blind", False):
|
|
real = c.hass.number
|
|
|
|
async def number(e, invert=False):
|
|
if e == "sensor.batt_a" and t0 <= c.hass.t < t1:
|
|
return None
|
|
return await real(e, invert)
|
|
c.hass.number, c.hass._blind = number, True
|
|
return before
|
|
|
|
|
|
STOP = {n: run(6500.0, n, dur=900.0, before=blind_at(300.0, 600.0))
|
|
for n in (1, 2)}
|
|
for n, (rows, _c) in STOP.items():
|
|
print(f" N={n} reading lost 300-600 s: at {rows[100][0]:.0f} s "
|
|
f"{rows[100][2]}, {rows[100][6]}; after {rows[-1][2]}")
|
|
check("⭐ with a reading gone the safety SET parks every battery at 0 W, at "
|
|
"N=1 and N=2, with nothing refused",
|
|
all(all(v == 0.0 for r in rows if 330.0 <= r[0] < 600.0
|
|
for v in cmd(r).values())
|
|
and any(r[6].startswith("safety -> 0 W") for r in rows)
|
|
and refused(rows) == 0 for rows, _c in STOP.values()))
|
|
check("...the hold gave before the stop and gives again after it",
|
|
all(cmd(rows[50])["a"] > 0.0 and cmd(rows[-1])["a"] > 0.0
|
|
for rows, _c in STOP.values()))
|
|
|
|
# ============================================================================ #
|
|
print("\n6. the level dips below the hold's target")
|
|
|
|
|
|
def draining(u, t):
|
|
"""A's level falls 1 % a minute from 99 % (set, not integrated); B 60 %."""
|
|
return max(20.0, 99.0 - t / 60.0) if u == "a" else 60.0
|
|
|
|
|
|
DIP = {n: run(6500.0, n, hold_min=30, dur=2400.0, soc=draining)
|
|
for n in (1, 2)}
|
|
for n, (rows, c) in DIP.items():
|
|
print(f" N={n}: phases {sorted({tuple(x[-1][3]) for x in rows})}, "
|
|
f"last event {[e for e in c.events if 'maintenance' in e][:1]}")
|
|
check("⭐ the level falls from 99 % to 70 % while the hold gives, and "
|
|
"the hold still ends on its 30 min clock: HOLD -> IDLE, cycle "
|
|
"COMPLETE, never back to CHARGE, at N=1 and N=2",
|
|
all(c.maints[0].phase == IDLE
|
|
and any("cycle COMPLETE" in e for e in c.events)
|
|
and {x[-1][3][0] for x in rows} == {HOLD, IDLE}
|
|
and all(x[-1][3][0] == HOLD for x in rows if x[0] < 1795.0)
|
|
for rows, c in DIP.values()))
|
|
check("...the hold gave on the way (the level really dipped)",
|
|
all(any(cmd(r)["a"] > 0.0 for r in rows if r[0] < 1795.0)
|
|
for rows, _c in DIP.values()))
|
|
|
|
# ============================================================================ #
|
|
print("\n7. the rule at its boundaries")
|
|
_c = run(1500.0, 2, dur=10.0)[1]
|
|
_c1 = run(1500.0, 1, dur=10.0)[1]
|
|
HOLDING = [MaintResult(HOLD, True, 0.0, False, []),
|
|
MaintResult(IDLE, False, None, False, [])]
|
|
|
|
|
|
def pin_after(floor, *extra, live=(A, B), results=HOLDING, c=_c):
|
|
fake = types.SimpleNamespace(inverters=c.inverters,
|
|
actuator_of=c.actuator_of,
|
|
live_units=lambda: tuple(live))
|
|
cl = [Claim.limit("supervised", P_SAFETY, -2500.0, 2500.0, actuator=A),
|
|
Claim.limit("supervised", P_SAFETY, -2500.0, 2500.0, actuator=B),
|
|
Claim.set("maintenance", P_MAINTENANCE, 0.0, HOLD, actuator=A),
|
|
Claim.limit("peak", P_PEAK, lo=floor, reason="quarter peak"),
|
|
*extra]
|
|
Controller._peak_over_hold(fake, cl, results)
|
|
return next((x.value for x in cl
|
|
if x.source == "maintenance" and x.kind == "set"), None), cl
|
|
|
|
|
|
check("⭐ at N=2 a floor under or at B's 2500 W leaves A at 0 W (never "
|
|
"charging); 0.5 W above it raises A's pin to 0.5 W",
|
|
pin_after(1000.0)[0] == 0.0 and pin_after(2500.0)[0] == 0.0
|
|
and pin_after(2500.5)[0] == 0.5)
|
|
check("⭐ B not live, contradicting itself, or refusing to give: A gives the "
|
|
"whole floor",
|
|
pin_after(2000.0, live=(A,))[0] == 2000.0
|
|
and pin_after(2000.0, Claim.limit("pack", P_SAFETY, 10.0, -10.0,
|
|
actuator=B))[0] == 2000.0
|
|
and pin_after(2000.0, Claim.limit("refused", P_SAFETY, hi=0.0,
|
|
actuator=B))[0] == 2000.0)
|
|
_probe = _ctl.PROBE_W
|
|
check("⭐ B probing: giving counts its probe (A gives the rest), charging "
|
|
"leaves A at 0 W so the site is never split both ways",
|
|
pin_after(2000.0, Claim.set("refused", P_SAFETY, _probe, "probe",
|
|
actuator=B))[0] == 2000.0 - _probe
|
|
and pin_after(2000.0, Claim.set("refused", P_SAFETY, -_probe, "probe",
|
|
actuator=B))[0] == 0.0)
|
|
_p, _cl = pin_after(4000.0, Claim.set("safety", P_SAFETY, 0.0,
|
|
"inputs missing"))
|
|
check("⭐ a site safety SET: A stays at 0 W and the site resolves to 0 W; "
|
|
"a unit safety SET on B does not stop A giving",
|
|
_p == 0.0 and resolve(_cl, units=(A, B), live={A, B}).target_w == 0.0
|
|
and pin_after(3000.0, Claim.set("failsafe", P_SAFETY, 0.0,
|
|
actuator=B))[0] == 3000.0)
|
|
check("only a HOLD: a DRAIN pin or an idle pack is not raised",
|
|
pin_after(4000.0, results=[MaintResult(DRAIN, True, 0.0, False, []),
|
|
HOLDING[1]])[0] == 0.0
|
|
and pin_after(4000.0, results=[HOLDING[1], HOLDING[1]])[0] == 0.0)
|
|
|
|
|
|
def refile(lo, results=(HOLDING[0],)):
|
|
fake = types.SimpleNamespace(inverters=_c1.inverters,
|
|
actuator_of=_c1.actuator_of)
|
|
cl = [Claim.set("maintenance", P_MAINTENANCE, 0.0, HOLD),
|
|
Claim.limit("peak", P_PEAK, lo=lo, reason="quarter peak")]
|
|
Controller._peak_over_hold(fake, cl, list(results))
|
|
return cl
|
|
|
|
|
|
check("⭐ at N=1 a floor of 0 W stays at P_PEAK, 0.5 W is refiled at "
|
|
"P_MAINTENANCE, and a blind claim (a ceiling only) is not refiled",
|
|
refile(0.0)[1].priority == P_PEAK
|
|
and refile(0.5)[1].priority == P_MAINTENANCE
|
|
and refile(float("-inf"))[1].priority == P_PEAK)
|
|
_stop = refile(3000.0) + [Claim.set("safety", P_SAFETY, 0.0,
|
|
"inputs missing")]
|
|
check("⭐ at N=1 a refiled floor stays under a safety SET: 0 W",
|
|
resolve(_stop).target_w == 0.0)
|
|
check("at N=1 outside a HOLD the claim is not touched by this rule",
|
|
refile(3000.0, (MaintResult(IDLE, False, None, False, []),))[1]
|
|
.priority == P_PEAK)
|
|
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED: " + ", ".join(fails))
|
|
sys.exit(1)
|
|
print("all SAFETY-86 checks passed")
|