"""Runnable check for SAFETY-94: a healthy battery that lags or reverses is not flagged as frozen or refusing. `python3 test_safety94.py` Two false alarms, both pre-existing: - The stale-reading gate (SAFETY-37) counted its 35 s from the register's last change. After a rest that clock was already past 35 s, so the first read after a new command tripped it, before the register could answer (SAFETY-58; live 2026-10-03, four trips, each 1 s after a command). - At N=1 the refusal (SAFETY-52) engaged after 12 frozen cycles if the reading was under 100 W at that moment. At a 1 s meter, a battery reversing out of a charge through its lag is passing 0 W just then (SAFETY-85 rig, O1). It was held at 0 W until the next probe, up to 15 min. Now the gate counts only the time the command asked the register to move, and a refusal needs the reading to have stayed under 100 W, not moving towards the ask, for 12 cycles spanning 22 s, at every roster size. DECISIONS/SAFETY-94-lagging-battery.md. 1. The four live trips, replayed through the gate. 2. The REAL Controller on a lagging battery (tools/safety94_rig.py), N=1 and N=2: rest then a step, maintenance drain start, drain -> charge, a charge reversing into a discharge. No stale trip and no refusal. 3. A frozen register and a refusing battery are still caught, and how fast. 4. The rules at their edges. Mutations (MUTATION.md, purged copies) are listed in the pull request. """ import logging import os import sys 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.CRITICAL) import safety94_rig as rig # noqa: E402 import app.inverter as app_inverter # noqa: E402 import app.main as app_main # noqa: E402 from app import control # noqa: E402 from app.freshness import BOUNDS, Freshness # noqa: E402 fails = [] def check(name, cond): print(f" {'ok ' if cond else 'FAIL'} {name}") if not cond: fails.append(name) class Before94(Freshness): """The gate before SAFETY-94: the clock ran from the last change.""" def obliged_s(self): return self.graded_s(self.last_read) def without_fix(fn): """Run `fn` with the gate's old clock and the old refusal evidence.""" real_f, real_c = app_inverter.Freshness, app_main.compute real_p = app_main.REFUSED_PROGRESS_W app_inverter.Freshness = Before94 app_main.compute = lambda *a, **k: real_c(*a, **dict(k, gave_nothing=None)) app_main.REFUSED_PROGRESS_W = float("inf") try: return fn() finally: app_inverter.Freshness, app_main.compute = real_f, real_c app_main.REFUSED_PROGRESS_W = real_p # ============================================================================ # print("1. the four live trips of 2026-10-03, replayed") # HA history (only changes are recorded), local time: the unit's setpoint # (what stands on its wire) and its battery power register (0x0518). LIVE = { "17:39:16 inverter-b": ( [("17:38:00", -24), ("17:38:14.7", 0), ("17:38:25.4", 5), ("17:38:40.5", 16), ("17:38:44.7", 26), ("17:38:55.7", 37), ("17:39:05.1", 42), ("17:39:09.8", 58), ("17:39:15.6", 268), ("17:39:25.0", 479), ("17:39:30.7", 689), ("17:39:34.8", 900)], [("17:38:00", -118), ("17:38:21.6", -99), ("17:39:21.6", 179), ("17:39:31.6", 399)]), "17:39:25 inverter-a": ( [("17:38:00", -16), ("17:38:14.9", 0), ("17:38:25.0", 5), ("17:38:40.0", 14), ("17:38:44.7", 24), ("17:38:55.2", 33), ("17:39:04.7", 38), ("17:39:09.3", 52), ("17:39:15.0", 242), ("17:39:24.6", 431), ("17:39:30.3", 621), ("17:39:34.8", 810)], [("17:38:00", -3), ("17:38:25.9", 0), ("17:39:25.9", 318), ("17:39:36.0", 654)]), "19:00:04 inverter-a": ( [("18:58:30.0", 290), ("18:58:43.3", 305), ("18:58:48.0", 314), ("18:58:52.5", 300), ("19:00:03.1", 2500)], [("18:58:30.0", 292), ("19:00:06.0", 951), ("19:00:16.0", 2360), ("19:00:26.0", 2376)]), "21:21:22 inverter-b": ( [("21:19:30.0", 0), ("21:21:21.9", 2100), ("21:21:26.6", 1700), ("21:21:31.1", 1300), ("21:21:36.9", 900), ("21:21:41.0", 500), ("21:21:47.3", 200)], [("21:19:30.0", -98), ("21:21:22.8", -97), ("21:21:32.8", 1411), ("21:21:42.8", 798)]), } def _sec(hms): h, m, s = hms.split(":") return int(h) * 3600 + int(m) * 60 + float(s) def replay(cls, setp, reg): """1 Hz reads, as `run_control()` does. Returns the trip times.""" f, trips = cls(BOUNDS["batt"]), [] t, end = _sec(setp[0][0]), _sec(reg[-1][0]) def at(series, t): return [float(v) for ts, v in series if _sec(ts) <= t][-1] while t <= end: was = f.stale f.submit(at(reg, t), at(setp, t), t) if f.stale and not was: trips.append(t) t += 1.0 return trips for name, (setp, reg) in LIVE.items(): old, new = replay(Before94, setp, reg), replay(Freshness, setp, reg) print(f" {name}: before {[f'{int(t // 3600):02d}:{int(t % 3600 // 60):02d}:{t % 60:04.1f}' for t in old]}" f", now {len(new)} trips") check(f"{name}: the shipped trip reproduces 1-2 s after the command, and " f"now there is none (the register answered within 11 s)", len(old) == 1 and abs(old[0] - _sec(name.split()[0])) < 1.0 and new == []) # ============================================================================ # print("\n2. the REAL Controller on a lagging battery (10 s lag)") # Two register shapes: published every 10 s and repeating its value (the # boards, live), and every 1 s with +-1 W of dither (the SAFETY-85 rig). SHAPES = ((10.0, False), (1.0, True)) NAMES = ("rest_step", "drain_start", "drain_charge", "reversal") def sweep(fn=lambda f: f()): out = {} for n in (1, 2): for name in NAMES: for tick in (1.0, 2.0, 5.0): for pub, dith in SHAPES: s, r, _ = fn(lambda: rig.counts(n, name, tick=tick, publish_s=pub, dither=dith)) out[(n, name, tick, pub)] = (s, r) return out NOW = sweep() OLD = sweep(without_fix) for n in (1, 2): for name in NAMES: now = [NOW[k] for k in NOW if k[:2] == (n, name)] old = [OLD[k] for k in OLD if k[:2] == (n, name)] print(f" N={n} {name:12s} stale/refused now " f"{sum(a for a, _ in now)}/{sum(b for _, b in now)}, before " f"{sum(a for a, _ in old)}/{sum(b for _, b in old)}") check("⭐ no stale trip and no refusal, at N=1 and N=2, at 1, 2 and 5 s meters, " "on both register shapes: a step after a rest, a drain start, drain -> " "charge and a charge reversing into a discharge", all(v == (0, 0) for v in NOW.values()) and len(NOW) == 48) check("non-vacuous: before, the same runs trip the gate (drain start and " "drain -> charge at both N, as live) and refuse at N=1 on a 1 s meter", sum(OLD[(n, nm, t, 10.0)][0] for n in (1, 2) for nm in ("drain_start", "drain_charge") for t in (1.0, 2.0, 5.0)) >= 12 and OLD[(1, "drain_charge", 1.0, 1.0)][1] >= 1) def reversals(n, tick, taus=(10.0, 20.0)): """A charge reversing into a discharge: (stale trips, refusals) summed.""" s = r = 0 for tau in taus: for charge in (-1500.0, -2500.0): for house in (300.0, 800.0, 2000.0): for pub, dith in SHAPES: x = rig.run(n, rig.reversal(300.0, charge, house), 600.0, tau=tau, tick=tick, publish_s=pub, dither=dith) s, r = s + len(x["stale"]), r + len(x["refused"]) return s, r REV = {(n, t): reversals(n, t) for n in (1, 2) for t in (1.0, 2.0)} REV_OLD = {(n, 1.0): without_fix(lambda n=n: reversals(n, 1.0)) for n in (1, 2)} print(f" reversals (lag 10 and 20 s): now {REV}, before {REV_OLD}") check("⭐ SAFETY-94's case: a battery reversing out of a charge (1500 or 2500 " "W) into a 300-2000 W house, lag 10 or 20 s, 1 or 2 s meter, is never " "refused or flagged, at N=1 or N=2", all(v == (0, 0) for v in REV.values())) check("non-vacuous: before, the 1 s meter refuses at N=1 (lag 20 s) and at " "N=2 (B, lag 20 s, its reading 100 W below its power)", REV_OLD[(1, 1.0)][1] >= 1 and REV_OLD[(2, 1.0)][1] >= 1) # ============================================================================ # print("\n3. a frozen register and a refusing battery are still caught") def obliged_at(rows, t_f): """First cycle from `t_f` where A's command is 250 W or more and has moved 250 W from where it stood at `t_f`.""" c0 = None for row in rows: if row[0] >= t_f and c0 is None: c0 = row[2]["a"] if c0 is not None and abs(row[2]["a"]) >= 250 \ and abs(row[2]["a"] - c0) >= 250: return row[0] return None CATCH = {} for n in (1, 2): for tick in (1.0, 5.0): # frozen while giving 1000 W, then the house steps r1 = rig.run(n, lambda t: 3000.0 if t >= 400 else 1000.0, 520.0, tick=tick, publish_s=1.0, dither=True, freeze_at=lambda u: 300.0 if u == "a" else None) ob = obliged_at(r1["rows"], 300.0) st = [t for t, u in r1["stale"] if u == "inverter-a"] # frozen at rest (HW-01's shape), the house steps 500 s later r2 = rig.run(n, lambda t: 2500.0 if t >= 600 else 0.0, 700.0, tick=tick, publish_s=1.0, dither=True, freeze_at=lambda u: 100.0 if u == "a" else None) ob2 = obliged_at(r2["rows"], 100.0) park = min([t for t, u in r2["stale"] if u == "inverter-a"] + [t for t, _u in r2["refused"]] or [None]) # a refusing battery from 300 s r3 = rig.run(n, lambda t: 3000.0, 420.0, tick=tick, publish_s=10.0, cut=lambda u, t: u == "a" and t >= 300.0) ref = [t for t, _u in r3["refused"]] CATCH[(n, tick)] = (st[0] - ob if st and ob is not None else None, park - ob2 if park is not None and ob2 is not None else None, ref[0] - 300.0 if ref else None) print(" (N, meter): (frozen giving: trip after the command moved, frozen " "at rest: 0 W after, refusing: refused after the cut), in s") print(f" {CATCH}") check("⭐ a register frozen while giving is caught 35 s after the command asks " "it to move (the SAFETY-37 bound, now counted from the ask)", all(v[0] == BOUNDS["batt"].unchanged_s for v in CATCH.values())) check("⭐ a register frozen at rest reads like a refusing battery: the " "battery is at 0 W, refused or stale, within 36 s of the ask", all(v[1] is not None and v[1] <= 36.0 for v in CATCH.values())) check("⭐ a refusing battery is still refused: within 22 s plus one 10 s " "publish at a 1 s meter, and within 12 cycles plus a publish at 5 s", all(CATCH[(n, 1.0)][2] is not None and CATCH[(n, 1.0)][2] <= 32.0 for n in (1, 2)) and all(CATCH[(n, 5.0)][2] is not None and CATCH[(n, 5.0)][2] <= 12 * 5.0 + 10.0 for n in (1, 2))) # ============================================================================ # print("\n4. the rules at their edges") # The gate counts time the register was asked to move, across a flapping # command too. f = Freshness(BOUNDS["batt"]) trip = None for t in range(0, 200): f.submit(-3.0, 1500.0 if (t // 5) % 2 else 0.0, float(t)) if f.stale and trip is None: trip = t print(f" command flapping 0 / 1500 W every 5 s, register frozen: trip at " f"{trip} s") check("a frozen register under a command that flaps in and out of the ask is " "still caught (after 35 s of asking, about 70 s here)", trip is not None and 65 <= trip <= 75) f = Freshness(BOUNDS["batt"]) for t in range(0, 30): f.submit(-3.0, 1500.0, float(t)) f.submit(400.0, 1500.0, 30.0) # it answers for t in range(31, 60): f.submit(400.0, 1500.0, float(t)) check("a register that answers starts a fresh count: 30 s asked, then it " "moves, then 29 s more is not stale", not f.stale and f.obliged_s() < 35.0) f = Freshness(BOUNDS["batt"]) for t in range(0, 600): f.submit(-3.0, 200.0 if t % 60 < 3 else 0.0, float(t)) check("a command that never asks 250 W is never an ask (SAFETY-52's 200 W " "probe cannot arm the gate)", not f.stale and f.obliged_s() == 0.0) # Review: time under 250 W is not counted, even after the command travelled. f = Freshness(BOUNDS["batt"]) for t in range(0, 61): f.submit(-300.0, -300.0 if t == 0 else 200.0, float(t)) for t in range(61, 131): f.submit(-300.0, 260.0, float(t)) if t == 90: early = f.stale check("a command that has travelled 500 W but stands at 200 W for a minute " "adds nothing: 30 s after it reaches 260 W the register is not stale, " "and by 70 s it is", not early and f.stale) # compute(): `gave_nothing` gates the N=1 engage; None keeps the old law. T = control.Tuning(max_w=4000.0, slew_w=400.0) args = dict(prev_w=1200.0, grid_w=1200.0, actual_w=-20.0, tuning=T, sat_count=control.SETTLE_CYCLES, i_w=1200.0, refused=0) check("compute(): a frozen loop at SETTLE_CYCLES reading -20 W engages with " "gave_nothing None (compute alone, as before) or True, and not with " "False", control.compute(**args).refused == 1 and control.compute(**args, gave_nothing=True).refused == 1 and control.compute(**args, gave_nothing=False).refused == 0) # The evidence restarts on progress towards the ask, not on jitter. print(f" REFUSED_PROGRESS_W {app_main.REFUSED_PROGRESS_W:.0f} W, span " f"{app_main.REFUSED_SPAN_S:.0f} s") class _Inv: def __init__(self): self.batt, self.ref_count, self.ref_since, self.ref_sign = 0.0, 0, None, 0 self.ref_from = None self.idle_w = 0.0 # SAFETY-97: A, ~0 W at idle self.ask = 800.0 def standing_w(self): return self.ask class _C: _now_mono = 0.0 def evidence(readings, tick=1.0, ask=lambda i: 800.0): c, inv = _C(), _Inv() got = [] for i, b in enumerate(readings): c._now_mono, inv.batt, inv.ask = i * tick, b, ask(i) got.append(app_main.Controller._gave_nothing(c, inv)) return got flat = evidence([-4.0 + (i % 3) for i in range(30)]) climb = evidence([-95.0 + 4.0 * i for i in range(48)]) check("a flat reading (+-1 W) under an 800 W ask is evidence after 22 s; one " "climbing 4 W/s through the band is never", 1 in flat and flat.index(1) == 22 and not any(climb)) flap = evidence([-4.0] * 60, ask=lambda i: 800.0 if i % 2 else -800.0) away = evidence([40.0] * 5 + [-40.0] * 25) check("a command that swaps direction every cycle is never evidence, and a " "reading moving 80 W away from the ask does not restart it (latched at " "22 s, counted from the first reading)", not any(flap) and 1 in away and away.index(1) == 22) check("REFUSED_PROGRESS_W is under the band's width and over the live " "readings' rest wander (25 W)", 25.0 < app_main.REFUSED_PROGRESS_W < 2 * control.REFUSED_BELOW_W) print() if fails: print(f"{len(fails)} FAILED: " + ", ".join(fails)) sys.exit(1) print("all checks passed")