"""Runnable check for SAFETY-73. `python3 test_refusal_units.py` SAFETY-52 per inverter. With two or more inverters, a battery that refuses (gives < 100 W for 12 cycles and 22 s while asked > 200 W, or trips the stale-reading gate on a ~0 W reading) is closed in that direction by its own P_SAFETY limit, so the allocator hands its share to the others. It is probed at 200 W every 15 minutes, one inverter at a time. When every inverter refuses the same way the site holds 0 W. When a refusal ends the site target restarts from the watts on the wire. DECISIONS/SAFETY-73 has the record. Drives the REAL `Controller` (read_inputs + cycle, the real freshness gate, allocator, OPT-09 pack claim and SAFETY-42 detector) with two inverters in simulated time. At the cut-out a battery reads -4 W for any request >= -4 W and accepts charge. Two register shapes: FLAT at -4 W (the 09-06 capture) and JITTER +-3 W (the gate cannot arm). `OLD` runs the same fixture with the per-unit refusal switched off, which is 652097a's behaviour at N>1. Mutations (MUTATION.md, `python3 -B`, copies purged of `__pycache__`), graded over this file / test_delivery / test_refusal. Every one is red here: # mutation reds --- --------------------------------------------------------- ------- U1 main.py: the per-unit engage clause never fires 7/2/0 U2 main.py: the N>1 stale-reading seed returns at once 14/0/0 U3 main.py: no `refused` claim is filed 13/2/0 U4 main.py: a probe never pins (always the closed limit) 7/0/0 U5 main.py: a probe pin stands whatever the site target 1/0/0 U6 main.py: no reseed of the site target when a refusal ends 1/0/0 U7 main.py: no site hold when every inverter refuses 2/0/0 U8 main.py: the other-direction exit removed (charge counter) 1/0/0 U9 main.py: the delivery exit removed 3/0/0 U10 main.py: probes are no longer one at a time 1/0/0 U11 main.py: the above-cut-out WARNING condition inverted 1/0/0 U12 main.py: SAFETY-42's account no longer sees a small ask 1/0/0 U13 main.py: a unit engages above `saturation_w`, not PROBE_W 1/0/0 U14 main.py: the evidence need not span REFUSED_SPAN_S (review) 1/0/0 U15 main.py: the WARNING also for a refusal to take (review) 1/0/0 On 652097a itself this file stops at the first sample (`Inverter` has no `refused`); its OLD arms reproduce that tree's N>1 behaviour instead, and every PRECONDITION line is graded against them. """ import asyncio import logging import sys from datetime import datetime, timedelta, timezone from app import control from app.control import PROBE_EVERY_S, PROBE_S, PROBE_W from app.main import Controller from app.arbiter import P_LOOP, P_PEAK, P_SAFETY, Claim, resolve import app.main as m fails = [] def check(name, cond): print(f" {'ok ' if cond else 'FAIL'} {name}") if not cond: fails.append(name) TICK = 5.0 CUT_W = -4.0 FULL_W = 40.0 H = 3600.0 BASE = datetime(2026, 9, 6, 19, 30, tzinfo=timezone.utc) UNITS = ("a", "b") class _Store: def __init__(self): from app.store import DEFAULTS self.data = dict(DEFAULTS) def set(self, k, v): self.data[k] = v def get_time(self, *a): return None class _Pub: enabled = False def publish(self, *a, **k): pass def close(self): pass class _Hass: """A house, one meter, and two batteries that follow unless cut out. `lag` (review): (dead s, publish s). Each battery reading then shows the power of `dead` s before its last publish; the power itself and the meter follow at once. `full[u](t)`: that battery takes at most FULL_W of charge. """ def __init__(self, load, cut, soc, jitter, lag=None, full=None): self.load, self.cut, self.soc, self.jitter = load, cut, soc, jitter self.lag, self.full = lag, full or {} self.cmd = {u: 0.0 for u in UNITS} self.hist = {u: [(-1e18, 0.0)] for u in UNITS} self.t = 0.0 def power(self, u, t=None): t = self.t if t is None else t c = next((v for tt, v in reversed(self.hist[u]) if tt <= t), 0.0) if self.cut[u](t) and c >= CUT_W: if self.jitter: return CUT_W + ((int(t / TICK) % 3) - 1) * 3.0 return CUT_W if u in self.full and self.full[u](t) and c < -FULL_W: return -FULL_W return float(round(c)) def batt(self, u): if self.lag is None: return self.power(u) dead, pub = self.lag return self.power(u, (self.t // pub) * pub - dead) def grid(self): return self.load(self.t) - sum(self.power(u) for u in UNITS) async def limits(self, e): return (-2000.0, 2000.0, 1.0) async def call(self, *a): return True async def number(self, e, invert=False): if e == "sensor.grid": return self.grid() u = e[-1] if e.startswith("sensor.soc"): v = self.soc[u] return v(self.t) if callable(v) else v if e.startswith("sensor.batt"): return self.batt(u) return None async def set_number(self, e, v): self.cmd[e[-1]] = float(v) self.hist[e[-1]].append((self.t, float(v))) return True def _off(c): """652097a's N>1 behaviour: nothing refuses per unit.""" c._unit_refusals = lambda: None c._unit_refusal_start = lambda *a, **k: None def run(load, cut, dur, *, soc=None, jitter=False, old=False, events=None, extra=None, tick=None, lag=None, full=None): """Real cycles every TICK of simulated time. Rows are dicts per cycle.""" async def go(): hass = _Hass(load, cut, soc or {"a": 50.0, "b": 50.0}, jitter, lag, full) dt = tick or TICK inv = [{"name": f"inverter-{u}", "setpoint_entity": f"number.sp_{u}", "soc_entity": f"sensor.soc_{u}", "batt_entity": f"sensor.batt_{u}", "max_w": 2000, "bms_soc_floor": 10} for u in UNITS] o = {"meter_entity": "sensor.grid", "meter_source": "off", "inverters": inv, "gain": 0.6, "slew_w": 400, "target_grid_w": -10, "deadband_w": 15, "step_w": 10, "heartbeat_s": 10, "stale_input_s": 15, "auto_start": True} o.update(extra or {}) c = Controller(o, hass, _Store(), _Pub(), stop=asyncio.Event()) for i in c.inverters: i.dev_min, i.dev_max = -2000.0, 2000.0 c.auto = True if old: _off(c) real_dt = m.datetime clock = [0.0] class _DT(real_dt): @classmethod def now(cls, tz=None): return BASE + timedelta(seconds=clock[0]) m.datetime = _DT rows = [] trips = {u: 0 for u in UNITS} was = {u: False for u in UNITS} pending = dict(events or {}) try: t = 0.0 while t < dur: hass.t = clock[0] = t for et in sorted(pending): if et <= t: pending.pop(et)(c) await c.read_inputs(now_mono=t) for u, i in zip(UNITS, c.inverters): s = i.batt_fresh.stale trips[u] += s and not was[u] was[u] = s await c.cycle() rows.append(dict( t=t, grid=hass.grid(), site=c.target, reason=c.loop_reason, a=hass.cmd["a"], b=hass.cmd["b"], ra=c.inverters[0].refused, rb=c.inverters[1].refused, da=c.inverters[0].delivery_wd.tripped, db=c.inverters[1].delivery_wd.tripped)) t += dt finally: m.datetime = real_dt return rows, trips, c return asyncio.run(go()) def after(rows, t0, t1=None): return [r for r in rows if r["t"] >= t0 and (t1 is None or r["t"] < t1)] def mean(rows, k): return sum(r[k] for r in rows) / len(rows) if rows else float("nan") def never(t): return False def cut_at(t0, t1=None): return lambda t: t >= t0 and (t1 is None or t < t1) def probes(rows, k): """Start times of runs where refused unit `k` is asked exactly PROBE_W.""" out, prev = [], False for r in rows: on = abs(r[k]) == PROBE_W and r["r" + k] != 0 if on and not prev: out.append(r["t"]) prev = on return out SHAPES = (("flat", False), ("jitter", True)) # ============================================================================== print("1. case a: house 3000 W, both give power, A cuts out at 120 s") # ============================================================================== for shape, jit in SHAPES: new, ntr, cn = run(lambda t: 3000.0, {"a": cut_at(120), "b": never}, H, jitter=jit) old, otr, _ = run(lambda t: 3000.0, {"a": cut_at(120), "b": never}, H, jitter=jit, old=True) tail_n, tail_o = after(new, 300), after(old, 300) held = [r for r in tail_n if abs(r["a"]) != PROBE_W] print(f" {shape}: old B {mean(tail_o, 'b'):.0f} W grid " f"{mean(tail_o, 'grid'):+.0f} W trips A {otr['a']}; new B " f"{mean(tail_n, 'b'):.0f} W grid {mean(tail_n, 'grid'):+.0f} W " f"trips A {ntr['a']}") check(f"{shape}: PRECONDITION: on the old law B never takes over (mean " f"under 1950 W) while A gives nothing", mean(tail_o, "b") < 1950 and mean(tail_o, "grid") > 1050) check(f"{shape}: A is refused and asked 0 W outside its probes, B carries " f"its rating, and the grid imports only what B cannot give", cn.inverters[0].refused == 1 and all(r["a"] == 0.0 for r in held) and min(r["b"] for r in tail_n) == 2000.0 and abs(mean(tail_n, "grid") - 1004) < 20) check(f"{shape}: A's stale-reading gate trips at most once, the site is " f"never stopped after the first 5 minutes, and SAFETY-42 does not " f"report A as not delivering", ntr["a"] <= 1 and all(r["reason"] != "not-running" for r in tail_n) and not any(r["da"] for r in new)) # ============================================================================== print("2. case b: house 1200 W, which B alone can carry, A cuts out") # ============================================================================== for shape, jit in SHAPES: new, ntr, cn = run(lambda t: 1200.0, {"a": cut_at(120), "b": never}, H, jitter=jit) old, _, _ = run(lambda t: 1200.0, {"a": cut_at(120), "b": never}, H, jitter=jit, old=True) tail_n, tail_o = after(new, 300), after(old, 300) imp = sum(max(0.0, r["grid"]) for r in tail_n) * TICK / H print(f" {shape}: old grid {mean(tail_o, 'grid'):+.0f} W, new grid " f"{mean(tail_n, 'grid'):+.0f} W, new import {imp:.0f} Wh") check(f"{shape}: PRECONDITION: the old law leaves a standing import", mean(tail_o, "grid") > 250) check(f"{shape}: B takes over the whole house and the grid sits near its " f"target", abs(mean(tail_n, "grid") + 10) < 25 and imp < 10) pa = probes(new, "a") in_probe = [r for r in new if any(p <= r["t"] < p + PROBE_S for p in pa)] check(f"{shape}: A is probed at 200 W every 15 minutes while B carries " f"the rest, and the two never pull opposite ways", len(pa) >= 3 and all(abs(q - p - PROBE_EVERY_S - PROBE_S) <= 2 * TICK for p, q in zip(pa, pa[1:])) and all(r["b"] > 0 for r in in_probe) and all(r["a"] * r["b"] >= 0 for r in new)) # ============================================================================== print("3. case c: both cut out; the site holds 0 W, probes one at a time") # ============================================================================== for shape, jit in SHAPES: new, ntr, cn = run(lambda t: 3000.0, {"a": cut_at(120), "b": cut_at(150)}, 1.5 * H, jitter=jit) old, _, _ = run(lambda t: 3000.0, {"a": cut_at(120), "b": cut_at(150)}, 1.5 * H, jitter=jit, old=True) tail_n, tail_o = after(new, 600), after(old, 600) pa, pb = probes(new, "a"), probes(new, "b") print(f" {shape}: old mean asks A/B {mean(tail_o, 'a'):.0f}/" f"{mean(tail_o, 'b'):.0f} W; new {mean(tail_n, 'a'):.0f}/" f"{mean(tail_n, 'b'):.0f} W, probes A {len(pa)} B {len(pb)}") check(f"{shape}: PRECONDITION: the old law keeps asking two empty " f"batteries", mean(tail_o, "a") > 150 and mean(tail_o, "b") > 150) check(f"{shape}: the site holds 0 W: the loop says `battery refuses` or " f"`battery probe` and each battery is asked 0 W or its probe", all(r["reason"] in (control.REFUSED, control.PROBING) for r in tail_n) and all(r["a"] in (0.0, PROBE_W) and r["b"] in (0.0, PROBE_W) for r in tail_n)) check(f"{shape}: both batteries are probed, never at the same time", len(pa) >= 3 and len(pb) >= 3 and not any(r["a"] and r["b"] for r in new if r["t"] > 300)) # ============================================================================== print("4. recovery: the reseed stops the parked site target reaching the grid") # ============================================================================== # Both cut, the house drops to 940 W at 1 h, A recovers at 1.5 h with no sign # change. Phase A, with the envelope but no reseed and no site hold: -1059 W. ld = (lambda t: 3000.0 if t < H else 940.0) for shape, jit in SHAPES: rows, _, c = run(ld, {"a": cut_at(120, 1.5 * H), "b": cut_at(150)}, 2 * H, jitter=jit) win = after(rows, 1.5 * H, 1.5 * H + 900) worst = min(r["grid"] for r in win) print(f" {shape}: worst grid after A recovers {worst:+.0f} W; A " f"refused {c.inverters[0].refused}") check(f"{shape}: A's recovery is found (at its next probe at the latest) " f"and never exports more than 250 W", c.inverters[0].refused == 0 and worst > -250 and any("BATTERY REFUSES ENDED inverter-a" in e for e in c.events)) # Case a: B at its rating, the site target parked at 3600 W while the house # still imports; the house falls to 2500 W and A recovers. Without the reseed A # is handed the parked 1600 W at once, 1100 W more than the house needs. ld2 = (lambda t: 3000.0 if t < 0.5 * H else 2500.0) rows, _, c = run(ld2, {"a": cut_at(120, 0.75 * H), "b": never}, 1.5 * H) win = after(rows, 0.75 * H, 0.75 * H + 900) print(f" a: worst grid when A recovers {min(r['grid'] for r in win):+.0f} W") check("case a: a recovery hands A no parked site target (worst export under " "250 W)", min(r["grid"] for r in win) > -250 and any("BATTERY REFUSES ENDED inverter-a" in e for e in c.events)) # ============================================================================== print("5. exits, words and the circulation guard") # ============================================================================== # Solar: the house exports from 30 min, so the loop wants to charge. rows, _, c = run(lambda t: 3000.0 if t < 0.5 * H else -2500.0, {"a": cut_at(120), "b": never}, H) check("the grid asking the other direction lets A charge at once (its limit " "closes only the refused direction) and a minute of charge ends the " "refusal", c.inverters[0].refused == 0 and min(r["a"] for r in after(rows, 0.6 * H)) < -500 and any("the other way" in e for e in c.events)) def _stop(c): c.auto = False rows, _, c = run(lambda t: 3000.0, {"a": cut_at(120), "b": never}, 0.5 * H, events={1200.0: _stop}) check("stopping control ends the refusal", c.inverters[0].refused == 0 and any("control stopped" in e for e in c.events)) # A small house: the site asks under 200 W, so a probe may not pin A (B would # have to charge while A gives power). The probe waits instead. rows, _, c = run(lambda t: 3000.0 if t < 600 else 120.0, {"a": cut_at(120), "b": never}, 1.2 * H) check("with the site asking under 200 W no probe is sent and no inverter is " "ever asked the opposite way to the other", c.inverters[0].refused == 1 and probes(after(rows, 700), "a") == [] and all(r["a"] * r["b"] >= 0 for r in rows) and not any("ARBITER" in e for e in c.events)) # The hold binds every strategy's ask: a peak-guard SET at P_PEAK on the site # is still placed around a refusing unit's closed limit. _res = resolve([Claim.set("peak", P_PEAK, 3000.0, "quarter peak"), Claim.set("loop", P_LOOP, 500.0, "tracking"), Claim.limit("supervised", P_SAFETY, -2000.0, 2000.0, actuator="inverter-a"), Claim.limit("supervised", P_SAFETY, -2000.0, 2000.0, actuator="inverter-b"), Claim.limit("refused", P_SAFETY, hi=0.0, reason=control.REFUSED, actuator="inverter-a")], units=("inverter-a", "inverter-b"), live=("inverter-a", "inverter-b")) check("a P_SAFETY refusal limit binds a higher-priority strategy too: the " "peak guard's 3000 W goes to B, A gets 0 W", _res.shares.get("inverter-a") == 0 and _res.shares.get("inverter-b") == 2000) class _Grab(logging.Handler): def __init__(self): super().__init__() self.lines = [] def emit(self, record): self.lines.append((record.levelno, record.getMessage())) _g = _Grab() _lg = logging.getLogger("goodwe") _lv = _lg.level _lg.setLevel(logging.INFO) _lg.addHandler(_g) run(lambda t: 3000.0, {"a": cut_at(120), "b": never}, 600) run(lambda t: 3000.0, {"a": cut_at(120), "b": never}, 600, soc={"a": 10.0, "b": 50.0}) _lg.removeHandler(_g) _lg.setLevel(_lv) warns = [x for lv, x in _g.lines if lv == logging.WARNING and "REFUSES_ABOVE_CUT_OUT" in x] infos = [x for lv, x in _g.lines if lv == logging.INFO and "BATTERY_REFUSES" in x] print(f" {warns} / {infos}") check("a refusal above the configured cut-out level logs INFO and one WARNING " "naming the inverter, and a battery at its cut-out logs neither " "(OPT-09's limit closes it first)", len(warns) == 1 and warns[0].startswith("inverter-a ") and len(infos) == 1 and infos[0].startswith("inverter-a:") and len(warns[0].split()) <= 25) # A night load of 400 W: each unit is asked about 400 W once the loop has # made up A's missing half, under `saturation_w`. Engaging at PROBE_W is what # refuses A here; at 500 W it would keep a phantom share that SAFETY-43 names. rows, _, c = run(lambda t: 400.0, {"a": cut_at(0), "b": never}, 1200) check("a 400 W night load: A is refused at a share under 500 W, B carries the " "house and nothing reports A as not delivering", c.inverters[0].refused == 1 and not any(r["da"] for r in rows) and abs(mean(after(rows, 300), "grid") + 10) < 25 and mean(after(rows, 300), "b") > 380) # A night load of 150 W: A's share never passes PROBE_W, so it is never # refused, and SAFETY-43 must not name it as not delivering either: a battery # asked under 200 W that gives under 100 W is a refusal too small to call. rows, _, c = run(lambda t: 150.0, {"a": cut_at(0), "b": never}, 1200) check("a small share on an empty battery is neither refused nor reported as " "not delivering, and the grid stays near its target", c.inverters[0].refused == 0 and not any(r["da"] for r in rows) and abs(mean(after(rows, 300), "grid") + 10) < 25) rows, _, c = run(lambda t: 3000.0, {"a": cut_at(120), "b": never}, 600) notes = c.liveness_notes() check("the status page names the refusing inverter", any(n.startswith("inverter-a: the battery refuses to give power") for n in notes)) # Shadow mode writes nothing, so nothing is asked and nothing can refuse. rows, _, c = run(lambda t: 3000.0, {"a": cut_at(120), "b": never}, 1200, extra={"shadow_mode": True}) check("shadow mode: no inverter is refused and both stay at 0 W", c.inverters[0].refused == 0 and c.inverters[1].refused == 0 and all(r["a"] == 0.0 and r["b"] == 0.0 for r in rows)) # Review. A 1 s meter makes 12 cycles 11 s, shorter than a healthy battery # reading's lag (6 s dead time plus a 13 s publish). Ordinary load steps then # refused both inverters and dropped the site to 0 W (16 times an hour). The # evidence must also span REFUSED_SPAN_S. _steps = (lambda t: 300.0 if int(t // 120) % 2 == 0 else 3000.0) rows, _, c = run(_steps, {"a": never, "b": never}, H, tick=1.0, lag=(6.0, 13.0)) check("a 1 s meter and slow battery readings: load steps for an hour refuse " "neither healthy inverter", not any(r["ra"] or r["rb"] for r in rows)) rows, _, c = run(lambda t: 3000.0, {"a": cut_at(120), "b": never}, 600, tick=1.0, lag=(6.0, 13.0)) _eng = next((r["t"] for r in rows if r["ra"]), None) check("...and the same slow readings still refuse a battery that is cut " "out, after the span and well inside the 185 s report", _eng is not None and 120 + m.REFUSED_SPAN_S <= _eng < 185 and not any(r["da"] for r in rows)) # Review. A full battery that stops taking charge is refused (charge), and # it is always above its cut-out level: that says nothing about the setting, # so it gets the INFO line and no REFUSES_ABOVE_CUT_OUT warning. _g2 = _Grab() _lg.setLevel(logging.INFO) _lg.addHandler(_g2) rows, _, c = run(lambda t: -3000.0, {"a": never, "b": never}, 600, soc={"a": 99.0, "b": 60.0}, full={"a": lambda t: t >= 120}) _lg.removeHandler(_g2) _lg.setLevel(_lv) check("a full battery that stops taking charge is refused that way only, " "with the INFO line and no warning about its cut-out level", c.inverters[0].refused == -1 and any(lv == logging.INFO and "BATTERY_REFUSES" in x for lv, x in _g2.lines) and not any("REFUSES_ABOVE_CUT_OUT" in x for _lv2, x in _g2.lines)) # ============================================================================== print("6. one inverter is untouched") # ============================================================================== def _boom(*a, **k): raise AssertionError("SAFETY-73 code ran at N=1") _saved = {k: getattr(Controller, k) for k in ( "_unit_refusals", "_unit_refusal_start", "_unit_refusal_end", "_unit_probe", "_site_refused")} for k in _saved: setattr(Controller, k, _boom) try: import test_refusal as _t1 _one_ok = not _t1.fails except (AssertionError, SystemExit) as e: _one_ok = isinstance(e, SystemExit) and not e.code finally: for k, v in _saved.items(): setattr(Controller, k, v) check("SAFETY-52's whole single-inverter suite passes with every SAFETY-73 " "method made to raise, so none of it runs at N=1", _one_ok) print() if fails: print(f"{len(fails)} FAILED: " + ", ".join(fails)) sys.exit(1) print("all checks passed")