"""Runnable check for SAFETY-125. `python3 test_safety125.py` A meter reading that is late near a quarter's end left seconds the projection did not count (SAFETY-99), and `charge_headroom_w()` read them as room: a grid charge went back to full power in the quarter's last seconds and the billed quarter went over the aim, above a banked month peak or the cap. And the peak guard's row said "under" while the period was over the limit at the current rate. DECISIONS/SAFETY-125-grid-charge-under-the-month-peak.md. Drives the REAL Controller (read_inputs + cycle, the tariff task on a day/night profile, a P1 meter on a simulated clock) in the live shape: modes on, inverter-a Cheap hours at 5000 W charging harder than asked (as it did live), inverter-b Solar first, a house of 1 kW, cap 3500 W, margin 50 W. 1. `charge_room_w()` on its own. 2. A reading 30 s late at each quarter's end: every quarter at or under the aim, with a month peak banked above the cap and with none, with the P1 projection and with a `peak_forecast_entity`; the charge is cut late in the quarter and back in the next one. 3. The same for a price trade with modes off, and a maintenance charge at N=2. 4. A healthy meter: the quarters as before, and the row and the month-peak line say what is true. 5. The words on their own. Mutations (MUTATION.md, purged copies) are listed in the pull request. """ import asyncio 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")) logging.disable(logging.WARNING) import opt11_identity as rig # noqa: E402 import opt11p3_replay as rp # noqa: E402 import safety84_identity as s84 # noqa: E402 import app.main as m # noqa: E402 import app.p1 as p1mod # noqa: E402 from named_absence import first, present # 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 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_PEAK = PEAK - 50.0 AIM_CAP = 3500.0 - 50.0 TOL = 10.0 # W over the aim a quarter may land (SAFETY-81) FC = "sensor.piek_prognose_kwartier" L = timezone(timedelta(hours=1)) START = datetime(2026, 10, 29, 0, 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}]}') def opts(*, fc=False, 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) if fc: o["peak_forecast_entity"] = FC return o class Harder(rp.Plant): """Charges harder than asked, as inverter-a did live (asked -4271 W, measured about -4750 W; asked -5000 W, about -5210 W).""" def step(self, dt, floor): c = self.cmd if c < 0: self.cmd = max(c * 1.11, -5216.0) 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 drive(o, *, peak=None, late=0.0, minutes=31, every=0): """The real Controller from 00:00 local for `minutes` at a 1 s cycle (OPT-15's lag). `late`: no P1 telegram in each quarter's last `late` s. -> (controller, rows, {quarter start: true grid average}, samples), the samples taken every `every` cycles from the second quarter (the first starts before the first reading): (projection, ceiling, engaged, peak_guard row, month-peak line).""" store = rp._Store() for u, mode in (("a", "cheap_hours"), ("b", "solar_first")): 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 rows, samples, sums = [], [], {} async def go(): lag = rp.Lag() hass = rp.Hass(("a", "b"), lambda t: 1000.0, {"a": 11.0, "b": 10.0}, limits={"a": (-5000.0, 5000.0), "b": (-4600.0, 4600.0)}, lag=lag) hass.plants["a"].__class__ = Harder fc_kw = [0.0] async def state(e): # The HA quarter projection, in kW like the live template. return None if e != FC else { "state": f"{fc_kw[0]:.3f}", "attributes": {"unit_of_measurement": "kW"}} hass.state = state tmp = tempfile.mkdtemp(prefix="s125-") 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(START), 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 = [START], [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()) try: hass.advance(0.0, 10.0) t, n = 0.0, 0 while t < minutes * 60: now = START + timedelta(seconds=t) hass.t, simt[0], clock[0] = t, t, now if t == 0.0: for _ in range(500): if c.tariff_plan is not None: break await asyncio.sleep(0) q = int(now.timestamp() // QUARTER_S) acc = sums.setdefault(q, [0.0, 0]) g_now = hass.grid() fc_kw[0] = (acc[0] + g_now * (QUARTER_S - acc[1])) \ / QUARTER_S / 1000.0 if now.timestamp() % QUARTER_S < QUARTER_S - late \ and 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 rows.append((now, c.target, hass.plants["a"].p, getattr(c.arb, "action", None))) pg = c.peak_guard if every and n % every == 0 and t >= QUARTER_S \ and pg is not None and pg.projected_w is not None: row = first((r["text"] for r in c.checks(now) if r["id"] == "peak_guard"), "guard row") samples.append((pg.projected_w, pg.ceiling_w, pg.engaged, row, homeview.month_peak(c, now)["line"])) t += lag.tick n += 1 finally: 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 c = asyncio.run(go()) full = {datetime.fromtimestamp(q * QUARTER_S, L).strftime("%H:%M"): s / k for q, (s, k) in sorted(sums.items()) if k >= QUARTER_S} return c, rows, full, samples def into(now): return now.timestamp() % QUARTER_S # ============================================================================ # print("1. charge_room_w(): a late reading's seconds count at that reading") _dec = peakguard.decide( Projection(5800.0, 5800.0, 450.0, 1.0, True, START, 5800.0), batt_w=-4800.0, cap_w=3500, margin_w=50, reading=types.SimpleNamespace(is_observed=True, peak_w=PEAK), achievable_w=9600.0, soc=50.0, soc_floor=20.0) _room = peakguard.charge_headroom_w(_dec) _late = Projection(5800.0, 5800.0, 450.0, 430.0 / 450.0, False, START, 5800.0) print(f" headroom {_room:.0f} W; 20 s late, 450 s left: " f"{peakguard.charge_room_w(_dec, _late):.0f} W") check("with every second counted it is charge_headroom_w() exactly", peakguard.charge_room_w( _dec, Projection(5800.0, 5800.0, 450.0, 1.0, True, START, 5800.0)) == _room and _room > 0.0) check("20 s late at a 5800 W reading with 450 s left: 5800 x 20 / 450 W " "less room", abs(peakguard.charge_room_w(_dec, _late) - (_room - 5800.0 * 20.0 / 450.0)) < 1e-6) check("never below 0 W, and an export held counts nothing", peakguard.charge_room_w(_dec, Projection( 5800.0, 5800.0, 880.0, 850.0 / 880.0, False, START, 5800.0)) == 0.0 and peakguard.charge_room_w(_dec, Projection( 5800.0, 5800.0, 450.0, 430.0 / 450.0, False, START, -900.0)) == _room) check("None, blind (0 W) and no projection pass through", peakguard.charge_room_w(None, _late) is None and peakguard.charge_room_w(_dec, None) == _room and peakguard.charge_room_w( peakguard.PeakDecision(engaged=True, floor_w=0.0, ceiling_w=3500.0, ceiling_from="cap", blind="frozen"), _late) == 0.0) # ============================================================================ # def graded(label, full, aim): print(f" {label}: quarters " f"{ {k: round(v, 1) for k, v in full.items()} } W, aim {aim:.0f}") return len(full) >= 2 and all(v <= aim + TOL for v in full.values()) def cut_and_back(rows): """The charge is off in each quarter's last 5 s, and back (A charging over 2 kW) within the next quarter's first minute.""" ends = [r for r in rows if into(r[0]) >= QUARTER_S - 5] starts = {} for r in rows: if into(r[0]) < 60: q = int(r[0].timestamp() // QUARTER_S) starts[q] = starts.get(q, False) or r[2] < -2000.0 return (present(first(iter(ends), "a quarter end")) and all(r[1] > -1.0 for r in ends) and len(starts) >= 2 and all(starts.values())) print("\n2. a reading 30 s late at each quarter's end, the live shape") GAP = {} for name, kw, aim in ( ("month peak 5413 W banked, P1 projection", dict(peak=PEAK), AIM_PEAK), ("month peak banked, peak_forecast_entity (kW)", dict(peak=PEAK, fc=True), AIM_PEAK), ("no peak banked (aim = cap - margin), P1 projection", dict(), AIM_CAP), ("no peak banked, peak_forecast_entity", dict(fc=True), AIM_CAP)): fc = kw.pop("fc", False) _c, _rows, _full, _s = drive(opts(fc=fc), late=30.0, **kw) GAP[name] = (_c, _rows, _full) check(f"{name}: every quarter at or under the aim " f"({aim:.0f} W, +{TOL:.0f} W landing)", graded(name, _full, aim)) check(f"{name}: the charge is off in the quarter's last seconds and " f"back in the next quarter", cut_and_back(_rows)) _c_fc = GAP["month peak banked, peak_forecast_entity (kW)"][0] check("the forecast entity is read, in kW, as the live template (SAFETY-118), " "and the cheap run's charge does not depend on it", _c_fc.peak_fc is not None and _c_fc.peak_fc > 1000.0 and [r[1] for r in GAP["month peak banked, peak_forecast_entity (kW)"] [1]] == [r[1] for r in GAP[ "month peak 5413 W banked, P1 projection"][1]]) # ============================================================================ # print("\n3. the same for a price trade (modes off) and a maintenance charge") _ca, _ra, _fa = drive(opts(modes=False, arb=True), peak=PEAK, late=30.0)[:3] check("modes off, price trading on, a reading 30 s late: every quarter at " "or under the aim", graded("trade", _fa, AIM_PEAK) and any(r[3] == "ARBITRAGE_CHARGE" for r in _ra)) def late_feed(gap): """safety84's cheap-window feed, no telegram in each quarter's last `gap` s.""" feed = s84.feed(True) def before(c, t): if (rig.BASE.timestamp() + t) % QUARTER_S < QUARTER_S - gap \ or not getattr(c.hass, "_s84", False): feed(c, t) return before for peak, house, aim in ((5413.0, 4000.0, 5363.0), (None, 2500.0, AIM_CAP)): _mr, _mc = s84.run(2, load=lambda t, h=house: h, dur=2700.0, month_peak=peak, before=late_feed(30.0)) _qs = {} for r in _mr: q = int((rig.BASE.timestamp() + r[0]) // QUARTER_S) a = _qs.setdefault(q, [0.0, 0]) a[0] += max(0.0, house - sum(v for _u, v in r[2])) a[1] += 1 _mf = {datetime.fromtimestamp(q * QUARTER_S, L).strftime("%H:%M"): s / k for q, (s, k) in sorted(_qs.items()) if k * rig.TICK >= QUARTER_S} check(f"N=2 maintenance charge, house {house:.0f} W, " f"{'peak banked' if peak else 'no peak banked'}, a reading 30 s " f"late: every quarter at or under the aim", graded("maintenance", _mf, aim) and any(mm.phase == "charge" for mm in _mc.maints)) # ============================================================================ # print("\n4. a healthy meter: the quarters, and the words") for name, kw, aim, kw_text in (("month peak banked", dict(peak=PEAK), AIM_PEAK, "5.4 kW"), ("no peak banked", dict(), AIM_CAP, "3.5 kW")): _c, _rows, _full, _s = drive(opts(), every=10, **kw) check(f"{name}: every quarter at or under the aim", graded(name, _full, aim)) over = [s for s in _s if not s[2] and s[0] > s[1]] under = [s for s in _s if not s[2] and s[0] <= s[1]] _o, _u = first(iter(over), "an over sample"), \ first(iter(under), "an under sample") print(f" over the limit at this rate ({len(over)} samples): " f"{_o[3] if present(_o) else _o!s} / " f"{_o[4] if present(_o) else _o!s}") print(f" under it ({len(under)}): " f"{_u[3] if present(_u) else _u!s} / " f"{_u[4] if present(_u) else _u!s}") check(f"{name}: over the limit at this rate the guard row never says " f"'under'; it says the grid charge is cut", present(_o) and all( "under" not in s[3] and "Peak guard: cutting the grid charge, " "at this rate" in s[3] and " over " in s[3] for s in over)) check(f"{name}: ...and the month-peak line says the rate, and that the " f"grid charge is cut back to stay under {kw_text}", present(_o) and all( s[4].startswith("At this rate this 15-minute period reaches") and s[4].endswith("The grid charge is cut back to stay " f"under {kw_text}.") for s in over)) check(f"{name}: under the limit both read as before", present(_u) and all( s[3].startswith("Peak guard: idle, heading for") and " under " in s[3] and s[4].startswith("This 15-minute period is heading for") and s[4].endswith(f"The batteries step in above {kw_text}.") for s in under)) # ============================================================================ # print("\n5. the words on their own") _en = messages.row_peak_guard_idle(5814.0, 5413.0, month_peak=True) _cut = messages.row_peak_guard_idle(5814.0, 5413.0, month_peak=True, cut=True) _ok = messages.row_peak_guard_idle(5200.0, 5413.0, month_peak=True) print(f" {_en}\n {_cut}\n {_ok}") check("over, nothing cut: 'idle, at this rate ..., over ...'", _en == "Peak guard: idle, at this rate 5 814 W, over this month's " "5 413 W peak") check("over, a grid charge cut", _cut == "Peak guard: cutting the grid charge, at this rate 5 814 W, " "over this month's 5 413 W peak") check("under: the 0.19.1 sentence, unchanged", _ok == "Peak guard: idle, heading for 5 200 W, under this month's " "5 413 W peak") check("the Dutch forms say the same", _cut.nl == "Piekbewaking: schroeft het laden van het net terug, aan " "dit tempo 5 814 W, boven de maandpiek van 5 413 W" and _en.nl.startswith("Piekbewaking: in rust, aan dit tempo")) check("the month-peak line in shadow mode says 'would'", homeview.line_period(5814.0, 5413.0, True, cut=True).endswith( "The grid charge would be cut back to stay under 5.4 kW.") and homeview.line_period(5814.0, 5413.0, False).endswith( "The batteries step in above 5.4 kW.")) print() if fails: print(f"{len(fails)} FAILED:") for f in fails: print(f" - {f}") sys.exit(1) print("all checks passed")