"""Runnable check for SAFETY-78: a maintenance drain at N=2 no longer stalls. `python3 test_safety78.py` At two inverters the drain pins its battery at the drain rate and the site target is split around the pin. Below the pin the other battery had to charge, the allocator refused the opposite directions, the site sat at 0 W and the drain never moved (OPT-11 P1 F1, UX-13 P3 check 5, OPT-11 P3 F4). Now a site `lo` (`drain`, P_MAINTENANCE) keeps the site target at or above what the pin places. Drives the REAL Controller (tools/opt11_identity.py's harness). DECISIONS/SAFETY-78-drain-at-two.md has the record. 1. N=2, house below, at and above the drain rate, a solar surplus: the drain runs at its rate, no opposite split, no refusal, no log spam, and it ends. 2. A drain that cannot give power waits with ONE log line. 3. Battery modes on, a Cheap hours battery in a cheap run during the drain: no grid charge, and its card says why. 4. The peak guard binds during a drain, and the arbiter's order holds. 5. The N=1 and N=2 digests hold, and the harness bites. 6. A seeded sweep of the floor against the loop, trades, the peak floor, a safety stop, refusals and an Off battery. Mutations (MUTATION.md, purged copies, graded over this file and the named tests) are listed in the pull request. """ import functools import logging import os import random import sys import tempfile 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 opt11p3_identity as p3 # noqa: E402 import safety78_identity as s78 # noqa: E402 import safety79_identity as s79 # noqa: E402 from app import allocator, arbitrage, homeview, packmode # noqa: E402 from app.arbiter import (Claim, P_ARBITRAGE, P_LOOP, P_MAINTENANCE, # noqa P_MODE, P_PEAK, P_SAFETY, pre_resolve, resolve) from app.main import Controller # noqa: E402 from app.maintenance import CHARGE, DRAIN # noqa: E402 from app.store import unit_key # noqa: E402 fails = [] def check(name, cond): print(f" {'ok ' if cond else 'FAIL'} {name}") if not cond: fails.append(name) M = {"maintenance_enabled": True, "maintenance_start_hour": 11} A, B = "inverter-a", "inverter-b" def cmd(r): return dict(r[2]) def refused(rows): return sum(1 for r in rows if r[7] and "REFUSED" in r[7]) def opposite(rows): return sum(1 for r in rows if cmd(r)["a"] * cmd(r)["b"] < 0) def drain_run(house, dur=600.0, soc=lambda u, t: 50.0, cut=None, **kw): return rig.run(2, extra=dict(M, **kw.pop("extra", {})), dur=dur, load=lambda t: house(t) if callable(house) else house, soc=soc, cut=cut or (lambda u, t: False), **kw) def with_peak_db(fn): """Run `fn()` with the Controller's peak store in a temporary file.""" db = os.path.join(tempfile.mkdtemp(prefix="s78t-"), "tariff.db") real = rig.Controller rig.Controller = functools.partial(real, peak_db_path=db) try: return fn() finally: rig.Controller = real # ============================================================================ # print("1. N=2: the drain runs whatever the house uses") RUNS = {h: drain_run(h) for h in (1200.0, 2000.0, 3000.0, -1500.0)} for h, (rows, c) in RUNS.items(): print(f" house {h:+.0f} W: last A {cmd(rows[-1])['a']:+.0f} W, B " f"{cmd(rows[-1])['b']:+.0f} W, grid {h - sum(cmd(rows[-1]).values()):+.0f}" f" W; {rows[-1][6]}; refusals {refused(rows)}") check("⭐ below, at and above the drain rate and under a solar surplus, " "battery A is commanded its 2000 W drain on EVERY cycle, so the drain " "moves (it used to sit at 0 W from the first cycle)", all(cmd(r)["a"] == 2000.0 for rows, _c in RUNS.values() for r in rows) and all(c.maints[0].phase == DRAIN for _r, c in RUNS.values())) check("⭐ no cycle splits the site both ways and the allocator refuses " "nothing: battery B never charges while A drains", all(opposite(rows) == 0 and refused(rows) == 0 and all(cmd(r)["b"] >= 0.0 for r in rows) for rows, _c in RUNS.values())) check("...and nothing is logged per cycle: no ARBITER line in the action log", not any("ARBITER" in e for _r, c in RUNS.values() for e in c.events)) _r3 = RUNS[3000.0][0] check("above the drain rate, B covers the rest of the house in the SAME " "direction and the grid settles at the loop's aim", 990.0 <= cmd(_r3[-1])["b"] <= 1030.0 and abs(3000.0 - sum(cmd(_r3[-1]).values()) + 10.0) <= 30.0) check("at or below the drain rate, B is held at 0 W and A exports the rest, " "as the drain does at one inverter ('exports the surplus')", all(cmd(RUNS[h][0][-1])["b"] == 0.0 for h in (1200.0, 2000.0, -1500.0)) and all(RUNS[h][0][-1][6].startswith("loop -> 2000 W") for h in (1200.0, 2000.0, -1500.0))) check("...and where the loop alone would ask less, the Why names the drain", "limited by drain" in RUNS[1200.0][0][-1][6]) # The drain ends: A's level falls to the floor, and the cycle moves on. _end_rows, _end = drain_run(1200.0, dur=500.0, soc=lambda u, t: max(5.0, 50.0 - t / 10.0) if u == "a" else 50.0) check("⭐ and the drain ENDS: A reaches the 11 % floor and the cycle moves " "to its charge phase, with no refusal on the way", _end.maints[0].phase == CHARGE and refused(_end_rows) == 0) # The check bites: without the floor, the same site stalls as it did. _real = Controller._drain_floor Controller._drain_floor = lambda self, claims, results: None try: _bite, _bc = drain_run(1200.0, dur=120.0) finally: Controller._drain_floor = _real print(f" without the drain floor: refusals {refused(_bite)}/{len(_bite)}, " f"last A {cmd(_bite[-1])['a']:+.0f} W") check("non-vacuous: with the floor taken out, the same fixture refuses every " "cycle and parks A at 0 W (the reported defect)", refused(_bite) == len(_bite) and cmd(_bite[-1])["a"] == 0.0 and any("ARBITER" in e for e in _bc.events)) # ============================================================================ # print("\n2. a drain that cannot give power waits, with one log line") _w_rows, _w = drain_run(1500.0, cut=lambda u, t: u == "a") _waits = [e for e in _w.events if "cannot give power now" in e] print(f" A refuses to give: {_waits}; last {_w_rows[-1][2]}") check("⭐ A refuses to give power (SAFETY-73): the cycle waits, the line is " "logged ONCE over 120 cycles, and nothing is refused", len(_waits) == 1 and _w.inverters[0].refused == 1 and refused(_w_rows) == 0 and _w.maints[0].phase == DRAIN) check("...and B covers the house on its own meanwhile", 1400.0 <= cmd(_w_rows[-1])["b"] <= 1600.0) # PR #154 review: one line per episode, across a probe and a second episode. _pr_rows, _pr = drain_run(1500.0, dur=1300.0, cut=lambda u, t: u == "a") check("the SAFETY-73 probe (A pinned at 200 W, not by maintenance) does not " "end the wait: still ONE line over 1300 s with a probe in it", any(cmd(r)["a"] == 200.0 for r in _pr_rows) and sum("cannot give power now" in e for e in _pr.events) == 1) _e2_rows, _e2 = drain_run(1500.0, dur=2600.0, cut=lambda u, t: u == "a" and ( t < 600.0 or t >= 1400.0)) check("...and a second episode, after the probe ends the first, logs again", sum("cannot give power now" in e for e in _e2.events) == 2) _cw_rows, _cw = drain_run(-1500.0, dur=300.0, cut=lambda u, t: u == "a") _cw_cards = homeview.battery_rows(_cw, False, True) check("while A waits, B may take the solar surplus, so its card does not " "say 'This one does not charge'", cmd(_cw_rows[-1])["b"] < 0.0 and _cw_cards[1]["why_key"] != "other_cycle") def reject_a(c, t): """Every write to A is rejected, so A's link goes stale (not live).""" if not getattr(c.hass, "_rej", False): real = c.hass.set_number async def set_number(e, v): return False if e == "number.sp_a" else await real(e, v) c.hass.set_number, c.hass._rej = set_number, True _l_rows, _l = drain_run(1500.0, dur=400.0, before=reject_a) print(f" A's writes rejected: {_l_rows[-1][2]}, {_l_rows[-1][7][-40:]}") check("⭐ a draining battery that is dropped (stale link) adds no floor: B " "covers the 1500 W house, not the 2000 W drain rate, and the wait is " "logged once", "dropped inverter-a" in _l_rows[-1][7] and cmd(_l_rows[-1])["b"] == 1500.0 and sum("cannot give power now" in e for e in _l.events) == 1) # ============================================================================ # print("\n3. battery modes on: no grid charge during the drain") def cheap_b(): s = rig._Store() s.data[unit_key(A, packmode.MODE_KEY)] = packmode.SOLAR_FIRST s.data[unit_key(B, packmode.MODE_KEY)] = packmode.CHEAP_HOURS return s def modes_run(maint): extra = dict(s79.P1, **p3.ARB, battery_modes_enabled=True, **(M if maint else {})) ticks = [] def on_cycle(c, t): ticks.append(c.mode_tick) rows, c = with_peak_db(lambda: rig.run( 2, extra=extra, dur=1800.0, load=lambda t: 800.0, soc=lambda u, t: 50.0, cut=lambda u, t: False, before=p3._feed, on_cycle=on_cycle, store=cheap_b())) cheap = [(r, k) for r, k in zip(rows, ticks) if 900.0 <= r[0] < 1800.0] return rows, c, cheap _base_rows, _base_c, _base_cheap = modes_run(False) _m_rows, _m_c, _m_cheap = modes_run(True) check("baseline: with no drain, B (Cheap hours only) charges from the grid " "in the cheap run, so the fixture reaches the charge", any(cmd(r)["b"] < 0.0 for r, _k in _base_cheap) and all(k.roles[B] == "charge" for _r, k in _base_cheap)) check("⭐ with A draining, B never charges in that run: its role is not " "'charge', no trade is filed, and the modes say why ('cycle')", all(cmd(r)["b"] >= 0.0 and k.roles[B] != "charge" and k.action is None and k.why == "cycle" for r, k in _m_cheap) and _m_c.maints[0].phase == DRAIN) check("...and the drain runs through the run: A at 2000 W, nothing refused", all(cmd(r)["a"] == 2000.0 for r in _m_rows) and refused(_m_rows) == 0) # Battery modes off: the site trade (`arbitrage.decide()`) buys in the cheap # period. It is a site SET at P_ARBITRAGE, below the drain floor. _t_trades = [] _t_rows, _t_c = with_peak_db(lambda: rig.run( 2, extra=dict(s79.P1, **p3.ARB, **M), dur=1800.0, load=lambda t: 800.0, soc=lambda u, t: 50.0, cut=lambda u, t: False, before=p3._feed, on_cycle=lambda c, t: _t_trades.append(c.arb and c.arb.action))) print(f" modes off, trades seen: {sorted({str(a) for a in _t_trades})}") check("⭐ battery modes off: the site trade still asks to buy in the cheap " "period, and the drain floor keeps it from charging B against the drain: " "A drains at 2000 W, B never charges, nothing is refused", arbitrage.CHARGE in _t_trades and all(cmd(r)["a"] == 2000.0 and cmd(r)["b"] >= 0.0 for r in _t_rows) and refused(_t_rows) == 0) _cards = homeview.battery_rows(_m_c, False, True) print(f" cards: {[(b['name'], b['now'], b['why']) for b in _cards]}") check("⭐ B's card tells the truth: it names the draining battery and says " "it does not charge (it said 'charges from the grid' while resting)", _cards[1]["why_key"] == "other_cycle" and A in _cards[1]["why"] and _cards[0]["why_key"] == "cycle_mode") # ============================================================================ # print("\n4. the peak guard during a drain, and the order of the claims") def off_b(): s = rig._Store() s.data[unit_key(B, packmode.USE_KEY)] = False return s _p_rows, _p = with_peak_db(lambda: rig.run( 2, extra=dict(s79.P1, battery_modes_enabled=True, **M), dur=600.0, load=lambda t: 6000.0, soc=lambda u, t: 60.0, cut=lambda u, t: False, before=s79._feed, store=off_b())) print(f" house 6000 W, limit {s79.CAP_W} W, B Off: {_p_rows[-1][2]}, " f"{_p_rows[-1][6]}") check("⭐ the peak guard binds during the drain: the site goes above the " "drain to the guard's floor, A keeps draining at 2000 W and B (Off) " "gives only the shortfall, in the same direction", _p.peak_guard.engaged and _p_rows[-1][1] >= _p.peak_guard.floor_w > 2000.0 and all(cmd(r)["a"] == 2000.0 and cmd(r)["b"] >= 0.0 for r in _p_rows) and "limited by peak" in _p_rows[-1][6] and refused(_p_rows) == 0) def blind_at(t0): """The site meter stops answering at `t0`: a safety SET at P_SAFETY.""" def before(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.grid" else await real(e, invert) c.hass.number, c.hass._blind = number, True return before _s_rows, _s = drain_run(3000.0, dur=300.0, before=blind_at(150.0)) print(f" meter lost at 150 s: {_s_rows[-1][2]}, {_s_rows[-1][6]}") check("⭐ a safety stop outranks the drain at N=2: with the meter gone the " "safety SET parks both batteries at 0 W, and the drain ran before", cmd(_s_rows[20])["a"] == 2000.0 and all(cmd(r) == {"a": 0.0, "b": 0.0} for r in _s_rows if r[0] >= 175.0) and _s_rows[-1][6].startswith("safety -> 0 W")) UNITS = (A, B) _drain = [Claim.set("maintenance", P_MAINTENANCE, 2000.0, "drain", actuator=A), Claim.limit("drain", P_MAINTENANCE, lo=2000.0, reason="drain")] _order = { "loop under the drain": ([Claim.set("loop", P_LOOP, 500.0)], 2000.0), "a grid charge (trade)": ([Claim.set("arbitrage", P_ARBITRAGE, -2500.0)], 2000.0), "a sale bigger than the drain": ([Claim.set("arbitrage", P_ARBITRAGE, 3000.0)], 3000.0), "a peak floor above the drain": ([Claim.set("loop", P_LOOP, 500.0), Claim.limit("peak", P_PEAK, lo=3500.0)], 3500.0), "a safety stop": ([Claim.set("loop", P_LOOP, 500.0), Claim.set("safety", P_SAFETY, 0.0, "stopped")], 0.0), } _got = {} for k, (extra, want) in _order.items(): r = resolve(_drain + extra, units=UNITS) _got[k] = (r.target_w, dict(r.shares), r.allocated) print(f" {k}: target {r.target_w:+.0f} W, shares {dict(r.shares)}") check("⭐ the order holds: a safety stop parks everything; the peak floor " "above the drain wins; a sale above it runs; a grid charge and the " "loop below it are lifted to it; never an opposite split", all(_got[k][0] == want and _got[k][2] and not (min(_got[k][1].values()) < 0 < max(_got[k][1].values())) for k, (_e, want) in _order.items()) and _got["a safety stop"][1] == {A: 0, B: 0} and _got["a grid charge (trade)"][1] == {A: 2000, B: 0}) # ============================================================================ # print("\n5. the N=1 and N=2 digests") # Printed by `python3 tools/safety78_identity.py`, goodwe-addon-checks:1, # 2026-09-29. 480 cycles each, draining throughout, every claim and the # newest action-log line hashed. N=2: the same on release/1.0 35cdffe. N=1: # re-pinned by SAFETY-83, whose `--diff` against 35cdffe changes only cycles # where the peak guard asked for more than the drain (test_safety83.py). # SAFETY-85 re-pinned two: n1_drain_owner loses its false 'not keeping up' # rows (commands unchanged); n2_drain's loop is bounded while A drains # (reasons and the loop claim change, commands unchanged). PINNED = { "n1_drain": "b0f9ef44149163bc0acece537fe047871c00f0d9ea213e7a7fc8efc4ad3aa500", "n1_drain_modes": "95ab82426e23bce308db3edebe18b1ab37d956952e9f521adce37150ce87ee8a", "n1_drain_owner": "607c2583c7ae4bdafb90d780b3414c62d76c430ca67798cd117d1a46f3bb40ae", "n2_drain": "5806a6f457b7baf8dfb44daddced502f1939c114e15b84029dbb2659be8c8b0f", "n2_drain_owner": "05750347f1829c249f99154613f068f442a59ef59f409093907f6d0add78dd4c", } _now = s78.digests() for k in PINNED: print(f" {k} {'same' if _now[k] == PINNED[k] else 'DIFFERS'} " f"{_now[k][:16]}") check("⭐ a drain under a house below, at and above its rate, a surplus and a " "load over the peak limit, at N=1 (modes off and on) and N=2, matches " "the pinned digests: commands, reasons, roles, trades and every claim", _now == PINNED) Controller._drain_floor = lambda self, claims, results: ( 2000.0 if any(r.phase == DRAIN for r in results) else None) try: _leak = s78.digest(s78.run(**s78.SCENARIOS["n1_drain"], dur=60.0)[0]) _same = s78.digest(s78.run(**s78.SCENARIOS["n1_drain"], dur=60.0)[0]) finally: Controller._drain_floor = _real _ref = s78.digest(s78.run(**s78.SCENARIOS["n1_drain"], dur=60.0)[0]) check("non-vacuous: filing the drain floor at N=1 (which moves no watt " "there) moves the digest, because the harness hashes the claims", _leak != _ref and _leak == _same) # ============================================================================ # print("\n6. the floor against every other claim (PR #154 review, seeded sweep)") _rnd = random.Random(78) _bad, _peak_above, _off_gave = [], 0, 0 for _i in range(20000): ma, mb = _rnd.choice((2500., 3000., 5000.)), _rnd.choice((1500., 2500.)) ua = [Claim.limit("supervised", P_SAFETY, -ma, ma, actuator=A), Claim.set("maintenance", P_MAINTENANCE, _rnd.choice((500., 2000., 6000.)), "drain", actuator=A)] if _rnd.random() < .2: ua.append(Claim.limit("pack", P_SAFETY, -ma, _rnd.choice((0., 800.)), actuator=A)) ub = [Claim.limit("supervised", P_SAFETY, -mb, mb, actuator=B)] _offb = _rnd.random() < .4 if _offb: ub.append(Claim.limit("mode", P_MODE, 0., 0., "off", actuator=B)) if _rnd.random() < .15: ub.append(Claim.limit("refused", P_SAFETY, -mb, 0., actuator=B)) pin = allocator.pin_w(pre_resolve(ua)) site = [Claim.set("loop", P_LOOP, _rnd.uniform(-6000., 8000.))] if pin is not None and pin > 0.0: site.append(Claim.limit("drain", P_MAINTENANCE, lo=pin)) if _rnd.random() < .3: site.append(Claim.set("arbitrage", P_ARBITRAGE, _rnd.choice((-3000., 1000., 4000.)))) pk = _rnd.uniform(0., 8000.) if _rnd.random() < .4 else None if pk is not None: site.append(Claim.limit("peak", P_PEAK, lo=pk)) stop = _rnd.random() < .05 if stop: site.append(Claim.set("safety", P_SAFETY, 0.0, "stopped")) r = resolve(ua + ub + site, units=UNITS) s = dict(r.shares) why = [] if min(s.values()) < 0 < max(s.values()) or r.contradiction: why.append("opposite or contradiction") if stop and (r.target_w != 0.0 or any(s.values())): why.append("stop not parked") if not stop and pin is not None and pin > 0.0: if r.target_w < pin or s[B] < 0: why.append("below the floor, or B charges") if pk is not None and pk > pin: _peak_above += 1 if r.allocated and r.target_w < min(pk, pin + mb): why.append("peak floor not honoured") if _offb and s[B] > 0: _off_gave += 1 if not (pk is not None and pk > pin): why.append("Off B used for the drain") if why: _bad.append((_i, why, r.target_w, s, pin, pk)) print(f" {_bad[:3]}; peak above the drain {_peak_above}, Off B gave for " f"the peak {_off_gave}") check("⭐ over 20000 seeded claim lists: never an opposite split or a " "contradiction; a safety stop parks both; the site never goes below " "the pin and B never charges; a peak floor above the drain is met; an " "Off battery gives only for the peak floor, never for the drain", not _bad and _peak_above > 1000 and _off_gave > 100) print() if fails: print(f"{len(fails)} FAILED: {', '.join(fails)}") sys.exit(1) print("all checks passed")