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
This commit is contained in:
glenn schrooyen
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
935878193b
commit
4b81387908
@@ -0,0 +1,303 @@
|
||||
"""Runnable check for SAFETY-85: no false 'not keeping up' while another owner
|
||||
sets the battery. `python3 test_safety85.py`
|
||||
|
||||
The loop's integrator bound fired every ~2.5 min during maintenance and under
|
||||
the peak guard, though the battery gave what it was told: the loop's own ask,
|
||||
re-seeded and pushed by a grid error it did not own, crossed its rail. Now the
|
||||
loop's clamp counts only on a cycle the loop decided, a battery another owner
|
||||
set alarms only when it gives less than told, and at N>=2 the loop is bounded
|
||||
while maintenance holds a battery. Drives the REAL Controller
|
||||
(tools/safety84_identity.py's harness). DECISIONS/SAFETY-85-no-false-not-keeping-up.md.
|
||||
|
||||
1. N=1 and N=2: a drain, a charge and a HOLD with the house above what the
|
||||
battery is set to, an hour each: no warning; without the fix they warn.
|
||||
2. A battery that gives less than told during a drain does alarm (N=1, N=2);
|
||||
a two-cycle dip does not.
|
||||
3. Control coming back from maintenance does not jump (N=2 hold end, N=2
|
||||
charge pin lifting, N=1 hold end).
|
||||
4. The loop's own alarm stays (house over the rating); a safety limit at N=2
|
||||
(a battery at its cut-out) does not warn.
|
||||
5. The status row counts episodes, not cycles.
|
||||
6. The rules at their boundaries.
|
||||
|
||||
Mutations (MUTATION.md, purged copies) are listed in the pull request.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
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 safety79_identity as s79 # noqa: E402
|
||||
import safety84_identity as s84 # noqa: E402
|
||||
import app.main as app_main # noqa: E402
|
||||
from app import clamp, messages # noqa: E402
|
||||
from app.main import Controller # noqa: E402
|
||||
from app.store import unit_key # noqa: E402
|
||||
|
||||
fails = []
|
||||
A = "inverter-a"
|
||||
HOUR = 3600.0
|
||||
SLEW = 400.0 # the rig's slew_w
|
||||
OPT0 = dict(s84.OPT)
|
||||
|
||||
|
||||
def check(name, cond):
|
||||
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
||||
if not cond:
|
||||
fails.append(name)
|
||||
|
||||
|
||||
async def _call(self, *a):
|
||||
"""The rig's Home Assistant, recording each notification id it is sent."""
|
||||
if len(a) >= 3 and a[1] == "create" and isinstance(a[2], dict):
|
||||
self.__dict__.setdefault("notes", []).append(
|
||||
a[2].get("notification_id"))
|
||||
return True
|
||||
|
||||
|
||||
rig._Hass.call = _call
|
||||
|
||||
|
||||
def store(n, phase):
|
||||
"""Battery A in `phase` from the first cycle ('drain': due at 11:00;
|
||||
'idle': only with `maint=False`)."""
|
||||
if phase == "drain":
|
||||
return rig._Store()
|
||||
if phase == "charge":
|
||||
return s79.store_charging(n)
|
||||
s = rig._Store()
|
||||
if phase == "hold":
|
||||
for k, v in (("phase", "hold"), ("phase_started",
|
||||
rig.BASE.isoformat())):
|
||||
s.data[k if n == 1 else unit_key(A, k)] = v
|
||||
return s
|
||||
|
||||
|
||||
def run(n, phase, house, cap, *, dur=HOUR, soc=None, cut=None, extra=None,
|
||||
maint=True):
|
||||
"""Rows and controller; `cap` is the peak limit (20000: never binds)."""
|
||||
s84.OPT = dict(OPT0, peak_cap_w=cap, maintenance_enabled=maint)
|
||||
if soc is None:
|
||||
def soc(u, t):
|
||||
return 99.0 if phase == "hold" and u == "a" else 60.0
|
||||
try:
|
||||
return s84.run(n, load=lambda t: house, cheap=(phase == "charge"),
|
||||
store=store(n, phase), soc=soc, dur=dur, cut=cut,
|
||||
extra=dict({"maintenance_hold_min": 600,
|
||||
"maintenance_discharge_w": 2000},
|
||||
**(extra or {})))
|
||||
finally:
|
||||
s84.OPT = OPT0
|
||||
|
||||
|
||||
def warned(c):
|
||||
"""'Battery is not keeping up': notifications, cycles, episodes."""
|
||||
return (getattr(c.hass, "notes", []).count(messages.ID_LOOP_BOUND),
|
||||
c.clamp_wd.bound, c.clamp_wd.bound_episodes)
|
||||
|
||||
|
||||
def patched(obj, name, value):
|
||||
"""Run `fn` with `obj.name` replaced (the 'without the fix' runs)."""
|
||||
def wrap(fn):
|
||||
real = getattr(obj, name)
|
||||
setattr(obj, name, value)
|
||||
try:
|
||||
return fn()
|
||||
finally:
|
||||
setattr(obj, name, real)
|
||||
return wrap
|
||||
|
||||
|
||||
no_gate = patched(Controller, "_loop_alarm_input", lambda self, r: None)
|
||||
no_bound = patched(Controller, "loop_bound", lambda self, claims: None)
|
||||
|
||||
|
||||
def cmd(r):
|
||||
return dict(r[2])
|
||||
|
||||
|
||||
def steps(rows, lo, hi):
|
||||
"""The largest change in the total command between cycles in [lo, hi]."""
|
||||
tot = [(r[0], sum(cmd(r).values())) for r in rows]
|
||||
return max(abs(b - a) for (t0, a), (t1, b) in zip(tot, tot[1:])
|
||||
if lo <= t1 <= hi)
|
||||
|
||||
|
||||
# ============================================================================ #
|
||||
print("1. the house above what the battery is set to, an hour each")
|
||||
CASES = [
|
||||
# (n, phase, house, cap): the peak guard raises the drain / the HOLD at a
|
||||
# 3500 W limit; the rest are the phase alone under a house that pushes the
|
||||
# loop's re-seeded ask past its rail.
|
||||
(1, "drain", 6500.0, 3500), (1, "drain", 7500.0, 20000),
|
||||
(1, "charge", 10000.0, 20000),
|
||||
(1, "hold", 6500.0, 3500), (1, "hold", 9000.0, 20000),
|
||||
(2, "drain", 6500.0, 3500), (2, "charge", 10000.0, 20000),
|
||||
(2, "hold", 4500.0, 20000), (2, "hold", 6500.0, 3500),
|
||||
]
|
||||
NEW, OLD = {}, {}
|
||||
for k in CASES:
|
||||
NEW[k] = run(*k)
|
||||
OLD[k] = no_gate(lambda: no_bound(lambda: run(*k)))
|
||||
print(f" N={k[0]} {k[1]:6} house {k[2]:5.0f} limit {k[3]:5}: phases "
|
||||
f"{sorted({p for r in NEW[k][0] for p in r[-1][3]})}, warnings "
|
||||
f"(notes, cycles, episodes) {warned(NEW[k][1])}, without the fix "
|
||||
f"{warned(OLD[k][1])}")
|
||||
check("⭐ no 'Battery is not keeping up' notification, cycle or episode in any "
|
||||
"of the nine hours, at N=1 and N=2",
|
||||
all(warned(c) == (0, 0, 0) for _r, c in NEW.values()))
|
||||
check("...the phase held all hour (the battery was set by maintenance, or by "
|
||||
"the guard over it, throughout)",
|
||||
all(all(r[-1][3][0] == k[1] for r in NEW[k][0]) for k in CASES))
|
||||
check("non-vacuous: without the fix every one of the nine warns, the "
|
||||
"notification re-created on the re-assert cadence",
|
||||
all(warned(c)[0] >= 1 and warned(c)[1] > 0 for _r, c in OLD.values())
|
||||
and sum(warned(c)[0] for _r, c in OLD.values()) >= 9 * 5)
|
||||
|
||||
# ============================================================================ #
|
||||
print("\n2. a battery that really gives less than told still alarms")
|
||||
|
||||
|
||||
def gone(t0, t1):
|
||||
"""Battery A gives nothing from t0 to t1."""
|
||||
return lambda u, t: u == "a" and t0 <= t < t1
|
||||
|
||||
|
||||
SHORT = {n: run(n, "drain", 1000.0, 20000, dur=1200.0, cut=gone(600, 1200))
|
||||
for n in (1, 2)}
|
||||
for n, (rows, c) in SHORT.items():
|
||||
print(f" N={n}: {warned(c)}, log "
|
||||
f"{[e for e in c.events if 'OUTPUT_CLAMP' in e][:1]}")
|
||||
_r1, _c1 = SHORT[1]
|
||||
check("⭐ N=1, a drain of 2000 W that gives nothing: it warns, and the log "
|
||||
"says what was asked and what came",
|
||||
warned(_c1)[0] >= 1 and any(
|
||||
"the battery gives -4 W of the +2000 W asked" in e
|
||||
for e in _c1.events))
|
||||
_r2, _c2 = SHORT[2]
|
||||
check("⭐ N=2, the same on battery A while B follows the house: it warns and "
|
||||
"names inverter-a",
|
||||
warned(_c2)[0] >= 1 and any(
|
||||
"inverter-a gives -4 W of the +2000 W asked" in e
|
||||
for e in _c2.events))
|
||||
_hb = run(2, "hold", 4500.0, 20000, dur=1200.0,
|
||||
cut=lambda u, t: u == "b" and t >= 600)
|
||||
print(f" N=2 HOLD on A, B gives nothing: {warned(_hb[1])}")
|
||||
check("⭐ N=2, A in its HOLD and B, which the loop drives, gives nothing: the "
|
||||
"loop stands at what B may give, and it warns naming inverter-b",
|
||||
warned(_hb[1])[0] >= 1 and any(
|
||||
"inverter-b gives -4 W of the +2500 W asked" in e
|
||||
for e in _hb[1].events))
|
||||
check("...it starts no earlier than the saturation cycles allow (3 x 5 s)",
|
||||
all(e.split(" ", 1)[0] >= "11:10:10" for c in (_c1, _c2)
|
||||
for e in c.events if "OUTPUT_CLAMP" in e))
|
||||
_quiet = patched(app_main, "delivery_short", lambda *a: False)(
|
||||
lambda: run(1, "drain", 1000.0, 20000, dur=1200.0, cut=gone(600, 1200)))
|
||||
check("non-vacuous: without the delivery rule the N=1 drain that gives "
|
||||
"nothing is silent (the loop's clamp never reaches its rail here)",
|
||||
warned(_quiet[1]) == (0, 0, 0))
|
||||
_dip = run(1, "drain", 1000.0, 20000, dur=900.0, cut=gone(600, 610))
|
||||
check("a dip of two cycles is not an alarm", warned(_dip[1]) == (0, 0, 0))
|
||||
|
||||
# ============================================================================ #
|
||||
print("\n3. control coming back from maintenance does not jump")
|
||||
END = {"maintenance_hold_min": 10}
|
||||
_h2 = run(2, "hold", 4500.0, 20000, dur=900.0, extra=END)
|
||||
_h2_old = no_bound(lambda: run(2, "hold", 4500.0, 20000, dur=900.0,
|
||||
extra=END))
|
||||
_h1 = run(1, "hold", 4500.0, 20000, dur=900.0, extra=END)
|
||||
s84.OPT = dict(OPT0)
|
||||
_cd = s84.run(2, cheap=False)
|
||||
_cd_old = no_bound(lambda: s84.run(2, cheap=False))
|
||||
print(f" N=2 hold end: largest step {steps(_h2[0], 560, 700):.0f} W "
|
||||
f"(without the bound {steps(_h2_old[0], 560, 700):.0f} W); N=1 "
|
||||
f"{steps(_h1[0], 560, 700):.0f} W; N=2 charge pin lifts at 1800 s: "
|
||||
f"{steps(_cd[0], 1780, 1840):.0f} W (without "
|
||||
f"{steps(_cd_old[0], 1780, 1840):.0f} W)")
|
||||
check("⭐ N=2, the HOLD ends under a 4500 W house: the batteries move at most "
|
||||
"one slew step a cycle",
|
||||
_h2[1].maints[0].phase != "hold" and steps(_h2[0], 560, 700) <= SLEW)
|
||||
check("⭐ N=2, A's charge pin lifts at a quarter's start: at most one slew "
|
||||
"step a cycle", steps(_cd[0], 1780, 1840) <= SLEW)
|
||||
check("...and N=1 already moved one slew step at a time (re-seeded each "
|
||||
"cycle), unchanged", steps(_h1[0], 560, 700) <= SLEW)
|
||||
check("non-vacuous: without the bound the N=2 loop stands wound at its rail "
|
||||
"and both jump (2500 W at the hold's end)",
|
||||
steps(_h2_old[0], 560, 700) >= 2000.0
|
||||
and steps(_cd_old[0], 1780, 1840) >= 1000.0)
|
||||
|
||||
# ============================================================================ #
|
||||
print("\n4. what still warns, and what a safety limit does")
|
||||
_rail = run(1, "drain", 6500.0, 20000, dur=600.0, maint=False)
|
||||
print(f" N=1 no maintenance, house 6500 over a 5000 W rating: "
|
||||
f"{warned(_rail[1])}")
|
||||
check("⭐ the loop in charge at its rail (house over the rating) still warns, "
|
||||
"as SAFETY-40 decided", warned(_rail[1])[0] >= 1)
|
||||
|
||||
|
||||
def at_cutout(u, t):
|
||||
return 9.0 if u == "a" else 60.0
|
||||
|
||||
|
||||
_cut = run(2, "idle", 6000.0, 20000, dur=900.0, soc=at_cutout, maint=False)
|
||||
_cut_old = no_gate(lambda: run(2, "idle", 6000.0, 20000, dur=900.0,
|
||||
soc=at_cutout, maint=False))
|
||||
print(f" N=2, A at its cut-out, house 6000: {warned(_cut[1])}, without "
|
||||
f"the fix {warned(_cut_old[1])}")
|
||||
check("⭐ N=2, A at its cut-out level (a safety limit sets it to 0 W) and B "
|
||||
"at its rating: no warning, each battery gives what it is told",
|
||||
warned(_cut[1]) == (0, 0, 0) and cmd(_cut[0][-1]) == {"a": 0.0,
|
||||
"b": 2500.0})
|
||||
check("non-vacuous: without the fix it warns",
|
||||
warned(_cut_old[1])[0] >= 1)
|
||||
|
||||
# ============================================================================ #
|
||||
print("\n5. the status row counts episodes")
|
||||
_wd = clamp.ClampWatchdog()
|
||||
_t = datetime(2026, 9, 20, 11, 0, tzinfo=timezone.utc)
|
||||
for ep in range(3):
|
||||
for i in range(10):
|
||||
_wd.note(True, "x", _t, bug=False)
|
||||
_t += timedelta(seconds=5)
|
||||
for i in range(clamp.CLEAR_CYCLES):
|
||||
_wd.note(False, "", _t)
|
||||
_t += timedelta(seconds=5)
|
||||
_wd.note(True, "y", _t, bug=True)
|
||||
check("three bound episodes of ten cycles and one over-rating: 3 episodes, "
|
||||
"30 cycles", (_wd.bound_episodes, _wd.bound, _wd.episodes) == (3, 30, 4))
|
||||
_rows = [x for x in _c1.checks() if x.get("id") == "not_keeping_up"]
|
||||
check("⭐ the page row reads the episodes, not the cycles "
|
||||
f"({_c1.clamp_wd.bound_episodes} vs {_c1.clamp_wd.bound})",
|
||||
_rows and _c1.clamp_wd.bound > _c1.clamp_wd.bound_episodes >= 1
|
||||
and _rows[0]["text"].startswith(messages.T_LOOP_BOUND + ": "
|
||||
+ messages.times(
|
||||
_c1.clamp_wd.bound_episodes)))
|
||||
|
||||
# ============================================================================ #
|
||||
print("\n6. the rules at their boundaries")
|
||||
DS = [((2000, 1499, 500), True), ((2000, 1500, 500), False),
|
||||
((2000, 2600, 500), False), ((-2000, -1499, 500), True),
|
||||
((-2000, -2600, 500), False), ((0, -900, 500), False),
|
||||
((0, 900, 500), False),
|
||||
((2000, -4, 500), True), ((-2000, 600, 500), True)]
|
||||
check("delivery_short: more than the margin short in the asked direction; "
|
||||
"more than asked, or 0 W asked, is never short",
|
||||
all(clamp.delivery_short(*a) is want for a, want in DS))
|
||||
check("the note names each battery, asked before given",
|
||||
clamp.shortfall_note([("inverter-a", 2000.0, -4.0)])
|
||||
== "OUTPUT_CLAMP: inverter-a gives -4 W of the +2000 W asked"
|
||||
and clamp.shortfall_note([(None, -2000.0, 0.0)])
|
||||
== "OUTPUT_CLAMP: the battery gives +0 W of the -2000 W asked")
|
||||
|
||||
print()
|
||||
if fails:
|
||||
print(f"{len(fails)} FAILED: " + ", ".join(fails))
|
||||
sys.exit(1)
|
||||
print("all SAFETY-85 checks passed")
|
||||
Reference in New Issue
Block a user