glenn schrooyenandClaude Opus 5.5
ba31e7c8b1
During a HOLD the holding battery gives what keeps the quarter at the peak aim. N=1: the peak floor is refiled at P_MAINTENANCE, as for a drain (SAFETY-83). N>=2: the hold's pin is raised to the floor minus what the other live batteries may place, so they give first. A site stop still decides alone. The hold ends on its clock even if the level dipped. Below the aim nothing changes (rows identical). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
308 lines
14 KiB
Python
308 lines
14 KiB
Python
"""Runnable check for SAFETY-83: the peak guard outranks the drain at N=1.
|
|
`python3 test_safety83.py`
|
|
|
|
At one inverter the monthly drain is the site SET at P_MAINTENANCE, above the
|
|
peak floor, so the house imported what the drain left and quarters went over
|
|
the peak limit. Now a peak floor above the drain rate is refiled at
|
|
P_MAINTENANCE and wins. Drives the REAL Controller (tools/opt11_identity.py's
|
|
harness on a P1 meter). DECISIONS/SAFETY-83-peak-over-drain.md has the record.
|
|
|
|
1. A house of 5000, 6000 and 7000 W during a drain, limits 2500 and 3500,
|
|
margins 0 and 50: every quarter banks at or under the aim, the drain keeps
|
|
running, no 'Load too high' alarm, and without the fix the same runs go over.
|
|
Under a month peak banked above the limit the aim is that peak - margin.
|
|
2. The drain never makes the banked quarter worse than no drain, below and
|
|
above the drain rate; below it the rest is still exported.
|
|
3. The drain still ends, and the order holds: a lost reading parks the
|
|
battery, the rating caps the floor without a clash.
|
|
4. Only cycles where the guard wanted more than the drain change, and N=2 is
|
|
untouched.
|
|
|
|
Mutations (MUTATION.md, purged copies) are listed in the pull request.
|
|
"""
|
|
|
|
import functools
|
|
import logging
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import types
|
|
|
|
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 safety78_identity as s78 # noqa: E402
|
|
import safety79_identity as s79 # noqa: E402
|
|
from app import tariff # noqa: E402
|
|
from app.arbiter import (Claim, P_MAINTENANCE, P_PEAK, P_SAFETY, # noqa
|
|
resolve)
|
|
from app.main import Controller # noqa: E402
|
|
from app.maintenance import CHARGE, DRAIN, MaintResult # noqa: E402
|
|
|
|
fails = []
|
|
TOL = 1.0 # watts over the aim a banked quarter may show
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
def run(house, cap, margin, drain_w, *, maint=True, dur=2700.0, soc=None,
|
|
before=None, store=None, month_peak=None, **extra):
|
|
"""N=1 on a P1 meter; returns (rows, controller, banked quarters).
|
|
`month_peak` banks that quarter earlier in the month."""
|
|
opts = dict(s79.P1, peak_cap_w=cap, peak_margin_w=margin, **extra)
|
|
if maint:
|
|
opts.update(maintenance_enabled=True, maintenance_start_hour=11,
|
|
maintenance_discharge_w=drain_w)
|
|
db = os.path.join(tempfile.mkdtemp(prefix="s83-"), "tariff.db")
|
|
if month_peak is not None:
|
|
tariff.PeakStore(db)._db.execute(
|
|
"INSERT INTO monthly_peak VALUES (?, ?, ?, NULL)",
|
|
(tariff.month_key(rig.BASE), month_peak,
|
|
"2026-09-02T10:00:00+00:00")).connection.commit()
|
|
real = rig.Controller
|
|
rig.Controller = functools.partial(real, peak_db_path=db)
|
|
try:
|
|
rows, c = rig.run(1, extra=opts, dur=dur,
|
|
load=house if callable(house) else lambda t: house,
|
|
soc=soc or (lambda u, t: 60.0),
|
|
cut=lambda u, t: False, before=before or s79._feed,
|
|
store=store)
|
|
finally:
|
|
rig.Controller = real
|
|
return rows, c, {b.start.strftime("%H:%M"): b.offtake_avg_w
|
|
for b in c.p1.blocks}
|
|
|
|
|
|
def without_fix(fn):
|
|
real = Controller._peak_over_drain
|
|
Controller._peak_over_drain = lambda self, claims, results: None
|
|
try:
|
|
return fn()
|
|
finally:
|
|
Controller._peak_over_drain = real
|
|
|
|
|
|
def cmd(r):
|
|
return dict(r[2])["a"]
|
|
|
|
|
|
def alarmed(c):
|
|
"""Did SAFETY-81's 'Load too high' alarm fire on this run?"""
|
|
return any("PEAK_GUARD_SATURATED" in e for e in c.events)
|
|
|
|
|
|
# ============================================================================ #
|
|
print("1. banked quarters during a drain, N=1")
|
|
CASES = [(cap, drain, m, h) for cap, drain in ((2500, 2000), (3500, 2500))
|
|
for m in (0, 50) for h in (5000.0, 6000.0, 7000.0)]
|
|
RES, OLD = {}, {}
|
|
for k in CASES:
|
|
cap, drain, m, h = k
|
|
RES[k] = run(h, cap, m, drain)
|
|
OLD[k] = without_fix(lambda: run(h, cap, m, drain))
|
|
print(f" limit {cap} drain {drain} margin {m:>2} house {h:.0f}: "
|
|
f"banked {[round(v) for v in RES[k][2].values()]} W "
|
|
f"(before the fix {[round(v) for v in OLD[k][2].values()]} W)")
|
|
check("⭐ every quarter banks at or under the aim (limit - margin) plus 1 W, "
|
|
"for a 5000, 6000 and 7000 W house, limits 2500 and 3500, margins 0 "
|
|
"and 50",
|
|
all(len(b) >= 2 and max(b.values()) <= k[0] - k[2] + TOL
|
|
for k, (_r, _c, b) in RES.items()))
|
|
check("⭐ the drain keeps running: the phase is drain on every run and the "
|
|
"battery gives at least the drain rate on every cycle",
|
|
all(c.maints[0].phase == DRAIN and all(cmd(r) >= k[1] for r in rows)
|
|
for k, (rows, c, _b) in RES.items()))
|
|
check("...and where the house is over drain + aim, the battery gives more "
|
|
"than the drain rate, bound by the peak floor",
|
|
all(any(cmd(r) > k[1] and "limited by peak" in r[6] for r in rows)
|
|
for k, (rows, _c, _b) in RES.items() if k[3] - k[1] > k[0] - k[2]))
|
|
check("non-vacuous: without the fix the same runs bank over the limit "
|
|
"wherever the house is above drain + aim (the reported defect: 3000 W "
|
|
"at a 5000 W house against 2500 W)",
|
|
all(max(b.values()) > k[0] + 100.0 for k, (_r, _c, b) in OLD.items()
|
|
if k[3] - k[1] > k[0])
|
|
and round(OLD[(2500, 2000, 50, 5000.0)][2]["11:15"]) == 3000)
|
|
# A hob cycling: one 5 s spike to 11000 W every 150 s on a 6000 W house. Each
|
|
# spike saturates the guard for one cycle, shorter than ALARM_HOLD_S.
|
|
_sat = []
|
|
|
|
|
|
def _spiky_feed(c, t):
|
|
s79._feed(c, t)
|
|
_sat.append(bool(c.peak_guard is not None and c.peak_guard.saturated))
|
|
|
|
|
|
_k_rows, _k, _kb = run(lambda t: 11000.0 if t % 150 < 5 else 6000.0,
|
|
3500, 50, 2500, before=_spiky_feed)
|
|
check("⭐ a drain the guard raises gives no 'Load too high' alarm (SAFETY-81), "
|
|
"steady or with 5 s spikes that saturate the guard",
|
|
not any(alarmed(c) for _r, c, _b in RES.values())
|
|
and any(_sat) and not alarmed(_k))
|
|
|
|
# Owner (SAFETY-79): under a month peak banked above the limit, the aim is that
|
|
# peak - margin, during a drain too. Owner's settings, peak banked at 4500 W.
|
|
BANK = {h: (run(h, 3500, 50, 2500, month_peak=4500.0),
|
|
without_fix(lambda: run(h, 3500, 50, 2500, month_peak=4500.0)))
|
|
for h in (6500.0, 7500.0)}
|
|
for h, (new, old) in BANK.items():
|
|
print(f" month peak 4500, house {h:.0f}: banked "
|
|
f"{[round(v) for v in new[2].values()]} W (before the fix "
|
|
f"{[round(v) for v in old[2].values()]} W)")
|
|
check("⭐ under a month peak of 4500 W: a 7500 W house banks at 4450 W "
|
|
"(peak - margin, not the 3450 W cap aim), was 5000 W; a 6500 W house "
|
|
"stays under the drain alone at 4000 W",
|
|
all(4430.0 <= v <= 4450.0 + TOL for v in BANK[7500.0][0][2].values())
|
|
and max(BANK[7500.0][1][2].values()) > 4900.0
|
|
and all(round(v) in (4000, 4014)
|
|
for v in BANK[6500.0][0][2].values()))
|
|
|
|
# ============================================================================ #
|
|
print("\n2. the drain never makes the banked quarter worse, N=1")
|
|
CAP, DRAIN_W, MARGIN = 3500, 2500, 50 # the owner's settings
|
|
AIM = CAP - MARGIN
|
|
WORSE = {}
|
|
for h in (1200.0, 3000.0, 5000.0, 6500.0, 7000.0):
|
|
_rows, _c, b = run(h, CAP, MARGIN, DRAIN_W)
|
|
_r0, _c0, b0 = run(h, CAP, MARGIN, DRAIN_W, maint=False)
|
|
WORSE[h] = (b, b0, _rows)
|
|
print(f" house {h:.0f}: with the drain {[round(v) for v in b.values()]}"
|
|
f" W, without {[round(v) for v in b0.values()]} W")
|
|
check("⭐ below and above the drain rate, every quarter with the drain banks "
|
|
"no more than the larger of the aim and the same quarter without it",
|
|
all(b[q] <= max(AIM, b0[q]) + TOL for b, b0, _r in WORSE.values()
|
|
for q in b))
|
|
_r12 = WORSE[1200.0][2]
|
|
check("below the drain rate the battery gives the drain rate and exports the "
|
|
"rest (owner, 2026-09-29): 2500 W against a 1200 W house",
|
|
all(cmd(r) == DRAIN_W for r in _r12[1:]))
|
|
|
|
# ============================================================================ #
|
|
print("\n3. the drain ends, and the order holds")
|
|
_e_rows, _e, _eb = run(6500.0, CAP, MARGIN, DRAIN_W, dur=1200.0,
|
|
soc=lambda u, t: max(5.0, 60.0 - t / 20.0))
|
|
check("⭐ with the guard binding, the drain still reaches its floor and moves "
|
|
"to the charge phase, with no abort",
|
|
_e.maints[0].phase == CHARGE
|
|
and not any("abort" in e.lower() for e in _e.events))
|
|
|
|
|
|
def blind_at(t0):
|
|
"""The battery power reading goes missing at `t0`: a safety SET."""
|
|
def before(c, t):
|
|
s79._feed(c, t)
|
|
if t >= t0 and not getattr(c.hass, "_blind", False):
|
|
real = c.hass.number
|
|
|
|
async def number(e, invert=False):
|
|
return None if e == "sensor.batt_a" else await real(e, invert)
|
|
c.hass.number, c.hass._blind = number, True
|
|
return before
|
|
|
|
|
|
_s_rows, _s, _sb = run(6500.0, CAP, MARGIN, DRAIN_W, dur=600.0,
|
|
before=blind_at(300.0))
|
|
print(f" reading lost at 300 s: last {cmd(_s_rows[-1]):+.0f} W, "
|
|
f"{_s_rows[-1][6]}")
|
|
check("⭐ a safety SET outranks the refiled peak floor: before the reading "
|
|
"goes the battery gives above the drain rate, after it it is at 0 W",
|
|
any(cmd(r) > DRAIN_W for r in _s_rows if r[0] < 300.0)
|
|
and all(cmd(r) == 0.0 for r in _s_rows if r[0] >= 400.0)
|
|
and _s_rows[-1][6].startswith("safety -> 0 W"))
|
|
_m_rows, _m, _mb = run(7000.0, 2500, 50, 2000, dur=900.0, max_w=3000)
|
|
print(f" max_w 3000, house 7000: last {cmd(_m_rows[-1]):+.0f} W, "
|
|
f"{_m_rows[-1][6]}")
|
|
check("⭐ the rating caps the floor without a clash: at max_w 3000 the "
|
|
"battery gives 3000 W, never 0 W, and nothing contradicts",
|
|
all(cmd(r) >= 2000.0 for r in _m_rows)
|
|
and cmd(_m_rows[-1]) == 3000.0
|
|
and not any("CONTRADICTION" in (r[6] or "") for r in _m_rows))
|
|
check("...and there, with the battery at its rating and the house still over, "
|
|
"the 'Load too high' alarm does fire", alarmed(_m))
|
|
|
|
# ============================================================================ #
|
|
print("\n4. only the cycles where the guard wanted more change")
|
|
|
|
|
|
def promoted(row):
|
|
return [c for c in row[-1][0] if c[0] == "peak"]
|
|
|
|
|
|
_id = {k: s78.run(**v)[0] for k, v in s78.SCENARIOS.items()}
|
|
_bad = []
|
|
for k, rows in _id.items():
|
|
for r in rows:
|
|
claims, guard = r[-1][0], r[-1][1]
|
|
rate = [c[3] for c in claims if c[0] == "maintenance" and c[2] == "set"]
|
|
pk = promoted(r)
|
|
if k.startswith("n2"):
|
|
ok = all(c[1] == P_PEAK for c in pk)
|
|
elif s78.wanted_more(r):
|
|
ok = ([c[1] for c in pk] == [P_MAINTENANCE]
|
|
and r[1] == guard[1] and "limited by peak" in r[6])
|
|
else:
|
|
ok = (all(c[1] == P_PEAK for c in pk)
|
|
and (not rate or r[1] == rate[0]))
|
|
if not ok:
|
|
_bad.append((k, r[0], r[1], r[6], pk))
|
|
_bad1 = [b for b in _bad if b[0].startswith("n1")]
|
|
_bad2 = [b for b in _bad if b[0].startswith("n2")]
|
|
_more = {k: sum(map(s78.wanted_more, rows)) for k, rows in _id.items()}
|
|
print(f" cycles with the guard above the drain: {_more}; bad {_bad[:3]}")
|
|
check("⭐ at N=1, on every cycle where the guard asks for more than the drain "
|
|
"the peak claim is refiled at P_MAINTENANCE and the target is its floor; "
|
|
"on every other cycle the peak claim stays at P_PEAK and the target is "
|
|
"the drain rate",
|
|
not _bad1 and all(_more[k] > 50 for k in _id if k.startswith("n1")))
|
|
check("⭐ at N=2 the peak claim is never refiled (the SAFETY-78 floor does "
|
|
"that job), and test_safety78.py part 5 holds N=2 to the digests of "
|
|
"release/1.0 35cdffe", not _bad2)
|
|
|
|
|
|
# The rule itself, at the boundary: a floor AT the drain rate changes nothing.
|
|
_one = types.SimpleNamespace(inverters=[None])
|
|
_edge = {}
|
|
for lo in (1999.5, 2000.0, 2000.5):
|
|
cl = [Claim.set("maintenance", P_MAINTENANCE, 2000.0, DRAIN),
|
|
Claim.limit("peak", P_PEAK, lo=lo, reason="quarter peak")]
|
|
Controller._peak_over_drain(_one, cl, [MaintResult(DRAIN, True, 2000.0,
|
|
False, [])])
|
|
_edge[lo] = cl[1].priority
|
|
check("at the boundary: a floor at or below the 2000 W drain stays at P_PEAK, "
|
|
"a floor 0.5 W above it is refiled at P_MAINTENANCE",
|
|
_edge == {1999.5: P_PEAK, 2000.0: P_PEAK, 2000.5: P_MAINTENANCE})
|
|
|
|
_stop = [Claim.set("maintenance", P_MAINTENANCE, 2000.0, DRAIN),
|
|
Claim.limit("peak", P_PEAK, lo=4000.0, reason="quarter peak"),
|
|
Claim.set("safety", P_SAFETY, 0.0, "inputs missing")]
|
|
Controller._peak_over_drain(_one, _stop, [MaintResult(DRAIN, True, 2000.0,
|
|
False, [])])
|
|
check("⭐ and a refiled floor stays under a safety SET: the site resolves "
|
|
"to 0 W", _stop[1].priority == P_MAINTENANCE
|
|
and resolve(_stop).target_w == 0.0)
|
|
|
|
|
|
def store_hold():
|
|
st = rig._Store()
|
|
st.data["phase"], st.data["phase_started"] = "hold", rig.BASE.isoformat()
|
|
return st
|
|
|
|
|
|
_h_rows, _h, _hb = run(6000.0, 2500, 50, 2000, dur=600.0, store=store_hold())
|
|
check("a maintenance HOLD at N=1 now gives way to the guard too (SAFETY-86, "
|
|
"test_safety86.py): the battery gives with the house over the limit",
|
|
_h.maints[0].phase == "hold" and _h.peak_guard.engaged
|
|
and cmd(_h_rows[-1]) > 0.0 and "limited by peak" in _h_rows[-1][6])
|
|
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED: {', '.join(fails)}")
|
|
sys.exit(1)
|
|
print("all checks passed")
|