"""Runnable check for SAFETY-95: a maintenance drain that ends hands back smoothly at N>=2. `python3 test_safety95.py` At N=2, when battery A's drain ended (A switched Off, say) with the house at 1200 W, the drain's floor stayed in the loop's site target and landed on B: B went 0 -> +1600 W in one cycle and exported. Now the loop starts from what the others stood at plus A's new command, raised to the house's own point when the house needs more (never above last cycle's start), and counts every battery at its command while A settles. Drives the REAL Controller on a lagging battery (10 s, 2 s meter, as on the rig), at N=2 and N=3. DECISIONS/SAFETY-95-drain-end.md. 1. Every drain end (complete -> charge, Off, timeout, operator abort, stale link, refusal) at houses of 1200 W and 4000 W and under a solar surplus. 2. The peak guard still binds across a drain end. 3. A drain end in the same cycle as another battery's drop; a missed poll. 4. N=1 is untouched; the rule at its boundaries. Mutations (MUTATION.md, purged copies) are listed in the pull request. """ import asyncio import functools import hashlib import logging import math import os import sys import tempfile from datetime import timedelta 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 safety79_identity as s79 # noqa: E402 from app import main as app_main, packmode # noqa: E402 from app.arbiter import Claim, P_MAINTENANCE, P_SAFETY # noqa: E402 from app.main import Controller # noqa: E402 from app.maintenance import (CHARGE, DRAIN, DRAIN_TIMEOUT_MIN, # noqa: E402 HOLD, IDLE, MaintResult) from app.store import unit_key # noqa: E402 fails = [] SLEW = 400.0 # the rig's slew_w TOL = SLEW / 4 # the loop's own overshoot on a lagging battery AIM = -10.0 # the rig's target_grid_w T = 300.0 # the drain ends here M = {"maintenance_enabled": True, "maintenance_start_hour": 11} A = "inverter-a" def check(name, cond): print(f" {'ok ' if cond else 'FAIL'} {name}") if not cond: fails.append(name) # None on a tree without the fix, where this file prints the base numbers # and runs red. fixed = getattr(Controller, "_reseed_drain_end", None) real_grid = getattr(Controller, "_drain_grid_w", None) real_drop = Controller._reseed_drop def _base_drop(self, live): """SAFETY-96's drop reseed as it was: a draining battery is not skipped.""" keep, self._drain_ended = self._drain_ended, frozenset() try: real_drop(self, live) finally: self._drain_ended = keep def without(fn): """Run `fn` with SAFETY-95 taken out (the base behaviour).""" if fixed is None: return fn() Controller._reseed_drain_end = lambda self, r, live: setattr( self, "_drain_ended", frozenset()) Controller._drain_grid_w = lambda self, live: float(self.grid) Controller._reseed_drop = _base_drop try: return fn() finally: Controller._reseed_drain_end = fixed Controller._drain_grid_w = real_grid Controller._reseed_drop = real_drop def options(n, extra): o = rig._options(n, extra) if n > 1: o["inverters"] = [ {"name": f"inverter-{u}", "setpoint_entity": f"number.sp_{u}", "soc_entity": f"sensor.soc_{u}", "batt_entity": f"sensor.batt_{u}", "max_w": 2500, "bms_soc_floor": 10} for u in "abc"[:n]] return o def run(n, house, *, store, extra=None, soc=None, reject=None, dead=None, before=None, dur=T + 120.0, lag=10.0, p1=False, back=None): """Rows of one run at N=`n` on plantsim's battery (800 W/s ramp, `lag` s first-order lag) at a 2 s meter tick. Writes to the units in `reject` (name -> from s) are refused, until `back` s; `dead` (u, from s[, to s]): u gives nothing from then on (until `to`). Each row: (t, the add-on's commands, site target, grid, phases, loop reason, peak floor). opt11_identity's cycle, with a third battery.""" units = tuple("abc"[:n]) soc = soc or (lambda u, t: 50.0) def refused(u, t): return bool(reject) and u in reject and t >= reject[u] and ( back is None or t < back) class Hass(rig._Hass): def power(self, u): st = self.__dict__.setdefault("_plant", {}) a, tgt, tl = st.get(u, (0.0, 0.0, self.t)) for k in range(int(round((self.t - tl) / 0.1))): now = tl + (k + 1) * 0.1 c = self.cmd[u] if dead is not None and u == dead[0] and dead[1] <= now < ( dead[2] if len(dead) > 2 else math.inf): c = 0.0 d = c - tgt tgt += math.copysign(min(abs(d), 80.0), d) a += (tgt - a) * (1 - math.exp(-0.1 / lag)) st[u] = (a, tgt, self.t) return float(round(a)) async def set_number(self, e, v): if refused(e[-1], self.t): return False self.cmd[e[-1]] = float(v) return True opts = options(n, dict(M, **(s79.P1 if p1 else {}), **(extra or {}))) db = os.path.join(tempfile.mkdtemp(prefix="s95-"), "tariff.db") async def go(): hass = Hass(units, soc, lambda u, t: False, lambda t: house(t) if callable(house) else house) c = Controller(opts, hass, store, rig._Pub(), stop=asyncio.Event(), peak_db_path=db) for i in c.inverters: i.dev_min, i.dev_max = -5000.0, 5000.0 c.auto = True real_dt, clock = app_main.datetime, [0.0] class _DT(real_dt): @classmethod def now(cls, tz=None): return rig.BASE + timedelta(seconds=clock[0]) app_main.datetime = _DT rows = [] try: t = 0.0 while t < dur: hass.t = clock[0] = t for inv in c.inverters: u = inv.name[-1] if refused(u, t) and inv.rejected_since is None: inv.rejected_since = rig.BASE + timedelta(seconds=t) if before is not None: r = before(c, t) if asyncio.iscoroutine(r): await r if p1: s79._feed(c, t) await c.read_inputs(now_mono=t) await c.cycle() g = c.peak_guard rows.append((t, tuple(i.target for i in c.inverters), c.target, hass.grid(), tuple(m.phase for m in c.maints), c.loop_reason, None if g is None or not g.engaged else g.floor_w)) t += 2.0 finally: app_main.datetime = real_dt return rows, c return asyncio.run(go()) def store(n, **a): """A due for its monthly cycle (it drains from the first cycle); the others finished theirs, so none starts here.""" s = rig._Store() if n == 1: for k, v in a.items(): s.data[k] = v return s for u in "bc"[:n - 1]: s.data[unit_key(f"inverter-{u}", "last_completed")] = \ rig.BASE.isoformat() for k, v in a.items(): s.data[unit_key(A, k)] = v return s def timed_store(n): """A's drain passes DRAIN_TIMEOUT_MIN at T; its last start was at 0 s, so the abort's cooldown keeps it from starting again.""" return store(n, phase=DRAIN, phase_started=( rig.BASE + timedelta(seconds=T - DRAIN_TIMEOUT_MIN * 60 - 1) ).isoformat(), last_start_attempt=rig.BASE.isoformat()) def off(c, t): if t >= T: c.store.data[unit_key(A, packmode.USE_KEY)] = False async def abort(c, t): if t == T: await c.handle_action("maint_abort") ENDS = { "complete": lambda n, h, **k: run( n, h, store=store(n), soc=lambda u, t: 5.0 if (u == "a" and t >= T) else 50.0, **k), "Off": lambda n, h, **k: run(n, h, store=store(n), before=off, extra={"battery_modes_enabled": True}, **k), "timeout": lambda n, h, **k: run(n, h, store=timed_store(n), **k), "abort": lambda n, h, **k: run(n, h, store=store(n), before=abort, **k), # A's link goes stale; its board's watchdog sends it to 0 W after 25 s. "stale link": lambda n, h, **k: run(n, h, store=store(n), reject={"a": T}, dead=("a", T + 25.0), dur=T + 200.0, **k), # A gives nothing from T; SAFETY-73 learns it ~55 s later. "refusal": lambda n, h, **k: run(n, h, store=store(n), dead=("a", T), dur=T + 300.0, **k), } HOUSES = (1200.0, 4000.0, -1500.0) def end_at(rows): """The first cycle A is not draining, or its command left the drain; None if its drain never ended.""" return next((r[0] for r in rows if r[0] >= T and ( r[4][0] != DRAIN or r[1][0] < 2000.0)), None) def measure(rows, n, house): """From the end on: the site target's largest step after the end cycle (the slew's subject); the commanded export in the 20 s after it (the site command over what the house needs, or over what the others stood at); the largest grid import; the export beyond what the site exported before the end; the largest grid step; the others' largest command step.""" te = end_at(rows) if te is None: # no end: every measure is NaN, so every check FAILs return dict(end=math.nan, site_step=math.nan, cmd_export=math.nan, imp=math.nan, exp=math.nan, gstep=math.nan, ostep=math.nan, rows=rows) pre = next((r for r in reversed(rows) if r[0] < te), rows[0]) seg = [r for r in rows if r[0] >= te - 2.0] aft = [r for r in rows if te <= r[0] <= te + 120.0] others0 = sum(pre[1][1:]) site = [r[2] for r in seg] return dict( end=te, site_step=max(abs(b - a) for a, b in zip(site[1:], site[2:])), cmd_export=max(sum(r[1]) - max(house - AIM, others0) for r in aft if r[0] <= te + 20.0), imp=max(r[3] for r in aft), exp=min(min(pre[3], AIM) - r[3] for r in aft), gstep=max(abs(b[3] - a[3]) for a, b in zip(seg, seg[1:])), ostep=max(abs(b[1][i] - a[1][i]) for a, b in zip(seg, seg[1:]) for i in range(1, n)), rows=rows) TABLE = [] def case(kind, n, house): now = measure(ENDS[kind](n, house)[0], n, house) old = measure(without(lambda: ENDS[kind](n, house))[0], n, house) TABLE.append((kind, n, house, now, old)) return now, old # ============================================================================ # print("1. every drain end, N=2 and N=3, houses 1200 W, 4000 W and -1500 W") for kind in ENDS: for n in (2, 3): for house in HOUSES: now, old = case(kind, n, house) tag = f"{kind}, N={n}, house {house:.0f} W" # Abort zeroes the site (operator); a stale link's start is # SAFETY-96's drop. Those are graded against base. own = kind not in ("abort", "stale link") check(f"{'⭐' if own else ' '} {tag}: no battery gives into an " f"export after the end (commanded export " f"{now['cmd_export']:.0f} W, base {old['cmd_export']:.0f}) " f"and the site target moves at most one slew step a cycle " f"after it ({now['site_step']:.0f} W, base " f"{old['site_step']:.0f})", now["cmd_export"] <= (TOL if own else max( TOL, old["cmd_export"] + TOL)) and now["site_step"] <= SLEW) check(f" {tag}: the grid imports no more than base (+50 W) " f"({now['imp']:.0f} W, base {old['imp']:.0f}) and exports " f"no more than it did before the end, or base (+50 W) " f"({now['exp']:.0f} W more, base {old['exp']:.0f})", now["imp"] <= old["imp"] + 50.0 and now["exp"] <= max(0.0, old["exp"]) + 50.0) _by = {(k, n, h): (a, b) for k, n, h, a, b in TABLE} for kind in ("complete", "Off", "timeout", "refusal"): _now, _old = _by[(kind, 2, 1200.0)] check(f"non-vacuous ({kind}, N=2, 1200 W): without the fix B steps " f"more than a slew step into an export", _old["cmd_export"] >= 300.0 and _old["ostep"] > SLEW and _now["ostep"] < _old["ostep"]) _now, _old = _by[(kind, 2, -1500.0)] check(f"non-vacuous ({kind}, solar surplus): without the fix B gives " f"into the surplus", _old["cmd_export"] >= 1000.0 and _now["cmd_export"] <= TOL) _now, _old = _by[("complete", 2, 1200.0)] check("⭐ the rig case (A's drain completes at 1200 W): B covers the house " "at once (0 -> 1210 W, not 1600), and the grid stays within 50 W of " "its aim while A settles (base: +559 W import, B down to 230 W)", [r[1][1] for r in _now["rows"] if r[0] == _now["end"]] == [1210.0] and _now["imp"] <= 50.0 and _old["imp"] > 400.0) _now, _old = _by[("complete", 3, 4000.0)] check("N=3 at 4000 W: the others take A's share at once, as before (the " "site does not drop), so the grid does not import", _now["imp"] <= 50.0 and abs(_now["imp"] - _old["imp"]) <= 50.0) _now, _old = _by[("stale link", 3, 4000.0)] check(f"⭐ stale link, N=3, 4000 W: a dropped draining battery is this " f"rule's, not the link-drop reseed's, so the grid imports at least " f"50 W less than base ({_now['imp']:.0f} W, base {_old['imp']:.0f})", _now["imp"] <= _old["imp"] - 50.0) # ============================================================================ # print("\n2. the peak guard still binds across a drain end") _g = ENDS["Off"](2, 4000.0, p1=True, dur=T + 120.0)[0] _gs = [r for r in _g if r[0] >= T - 20 and r[6] is not None] print(f" guard engaged on {len(_gs)} cycles around the end; site " f"{[sum(r[1]) for r in _g if T - 4 <= r[0] <= T + 8]}") check("⭐ house 4000 W over a 2500 W limit, A switched Off mid-drain: on " "every cycle the guard is engaged, the site gives at least its floor " "(or every battery is at its rating)", _gs and all(sum(r[1]) >= min(r[6], 2500.0 * 1) - 1.0 for r in _gs)) # ============================================================================ # print("\n3. a drain end with another edge in the same cycle; a missed poll") def c_drops(c, t): """C's writes are refused from T - 30 s, so it drops at T, the cycle A is switched Off; its board goes to 0 W at once (T - 30 s).""" off(c, t) _e3 = run(3, 4000.0, store=store(3), before=c_drops, reject={"c": T - 30.0}, dead=("c", T - 30.0), extra={"battery_modes_enabled": True})[0] _e3m = measure(_e3, 3, 4000.0) print(f" A Off as C drops, house 4000 W: B " f"{[r[1][1] for r in _e3 if T - 4 <= r[0] <= T + 10]}") check("⭐ A switched Off on the cycle C drops (N=3, 4000 W): B takes over " "with no export and the site moves one slew step a cycle after", _e3m["cmd_export"] <= TOL and _e3m["site_step"] <= SLEW and _e3m["end"] == T) def c_back(c, t): """C's link (stale since T - 60 s, its board holding 0 W) answers again at T, the cycle A is switched Off.""" off(c, t) if t == T: c.inverters[2].note_write(True, rig.BASE + timedelta(seconds=t)) _r3 = run(3, 1200.0, store=store(3), before=c_back, reject={"c": T - 60.0}, back=T, dead=("c", T - 60.0, T), extra={"battery_modes_enabled": True})[0] _r3m = measure(_r3, 3, 1200.0) print(f" A Off as C comes back, house 1200 W: B, C " f"{[r[1][1:] for r in _r3 if T - 4 <= r[0] <= T + 6]}") check("⭐ A switched Off on the cycle C comes back (N=3, 1200 W): no " "export, one slew step a cycle after", _r3m["cmd_export"] <= TOL and _r3m["site_step"] <= SLEW and _r3m["end"] == T) def missed_poll(at_t): """Battery B's power reading is missing on the cycle at `at_t` s.""" def before(c, t): off(c, t) if not getattr(c.hass, "_s95", False): read = c.hass.number async def number(e, invert=False): if e == "sensor.batt_b" and c.hass.t == at_t: return None return await read(e, invert) c.hass.number, c.hass._s95 = number, True return before _mp = run(2, 1200.0, store=store(2), before=missed_poll(T), extra={"battery_modes_enabled": True})[0] _mpo = without(lambda: run(2, 1200.0, store=store(2), before=missed_poll(T), extra={"battery_modes_enabled": True}))[0] print(f" a missed poll on the end cycle: B " f"{[r[1][1] for r in _mp if T - 2 <= r[0] <= T + 8]}, without the fix " f"{[r[1][1] for r in _mpo if T - 2 <= r[0] <= T + 8]}") check("⭐ a missed poll on the cycle A is switched Off: the end still " "reaches the loop on the next cycle (B covers the house, no export)", max(sum(r[1]) for r in _mp if r[0] > T) <= 1200.0 - AIM + TOL and max(sum(r[1]) for r in _mpo if r[0] > T) >= 1500.0) # ============================================================================ # print("\n4. N=1 is untouched; the rule at its boundaries") def digest(rows): return hashlib.sha256(repr(rows).encode()).hexdigest() _n1 = [functools.partial(ENDS[k], 1, h) for k in ENDS for h in (1200.0, -1500.0)] check("⭐ N=1, every drain end (complete, Off, timeout, abort, stale link, " "refusal), at 1200 W and under a surplus: every row byte-identical " "with and without SAFETY-95", [digest(f()[0]) for f in _n1] == [digest(without(f)[0]) for f in _n1]) _wave = functools.partial(run, 2, lambda t: 2400.0 + 1500 * math.sin(t / 40), store=store(2, last_completed=rig.BASE.isoformat()), dur=600.0) check("N=2 healthy (a moving house, no maintenance): every row " "byte-identical with and without SAFETY-95", digest(_wave()[0]) == digest(without(_wave)[0])) if fixed is None: print(f"\n{len(fails)} FAILED (no _reseed_drain_end): " + ", ".join(fails)) sys.exit(1) _c = run(2, 1200.0, store=store(2), dur=4.0)[1] _pin = MaintResult(IDLE, True, 0.0, False, []) _free = MaintResult(CHARGE, False, None, True, []) _b = MaintResult(IDLE, False, None, False, []) AB = {"inverter-a", "inverter-b"} def reseeded(target, shares, batts, grid, res=_pin, live=AB, ended=(A,)): """`_reseed_drain_end` on a 2-battery controller: (target, i_w, A's start).""" _c.target, _c.i_w, _c._drain_ended = target, 123.0, frozenset(ended) _c._loop_start, _c._settle, _c._drain_tail = {}, {}, set() _c._now_mono, _c.grid = 50.0, grid for inv in _c.inverters: inv.target, inv.batt = shares[inv.name[-1]], batts[inv.name[-1]] _c._reseed_drain_end([res, _b], set(live)) return _c.target, round(_c.i_w, 6), _c._loop_start.get(A) # House 1200 W: A drains 2000, B 0, grid -800. check("the house needs more than the others stand at: the start is the " "house's own point (1210 W), the integrator with it", reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0}, -800.0) == (1210.0, 1210.0, 0.0) and _c._settle == {A: 50.0} and _c._drain_tail == {A} and _c._drain_ended == frozenset()) check("...and while A settles the loop's error is the house's point minus " "its start (0 here: it holds), not the meter's -790 W", _c._drain_grid_w(AB) - AIM == 0.0) check("a solar surplus (house -1500 W): the start is what the others stood " "at plus A's new command (0 W), and the loop moves down from there", reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0}, -3500.0)[0] == 0.0) check("the house needs more than stood placed (4000 W, B at 2000): never " "above last cycle's start", reseeded(4000.0, {"a": 2000.0, "b": 2000.0}, {"a": 2000.0, "b": 2000.0}, 0.0)[0] == 4000.0) check("a free battery (drain complete -> charge): its drain leaves the " "start; a pin's value is its new command", reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0}, -3500.0, res=_free)[2] == 0.0 and reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0}, -3500.0, res=MaintResult(HOLD, True, 0.0, False, []))[2] == 0.0) check("a dropped battery's power is the meter's: a board stuck at 2000 W " "keeps B where it stood, and there is no tail to count", reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0}, -800.0, live={"inverter-b"})[0] == 0.0 and _c._drain_tail == set()) _chg = MaintResult(CHARGE, True, -1500.0, False, []) check("a drain that ends into a pinned charge: the pin is A's new command " "while A is live, but 0 W once it is dropped (it is sent 0 W)", reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0}, -3500.0, res=_chg)[2] == -1500.0 and reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0}, -3500.0, res=_chg, live={"inverter-b"}) == (0.0, 0.0, 0.0)) check("no drain ended: nothing moves (the integrator either)", reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0}, -800.0, ended=()) == (2000.0, 123.0, None)) _c._drain_tail, _c._settle, _c.grid = {A}, {}, -800.0 check("the tail ends with A's settling (SAFETY-96's window)", _c._drain_grid_w(AB) == -800.0 and _c._drain_tail == set()) _UA, _UB = (_c.actuator_of("maintenance", i) for i in _c.inverters) _drain = MaintResult(DRAIN, True, 2000.0, False, []) def drain_ends(extra, refused=0): """`_charge_track` with A draining last cycle, its drain SET and the claims `extra` on A: which drains it ends.""" _c._draining_was, _c._drain_ended = frozenset({A}), frozenset() _c._charging_was, _c._charge_ended = frozenset(), frozenset() _c.live_units = lambda: (_UA, _UB) _c.inverters[0].refused = refused try: _c._charge_track([Claim.set("maintenance", P_MAINTENANCE, 2000.0, actuator=_UA)] + extra, [_drain, _b]) finally: del _c.live_units _c.inverters[0].refused = 0 return set(_c._drain_ended) _rating = [Claim.limit("device", P_SAFETY, -2500.0, 2500.0, actuator=_UA)] check("a drain that still stands does not end (its rating binds, no " "contradiction, no refusal)", drain_ends(_rating) == set()) check("a drain whose limits contradict ends: A is sent 0 W (SAFETY-91's " "pin as a floor against a closed pack)", drain_ends(_rating + [ Claim.limit("maintenance", P_MAINTENANCE, lo=2000.0, actuator=_UA), Claim.limit("pack", P_SAFETY, hi=0.0, actuator=_UA)]) == {A}) check("a refusal ends a drain only if A will not give (+1); one that will " "not take (-1) leaves the drain standing", drain_ends(_rating, refused=1) == {A} and drain_ends(_rating, refused=-1) == set()) _n1c = run(1, 1200.0, store=store(1), dur=4.0)[1] _n1c.target, _n1c.i_w, _n1c._drain_ended = 2000.0, 5.0, frozenset({A}) _n1c._reseed_drain_end([_pin], {A}) check("N=1: nothing moves", (_n1c.target, _n1c.i_w) == (2000.0, 5.0)) _seen = [] def _spy(self, maint_results, live): _seen.append(sorted(self._drain_ended)) return fixed(self, maint_results, live) Controller._reseed_drain_end = _spy try: run(2, 1200.0, store=store(2), soc=lambda u, t: 5.0 if (u == "a" and T <= t < T + 200) else 99.5 if u == "a" and t >= T + 200 else 50.0, dur=T + 400.0, extra={"maintenance_hold_min": 2}) finally: Controller._reseed_drain_end = fixed check("it fires once, at the drain end: not at the charge end, the HOLD " "end, or on any other cycle", [x for x in _seen if x] == [[A]] and len(_seen) > 300) print("\n drain end | N | house W | site step W head/base | commanded " "export W head/base | others' step W head/base | grid import W " "head/base | grid step W head/base | export beyond the end's W head/base") for k, n, h, a, b in TABLE: print(f" {k} | {n} | {h:.0f} | {a['site_step']:.0f}/" f"{b['site_step']:.0f} | {a['cmd_export']:.0f}/{b['cmd_export']:.0f}" f" | {a['ostep']:.0f}/{b['ostep']:.0f} | {a['imp']:.0f}/" f"{b['imp']:.0f} | {a['gstep']:.0f}/{b['gstep']:.0f} | " f"{a['exp']:.0f}/{b['exp']:.0f}") print() if fails: print(f"{len(fails)} FAILED: " + ", ".join(fails)) sys.exit(1) print("all SAFETY-95 checks passed")