"""Runnable check for SAFETY-87: the peak guard uses every battery that can still give. `python3 test_safety87.py` At N>=2 the guard graded the EMPTIEST battery (SAFETY-32), so one battery under the peak reserve stopped the guard for all of them: it asked for nothing and raised the reserve alarm while the other battery was full. Now a battery at or under its own reserve is HELD (it gives nothing for the peak, weight 0) and the guard's need is placed on the others; the reserve alarm fires only when they cannot hold the limit. Drives the REAL Controller (tools/ safety84_identity.py's harness). DECISIONS/SAFETY-87-guard-uses-able-batteries.md. 1. Maintenance charge at N=2 (A under the reserve, B able), house over the aim: no reserve alarm, the quarter on the aim, A gives nothing for the peak while B can carry it. When B cannot, the alarm fires and names A. 2. HOLD at N=2 (A full holding, B under the reserve): A gives, B gives nothing, the quarter lands on the aim; at a banked month peak too. 3. Both under the reserve, or the other one refusing: the alarm still fires and names the right battery. 4. The alarm at the quarter's edge: a one-cycle miss does not fire it. 5. place_on_able() at its boundaries, and the claim it files. 6. The allocator: a held battery gives last, after an Off one. 7. A held battery draining for maintenance counts for its drain. 8. The page: a peak alarm does not say "Not ready" or "0 W until the alarm clears" while control goes on. 9. N=1 is byte-identical to release/1.0 b8ddfd7 (tools/safety87_identity.py). Mutations (MUTATION.md, purged copies) are listed in the pull request. """ import asyncio import dataclasses import json import logging import os import sys import types 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 safety84_identity as s84 # noqa: E402 import safety87_identity as s87 # noqa: E402 from app import messages, packmode, peakguard # noqa: E402 from app.arbiter import (Claim, P_MAINTENANCE, P_MODE, P_PEAK, # noqa: E402 P_SAFETY, UnitBounds, resolve) from app.main import Controller # noqa: E402 from app.maintenance import DRAIN, HOLD, IDLE, MaintResult # noqa: E402 from app.store import unit_key # noqa: E402 fails = [] A, B = "inverter-a", "inverter-b" AIM = 3450.0 # peak_cap_w 3500 - peak_margin_w 50 TOL = 10.0 # W over the aim a banked quarter may show (SAFETY-81) def check(name, cond): print(f" {'ok ' if cond else 'FAIL'} {name}") if not cond: fails.append(name) def store(phase, *, b_off=False): """Battery A in `phase` from the first cycle (B idle, cycle done).""" s = rig._Store() s.data[unit_key(A, "phase")] = phase s.data[unit_key(A, "phase_started")] = rig.BASE.isoformat() s.data[unit_key(B, "last_completed")] = rig.BASE.isoformat() if phase == IDLE: s.data[unit_key(A, "last_completed")] = rig.BASE.isoformat() if b_off: s.data[unit_key(B, packmode.USE_KEY)] = False return s def run(house, phase, socs, *, cheap=False, n=2, **kw): """Rows, controller, the notifications raised and one record per cycle: (floor_limited, held, floor_w, reserve set, alarm latched).""" notes, seen = [], [] real_check = Controller.check_peak_floor_limited async def spy(self): await real_check(self) g = self.peak_guard seen.append(None if g is None else ( g.floor_limited, g.held, g.floor_w, self._peak_reserve(), self._peak_floor_limited)) def before(c, t): if not getattr(c.hass, "_s87", False): real = c.hass.call async def call(*a): if a[:2] == ("persistent_notification", "create"): notes.append(a[2]) return await real(*a) c.hass.call, c.hass._s87 = call, True return s84.feed(cheap)(c, t) kw.setdefault("dur", 2700.0) extra = dict({"maintenance_hold_min": 600}, **kw.pop("extra", {})) Controller.check_peak_floor_limited = spy try: rows, c = s84.run( n, load=house if callable(house) else lambda t: house, cheap=cheap, store=kw.pop("store", None) or store(phase), soc=lambda u, t: socs[u], before=before, extra=extra, **kw) finally: Controller.check_peak_floor_limited = real_check return rows, c, notes, seen def without_fix(fn): """The shipped b8ddfd7 behaviour: no battery is ever held back.""" real = Controller._peak_on_able Controller._peak_on_able = lambda self, decision, results: decision try: return fn() finally: Controller._peak_on_able = real def cmd(r): return dict(r[2]) def banked(c): return [b.offtake_avg_w for b in c.p1.blocks] def on_aim(c, aim=AIM): b = banked(c) return len(b) >= 2 and all(aim - TOL <= v <= aim + TOL for v in b) def reserve_notes(notes): return [x for x in notes if x["notification_id"] == messages.ID_PEAK_RESERVE] def limited_events(c): return sum(1 for e in c.events if "PEAK_GUARD_FLOOR_LIMITED" in e) def show(tag, rows, c, notes, seen): a = [cmd(r)["a"] for r in rows] b = [cmd(r)["b"] for r in rows] print(f" {tag}: banked {[round(v, 1) for v in banked(c)]} W, mean A " f"{sum(a) / len(a):.0f} W, B {sum(b) / len(b):.0f} W, reserve " f"alarms {len(reserve_notes(notes))}, held " f"{sorted({h for s in seen if s for h in s[1]})}") # ============================================================================ # print("1. maintenance charge at N=2: A under the reserve, B able") LOW_A = {"a": 11.0, "b": 60.0} CH = run(4500.0, "charge", LOW_A, cheap=True) CH_OLD = without_fix(lambda: run(4500.0, "charge", LOW_A, cheap=True)) show("house 4500", *CH) show("house 4500, before the fix", *CH_OLD) check("⭐ house 4500 W, B can carry the guard alone: no reserve alarm (no " "notification, no PEAK_GUARD_FLOOR_LIMITED line) and every quarter on " "the aim", not reserve_notes(CH[2]) and limited_events(CH[1]) == 0 and on_aim(CH[1])) check("...A is held (it gives nothing for the peak) on every cycle the guard " "asks for power, and the guard asks for a positive floor", any(s and s[2] and s[2] > 0.0 for s in CH[3]) and all(s[1] == (A,) and s[3] == {A} for s in CH[3] if s and s[2] and s[2] > 0.0)) check("non-vacuous: before the fix the same run raised the reserve alarm " "(the SAFETY-84 rig test's check 2/3 false alarm)", len(reserve_notes(CH_OLD[2])) >= 1 and limited_events(CH_OLD[1]) >= 1) _b_carry = [r for r, s in zip(CH[0], CH[3]) if s and s[3] == {A} and r[1] <= 2500.0 and r[1] > 0.0] check("...and while B can carry the site target (at most its 2500 W) A is " "not asked to give at all (weight 0 behind B)", len(_b_carry) >= 3 and all(cmd(r)["a"] <= 0.0 for r in _b_carry)) CH65 = run(6500.0, "charge", LOW_A, cheap=True) show("house 6500", *CH65) _n65 = reserve_notes(CH65[2]) check("⭐ house 6500 W, B (2500 W) cannot hold the 3450 W aim alone: the " "reserve alarm fires and names A, the battery under its reserve, and " "says the others cannot hold the limit alone", len(_n65) >= 1 and all("inverter-a is at its peak reserve" in x["message"] and "inverter-b" not in x["message"] and "cannot keep this 15-minute period under your 3 500 W peak " "limit alone" in x["message"] for x in _n65)) check("...and while it is limited the guard still asks B for everything B " "has (floor 2500 W), so B is never left idle", any(s and s[0] and s[2] == 2500.0 for s in CH65[3]) and all(s[2] == 2500.0 for s in CH65[3] if s and s[0])) # ============================================================================ # print("\n2. HOLD at N=2: A full and holding, B under the reserve") LOW_B = {"a": 99.0, "b": 15.0} HD = run(5500.0, HOLD, LOW_B) HD_OLD = without_fix(lambda: run(5500.0, HOLD, LOW_B)) show("house 5500", *HD) show("house 5500, before the fix", *HD_OLD) check("⭐ house 5500 W: A gives, B gives nothing, every quarter lands on the " "aim, no reserve alarm, and A stays in its HOLD", on_aim(HD[1]) and not reserve_notes(HD[2]) and limited_events(HD[1]) == 0 and all(cmd(r)["b"] == 0.0 for r in HD[0]) and max(cmd(r)["a"] for r in HD[0]) > 2000.0 and HD[1].maints[0].phase == HOLD) check("non-vacuous: before the fix A sat at 0 W, B (at 15 %) gave through " "the loop and the reserve alarm fired (SAFETY-86 rig case 7a)", all(cmd(r)["a"] == 0.0 for r in HD_OLD[0]) and max(cmd(r)["b"] for r in HD_OLD[0]) > 2000.0 and len(reserve_notes(HD_OLD[2])) >= 1) HD65 = run(6500.0, HOLD, LOW_B) HD65_OLD = without_fix(lambda: run(6500.0, HOLD, LOW_B)) show("house 6500", *HD65) show("house 6500, before the fix", *HD65_OLD) _n = reserve_notes(HD65[2]) check("⭐ house 6500 W, A alone (2500 W) cannot hold the aim: A gives all it " "has, B still gives nothing, and the reserve alarm fires ONCE naming B", all(cmd(r)["b"] == 0.0 for r in HD65[0]) and max(cmd(r)["a"] for r in HD65[0]) == 2500.0 and len(_n) == 1 and "inverter-b is at its peak reserve" in _n[0]["message"] and "inverter-a" not in _n[0]["message"]) check("...and the quarter is no worse than before the fix (4051 W in the " "SAFETY-86 review), with B kept at its reserve", max(banked(HD65[1])) <= max(banked(HD65_OLD[1])) + 1.0) BK = run(6000.0, HOLD, LOW_B, month_peak=4000.0) show("house 6000, month peak 4000", *BK) check("⭐ under a banked month peak of 4000 W the aim is 3950 W (peak - " "margin): A gives, B nothing, every quarter on 3950 W, no alarm", on_aim(BK[1], 3950.0) and not reserve_notes(BK[2]) and all(cmd(r)["b"] == 0.0 for r in BK[0])) # ============================================================================ # print("\n3. a real loss of peak limiting still alarms") BOTH = run(5500.0, IDLE, {"a": 15.0, "b": 12.0}) show("both under", *BOTH) _nb = reserve_notes(BOTH[2]) check("⭐ both batteries under the reserve: the reserve alarm fires and " "names both, the guard asks for nothing (claim lo 0 W, 'SoC floor')", len(_nb) >= 1 and all("inverter-a and inverter-b" in x["message"] for x in _nb) and all(not s[2] for s in BOTH[3] if s and s[0]) and any(c[0] == "peak" and c[4] == 0.0 and c[6] == "SoC floor" for r in BOTH[0] for c in r[-1][0])) _ctl2 = BOTH[1] for _inv, _s, _r in zip(_ctl2.inverters, (15.0, 60.0), (0, 1)): _inv.soc, _inv.refused = _s, _r # SAFETY-73: B refuses to give _ref = _ctl2._peak_on_able(_lim_all := peakguard.PeakDecision( engaged=True, floor_w=0.0, ceiling_w=3500.0, ceiling_from=peakguard.CEILING_CAP, projected_w=4000.0, excess_w=550.0, wanted_w=2000.0, target_w=3450.0, floor_limited=True, soc=15.0, soc_floor=20.0, soc_floor_reason=peakguard.FLOOR_AT_SOC_FLOOR, reason="PEAK_GUARD_FLOOR_LIMITED: x"), [MaintResult(IDLE, False, None, False, [])] * 2) check("⭐ A under the reserve and B refusing to give (SAFETY-73): nothing can " "give for the peak, so it stays limited at 0 W (the alarm fires) and " "only A is named (test_peakguard.py part 12h drives the notification)", _ref.floor_limited and _ref.floor_w == 0.0 and _ref.held == (A,) and _ref.able_w is None and _ctl2._peak_reserve() == frozenset()) _ctl2.inverters[1].refused = 0 _ctl2.live_units = lambda: (A,) # B dropped from the split _drop = _ctl2._peak_on_able(_lim_all, [MaintResult(IDLE, False, None, False, [])] * 2) del _ctl2.live_units check("...and the same with B dropped (not live): nothing can give, the " "alarm state stands at 0 W, only A named", _drop.floor_limited and _drop.floor_w == 0.0 and _drop.held == (A,) and _drop.able_w is None) async def _notify(ctl, socs, cutouts, refused): """The shipped check_peak_floor_limited() on `ctl`, its notification.""" sent = [] async def call(*a): sent.append(a) return True for inv, so, cut, ref in zip(ctl.inverters, socs, cutouts, refused): inv.soc, inv.bms_soc_floor, inv.refused = so, cut, ref real, ctl.hass.call = ctl.hass.call, call ctl.peak_guard = ctl._peak_on_able(dataclasses.replace( _lim_all, soc=min(socs), soc_floor=peakguard.binding_soc_floor( ctl.packs, 20.0)), [MaintResult(IDLE, False, None, False, [])] * 2) ctl._peak_floor_limited = False try: await ctl.check_peak_floor_limited() finally: ctl.hass.call = real for inv in ctl.inverters: inv.bms_soc_floor, inv.refused = 10, 0 return [a[2]["message"] for a in sent if a[2].get("notification_id") == messages.ID_PEAK_RESERVE] # Cut-outs A 10 % / B 35 % (simulated; rule 8 refuses it, sim/inject.py does # not run rule 8), A at 25 % refusing, B at 30 %. Only B is at its own floor; # A is above its own but under the site's highest (35 %). _msg = asyncio.run(_notify(_ctl2, (25.0, 30.0), (10, 35), (1, 0))) print(f" differing cut-outs: {_msg}") check("⭐ the alarm names the battery at ITS OWN floor (B, 30 % <= 35 %), not " "every battery under the site's highest floor (A at 25 % is above its " "own 20 %)", len(_msg) == 1 and "inverter-b is at its peak reserve" in _msg[0] and "inverter-a" not in _msg[0]) # ============================================================================ # print("\n4. the reserve alarm at the quarter's edge") def spiky(t): """A 5000 W house (B alone can hold the aim) with a 5 s spike to 11000 W every 150 s, which B alone cannot.""" return 11000.0 if t % 150 < 5 else 5000.0 LOW_A_IDLE = {"a": 15.0, "b": 60.0} SP = run(spiky, IDLE, LOW_A_IDLE) _flips = sum(1 for s in SP[3] if s and s[0]) show("spiky house", *SP) print(f" {_flips} cycles floor-limited (B alone short during a spike)") check("⭐ a 5 s spike that B alone cannot cover makes the guard limited for a " "cycle, and that does not fire the alarm: the miss must hold for " "ALARM_HOLD_S, as for 'Load too high'. No quarter goes over the aim", _flips >= 5 and not reserve_notes(SP[2]) and not any(s[4] for s in SP[3] if s) and len(banked(SP[1])) >= 2 and max(banked(SP[1])) <= AIM) def ungated(fn): real = Controller._reserve_short Controller._reserve_short = lambda self, g: True try: return fn() finally: Controller._reserve_short = real _ug = ungated(lambda: run(spiky, IDLE, LOW_A_IDLE)) check("non-vacuous: without the hold the same run raises the alarm", len(reserve_notes(_ug[2])) >= 1) # ============================================================================ # print("\n5. place_on_able() at its boundaries") _lim = peakguard.PeakDecision( engaged=True, floor_w=0.0, ceiling_w=3500.0, ceiling_from=peakguard.CEILING_CAP, projected_w=4000.0, excess_w=550.0, wanted_w=2000.0, target_w=3450.0, floor_limited=True, soc=15.0, soc_floor=20.0, soc_floor_reason=peakguard.FLOOR_AT_SOC_FLOOR, reason="PEAK_GUARD_FLOOR_LIMITED: x") def place(d=_lim, held=("b",), able_w=2000.0, soc=60.0, floor=20.0): return peakguard.place_on_able(d, held, able_w=able_w, soc=soc, soc_floor=floor) _eq, _short = place(), place(able_w=1999.0) check("⭐ the others can give exactly the need (2000 W): not limited, floor " "2000 W, the claim reads 'quarter peak' (the page says 'the peak " "limit')", not _eq.floor_limited and _eq.engaged and _eq.floor_w == 2000.0 and _eq.held == ("b",) and _eq.able_w == 2000.0 and peakguard.claim(_eq).lo == 2000.0 and peakguard.claim(_eq).reason == "quarter peak" and _eq.reason.startswith("PEAK_GUARD: ")) check("...1 W short: limited, but the claim still asks them for all they " "have (1999 W), not 0 W", _short.floor_limited and _short.floor_w == 1999.0 and peakguard.claim(_short).lo == 1999.0 and peakguard.claim(_short).reason == "quarter peak" and _short.reason.startswith("PEAK_GUARD_FLOOR_LIMITED: ")) check("...the others' own permission is graded: at or under their floor, or " "unread, nothing is asked (0 W, 'SoC floor')", all(place(soc=s).floor_w == 0.0 and place(soc=s).floor_limited and peakguard.claim(place(soc=s)).reason == "SoC floor" for s in (20.0, 19.0, None))) check("...the roster's verdict is kept as decide() echoed it (soc 15 %, " "floor 20 %), and a decision that was not limited is only labelled", _eq.soc == 15.0 and _eq.soc_floor == 20.0 and all(place(d=d) == dataclasses.replace(d, held=("b",)) for d in (dataclasses.replace( _lim, floor_limited=False, engaged=False, floor_w=None), dataclasses.replace(_lim, wanted_w=None)))) check("...and at N=1 the floor-limited claim is unchanged: 0 W, 'SoC floor'", peakguard.claim(_lim).lo == 0.0 and peakguard.claim(_lim).reason == "SoC floor") def three(socs, cuts): """`_peak_on_able()` on a stub roster of three (SIMULATED packs).""" names = ("x", "y", "z") invs = [types.SimpleNamespace(name=n, refused=0, rating_w=lambda: 2500.0) for n in names] stub = types.SimpleNamespace( inverters=invs, o={"peak_soc_floor": 20}, packs=[peakguard.Pack(n, so, c) for n, so, c in zip(names, socs, cuts)], live_units=lambda: names) return Controller._peak_on_able(stub, _lim, [MaintResult( IDLE, False, None, False, [])] * 3) # x 25 % (cut-out 10 -> floor 20), y 40 % (cut-out 30), z 15 % (held). x and # y each clear their own floor, but 25 % is not above y's 30 %: the pair is # graded min SoC against max floor (SAFETY-32), so nothing is asked. _t3 = three((25.0, 40.0, 15.0), (10, 30, 10)) _t3ok = three((35.0, 40.0, 15.0), (10, 30, 10)) check("⭐ three batteries, z held: the two others are graded as a pair, the " "EMPTIEST of them (25 %) against the HIGHEST of their floors (30 %), " "so nothing is asked; at 35 % they are asked for the need", _t3.held == ("z",) and _t3.floor_limited and _t3.floor_w == 0.0 and _t3ok.held == ("z",) and not _t3ok.floor_limited and _t3ok.floor_w == 2000.0) # ============================================================================ # print("\n6. the allocator: a held battery gives last") def split(target, reserve, modes=(), floor=1500.0): cl = [Claim.set("loop", 10, target), Claim.limit("supervised", P_SAFETY, -2500.0, 2500.0, actuator=A), Claim.limit("supervised", P_SAFETY, -2500.0, 2500.0, actuator=B), Claim.limit("peak", P_PEAK, lo=min(target, floor))] cl += [Claim.limit("mode", P_MODE, 0.0, 0.0, "off", actuator=u) for u in modes] r = resolve(cl, units=(A, B), live=(A, B), weights={A: 1.0, B: 3.0}, reserve=reserve) return {u: round(v) for u, v in r.shares.items()} check("⭐ B held: A carries the site target first (B's weight 3.0 ignored) " "and B only takes what A cannot: 1500 -> A 1500 / B 0, 4000 -> A 2500 " "/ B 1500", split(1500.0, {B}) == {A: 1500, B: 0} and split(4000.0, {B}) == {A: 2500, B: 1500}) check("...no one held: the weights decide, as before", split(1500.0, set()) == {A: 375, B: 1125}) check("...charging is not touched: a held battery may still charge first", split(-2000.0, {B}) == split(-2000.0, set())) check("⭐ A Off and B under the reserve, a 3000 W floor B cannot place: the " "Off battery gives first (2500 W) and B only the last 500 W - not an " "equal 1500/1500 split", split(3000.0, {B}, modes=(A,), floor=3000.0) == {A: 2500, B: 500}) check("...OPEN, recorded: when B alone CAN place the floor, OPT-11 keeps the " "Off battery resting and B gives it (1500 W). See the DECISIONS entry", split(1500.0, {B}, modes=(A,)) == {A: 0, B: 1500}) check("...and with B Off and nobody held, Off still gives way to the other " "battery, as before (OPT-11)", split(1500.0, set(), modes=(B,)) == {A: 1500, B: 0}) # ============================================================================ # print("\n7. a held battery draining for maintenance") _ctl = HD[1] def able_w(results, socs=(15.0, 60.0)): for inv, s in zip(_ctl.inverters, socs): inv.soc, inv.refused = s, 0 d = _ctl._peak_on_able(dataclasses.replace( _lim, wanted_w=6000.0), results) return d.able_w, d.held, d.floor_w _idle = [MaintResult(IDLE, False, None, False, [])] * 2 _drain = [MaintResult(DRAIN, True, 2000.0, False, []), MaintResult(IDLE, False, None, False, [])] print(f" A at 15 % idle: {able_w(_idle)}; draining 2000 W: " f"{able_w(_drain)}") check("⭐ A under the reserve but draining 2000 W for its monthly cycle: " "that drain still counts (able 4500 W = B 2500 + 2000), so B is not " "asked to carry it; idle, only B's 2500 W", able_w(_idle) == (2500.0, (A,), 2500.0) and able_w(_drain) == (4500.0, (A,), 4500.0)) # ============================================================================ # print("\n8. the page while a peak alarm stands and control goes on") _rows65, _c65 = CH65[0], CH65[1] _c65._peak_floor_limited = True _c65.peak_guard = dataclasses.replace( _lim, floor_w=2500.0, held=(A,), able_w=2500.0) _pay = _c65.status_payload() _txt = json.dumps(_pay, ensure_ascii=False) _row = [r for r in _pay["checks"] if r.get("id") == "peak_guard"] print(f" banner: {_pay['banner']}; row: {_row[0]['text'] if _row else '-'}") check("⭐ the banner is the alarm's title without 'Not ready', and nothing " "on the page says the inverters go to 0 W or that an alarm stopped " "control", _pay["banner"] == messages.T_PEAK_RESERVE and _pay["level"] == "bad" and "Not ready" not in _txt and "until the alarm clears" not in _txt and "An alarm stopped control" not in _txt) check("...the row names A and says the others cannot hold the limit alone, " "in three short sentences", len(_row) == 1 and _row[0]["text"].startswith( "inverter-a battery at its peak reserve. The others cannot hold " "the 3 500 W peak limit alone.")) _c65._peak_floor_limited = False _blip = [r for r in _c65.checks() if r.get("id") == "peak_guard"] _c65._peak_floor_limited = True check("...and while the alarm is not latched (a one-cycle miss) the row is " "the ordinary amber guard row, not the red reserve row", len(_blip) == 1 and _blip[0]["warn"] and not _blip[0].get("runs") and "peak reserve" not in _blip[0]["text"]) _c65.auto = False _stopped = _c65.status_payload() _c65.auto = True check("...a stopped controller still says so: the peak alarm does not take " "the banner from it", _stopped["banner"] != messages.T_PEAK_RESERVE and _stopped["level"] != "bad") # ============================================================================ # print("\n9. N=1 is byte-identical to release/1.0 b8ddfd7") PINNED = { # tools/safety87_identity.py on b8ddfd7 "n1_under": "d2395ef8d74fb23607ad864d5a4bf62c03044bbd9367e873675d7ad10e8c169f", "n1_crossing": "97f7321017d9d34ee17a40dfd4abf73d915f25b9faef086a483d2b5cd124b198", "n1_above": "45726a285aa84f40542d56482652397c3c1e212ba4fe2009f6cc52f537ccad52", "n1_under_banked": "a2876c7cc873bc4c974d2b4bd98e23bba2d007b84545587d2fe0b713b0a6249b", "n1_charge_low": "91336e3c0314985a3b00f314c79b5ffb5a740d218de3ad44f88a4db6e4235b32", "n1_hold": "0a1df347754c69166b20d04c15ee878bbcefb02abb824b9b452334f68f03e46d", } _now = s87.digests() for k in PINNED: print(f" {k} {'same' if _now[k] == PINNED[k] else 'DIFFERS'}") check("⭐ every N=1 digest (claims, commands, the guard's decision and the " "log) equals b8ddfd7's, under, crossing and above the reserve, in a " "charge and a HOLD, and under a banked peak", _now == PINNED) # ============================================================================ # print("\n10. review: maintenance pins beside a held battery, N=3, the page") def _drain_store(): s = rig._Store() s.data[unit_key(B, "phase")] = DRAIN s.data[unit_key(B, "phase_started")] = rig.BASE.isoformat() s.data[unit_key(A, "last_completed")] = rig.BASE.isoformat() return s def _drain_run(): return run(8000.0, IDLE, {"a": 15.0, "b": 60.0}, store=_drain_store(), dur=900.0) DR = _drain_run() _real_dor = Controller._drain_over_reserve _real_rise = Controller._drain_rise # SAFETY-91 would raise B as well Controller._drain_over_reserve = lambda self, claims, results: None Controller._drain_rise = lambda self, claims, results: None try: DR_OLD = _drain_run() finally: Controller._drain_over_reserve = _real_dor Controller._drain_rise = _real_rise def _a_for_peak(rows): """A's command on the cycles where the site target is the peak floor and B alone (2500 W) could place it.""" return [cmd(r)["a"] for r in rows if r[-1][1] and r[-1][1][1] and r[1] <= 2500.0 and abs(r[1] - r[-1][1][1]) < 1.0] print(f" B draining, A held: A for the peak {_a_for_peak(DR[0])[:4]}, " f"without the raise {_a_for_peak(DR_OLD[0])[:4]}") check("⭐ B draining (2000 W pin) while A is held: B is raised to its rating " "for the peak floor, so A gives nothing for it (peak > drain)", DR[1].maints[1].phase == DRAIN and _a_for_peak(DR[0]) and all(a <= 0.0 for a in _a_for_peak(DR[0]))) check("non-vacuous: without the raise A gave the floor's rest above B's pin", any(a > 0.0 for a in _a_for_peak(DR_OLD[0]))) def _stub(names, reserve, live=None): """A Controller stand-in for the pin rules, on SIMULATED units.""" s = types.SimpleNamespace( inverters=[types.SimpleNamespace(name=n) for n in names], actuator_of=lambda src, inv=None: inv.name, live_units=lambda: tuple(live or names), _peak_reserve=lambda: frozenset(reserve)) s._held_pin = lambda cl, u: Controller._held_pin(s, cl, u) return s def _hold(floor, names, reserve, extra=(), results=None): """`_peak_over_hold()`: x HOLDs; returns (x's pin, resolved shares).""" s = _stub(names, reserve) cl = [Claim.limit("supervised", P_SAFETY, -2500.0, 2500.0, actuator=n) for n in names] cl += [Claim.set("maintenance", P_MAINTENANCE, 0.0, HOLD, actuator="x"), Claim.limit("peak", P_PEAK, lo=floor, reason="quarter peak"), *extra] res = results or ([MaintResult(HOLD, True, 0.0, False, [])] + [MaintResult(IDLE, False, None, False, [])] * (len(names) - 1)) hi = Controller._peak_over_hold(s, cl, res) if hi is not None: cl.append(Claim.limit("hold", P_MAINTENANCE, hi=hi)) r = resolve(cl + [Claim.set("loop", 10, floor)], units=tuple(names), live=set(names), reserve=reserve) pin = next((c.value for c in cl if c.source == "maintenance" and c.actuator == "x"), None) return pin, {u: round(v) for u, v in r.shares.items()} _h3 = _hold(3000.0, ("x", "y", "z"), {"z"}) _h3b = _hold(2000.0, ("x", "y", "z"), {"z"}) print(f" N=3 HOLD x, y able, z held: floor 3000 {_h3}, 2000 {_h3b}") check("⭐ HOLD at N=3 with one held: the able battery gives first, the " "holding one the rest (3000 -> y 2500, x 500), the held one nothing", _h3 == (500.0, {"x": 500, "y": 2500, "z": 0}) and _h3b == (0.0, {"x": 0, "y": 2000, "z": 0})) _hd = _hold(2000.0, ("x", "z"), {"z"}, (Claim.set("maintenance", P_MAINTENANCE, 1500.0, DRAIN, actuator="z"),), [MaintResult(HOLD, True, 0.0, False, []), MaintResult(DRAIN, True, 1500.0, False, [])]) _hp = _hold(2000.0, ("x", "z"), {"z"}, (Claim.set("refused", P_SAFETY, -50.0, "probe", actuator="z"),)) print(f" held z draining 1500: {_hd}; held z charge probe: {_hp}") check("⭐ a held battery's own pin counts: draining 1500 W, the hold gives " "only the 500 W rest (no over-placement); a charge probe leaves the " "hold at 0 W (never split both ways, no park)", _hd == (500.0, {"x": 500, "z": 1500}) and _hp[0] == 0.0) _unk = three((60.0, 60.0, 60.0), (10, None, 10)) _unr = three((60.0, None, 60.0), (10, 10, 10)) check("⭐ N=3: an unknown cut-out or an unread level holds that battery " "only; the two others are asked for the need", _unk.held == ("y",) and not _unk.floor_limited and _unk.floor_w == 2000.0 and _unr.held == ("y",) and not _unr.floor_limited and _unr.floor_w == 2000.0) _stop_row = {"ok": False, "warn": False, "id": "meter", "text": "Meter down."} _peak_rows = [r for r in _c65.checks() if r.get("id") == "peak_guard"] _c65.checks = lambda: _peak_rows[:1] + [_stop_row] _both = _c65.status_payload() del _c65.checks check("⭐ a row that stops control outranks the running peak alarm: the " "banner says 'Not ready' for it and the Home page shows the hold", len(_peak_rows) == 1 and _peak_rows[0].get("runs") and _both["banner"].startswith(messages.NOT_READY) and "Meter down" in _both["banner"] and "until the alarm clears" in json.dumps(_both, ensure_ascii=False)) for _inv, _s, _r in zip(_ctl2.inverters, (15.0, 60.0), (0, -1)): _inv.soc, _inv.refused = _s, _r # B refuses only to CHARGE _chg = _ctl2._peak_on_able(_lim_all, [MaintResult(IDLE, False, None, False, [])] * 2) _ctl2.inverters[1].refused = 0 check("⭐ a battery that refuses only to charge can still give: it stays " "able and carries the floor (2000 W of 2500 W)", not _chg.floor_limited and _chg.floor_w == 2000.0 and _chg.able_w == 2500.0 and _chg.held == (A,)) def _short_for(held_w, secs=30.0): """`_reserve_short()` over `secs` of a steady quarter: the meter at `held_w` with the able batteries already at their full 2500 W.""" s = types.SimpleNamespace(_reserve_batt_prev=None, _reserve_high_since=None, _peak_floor_limited=False, site_batt_w=2500.0) g = dataclasses.replace(_lim, floor_w=2500.0, able_w=2500.0) out = [] for k in range(int(secs / 5.0) + 1): s._peak_proj = types.SimpleNamespace( block_start=rig.BASE, elapsed_s=450.0 + 5.0 * k, projected_w=held_w, held_w=held_w) out.append(Controller._reserve_short(s, g)) return out _aim_miss, _limit_miss = _short_for(3480.0), _short_for(3600.0) check("⭐ the reserve alarm is graded on the real limit, not the aim: a " "quarter ending at 3480 W (over the 3450 W aim, under 3500 W) never " "alarms; one ending at 3600 W alarms once it holds ALARM_HOLD_S", not any(_aim_miss) and not any(_limit_miss[:2]) and all(_limit_miss[2:])) _runs_src = {} for _f in ("main.py", "homeview.py"): with open(os.path.join(HERE, "app", _f), encoding="utf-8") as _fh: _src = _fh.read() for _part in _src.split('out.append({')[1:]: if '"runs": True' in _part.split("})", 1)[0]: _runs_src.setdefault(_f, []).append( _part.split('"id": "', 1)[-1].split('"', 1)[0]) check("⭐ 'control still runs' is set on the two peak rows only (saturated " "and reserve); no row that stops control carries it", _runs_src == {"main.py": ["peak_guard", "peak_guard"]}) print() if fails: print(f"{len(fails)} FAILED: " + ", ".join(fails)) sys.exit(1) print("all SAFETY-87 checks passed")