"""Runnable check for SAFETY-83: the peak guard outranks the drain at N=1. `python3 test_safety83.py` At one inverter the monthly drain is the site SET at P_MAINTENANCE, above the peak floor, so the house imported what the drain left and quarters went over the peak limit. Now a peak floor above the drain rate is refiled at P_MAINTENANCE and wins. Drives the REAL Controller (tools/opt11_identity.py's harness on a P1 meter). DECISIONS/SAFETY-83-peak-over-drain.md has the record. 1. A house of 5000, 6000 and 7000 W during a drain, limits 2500 and 3500, margins 0 and 50: every quarter banks at or under the aim, the drain keeps running, and without the fix the same runs go over. 2. The drain never makes the banked quarter worse than no drain, below and above the drain rate; below it the rest is still exported. 3. The drain still ends, and the order holds: a lost reading parks the battery, the rating caps the floor without a clash. 4. Only cycles where the guard wanted more than the drain change, and N=2 is untouched. Mutations (MUTATION.md, purged copies) are listed in the pull request. """ import functools import logging import os import sys import tempfile 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 safety78_identity as s78 # noqa: E402 import safety79_identity as s79 # noqa: E402 from app.arbiter import Claim, P_MAINTENANCE, P_PEAK # noqa: E402 from app.main import Controller # noqa: E402 from app.maintenance import CHARGE, DRAIN, MaintResult # noqa: E402 fails = [] TOL = 1.0 # watts over the aim a banked quarter may show def check(name, cond): print(f" {'ok ' if cond else 'FAIL'} {name}") if not cond: fails.append(name) def run(house, cap, margin, drain_w, *, maint=True, dur=2700.0, soc=None, before=None, store=None, **extra): """N=1 on a P1 meter; returns (rows, controller, banked quarters).""" opts = dict(s79.P1, peak_cap_w=cap, peak_margin_w=margin, **extra) if maint: opts.update(maintenance_enabled=True, maintenance_start_hour=11, maintenance_discharge_w=drain_w) db = os.path.join(tempfile.mkdtemp(prefix="s83-"), "tariff.db") real = rig.Controller rig.Controller = functools.partial(real, peak_db_path=db) try: rows, c = rig.run(1, extra=opts, dur=dur, load=house if callable(house) else lambda t: house, soc=soc or (lambda u, t: 60.0), cut=lambda u, t: False, before=before or s79._feed, store=store) finally: rig.Controller = real return rows, c, {b.start.strftime("%H:%M"): b.offtake_avg_w for b in c.p1.blocks} def without_fix(fn): real = Controller._peak_over_drain Controller._peak_over_drain = lambda self, claims, results: None try: return fn() finally: Controller._peak_over_drain = real def cmd(r): return dict(r[2])["a"] # ============================================================================ # print("1. banked quarters during a drain, N=1") CASES = [(cap, drain, m, h) for cap, drain in ((2500, 2000), (3500, 2500)) for m in (0, 50) for h in (5000.0, 6000.0, 7000.0)] RES, OLD = {}, {} for k in CASES: cap, drain, m, h = k RES[k] = run(h, cap, m, drain) OLD[k] = without_fix(lambda: run(h, cap, m, drain)) print(f" limit {cap} drain {drain} margin {m:>2} house {h:.0f}: " f"banked {[round(v) for v in RES[k][2].values()]} W " f"(before the fix {[round(v) for v in OLD[k][2].values()]} W)") check("⭐ every quarter banks at or under the aim (limit - margin) plus 1 W, " "for a 5000, 6000 and 7000 W house, limits 2500 and 3500, margins 0 " "and 50", all(len(b) >= 2 and max(b.values()) <= k[0] - k[2] + TOL for k, (_r, _c, b) in RES.items())) check("⭐ the drain keeps running: the phase is drain on every run and the " "battery gives at least the drain rate on every cycle", all(c.maints[0].phase == DRAIN and all(cmd(r) >= k[1] for r in rows) for k, (rows, c, _b) in RES.items())) check("...and where the house is over drain + aim, the battery gives more " "than the drain rate, bound by the peak floor", all(any(cmd(r) > k[1] and "limited by peak" in r[6] for r in rows) for k, (rows, _c, _b) in RES.items() if k[3] - k[1] > k[0] - k[2])) check("non-vacuous: without the fix the same runs bank over the limit " "wherever the house is above drain + aim (the reported defect: 3000 W " "at a 5000 W house against 2500 W)", all(max(b.values()) > k[0] + 100.0 for k, (_r, _c, b) in OLD.items() if k[3] - k[1] > k[0]) and round(OLD[(2500, 2000, 50, 5000.0)][2]["11:15"]) == 3000) # ============================================================================ # print("\n2. the drain never makes the banked quarter worse, N=1") CAP, DRAIN_W, MARGIN = 3500, 2500, 50 # the owner's settings AIM = CAP - MARGIN WORSE = {} for h in (1200.0, 3000.0, 5000.0, 6500.0, 7000.0): _rows, _c, b = run(h, CAP, MARGIN, DRAIN_W) _r0, _c0, b0 = run(h, CAP, MARGIN, DRAIN_W, maint=False) WORSE[h] = (b, b0, _rows) print(f" house {h:.0f}: with the drain {[round(v) for v in b.values()]}" f" W, without {[round(v) for v in b0.values()]} W") check("⭐ below and above the drain rate, every quarter with the drain banks " "no more than the larger of the aim and the same quarter without it", all(b[q] <= max(AIM, b0[q]) + TOL for b, b0, _r in WORSE.values() for q in b)) _r12 = WORSE[1200.0][2] check("below the drain rate the battery gives the drain rate and exports the " "rest (owner, 2026-09-29): 2500 W against a 1200 W house", all(cmd(r) == DRAIN_W for r in _r12[1:])) # ============================================================================ # print("\n3. the drain ends, and the order holds") _e_rows, _e, _eb = run(6500.0, CAP, MARGIN, DRAIN_W, dur=1200.0, soc=lambda u, t: max(5.0, 60.0 - t / 20.0)) check("⭐ with the guard binding, the drain still reaches its floor and moves " "to the charge phase, with no abort", _e.maints[0].phase == CHARGE and not any("abort" in e.lower() for e in _e.events)) def blind_at(t0): """The battery power reading goes missing at `t0`: a safety SET.""" def before(c, t): s79._feed(c, t) if t >= t0 and not getattr(c.hass, "_blind", False): real = c.hass.number async def number(e, invert=False): return None if e == "sensor.batt_a" else await real(e, invert) c.hass.number, c.hass._blind = number, True return before _s_rows, _s, _sb = run(6500.0, CAP, MARGIN, DRAIN_W, dur=600.0, before=blind_at(300.0)) print(f" reading lost at 300 s: last {cmd(_s_rows[-1]):+.0f} W, " f"{_s_rows[-1][6]}") check("⭐ a safety SET outranks the refiled peak floor: before the reading " "goes the battery gives above the drain rate, after it it is at 0 W", any(cmd(r) > DRAIN_W for r in _s_rows if r[0] < 300.0) and all(cmd(r) == 0.0 for r in _s_rows if r[0] >= 400.0) and _s_rows[-1][6].startswith("safety -> 0 W")) _m_rows, _m, _mb = run(7000.0, 2500, 50, 2000, dur=900.0, max_w=3000) print(f" max_w 3000, house 7000: last {cmd(_m_rows[-1]):+.0f} W, " f"{_m_rows[-1][6]}") check("⭐ the rating caps the floor without a clash: at max_w 3000 the " "battery gives 3000 W, never 0 W, and nothing contradicts", all(cmd(r) >= 2000.0 for r in _m_rows) and cmd(_m_rows[-1]) == 3000.0 and not any("CONTRADICTION" in (r[6] or "") for r in _m_rows)) # ============================================================================ # print("\n4. only the cycles where the guard wanted more change") def promoted(row): return [c for c in row[-1][0] if c[0] == "peak"] _id = {k: s78.run(**v)[0] for k, v in s78.SCENARIOS.items()} _bad = [] for k, rows in _id.items(): for r in rows: claims, guard = r[-1][0], r[-1][1] rate = [c[3] for c in claims if c[0] == "maintenance" and c[2] == "set"] pk = promoted(r) if k.startswith("n2"): ok = all(c[1] == P_PEAK for c in pk) elif s78.wanted_more(r): ok = ([c[1] for c in pk] == [P_MAINTENANCE] and r[1] == guard[1] and "limited by peak" in r[6]) else: ok = (all(c[1] == P_PEAK for c in pk) and (not rate or r[1] == rate[0])) if not ok: _bad.append((k, r[0], r[1], r[6], pk)) _bad1 = [b for b in _bad if b[0].startswith("n1")] _bad2 = [b for b in _bad if b[0].startswith("n2")] _more = {k: sum(map(s78.wanted_more, rows)) for k, rows in _id.items()} print(f" cycles with the guard above the drain: {_more}; bad {_bad[:3]}") check("⭐ at N=1, on every cycle where the guard asks for more than the drain " "the peak claim is refiled at P_MAINTENANCE and the target is its floor; " "on every other cycle the peak claim stays at P_PEAK and the target is " "the drain rate", not _bad1 and all(_more[k] > 50 for k in _id if k.startswith("n1"))) check("⭐ at N=2 the peak claim is never refiled (the SAFETY-78 floor does " "that job), and test_safety78.py part 5 holds N=2 to the digests of " "release/1.0 35cdffe", not _bad2) # The rule itself, at the boundary: a floor AT the drain rate changes nothing. _one = types.SimpleNamespace(inverters=[None]) _edge = {} for lo in (1999.5, 2000.0, 2000.5): cl = [Claim.set("maintenance", P_MAINTENANCE, 2000.0, DRAIN), Claim.limit("peak", P_PEAK, lo=lo, reason="quarter peak")] Controller._peak_over_drain(_one, cl, [MaintResult(DRAIN, True, 2000.0, False, [])]) _edge[lo] = cl[1].priority check("at the boundary: a floor at or below the 2000 W drain stays at P_PEAK, " "a floor 0.5 W above it is refiled at P_MAINTENANCE", _edge == {1999.5: P_PEAK, 2000.0: P_PEAK, 2000.5: P_MAINTENANCE}) def store_hold(): st = rig._Store() st.data["phase"], st.data["phase_started"] = "hold", rig.BASE.isoformat() return st _h_rows, _h, _hb = run(6000.0, 2500, 50, 2000, dur=600.0, store=store_hold()) check("a maintenance HOLD at N=1 is not changed (DECISIONS/SAFETY-83, not " "changed): the battery stays at 0 W with the house over the limit", _h.maints[0].phase == "hold" and _h.peak_guard.engaged and _h.peak_guard.floor_w > 0.0 and all(cmd(r) == 0.0 for r in _h_rows)) print() if fails: print(f"{len(fails)} FAILED: {', '.join(fails)}") sys.exit(1) print("all checks passed")