adminandClaude Opus 5.5
214051d1a8
A battery still in CHARGE but dropped by the allocator (stale link, or a SAFETY-73 refusal contradicting its charge) left its -2000 W in the loop's start point, and B went 0 -> -1600 W in one cycle. The end is now noted after the unit claims are filed: in CHARGE, live and not contradicted. Tests: part 6 (stale link, refusal) and a spy on the guard suspension. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
423 lines
16 KiB
Python
423 lines
16 KiB
Python
"""Runnable check for SAFETY-93: the maintenance charge end hands back
|
|
smoothly, with no grid charging at a high house. `python3 test_safety93.py`
|
|
|
|
At N=2, when battery A's maintenance charge ended, its watts stayed in the
|
|
loop's site target and landed on B: B went 0 -> -1600 W in one cycle and
|
|
charged from the grid with the house at 6000 W (SAFETY-85 rig test, check 4).
|
|
Now the loop starts from what B stood at plus A's new command. Drives the
|
|
REAL Controller on a lagging battery (10 s, 2 s meter, as on the rig).
|
|
DECISIONS/SAFETY-93-charge-end.md.
|
|
|
|
1. The rig case: A charges 2000 W late in a quarter with the house at 6000 W,
|
|
then its charge ends (HOLD).
|
|
2. The same at a low house.
|
|
3. The charge ends another way: a timeout (abort) and the battery switched
|
|
Off.
|
|
4. The charge ends while the guard has suspended it (A gives under the loop).
|
|
5. N=1 is untouched; a missed poll on the end cycle; the rule at its
|
|
boundaries.
|
|
|
|
Mutations (MUTATION.md, purged copies) are listed in the pull request.
|
|
"""
|
|
|
|
import hashlib
|
|
import logging
|
|
import math
|
|
import os
|
|
import sys
|
|
from datetime import timedelta
|
|
|
|
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
|
|
from app import packmode # noqa: E402
|
|
from app.main import Controller # noqa: E402
|
|
from app.maintenance import (CHARGE_TIMEOUT_MIN, HOLD, IDLE, # noqa: E402
|
|
MaintResult)
|
|
from app.store import unit_key # noqa: E402
|
|
from named_absence import first, present # noqa: E402
|
|
|
|
fails = []
|
|
A = "inverter-a"
|
|
SLEW = 400.0 # the rig's slew_w
|
|
AIM = 3500 - 50 # peak_cap_w - peak_margin_w
|
|
OPT0 = dict(s84.OPT)
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
def lagging(fn, lag=10.0):
|
|
"""plantsim's battery (800 W/s ramp, `lag` s first-order lag) at the
|
|
live 2 s meter tick (test_safety85.py part 7)."""
|
|
real_p, real_t = rig._Hass.power, rig.TICK
|
|
|
|
def power(self, u):
|
|
st = self.__dict__.setdefault("_plant", {})
|
|
a, tgt, tl = st.get(u, (0.0, 0.0, self.t))
|
|
for _ in range(int(round((self.t - tl) / 0.1))):
|
|
d = self.cmd[u] - tgt
|
|
tgt += math.copysign(min(abs(d), 80.0), d)
|
|
a += (tgt - a) * (1 - math.exp(-0.1 / lag))
|
|
st[u] = (a, tgt, self.t)
|
|
return float(round(a))
|
|
rig._Hass.power, rig.TICK = power, 2.0
|
|
try:
|
|
return fn()
|
|
finally:
|
|
rig._Hass.power, rig.TICK = real_p, real_t
|
|
|
|
|
|
def run(n, house, end, *, step=None, dur=None, store=None, extra=None,
|
|
before=None):
|
|
"""A charging in a cheap window from the first cycle (quarters start at
|
|
0, 900, 1800 s); the house is 1200 W, then `house` from `step`; A reads
|
|
99 % (its charge target) from `end`. Returns (rows, controller)."""
|
|
step = end if step is None else step
|
|
s84.OPT = dict(OPT0, peak_cap_w=3500)
|
|
feed = s84.feed(True)
|
|
|
|
def both(c, t):
|
|
if before is not None:
|
|
before(c, t)
|
|
feed(c, t)
|
|
try:
|
|
return lagging(lambda: s84.run(
|
|
n, load=lambda t: house if t >= step else 1200.0, cheap=True,
|
|
store=store or s79.store_charging(n),
|
|
soc=lambda u, t: (99.0 if t >= end else 90.0) if u == "a"
|
|
else 60.0, dur=dur or end + 300.0, before=both,
|
|
extra=dict({"maintenance_hold_min": 600}, **(extra or {}))))
|
|
finally:
|
|
s84.OPT = OPT0
|
|
|
|
|
|
def cmd(r):
|
|
return dict(r[2])
|
|
|
|
|
|
def moves(rows, u, lo, hi):
|
|
"""The largest change in battery `u`'s command between cycles in
|
|
(lo, hi]."""
|
|
return max(abs(cmd(b)[u] - cmd(a)[u]) for a, b in zip(rows, rows[1:])
|
|
if lo < b[0] <= hi)
|
|
|
|
|
|
def charging_hi(rows, lo, hi, house_at):
|
|
"""Cycles in [lo, hi] where a battery charges while the house is above
|
|
the peak aim."""
|
|
return sum(1 for r in rows if lo <= r[0] <= hi and house_at(r[0]) > AIM
|
|
and min(cmd(r).values()) < 0.0)
|
|
|
|
|
|
# None on a tree without the fix (f3e409c), where this file runs red.
|
|
no_fix = getattr(Controller, "_reseed_charge_end", None)
|
|
|
|
|
|
def without(fn):
|
|
"""Run `fn` with the SAFETY-93 reseed taken out."""
|
|
if no_fix is None:
|
|
return fn()
|
|
Controller._reseed_charge_end = lambda self, ended: None
|
|
try:
|
|
return fn()
|
|
finally:
|
|
Controller._reseed_charge_end = no_fix
|
|
|
|
|
|
def phase_at(rows, t):
|
|
return first((r[-1][3][0] for r in rows if r[0] == t), f"row {t}")
|
|
|
|
|
|
def at(rows, t, u):
|
|
"""Battery `u`'s command at `t` s."""
|
|
r = first((r for r in rows if r[0] == t), f"row {t}")
|
|
return dict(r[2])[u] if present(r) else r
|
|
|
|
|
|
# ============================================================================ #
|
|
print("1. the rig case: A charges 2000 W late in a quarter, house 6000 W")
|
|
# House 1200 W until 1760 s, then 6000 W; the quarter (900-1800 s) has room,
|
|
# so the charge stays at 2000 W. A is full at 1780 s. Graded to 1798 s: the
|
|
# guard grades a new quarter from 1800 s.
|
|
R1 = run(2, 6000.0, 1780.0, step=1760.0)
|
|
R1_old = without(lambda: run(2, 6000.0, 1780.0, step=1760.0))
|
|
|
|
|
|
def hi1(t):
|
|
return 6000.0 if t >= 1760.0 else 1200.0
|
|
|
|
|
|
for name, (rows, _c) in (("now", R1), ("without the fix", R1_old)):
|
|
print(f" {name}: " + ", ".join(
|
|
f"{r[0]:.0f}s A {cmd(r)['a']:.0f} B {cmd(r)['b']:.0f}"
|
|
for r in rows if 1776 <= r[0] <= 1786))
|
|
_r, _ = R1
|
|
check("the case is the rig's: A charges 2000 W with the house at 6000 W the "
|
|
"cycle before its charge ends, then holds",
|
|
at(_r, 1778.0, "a") == -2000.0
|
|
and phase_at(_r, 1778.0) == "charge" and phase_at(_r, 1780.0) == HOLD)
|
|
check("⭐ no battery charges from the grid after the charge ends (house above "
|
|
"the aim)", charging_hi(_r, 1780, 1798, hi1) == 0)
|
|
check("⭐ B moves at most one slew step a cycle across the charge end, and A "
|
|
"after its HOLD pin", moves(_r, "b", 1776, 1798) <= SLEW
|
|
and moves(_r, "a", 1780, 1798) <= SLEW)
|
|
check("...and B gives toward the house at once (+400 W at the end cycle)",
|
|
at(_r, 1780.0, "b") == SLEW)
|
|
check("non-vacuous: without the fix B jumps 0 -> -1600 W and charges from "
|
|
"the grid for 4 cycles", moves(R1_old[0], "b", 1776, 1798) >= 1600
|
|
and charging_hi(R1_old[0], 1780, 1798, hi1) >= 4)
|
|
|
|
# ============================================================================ #
|
|
print("\n2. the same at a low house")
|
|
R2 = run(2, 1200.0, 1300.0)
|
|
R2_old = without(lambda: run(2, 1200.0, 1300.0))
|
|
print(f" B's largest move: {moves(R2[0], 'b', 1296, 1600):.0f} W, without "
|
|
f"the fix {moves(R2_old[0], 'b', 1296, 1600):.0f} W; B's least "
|
|
f"command after the end "
|
|
f"{min(cmd(r)['b'] for r in R2[0] if r[0] >= 1300):.0f}"
|
|
f" W, without {min(cmd(r)['b'] for r in R2_old[0] if r[0] >= 1300):.0f}"
|
|
" W")
|
|
check("⭐ house 1200 W: B moves at most one slew step a cycle and never "
|
|
"charges from the grid after A's charge ends",
|
|
moves(R2[0], "b", 1296, 1600) <= SLEW
|
|
and min(cmd(r)["b"] for r in R2[0] if r[0] >= 1300) >= 0.0)
|
|
check("non-vacuous: without the fix B jumps to -1600 W (2800 W from the grid)",
|
|
min(cmd(r)["b"] for r in R2_old[0] if r[0] >= 1300) <= -1600.0)
|
|
|
|
# ============================================================================ #
|
|
print("\n3. the charge ends another way")
|
|
|
|
|
|
def timed_out(end):
|
|
"""A store whose charge passes CHARGE_TIMEOUT_MIN at `end` s."""
|
|
s = s79.store_charging(2)
|
|
s.data[unit_key(A, "phase_started")] = (
|
|
rig.BASE + timedelta(seconds=end - CHARGE_TIMEOUT_MIN * 60 - 1)
|
|
).isoformat()
|
|
return s
|
|
|
|
|
|
def b_done():
|
|
"""A charging; B finished its monthly cycle, so it starts none here."""
|
|
s = s79.store_charging(2)
|
|
s.data[unit_key("inverter-b", "last_completed")] = rig.BASE.isoformat()
|
|
return s
|
|
|
|
|
|
def switch_off(at):
|
|
"""Battery A switched Off at `at` s (OPT-11 modes on)."""
|
|
def before(c, t):
|
|
if t >= at:
|
|
c.store.data[unit_key(A, packmode.USE_KEY)] = False
|
|
return before
|
|
|
|
|
|
ENDS = {
|
|
"timeout": lambda: run(2, 6000.0, 9999.0, step=1760.0, dur=2080.0,
|
|
store=timed_out(1780.0)),
|
|
"Off": lambda: run(2, 6000.0, 9999.0, step=1760.0, dur=2080.0,
|
|
store=b_done(), before=switch_off(1780.0),
|
|
extra={"battery_modes_enabled": True}),
|
|
}
|
|
for k, fn in ENDS.items():
|
|
rows, c = fn()
|
|
old = without(fn)[0]
|
|
print(f" {k}: A {[cmd(r)['a'] for r in rows if 1776 <= r[0] <= 1784]}"
|
|
f" B {[cmd(r)['b'] for r in rows if 1776 <= r[0] <= 1784]}; "
|
|
f"without the fix B "
|
|
f"{[cmd(r)['b'] for r in old if 1776 <= r[0] <= 1784]}")
|
|
check(f"{k}: the charge ends at 1780 s",
|
|
phase_at(rows, 1778.0) == "charge"
|
|
and phase_at(rows, 1780.0) == IDLE)
|
|
check(f"⭐ {k}: no battery charges from the grid after it, and B moves at "
|
|
f"most one slew step a cycle",
|
|
charging_hi(rows, 1780, 1798, hi1) == 0
|
|
and moves(rows, "b", 1776, 1798) <= SLEW)
|
|
check(f"non-vacuous ({k}): without the fix B jumps to -1600 W, from "
|
|
f"the grid", at(old, 1780.0, "b") == -1600.0
|
|
and charging_hi(old, 1780, 1798, hi1) >= 1)
|
|
|
|
# ============================================================================ #
|
|
print("\n4. the charge ends while the guard has suspended it")
|
|
# House 6000 W from 1500 s: the guard suspends the shaping, the loop has both
|
|
# batteries give. A is full at 1530 s; its HOLD pin takes it to 0 W.
|
|
R4 = run(2, 6000.0, 1530.0, step=1500.0)
|
|
R4_old = without(lambda: run(2, 6000.0, 1530.0, step=1500.0))
|
|
_a4 = at(R4[0], 1528.0, "a")
|
|
print(f" A gave {_a4:.0f} W before its HOLD; B's largest move "
|
|
f"{moves(R4[0], 'b', 1526, 1798):.0f} W, without the fix "
|
|
f"{moves(R4_old[0], 'b', 1526, 1798):.0f} W")
|
|
check("A gives under the loop before its charge ends",
|
|
present(_a4) and _a4 > SLEW and R4[1].maints[0].phase == HOLD)
|
|
check("⭐ B takes A's share at one slew step a cycle (A's own give stops "
|
|
"with its HOLD pin, as before)", moves(R4[0], "b", 1526, 1798) <= SLEW
|
|
and charging_hi(R4[0], 1530, 1798, lambda t: 6000.0) == 0)
|
|
check("non-vacuous: without the fix B jumps more than a slew step",
|
|
moves(R4_old[0], "b", 1526, 1798) > SLEW)
|
|
|
|
# ============================================================================ #
|
|
print("\n5. N=1 is untouched; a missed poll; the rule at its boundaries")
|
|
|
|
|
|
def digest(rows):
|
|
return hashlib.sha256(repr(rows).encode()).hexdigest()
|
|
|
|
|
|
_n1 = [digest(run(1, h, e, step=s)[0]) for h, e, s in
|
|
((6000.0, 1780.0, 1760.0), (1200.0, 1300.0, None))]
|
|
_n1_old = [digest(without(lambda: run(1, h, e, step=s))[0]) for h, e, s in
|
|
((6000.0, 1780.0, 1760.0), (1200.0, 1300.0, None))]
|
|
check("⭐ N=1, the charge ending at a high and a low house: every row "
|
|
"byte-identical with and without the reseed", _n1 == _n1_old)
|
|
|
|
if no_fix is None:
|
|
print(f"\n{len(fails)} FAILED (no _reseed_charge_end): " + ", ".join(fails))
|
|
sys.exit(1)
|
|
|
|
|
|
def missed_poll(at_t):
|
|
"""Battery B's power reading is missing on the cycle at `at_t` s."""
|
|
def before(c, t):
|
|
if not getattr(c.hass, "_s93", False):
|
|
read = c.hass.number
|
|
|
|
async def number(e, invert=False):
|
|
if e == "sensor.batt_b" and c.hass.t == at_t:
|
|
return None
|
|
return await read(e, invert)
|
|
c.hass.number, c.hass._s93 = number, True
|
|
return before
|
|
|
|
|
|
R6 = run(2, 6000.0, 1780.0, step=1760.0, before=missed_poll(1780.0))
|
|
R6_old = without(lambda: run(2, 6000.0, 1780.0, step=1760.0,
|
|
before=missed_poll(1780.0)))
|
|
print(f" a missed poll on the end cycle: B "
|
|
f"{[cmd(r)['b'] for r in R6[0] if 1778 <= r[0] <= 1786]}, without "
|
|
f"the fix {[cmd(r)['b'] for r in R6_old[0] if 1778 <= r[0] <= 1786]}")
|
|
check("a missed poll on the cycle the charge ends: the loop holds that cycle",
|
|
phase_at(R6[0], 1780.0) == HOLD
|
|
and "inputs missing" in str(first(r[6] for r in R6[0] if r[0] == 1780.0)))
|
|
check("⭐ ...and the end still reaches the loop on the next cycle: no grid "
|
|
"charge (A's charge stands through the held cycle), one slew step a "
|
|
"cycle", charging_hi(R6[0], 1782, 1798, hi1) == 0
|
|
and moves(R6[0], "b", 1776, 1798) <= SLEW)
|
|
check("non-vacuous: without the fix B charges from the grid",
|
|
charging_hi(R6_old[0], 1782, 1798, hi1) >= 1)
|
|
_c = R1[1]
|
|
|
|
|
|
def reseeded(target, shares, res, ended=frozenset({A})):
|
|
"""`_reseed_charge_end` on the rig's controller, A's charge ended and
|
|
its result `res`: (target, i_w)."""
|
|
_c.target, _c.i_w, _c._charge_ended = target, 123.0, ended
|
|
for inv in _c.inverters:
|
|
inv.target = shares[inv.name[-1]]
|
|
_c._reseed_charge_end([res, _free])
|
|
return _c.target, _c.i_w
|
|
|
|
|
|
_pin = MaintResult(HOLD, True, 0.0, False, [])
|
|
_free = MaintResult(IDLE, False, None, False, [])
|
|
check("a HOLD pin: the start is B's command plus the pin, integrator with it",
|
|
reseeded(-2000.0, {"a": -2000.0, "b": 0.0}, _pin) == (0.0, 0.0)
|
|
and reseeded(2770.0, {"a": 1385.0, "b": 1385.0}, _pin)
|
|
== (1385.0, 1385.0) and _c._charge_ended == frozenset())
|
|
check("a free battery: its charge leaves the start, its discharge stays",
|
|
reseeded(-2000.0, {"a": -2000.0, "b": 0.0}, _free) == (0.0, 0.0)
|
|
and reseeded(2770.0, {"a": 1385.0, "b": 1385.0}, _free)
|
|
== (2770.0, 2770.0))
|
|
check("no charge ended: nothing moves (the integrator either)",
|
|
reseeded(-2000.0, {"a": -2000.0, "b": 0.0}, _pin, frozenset())
|
|
== (-2000.0, 123.0))
|
|
_n1c = run(1, 6000.0, 1780.0, step=1760.0)[1]
|
|
_n1c.target, _n1c.i_w = -2000.0, 5.0
|
|
_n1c.inverters[0].target, _n1c._charge_ended = -2000.0, frozenset({A})
|
|
_n1c._reseed_charge_end([_pin])
|
|
check("N=1: nothing moves", (_n1c.target, _n1c.i_w) == (-2000.0, 5.0))
|
|
_seen = []
|
|
|
|
|
|
def _spy(self, maint_results):
|
|
_seen.append(sorted(self._charge_ended))
|
|
return no_fix(self, maint_results)
|
|
|
|
|
|
Controller._reseed_charge_end = _spy
|
|
try:
|
|
run(2, 6000.0, 1780.0, step=1760.0, dur=2400.0,
|
|
extra={"maintenance_hold_min": 5})
|
|
finally:
|
|
Controller._reseed_charge_end = no_fix
|
|
check("it fires once, at the charge end: not at the HOLD end or on any "
|
|
"other cycle", [x for x in _seen if x] == [[A]]
|
|
and len(_seen) > 300)
|
|
|
|
# ============================================================================ #
|
|
print("\n6. A still in CHARGE but no longer placed: stale link, refusal")
|
|
|
|
|
|
def link_drop(at):
|
|
"""A's writes rejected from `at` s, so it is dropped as a stale link."""
|
|
def before(c, t):
|
|
a = c.inverters[0]
|
|
if t >= at and a.rejected_since is None:
|
|
a.rejected_since = rig.BASE + timedelta(seconds=at)
|
|
return before
|
|
|
|
|
|
def refuses(at):
|
|
"""A stops charging from `at` s (reads 0 W); SAFETY-73 drops it."""
|
|
def before(c, t):
|
|
if not getattr(c.hass, "_s93r", False):
|
|
real = c.hass.power
|
|
|
|
def power(u):
|
|
p = real(u)
|
|
return 0.0 if u == "a" and c.hass.t >= at and p < 0 else p
|
|
c.hass.power, c.hass._s93r = power, True
|
|
return before
|
|
|
|
|
|
for k, hook in (("stale link", link_drop), ("refusal", refuses)):
|
|
def fn(hook=hook):
|
|
return run(2, 1200.0, 9999.0, dur=1500.0, before=hook(1300.0))
|
|
rows, _c6 = fn()
|
|
old = without(fn)[0]
|
|
drop = first((r[0] for r in rows if r[0] >= 1300
|
|
and "dropped inverter-a" in str(r[7])), "A dropped")
|
|
print(f" {k}: A dropped at {drop:.0f} s; B "
|
|
f"{[cmd(r)['b'] for r in rows if drop - 2 <= r[0] <= drop + 4]}, "
|
|
f"without the fix {[cmd(r)['b'] for r in old if drop - 2 <= r[0] <= drop + 4]}")
|
|
check(f"⭐ {k}: A is dropped while still in CHARGE, and B never charges "
|
|
f"from the grid and moves at most one slew step a cycle",
|
|
phase_at(rows, drop) == "charge"
|
|
and min(cmd(r)["b"] for r in rows if r[0] >= 1300) >= 0.0
|
|
and moves(rows, "b", 1296, 1500) <= SLEW)
|
|
check(f"non-vacuous ({k}): without the fix B jumps to -1600 W",
|
|
min(cmd(r)["b"] for r in old if r[0] >= 1300) <= -1600.0)
|
|
|
|
_seen.clear()
|
|
Controller._reseed_charge_end = _spy
|
|
try:
|
|
run(2, 6000.0, 1530.0, step=1500.0)
|
|
finally:
|
|
Controller._reseed_charge_end = no_fix
|
|
check("the guard suspending the charge is no end: it fires once, at the HOLD",
|
|
[x for x in _seen if x] == [[A]])
|
|
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED: " + ", ".join(fails))
|
|
sys.exit(1)
|
|
print("all SAFETY-93 checks passed")
|