adminandClaude Opus 5.5
98dd110a4f
Mutation M13 (a battery out of the split counted in the house point) survived; a direct check now kills it. win[0] is read through a guard (test_named_absence). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
402 lines
17 KiB
Python
402 lines
17 KiB
Python
"""Runnable check for SAFETY-103: no swing on old battery readings after a
|
|
drain ends. `python3 test_safety103.py`
|
|
|
|
While a battery settles from its ended drain (SAFETY-95), the loop works
|
|
from the grid plus each battery's reading. A reading can be one register
|
|
period (10 s) older than the meter: on the rig (SAFETY-95 test, check 5) A's
|
|
reading still said 1238 W while A was at 616 W, and B went 860 -> 1650 ->
|
|
870 W. Now the house point is held, and moved only as far as the meter
|
|
proves (each battery between its last changed reading and its commands
|
|
since). Drives the REAL Controller on tools/safety103_plant.py: a lagging
|
|
battery, the meter every 1 s, registers every 10 s at the rig's phases, B
|
|
idling at -100 W. DECISIONS/SAFETY-103-old-readings-after-a-drain.md.
|
|
|
|
1. The rig case and every drain end at N=2 and N=3, 10 s registers: no
|
|
swing beyond one slew step; release/1.0 swings (non-vacuous).
|
|
2. SAFETY-95/96's gains stay: no jump at the end, no give into a surplus,
|
|
one slew step a cycle, refusal and stale link still learned.
|
|
3. A house step inside the window is still followed.
|
|
4. With readings as fresh as the meter, every row is release/1.0's; N=1 is
|
|
untouched.
|
|
5. The rule's parts, one by one.
|
|
|
|
Mutations (MUTATION.md, purged copies) are listed in the pull request.
|
|
"""
|
|
|
|
import hashlib
|
|
import logging
|
|
import math
|
|
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.WARNING)
|
|
|
|
import safety103_plant as p # noqa: E402
|
|
from app.main import Controller # noqa: E402
|
|
from app.maintenance import DRAIN # noqa: E402
|
|
|
|
fails = []
|
|
SLEW = 400.0 # the rig's slew_w
|
|
AIM = -10.0 # the rig's target_grid_w
|
|
T = p.T
|
|
A = p.A
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
fixed = Controller._drain_grid_w
|
|
track = getattr(Controller, "_track_reads", None)
|
|
|
|
|
|
def _release(self, live):
|
|
"""release/1.0's `_drain_grid_w` (SAFETY-95 as merged): every battery at
|
|
its reading."""
|
|
grid = float(self.grid)
|
|
self._drain_tail &= set(self._settle) & set(live)
|
|
if not self._drain_tail:
|
|
return grid
|
|
return grid - float(self.target) + sum(
|
|
float(inv.batt) for inv in self.inverters if inv.name in live)
|
|
|
|
|
|
def without(fn):
|
|
"""Run `fn` with SAFETY-103 taken out (release/1.0's rule)."""
|
|
Controller._drain_grid_w = _release
|
|
try:
|
|
return fn()
|
|
finally:
|
|
Controller._drain_grid_w = fixed
|
|
|
|
|
|
def soc_done(u, t):
|
|
return 5.0 if (u == "a" and t >= T) else 50.0
|
|
|
|
|
|
# A's drain ends at T: switched Off (the rig case), completed on SoC, its
|
|
# battery refusing (gives nothing from T; SAFETY-73 learns it later), or its
|
|
# link going stale (writes refused from T, its board to 0 W 25 s later).
|
|
ENDS = {
|
|
"Off": dict(),
|
|
"complete": dict(before=None, soc=soc_done),
|
|
"refusal": dict(before=None, dead=("a", T), dur=T + 150.0),
|
|
"stale link": dict(before=None, reject={"a": T}, dead=("a", T + 25.0),
|
|
dur=T + 150.0),
|
|
}
|
|
|
|
|
|
def end_of(rows):
|
|
"""The first cycle A's drain is over (its command left 2000 W, or its
|
|
phase left DRAIN); None if never."""
|
|
return next((r["t"] for r in rows if r["t"] >= T and (
|
|
r["cmd"][0] < 1990.0 or r["phase"][0] != DRAIN)), None)
|
|
|
|
|
|
def measure(rows, n, house):
|
|
"""From the end on: the largest swing of the others' summed command in
|
|
the 23 s window (a move later undone; the slew step is the site's); their
|
|
largest command on the end cycle; the site target's largest step after
|
|
the end cycle; the others' largest command after the end (for a
|
|
surplus); the grid's range in the window."""
|
|
te = end_of(rows)
|
|
if te is None:
|
|
return dict(end=math.nan, swing=math.nan, first=math.nan,
|
|
step=math.nan, give=math.nan, gmin=math.nan,
|
|
gmax=math.nan, rows=rows)
|
|
win = [r for r in rows if te <= r["t"] <= te + 23.0]
|
|
sw = 0.0
|
|
w = [sum(r["cmd"][1:]) for r in win][1:]
|
|
for j in range(len(w)):
|
|
sw = max(sw, min(w[j] - min(w[:j + 1]), w[j] - min(w[j:])),
|
|
min(max(w[:j + 1]) - w[j], max(w[j:]) - w[j]))
|
|
seg = [r["target"] for r in rows if r["t"] >= te]
|
|
return dict(
|
|
end=te, swing=sw,
|
|
first=max(win[0]["cmd"][1:]) if win else math.nan,
|
|
step=max(abs(b - a) for a, b in zip(seg, seg[1:])),
|
|
give=max(max(r["cmd"][1:]) for r in rows if r["t"] >= te),
|
|
gmin=min(r["grid"] for r in win), gmax=max(r["grid"] for r in win),
|
|
rows=rows)
|
|
|
|
|
|
def run(kind, n, house, **k):
|
|
return p.run(n, house, **dict(ENDS[kind], **k))[0]
|
|
|
|
|
|
TABLE = []
|
|
# A settles from its drain: the tail this rule is about. A refusing battery
|
|
# already reads ~0 W, and a dropped one is out of the split (the meter has
|
|
# it), so their tails last at most a cycle.
|
|
SETTLE = ("Off", "complete")
|
|
|
|
|
|
def case(kind, n, house, **k):
|
|
now = measure(run(kind, n, house, **k), n, house)
|
|
old = measure(without(lambda: run(kind, n, house, **k)), n, house)
|
|
fresh = measure(run(kind, n, house, reg_s=1.0, **k), n, house)
|
|
TABLE.append((kind, n, house, now, old, fresh))
|
|
return now, old, fresh
|
|
|
|
|
|
# ============================================================================ #
|
|
print("1. no swing on an old reading: every drain end, N=2 and N=3, 10 s "
|
|
"registers")
|
|
for kind in ENDS:
|
|
for n in (2, 3):
|
|
for house in (1200.0, 4000.0, -1500.0):
|
|
now, old, fresh = case(kind, n, house)
|
|
own = kind in SETTLE
|
|
check(f"{'⭐' if kind == 'Off' else ' '} {kind}, N={n}, house "
|
|
f"{house:.0f} W: old readings add at most one slew step "
|
|
f"to the others' swing in the 23 s after the end"
|
|
f"{', and it is at most one slew step' if own else ''} "
|
|
f"({now['swing']:.0f} W; 1 s registers "
|
|
f"{fresh['swing']:.0f}; release/1.0 {old['swing']:.0f})",
|
|
now["swing"] - fresh["swing"] <= SLEW
|
|
and (not own or now["swing"] <= SLEW))
|
|
_by = {(k, n, h): (a, b) for k, n, h, a, b, f in TABLE}
|
|
_now, _old = _by[("Off", 2, 1200.0)]
|
|
_b = [r["cmd"][1] for r in _now["rows"] if 0 <= r["t"] - _now["end"] <= 23]
|
|
_bo = [r["cmd"][1] for r in _old["rows"] if 0 <= r["t"] - _old["end"] <= 23]
|
|
print(f" rig check 5 (N=2, 1200 W, A Off): B {_b}\n"
|
|
f" release/1.0: B {_bo}")
|
|
check("⭐ the rig case (SAFETY-95 check 5): release/1.0 swings B by more "
|
|
"than a slew step on A's old reading; now B holds within 50 W",
|
|
_old["swing"] > SLEW and max(_b) - min(_b) <= 50.0)
|
|
for n in (2, 3):
|
|
check(f"non-vacuous (N={n}): on release/1.0 old readings add more than "
|
|
f"a slew step in at least one of these cases",
|
|
any(b["swing"] - f["swing"] > SLEW
|
|
for k, nn, h, a, b, f in TABLE if nn == n))
|
|
_s5 = []
|
|
for lag in (5.0, 20.0):
|
|
for reg in (5.0, 10.0):
|
|
for n in (2, 3):
|
|
_s5.append(measure(run("Off", n, 1200.0, lag=lag, reg_s=reg), n,
|
|
1200.0)["swing"])
|
|
check(f"no plant constant is assumed: battery lag 5 s and 20 s, registers "
|
|
f"every 5 s and 10 s, N=2 and N=3: largest swing {max(_s5):.0f} W",
|
|
max(_s5) <= SLEW)
|
|
_ph = [measure(run("Off", 2, 1200.0, phase={"a": a, "b": b}), 2, 1200.0)
|
|
["swing"] for a, b in ((0.0, 0.5), (0.9, 0.1), (0.3, 0.3))]
|
|
check(f"other register phases: largest swing {max(_ph):.0f} W",
|
|
max(_ph) <= SLEW)
|
|
|
|
# ============================================================================ #
|
|
print("\n2. SAFETY-95/96's gains stay")
|
|
for kind in ENDS:
|
|
for n in (2, 3):
|
|
now, old = _by[(kind, n, 1200.0)]
|
|
check(f"{kind}, N={n}, 1200 W: no jump on the end cycle (others "
|
|
f"<= the house, {now['first']:.0f} W) and the site moves at "
|
|
f"most one slew step a cycle after ({now['step']:.0f} W)",
|
|
now["first"] <= 1200.0 - AIM + 50.0 and now["step"] <= SLEW)
|
|
now, old = _by[(kind, n, -1500.0)]
|
|
check(f"{kind}, N={n}, solar surplus: no battery gives into it "
|
|
f"(largest command after the end {now['give']:.0f} W"
|
|
+ (f"; a lost link as release/1.0, {old['give']:.0f} W)"
|
|
if kind == "stale link" else ")"),
|
|
now["give"] <= (max(50.0, old["give"]) if kind == "stale link"
|
|
else 50.0))
|
|
_now, _old = _by[("Off", 2, 1200.0)]
|
|
check(f"the grid in the window stays inside release/1.0's range "
|
|
f"([{_now['gmin']:.0f}, {_now['gmax']:.0f}] W, release/1.0 "
|
|
f"[{_old['gmin']:.0f}, {_old['gmax']:.0f}])",
|
|
_now["gmin"] >= _old["gmin"] - 50.0 and _now["gmax"] <= _old["gmax"])
|
|
|
|
|
|
_ref = {}
|
|
for name, f in (("now", lambda: p.run(2, 1200.0, **ENDS["refusal"])),
|
|
("old", lambda: without(lambda: p.run(
|
|
2, 1200.0, **ENDS["refusal"])))):
|
|
rows, c = f()
|
|
_ref[name] = (c.inverters[0].refused, c.inverters[0].refused_at)
|
|
check(f"a refusal is still learned, at the same instant as release/1.0 "
|
|
f"({_ref['now']})", _ref["now"] == _ref["old"]
|
|
and _ref["now"][0] == 1)
|
|
_st = {}
|
|
for name, f in (("now", lambda: p.run(2, 1200.0, **ENDS["stale link"])),
|
|
("old", lambda: without(lambda: p.run(
|
|
2, 1200.0, **ENDS["stale link"])))):
|
|
rows, c = f()
|
|
_st[name] = sorted(c.live_units())
|
|
check(f"a stale link is still dropped, as in release/1.0 "
|
|
f"(live: {_st['now']})", _st["now"] == _st["old"]
|
|
and len(_st["now"]) == 1)
|
|
|
|
# ============================================================================ #
|
|
print("\n3. a house step inside the window is still followed")
|
|
|
|
|
|
def step(sign):
|
|
return lambda t: 1200.0 + (sign * 2000.0 if t >= T + 5.0 else 0.0)
|
|
|
|
|
|
for sign, what in ((1, "kettle +2000 W"), (-1, "solar -2000 W")):
|
|
for n in (2, 3):
|
|
rows = p.run(n, step(sign))[0]
|
|
rows_o = without(lambda: p.run(n, step(sign)))[0]
|
|
te = end_of(rows)
|
|
|
|
def wh(rs):
|
|
return sum(abs(r["grid"] - AIM) for r in rs
|
|
if te + 5.0 <= r["t"] <= te + 35.0) / 3600.0
|
|
sw = measure(rows, n, 1200.0)["swing"]
|
|
check(f"{what} at E+5 s, N={n}: the grid's distance from its aim "
|
|
f"over the next 30 s is within 1.5 Wh of release/1.0's "
|
|
f"({wh(rows):.1f} Wh, release/1.0 {wh(rows_o):.1f}); swing "
|
|
f"{sw:.0f} W", wh(rows) <= wh(rows_o) + 1.5 and sw <= SLEW)
|
|
|
|
# ============================================================================ #
|
|
print("\n4. fresh readings and N=1: release/1.0's rows")
|
|
|
|
|
|
def digest(rows):
|
|
return hashlib.sha256(repr(rows).encode()).hexdigest()
|
|
|
|
|
|
_fresh = [(n, h) for n in (2, 3) for h in (1200.0, 4000.0, -1500.0)]
|
|
check("⭐ registers as fresh as the meter (1 s), N=2 and N=3, three houses: "
|
|
"every row byte-identical with release/1.0 (a reading that changed "
|
|
"this cycle is taken as it is)",
|
|
[digest(p.run(n, h, reg_s=1.0)[0]) for n, h in _fresh]
|
|
== [digest(without(lambda: p.run(n, h, reg_s=1.0))[0])
|
|
for n, h in _fresh])
|
|
_rf = [(n, h) for n in (2, 3) for h in (1200.0, -1500.0)]
|
|
check("⭐ a refusal end, 10 s registers, N=2 and N=3, 1200 W and a surplus: "
|
|
"every row byte-identical with release/1.0 (A reads ~0 W and will not "
|
|
"give, so its side is its reading)",
|
|
[digest(run("refusal", n, h)) for n, h in _rf]
|
|
== [digest(without(lambda: run("refusal", n, h))) for n, h in _rf])
|
|
_n1 = [dict(), dict(before=None, soc=soc_done)]
|
|
check("⭐ N=1, A switched Off and A's drain completing, 10 s registers: "
|
|
"every row byte-identical with release/1.0",
|
|
[digest(p.run(1, 1200.0, **k)[0]) for k in _n1]
|
|
== [digest(without(lambda: p.run(1, 1200.0, **k))[0]) for k in _n1])
|
|
|
|
# ============================================================================ #
|
|
print("\n5. the rule's parts")
|
|
_c = p.run(2, 1200.0, dur=4.0)[1]
|
|
AB = {"inverter-a", "inverter-b"}
|
|
ia, ib = _c.inverters
|
|
|
|
|
|
def at(now, batts, cmds, changed):
|
|
"""Set the clock, each battery's reading, its standing command and the
|
|
instant its reading last changed; then track."""
|
|
_c._now_mono = now
|
|
for inv, b, cmd, ch in zip(_c.inverters, batts, cmds, changed):
|
|
inv.batt, inv.last_write = b, cmd
|
|
inv.batt_fresh.changed_mono = ch
|
|
_c._track_reads()
|
|
|
|
|
|
_c._read_span = {}
|
|
at(10.0, (2000.0, -100.0), (2000.0, 0.0), (10.0, 10.0))
|
|
check("a reading that changed is the start of its range: (R, R) joined "
|
|
"with the command on its wire",
|
|
_c._read_span[A] == (2000.0, 2000.0, 10.0)
|
|
and _c._read_span["inverter-b"] == (-100.0, 0.0, 10.0))
|
|
at(11.0, (2000.0, -100.0), (0.0, 1140.0), (10.0, 10.0))
|
|
at(12.0, (2000.0, -100.0), (0.0, 800.0), (10.0, 10.0))
|
|
check("every command since the reading widens the range (a lagging "
|
|
"battery moves toward each in turn): A [0, 2000], B [-100, 1140]",
|
|
_c._read_span[A][:2] == (0.0, 2000.0)
|
|
and _c._read_span["inverter-b"][:2] == (-100.0, 1140.0))
|
|
at(13.0, (1241.0, -100.0), (0.0, 800.0), (13.0, 10.0))
|
|
check("a new reading starts the range again: A [0, 1241]",
|
|
_c._read_span[A][:2] == (0.0, 1241.0))
|
|
_c._read_span = {}
|
|
_c.inverters = _c.inverters[:1]
|
|
at(14.0, (2000.0,), (2000.0,), (14.0,))
|
|
_c.inverters = [ia, ib]
|
|
check("N=1: nothing is tracked", _c._read_span == {})
|
|
|
|
|
|
def tail(now, grid, target, held=None, live=AB):
|
|
"""`_drain_grid_w` with A settling: the error it hands the loop (grid
|
|
minus aim) and the house point it holds."""
|
|
_c._now_mono, _c.grid, _c.target = now, grid, target
|
|
_c._drain_tail, _c._settle = {A}, {A: 0.0}
|
|
_c._drain_house = held
|
|
g = _c._drain_grid_w(live)
|
|
return round(g - AIM, 6), _c._drain_house
|
|
|
|
|
|
# A: reading 1241 W from 13 s, range [0, 1241]; B: reading -100 W from 10 s,
|
|
# range [-100, 1140]. Grid -330 W: the house is in
|
|
# [-330 + 10 + 0 - 100, -330 + 10 + 1241 + 1140] = [-420, 2061].
|
|
_c._read_span = {A: (0.0, 1241.0, 13.0), "inverter-b": (-100.0, 1140.0, 10.0)}
|
|
ia.batt, ib.batt = 1241.0, -100.0
|
|
ia.batt_fresh.changed_mono, ib.batt_fresh.changed_mono = 13.0, 10.0
|
|
check("⭐ held: the first window cycle holds the loop's start (1140 W) while "
|
|
"the meter allows it (error 0; release/1.0's would be -330 + 1241 - "
|
|
"100 + 10 - 1140 = -319 W)", tail(14.0, -330.0, 1140.0) == (0.0, 1140.0)
|
|
and round(_release(_c, AB) - AIM) == -319)
|
|
check("⭐ held, then kept: a later cycle keeps the held point, not the "
|
|
"loop's target", tail(15.0, -330.0, 1000.0, held=1140.0)
|
|
== (140.0, 1140.0))
|
|
check("⭐ moved as far as the meter proves: a grid of +2000 W puts the "
|
|
"house at least at 2000 + 10 + 0 - 100 = 1910 W",
|
|
tail(15.0, 2000.0, 1140.0, held=1140.0) == (770.0, 1910.0))
|
|
check("...and down: a grid of -2500 W puts it at most at -2500 + 10 + 1241 "
|
|
"+ 1140 = -109 W", tail(15.0, -2500.0, 1140.0, held=1140.0)
|
|
== (-1249.0, -109.0))
|
|
ia.batt_fresh.changed_mono = 15.0
|
|
check("⭐ a reading that changed this cycle is taken as it is: A at 1241 W, "
|
|
"not [0, 1241], so a grid of +2000 W puts the house at least at "
|
|
"2000 + 10 + 1241 - 100 = 3151 W", tail(15.0, 2000.0, 1140.0,
|
|
held=1140.0) == (2011.0, 3151.0))
|
|
ia.batt_fresh.changed_mono = 13.0
|
|
# A refuses to give (SAFETY-73): it reads 8 W, though it was asked 2000 W
|
|
# since that reading. Grid -1500 W: the house is at most -1500 + 10 + 8 +
|
|
# 1140 = -342 W, not -1500 + 10 + 2000 + 1140 = 1650 W.
|
|
_c._read_span[A] = (8.0, 2000.0, 13.0)
|
|
ia.batt = 8.0
|
|
ia.refused = 1
|
|
_ref1 = tail(15.0, -1500.0, 1140.0, held=1140.0)
|
|
ia.refused = -1
|
|
_c._read_span[A] = (-2000.0, 8.0, 13.0)
|
|
_ref2 = tail(15.0, 1000.0, -500.0, held=-500.0)
|
|
ia.refused = 0
|
|
check("⭐ a battery that refuses one way is taken at its reading on that "
|
|
"side: will not give, asked 2000 W, reads 8 W -> the house at most "
|
|
"-342 W; will not take, asked -2000 W -> at least 1000 + 10 + 8 - 100 "
|
|
"= 918 W", _ref1 == (-1482.0, -342.0) and _ref2 == (1418.0, 918.0))
|
|
_c._read_span[A], ia.batt = (0.0, 1241.0, 10.0), 1241.0
|
|
check("a range from an older reading than the one in hand (a cycle the "
|
|
"loop did not run) is not used: A at its reading, 1241 W",
|
|
tail(15.0, 2000.0, 1140.0, held=1140.0) == (2011.0, 3151.0))
|
|
_c._read_span[A] = (0.0, 1241.0, 13.0)
|
|
check("⭐ a battery out of the split is not counted (the meter has its "
|
|
"power): B dropped, a grid of +2000 W puts the house at least at "
|
|
"2000 + 10 + 0 = 2010 W, not 1910 W",
|
|
tail(15.0, 2000.0, 1140.0, held=1140.0, live={A}) == (870.0, 2010.0))
|
|
_c._drain_tail, _c._settle, _c._drain_house = {A}, {}, 1140.0
|
|
_c.grid = -330.0
|
|
check("⭐ the tail's end lets the hold go: the meter again, and the next "
|
|
"tail starts from the loop's start",
|
|
_c._drain_grid_w(AB) == -330.0 and _c._drain_house is None)
|
|
|
|
# ============================================================================ #
|
|
print("\n drain end | N | house W | swing W now/release/1 s registers | "
|
|
"end-cycle others W now/release | site step W now/release | grid in "
|
|
"window W now | release")
|
|
for k, n, h, a, b, f in TABLE:
|
|
print(f" {k} | {n} | {h:.0f} | {a['swing']:.0f}/{b['swing']:.0f}/"
|
|
f"{f['swing']:.0f} | "
|
|
f"{a['first']:.0f}/{b['first']:.0f} | {a['step']:.0f}/"
|
|
f"{b['step']:.0f} | [{a['gmin']:.0f}, {a['gmax']:.0f}] | "
|
|
f"[{b['gmin']:.0f}, {b['gmax']:.0f}]")
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED: " + ", ".join(fails))
|
|
sys.exit(1)
|
|
print("all SAFETY-103 checks passed")
|