Files
glenn schrooyenandClaude Opus 5.5 4b81387908
checks / unit (pull_request) Successful in 25m35s
infra-pin / fresh (pull_request) Successful in 2s
SAFETY-85: no false 'not keeping up' while maintenance or the peak guard sets the battery
The loop's integrator bound counts only on a cycle the loop decided. A battery
another owner set (maintenance, the peak guard, a safety limit) alarms only
when it gives more than saturation_w less than told for saturation_cycles
cycles. At N>=2 the loop is bounded while maintenance holds a battery, so it
does not wind and control comes back one slew step a cycle. The status row
counts episodes, not cycles. Digests re-pinned where behaviour changed on
purpose; DECISIONS/SAFETY-85-no-false-not-keeping-up.md has the list.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019DBUfvndgHwaV339Fvd2gr
2026-10-01 19:05:27 +02:00

511 lines
22 KiB
Python

"""Runnable check for SAFETY-84: a maintenance charge at N=2 no longer stalls.
`python3 test_safety84.py`
In a cheap window the charging battery's own `hi` makes it charge. At two
inverters the loop's site target sat above it, the other battery had to give,
the allocator refused the opposite directions and the site sat at 0 W. Now a
site `hi` (`charge`, P_MAINT_SHAPE) keeps the site target at or below the
charge, and the charge is sized to the quarter's headroom so it yields to the
peak guard. Drives the REAL Controller (tools/safety84_identity.py's harness).
DECISIONS/SAFETY-84-charge-at-two.md has the record.
1. N=2 in a cheap window, house below, at and above the charge rate and a
solar surplus: the charge runs, no opposite split, no refusal, and it ends.
2. The peak guard: the charge yields just enough to hold the aim (cap -
margin, and a banked month peak - margin); over the aim the guard wins,
the charging battery gives too, and nothing parks; no false 'Load too
high'. Without P1 the charge is not sized, as at N=1.
3. Outside a cheap window nothing changes.
4. A safety stop parks both batteries without a refusal, and the charge
resumes after it.
5. One battery refusing, or dropped, and B Off with battery modes on.
6. A trade outranked by the charge (P_ARBITRAGE < P_MAINT_SHAPE).
7. A seeded sweep of the ceiling against the loop, trades, refusals, packs
and modes.
8. The N=1 digests equal release/1.0 9d5b7d1, and the N=2 ones are pinned.
Mutations (MUTATION.md, purged copies) are listed in the pull request.
"""
import logging
import os
import random
import sys
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 opt11_identity as rig # noqa: E402
import opt11p3_identity as p3 # noqa: E402
import safety84_identity as s84 # noqa: E402
from app import arbitrage, packmode # noqa: E402
from app.arbiter import (Claim, P_ARBITRAGE, P_LOOP, P_MAINT_SHAPE, # noqa
P_MODE, P_SAFETY, resolve)
from app.main import Controller # noqa: E402
from app.maintenance import CHARGE, HOLD # 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)
A, B = "inverter-a", "inverter-b"
AIM = 3450.0 # peak_cap_w 3500 - peak_margin_w 50
TOL = 10.0 # W over the aim a banked quarter may show (SAFETY-81)
def cmd(r):
return dict(r[2])
def refused(rows):
return sum(1 for r in rows if r[7] and "REFUSED" in r[7])
def opposite(rows):
return sum(1 for r in rows if cmd(r)["a"] * cmd(r)["b"] < 0)
def charge_claims(r):
"""The site `charge` claims main.py handed resolve() on row `r`."""
return [c for c in r[-1][0] if c[0] == "charge"]
def banked(c):
return [b.offtake_avg_w for b in c.p1.blocks]
def alarmed(c):
"""Did SAFETY-81's 'Load too high' alarm fire on this run? Its quarter
stamp too, since ALLOCATOR rows can push its row out of the 40-row log
(SAFETY-85)."""
return c._peak_alarm_block is not None or any(
"PEAK_GUARD_SATURATED" in e for e in c.events)
def run(house, n=2, **kw):
return s84.run(n, load=house if callable(house) else lambda t: house,
**kw)
def rise(u, t):
"""A fills from 60 % (1 % a cycle); B stays at 50 %."""
return min(100.0, 60.0 + t / 5.0) if u == "a" else 50.0
# ============================================================================ #
print("1. N=2 in a cheap window: the charge runs whatever the house uses")
RUNS = {h: run(h, dur=900.0) for h in (800.0, 2000.0, 3000.0, -1500.0)}
for h, (rows, c) in RUNS.items():
print(f" house {h:+.0f} W: last A {cmd(rows[-1])['a']:+.0f} W, B "
f"{cmd(rows[-1])['b']:+.0f} W; {rows[-1][6]}; refusals "
f"{refused(rows)}")
check("⭐ below, at and above the charge rate and under a solar surplus, "
"battery A charges on EVERY cycle (it used to sit at 0 W from the "
"first cycle)",
all(cmd(r)["a"] < 0.0 for rows, _c in RUNS.values() for r in rows)
and all(c.maints[0].phase == CHARGE for _r, c in RUNS.values()))
check("⭐ no cycle splits the site both ways and the allocator refuses "
"nothing: battery B never gives while A charges",
all(opposite(rows) == 0 and refused(rows) == 0
and all(cmd(r)["b"] <= 0.0 for r in rows)
for rows, _c in RUNS.values()))
check("...and nothing is logged per cycle: no ARBITER line in the action log",
not any("ARBITER" in e for _r, c in RUNS.values() for e in c.events))
check("with the quarter's headroom above the charge rate, A charges at the "
"full 2000 W and the Why names the charge",
all(cmd(r)["a"] == -2000.0 for h in (800.0, -1500.0)
for r in RUNS[h][0])
and "limited by charge" in RUNS[800.0][0][-1][6])
print(f" Why: {RUNS[800.0][1].why_words}")
check("...and the page's Why reads as at one battery: 'held back by battery "
"maintenance'",
RUNS[800.0][1].why_words.endswith("held back by battery maintenance"))
_end_rows, _end = run(800.0, dur=600.0, soc=rise)
check("⭐ and the charge ENDS: A reaches its 99 % target and the cycle moves "
"on to its hold, with no refusal on the way",
_end.maints[0].phase == HOLD and refused(_end_rows) == 0
and any("charge complete" in e for e in _end.events))
_real = Controller._charge_ceiling
Controller._charge_ceiling = lambda self, claims: None
try:
_bite, _bc = run(800.0, dur=120.0)
finally:
Controller._charge_ceiling = _real
print(f" without the ceiling: refusals {refused(_bite)}/{len(_bite)}, "
f"last A {cmd(_bite[-1])['a']:+.0f} W")
check("non-vacuous: with the ceiling taken out, the same fixture refuses "
"every cycle and parks A at 0 W (the reported defect)",
refused(_bite) == len(_bite) and cmd(_bite[-1])["a"] == 0.0)
# ============================================================================ #
print("\n2. the peak guard outranks the charge")
_q_rows, _q = run(3000.0, dur=2700.0)
print(f" house 3000: banked {[round(v) for v in banked(_q)]} W, last A "
f"{cmd(_q_rows[-1])['a']:+.0f} W")
check("⭐ house 3000 W, aim 3450 W: the charge yields just enough - every "
"banked quarter lands on the aim, not on house + 2000 W",
len(banked(_q)) >= 2
and all(AIM - TOL <= v <= AIM + TOL for v in banked(_q)))
check("...and it yields by charging less, not by stopping: A charges and B "
"is 0 W on every cycle, nothing is refused, no 'Load too high'",
all(cmd(r)["a"] < 0.0 and cmd(r)["b"] == 0.0 for r in _q_rows)
and refused(_q_rows) == 0 and not alarmed(_q))
check("...and the peak guard never has to step in (the charge stays under "
"what the quarter can afford)",
not any(r[-1][1] and r[-1][1][0] for r in _q_rows))
_o_rows, _o = run(5000.0, dur=2700.0)
_both = [r for r in _o_rows
if charge_claims(r) and any(c[0] == "peak" for c in r[-1][0])]
print(f" house 5000: banked {[round(v) for v in banked(_o)]} W, last "
f"{cmd(_o_rows[-1])}, {_o_rows[-1][6]}")
check("⭐ house 5000 W, over the aim alone: the peak guard wins - every "
"banked quarter is at or under the aim, both batteries give in the "
"same direction, nothing is refused and nothing parks at 0 W",
all(v <= AIM + TOL for v in banked(_o)) and opposite(_o_rows) == 0
and refused(_o_rows) == 0
and sum(cmd(r)["a"] + cmd(r)["b"] > 0.0 for r in _o_rows) > 100)
check("...and the guard's `lo` never meets the charge's `hi` in one cycle "
"(the charge is suspended while the guard holds)", not _both)
check("...with no false 'Load too high' alarm", not alarmed(_o))
_k_rows, _k = run(3000.0, dur=2700.0, month_peak=4000.0)
print(f" month peak 4000 banked, house 3000: banked "
f"{[round(v) for v in banked(_k)]} W")
check("⭐ under a month peak banked at 4000 W the aim is 3950 W: the charge "
"takes the extra headroom and every banked quarter lands on 3950 W",
len(banked(_k)) >= 2
and all(3950.0 - TOL <= v <= 3950.0 + TOL for v in banked(_k))
and refused(_k_rows) == 0 and not alarmed(_k))
_real = Controller._charge_headroom
Controller._charge_headroom = lambda self, floor, claims: floor
try:
_nh_rows, _nh = run(3000.0, dur=2700.0)
finally:
Controller._charge_headroom = _real
_gave = sum(1 for r in _nh_rows if cmd(r)["a"] > 0.0 or cmd(r)["b"] > 0.0)
print(f" without the headroom: cycles a battery gives {_gave}, banked "
f"{[round(v) for v in banked(_nh)]} W")
check("non-vacuous: charging at the full rate instead, the guard has to "
"step in and the batteries give back on many cycles",
_gave > 50)
def peak_run(house):
rows, c = run(house, dur=1800.0)
return rows, c, [r for r in rows if r[-1][1] and r[-1][1][0]]
# A 6500 W house needs 3050 W off the meter: more than B's 2500 W alone.
_p_rows, _p, _p_eng = peak_run(6500.0)
print(f" house 6500: banked {[round(v) for v in banked(_p)]} W, A gives "
f"on {sum(cmd(r)['a'] > 0.0 for r in _p_eng)} of {len(_p_eng)} "
f"engaged cycles")
check("⭐ house 6500 W, more than B alone can cover: the charge is suspended "
"and A, the battery being filled, gives too; the quarter lands on the "
"aim, nothing is refused, no 'Load too high'",
len(_p_eng) > 100 and all(cmd(r)["a"] > 0.0 for r in _p_eng)
and all(v <= AIM + TOL for v in banked(_p))
and refused(_p_rows) == 0 and not alarmed(_p))
_real = s84.m.peakguard.shaping_suspended
s84.m.peakguard.shaping_suspended = lambda *a: False
try:
_ns_rows, _ns, _ns_eng = peak_run(6500.0)
finally:
s84.m.peakguard.shaping_suspended = _real
check("non-vacuous: with the suspension taken out, A is held at 0 W, the "
"quarter goes over the cap and 'Load too high' fires",
not any(cmd(r)["a"] > 0.0 for r in _ns_eng)
and max(banked(_ns)) > 3500.0 and alarmed(_ns))
def prices_only(c, t):
"""A cheap price and no P1 telegram (`meter_source: off`)."""
if not getattr(c.hass, "_px", False):
read, px = c.hass.number, {"sensor.pnow": 0.05, "sensor.pavg": 0.20}
async def number(e, invert=False):
return px[e] if e in px else await read(e, invert)
c.hass.number, c.hass._px = number, True
_p1 = s84.s79.P1
s84.s79.P1 = {"meter_source": "off", "peak_cap_w": 3500}
try:
_off_rows, _off = run(800.0, dur=600.0, before=prices_only)
finally:
s84.s79.P1 = _p1
check("⭐ with no P1 meter nothing measures the quarter and the charge is not "
"sized, as at N=1: A charges at 2000 W on every cycle, nothing refused",
_off.peak_guard is None and all(cmd(r)["a"] == -2000.0 for r in _off_rows)
and refused(_off_rows) == 0)
# ============================================================================ #
print("\n3. outside a cheap window nothing changes")
_d = {h: run(h, cheap=False, dur=600.0) for h in (800.0, -1500.0)}
check("⭐ no `charge` claim is filed outside a cheap window: B covers the "
"house, the charge takes solar only, nothing is refused",
not any(charge_claims(r) for rows, _c in _d.values() for r in rows)
and cmd(_d[800.0][0][-1]) == {"a": 0.0, "b": 800.0}
and cmd(_d[-1500.0][0][-1])["a"] < 0.0
and all(refused(rows) == 0 for rows, _c in _d.values()))
# ============================================================================ #
print("\n4. a safety stop outranks the charge")
def blind_at(t0, t1):
"""A's power reading is missing from `t0` to `t1`: a safety SET."""
def before(c, t):
s84.feed(True)(c, t)
if not getattr(c.hass, "_blind", False):
real = c.hass.number
async def number(e, invert=False):
if e == "sensor.batt_a" and t0 <= c.hass.t < t1:
return None
return await real(e, invert)
c.hass.number, c.hass._blind = number, True
return before
_s_rows, _s = run(800.0, dur=600.0, before=blind_at(150.0, 400.0))
_stop = [r for r in _s_rows if 175.0 <= r[0] < 400.0]
_last = _s_rows[79] # t = 395 s, still without the reading
print(f" A's reading lost 150-400 s: {_last[2]}, {_last[6]}; after: "
f"{_s_rows[-1][2]}")
check("⭐ with a reading gone the safety SET parks both batteries at 0 W "
"with no refusal and no ARBITER line (it used to refuse every cycle)",
all(cmd(r) == {"a": 0.0, "b": 0.0} for r in _stop)
and _last[0] == 395.0 and _last[6].startswith("safety -> 0 W")
and refused(_s_rows) == 0 and not any("ARBITER" in e for e in _s.events))
check("...and the charge ran before and runs again after",
cmd(_s_rows[20])["a"] == -2000.0 and cmd(_s_rows[-1])["a"] == -2000.0)
_full = _s._charge_headroom(2000.0, [])
check("the stop test reads only SITE SETs: a unit's own safety SET (on B) "
"or a site limit does not cut A's charge, a site safety SET does",
_s._charge_headroom(2000.0, [Claim.set("failsafe", P_SAFETY, 0.0,
actuator=B)]) == _full > 0.0
and _s._charge_headroom(2000.0, [Claim.limit("peak", P_SAFETY,
lo=0.0)]) == _full
and _s._charge_headroom(2000.0, [Claim.set("safety", P_SAFETY,
0.0)]) == 0.0)
def stop_at(t0):
def before(c, t):
s84.feed(True)(c, t)
if t >= t0:
c.auto = False
return before
_st_rows, _st = run(800.0, dur=400.0, before=stop_at(200.0))
check("control switched off mid-charge: both batteries at 0 W, no refusal",
cmd(_st_rows[-1]) == {"a": 0.0, "b": 0.0} and refused(_st_rows) == 0)
# ============================================================================ #
print("\n5. one battery refusing, dropped, or Off")
def a_takes_nothing(c, t):
"""A reads 0 W whenever it is asked to charge (a battery that refuses)."""
s84.feed(True)(c, t)
if not getattr(c.hass, "_rt", False):
real = c.hass.power
c.hass.power = lambda u: (0.0 if u == "a" and c.hass.cmd["a"] < 0.0
else real(u))
c.hass._rt = True
_r_rows, _r = run(800.0, dur=900.0, before=a_takes_nothing)
print(f" A refuses to charge: {_r_rows[-1][2]}, {_r_rows[-1][6]}")
check("⭐ A refuses to charge (SAFETY-73): no ceiling is filed for it, B "
"covers the 800 W house, and no cycle is split both ways or refused",
_r.inverters[0].refused == -1 and cmd(_r_rows[-1]) == {"a": 0.0,
"b": 800.0}
and not charge_claims(_r_rows[-1])
and opposite(_r_rows) == 0 and refused(_r_rows) == 0)
_g_rows, _g = run(3000.0, dur=600.0, cut=lambda u, t: u == "b")
check("B refuses to give: A still charges and nothing is refused",
all(cmd(r)["a"] < 0.0 for r in _g_rows) and refused(_g_rows) == 0)
def reject_a(c, t):
"""Every write to A is rejected, so A's link goes stale (not live)."""
s84.feed(True)(c, t)
if not getattr(c.hass, "_rej", False):
real = c.hass.set_number
async def set_number(e, v):
return False if e == "number.sp_a" else await real(e, v)
c.hass.set_number, c.hass._rej = set_number, True
_l_rows, _l = run(1500.0, dur=400.0, before=reject_a)
print(f" A's writes rejected: {_l_rows[-1][2]}, {_l_rows[-1][7][-40:]}")
check("⭐ a charging battery that is dropped (stale link) adds no ceiling: "
"B covers the 1500 W house instead of charging for A",
"dropped inverter-a" in _l_rows[-1][7]
and 1490.0 <= cmd(_l_rows[-1])["b"] <= 1530.0
and not charge_claims(_l_rows[-1]))
def modes_store(b_mode=None, b_off=False):
s = rig._Store()
for k in ("phase", "phase_started"):
s.data[unit_key(A, k)] = ("charge" if k == "phase"
else rig.BASE.isoformat())
if b_mode:
s.data[unit_key(B, packmode.MODE_KEY)] = b_mode
if b_off:
s.data[unit_key(B, packmode.USE_KEY)] = False
return s
_m = {h: run(h, dur=600.0, extra={"battery_modes_enabled": True},
store=modes_store(b_off=True)) for h in (800.0, -3000.0)}
print(f" B Off: {[(h, cmd(rows[-1])) for h, (rows, _c) in _m.items()]}")
check("⭐ battery modes on, B Off: A charges at 2000 W over an 800 W house "
"and takes 2500 W of a 3000 W surplus, B stays at 0 W, nothing is "
"refused",
cmd(_m[800.0][0][-1]) == {"a": -2000.0, "b": 0.0}
and cmd(_m[-3000.0][0][-1]) == {"a": -2500.0, "b": 0.0}
and all(refused(rows) == 0 and opposite(rows) == 0
for rows, _c in _m.values()))
# ============================================================================ #
print("\n6. a trade does not outrank the charge")
def sell_feed(c, t):
p3._feed(c, t)
s84.feed(True)(c, t)
_trades = []
_t_rows, _t = run(800.0, dur=5400.0, before=sell_feed,
extra=dict(p3.ARB),
soc=lambda u, t: 60.0)
print(f" modes off, a sell at 75-90 min: last {_t_rows[-1][2]}")
_sells = [r for r in _t_rows if any(c[0] == "arbitrage" and c[3] > 0.0
for c in r[-1][0])]
check("⭐ battery modes off: the site trade asks to SELL while A charges; "
"the charge (P_MAINT_SHAPE) outranks it, so A keeps charging and "
"nothing is refused",
len(_sells) > 50
and all(cmd(r)["a"] < 0.0 and cmd(r)["b"] <= 0.0 for r in _sells)
and refused(_t_rows) == 0)
# ============================================================================ #
print("\n7. a seeded sweep of the ceiling")
_c = run(800.0, dur=30.0)[1]
_rnd = random.Random(84)
_bad = []
for _i in range(20000):
ra, rb = _rnd.choice((2000.0, 2500.0)), _rnd.choice((1000.0, 2500.0))
f = _rnd.uniform(1.0, ra)
cl = [Claim.limit("supervised", P_SAFETY, -ra, ra, actuator=A),
Claim.limit("supervised", P_SAFETY, -rb, rb, actuator=B),
Claim.limit("maintenance", P_MAINT_SHAPE, hi=-f, actuator=A)]
x = _rnd.random()
if x < 0.1:
cl.append(Claim.limit("refused", P_SAFETY, 0.0, actuator=A))
elif x < 0.2:
cl.append(Claim.limit("pack", P_SAFETY, 0.0, actuator=A))
y = _rnd.random()
if y < 0.15:
cl.append(Claim.limit("refused", P_SAFETY, *_rnd.choice(
((0.0, float("inf")), (float("-inf"), 0.0))), actuator=B))
elif y < 0.3:
cl.append(Claim.limit("mode", P_MODE, *_rnd.choice(
((0.0, 0.0), (0.0, rb), (-rb, 0.0))), actuator=B))
cl.append(Claim.set("loop", P_LOOP, _rnd.uniform(-6000, 6000)))
if _rnd.random() < 0.4:
cl.append(Claim.set("arbitrage", P_ARBITRAGE,
_rnd.uniform(-5000, 5000)))
ceil = _c._charge_ceiling(cl)
if ceil is not None:
cl.append(Claim.limit("charge", P_MAINT_SHAPE, hi=ceil))
res = resolve(cl, units=(A, B), live={A, B})
sh = res.shares or {}
why = []
if not res.allocated or res.contradiction:
why.append("refused or contradicted")
if sh.get(A, 0.0) * sh.get(B, 0.0) < 0.0:
why.append("opposite")
a_free = x >= 0.2
if a_free and (ceil is None or sh.get(A, 0.0) > -f + 1.0):
why.append("A does not charge")
if not a_free and ceil is not None:
why.append("ceiling filed for a pack that cannot charge")
if why:
_bad.append((_i, why))
print(f" 20000 claim lists, bad {len(_bad)}: {_bad[:3]}")
check("⭐ 20000 random N=2 claim lists (a charge on A, refusals, a full "
"pack, B's mode, the loop and trades in both directions): never "
"refused, never both ways, A always charges at least its rate, and "
"no ceiling for a pack that cannot charge", not _bad)
# ============================================================================ #
print("\n8. the N=1 and N=2 digests")
# Printed by `python3 tools/safety84_identity.py` in goodwe-addon-checks:1,
# 2026-09-30, 480 cycles each, every claim hashed. N=1 and n2_charge_dear:
# the same on release/1.0 9d5b7d1 (the tool copied in). N=2 in a cheap
# window: new, the charge that 9d5b7d1 refused on every cycle.
# SAFETY-85 re-pinned the three N=2 ones: the loop is bounded while A's
# charge pin stands. n2_charge and _banked change only the loop's reason and
# claim on 2 cycles; n2_charge_dear changes commands on 98 cycles, where A's
# pin lifts (t=1800) and B no longer jumps 1390 W, and after the surplus step.
PINNED = {
"n1_charge":
"0ed25c856456b97acb07a39299a8cb7b1a762c4d0005b88a0823c03057a80be5",
"n1_charge_dear":
"c25340b56060d4581f7204e200b07e7c8618711be2baca3b1667b7ebeb600dc3",
"n1_charge_banked":
"2619acdd06b04e56c000b168d64e1d044346c6ba3f929c94258d6ad8189ec609",
"n2_charge":
"b16d692128f14bcce3d87b142d290e0d6b13724289680e44a05215c9eaa24a7c",
"n2_charge_dear":
"00a197d89a5dd1e8b3ae6a5d1af66329a4dc4850476569c5931c125fa4d9eaee",
"n2_charge_banked":
"a6973faea916e4c88ca1770047f94478346eeb1efa4889b154fad0747f91267f",
}
_now = s84.digests()
for k in PINNED:
print(f" {k} {'same' if _now[k] == PINNED[k] else 'DIFFERS'} "
f"{_now[k][:16]}")
check("⭐ at N=1 a charge in and out of a cheap window and under a banked "
"month peak is byte-identical to 9d5b7d1 (commands, reasons and every "
"claim), and N=2 outside a cheap window matches its SAFETY-85 pin",
all(_now[k] == PINNED[k] for k in ("n1_charge", "n1_charge_dear",
"n1_charge_banked",
"n2_charge_dear")))
check("...and the N=2 cheap-window digests match the pins",
_now["n2_charge"] == PINNED["n2_charge"]
and _now["n2_charge_banked"] == PINNED["n2_charge_banked"])
_n1 = s84.run(1, dur=600.0)[0]
check("...and no N=1 cycle files a `charge` claim",
not any(charge_claims(r) for r in _n1))
print()
if fails:
print(f"{len(fails)} FAILED: " + ", ".join(fails))
sys.exit(1)
print("all SAFETY-84 checks passed")