glenn schrooyenandClaude
e27e62ff72
The rolling budget now holds without a monotonic clock (it never refills). Quarter-edge tests use seeds 1-3 at 2, 5, 10 and 15 s (no hand-picked seeds). Re-pin comments point at part 6. DECISIONS lists the limits the owner accepted with N=8 on 2026-10-20. Co-Authored-By: Claude <noreply@anthropic.com>
515 lines
22 KiB
Python
515 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.
|
|
# SAFETY-119 re-pinned all six: only the log. OUTPUT_CLAMP rows from the
|
|
# loop's bound just after the guard hands back are gone; every command and
|
|
# claim is the same (test_safety119.py part 6).
|
|
PINNED = {
|
|
"n1_charge":
|
|
"19781f90b9751b10994dedeaafb38e42d48ad2551640792e601336b7f9b5bb4f",
|
|
"n1_charge_dear":
|
|
"eeeb0fa19a6fdc04d44a0838bb2bb911f506a936b0f93d8a6f4d87159053ec74",
|
|
"n1_charge_banked":
|
|
"a51f125a8c1c98fdbb009fce72b6781d20d384bdf0f5ce7bbfe367d86e5477db",
|
|
"n2_charge":
|
|
"e37e98390baa5dad9128991fabf8b170894395fe026d8a33c8a51de7ed0789cc",
|
|
"n2_charge_dear":
|
|
"7727372de423f62d77b4426a7226eae31a30f3bb89a3cf70272cc7c1823d7263",
|
|
"n2_charge_banked":
|
|
"c77e26006c026b953e9c3dcc9a02f3993589f885cb52c7cbceac6bd59b600551",
|
|
}
|
|
_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; the log as re-pinned by SAFETY-119), 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")
|