adminandClaude Opus 5.5
6872353e2e
At N>=2 a battery at or under its own peak reserve now weighs 0 whenever the site gives (not only for the peak), a draining battery's pin becomes a floor at its drain rate so it rises before the held one gives, and a quarter's edge is graded on the next quarter when its first telegram is late. SAFETY-25 option A stays: the held battery is still drawn once the others are full. N=1 unchanged. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
530 lines
25 KiB
Python
530 lines
25 KiB
Python
"""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 = {k: rig.digest(rig.run(**v)[0])
|
|
for k, v in rig.SCENARIOS.items() if k != "n2_maint"}
|
|
_now["resolve"] = rig.digest(rig.resolve_rows())
|
|
finally:
|
|
_undo_b()
|
|
# SAFETY-85: n2_maint (maintenance at N=2) is the one run with a bound.
|
|
_bounds_m, _undo_b = spy_bounds()
|
|
try:
|
|
_now["n2_maint"] = rig.digest(rig.run(**rig.SCENARIOS["n2_maint"])[0])
|
|
finally:
|
|
_undo_c()
|
|
_undo_b()
|
|
# SAFETY-78 changes n2_maint ON PURPOSE: on a5dddad and 5a4cdc2 its N=2 drain
|
|
# was refused on 392 of 540 cycles (site at 0 W); now none is. Printed on the
|
|
# SAFETY-78 tree the same way. Every other digest is unchanged.
|
|
# SAFETY-85 changes it again ON PURPOSE: while A's drain pin stands the loop
|
|
# is bounded by what the batteries can place, so when A drops out at t=1655 B
|
|
# follows the house down at once instead of unwinding 920 W too high. Was
|
|
# cd4d9cb5b7e29ff4ab4fe65dab5106683d8a599f09323fbdee4018ad9b28fcd1.
|
|
SAFETY78_N2_MAINT = \
|
|
"1afcd46b1b987868b110d2b0f4a7555848c9aa4728a836d4a2ce2a2107268497"
|
|
# SAFETY-91 changes n2 and n2_off ON PURPOSE: from t=1600 battery A is at its
|
|
# 20 % reserve, so B carries the house first and A gives last. With
|
|
# `_held_back()` reverted to SAFETY-87's (held only for the peak) both are
|
|
# a5dddad's again, which the check below grades.
|
|
SAFETY91_N2 = \
|
|
"9535c08f556d6d50bec8fb0872445173480c29b10df1a7c437ef73d6ddb2814d"
|
|
_real_hb = app_main.Controller._held_back
|
|
app_main.Controller._held_back = lambda self: self._peak_reserve()
|
|
try:
|
|
_n2_s87 = rig.digest(rig.run(**rig.SCENARIOS["n2"])[0])
|
|
finally:
|
|
app_main.Controller._held_back = _real_hb
|
|
_S91 = ("n2", "n2_off", "n2_maint")
|
|
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; n2, n2_off and n2_maint apart, below), and so "
|
|
"is the arbiter on 6000 claim lists",
|
|
{k: v for k, v in _now.items() if k not in _S91}
|
|
== {k: v for k, v in A5DDDAD.items() if k not in _S91})
|
|
check("...n2 and n2_off are the SAFETY-91 digest (A at its reserve gives "
|
|
"last), and SAFETY-87's held rule gives a5dddad's back",
|
|
_now["n2"] == _now["n2_off"] == SAFETY91_N2 != A5DDDAD["n2"]
|
|
and _n2_s87 == A5DDDAD["n2"])
|
|
check("...and n2_maint is the SAFETY-78 digest: the N=2 drain that a5dddad "
|
|
"refused on 392 of 540 cycles now runs (test_safety78.py), bounded by "
|
|
"SAFETY-85",
|
|
_now["n2_maint"] == SAFETY78_N2_MAINT != A5DDDAD["n2_maint"])
|
|
check("...and no cycle in those runs filed a `mode` claim, and only "
|
|
"n2_maint's (SAFETY-85: maintenance at N=2) 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}
|
|
and any(b is not None for b in _bounds_m))
|
|
|
|
|
|
# ============================================================================ #
|
|
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="sell_only")),
|
|
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 "
|
|
"that does not exist 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")
|