"""Runnable check for MAINT-24. `python3 test_maint24.py` A started maintenance charge at N=2 must get the cheap night. On the live site (2026-11-08..11) battery B's grid charge (battery modes) took the night, and A's own charge was switched off or sized at 200-450 W: its pause and its floor compared the peak forecast with `cap - margin`, not with this month's peak that the guard and B's trade aim at (SAFETY-80), and its floor counted its own charge in the forecast (SAFETY-111). All on the real Controller through `cycle()` (the lagged N=2 replay, tools/ opt20_scen.py): cheap window, house 3000 W, cap 3500 / margin 50, a month peak of 6000 W banked (aim 5950 W), a peak forecast that follows the grid. 1. A in maintenance CHARGE, B a Cheap hours charger: A charges at the room under the month aim (half of it: the forecast holds A's own charge, SAFETY-111 stays on its card), B files no grid trade, B's why is "cycle", every quarter is at or under the month aim. 2. Forecast above 3450 W but under the month aim: shaping is not paused and A gives nothing. A real peak (house over the month aim) still pauses it. 3. The mirror: B in maintenance CHARGE, A the charger. 4. modetrade.decide(): `maint_charging` files no charge, why "cycle"; solar roles stay. 5. Unchanged: N=1 with no forecast entity, the SAFETY-78 drain, battery modes off (digests printed on release/1.0 02eac3b). Mutations: DECISIONS/MAINT-24-charge-first.md. Read MUTATION.md before you report a survivor. """ import logging import os import sys from types import SimpleNamespace 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 opt11p3_replay as rp # noqa: E402 import opt20_scen as sc # noqa: E402 from app import modetrade, packmode, tariff # noqa: E402 from app.maintenance import DRAIN, MaintResult # noqa: E402 fails = [] def check(name, cond): print(f" {'ok ' if cond else 'FAIL'} {name}") if not cond: fails.append(name) A, B = sc.A, sc.B MONTH_PEAK = 6000.0 AIM = MONTH_PEAK - 50.0 # the guard's aim this month DRAIN_R = MaintResult(DRAIN, True, 2000.0, False, []) def scen(*, a=None, b=None, soc=(49.0, 30.0), house_w=3000.0, peak=MONTH_PEAK, fc=True, n=2, variant="p3_ch_sell", choice=None, hours=0.75): """`a` / `b`: that pack's forced maintenance result (or None). `peak`: the banked month peak or None. `fc`: a peak forecast entity that follows the grid. Returns (rows, stats, controller).""" rows = [] extra = {"cycle_cost_threshold_eur": 0.10} def patch(c): for i, res in enumerate((a, b)): if res is not None and i < len(c.maints): c.maints[i].tick = ( lambda now, soc_, blocked_by=None, gate_open=True, res=res: res) if peak is not None: orig_peak = c.peak.peak def peak_(now): r = orig_peak(now) return tariff.PeakReading(tariff.OBSERVED, getattr(r, "month", ""), peak) c.peak.peak = peak_ orig = c.cycle susp = [] c._m24_susp = susp async def cyc(): c.cheap = True if fc and c.grid is not None: c.peak_fc = float(c.grid) await orig() susp.append(bool(c._maint_shape_suspended)) c.cycle = cyc def cb(c, h, now, pr): sh = dict(c.allocation.shares) if c.allocation is not None \ and c.allocation.shares else {} rows.append(dict( now=now, grid=h.grid(), shares=sh, action=getattr(c.arb, "action", None), reason=getattr(c.arb, "reason", None) or "", roles=dict(c.mode_tick.roles) if c.mode_tick else {}, why=c.mode_tick.why if c.mode_tick else None, p={u: p.p for u, p in h.plants.items()})) stats, c = rp.run(variant, [(sc.UTC0.timestamp() - 10, house_w)], sc.WPTS, sc.at(2.0), 3600 * hours, check=cb, lag=sc.LAG, n=n, soc0={"a": soc[0], "b": soc[1]}, choice=choice or sc.CH_CH, refuse=(), patch=patch, extra=extra) return rows, stats, c def mean(rows, u): r = rows[len(rows) // 6:] return sum(x["p"][u] for x in r) / len(r) def worst_quarter(stats): return max(a[0] / a[1] for a in stats["quarters"].values()) CH = sc.MAINT_CHARGE # ============================================================================ # print("1. A in maintenance CHARGE, B a charger in the cheap window") rows, st, c = scen(a=CH) ma, mb = mean(rows, "a"), mean(rows, "b") print(f" A {ma:+.0f} W, B {mb:+.0f} W, worst quarter {worst_quarter(st):.0f}" f" W, why {rows[-1]['why']}, roles {rows[-1]['roles']}") check("A charges at (5950 - 3000) / 2 = 1475 W, not 0 or a few hundred W", -1500.0 <= ma <= -1440.0) check("B files no grid trade: no trade action in the run, B near 0 W", all("CHARGE" not in str(r["action"]) for r in rows) and mb > -50.0) check("roles: B does not charge, and the why is 'cycle'", all(r["roles"].get(B) != "charge" for r in rows) and rows[-1]["why"] == "cycle") check("every quarter at or under the month aim", worst_quarter(st) <= AIM + 1.0) check("A never gives back while there is no real peak", max(r["p"]["a"] for r in rows) <= 0.0) check("the shaping was never paused", not any(c._m24_susp)) # ============================================================================ # print("\n2. forecast between 3450 W and the month aim; a real peak") rows2, st2, c2 = scen(a=CH, house_w=3800.0) ma2 = mean(rows2, "a") print(f" house 3800: A {ma2:+.0f} W, worst quarter " f"{worst_quarter(st2):.0f} W") check("forecast 4900 W (over 3450, under 5950): no pause, A charges " "(5950 - 3800) / 2 = 1075 W and gives nothing", not any(c2._m24_susp) and -1100.0 <= ma2 <= -1040.0 and max(r["p"]["a"] for r in rows2) <= 0.0 and worst_quarter(st2) <= AIM + 1.0) rows3, st3, c3 = scen(a=CH, house_w=6500.0) print(f" house 6500: paused {sum(c3._m24_susp)}/{len(c3._m24_susp)} " f"cycles, A max {max(r['p']['a'] for r in rows3):+.0f} W") check("a real peak (house over the month aim) still pauses the shaping " "(SAFETY-07)", sum(c3._m24_susp) > len(c3._m24_susp) // 2) rows4, st4, c4 = scen(a=CH, house_w=3000.0, peak=None) print(f" no banked peak: A {mean(rows4, 'a'):+.0f} W, worst quarter " f"{worst_quarter(st4):.0f} W") check("no banked peak: the aim is cap - margin (3450 W) as before, so A " "charges (3450 - 3000) / 2 = 225 W, B does not trade, the quarter " "stays under 3450 W", -240.0 <= mean(rows4, "a") <= -210.0 and not any(c4._m24_susp) and all("CHARGE" not in str(r["action"]) for r in rows4) and worst_quarter(st4) <= 3450.0 + 1.0) # ============================================================================ # print("\n3. the mirror: B in maintenance CHARGE, A the charger") rowsm, stm, cm = scen(b=CH, soc=(30.0, 49.0)) mam, mbm = mean(rowsm, "a"), mean(rowsm, "b") print(f" A {mam:+.0f} W, B {mbm:+.0f} W, worst quarter " f"{worst_quarter(stm):.0f} W, why {rowsm[-1]['why']}") check("B charges at 1475 W, A files no grid trade, why 'cycle'", -1500.0 <= mbm <= -1440.0 and mam > -50.0 and all("CHARGE" not in str(r["action"]) for r in rowsm) and rowsm[-1]["why"] == "cycle" and all(r["roles"].get(A) != "charge" for r in rowsm)) check("...every quarter at or under the month aim", worst_quarter(stm) <= AIM + 1.0) # ============================================================================ # print("\n4. modetrade.decide(maint_charging=)") _runs = SimpleNamespace(on=True, cheap=SimpleNamespace(holds=lambda n: True), charge_useful=lambda: True, give_at=lambda n: None, sell=None) def _dec(packs, **kw): args = dict(charge_ok=True, sell_ok=True, trade_why=None, price_now=0.1, headroom_w=3000.0, floor_pct=20.0, ceiling_pct=95.0, house_w=800.0) args.update(kw) return modetrade.decide(_runs, sc.UTC0, packs, **args) _pa = modetrade.Pack(A, True, packmode.CHEAP_HOURS, False, 2500.0, 50.0, 10.0, idle=False) _pb = modetrade.Pack(B, True, packmode.CHEAP_HOURS, False, 2500.0, 30.0, 10.0) _off = _dec([_pa, _pb]) _on = _dec([_pa, _pb], maint_charging=True) _dr = _dec([_pa, _pb], draining=True) check("without it B is the charger (as before)", _off.action == modetrade.CHARGE and _off.roles[B] == modetrade.CHARGE) check("with it: no charge, why 'cycle', B keeps a hold role, no ask", _on.action is None and _on.why == "cycle" and _on.ask_w == 0.0 and _on.roles[B] == modetrade.HOLD) check("same answer as the SAFETY-78 drain", _on == _dr) _pr = modetrade.Pack(B, True, packmode.CHEAP_HOURS, False, 2500.0, 30.0, 10.0, refused=-1, probing=True) _tp = _dec([_pa, _pr], maint_charging=True) check("a refusal probe beside the maintenance charge goes on: PROBE_W alone " "(OPT-20); beside a drain it does not", _tp.action == modetrade.CHARGE and _tp.ask_w == -modetrade.PROBE_W and _dec([_pa, _pr], draining=True).action is None) check("a probe does not carry a normal charger along", _dec([_pa, _pr, modetrade.Pack("inverter-c", True, packmode.CHEAP_HOURS, False, 2500.0, 30.0, 10.0)], maint_charging=True).ask_w == -modetrade.PROBE_W) check("it files nothing outside a cheap run (why None)", _dec([_pa, _pb], maint_charging=True, ).why == "cycle" and modetrade.decide( SimpleNamespace(on=False), sc.UTC0, [_pa, _pb], charge_ok=True, sell_ok=True, trade_why=None, price_now=0.1, headroom_w=3000.0, floor_pct=20.0, ceiling_pct=95.0, house_w=800.0, maint_charging=True).why is None) # ============================================================================ # print("\n5. unchanged: N=1 (no forecast entity), the drain, modes off") d1 = sc.digest(scen(a=CH, n=1, fc=False, peak=None, soc=(49.0, 30.0))[0]) dd = sc.digest(scen(a=DRAIN_R, soc=(60.0, 30.0), peak=None, fc=False)[0]) dm = sc.digest(scen(a=CH, variant="modes_off", peak=None, fc=False)[0]) AT_BASE = { "n1": "44816d132984ecd64d5a2510223debe102f29decd1d5127ae2b255b45dbf8bd4", "drain": "8fd86d0816626ee5ae86bc973cb0121b1b5108e56c3b0fdd4b4be6c1f403f748", "off": "a007f962793611bce15b11f576a76aa52063adcfde82b2a2dc0b0c075b60e60b", } got = {"n1": d1, "drain": dd, "off": dm} for k in AT_BASE: print(f" {k} {'same' if got[k] == AT_BASE[k] else 'DIFFERS'} " f"{got[k][:16]}") check("N=1 with A in maintenance charge and no forecast entity: shares, " "grid, trade, roles and plant powers equal release/1.0 02eac3b", got["n1"] == AT_BASE["n1"]) check("the SAFETY-78 drain beside a charger: equal to 02eac3b", got["drain"] == AT_BASE["drain"]) check("battery modes off: equal to 02eac3b", got["off"] == AT_BASE["off"]) print() if fails: print(f"{len(fails)} FAILED:") for f in fails: print(f" - {f}") sys.exit(1) print("all checks passed")