"""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 R10 main.py: seed from a stale stop below `saturation_w` again, 1 not below REFUSED_BELOW_W (review of #141). The same mutation in control.py's engage clause SURVIVES, and that is expected: the delivery exit (R11) ends such a refusal in the same cycle. R11 control.py: only an answered PROBE ends the refusal 2 (`and probe` back in the delivery exit; review of #141) """ 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("5b. review of #141: a pack that gives SOME power, and a reading that is " "only late, are not a refusal") # ============================================================================== # A pack derated to 100-499 W (cold, BMS taper) gives what it can against a # larger ask. Engaging below `saturation_w` (500 W) held it at 0 W for ~93 % of # the time; "gives nothing" is below the probe's own answer. # `compute()` alone: through the Controller the stale-reading gate trips on the # flat 300 W first, and that fixture then passes whatever this threshold is. def derated(carry, cap_w=300.0, n=1800): prev, i_w, sat, ref, gave, held = 0.0, 0.0, 0, carry, 0.0, 0 b = 0.0 for _ in range(n): # 2.5 h at 5 s d = control.compute(prev, 2000.0 - b, b, T, sat, i_w, refused=ref) prev, i_w, sat, ref = d.target_w, d.i_w, d.sat_count, d.refused held += bool(ref) b = max(0.0, min(prev, cap_w)) # one tick late gave += b return gave / n, held T = control.Tuning(**LIVE) gave_old, _ = derated(None) gave_new, held_new = derated(0) print(f" pack derated to 300 W: old law {gave_old:.0f} W average, new " f"{gave_new:.0f} W, held as refusing on {held_new} cycles") check("a pack that gives a steady 300 W against a 2000 W ask is never held at " "0 W as refusing, and keeps giving what the old law got from it", held_new == 0 and gave_new == gave_old and gave_old > 290.0) def lagging(tick, dead_s=6.0, publish_s=10.0, phase_s=5.0, dur=600.0): """`compute()` on a HEALTHY pack whose reading lags: it follows the command of `dead_s` ago, published every `publish_s`. A 1 s meter (`meter_poll_s` 1) makes SETTLE_CYCLES only 12 s, shorter than that lag.""" prev, i_w, sat, ref, hist, reg, pub = 0.0, 0.0, 0, 0, [], 0.0, -phase_s held = engaged = 0 t = 0.0 while t < dur: then = [c for tc, c in hist if tc <= t - dead_s] b = then[-1] if then else 0.0 if t - pub >= publish_s: reg, pub = b, t d = control.compute(prev, (3000.0 if t >= 10 else 0.0) - b, reg, T, sat, i_w, refused=ref) engaged += bool(d.refused) and not ref prev, i_w, sat, ref = d.target_w, d.i_w, d.sat_count, d.refused hist.append((t, prev)) held += tick * bool(ref) t += tick return engaged, held lag1 = lagging(1.0) print(f" 1 s meter, healthy pack with a late reading: engaged {lag1[0]} " f"time(s), held {lag1[1]:.0f} s") check("a healthy pack whose reading is merely late is let go as soon as the " "reading shows it delivering - not held until the next probe", lag1[1] <= 10.0) async def _seed(held_w): """`_refusal_from_stale()` on a unit standing at +2000 W, frozen.""" c = Controller(dict( {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid", "soc_entity": "sensor.soc", "batt_entity": "sensor.batt", "meter_source": "off"}, **LIVE), _Hass(lambda t: 0.0, lambda t: False), _Store(), _Pub(), stop=asyncio.Event()) c.auto, c.sat_count = True, c.tuning.saturation_cycles inv = c.inverters[0] inv.last_write, inv.batt_fresh.held = 2000.0, held_w c._refusal_from_stale(inv, 100.0) return c.batt_refused check("a stale-reading stop on a register standing at 300 W is not a refusal " "(it gives something), and one at -4 W is", asyncio.run(_seed(300.0)) == 0 and asyncio.run(_seed(CUT_W)) == 1) d = control.compute(0.0, 1500.0, 1400.0, T, 0, 0.0, refused=1, probe=False) check("...i.e. a held direction that reads 100 W or more ends the refusal " "without waiting for a probe", d.refused == 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")