"""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 (CHARGE, DRAIN, HOLD, IDLE, # noqa: E402 MaintResult) 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? Its quarter stamp too, since ALLOCATOR rows can push its row out of the 40-row log (SAFETY-85).""" return c._peak_alarm_block is not None or 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)) def noisy(t): """A 6500 W house that moves up to 150 W from one second to the next.""" return 6500.0 + ((int(t) * 7919) % 301) - 150.0 def no_ceiling(fn): """Run with the pin raised but no `hold` site ceiling (review check).""" real = Controller._peak_over_hold def pin_only(self, claims, results): real(self, claims, results) Controller._peak_over_hold = pin_only try: return fn() finally: Controller._peak_over_hold = real def rows_logged(c, word): return sum(1 for e in c.events if word in e) def no_bound(fn): """Run without SAFETY-85's loop bound (modes are off here).""" real = Controller.loop_bound Controller.loop_bound = lambda self, claims: None try: return fn() finally: Controller.loop_bound = real _nz_rows, _nz = run(noisy, 2) # SAFETY-85's loop bound also keeps the loop inside what the batteries can # give, so the flood needs both the ceiling and the bound taken out. _nz_old = no_bound(lambda: no_ceiling(lambda: run(noisy, 2)))[1] _nz_nc = no_ceiling(lambda: run(noisy, 2))[1] print(f" noisy house: ALLOCATOR rows {rows_logged(_nz, 'ALLOCATOR')}, " f"clamp rows {rows_logged(_nz, 'OUTPUT_CLAMP')} (without the site " f"ceiling {rows_logged(_nz_old, 'ALLOCATOR')} and " f"{rows_logged(_nz_old, 'OUTPUT_CLAMP')})") check("⭐ while A gives, the site target is the guard's floor (the `hold` " "ceiling), so a noisy house does not fill the action log: no ALLOCATOR " "row and at most one clamp row", on_aim(_nz) and all(abs(r[1] - r[-1][1][1]) < 1e-6 for r in _nz_rows if cmd(r)["a"] > 0.0) and rows_logged(_nz, "ALLOCATOR") == 0 and rows_logged(_nz, "OUTPUT_CLAMP") <= 1) check("non-vacuous: without the ceiling and SAFETY-85's bound the loop asks " "5000 W and the ALLOCATOR and clamp rows fill the 40-row log", rows_logged(_nz_old, "ALLOCATOR") >= 10 and len(_nz_old.events) == 40) check("...and either one alone keeps it clean (without the ceiling: " f"{rows_logged(_nz_nc, 'ALLOCATOR')} ALLOCATOR rows)", on_aim(_nz_nc) and rows_logged(_nz_nc, "ALLOCATOR") == 0 and rows_logged(_nz_nc, "OUTPUT_CLAMP") <= 1) # ============================================================================ # 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), _peak_reserve=frozenset) 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] ceiling = Controller._peak_over_hold(fake, cl, results) pin_after.ceiling = ceiling 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) _p, _cl = pin_after(2600.0) _hi = pin_after.ceiling _cl += [Claim.limit("hold", P_MAINTENANCE, hi=_hi, actuator="site"), Claim.set("loop", 10, 5000.0, "the loop asks for the whole house")] pin_after(2500.0) check("⭐ at N=2 the `hold` site ceiling is the floor itself, only when A " "gives, and a 5000 W loop ask resolves to that floor", _hi == 2600.0 and pin_after.ceiling is None and resolve(_cl, units=(A, B), live={A, B}).target_w == 2600.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 (idle, or a CHARGE) the claim is not touched " "by this rule", refile(3000.0, (MaintResult(IDLE, False, None, False, []),))[1] .priority == P_PEAK and refile(3000.0, (MaintResult(CHARGE, False, None, True, []),))[1] .priority == P_PEAK) print() if fails: print(f"{len(fails)} FAILED: " + ", ".join(fails)) sys.exit(1) print("all SAFETY-86 checks passed")