"""Runnable check for SAFETY-126. `python3 test_safety126.py` During the night grid charge, with both batteries under the peak reserve, the add-on logged ERROR PEAK_GUARD_FLOOR_LIMITED and raised "Battery stopped limiting the peak" at quarter ends, and the Batteries "Why" blinked "The peak limit needs power from the batteries.", while every quarter landed on the aim. The cause: the charge lands the quarter on the aim, a 1 W landing error in the quarter's last second becomes tens of kW through `excess_w()`'s 900 / remaining, and the alarm fired on any `wanted > 0` when no battery can give. DECISIONS/SAFETY-126-landing-error-is-not-an-overrun.md. Drives the REAL Controller (read_inputs + cycle, the tariff task on a day/night profile, a P1 meter on a simulated clock, OPT-15's lag), as test_safety125.py does, with the cycle a fraction of a second off the quarter's edge as on the rig. 1. The night grid charge under the reserve: no ERROR, no notification, no red peak row, no "Why" peak, every quarter on the aim; the guard did want power at the quarter's end (not vacuous); with the fix reverted the same run alarms. The commands are the same with and without the fix. 2. A real overrun under the reserve still alarms, a step load at the same cycle as before, and names both batteries. After it, the latch is reset: the night's later landings are quiet again. 3. `misses_aim()` on its own. 4. The page and the "Why" on a decision within the aim. Mutations (MUTATION.md, purged copies) are listed in the pull request. """ import asyncio import dataclasses import io import logging import os import sys import tempfile import types from datetime import datetime, timedelta, timezone HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) sys.path.insert(0, os.path.join(HERE, "tools")) import opt11p3_replay as rp # noqa: E402 from named_absence import first, nth, present # noqa: E402 import app.main as m # noqa: E402 import app.p1 as p1mod # noqa: E402 from app import homeview, messages, peakguard, tariff # noqa: E402 from app import price as _price, static_tariff as _st # noqa: E402 from app.main import Controller # noqa: E402 from app.p1 import make_sample, split_signed # noqa: E402 from app.store import unit_key # noqa: E402 from app.tariff import QUARTER_S, Projection # noqa: E402 logging.disable(logging.NOTSET) logging.getLogger().setLevel(logging.CRITICAL) _ERRORS = io.StringIO() _h = logging.StreamHandler(_ERRORS) _h.setLevel(logging.ERROR) logging.getLogger("goodwe").addHandler(_h) logging.getLogger("goodwe").setLevel(logging.ERROR) logging.getLogger("goodwe").propagate = False fails = [] def check(name, cond): print(f" {'ok ' if cond else 'FAIL'} {name}") if not cond: fails.append(name) PEAK = 5413.17 # October's banked peak on the live site AIM_CAP = 3500.0 - 50.0 AIM_PEAK = PEAK - 50.0 TOL = 10.0 # W over the aim a quarter may land (SAFETY-81) L = timezone(timedelta(hours=1)) NIGHT = datetime(2026, 10, 30, 22, 0, tzinfo=L).astimezone(timezone.utc) PROFILE = ('{"periods": [' '{"name": "night", "days": "all", "from": "22:00", "to": "07:00",' ' "eur_per_kwh": 0.2333},' '{"name": "day", "days": "all", "from": "07:00", "to": "22:00",' ' "eur_per_kwh": 0.2801}]}') OWNER = {"a": "leftover_solar", "b": "cheap_hours"} def opts(*, modes=True, arb=False): o = dict(rp.options(modes, arb, 2)) o.update(cycle_cost_threshold_eur=0.03, static_profile=PROFILE, peak_cap_w=3500, peak_margin_w=50) o["inverters"][0].update(max_w=5000, capacity_kwh=15.36) o["inverters"][1].update(max_w=4600, capacity_kwh=20.48) return o class Harder(rp.Plant): """Charges harder than asked, as inverter-a did live (test_safety125.py: asked -4271 W, measured about -4750 W; asked -5000 W, about -5210 W), up to `cap` W. So a charge sized to the aim lands a few W over it, as on the rig (worst +7 W).""" cap = -5216.0 def step(self, dt, floor): c = self.cmd if c < 0: self.cmd = max(c * 1.11, self.cap) try: super().step(dt, floor) finally: self.cmd = c async def _no_fetch(*a, **k): raise AssertionError("the tariff task reached for a price feed") def without_fix(fn): """release/1.0 cc9bbeb: no decision is ever within the aim.""" real = getattr(Controller, "_peak_landing", None) if real is None: return fn() Controller._peak_landing = lambda self, decision: decision try: return fn() finally: Controller._peak_landing = real def drive(o, *, modes=OWNER, socs=(10.0, 10.0), house=600.0, minutes=46, start=NIGHT, off=0.7, peak=None): """The real Controller from `start` + `off` s for `minutes` at a 1 s cycle with OPT-15's lag, so a cycle lands `1 - off` s before each quarter's end. `house` is W or a function of t. Returns a dict: the quarters' true grid averages, the commands, the reserve notifications, the ERROR lines, and per cycle where the guard engaged: (t, floor limited, within the aim, battery "Why" peak, red reserve row, red saturated row, alarm latched).""" store = rp._Store() for u, mode in (modes or {}).items(): store.data[unit_key(f"inverter-{u}", "pack_use")] = True store.data[unit_key(f"inverter-{u}", "pack_mode")] = mode store.data[unit_key(f"inverter-{u}", "pack_sell")] = False load = house if callable(house) else (lambda t: float(house)) t0 = start + timedelta(seconds=off) out = {"cmds": [], "notes": [], "seen": [], "sums": {}} _ERRORS.seek(0) _ERRORS.truncate() async def go(): lag = rp.Lag() hass = rp.Hass(("a", "b"), load, {"a": socs[0], "b": socs[1]}, limits={"a": (-5000.0, 5000.0), "b": (-4600.0, 4600.0)}, lag=lag) for u, cap in (("a", -5216.0), ("b", -4800.0)): hass.plants[u].__class__ = Harder hass.plants[u].cap = cap real_set = hass.set_number async def set_number(e, v): out["cmds"].append((round(hass.t, 3), e, float(v))) return await real_set(e, v) async def call(*a): if a[:2] == ("persistent_notification", "create") \ and a[2]["notification_id"] == messages.ID_PEAK_RESERVE: out["notes"].append((hass.t, a[2]["message"])) return True hass.set_number, hass.call = set_number, call tmp = tempfile.mkdtemp(prefix="s126-") db = os.path.join(tmp, "tariff.db") if peak is not None: tariff.PeakStore(db)._db.execute( "INSERT INTO monthly_peak VALUES (?, ?, ?, NULL)", (tariff.month_key(t0), peak, "2026-10-21T20:15:00+00:00")).connection.commit() c = Controller(dict(o), hass, store, rp._Pub(), stop=asyncio.Event(), peak_db_path=db, price_history_path=os.path.join(tmp, "ph.json")) c.inverters[0].dev_min, c.inverters[0].dev_max = -5000.0, 5000.0 c.inverters[1].dev_min, c.inverters[1].dev_max = -4600.0, 4600.0 c.auto = True real = m.datetime, homeview.datetime, p1mod.time clock, simt = [t0], [0.0] class _DT(real[0]): @classmethod def now(cls, tz=None): return clock[0] if tz is None else clock[0].astimezone(tz) m.datetime = homeview.datetime = _DT p1mod.time = types.SimpleNamespace(monotonic=lambda: 1e6 + simt[0]) real_load = _price.load_prices, _st.load_prices _price.load_prices = _st.load_prices = _no_fetch task = (asyncio.create_task(c.run_tariff_refresh()) if o.get("battery_modes_enabled") or o.get("arbitrage_enabled") else None) try: hass.advance(0.0, 10.0) t = 0.0 while t < minutes * 60: now = t0 + timedelta(seconds=t) hass.t, simt[0], clock[0] = t, t, now if t == 0.0 and task is not None: for _ in range(500): if c.tariff_plan is not None: break await asyncio.sleep(0) q = int(now.timestamp() // QUARTER_S) acc = out["sums"].setdefault(q, [0.0, 0]) if t >= lag.p1_s: imp, exp = split_signed(hass._past(lag.meter_s + lag.p1_s)[0]) c.p1.submit(make_sample( "ha_signed", imp, exp, phases=1, ingest_ts=now - timedelta(seconds=lag.p1_s), ingest_mono=1e6 + t)) await c.read_inputs(now_mono=t) await c.cycle() hass.advance(lag.tick, 10.0) acc[0] += max(0.0, hass.grid()) acc[1] += 1 g = c.peak_guard if g is not None and g.engaged and g.blind is None: rows = {r["id"]: r for r in c.checks(now)} row = rows.get("peak_guard", {}) red = row.get("ok") is False and not row.get("warn") out["seen"].append(( t, g.floor_limited, getattr(g, "within_aim", False), homeview.peak_engaged(c), red and "banner" in row, red and "banner" not in row, c._peak_floor_limited)) t += lag.tick finally: if task is not None: task.cancel() try: await task except asyncio.CancelledError: pass m.datetime, homeview.datetime, p1mod.time = real _price.load_prices, _st.load_prices = real_load return c out["c"] = asyncio.run(go()) out["quarters"] = { datetime.fromtimestamp(q * QUARTER_S, L).strftime("%H:%M"): s / k for q, (s, k) in sorted(out["sums"].items()) if k >= QUARTER_S - 1} out["errors"] = [x for x in _ERRORS.getvalue().splitlines() if "PEAK_GUARD_FLOOR_LIMITED" in x] return out def show(tag, r): print(f" {tag}: quarters " f"{ {k: round(v, 1) for k, v in r['quarters'].items()} }, " f"guard cycles {len(r['seen'])}, floor-limited " f"{sum(1 for s in r['seen'] if s[1])}, within-aim " f"{sum(1 for s in r['seen'] if s[2])}, Why peak " f"{sum(1 for s in r['seen'] if s[3])}, notes {len(r['notes'])}, " f"errors {len(r['errors'])}") def quiet(r): """No reserve alarm in any form: ERROR, notification, red row, latch, "Why" peak, and no red saturated row.""" return (not r["errors"] and not r["notes"] and not any(s[3] or s[4] or s[5] or s[6] for s in r["seen"])) def on_aim(r, aim): q = r["quarters"] return len(q) >= 3 and all(v <= aim + TOL for v in q.values()) # ============================================================================ # print("1. the night grid charge under the peak reserve") NIGHTS = {} # The owner's shape at the cap's aim. At the banked peak's aim it is # test_safety125.py's live shape (A Cheap hours, B Solar first), because # B's 4 600 W alone stays under that aim. LIVE = {"a": "cheap_hours", "b": "solar_first"} for tag, kw, aim in ( ("aim 3450, cycle 0.3 s before the edge", dict(off=0.7), AIM_CAP), ("aim 3450, cycle 0.01 s before the edge", dict(off=0.99), AIM_CAP), ("month peak 5413, cycle 0.01 s before the edge", dict(off=0.99, peak=PEAK, modes=LIVE, socs=(11.0, 10.0)), AIM_PEAK)): head = drive(opts(), **kw) base = without_fix(lambda: drive(opts(), **kw)) NIGHTS[tag] = (head, base) show("head " + tag, head) show("fix reverted", base) check(f"⭐ {tag}: no ERROR, no notification, no red peak row and no " f"'Why' peak", quiet(head)) check(f"{tag}: every quarter at or under the aim ({aim:.0f} W + " f"{TOL:.0f} W)", on_aim(head, aim)) check(f"{tag}: not vacuous, the guard wanted power under the reserve " f"at a quarter's end (floor-limited, within the aim)", any(s[1] and s[2] for s in head["seen"])) check(f"{tag}: with the fix reverted the same run alarms", not quiet(base)) check(f"{tag}: the commands are the same with and without the fix", head["cmds"] == base["cmds"] and len(head["cmds"]) > 100) _h, _b = NIGHTS["aim 3450, cycle 0.01 s before the edge"] check("fix reverted, at 0.01 s: the ERROR, the notification and the " "'Why' peak all come back (the rig's three symptoms)", _b["errors"] and _b["notes"] and any(s[3] for s in _b["seen"])) # ============================================================================ # print("\n2. a real overrun under the reserve still alarms") def evening(t): """600 W, then an oven and a car charger from 20 minutes on.""" return 600.0 if t < 1200 else 6500.0 for tag, onset, kw in ( ("night grid charge, then 6500 W", 1200.0, dict(house=evening)), ("modes off, 5500 W all along", 0.0, dict(house=5500.0, modes=None, socs=(15.0, 12.0)))): o = opts(modes=kw.get("modes", OWNER) is not None) head = drive(o, minutes=35, **kw) base = without_fix(lambda: drive(o, minutes=35, **kw)) show("head " + tag, head) show("fix reverted", base) # From the load's onset: before it, the fix reverted shows the night's # false alarm (part 1). first_h = next((s[0] for s in head["seen"] if s[6] and s[0] >= onset), None) first_b = next((s[0] for s in base["seen"] if s[6] and s[0] >= onset), None) check(f"⭐ {tag}: the ERROR and the notification, naming both " f"batteries", head["errors"] and head["notes"] and all("inverter-a and inverter-b" in n for _t, n in head["notes"])) _th = [t for t, _n in head["notes"]] _tb = [t for t, _n in base["notes"] if t >= onset] print(f" notifications at {_th} s; with the fix reverted {_tb} s") check(f"{tag}: the first at the same cycle as before the fix " f"({first_h} s, {first_b} s), and none that was not raised before", first_h is not None and first_h == first_b and present(nth(_th, 0)) and nth(_th, 0) == nth(_tb, 0) and set(_th) <= set(_tb)) check(f"{tag}: the red peak row and its banner while it stands", any(s[4] for s in head["seen"])) def burst(t): """600 W, a 6 500 W load from 20 to 25 minutes, then 600 W again while the night charge goes on.""" return 6500.0 if 1200 <= t < 1500 else 600.0 # F3 (review): `_peak_landing()` resets its latch when the guard lets go, so # a real overrun does not keep the later landing errors alarming. _bh = drive(opts(), house=burst, off=0.99, minutes=61) _bb = without_fix(lambda: drive(opts(), house=burst, off=0.99, minutes=61)) show("head real overrun, then the night again", _bh) show("fix reverted", _bb) # The burst banked a new month peak, which the later quarters stay under. _new_peak = _bh["c"].peak_guard.ceiling_w def _late(r): """From the next quarter on: the overrun's own quarter ends over.""" return [x for x in r["seen"] if x[0] >= 1800] print(f" notifications at {[t for t, _n in _bh['notes']]} s; with the fix " f"reverted {[t for t, _n in _bb['notes']]} s; limit after the burst " f"{_new_peak:.0f} W") check("⭐ after a real overrun ends, the latch is reset: the alarm came " "during the overrun, and the later quarters' landings are quiet " "(no notification, no 'Why' peak, no red row) and under the new peak", any(1200 <= t < 1500 for t, _n in _bh["notes"]) and not [t for t, _n in _bh["notes"] if t >= 1800] and not any(x[3] or x[4] or x[5] for x in _late(_bh)) and present(nth([v for k, v in _bh["quarters"].items() if k >= "22:30"], 0)) and all(v <= _new_peak for k, v in _bh["quarters"].items() if k >= "22:30")) check("...not vacuous: the guard wanted power at those landings, and with " "the fix reverted they show the 'Why' peak or an alarm", any(x[2] for x in _late(_bh)) and (any(x[3] or x[4] or x[5] for x in _late(_bb)) or bool([t for t, _n in _bb["notes"] if t >= 1800]))) # ============================================================================ # print("\n3. misses_aim() on its own") START = datetime(2026, 10, 30, 21, 15, tzinfo=timezone.utc) def proj(projected, elapsed, held): return Projection(projected, projected, elapsed, 1.0, True, START, held) check("the rig's case: 3452 W against 3450 W, 0.3 s left, charging " "1250 W from the grid: not a miss", not peakguard.misses_aim(proj(3452.0, 899.7, 3452.0), batt_w=-1250.0, target_w=AIM_CAP)) check("...although decide() wants power there (its gain, 900 / 0.3 s)", (peakguard.decide(proj(3452.0, 899.7, 3452.0), batt_w=-1250.0, cap_w=3500, margin_w=50, reading=None, achievable_w=0.0, soc=10.0, soc_floor=20.0) .wanted_w or 0.0) > 1000.0) check("a quarter heading 11 W over the aim with the batteries at 0 W is a " "miss", peakguard.misses_aim(proj(3461.0, 899.7, 3461.0), batt_w=0.0, target_w=AIM_CAP)) check("...and 10 W is not (ALARM_MISS_W, as for 'Load too high')", not peakguard.misses_aim(proj(3460.0, 899.7, 3460.0), batt_w=0.0, target_w=AIM_CAP)) check("the add-on's own grid charge counts as stoppable: 4000 W on the " "meter mid-quarter, 3000 W of it charging, is not a miss", not peakguard.misses_aim(proj(2500.0, 450.0, 4000.0), batt_w=-3000.0, target_w=AIM_CAP)) check("...and a house of 6000 W the batteries are covering is one", peakguard.misses_aim(proj(1000.0, 450.0, 0.0), batt_w=6000.0, target_w=AIM_CAP)) check("no battery reading, or no seconds left: not a miss (decide() files " "nothing there)", not peakguard.misses_aim(proj(9000.0, 450.0, 9000.0), batt_w=None, target_w=AIM_CAP) and not peakguard.misses_aim(proj(9000.0, 900.0, 9000.0), batt_w=0.0, target_w=AIM_CAP)) # ============================================================================ # print("\n4. the page and the 'Why' on a decision within the aim") _c = NIGHTS["aim 3450, cycle 0.01 s before the edge"][0]["c"] _dec = peakguard.PeakDecision( engaged=True, floor_w=5000.0, ceiling_w=3500.0, ceiling_from=peakguard.CEILING_CAP, projected_w=3451.0, excess_w=9e4, wanted_w=9e4, target_w=AIM_CAP, saturated=True, soc=30.0, soc_floor=20.0) def page(d): """(the peak row is red, the "Why" says peak) for decision `d`.""" _c.peak_guard, _c._peak_floor_limited = d, False row = first((r for r in _c.checks(datetime.now(timezone.utc)) if r["id"] == "peak_guard"), "peak row") if not present(row): return {"red": None}, None return dict(row, red=row["ok"] is False and not row.get("warn")), \ homeview.peak_engaged(_c) _r1, _w1 = page(_dec) _r2, _w2 = page(dataclasses.replace(_dec, within_aim=True)) check("a saturated decision: red 'Load too high' row and the 'Why' peak, " "as before", _r1["red"] and _w1) check("...within the aim: neither", _r2["red"] is False and _w2 is False) _lim = dataclasses.replace(_dec, floor_w=0.0, saturated=False, floor_limited=True, soc=10.0, soc_floor_reason=peakguard.FLOOR_AT_SOC_FLOOR) _r3, _w3 = page(_lim) _r4, _w4 = page(dataclasses.replace(_lim, within_aim=True)) check("floor-limited at 0 W: the red reserve row with its banner, as " "before", _r3["red"] and "banner" in _r3 and not _w3) check("...within the aim: no red row", _r4["red"] is False and _w4 is False) # ============================================================================ # print() if fails: print(f"{len(fails)} SAFETY-126 check(s) FAILED:") for f in fails: print(f" - {f}") sys.exit(1) print("all SAFETY-126 checks passed")