"""Runnable check for OPT-11 P1, battery modes. `python3 test_modes.py` Each battery gets an on/off switch and a mode (Solar first, Leftover solar), kept in the store and filed as a SOFT unit limit at P_MODE (35). Drives the REAL `Controller` (read_inputs + cycle) through tools/opt11_identity.py's harness, in simulated time. DECISIONS/OPT-11-battery-modes.md has the record. 1. With battery modes off, every command is byte-identical to release/1.0 a5dddad, at one inverter and at two (digests measured on that tree). 2. Solar order: surplus solar fills the Solar first battery, then spills to the Leftover solar one; a full Solar first battery hands all of it on. 3. Off: the battery is asked for 0 W, the others carry the site, the loop is bounded and never freezes on watts no battery may give; all off at N=1 and N=2 asks 0 W. 4. Off stops and skips maintenance, with a log line; a battery in maintenance gets no mode claim. 5. The actions: `pack_use` and `pack_mode`, what they store, log and refuse. 6. A mode range without 0 W is dropped as a software fault, logged once. 7. The Batteries screen's data. Mutations (MUTATION.md, purged copies, graded over this file and the named tests) are listed in the pull request and DECISIONS/OPT-11-battery-modes.md. """ import asyncio import logging import os import sys HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, os.path.join(HERE, "tools")) import opt11_identity as rig # noqa: E402 from named_absence import nth, present # noqa: E402 import app.main as app_main # noqa: E402 from app import homeview, messages, packmode # noqa: E402 from app.control import MODE_BOUND # noqa: E402 from app.maintenance import DRAIN, IDLE # noqa: E402 from app.store import unit_key # noqa: E402 fails = [] def check(name, cond): print(f" {'ok ' if cond else 'FAIL'} {name}") if not cond: fails.append(name) ON = {"battery_modes_enabled": True} A, B = "inverter-a", "inverter-b" def store_with(**modes): """A store where each named battery has (use, mode).""" s = rig._Store() for name, (use, mode) in modes.items(): s.data[unit_key(name, packmode.USE_KEY)] = use s.data[unit_key(name, packmode.MODE_KEY)] = mode return s def spy_claims(): """Record every claim list main.py's own `resolve` is handed.""" seen, real = [], app_main.resolve def _spy(claims, **kw): seen.append(list(claims)) return real(claims, **kw) app_main.resolve = _spy return seen, lambda: setattr(app_main, "resolve", real) def spy_bounds(): """Record the `bound` main.py hands `compute()` on every cycle.""" seen, real = [], app_main.compute def _spy(*a, **kw): seen.append(kw.get("bound", "absent")) return real(*a, **kw) app_main.compute = _spy return seen, lambda: setattr(app_main, "compute", real) def flat(t): return 0.0 # ============================================================================ # print("1. battery modes off: byte-identical to release/1.0 a5dddad") # The digests below were printed by `python3 tools/opt11_identity.py` run on # the a5dddad tree (the same file copied in, nothing else changed), in the CI # image, on 2026-09-28. Each hashes 540 cycles: site target, every inverter's # written value, the loop's reason, integrator and saturation count, the # arbiter's sentence and the allocation record. A5DDDAD = { "n1": "fc3bca170c46920f08d86baee62ea883954a89ea6222d6bbf7d6652877226262", "n2": "a60003b1930fa3cba21f3220670d9172b9ff438347d77ba9bc3383fca1ceb15a", "n1_off": "fc3bca170c46920f08d86baee62ea883954a89ea6222d6bbf7d6652877226262", "n2_off": "a60003b1930fa3cba21f3220670d9172b9ff438347d77ba9bc3383fca1ceb15a", # Added in review (PR #149), printed on a5dddad the same way: both # batteries due for maintenance, and the arbiter alone on 6000 seeded # claim lists with the peak floor, trades, maintenance and refusals. "n1_maint": "5a21d43ec7cbf95a224bbb9bca13a8e62a00161101b586ae9fa68ab565992977", "n2_maint": "48a1ad96121592e207a4664fb2da9f5208a49b88e5802a358873a5d30cb8c18e", "resolve": "43a874216711936e14b9185bcc8d26c58fe4f3cfc3c1c8eaf89bffd95d6e81ef", } _claims, _undo_c = spy_claims() _bounds, _undo_b = spy_bounds() try: _now = rig.digests() finally: _undo_c() _undo_b() for k in A5DDDAD: print(f" {k:7} {'same' if _now[k] == A5DDDAD[k] else 'DIFFERS'} " f"{_now[k][:16]}") check("⭐ at N=1 and N=2, with the option absent and with it false, every " "command, reason and allocation record over 540 cycles each is " "byte-identical to a5dddad (house steps, a solar surplus, a battery " "at its cut-out level, at N=2 a refusal with probes, both batteries " "due for maintenance), and so is the arbiter on 6000 claim lists", _now == A5DDDAD) check("...and no cycle in those runs filed a `mode` claim or handed " "`compute()` a bound", _claims and not any(c.source == "mode" for cl in _claims for c in cl) and _bounds and set(_bounds) == {None}) # ============================================================================ # print("\n2. the solar order: Solar first fills, the rest spills to Leftover") _surplus = (lambda t: -3200.0) # 3200 W more solar than the house uses _full_a = (lambda u, t: 100.0 if (u == "a" and t >= 300) else 50.0) _rows, _c = rig.run(2, extra=ON, dur=600.0, load=_surplus, soc=_full_a, cut=lambda u, t: False, store=store_with(**{A: (True, packmode.SOLAR_FIRST), B: (True, packmode.LEFTOVER_SOLAR)})) _room = [r for r in _rows if 200 <= r[0] < 300] _full = [r for r in _rows if r[0] >= 500] def _cmd(r): """One row's written values, by battery; ABSENTs if the row is missing.""" return (dict(r[2]) if present(r) else {"a": nth((), 0, "a"), "b": nth((), 0, "b")}) print(f" A has room: A {_cmd(nth(_room, -1))['a']:+.0f} W, B " f"{_cmd(nth(_room, -1))['b']:+.0f} W | A full: A {_cmd(nth(_full, -1))['a']:+.0f} " f"W, B {_cmd(nth(_full, -1))['b']:+.0f} W") check("⭐ while A (Solar first) has room it takes all it can, 2500 W, and B " "(Leftover solar) gets only the spill, about 700 W", present(nth(_room, -1)) and all(_cmd(r)["a"] == -2500.0 and -800.0 <= _cmd(r)["b"] <= -600.0 for r in _room)) check("⭐ once A is full, the surplus spills to B, up to B's rating", present(nth(_full, -1)) and all(_cmd(r)["a"] == 0.0 and _cmd(r)["b"] == -2500.0 for r in _full)) _rows_off, _ = rig.run(2, dur=300.0, load=_surplus, soc=lambda u, t: 50.0, cut=lambda u, t: False) check("...and the same site with battery modes off splits 50/50, which is " "what says the order above comes from the modes", _cmd(_rows_off[-1])["a"] == _cmd(_rows_off[-1])["b"] < -1500.0) _give = rig.run(2, extra=ON, dur=300.0, load=lambda t: 3000.0, soc=lambda u, t: 50.0, cut=lambda u, t: False, store=store_with(**{A: (True, packmode.SOLAR_FIRST), B: (True, packmode.LEFTOVER_SOLAR)}))[0] check("while giving power the modes keep OPT-09's weights: A Solar first and " "B Leftover solar, both half full, share a 3000 W house equally", _cmd(_give[-1])["a"] == _cmd(_give[-1])["b"] > 1400.0) # ============================================================================ # print("\n3. Off: 0 W on that battery, the loop bounded, no freeze") _rows, _c = rig.run(2, extra=ON, dur=600.0, load=lambda t: 4000.0, soc=lambda u, t: 50.0, cut=lambda u, t: False, store=store_with(**{B: (False, packmode.SOLAR_FIRST)})) _tail = _rows[40:] print(f" house 4000 W, B off: A {_cmd(_tail[-1])['a']:+.0f} W, B " f"{_cmd(_tail[-1])['b']:+.0f} W, site {_tail[-1][1]:+.0f} W, reason " f"{_tail[-1][3]}, {_tail[-1][7]}") check("⭐ B (Off) is asked for 0 W on every cycle and A carries the house up " "to its rating", all(_cmd(r)["b"] == 0.0 for r in _rows) and all(_cmd(r)["a"] == 2500.0 for r in _tail)) check("⭐ the loop stays inside what the modes allow (2500 W) and never " "freezes: no saturated-freeze reason, and the saturation count stays " "below the freeze", all(r[1] <= 2500.0 for r in _rows) and not any(r[3] == "saturated-freeze" for r in _rows) and max(r[5] for r in _rows) < _c.tuning.saturation_cycles and _tail[-1][3] == MODE_BOUND) check("...and nothing is left unmet: the allocation places the whole target", all("UNMET" not in (r[7] or "") for r in _tail)) _bounds, _undo_b = spy_bounds() try: rig.run(2, extra=ON, dur=100.0, load=lambda t: 4000.0, soc=lambda u, t: 50.0, cut=lambda u, t: False, store=store_with(**{B: (False, packmode.SOLAR_FIRST)})) finally: _undo_b() check("...because main.py hands compute() the loop bound, B's 0 W included: " "[-2500, 2500]", _bounds and set(_bounds) == {(-2500.0, 2500.0)}) # Review (PR #149): a unit the allocator drops (stale link) must not count in # the loop bound either, or the loop asks for watts nobody live can place. _rows, _c = rig.run(2, extra=ON, dur=10.0, load=flat, soc=lambda u, t: 50.0, cut=lambda u, t: False) _both = _c.loop_bound([app_main.Claim.limit( "supervised", app_main.P_SAFETY, -2500.0, 2500.0, actuator=_c.actuator_of("supervised", i)) for i in _c.inverters]) _c.live_units = lambda: (A,) _one_live = _c.loop_bound([app_main.Claim.limit( "supervised", app_main.P_SAFETY, -2500.0, 2500.0, actuator=_c.actuator_of("supervised", i)) for i in _c.inverters]) check("the loop bound counts only live units: both live [-5000, 5000], B's " "link stale [-2500, 2500]", _both == (-5000.0, 5000.0) and _one_live == (-2500.0, 2500.0)) for _n in (1, 2): _st = store_with(**{A: (False, packmode.SOLAR_FIRST), B: (False, packmode.LEFTOVER_SOLAR)}) if _n == 1: _st = rig._Store() _st.data[unit_key(app_main.UNIT_A, packmode.USE_KEY)] = False _rows, _c = rig.run(_n, extra=ON, dur=300.0, load=lambda t: 3000.0, soc=lambda u, t: 50.0, cut=lambda u, t: False, store=_st) check(f"every battery off at N={_n}: every command is 0 W, the site " f"target is 0 W, the integrator stays at 0 and nothing freezes", all(all(v == 0.0 for _u, v in r[2]) and r[1] == 0.0 and r[4] == 0.0 and r[3] != "saturated-freeze" for r in _rows)) _rows1, _ = rig.run(1, extra=ON, dur=300.0, load=lambda t: 1800.0, soc=lambda u, t: 50.0) _rows1o, _ = rig.run(1, dur=300.0, load=lambda t: 1800.0, soc=lambda u, t: 50.0) check("at N=1 a battery that is on (Solar first by default) follows the " "house exactly as with battery modes off", [r[:6] for r in _rows1] == [r[:6] for r in _rows1o] and _cmd(_rows1[-1])["a"] > 1700.0) # ============================================================================ # print("\n4. Off stops and skips maintenance; maintenance outranks the mode") MAINT = {"maintenance_enabled": True, "maintenance_start_hour": 11} _st = store_with(**{A: (False, packmode.SOLAR_FIRST)}) _rows, _c = rig.run(2, extra={**ON, **MAINT}, dur=60.0, load=lambda t: 500.0, soc=lambda u, t: 60.0, cut=lambda u, t: False, store=_st) print(f" both due at 11:00, A off: A {_c.maints[0].phase}, " f"B {_c.maints[1].phase}") check("⭐ a battery that is off starts no cycle, and does not hold the slot: " "both are due, A (first in the roster) is off, so B starts", _c.maints[0].phase == IDLE and _c.maints[1].phase == DRAIN) _st2 = store_with() _seen, _undo_c = spy_claims() _msgs = [] async def _switch_b_off(c, t): if t == 30.0: _msgs.append(await c.handle_action("pack_use", unit=B, value=False)) try: _rows, _c = rig.run(2, extra={**ON, **MAINT}, dur=60.0, load=lambda t: 500.0, soc=lambda u, t: 60.0, cut=lambda u, t: False, store=_st2, before=_switch_b_off) finally: _undo_c() _during = [cl for cl in _seen[:6]] print(f" B off at 30 s while A drains: A {_c.maints[0].phase}, " f"B {_c.maints[1].phase}; events: " f"{[e for e in _c.events if 'switched' in str(e) or 'ABORT' in str(e)]}") check("while A runs its cycle it gets no mode claim (maintenance at 60/40 " "outranks the mode) and B, idle, gets one", all({c.actuator for c in cl if c.source == "mode"} == {B} for cl in _during)) _st3 = store_with() _st3.data[unit_key(B, "phase")] = DRAIN _st3.data["maint_schema_version"] = 1 async def _b_off_now(c, t): if t == 10.0: _msgs.append(await c.handle_action("pack_use", unit=B, value=False)) _rows, _c = rig.run(2, extra=ON, dur=30.0, load=lambda t: 500.0, soc=lambda u, t: 60.0, cut=lambda u, t: False, store=_st3, before=_b_off_now) _ev = [str(e) for e in _c.events] print(f" B mid-drain, switched off at 10 s: B {_c.maints[1].phase}; " f"{[e for e in _ev if 'switched off' in e or 'ABORT' in e]}") check("⭐ switching off a battery that is mid-cycle stops its cycle at once, " "with a log line saying why", _c.maints[1].phase == IDLE and any("battery switched off" in e for e in _ev) and _cmd(_rows[-1])["b"] == 0.0) _st4 = store_with(**{B: (False, packmode.SOLAR_FIRST)}) _st4.data[unit_key(B, "phase")] = DRAIN _st4.data["maint_schema_version"] = 1 _rows, _c = rig.run(2, extra=ON, dur=20.0, load=lambda t: 500.0, soc=lambda u, t: 60.0, cut=lambda u, t: False, store=_st4) check("...and a store that says Off for a battery mid-cycle (a restart) stops " "it on the first cycle, and it is never asked to drain", _c.maints[1].phase == IDLE and all(_cmd(r)["b"] == 0.0 for r in _rows)) # ============================================================================ # print("\n5. the actions: pack_use and pack_mode") _rows, _c = rig.run(2, extra=ON, dur=10.0, load=flat, soc=lambda u, t: 50.0, cut=lambda u, t: False) _r1 = asyncio.run(_c.handle_action("pack_mode", unit=B, value=packmode.LEFTOVER_SOLAR)) _r2 = asyncio.run(_c.handle_action("pack_use", unit=A, value=False)) print(f" {_r1} {_r2}") check("pack_mode stores the mode under the battery's own key and says what " "it now is", _r1 == {"unit": B, "use": True, "mode": packmode.LEFTOVER_SOLAR} and _c.store.data[unit_key(B, packmode.MODE_KEY)] == packmode.LEFTOVER_SOLAR) check("pack_use stores the switch, and both are written to the action log", _r2 == {"unit": A, "use": False, "mode": packmode.SOLAR_FIRST} and _c.store.data[unit_key(A, packmode.USE_KEY)] is False and any("inverter-a switched off by operator" in str(e) for e in _c.events) and any("inverter-b mode set to leftover solar" in str(e) for e in _c.events)) _bad = [asyncio.run(_c.handle_action("pack_use", unit="inverter-z", value=True)), asyncio.run(_c.handle_action("pack_use", unit=A, value="yes")), # 0 == False in Python: a stored 0 would read back as ON. asyncio.run(_c.handle_action("pack_use", unit=A, value=0)), asyncio.run(_c.handle_action("pack_use", unit=A, value=1)), asyncio.run(_c.handle_action("pack_mode", unit=A, value=packmode.CHEAP_HOURS)), asyncio.run(_c.handle_action("pack_mode", unit=A, value=None))] check("an unknown battery, a switch that is not true or false (0 and 1 " "included), and a mode " "this release does not offer (Cheap hours arrives in P3) are refused, " "with a plain reason, and store nothing", all(isinstance(x, dict) and set(x) == {"error"} for x in _bad) and unit_key(A, packmode.MODE_KEY) not in _c.store.data and _c.store.data[unit_key(A, packmode.USE_KEY)] is False) _rows, _c0 = rig.run(2, dur=10.0, load=flat, soc=lambda u, t: 50.0, cut=lambda u, t: False) _r0 = asyncio.run(_c0.handle_action("pack_use", unit=A, value=False)) check("with battery modes off the actions are refused and store nothing", set(_r0) == {"error"} and unit_key(A, packmode.USE_KEY) not in _c0.store.data) check("the old actions still take no battery and answer as before", asyncio.run(_c0.handle_action("nonsense")) is None) check("a store value from a later version (an unknown mode) reads as the " "default, Solar first and on, never as Off", packmode.read({unit_key(A, packmode.MODE_KEY): "grid_only", unit_key(A, packmode.USE_KEY): "maybe"}, A) == packmode.PackMode(True, packmode.SOLAR_FIRST)) # ============================================================================ # print("\n6. a mode range without 0 W is a software fault, never obeyed") _real_range = packmode.mode_range packmode.mode_range = lambda pm, r: (100.0, 900.0) _log = [] class _H(logging.Handler): def emit(self, rec): _log.append((rec.levelno, rec.getMessage())) _h = _H() logging.getLogger("goodwe").addHandler(_h) try: _rows, _c = rig.run(2, extra=ON, dur=60.0, load=lambda t: -2000.0, soc=lambda u, t: 50.0, cut=lambda u, t: False) finally: packmode.mode_range = _real_range logging.getLogger("goodwe").removeHandler(_h) _errs = [msg for lv, msg in _log if lv >= logging.ERROR and "MODE_RANGE" in msg] print(f" {_errs}") check("⭐ both batteries still charge from the surplus (the forced range is " "dropped, so no battery is made to give power) and no split is refused", all(v <= 0.0 for _u, v in _rows[-1][2]) and _cmd(_rows[-1])["a"] < 0.0 and not any("REFUSED" in (r[7] or "") for r in _rows)) check("...and it is logged once per battery, as an ERROR naming the fault " "code, not once per cycle", len(_errs) == 2 and any(A in e for e in _errs) and any(B in e for e in _errs)) # ============================================================================ # print("\n7. the Batteries screen's data") _rows, _c = rig.run(2, extra=ON, dur=10.0, load=flat, soc=lambda u, t: 50.0, cut=lambda u, t: False, store=store_with(**{B: (False, packmode.LEFTOVER_SOLAR)})) _bat = homeview.battery_rows(_c, False, True) _bat_nl = homeview.build(_c, level="ok", checks=[], lang="nl")["batteries"] print(f" {[(b['name'], b['use'], b['mode_key'], b['mode']) for b in _bat]}") check("with battery modes on each row carries its switch, its state and its " "stored mode, in the page's words", [(b["use"], b["mode_key"], b["mode_choice"], b["mode"]) for b in _bat] == [(True, "solar_first", "solar_first", "Solar first"), (False, "off", "leftover_solar", "Off")] and _bat_nl[1]["mode"] == "Uit") _rows, _c0 = rig.run(2, dur=10.0, load=flat, soc=lambda u, t: 50.0, cut=lambda u, t: False) check("...and with them off the rows are what they were: no new keys, the " "mode reads as before", all("mode_key" not in b and b["mode"] == homeview.mode_house() for b in homeview.battery_rows(_c0, False, True))) # ============================================================================ # print("\n8. the page's route: /action and /status over the real web app") import socket # noqa: E402 import aiohttp # noqa: E402 from aiohttp import web as aioweb # noqa: E402 from app import web as app_web # noqa: E402 _rows, _c = rig.run(2, extra=ON, dur=10.0, load=flat, soc=lambda u, t: 50.0, cut=lambda u, t: False) async def _page(): # The configuration routes are not used here; a stand-in keeps the app # from building a Supervisor client. app = app_web.build_app(_c, selfcfg=object()) runner = aioweb.AppRunner(app) await runner.setup() sock = socket.socket() sock.bind(("127.0.0.1", 0)) port = sock.getsockname()[1] sock.close() await aioweb.TCPSite(runner, "127.0.0.1", port).start() out = {} try: async with aiohttp.ClientSession() as s: root = f"http://127.0.0.1:{port}/" for key, body in ( ("mode", {"action": "pack_mode", "unit": B, "value": packmode.LEFTOVER_SOLAR}), ("off", {"action": "pack_use", "unit": A, "value": False}), ("bad", {"action": "pack_mode", "unit": A, "value": "grid_only"})): async with s.post(root + "action", json=body) as r: out[key] = (r.status, await r.json(content_type=None)) async with s.get(root + "status") as r: out["status"] = (r.status, await r.json(content_type=None)) async with s.get(root) as r: out["page"] = (r.status, await r.text()) finally: await runner.cleanup() return out _web = asyncio.run(_page()) print(f" {_web['mode']} {_web['off']} {_web['bad']}") check("POST /action with pack_mode and pack_use names a battery and a value, " "and answers what it stored", _web["mode"] == (200, {"ok": True, "result": { "unit": B, "use": True, "mode": packmode.LEFTOVER_SOLAR}}) and _web["off"][0] == 200 and _web["off"][1]["result"]["use"] is False) check("...a value this release does not offer is a 400 with the plain reason " "the page shows", _web["bad"][0] == 400 and _web["bad"][1]["ok"] is False and _web["bad"][1]["error"] == messages.pack_refused("value")) _sb = {b["name"]: b for b in _web["status"][1].get("batteries", [])} check("...and /status then shows both batteries' switch and mode for the " "Batteries screen", _web["status"][0] == 200 and (_sb.get(A, {}).get("mode_key"), _sb.get(B, {}).get("mode_key")) == ("off", packmode.LEFTOVER_SOLAR)) check("the page carries the mode names from app/ui_text for its cards", _web["page"][0] == 200 and "Leftover solar" in _web["page"][1] and "Overschot zonne-energie" in _web["page"][1]) print() if fails: print(f"{len(fails)} FAILED: {', '.join(fails)}") sys.exit(1) print("all checks passed")