adminandClaude Opus 5.5
7d06aa674d
Owner decision 2026-10-04 (review R1). A wrong idle reading more than about 100 W off hid every refusal on that battery. Clamped under the band, the band keeps 0 W +-25 W, so a battery reading about 0 W while refusing is caught whatever was learned or restored. Samples are still taken to 250 W. Beyond about +-170 W of idle a battery behaves as in 0.14.0. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
442 lines
18 KiB
Python
442 lines
18 KiB
Python
"""Runnable check for SAFETY-97: "the battery gives nothing" is graded against
|
|
each battery's own idle reading. `python3 test_safety97.py`
|
|
|
|
B's power reading shows -95 to -118 W at idle (standby draw; live
|
|
2026-10-03). The refusal evidence (SAFETY-52/73, shared since SAFETY-94)
|
|
needed the reading under 100 W in magnitude, so a refusing B never latched,
|
|
at N=2 ever, and at N=1 since SAFETY-94 (review F1).
|
|
|
|
Now each unit learns its idle reading (app/idle.py) and the band, the
|
|
refusal's end and the stale-path refusal are centred on it.
|
|
DECISIONS/SAFETY-97-idle-offset.md.
|
|
|
|
1. A refusing B is caught at N=1 and N=2, 1/2/5 s meters; A unchanged.
|
|
2. A healthy B with its offset is never refused or flagged (SAFETY-94's
|
|
scenarios), learned or restored.
|
|
3. Learning: not while refusing or asked, bounded, restored, follows drift.
|
|
4. The rules at their edges.
|
|
|
|
Mutations (MUTATION.md, purged copies) are listed in the pull request.
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
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.CRITICAL)
|
|
|
|
import safety94_rig as rig # noqa: E402
|
|
import app.idle as idle # noqa: E402
|
|
import app.main as app_main # noqa: E402
|
|
from app import control # 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)
|
|
|
|
|
|
def without_fix(fn):
|
|
"""Run `fn` with every idle reading at 0 W (the rule before SAFETY-97)."""
|
|
real = idle.IdleReading.w
|
|
idle.IdleReading.w = property(lambda self: 0.0)
|
|
try:
|
|
return fn()
|
|
finally:
|
|
idle.IdleReading.w = real
|
|
|
|
|
|
def with_offset(n, fn):
|
|
"""At N=1 the rig's only unit is "a": give it B's offset."""
|
|
saved = dict(rig.OFFSET)
|
|
if n == 1:
|
|
rig.OFFSET["a"] = rig.OFFSET["b"]
|
|
try:
|
|
return fn()
|
|
finally:
|
|
rig.OFFSET.clear()
|
|
rig.OFFSET.update(saved)
|
|
|
|
|
|
REST, CUT = 400.0, 600.0 # rest (it learns), then ask, then cut
|
|
REST_LOAD = (lambda t: 0.0 if t < REST else 3000.0)
|
|
|
|
|
|
def caught(n, unit, tick, *, offset=True, dur=None, store=None, load=None,
|
|
cut_at=CUT, **kw):
|
|
"""Seconds from the cut to the refusal, or None, and the run."""
|
|
cu = "a" if n == 1 else unit
|
|
|
|
def go():
|
|
return rig.run(n, load or REST_LOAD, dur or cut_at + 300.0,
|
|
tick=tick, noise=4.0, store=store,
|
|
cut=lambda u, t: u == cu and t >= cut_at, **kw)
|
|
r = with_offset(n, go) if offset and unit == "b" else go()
|
|
ref = [t - cut_at for t, _u in r["refused"]]
|
|
return (ref[0] if ref else None), r
|
|
|
|
|
|
# ============================================================================ #
|
|
print("1. a refusing battery, B with its -100 W idle reading (+-4 W noise)")
|
|
TIMES, OLD = {}, {}
|
|
for unit in ("a", "b"):
|
|
for n in (1, 2):
|
|
for tick in (1.0, 2.0, 5.0):
|
|
TIMES[(unit, n, tick)] = caught(n, unit, tick)[0]
|
|
OLD[(unit, n, tick)] = without_fix(
|
|
lambda: caught(n, unit, tick)[0])
|
|
for k in TIMES:
|
|
print(f" refusing {k[0].upper()} N={k[1]} meter {k[2]:.0f} s: refused "
|
|
f"{TIMES[k]} s after the cut (before {OLD[k]})")
|
|
check("⭐ a refusing B is caught at N=1 and N=2: within 22 s plus one 10 s "
|
|
"publish at 1 and 2 s meters, 12 cycles plus a publish at 5 s",
|
|
all(TIMES[("b", n, t)] is not None
|
|
and TIMES[("b", n, t)] <= (32.0 if t < 5 else 12 * 5.0 + 10.0)
|
|
for n in (1, 2) for t in (1.0, 2.0, 5.0)))
|
|
check("non-vacuous: before, a refusing B is never caught (300 s), at N=1 "
|
|
"or N=2",
|
|
all(OLD[("b", n, t)] is None for n in (1, 2) for t in (1.0, 2.0, 5.0)))
|
|
check("⭐ a refusing A is caught exactly as before",
|
|
all(TIMES[("a", n, t)] == OLD[("a", n, t)] is not None
|
|
for n in (1, 2) for t in (1.0, 2.0, 5.0)))
|
|
|
|
# B refusing for 800 s: one refusal, held, not ended by its own idle reading.
|
|
_t, r = caught(2, "b", 1.0, dur=CUT + 800.0)
|
|
starts = [x for x in r["refused"] if x[1] == "inverter-b"]
|
|
check("a refusing B stays refused until its probe: its idle reading is not "
|
|
"read as taking power the other way (one refusal in 800 s)",
|
|
len(starts) == 1 and r["c"].inverters[1].refused == 1)
|
|
|
|
# A register frozen at rest on B: refused or stale within 36 s of the ask.
|
|
r = rig.run(2, lambda t: 2500.0 if t >= 600 else 0.0, 700.0, tick=1.0,
|
|
publish_s=1.0, dither=True,
|
|
freeze_at=lambda u: 300.0 if u == "b" else None)
|
|
park = min([t for t, u in r["stale"] if u == "inverter-b"]
|
|
+ [t for t, u in r["refused"] if u == "inverter-b"] or [None])
|
|
print(f" B's register frozen at rest from 300 s, asked at 600 s: 0 W at "
|
|
f"{park}")
|
|
check("a register frozen at rest on B is caught as on A (within 36 s of "
|
|
"the ask)", park is not None and park - 600.0 <= 36.0)
|
|
# At a 5 s meter the stale trip comes first: its refusal is centred too.
|
|
saved = dict(rig.OFFSET)
|
|
rig.OFFSET["b"] = -110.0
|
|
try:
|
|
r = rig.run(2, lambda t: 2500.0 if t >= 600 else 0.0, 760.0, tick=5.0,
|
|
publish_s=1.0, dither=True,
|
|
freeze_at=lambda u: 300.0 if u == "b" else None)
|
|
finally:
|
|
rig.OFFSET.clear()
|
|
rig.OFFSET.update(saved)
|
|
st = [t for t, u in r["stale"] if u == "inverter-b"]
|
|
ref = [t for t, u in r["refused"] if u == "inverter-b"]
|
|
print(f" the same at a 5 s meter, B idle -110 W: stale {st}, refused {ref}")
|
|
check("...and at a 5 s meter the stale trip on B's held -110 W is a refusal "
|
|
"of B (one trip), not a repeating site stop",
|
|
len(st) == 1 and ref == st)
|
|
|
|
# Review: the two centrings with no check of their own.
|
|
# N=1, 5 s meter: the stale trip on a held -110 W idle reading is a refusal.
|
|
saved = dict(rig.OFFSET)
|
|
rig.OFFSET["a"] = -110.0
|
|
try:
|
|
r = rig.run(1, lambda t: 2500.0 if t >= 600 else 0.0, 800.0, tick=5.0,
|
|
publish_s=1.0, dither=True, freeze_at=lambda u: 300.0)
|
|
finally:
|
|
rig.OFFSET.clear()
|
|
rig.OFFSET.update(saved)
|
|
st = [t for t, _u in r["stale"]]
|
|
ref = [t for t, _u in r["refused"]]
|
|
print(f" N=1, 5 s meter, register frozen on -110 W: stale {st}, refused "
|
|
f"{ref}")
|
|
check("...and at N=1 the same stale trip is a refusal (one trip), not a "
|
|
"repeating site stop", len(st) == 1 and ref == st)
|
|
|
|
# SAFETY-42: B asked under 200 W while giving nothing (its idle reading) is
|
|
# parked, not named as not delivering; A carries the house.
|
|
def small_ask():
|
|
return rig.run(2, lambda t: 150.0, 700.0, tick=1.0, noise=4.0,
|
|
store={unit_key("inverter-b", "idle_w"): -100.0},
|
|
cut=lambda u, t: u == "b")["c"].inverters[1]
|
|
|
|
|
|
b, b_old = small_ask(), without_fix(small_ask)
|
|
check("SAFETY-42: B giving nothing at a small ask, idle -100 W, is parked "
|
|
"(not latched as not delivering in 700 s; with idle 0 W it is)",
|
|
not b.delivery_wd.tripped and b.idle.value == -100.0
|
|
and b_old.delivery_wd.tripped)
|
|
|
|
# ============================================================================ #
|
|
print("\n2. a healthy B with its offset is never refused or flagged")
|
|
SHAPES = ((10.0, False), (1.0, True))
|
|
NAMES = ("rest_step", "drain_start", "drain_charge", "reversal")
|
|
SAVED = {"b": {unit_key("inverter-b", "idle_w"): -100.0},
|
|
"a": {unit_key("inverter-a", "idle_w"): -100.0}}
|
|
bad = []
|
|
runs = 0
|
|
for n in (1, 2):
|
|
for name in NAMES:
|
|
for tick in (1.0, 2.0, 5.0):
|
|
for pub, dith in SHAPES:
|
|
for store in (None, SAVED["a" if n == 1 else "b"]):
|
|
s, f, _r = with_offset(n, lambda: rig.counts(
|
|
n, name, tick=tick, publish_s=pub, dither=dith,
|
|
store=store))
|
|
runs += 1
|
|
if s or f:
|
|
bad.append((n, name, tick, pub, bool(store), s, f))
|
|
print(f" {runs} runs, stale or refused: {bad}")
|
|
check("⭐ SAFETY-94's scenarios with B's offset (N=1 and N=2, 1/2/5 s "
|
|
"meters, both register shapes, idle learned or restored): no stale "
|
|
"trip and no refusal", not bad and runs == 96)
|
|
rev = []
|
|
for n in (1, 2):
|
|
for tau in (10.0, 20.0):
|
|
for charge in (-1500.0, -2500.0):
|
|
for house in (300.0, 800.0, 2000.0):
|
|
x = with_offset(n, lambda: rig.run(
|
|
n, rig.reversal(300.0, charge, house), 600.0, tau=tau,
|
|
tick=1.0, store=SAVED["a" if n == 1 else "b"]))
|
|
if x["stale"] or x["refused"]:
|
|
rev.append((n, tau, charge, house))
|
|
check("⭐ a charge reversing into a discharge (lag 10 and 20 s, 1 s meter) "
|
|
"with B's idle reading restored is never refused or flagged",
|
|
not rev)
|
|
|
|
# ============================================================================ #
|
|
print("\n3. learning the idle reading")
|
|
_t, r = caught(2, "b", 1.0)
|
|
c = r["c"]
|
|
check("learned during the rest: A about 0 W, B about -100 W (used: -75 W), "
|
|
"and saved",
|
|
abs(c.inverters[0].idle_w) < 5
|
|
and abs(c.inverters[1].idle.value + 100) < 5
|
|
and c.inverters[1].idle_w == -idle.IDLE_MAX_W
|
|
and abs(c.store.data.get(unit_key("inverter-b", "idle_w"), 0) + 100)
|
|
< 5)
|
|
|
|
# A refusing B from the start, no rest, nothing saved: nothing to learn from.
|
|
_t, r = caught(2, "b", 1.0, load=lambda t: 3000.0, cut_at=60.0, dur=1200.0)
|
|
check("a battery refusing from the start teaches nothing: asked the whole "
|
|
"time, then held at 0 W by its refusal (none learned, none saved)",
|
|
r["c"].inverters[1].idle.value is None
|
|
and unit_key("inverter-b", "idle_w") not in r["c"].store.data)
|
|
# ... which is why the value is saved: restored, the same battery is caught.
|
|
t_saved, _r = caught(2, "b", 1.0, load=lambda t: 3000.0, cut_at=60.0,
|
|
store=SAVED["b"])
|
|
t_none, _r = caught(2, "b", 1.0, load=lambda t: 3000.0, cut_at=60.0)
|
|
print(f" refusing from the start: saved idle {t_saved} s, none {t_none}")
|
|
check("⭐ after a restart the saved idle reading is used: a B refusing from "
|
|
"the start is caught in 32 s; with nothing saved it is not",
|
|
t_saved is not None and t_saved <= 32.0 and t_none is None)
|
|
|
|
# Not learned while refused: a refusing B reading -160 W (in the band) does
|
|
# not pull its idle reading during the 0 W its refusal holds.
|
|
class _P(rig.Plant):
|
|
def advance(self, t):
|
|
super().advance(t)
|
|
for u in self.units:
|
|
if self.cut(u, self.t):
|
|
self.reg[u] = -160.0
|
|
|
|
|
|
real_plant = rig.Plant
|
|
rig.Plant = _P
|
|
try:
|
|
_t, r = caught(2, "b", 1.0, dur=CUT + 800.0)
|
|
finally:
|
|
rig.Plant = real_plant
|
|
b = r["c"].inverters[1]
|
|
print(f" refusing B reading -160 W: refused at {_t} s, idle "
|
|
f"{b.idle.value:.1f}")
|
|
check("⭐ not learned while refused: 800 s held at 0 W reading -160 W leaves "
|
|
"B's idle reading at -100 W",
|
|
_t is not None and b.refused and abs(b.idle.value + 100.0) < 1.0)
|
|
|
|
# Not learned while control is stopped (another owner may drive the battery)
|
|
# or while the unit's sign watchdog has latched.
|
|
class _Stopped(app_main.Controller):
|
|
auto = property(lambda self: False, lambda self, v: None)
|
|
|
|
|
|
rig.Controller = _Stopped
|
|
try:
|
|
r = rig.run(2, lambda t: 0.0, 600.0, tick=1.0)
|
|
finally:
|
|
rig.Controller = app_main.Controller
|
|
r2 = rig.run(2, lambda t: 0.0, 600.0, tick=1.0,
|
|
store={unit_key("inverter-b", "sign_tripped"): True})
|
|
check("not learned while control is stopped, nor on a unit whose sign "
|
|
"watchdog has latched (600 s at 0 W); its sibling learns",
|
|
all(i.idle.value is None for i in r["c"].inverters)
|
|
and r2["c"].inverters[1].idle.value is None
|
|
and r2["c"].inverters[0].idle.value is not None)
|
|
|
|
# A battery capped at 150 W under a big ask is not idle.
|
|
x = idle.IdleReading()
|
|
for t in range(0, 3600):
|
|
x.observe(150.0, 2000.0, float(t), True)
|
|
check("a reading under a command over 50 W is never a sample (a capped "
|
|
"battery is not idle)", x.value is None and x.w == 0.0)
|
|
|
|
x = idle.IdleReading()
|
|
for t in range(0, 239):
|
|
x.observe(-100.0, 10.0, float(t), True)
|
|
first = x.value
|
|
x.observe(-100.0, 10.0, 239.0, True)
|
|
x.observe(-100.0, 10.0, 240.0, True)
|
|
check("the first value needs 120 s asked ~0 and then 120 s of samples",
|
|
first is None and abs(x.value + 100.0) < 1e-6)
|
|
y = idle.IdleReading()
|
|
for t in range(0, 400):
|
|
y.observe(-100.0, 60.0 if t % 100 == 0 else 0.0, float(t), True)
|
|
check("a command over 50 W restarts the quiet time (a 60 W blip every "
|
|
"100 s: never learned)", y.value is None)
|
|
z = idle.IdleReading()
|
|
for t in range(0, 400):
|
|
z.observe(-100.0, 0.0, float(t), False)
|
|
check("nothing is learned while learning is not allowed (refused, stale, "
|
|
"tripped or stopped)", z.value is None)
|
|
|
|
for t in range(240, 300):
|
|
x.observe(-130.0, 0.0, float(t), True)
|
|
after_min = x.value
|
|
for t in range(300, 240 + 3600):
|
|
x.observe(-130.0, 0.0, float(t), True)
|
|
print(f" drift -100 -> -130 W: {after_min:.1f} after 60 s, {x.value:.1f} "
|
|
f"after an hour")
|
|
check("drift is followed slowly: under 2 W in the first minute, most of it "
|
|
"in an hour", -102.0 < after_min < -100.0 and -130 < x.value < -124)
|
|
|
|
o = idle.IdleReading()
|
|
for t in range(0, 600):
|
|
o.observe(-300.0, 0.0, float(t), True)
|
|
p = idle.IdleReading()
|
|
for t in range(0, 600):
|
|
p.observe(2000.0 if t < 300 else -240.0, 0.0, float(t), True)
|
|
check("a reading beyond 250 W is not a sample (-300 W: never learned; a "
|
|
"register held at 2000 W then -240 W: -240 W)",
|
|
o.value is None and p.value is not None and abs(p.value + 240.0) < 1e-6)
|
|
check("a saved value is clamped to 250 W, the band's centre to 75 W, and "
|
|
"garbage is ignored",
|
|
idle.IdleReading(900).value == idle.IDLE_SAMPLE_W
|
|
and idle.IdleReading(900).w == idle.IDLE_MAX_W
|
|
and idle.IdleReading(-900.0).w == -idle.IDLE_MAX_W
|
|
and idle.IdleReading(-60.0).w == -60.0
|
|
and idle.IdleReading("x").value is None
|
|
and idle.IdleReading(float("nan")).value is None)
|
|
|
|
# ============================================================================ #
|
|
print("\n3b. a wrong idle reading (owner, 2026-10-04: centre clamped to 75 W)")
|
|
|
|
|
|
def wrong_idle(n, unit, err, restore):
|
|
"""Seconds from the cut to the refusal when the idle reading is `err` W
|
|
off: taught wrong during the rest, or restored wrong."""
|
|
cu = "a" if n == 1 else unit
|
|
true = -100.0 if unit == "b" else 0.0
|
|
if restore:
|
|
return caught(n, unit, 1.0, store={
|
|
unit_key(f"inverter-{cu}", "idle_w"): true + err})[0]
|
|
|
|
class _W(rig.Plant):
|
|
def advance(self, t):
|
|
rig.OFFSET[cu] = true + (err if t < REST else 0.0)
|
|
super().advance(t)
|
|
saved, real = dict(rig.OFFSET), rig.Plant
|
|
rig.Plant = _W
|
|
try:
|
|
return caught(n, unit, 1.0)[0]
|
|
finally:
|
|
rig.Plant = real
|
|
rig.OFFSET.clear()
|
|
rig.OFFSET.update(saved)
|
|
|
|
|
|
A_WRONG = {(n, e, rs): wrong_idle(n, "a", e, rs) for n in (1, 2)
|
|
for e, rs in ((-250.0, False), (-150.0, False), (150.0, False),
|
|
(250.0, False), (-250.0, True), (200.0, True),
|
|
(250.0, True))}
|
|
print(f" A (reads about 0 W) refusing, idle off by E: {A_WRONG}")
|
|
check("⭐ a battery reading about 0 W while refusing is caught whatever idle "
|
|
"reading was learned or restored (A, E -250..+250 W, N=1 and N=2)",
|
|
all(v is not None and v <= 32.0 for v in A_WRONG.values()))
|
|
check("⭐ ...including a wrong saved idle_w of +200 W on A",
|
|
all(A_WRONG[(n, 200.0, True)] is not None for n in (1, 2)))
|
|
B_WRONG = {(n, e): wrong_idle(n, "b", e, False) for n in (1, 2)
|
|
for e in (-150.0, -50.0, 90.0)}
|
|
print(f" B refusing, idle taught wrong by E: {B_WRONG}")
|
|
check("B is caught across -150..+90 W of error in its idle reading",
|
|
all(v is not None and v <= 32.0 for v in B_WRONG.values()))
|
|
|
|
# ============================================================================ #
|
|
print("\n4. the rules at their edges")
|
|
|
|
|
|
class _Inv:
|
|
def __init__(self, idle_w):
|
|
self.batt, self.ref_count, self.ref_since, self.ref_sign = 0.0, 0, None, 0
|
|
self.ref_from, self.idle_w = None, idle_w
|
|
|
|
def standing_w(self):
|
|
return 800.0
|
|
|
|
|
|
class _C:
|
|
_now_mono = 0.0
|
|
|
|
|
|
def evidence(readings, idle_w):
|
|
c, inv = _C(), _Inv(idle_w)
|
|
got = []
|
|
for i, b in enumerate(readings):
|
|
c._now_mono, inv.batt = float(i), b
|
|
got.append(app_main.Controller._gave_nothing(c, inv))
|
|
return got
|
|
|
|
|
|
flat_b = evidence([-104.0 + (i % 3) for i in range(30)], -100.0)
|
|
flat_b0 = evidence([-104.0 + (i % 3) for i in range(30)], 0.0)
|
|
gives = evidence([0.0] * 30, -100.0)
|
|
takes = evidence([-200.0] * 30, -100.0)
|
|
check("the band is centred on the idle reading: B flat at -104 W is evidence "
|
|
"at 22 s with idle -100 W and never with idle 0 W; reading 0 W (giving "
|
|
"100 W) or -200 W (taking 100 W) is not",
|
|
1 in flat_b and flat_b.index(1) == 22 and not any(flat_b0)
|
|
and not any(gives) and not any(takes))
|
|
|
|
T = control.Tuning(max_w=4000.0, slew_w=400.0)
|
|
args = dict(prev_w=1200.0, grid_w=1200.0, tuning=T,
|
|
sat_count=control.SETTLE_CYCLES, i_w=1200.0, refused=0,
|
|
gave_nothing=True)
|
|
check("compute(): the N=1 engage is centred on idle_w (-104 W refuses with "
|
|
"idle -100, not with 0; -4 W refuses with idle 0, not with -150)",
|
|
control.compute(actual_w=-104.0, idle_w=-100.0, **args).refused == 1
|
|
and control.compute(actual_w=-104.0, **args).refused == 0
|
|
and control.compute(actual_w=-4.0, **args).refused == 1
|
|
and control.compute(actual_w=-4.0, idle_w=-150.0, **args).refused == 0)
|
|
args.update(refused=-1, i_w=-1200.0, prev_w=0.0, grid_w=-1200.0)
|
|
check("compute(): a charge refusal ends when the battery takes 100 W more "
|
|
"than its idle reading, not at its idle reading",
|
|
control.compute(actual_w=-110.0, idle_w=-100.0, **args).refused == -1
|
|
and control.compute(actual_w=-205.0, idle_w=-100.0, **args).refused
|
|
== 0
|
|
and control.compute(actual_w=-110.0, **args).refused == 0)
|
|
check("the band's centre stays within 75 W of 0 W, so the band keeps 0 W "
|
|
"+-25 W; samples to 250 W",
|
|
idle.IDLE_MAX_W == 75.0 and idle.IDLE_SAMPLE_W == 250.0
|
|
and control.REFUSED_BELOW_W - idle.IDLE_MAX_W >= 25.0
|
|
and idle.IDLE_ASK_W < control.PROBE_W)
|
|
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED: " + ", ".join(fails))
|
|
sys.exit(1)
|
|
print("all checks passed")
|