"""Runnable check for SAFETY-52. `python3 test_refusal.py` A battery that refuses to give power (the BMS cut-out) used to leave the command standing at its last ask for hours: +5000 W for 11.87 h on 2026-09-06, +1940 W for 3 h 55 m on the bus on 2026-09-19. Now a refused direction is held at 0 W, a 200 W probe asks every 15 minutes, and the refusal ends on charge, on an answered probe, on stop/start and when another writer takes the wire. DECISIONS/SAFETY-52 has the record and the replay figures. Mutating the modules this file grades? Purge `__pycache__` first, or run `tools/mutation_run.sh`, which does - see MUTATION.md. Parts 1-5 drive the REAL `Controller` (read_inputs + cycle, the real SAFETY-37/57 gate) against a scripted battery whose reading stands at exactly -4 W for any request >= -4 W at the cut-out, and follows otherwise. Time is the `now_mono` handed to `read_inputs()`, so 12 hours run in seconds. `OLD` runs the same fixture with `refused=None` forced into `compute()` - release/1.0's law - so every "new" figure is shown beside the one it replaces. Part 4 also runs `compute()` alone, as the bus saw 09-19 before the gate existed. Mutations, each on a copy of this tree purged of `__pycache__`, graded with `python3 -B` over test_refusal, test_control, test_clamp, test_telemetry and test_messages (MUTATION.md). Reds are failing checks, on this tree: # mutation reds -- ------------------------------------------------------- --------------- R1 control.py: the engage `if` -> `if False and ...` 8 here + 1 clamp (release/1.0's behaviour) R2 control.py: the exit on the other direction removed 1 R3 control.py: the exit on an answered probe removed 2 R4 control.py: `PROBE_W = 300.0`, above the gate's 250 W 4 R5 control.py: the hold keeps the old target, not 0 W 4 R6 main.py: `_refusal_from_stale()` returns at once 1 R7 main.py: another writer no longer ends the refusal 1 R8 main.py: stop/start no longer ends the refusal 1 R9 main.py: the probe window never opens 4 """ import asyncio import sys from datetime import datetime, timedelta, timezone from app import control, freshness from app.control import PROBE_EVERY_S, PROBE_S, PROBE_W, REFUSED, PROBING from app.main import Controller, why_key from app.arbiter import Claim from app import messages fails = [] def check(name, cond): print(f" {'ok ' if cond else 'FAIL'} {name}") if not cond: fails.append(name) LIVE = {"max_w": 5000.0, "gain": 0.6, "slew_w": 400.0, "target_grid_w": -10.0, "deadband_w": 15, "step_w": 10} TICK = 5.0 # the meter's measured median tick (replay_recorded_loop) CUT_W = -4.0 # what the register reads at the cut-out, 2026-09-06 H = 3600.0 BASE = datetime(2026, 9, 6, 19, 30, tzinfo=timezone.utc) class _Store: def __init__(self): self.data = {} def set(self, *a): pass def get_time(self, *a): return None class _Pub: enabled = False def publish(self, values): pass def close(self): pass class _Hass: """House load, and a battery that follows unless `cut(t)` says it refuses. At the cut-out the reading is exactly CUT_W for any request >= CUT_W (a flat register, which is what arms or does not arm the stale-reading gate). A charge request is followed: the pack accepts charge at the floor. `cap(t)` limits discharge while not cut (the 09-06 pack gave 4570 W at 10 %). """ def __init__(self, load, cut, cap=lambda t: 1e9): self.load, self.cut, self.cap = load, cut, cap self.t = 0.0 self.cmd = 0.0 self.writes = [] def batt(self): if self.cut(self.t) and self.cmd >= CUT_W: return CUT_W return float(round(min(self.cmd, self.cap(self.t)))) async def limits(self, entity): return (-LIVE["max_w"], LIVE["max_w"], 1.0) async def number(self, entity, invert=False): b = self.batt() return {"sensor.grid": self.load(self.t) - b, "sensor.soc": 9.0, "sensor.batt": b}.get(entity) async def set_number(self, entity, value): self.writes.append((self.t, float(value))) self.cmd = float(value) return True async def call(self, *a): return True def run(load, cut, dur, cap=lambda t: 1e9, old=False, events=None): """Real cycles every TICK seconds of `now_mono`. `events[t] = fn(c)` pokes the controller (stop, start, another writer) at that instant. Returns [(t, grid, batt, setpoint, i_w, loop_reason, batt_refused)], the number of gate trips, and the controller.""" async def go(): hass = _Hass(load, cut, cap) c = Controller(dict( {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid", "soc_entity": "sensor.soc", "batt_entity": "sensor.batt", "meter_source": "off", "heartbeat_s": 10, "stale_input_s": 15}, **LIVE), hass, _Store(), _Pub(), stop=asyncio.Event()) c.auto = True c.dev_min, c.dev_max = -LIVE["max_w"], LIVE["max_w"] import app.main as m real, real_dt = m.compute, m.datetime clock = [0.0] class _DT(real_dt): """`cycle()`'s wall clock follows `now_mono`, so the 15 s input hold and the gate's release run in simulated seconds too.""" @classmethod def now(cls, tz=None): return BASE + timedelta(seconds=clock[0]) m.datetime = _DT if old: m.compute = lambda *a, **k: real(*a, **dict(k, refused=None)) rows, trips, was = [], 0, False try: t = 0.0 while t < dur: hass.t = clock[0] = t for et, fn in list((events or {}).items()): if et <= t: fn(c) del events[et] await c.read_inputs(now_mono=t) stale = c.inverters[0].batt_fresh.stale trips += stale and not was was = stale await c.cycle() rows.append((t, hass.load(t) - hass.batt(), hass.batt(), hass.cmd, c.i_w, c.loop_reason, c.batt_refused)) t += TICK finally: m.compute, m.datetime = real, real_dt return rows, trips, c return asyncio.run(go()) def at(rows, t): return min(rows, key=lambda r: abs(r[0] - t)) def longest_hold_h(rows, floor_w=250.0, until=None): """Longest run of one identical request at or above `floor_w`, in hours.""" best = cur = 0.0 last = None for r in rows: if until is not None and r[0] >= until: break cur = cur + TICK if (r[3] == last and abs(r[3]) >= floor_w) else 0.0 last = r[3] best = max(best, cur) return best / H # ============================================================================== print("1. the 09-06 shape: a load step takes the request to 5000 W, the battery " "cuts out, the load ends") # ============================================================================== def load906(t): return 1020 if t < 60 else 4800 if t < 162 else 940 if t < 11.9 * H else -600 def cut906(t): return t >= 132 def cap906(t): return 4570.0 old906, otrips906, _ = run(load906, cut906, 12.2 * H, cap906, old=True) new906, ntrips906, c906 = run(load906, cut906, 12.2 * H, cap906) for name, rows, tr in (("old", old906, otrips906), ("new", new906, ntrips906)): print(f" {name}: request at 2 min {at(rows, 120)[3]:+.0f} W, at 6 h " f"{at(rows, 6 * H)[3]:+.0f} W (i_w {at(rows, 6 * H)[4]:+.0f}), longest " f"hold {longest_hold_h(rows):.2f} h, gate trips {tr}") check("PRECONDITION: the old law holds +5000 W for the whole night on this " "fixture, as it did on 2026-09-06 - the fixture reproduces the card", at(old906, 6 * H)[3] == 5000.0 and longest_hold_h(old906) > 11.5) check("the request reaches the rail before the cut, so the new law starts from " "the same place", at(new906, 120)[3] == 5000.0) t0 = next((r[0] for r in new906 if r[5] == REFUSED), None) check(f"after the cut the battery-refuses hold takes the request to 0 W within " f"about a minute (at t={t0})", t0 is not None and 132 < t0 <= 132 + 90 and at(new906, t0)[3] == 0.0) check("...and it stays refused overnight, with the integrator at 0, and no " "request of 250 W or more stands for longer than one probe interval", at(new906, 6 * H)[6] == 1 and at(new906, 6 * H)[4] == 0.0 and longest_hold_h(new906, until=11.9 * H) < PROBE_EVERY_S / H) check(f"...and the stale-reading gate does not cycle: {ntrips906} trips in " f"12 hours (the old law's latch also kept it quiet: {otrips906})", ntrips906 == 0) morning = [r for r in new906 if r[0] >= 11.9 * H + 60] check("when solar makes the grid export, the refusal ends and the battery " "charges - the exit on the other direction", morning and morning[-1][6] == 0 and morning[-1][3] < -400) # ============================================================================== print("2. the probe: 200 W every 15 minutes, below the gate's 250 W") # ============================================================================== probes = [] for r in new906: if r[3] == PROBE_W and (not probes or r[0] - probes[-1][1] > TICK): probes.append([r[0], r[0]]) elif r[3] == PROBE_W: probes[-1][1] = r[0] starts = [p[0] for p in probes] gaps = [b - a for a, b in zip(starts, starts[1:])] print(f" {len(probes)} probes, first at {starts[0] / 60 if starts else 0:.1f} " f"min, gaps {sorted(set(round(g) for g in gaps))} s, lengths " f"{sorted(set(round(p[1] - p[0] + TICK) for p in probes))} s") check("the probe asks exactly PROBE_W, which is below the stale-reading gate's " "threshold, so a probe can never arm it", PROBE_W < freshness.MOVE_W and all(r[3] in (0.0, PROBE_W) for r in new906 if r[6] == 1)) check("there is a probe every 15 minutes through the night (at least 40 in " "11.8 hours), each about 34 s long", len(probes) >= 40 and all(PROBE_EVERY_S <= g <= PROBE_EVERY_S + PROBE_S + 2 * TICK for g in gaps) and all(PROBE_S - TICK <= p[1] - p[0] + TICK <= PROBE_S + TICK for p in probes)) check("...and a probe reads as a probe on the loop's own reason row", any(r[5] == PROBING for r in new906)) # ============================================================================== print("3. recovery with no sign change: the battery gives power again at 2 h") # ============================================================================== def cut_rec(t): return 132 <= t < 2 * H oldr, _, _ = run(load906, cut_rec, 2.6 * H, cap906, old=True) newr, _, crec = run(load906, cut_rec, 2.6 * H, cap906) after_o = [r for r in oldr if 2 * H <= r[0] < 2 * H + 600] after_n = [r for r in newr if 2 * H <= r[0] < 2.6 * H] exp_o = sum(max(0.0, -r[1]) * TICK for r in after_o) / 3600 exp_n = sum(max(0.0, -r[1]) * TICK for r in after_n) / 3600 print(f" old: worst grid {min(r[1] for r in after_o):+.0f} W, export " f"{exp_o:.0f} Wh; new: worst grid {min(r[1] for r in after_n):+.0f} W, " f"export {exp_n:.0f} Wh") check("PRECONDITION: the old law answers a recovering battery with its " "standing +5000 W and exports kilowatts into a 940 W house", min(r[1] for r in after_o) < -3000) check("the new law never exports more than the probe while it waits, and the " "answered probe ends the refusal so control resumes", min(r[1] for r in after_n) > -PROBE_W - 50 and after_n[-1][6] == 0 and abs(after_n[-1][1] - (-10)) < 100) check("...and the event log says the probe was answered", any("BATTERY REFUSES ENDED: the probe was answered" in e for e in crec.events)) # ============================================================================== print("4. the 09-19 shape: a stale-reading stop on a frozen ~0 W reading " "is a refusal") # ============================================================================== # The request steps +1540 -> +1940 and more as the battery cuts out mid-ramp: # the request moves after the reading went flat, so the gate trips before the # hold's 12 cycles - the ~70 s cycle SAFETY-59 measured. The trip seeds the hold. def load919(t): return 1530 if t < 60 else 2400 if t < 1800 else 900 if t < 4 * H else -600 def cut919(t): return t >= 65 def run919(old): return run(load919, cut919, 4.2 * H, old=old) old919, otr919, _ = run919(True) new919, ntr919, c919 = run919(False) print(f" old: gate trips {otr919}, longest hold {longest_hold_h(old919):.2f} h;" f" new: gate trips {ntr919}, longest hold {longest_hold_h(new919):.2f} h") check("PRECONDITION: on the old law the gate trips over and over all night " "(the SAFETY-59 cycle) or the request stands for hours", otr919 > 20 or longest_hold_h(old919) > 1.0) check("the new law stops after at most two trips and holds 0 W, and the event " "log says the battery refuses", ntr919 <= 2 and at(new919, 2 * H)[3] in (0.0, PROBE_W) and any("BATTERY REFUSES" in e for e in c919.events)) # The same shape through `compute()` alone, as the bus saw it on 2026-09-19 # before the gate existed: +1540 -> +1940, then a battery giving nothing. T = control.Tuning(**LIVE) res = {} for name, carry in (("old", None), ("new", 0)): # sat_count 2: the +1540 and +1940 cycles were already short (the plan) prev, i_w, sat, ref, cmds = 1940.0, 2340.0, 2, carry, [] for _ in range(2000): # ~2.8 h at 5 s d = control.compute(prev, 1950.0, CUT_W, T, sat, i_w, refused=ref) prev, i_w, sat, ref = d.target_w, d.i_w, d.sat_count, d.refused cmds.append(prev) res[name] = cmds print(f" compute() alone: old stands at {res['old'][-1]:+.0f} W " f"({sum(v == 1940.0 for v in res['old'])} of 2000 cycles at +1940 W), " f"new at {res['new'][-1]:+.0f} W from cycle " f"{res['new'].index(0.0) if 0.0 in res['new'] else None}") check("PRECONDITION: without the gate the old law parks at +1940 W, the 7 065 " "identical writes the bus recorded", set(res["old"]) == {1940.0}) check("...and the new law holds 0 W within SETTLE_CYCLES cycles", res["new"].index(0.0) < control.SETTLE_CYCLES and set(res["new"][control.SETTLE_CYCLES:]) == {0.0}) # ============================================================================== print("5. the other exits: stop, start, another writer") # ============================================================================== def stop(c): c.auto = False def start(c): c.auto = True rows, _, c = run(load906, cut906, 1200, cap906, events={600.0: stop, 700.0: start}) check("stopping control ends the refusal, and starting again begins from " "nothing refused", at(rows, 590)[6] == 1 and at(rows, 650)[6] == 0 and at(rows, 705)[6] == 0) def other_writer(c): c.loop_w = 1234.0 # the arbiter gave the wire something the loop did not ask rows, _, c = run(load906, cut906, 900, cap906, events={600.0: other_writer}) check("another writer on the wire ends the refusal (the arbiter reseed)", at(rows, 590)[6] == 1 and at(rows, 605)[6] == 0) # ============================================================================== print("6. words") # ============================================================================== w = messages.why(why_key("loop", [Claim.set("loop", 0, 0.0, REFUSED)]), 0.0) wp = messages.why(why_key("loop", [Claim.set("loop", 0, PROBE_W, PROBING)]), PROBE_W) print(f" {w!r} / {wp!r}") check("the page's Why line says the battery refuses, not 'Keeping the grid " "near zero: 0 W'", w == "The battery refuses to give power: 0 W") check("...and a probe says what it is", wp.startswith("Testing if the battery")) check("...and an ordinary loop cycle still reads as the loop", why_key("loop", [Claim.set("loop", 0, 10.0, "tracking")]) == "loop") print() if fails: print(f"{len(fails)} FAILED: " + ", ".join(fails)) sys.exit(1) print("all checks passed")