268 lines
12 KiB
Python
268 lines
12 KiB
Python
"""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")
|