"""Runnable check for SAFETY-93: the maintenance charge end hands back smoothly, with no grid charging at a high house. `python3 test_safety93.py` At N=2, when battery A's maintenance charge ended, its watts stayed in the loop's site target and landed on B: B went 0 -> -1600 W in one cycle and charged from the grid with the house at 6000 W (SAFETY-85 rig test, check 4). Now the loop starts from what B stood at plus A's new command. Drives the REAL Controller on a lagging battery (10 s, 2 s meter, as on the rig). DECISIONS/SAFETY-93-charge-end.md. 1. The rig case: A charges 2000 W late in a quarter with the house at 6000 W, then its charge ends (HOLD). 2. The same at a low house. 3. The charge ends another way: a timeout (abort) and the battery switched Off. 4. The charge ends while the guard has suspended it (A gives under the loop). 5. N=1 is untouched; a missed poll on the end cycle; the rule at its boundaries. Mutations (MUTATION.md, purged copies) are listed in the pull request. """ import hashlib import logging import math import os import sys 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 import safety84_identity as s84 # noqa: E402 from app import packmode # noqa: E402 from app.main import Controller # noqa: E402 from app.maintenance import (CHARGE_TIMEOUT_MIN, DRAIN, HOLD, IDLE, # noqa: E402 MaintResult) from app.store import unit_key # noqa: E402 from named_absence import first, present # noqa: E402 fails = [] A = "inverter-a" SLEW = 400.0 # the rig's slew_w AIM = 3500 - 50 # peak_cap_w - peak_margin_w OPT0 = dict(s84.OPT) def check(name, cond): print(f" {'ok ' if cond else 'FAIL'} {name}") if not cond: fails.append(name) def lagging(fn, lag=10.0): """plantsim's battery (800 W/s ramp, `lag` s first-order lag) at the live 2 s meter tick (test_safety85.py part 7).""" real_p, real_t = rig._Hass.power, rig.TICK def power(self, u): st = self.__dict__.setdefault("_plant", {}) a, tgt, tl = st.get(u, (0.0, 0.0, self.t)) for _ in range(int(round((self.t - tl) / 0.1))): d = self.cmd[u] - 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)) rig._Hass.power, rig.TICK = power, 2.0 try: return fn() finally: rig._Hass.power, rig.TICK = real_p, real_t def run(n, house, end, *, step=None, dur=None, store=None, extra=None, before=None): """A charging in a cheap window from the first cycle (quarters start at 0, 900, 1800 s); the house is 1200 W, then `house` from `step`; A reads 99 % (its charge target) from `end`. Returns (rows, controller).""" step = end if step is None else step s84.OPT = dict(OPT0, peak_cap_w=3500) feed = s84.feed(True) def both(c, t): if before is not None: before(c, t) feed(c, t) try: return lagging(lambda: s84.run( n, load=lambda t: house if t >= step else 1200.0, cheap=True, store=store or s79.store_charging(n), soc=lambda u, t: (99.0 if t >= end else 90.0) if u == "a" else 60.0, dur=dur or end + 300.0, before=both, extra=dict({"maintenance_hold_min": 600}, **(extra or {})))) finally: s84.OPT = OPT0 def cmd(r): return dict(r[2]) def moves(rows, u, lo, hi): """The largest change in battery `u`'s command between cycles in (lo, hi].""" return max(abs(cmd(b)[u] - cmd(a)[u]) for a, b in zip(rows, rows[1:]) if lo < b[0] <= hi) def charging_hi(rows, lo, hi, house_at): """Cycles in [lo, hi] where a battery charges while the house is above the peak aim.""" return sum(1 for r in rows if lo <= r[0] <= hi and house_at(r[0]) > AIM and min(cmd(r).values()) < 0.0) # None on a tree without the fix (f3e409c), where this file runs red. no_fix = getattr(Controller, "_reseed_charge_end", None) def without(fn): """Run `fn` with the SAFETY-93 reseed taken out.""" if no_fix is None: return fn() Controller._reseed_charge_end = lambda self, ended: None try: return fn() finally: Controller._reseed_charge_end = no_fix def phase_at(rows, t): return first((r[-1][3][0] for r in rows if r[0] == t), f"row {t}") def at(rows, t, u): """Battery `u`'s command at `t` s.""" r = first((r for r in rows if r[0] == t), f"row {t}") return dict(r[2])[u] if present(r) else r # ============================================================================ # print("1. the rig case: A charges 2000 W late in a quarter, house 6000 W") # House 1200 W until 1760 s, then 6000 W; the quarter (900-1800 s) has room, # so the charge stays at 2000 W. A is full at 1780 s. Graded to 1798 s: the # guard grades a new quarter from 1800 s. R1 = run(2, 6000.0, 1780.0, step=1760.0) R1_old = without(lambda: run(2, 6000.0, 1780.0, step=1760.0)) def hi1(t): return 6000.0 if t >= 1760.0 else 1200.0 for name, (rows, _c) in (("now", R1), ("without the fix", R1_old)): print(f" {name}: " + ", ".join( f"{r[0]:.0f}s A {cmd(r)['a']:.0f} B {cmd(r)['b']:.0f}" for r in rows if 1776 <= r[0] <= 1786)) _r, _ = R1 check("the case is the rig's: A charges 2000 W with the house at 6000 W the " "cycle before its charge ends, then holds", at(_r, 1778.0, "a") == -2000.0 and phase_at(_r, 1778.0) == "charge" and phase_at(_r, 1780.0) == HOLD) check("⭐ no battery charges from the grid after the charge ends (house above " "the aim)", charging_hi(_r, 1780, 1798, hi1) == 0) check("⭐ B moves at most one slew step a cycle across the charge end, and A " "after its HOLD pin", moves(_r, "b", 1776, 1798) <= SLEW and moves(_r, "a", 1780, 1798) <= SLEW) check("...and B gives toward the house at once (+400 W at the end cycle)", at(_r, 1780.0, "b") == SLEW) check("non-vacuous: without the fix B jumps 0 -> -1600 W and charges from " "the grid for 4 cycles", moves(R1_old[0], "b", 1776, 1798) >= 1600 and charging_hi(R1_old[0], 1780, 1798, hi1) >= 4) # ============================================================================ # print("\n2. the same at a low house") R2 = run(2, 1200.0, 1300.0) R2_old = without(lambda: run(2, 1200.0, 1300.0)) print(f" B's largest move: {moves(R2[0], 'b', 1296, 1600):.0f} W, without " f"the fix {moves(R2_old[0], 'b', 1296, 1600):.0f} W; B's least " f"command after the end " f"{min(cmd(r)['b'] for r in R2[0] if r[0] >= 1300):.0f}" f" W, without {min(cmd(r)['b'] for r in R2_old[0] if r[0] >= 1300):.0f}" " W") check("⭐ house 1200 W: B moves at most one slew step a cycle and never " "charges from the grid after A's charge ends", moves(R2[0], "b", 1296, 1600) <= SLEW and min(cmd(r)["b"] for r in R2[0] if r[0] >= 1300) >= 0.0) check("non-vacuous: without the fix B jumps to -1600 W (2800 W from the grid)", min(cmd(r)["b"] for r in R2_old[0] if r[0] >= 1300) <= -1600.0) # ============================================================================ # print("\n3. the charge ends another way") def timed_out(end): """A store whose charge passes CHARGE_TIMEOUT_MIN at `end` s.""" s = s79.store_charging(2) s.data[unit_key(A, "phase_started")] = ( rig.BASE + timedelta(seconds=end - CHARGE_TIMEOUT_MIN * 60 - 1) ).isoformat() return s def b_done(): """A charging; B finished its monthly cycle, so it starts none here.""" s = s79.store_charging(2) s.data[unit_key("inverter-b", "last_completed")] = rig.BASE.isoformat() return s def switch_off(at): """Battery A switched Off at `at` s (OPT-11 modes on).""" def before(c, t): if t >= at: c.store.data[unit_key(A, packmode.USE_KEY)] = False return before def timed_out_b_done(end): """timed_out(), with B's monthly cycle finished so it starts none.""" s = timed_out(end) s.data[unit_key("inverter-b", "last_completed")] = rig.BASE.isoformat() return s ENDS = { "timeout": lambda: run(2, 6000.0, 9999.0, step=1760.0, dur=2080.0, store=timed_out_b_done(1780.0)), "Off": lambda: run(2, 6000.0, 9999.0, step=1760.0, dur=2080.0, store=b_done(), before=switch_off(1780.0), extra={"battery_modes_enabled": True}), } for k, fn in ENDS.items(): rows, c = fn() old = without(fn)[0] print(f" {k}: A {[cmd(r)['a'] for r in rows if 1776 <= r[0] <= 1784]}" f" B {[cmd(r)['b'] for r in rows if 1776 <= r[0] <= 1784]}; " f"without the fix B " f"{[cmd(r)['b'] for r in old if 1776 <= r[0] <= 1784]}") check(f"{k}: the charge ends at 1780 s", phase_at(rows, 1778.0) == "charge" and phase_at(rows, 1780.0) == IDLE) check(f"⭐ {k}: no battery charges from the grid after it, and B moves at " f"most one slew step a cycle", charging_hi(rows, 1780, 1798, hi1) == 0 and moves(rows, "b", 1776, 1798) <= SLEW) check(f"non-vacuous ({k}): without the fix B jumps to -1600 W, from " f"the grid", at(old, 1780.0, "b") == -1600.0 and charging_hi(old, 1780, 1798, hi1) >= 1) # MAINT-13: A's timed-out charge is a failed cycle and backs off, so A no longer # restarts at once; a due B takes the slot instead and starts ITS drain. That is # a drain start (a step to the drain rate by design), not a charge end. rows, c = run(2, 6000.0, 9999.0, step=1760.0, dur=2080.0, store=timed_out(1780.0)) print(f" timeout, B due: A {[cmd(r)['a'] for r in rows if 1776 <= r[0] <= 1790]}" f" B {[cmd(r)['b'] for r in rows if 1776 <= r[0] <= 1790]}") check("timeout with B due: the charge ends at 1780 s and A is not restarted", phase_at(rows, 1778.0) == "charge" and phase_at(rows, 1780.0) == IDLE and c.maints[0].phase == IDLE and c.maints[0].backoff_until(rig.BASE + timedelta(seconds=1800)) is not None) # A command below 0 W is a charge, so B's commands must stay at or above 0; # the drain rate is the one this rig's controller was configured with. drain_w = c.maints[1].cfg.discharge_w print(f" B drain rate configured on the rig: {drain_w:.0f} W") check("⭐ timeout with B due: no battery charges from the grid after it", charging_hi(rows, 1780, 1798, hi1) == 0 and min(cmd(r)["b"] for r in rows if r[0] >= 1780) >= 0.0) check("...B takes the slot and begins its own drain, which is the only large " "step, to the configured drain rate and no higher", c.maints[1].phase == DRAIN and max(cmd(r)["b"] for r in rows if r[0] >= 1780) <= drain_w and moves(rows, "a", 1780, 1798) <= SLEW) # ============================================================================ # print("\n4. the charge ends while the guard has suspended it") # House 6000 W from 1500 s: the guard suspends the shaping, the loop has both # batteries give. A is full at 1530 s; its HOLD pin takes it to 0 W. R4 = run(2, 6000.0, 1530.0, step=1500.0) R4_old = without(lambda: run(2, 6000.0, 1530.0, step=1500.0)) _a4 = at(R4[0], 1528.0, "a") print(f" A gave {_a4:.0f} W before its HOLD; B's largest move " f"{moves(R4[0], 'b', 1526, 1798):.0f} W, without the fix " f"{moves(R4_old[0], 'b', 1526, 1798):.0f} W") check("A gives under the loop before its charge ends", present(_a4) and _a4 > SLEW and R4[1].maints[0].phase == HOLD) check("⭐ B takes A's share at one slew step a cycle (A's own give stops " "with its HOLD pin, as before)", moves(R4[0], "b", 1526, 1798) <= SLEW and charging_hi(R4[0], 1530, 1798, lambda t: 6000.0) == 0) check("non-vacuous: without the fix B jumps more than a slew step", moves(R4_old[0], "b", 1526, 1798) > SLEW) # ============================================================================ # print("\n5. N=1 is untouched; a missed poll; the rule at its boundaries") def digest(rows): return hashlib.sha256(repr(rows).encode()).hexdigest() _n1 = [digest(run(1, h, e, step=s)[0]) for h, e, s in ((6000.0, 1780.0, 1760.0), (1200.0, 1300.0, None))] _n1_old = [digest(without(lambda: run(1, h, e, step=s))[0]) for h, e, s in ((6000.0, 1780.0, 1760.0), (1200.0, 1300.0, None))] check("⭐ N=1, the charge ending at a high and a low house: every row " "byte-identical with and without the reseed", _n1 == _n1_old) if no_fix is None: print(f"\n{len(fails)} FAILED (no _reseed_charge_end): " + ", ".join(fails)) sys.exit(1) def missed_poll(at_t): """Battery B's power reading is missing on the cycle at `at_t` s.""" def before(c, t): if not getattr(c.hass, "_s93", 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._s93 = number, True return before R6 = run(2, 6000.0, 1780.0, step=1760.0, before=missed_poll(1780.0)) R6_old = without(lambda: run(2, 6000.0, 1780.0, step=1760.0, before=missed_poll(1780.0))) print(f" a missed poll on the end cycle: B " f"{[cmd(r)['b'] for r in R6[0] if 1778 <= r[0] <= 1786]}, without " f"the fix {[cmd(r)['b'] for r in R6_old[0] if 1778 <= r[0] <= 1786]}") check("a missed poll on the cycle the charge ends: the loop holds that cycle", phase_at(R6[0], 1780.0) == HOLD and "inputs missing" in str(first(r[6] for r in R6[0] if r[0] == 1780.0))) check("⭐ ...and the end still reaches the loop on the next cycle: no grid " "charge (A's charge stands through the held cycle), one slew step a " "cycle", charging_hi(R6[0], 1782, 1798, hi1) == 0 and moves(R6[0], "b", 1776, 1798) <= SLEW) check("non-vacuous: without the fix B charges from the grid", charging_hi(R6_old[0], 1782, 1798, hi1) >= 1) _c = R1[1] def reseeded(target, shares, res, ended=frozenset({A})): """`_reseed_charge_end` on the rig's controller, A's charge ended and its result `res`: (target, i_w).""" _c.target, _c.i_w, _c._charge_ended = target, 123.0, ended for inv in _c.inverters: inv.target = shares[inv.name[-1]] _c._reseed_charge_end([res, _free]) return _c.target, _c.i_w _pin = MaintResult(HOLD, True, 0.0, False, []) _free = MaintResult(IDLE, False, None, False, []) check("a HOLD pin: the start is B's command plus the pin, integrator with it", reseeded(-2000.0, {"a": -2000.0, "b": 0.0}, _pin) == (0.0, 0.0) and reseeded(2770.0, {"a": 1385.0, "b": 1385.0}, _pin) == (1385.0, 1385.0) and _c._charge_ended == frozenset()) check("a free battery: its charge leaves the start, its discharge stays", reseeded(-2000.0, {"a": -2000.0, "b": 0.0}, _free) == (0.0, 0.0) and reseeded(2770.0, {"a": 1385.0, "b": 1385.0}, _free) == (2770.0, 2770.0)) check("no charge ended: nothing moves (the integrator either)", reseeded(-2000.0, {"a": -2000.0, "b": 0.0}, _pin, frozenset()) == (-2000.0, 123.0)) _n1c = run(1, 6000.0, 1780.0, step=1760.0)[1] _n1c.target, _n1c.i_w = -2000.0, 5.0 _n1c.inverters[0].target, _n1c._charge_ended = -2000.0, frozenset({A}) _n1c._reseed_charge_end([_pin]) check("N=1: nothing moves", (_n1c.target, _n1c.i_w) == (-2000.0, 5.0)) _seen = [] def _spy(self, maint_results): _seen.append(sorted(self._charge_ended)) return no_fix(self, maint_results) Controller._reseed_charge_end = _spy try: run(2, 6000.0, 1780.0, step=1760.0, dur=2400.0, extra={"maintenance_hold_min": 5}) finally: Controller._reseed_charge_end = no_fix check("it fires once, at the charge end: not at the HOLD end or on any " "other cycle", [x for x in _seen if x] == [[A]] and len(_seen) > 300) # ============================================================================ # print("\n6. A still in CHARGE but no longer placed: stale link, refusal") def link_drop(at): """A's writes rejected from `at` s, so it is dropped as a stale link.""" def before(c, t): a = c.inverters[0] if t >= at and a.rejected_since is None: a.rejected_since = rig.BASE + timedelta(seconds=at) return before def refuses(at): """A stops charging from `at` s (reads 0 W); SAFETY-73 drops it.""" def before(c, t): if not getattr(c.hass, "_s93r", False): real = c.hass.power def power(u): p = real(u) return 0.0 if u == "a" and c.hass.t >= at and p < 0 else p c.hass.power, c.hass._s93r = power, True return before for k, hook in (("stale link", link_drop), ("refusal", refuses)): def fn(hook=hook): return run(2, 1200.0, 9999.0, dur=1500.0, before=hook(1300.0)) rows, _c6 = fn() old = without(fn)[0] drop = first((r[0] for r in rows if r[0] >= 1300 and "dropped inverter-a" in str(r[7])), "A dropped") print(f" {k}: A dropped at {drop:.0f} s; B " f"{[cmd(r)['b'] for r in rows if drop - 2 <= r[0] <= drop + 4]}, " f"without the fix {[cmd(r)['b'] for r in old if drop - 2 <= r[0] <= drop + 4]}") check(f"⭐ {k}: A is dropped while still in CHARGE, and B never charges " f"from the grid and moves at most one slew step a cycle", phase_at(rows, drop) == "charge" and min(cmd(r)["b"] for r in rows if r[0] >= 1300) >= 0.0 and moves(rows, "b", 1296, 1500) <= SLEW) check(f"non-vacuous ({k}): without the fix B jumps to -1600 W", min(cmd(r)["b"] for r in old if r[0] >= 1300) <= -1600.0) _seen.clear() Controller._reseed_charge_end = _spy try: run(2, 6000.0, 1530.0, step=1500.0) finally: Controller._reseed_charge_end = no_fix check("the guard suspending the charge is no end: it fires once, at the HOLD", [x for x in _seen if x] == [[A]]) print() if fails: print(f"{len(fails)} FAILED: " + ", ".join(fails)) sys.exit(1) print("all SAFETY-93 checks passed")