"""Runnable check for OPT-20. `python3 test_opt20.py` A refusal probe (OPT-15 F3, SAFETY-73) asks the probing battery PROBE_W and nothing else. On TEST-23 (OPT-15 g7) the probe made the trade ask 2500 W: B was pinned at 200 W and the allocator put the other 2300 W on A, which sat in its maintenance charge phase at 0 W, with the grid at 3100 W for 34 s. 1. modetrade: the ask carries PROBE_W for the probing battery, the other chargers' ratings as without the probe, and `size()` adds the probe on top of their share (or asks PROBE_W exactly, or nothing). 2. main: the ask carries PROBE_W for exactly the pins filed this tick, so the first and the last cycle of a window do not move another battery's share. 3. The real Controller in the lagged replay, with B refusing and probed while A is (a) in its maintenance charge phase (also with the charge shaped to 0 W), (b) idle, (c) a normal charger in the same run, and the OPT-15 rig's g7 sequence: A's command is the same through the window as just before it, B gets PROBE_W, and the commanded charge rises by PROBE_W at most. 4. Battery modes on with no probe: byte-identical to release/1.0 bf9019e. Out of scope, and unchanged: a normal (non-probing) trade placed on a battery in maintenance charge (OPT-11 P4 F4) and the action-log wording (UX-38). Mutations: MUTATION.md says how to run them (purged copies); the 12 of OPT-20 are listed in the pull request. """ import dataclasses 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 opt20_scen as sc # noqa: E402 from app import arbitrage, modetrade, packmode # noqa: E402 from app import control as ctl # noqa: E402 from app import main # noqa: E402 from app.arbiter import Claim, P_SAFETY # noqa: E402 fails = [] def check(name, cond): print(f" {'ok ' if cond else 'FAIL'} {name}") if not cond: fails.append(name) PW = ctl.PROBE_W A, B = sc.A, sc.B # ============================================================================ # print("1. modetrade: the ask and the size") _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) _pb = modetrade.Pack(B, True, packmode.CHEAP_HOURS, False, 2500.0, 30.0, 10.0, refused=-1, probing=True) _pa = modetrade.Pack(A, True, packmode.CHEAP_HOURS, False, 2500.0, 50.0, 10.0) _pm = dataclasses.replace(_pa, mode=packmode.SOLAR_FIRST, idle=False) _pi = dataclasses.replace(_pa, mode=packmode.SOLAR_FIRST) _t_alone = _dec([_pb, _pm]) check("a probe beside a battery in maintenance asks PROBE_W and no more", _t_alone.action == modetrade.CHARGE and _t_alone.ask_w == -PW and _t_alone.probe_w == PW) check("...beside an idle battery that cannot charge from the grid too", _dec([_pb, _pi]).ask_w == -PW and _dec([_pb, _pi]).probe_w == PW) _t_two = _dec([_pb, _pa]) check("...beside a normal charger the ask is that charger's rating plus " "PROBE_W", _t_two.ask_w == -2500.0 - PW and _t_two.probe_w == PW) _t_np = _dec([dataclasses.replace(_pb, probing=False), _pa]) check("no probe: the ask is the chargers' ratings, probe_w is 0 (as before)", _t_np.ask_w == -2500.0 and _t_np.probe_w == 0.0 and _dec([_pa, _pm]).ask_w == -2500.0 and modetrade.Tick({}, {}, None, 0.0, None).probe_w == 0.0) def sz(ask, probe_w, lo, room, action=modetrade.CHARGE): t = modetrade.Tick({}, {}, action, ask, None, probe_w) return modetrade.size(t, (lo, 2500.0), room) check("size, a probe alone: PROBE_W, whatever the headroom above it", sz(-PW, PW, -(PW), 3000.0) == -PW and sz(-PW, PW, -2700.0, 9000.0) == -PW) check("...exactly PROBE_W or nothing: a headroom or a bound under it files " "no trade (a smaller ask voids the pin and the allocator would place " "it elsewhere)", sz(-PW, PW, -2700.0, PW - 1) == 0.0 and sz(-PW, PW, -(PW - 1), 3000.0) == 0.0 and sz(-PW, PW, -2700.0, 0.0) == 0.0 and sz(-PW, PW, -2700.0, PW) == -PW) check("size with another charger: its share is sized as without the probe, " "the probe's PROBE_W is added on top", sz(-2700.0, PW, -2700.0, 3000.0) == -2700.0 and sz(-2700.0, PW, -2700.0, 1000.0) == -1200.0 and sz(-2700.0, PW, -1700.0, 3000.0) == -1700.0 and sz(-2700.0, PW, -2700.0, 0.0) == 0.0) check("size with no probe is what it was (ask, bound, headroom)", sz(-2500.0, 0.0, -2700.0, 3000.0) == -2500.0 and sz(-2500.0, 0.0, -2700.0, 1000.0) == -1000.0 and sz(-2500.0, 0.0, -1800.0, 3000.0) == -1800.0 and sz(-2500.0, 0.0, -2700.0, None) == 0.0 and modetrade.size(modetrade.Tick({}, {}, None, 0.0, None), (-5.0, 5.0), 3000.0) == 0.0) # ============================================================================ # print("\n2. main: the ask carries the pins filed this tick") def trade(tick, pins, bound=(-2700.0, 2500.0), room=3000.0): """The trade `_mode_trade()` files: (claims, size or None).""" claims = [Claim.set("refused", P_SAFETY, -PW * 1.0, ctl.PROBING, actuator="x") for _ in range(pins)] me = SimpleNamespace( mode_tick=tick, loop_bound=lambda cl: bound, arb_headroom_w=room, mode_runs=SimpleNamespace( cheap=SimpleNamespace(start=sc.UTC0, end=sc.UTC0, mean_eur_kwh=0.1), sell=None, level_eur_kwh=0.1), arb=None, _arb_action=None, log_event=lambda *a, **k: None) filed = main.Controller._mode_trade(me, claims, sc.UTC0) t = [c for c in claims if c.source == arbitrage.SOURCE] return filed, (t[0].value if t else None), me.mode_tick _t0 = modetrade.Tick({}, {A: modetrade.CHARGE}, modetrade.CHARGE, -2500.0, None, 0.0) check("the pin is filed but decide() ran before it (the window's first " "cycle): the charger keeps its whole ask and the pin is added", trade(_t0, 1)[1] == -2700.0 and trade(_t0, 1)[2].probe_w == PW) check("the window closed after decide() (its last cycle): the probe's PROBE_W " "leaves the ask, so no battery is handed it", trade(_t_two, 0)[1] == -2500.0 and trade(_t_alone, 0) == (False, None, trade(_t_alone, 0)[2]) and trade(_t_alone, 0)[2].action is None) check("pin and ask agree: nothing changes", trade(_t_two, 1)[1] == -2700.0 and trade(_t_alone, 1)[1] == -PW and trade(_t0, 0)[1] == -2500.0) # ============================================================================ # print("\n3. the real Controller: B refuses and is probed") def charge(r): return -sum(min(0.0, v) for v in r["shares"].values()) def window_checks(rows, label): """Per probe window: A as just before, B at PROBE_W (or 0 when nothing is traded), the charge up by at most PROBE_W.""" win = sc.windows(rows) ok_a = ok_b = ok_sum = ok_grid = True for i, j in win: pre = rows[i - 1] lo, hi = max(0, i - 3), min(len(rows) - 1, j + 3) ext = rows[lo:hi + 1] base_sum = sum(min(0.0, v) for v in pre["shares"].values()) ok_a &= all(abs(r["shares"].get(A, 0.0) - pre["shares"].get(A, 0.0)) < 1e-6 for r in ext) bs = [r["shares"].get(B, 0.0) for r in ext] if any(r["action"] for r in rows[i:j + 1]): ok_b &= set(bs) <= {0.0, -PW} and bs.count(-PW) >= 20 else: # a drain comes first: the window trades nothing ok_b &= set(bs) == {0.0} ok_sum &= all(charge(r) <= charge(pre) + PW + 1e-6 for r in ext) ok_grid &= max(r["grid"] for r in ext) <= pre["grid"] + PW + 50.0 n_ch = sum(1 for r in rows if r["probing"] and r["action"]) print(f" {label}: {len(win)} window(s), {n_ch} probing cycles with a " f"trade, highest grid " f"{max(r['grid'] for r in rows):.0f} W") return win, ok_a, ok_b, ok_sum, ok_grid SCEN = { "(a) A in its maintenance charge phase": ( sc.day(sc.SF_CH, a_maint=sc.MAINT_CHARGE), "same"), "(a') ...with the charge shaped to 0 W": ( sc.day(sc.SF_CH, a_maint=sc.MAINT_CHARGE, shaped=True), "same"), "(b) A idle at 0 W (Solar first, the sun covers the house)": ( sc.day(sc.SF_CH, house_w=-10.0), "same"), "(c) A a normal charger in the same run": (sc.day(sc.CH_CH), "same"), "(g7) the OPT-15 rig: drain, then maintenance charge at 0 W": ( sc.g7(), "same"), # Not the fix's: A gives the house, and a charge ask means no other # battery gives (OPT-11 owner answer 3), so A rests, as on bf9019e. "(b2) A gives 800 W to the house": (sc.day(sc.SF_CH), "rests"), } for _name, (_rows, _kind) in SCEN.items(): win, ok_a, ok_b, ok_sum, ok_grid = window_checks(_rows, _name) check(f"{_name}: a probe happens (non-vacuous)", len(win) >= 1) if _kind == "same": check(f"{_name}: A's command is the same through each window, first " "and last cycle included", ok_a) check(f"{_name}: the grid rises by PROBE_W at most (plus 50 W of " "noise)", ok_grid) else: check(f"{_name}: A stops giving for the window (the one-sign rule, " "as on bf9019e), charges nothing and gives back after it", all(r["shares"].get(A, 0.0) == 0.0 for i, j in win for r in _rows[i + 1:j]) and all( _rows[j + 4]["shares"].get(A, 0.0) > 0.0 for i, j in win)) check(f"{_name}: B gets PROBE_W and nothing else", ok_b) check(f"{_name}: the commanded charge rises by PROBE_W at most", ok_sum) # The g7 sequence in detail: the probe after the drain. _g7 = SCEN["(g7) the OPT-15 rig: drain, then maintenance charge at 0 W"][0] _w7 = sc.windows(_g7) _after = [w for w in _w7 if _g7[w[0]]["now"] > sc.at(18.3, sc.GUTC0)] _peak7 = max(r["grid"] for r in _g7 if r["now"] > sc.at(18.3, sc.GUTC0)) print(f" g7: probes after the drain at " f"{[_g7[w[0]]['now'].astimezone(sc.L).strftime('%H:%M:%S') for w in _after]}" f", grid after the drain at most {_peak7:.0f} W (TEST-23: 3100 W)") check("g7: the probe after A's drain asks B for 200 W with A at 0 W " "(TEST-23: -2300 W on A, the grid at 3100 W)", _after and all( abs(r["shares"].get(A, 0.0)) < 1e-6 for i, j in _after for r in _g7[i:j + 4]) and _peak7 <= 800.0 + PW + 50.0) check("g7: during the drain no probe trades (a drain comes first)", all(r["action"] is None for r in _g7 if sc.at(18.0, sc.GUTC0) <= r["now"] < sc.at(18.28, sc.GUTC0))) _acts = [r["reason"] for i, j in _after for r in _g7[i:j + 1] if r["action"]] check("g7: the action log's trade line names the probe's watts, 200 W", _acts and all(f"{PW:.0f} W on {B}" in a for a in _acts)) check("no opposite-direction split on any of these cycles", not [r for _r, _ in SCEN.values() for r in _r if any(v > 0 for k, v in r["shares"].items() if r["action"]) and any(v < 0 for v in r["shares"].values())]) # ============================================================================ # print("\n4. battery modes on, no probe: byte-identical to release/1.0 bf9019e") # Printed by tools/opt20_scen.py on release/1.0 bf9019e (2026-10-15), # goodwe-addon-checks:1. AT_BASE = { "sf_ch": "422d6578a7decf1ae3749edd1898ff3cd0541d904c95f636cdd9b1de8c1a07c9", "ch_ch": "15fc25dd34186e56d34b752d37c9e8613295ba737938c92454acca08c83080fb", "ch_ch_maint": "e46e3c666dd63ab33b56e83c0a560c07cec7774a9a70560f55c62a6ae6f55920", } _got = sc.no_probe_digests() for _k in AT_BASE: print(f" {_k} {'same' if _got[_k] == AT_BASE[_k] else 'DIFFERS'} " f"{_got[_k][:16]}") check("with no refusal every share, grid, trade sentence, role and plant " "power of a scripted cheap run equals bf9019e's (A Solar first, A " "charging, A in maintenance charge)", _got == AT_BASE) print() if fails: print(f"{len(fails)} FAILED:") for f in fails: print(f" - {f}") sys.exit(1) print("all checks passed")