glenn schrooyenandClaude Opus 5.5
4707d37019
F1: test_safety95 checks that a dropped draining battery is this rule's (stale link, N=3, 4000 W: grid import 676 W vs base 786 W); kills R2. F2: _charge_track tested directly with a contradicting drain (SAFETY-91's pin as a floor against a closed pack); kills R3. Today a drain contradicts only through a refusal; the clause stays, as a contradiction sends 0 W. F3: a drain ends on a refusal only when the battery will not give (refused > 0, as _held_back()); one that will not take still drains. Tested both signs; kills R8. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
579 lines
24 KiB
Python
579 lines
24 KiB
Python
"""Runnable check for SAFETY-95: a maintenance drain that ends hands back
|
|
smoothly at N>=2. `python3 test_safety95.py`
|
|
|
|
At N=2, when battery A's drain ended (A switched Off, say) with the house at
|
|
1200 W, the drain's floor stayed in the loop's site target and landed on B:
|
|
B went 0 -> +1600 W in one cycle and exported. Now the loop starts from what
|
|
the others stood at plus A's new command, raised to the house's own point
|
|
when the house needs more (never above last cycle's start), and counts every
|
|
battery at its command while A settles. Drives the REAL Controller on a
|
|
lagging battery (10 s, 2 s meter, as on the rig), at N=2 and N=3.
|
|
DECISIONS/SAFETY-95-drain-end.md.
|
|
|
|
1. Every drain end (complete -> charge, Off, timeout, operator abort, stale
|
|
link, refusal) at houses of 1200 W and 4000 W and under a solar surplus.
|
|
2. The peak guard still binds across a drain end.
|
|
3. A drain end in the same cycle as another battery's drop; a missed poll.
|
|
4. N=1 is untouched; the rule at its boundaries.
|
|
|
|
Mutations (MUTATION.md, purged copies) are listed in the pull request.
|
|
"""
|
|
|
|
import asyncio
|
|
import functools
|
|
import hashlib
|
|
import logging
|
|
import math
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
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
|
|
from app import main as app_main, packmode # noqa: E402
|
|
from app.arbiter import Claim, P_MAINTENANCE, P_SAFETY # noqa: E402
|
|
from app.main import Controller # noqa: E402
|
|
from app.maintenance import (CHARGE, DRAIN, DRAIN_TIMEOUT_MIN, # noqa: E402
|
|
HOLD, IDLE, MaintResult)
|
|
from app.store import unit_key # noqa: E402
|
|
|
|
fails = []
|
|
SLEW = 400.0 # the rig's slew_w
|
|
TOL = SLEW / 4 # the loop's own overshoot on a lagging battery
|
|
AIM = -10.0 # the rig's target_grid_w
|
|
T = 300.0 # the drain ends here
|
|
M = {"maintenance_enabled": True, "maintenance_start_hour": 11}
|
|
A = "inverter-a"
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
# None on a tree without the fix, where this file prints the base numbers
|
|
# and runs red.
|
|
fixed = getattr(Controller, "_reseed_drain_end", None)
|
|
real_grid = getattr(Controller, "_drain_grid_w", None)
|
|
real_drop = Controller._reseed_drop
|
|
|
|
|
|
def _base_drop(self, live):
|
|
"""SAFETY-96's drop reseed as it was: a draining battery is not
|
|
skipped."""
|
|
keep, self._drain_ended = self._drain_ended, frozenset()
|
|
try:
|
|
real_drop(self, live)
|
|
finally:
|
|
self._drain_ended = keep
|
|
|
|
|
|
def without(fn):
|
|
"""Run `fn` with SAFETY-95 taken out (the base behaviour)."""
|
|
if fixed is None:
|
|
return fn()
|
|
Controller._reseed_drain_end = lambda self, r, live: setattr(
|
|
self, "_drain_ended", frozenset())
|
|
Controller._drain_grid_w = lambda self, live: float(self.grid)
|
|
Controller._reseed_drop = _base_drop
|
|
try:
|
|
return fn()
|
|
finally:
|
|
Controller._reseed_drain_end = fixed
|
|
Controller._drain_grid_w = real_grid
|
|
Controller._reseed_drop = real_drop
|
|
|
|
|
|
def options(n, extra):
|
|
o = rig._options(n, extra)
|
|
if n > 1:
|
|
o["inverters"] = [
|
|
{"name": f"inverter-{u}", "setpoint_entity": f"number.sp_{u}",
|
|
"soc_entity": f"sensor.soc_{u}", "batt_entity": f"sensor.batt_{u}",
|
|
"max_w": 2500, "bms_soc_floor": 10} for u in "abc"[:n]]
|
|
return o
|
|
|
|
|
|
def run(n, house, *, store, extra=None, soc=None, reject=None, dead=None,
|
|
before=None, dur=T + 120.0, lag=10.0, p1=False, back=None):
|
|
"""Rows of one run at N=`n` on plantsim's battery (800 W/s ramp, `lag`
|
|
s first-order lag) at a 2 s meter tick. Writes to the units in `reject`
|
|
(name -> from s) are refused, until `back` s; `dead` (u, from s[, to
|
|
s]): u gives nothing from then on (until `to`). Each row: (t, the
|
|
add-on's commands, site target, grid, phases, loop reason, peak floor).
|
|
opt11_identity's cycle, with a third battery."""
|
|
units = tuple("abc"[:n])
|
|
soc = soc or (lambda u, t: 50.0)
|
|
|
|
def refused(u, t):
|
|
return bool(reject) and u in reject and t >= reject[u] and (
|
|
back is None or t < back)
|
|
|
|
class Hass(rig._Hass):
|
|
def power(self, u):
|
|
st = self.__dict__.setdefault("_plant", {})
|
|
a, tgt, tl = st.get(u, (0.0, 0.0, self.t))
|
|
for k in range(int(round((self.t - tl) / 0.1))):
|
|
now = tl + (k + 1) * 0.1
|
|
c = self.cmd[u]
|
|
if dead is not None and u == dead[0] and dead[1] <= now < (
|
|
dead[2] if len(dead) > 2 else math.inf):
|
|
c = 0.0
|
|
d = c - 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))
|
|
|
|
async def set_number(self, e, v):
|
|
if refused(e[-1], self.t):
|
|
return False
|
|
self.cmd[e[-1]] = float(v)
|
|
return True
|
|
|
|
opts = options(n, dict(M, **(s79.P1 if p1 else {}), **(extra or {})))
|
|
db = os.path.join(tempfile.mkdtemp(prefix="s95-"), "tariff.db")
|
|
|
|
async def go():
|
|
hass = Hass(units, soc, lambda u, t: False,
|
|
lambda t: house(t) if callable(house) else house)
|
|
c = Controller(opts, hass, store, rig._Pub(),
|
|
stop=asyncio.Event(), peak_db_path=db)
|
|
for i in c.inverters:
|
|
i.dev_min, i.dev_max = -5000.0, 5000.0
|
|
c.auto = True
|
|
real_dt, clock = app_main.datetime, [0.0]
|
|
|
|
class _DT(real_dt):
|
|
@classmethod
|
|
def now(cls, tz=None):
|
|
return rig.BASE + timedelta(seconds=clock[0])
|
|
app_main.datetime = _DT
|
|
rows = []
|
|
try:
|
|
t = 0.0
|
|
while t < dur:
|
|
hass.t = clock[0] = t
|
|
for inv in c.inverters:
|
|
u = inv.name[-1]
|
|
if refused(u, t) and inv.rejected_since is None:
|
|
inv.rejected_since = rig.BASE + timedelta(seconds=t)
|
|
if before is not None:
|
|
r = before(c, t)
|
|
if asyncio.iscoroutine(r):
|
|
await r
|
|
if p1:
|
|
s79._feed(c, t)
|
|
await c.read_inputs(now_mono=t)
|
|
await c.cycle()
|
|
g = c.peak_guard
|
|
rows.append((t, tuple(i.target for i in c.inverters),
|
|
c.target,
|
|
hass.grid(), tuple(m.phase for m in c.maints),
|
|
c.loop_reason,
|
|
None if g is None or not g.engaged
|
|
else g.floor_w))
|
|
t += 2.0
|
|
finally:
|
|
app_main.datetime = real_dt
|
|
return rows, c
|
|
return asyncio.run(go())
|
|
|
|
|
|
def store(n, **a):
|
|
"""A due for its monthly cycle (it drains from the first cycle); the
|
|
others finished theirs, so none starts here."""
|
|
s = rig._Store()
|
|
if n == 1:
|
|
for k, v in a.items():
|
|
s.data[k] = v
|
|
return s
|
|
for u in "bc"[:n - 1]:
|
|
s.data[unit_key(f"inverter-{u}", "last_completed")] = \
|
|
rig.BASE.isoformat()
|
|
for k, v in a.items():
|
|
s.data[unit_key(A, k)] = v
|
|
return s
|
|
|
|
|
|
def timed_store(n):
|
|
"""A's drain passes DRAIN_TIMEOUT_MIN at T; its last start was at 0 s,
|
|
so the abort's cooldown keeps it from starting again."""
|
|
return store(n, phase=DRAIN, phase_started=(
|
|
rig.BASE + timedelta(seconds=T - DRAIN_TIMEOUT_MIN * 60 - 1)
|
|
).isoformat(), last_start_attempt=rig.BASE.isoformat())
|
|
|
|
|
|
def off(c, t):
|
|
if t >= T:
|
|
c.store.data[unit_key(A, packmode.USE_KEY)] = False
|
|
|
|
|
|
async def abort(c, t):
|
|
if t == T:
|
|
await c.handle_action("maint_abort")
|
|
|
|
|
|
ENDS = {
|
|
"complete": lambda n, h, **k: run(
|
|
n, h, store=store(n), soc=lambda u, t:
|
|
5.0 if (u == "a" and t >= T) else 50.0, **k),
|
|
"Off": lambda n, h, **k: run(n, h, store=store(n), before=off,
|
|
extra={"battery_modes_enabled": True}, **k),
|
|
"timeout": lambda n, h, **k: run(n, h, store=timed_store(n), **k),
|
|
"abort": lambda n, h, **k: run(n, h, store=store(n), before=abort, **k),
|
|
# A's link goes stale; its board's watchdog sends it to 0 W after 25 s.
|
|
"stale link": lambda n, h, **k: run(n, h, store=store(n),
|
|
reject={"a": T}, dead=("a", T + 25.0),
|
|
dur=T + 200.0, **k),
|
|
# A gives nothing from T; SAFETY-73 learns it ~55 s later.
|
|
"refusal": lambda n, h, **k: run(n, h, store=store(n), dead=("a", T),
|
|
dur=T + 300.0, **k),
|
|
}
|
|
HOUSES = (1200.0, 4000.0, -1500.0)
|
|
|
|
|
|
def end_at(rows):
|
|
"""The first cycle A is not draining, or its command left the drain;
|
|
None if its drain never ended."""
|
|
return next((r[0] for r in rows if r[0] >= T and (
|
|
r[4][0] != DRAIN or r[1][0] < 2000.0)), None)
|
|
|
|
|
|
def measure(rows, n, house):
|
|
"""From the end on: the site target's largest step after the end
|
|
cycle (the slew's subject); the commanded export in the 20 s after it (the site command over
|
|
what the house needs, or over what the others stood at); the largest
|
|
grid import; the export beyond what the site exported before the end;
|
|
the largest grid step; the others' largest command step."""
|
|
te = end_at(rows)
|
|
if te is None: # no end: every measure is NaN, so every check FAILs
|
|
return dict(end=math.nan, site_step=math.nan, cmd_export=math.nan,
|
|
imp=math.nan, exp=math.nan, gstep=math.nan,
|
|
ostep=math.nan, rows=rows)
|
|
pre = next((r for r in reversed(rows) if r[0] < te), rows[0])
|
|
seg = [r for r in rows if r[0] >= te - 2.0]
|
|
aft = [r for r in rows if te <= r[0] <= te + 120.0]
|
|
others0 = sum(pre[1][1:])
|
|
site = [r[2] for r in seg]
|
|
return dict(
|
|
end=te,
|
|
site_step=max(abs(b - a) for a, b in zip(site[1:], site[2:])),
|
|
cmd_export=max(sum(r[1]) - max(house - AIM, others0) for r in aft
|
|
if r[0] <= te + 20.0),
|
|
imp=max(r[3] for r in aft),
|
|
exp=min(min(pre[3], AIM) - r[3] for r in aft),
|
|
gstep=max(abs(b[3] - a[3]) for a, b in zip(seg, seg[1:])),
|
|
ostep=max(abs(b[1][i] - a[1][i]) for a, b in zip(seg, seg[1:])
|
|
for i in range(1, n)),
|
|
rows=rows)
|
|
|
|
|
|
TABLE = []
|
|
|
|
|
|
def case(kind, n, house):
|
|
now = measure(ENDS[kind](n, house)[0], n, house)
|
|
old = measure(without(lambda: ENDS[kind](n, house))[0], n, house)
|
|
TABLE.append((kind, n, house, now, old))
|
|
return now, old
|
|
|
|
|
|
# ============================================================================ #
|
|
print("1. every drain end, N=2 and N=3, houses 1200 W, 4000 W and -1500 W")
|
|
for kind in ENDS:
|
|
for n in (2, 3):
|
|
for house in HOUSES:
|
|
now, old = case(kind, n, house)
|
|
tag = f"{kind}, N={n}, house {house:.0f} W"
|
|
# Abort zeroes the site (operator); a stale link's start is
|
|
# SAFETY-96's drop. Those are graded against base.
|
|
own = kind not in ("abort", "stale link")
|
|
check(f"{'⭐' if own else ' '} {tag}: no battery gives into an "
|
|
f"export after the end (commanded export "
|
|
f"{now['cmd_export']:.0f} W, base {old['cmd_export']:.0f}) "
|
|
f"and the site target moves at most one slew step a cycle "
|
|
f"after it ({now['site_step']:.0f} W, base "
|
|
f"{old['site_step']:.0f})",
|
|
now["cmd_export"] <= (TOL if own else max(
|
|
TOL, old["cmd_export"] + TOL))
|
|
and now["site_step"] <= SLEW)
|
|
check(f" {tag}: the grid imports no more than base (+50 W) "
|
|
f"({now['imp']:.0f} W, base {old['imp']:.0f}) and exports "
|
|
f"no more than it did before the end, or base (+50 W) "
|
|
f"({now['exp']:.0f} W more, base {old['exp']:.0f})",
|
|
now["imp"] <= old["imp"] + 50.0
|
|
and now["exp"] <= max(0.0, old["exp"]) + 50.0)
|
|
|
|
_by = {(k, n, h): (a, b) for k, n, h, a, b in TABLE}
|
|
for kind in ("complete", "Off", "timeout", "refusal"):
|
|
_now, _old = _by[(kind, 2, 1200.0)]
|
|
check(f"non-vacuous ({kind}, N=2, 1200 W): without the fix B steps "
|
|
f"more than a slew step into an export",
|
|
_old["cmd_export"] >= 300.0 and _old["ostep"] > SLEW
|
|
and _now["ostep"] < _old["ostep"])
|
|
_now, _old = _by[(kind, 2, -1500.0)]
|
|
check(f"non-vacuous ({kind}, solar surplus): without the fix B gives "
|
|
f"into the surplus", _old["cmd_export"] >= 1000.0
|
|
and _now["cmd_export"] <= TOL)
|
|
_now, _old = _by[("complete", 2, 1200.0)]
|
|
check("⭐ the rig case (A's drain completes at 1200 W): B covers the house "
|
|
"at once (0 -> 1210 W, not 1600), and the grid stays within 50 W of "
|
|
"its aim while A settles (base: +559 W import, B down to 230 W)",
|
|
[r[1][1] for r in _now["rows"] if r[0] == _now["end"]] == [1210.0]
|
|
and _now["imp"] <= 50.0 and _old["imp"] > 400.0)
|
|
_now, _old = _by[("complete", 3, 4000.0)]
|
|
check("N=3 at 4000 W: the others take A's share at once, as before (the "
|
|
"site does not drop), so the grid does not import",
|
|
_now["imp"] <= 50.0 and abs(_now["imp"] - _old["imp"]) <= 50.0)
|
|
_now, _old = _by[("stale link", 3, 4000.0)]
|
|
check(f"⭐ stale link, N=3, 4000 W: a dropped draining battery is this "
|
|
f"rule's, not the link-drop reseed's, so the grid imports at least "
|
|
f"50 W less than base ({_now['imp']:.0f} W, base {_old['imp']:.0f})",
|
|
_now["imp"] <= _old["imp"] - 50.0)
|
|
|
|
# ============================================================================ #
|
|
print("\n2. the peak guard still binds across a drain end")
|
|
_g = ENDS["Off"](2, 4000.0, p1=True, dur=T + 120.0)[0]
|
|
_gs = [r for r in _g if r[0] >= T - 20 and r[6] is not None]
|
|
print(f" guard engaged on {len(_gs)} cycles around the end; site "
|
|
f"{[sum(r[1]) for r in _g if T - 4 <= r[0] <= T + 8]}")
|
|
check("⭐ house 4000 W over a 2500 W limit, A switched Off mid-drain: on "
|
|
"every cycle the guard is engaged, the site gives at least its floor "
|
|
"(or every battery is at its rating)",
|
|
_gs and all(sum(r[1]) >= min(r[6], 2500.0 * 1) - 1.0 for r in _gs))
|
|
|
|
# ============================================================================ #
|
|
print("\n3. a drain end with another edge in the same cycle; a missed poll")
|
|
|
|
|
|
def c_drops(c, t):
|
|
"""C's writes are refused from T - 30 s, so it drops at T, the cycle A
|
|
is switched Off; its board goes to 0 W at once (T - 30 s)."""
|
|
off(c, t)
|
|
|
|
|
|
_e3 = run(3, 4000.0, store=store(3), before=c_drops, reject={"c": T - 30.0},
|
|
dead=("c", T - 30.0), extra={"battery_modes_enabled": True})[0]
|
|
_e3m = measure(_e3, 3, 4000.0)
|
|
print(f" A Off as C drops, house 4000 W: B "
|
|
f"{[r[1][1] for r in _e3 if T - 4 <= r[0] <= T + 10]}")
|
|
check("⭐ A switched Off on the cycle C drops (N=3, 4000 W): B takes over "
|
|
"with no export and the site moves one slew step a cycle after",
|
|
_e3m["cmd_export"] <= TOL and _e3m["site_step"] <= SLEW
|
|
and _e3m["end"] == T)
|
|
|
|
|
|
def c_back(c, t):
|
|
"""C's link (stale since T - 60 s, its board holding 0 W) answers again
|
|
at T, the cycle A is switched Off."""
|
|
off(c, t)
|
|
if t == T:
|
|
c.inverters[2].note_write(True, rig.BASE + timedelta(seconds=t))
|
|
|
|
|
|
_r3 = run(3, 1200.0, store=store(3), before=c_back, reject={"c": T - 60.0},
|
|
back=T, dead=("c", T - 60.0, T), extra={"battery_modes_enabled": True})[0]
|
|
_r3m = measure(_r3, 3, 1200.0)
|
|
print(f" A Off as C comes back, house 1200 W: B, C "
|
|
f"{[r[1][1:] for r in _r3 if T - 4 <= r[0] <= T + 6]}")
|
|
check("⭐ A switched Off on the cycle C comes back (N=3, 1200 W): no "
|
|
"export, one slew step a cycle after",
|
|
_r3m["cmd_export"] <= TOL and _r3m["site_step"] <= SLEW
|
|
and _r3m["end"] == T)
|
|
|
|
|
|
def missed_poll(at_t):
|
|
"""Battery B's power reading is missing on the cycle at `at_t` s."""
|
|
def before(c, t):
|
|
off(c, t)
|
|
if not getattr(c.hass, "_s95", 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._s95 = number, True
|
|
return before
|
|
|
|
|
|
_mp = run(2, 1200.0, store=store(2), before=missed_poll(T),
|
|
extra={"battery_modes_enabled": True})[0]
|
|
_mpo = without(lambda: run(2, 1200.0, store=store(2), before=missed_poll(T),
|
|
extra={"battery_modes_enabled": True}))[0]
|
|
print(f" a missed poll on the end cycle: B "
|
|
f"{[r[1][1] for r in _mp if T - 2 <= r[0] <= T + 8]}, without the fix "
|
|
f"{[r[1][1] for r in _mpo if T - 2 <= r[0] <= T + 8]}")
|
|
check("⭐ a missed poll on the cycle A is switched Off: the end still "
|
|
"reaches the loop on the next cycle (B covers the house, no export)",
|
|
max(sum(r[1]) for r in _mp if r[0] > T) <= 1200.0 - AIM + TOL
|
|
and max(sum(r[1]) for r in _mpo if r[0] > T) >= 1500.0)
|
|
|
|
# ============================================================================ #
|
|
print("\n4. N=1 is untouched; the rule at its boundaries")
|
|
|
|
|
|
def digest(rows):
|
|
return hashlib.sha256(repr(rows).encode()).hexdigest()
|
|
|
|
|
|
_n1 = [functools.partial(ENDS[k], 1, h) for k in ENDS
|
|
for h in (1200.0, -1500.0)]
|
|
check("⭐ N=1, every drain end (complete, Off, timeout, abort, stale link, "
|
|
"refusal), at 1200 W and under a surplus: every row byte-identical "
|
|
"with and without SAFETY-95",
|
|
[digest(f()[0]) for f in _n1] == [digest(without(f)[0]) for f in _n1])
|
|
_wave = functools.partial(run, 2, lambda t: 2400.0 + 1500 * math.sin(t / 40),
|
|
store=store(2, last_completed=rig.BASE.isoformat()),
|
|
dur=600.0)
|
|
check("N=2 healthy (a moving house, no maintenance): every row "
|
|
"byte-identical with and without SAFETY-95",
|
|
digest(_wave()[0]) == digest(without(_wave)[0]))
|
|
|
|
if fixed is None:
|
|
print(f"\n{len(fails)} FAILED (no _reseed_drain_end): " + ", ".join(fails))
|
|
sys.exit(1)
|
|
|
|
_c = run(2, 1200.0, store=store(2), dur=4.0)[1]
|
|
_pin = MaintResult(IDLE, True, 0.0, False, [])
|
|
_free = MaintResult(CHARGE, False, None, True, [])
|
|
_b = MaintResult(IDLE, False, None, False, [])
|
|
AB = {"inverter-a", "inverter-b"}
|
|
|
|
|
|
def reseeded(target, shares, batts, grid, res=_pin, live=AB, ended=(A,)):
|
|
"""`_reseed_drain_end` on a 2-battery controller: (target, i_w,
|
|
A's start)."""
|
|
_c.target, _c.i_w, _c._drain_ended = target, 123.0, frozenset(ended)
|
|
_c._loop_start, _c._settle, _c._drain_tail = {}, {}, set()
|
|
_c._now_mono, _c.grid = 50.0, grid
|
|
for inv in _c.inverters:
|
|
inv.target, inv.batt = shares[inv.name[-1]], batts[inv.name[-1]]
|
|
_c._reseed_drain_end([res, _b], set(live))
|
|
return _c.target, round(_c.i_w, 6), _c._loop_start.get(A)
|
|
|
|
|
|
# House 1200 W: A drains 2000, B 0, grid -800.
|
|
check("the house needs more than the others stand at: the start is the "
|
|
"house's own point (1210 W), the integrator with it",
|
|
reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0},
|
|
-800.0) == (1210.0, 1210.0, 0.0)
|
|
and _c._settle == {A: 50.0} and _c._drain_tail == {A}
|
|
and _c._drain_ended == frozenset())
|
|
check("...and while A settles the loop's error is the house's point minus "
|
|
"its start (0 here: it holds), not the meter's -790 W",
|
|
_c._drain_grid_w(AB) - AIM == 0.0)
|
|
check("a solar surplus (house -1500 W): the start is what the others stood "
|
|
"at plus A's new command (0 W), and the loop moves down from there",
|
|
reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0},
|
|
-3500.0)[0] == 0.0)
|
|
check("the house needs more than stood placed (4000 W, B at 2000): never "
|
|
"above last cycle's start",
|
|
reseeded(4000.0, {"a": 2000.0, "b": 2000.0},
|
|
{"a": 2000.0, "b": 2000.0}, 0.0)[0] == 4000.0)
|
|
check("a free battery (drain complete -> charge): its drain leaves the "
|
|
"start; a pin's value is its new command",
|
|
reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0},
|
|
-3500.0, res=_free)[2] == 0.0
|
|
and reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0},
|
|
-3500.0, res=MaintResult(HOLD, True, 0.0, False, []))[2]
|
|
== 0.0)
|
|
check("a dropped battery's power is the meter's: a board stuck at 2000 W "
|
|
"keeps B where it stood, and there is no tail to count",
|
|
reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0},
|
|
-800.0, live={"inverter-b"})[0] == 0.0
|
|
and _c._drain_tail == set())
|
|
_chg = MaintResult(CHARGE, True, -1500.0, False, [])
|
|
check("a drain that ends into a pinned charge: the pin is A's new command "
|
|
"while A is live, but 0 W once it is dropped (it is sent 0 W)",
|
|
reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0},
|
|
-3500.0, res=_chg)[2] == -1500.0
|
|
and reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0},
|
|
-3500.0, res=_chg, live={"inverter-b"}) == (0.0, 0.0, 0.0))
|
|
check("no drain ended: nothing moves (the integrator either)",
|
|
reseeded(2000.0, {"a": 2000.0, "b": 0.0}, {"a": 2000.0, "b": 0.0},
|
|
-800.0, ended=()) == (2000.0, 123.0, None))
|
|
_c._drain_tail, _c._settle, _c.grid = {A}, {}, -800.0
|
|
check("the tail ends with A's settling (SAFETY-96's window)",
|
|
_c._drain_grid_w(AB) == -800.0 and _c._drain_tail == set())
|
|
_UA, _UB = (_c.actuator_of("maintenance", i) for i in _c.inverters)
|
|
_drain = MaintResult(DRAIN, True, 2000.0, False, [])
|
|
|
|
|
|
def drain_ends(extra, refused=0):
|
|
"""`_charge_track` with A draining last cycle, its drain SET and the
|
|
claims `extra` on A: which drains it ends."""
|
|
_c._draining_was, _c._drain_ended = frozenset({A}), frozenset()
|
|
_c._charging_was, _c._charge_ended = frozenset(), frozenset()
|
|
_c.live_units = lambda: (_UA, _UB)
|
|
_c.inverters[0].refused = refused
|
|
try:
|
|
_c._charge_track([Claim.set("maintenance", P_MAINTENANCE, 2000.0,
|
|
actuator=_UA)] + extra, [_drain, _b])
|
|
finally:
|
|
del _c.live_units
|
|
_c.inverters[0].refused = 0
|
|
return set(_c._drain_ended)
|
|
|
|
|
|
_rating = [Claim.limit("device", P_SAFETY, -2500.0, 2500.0, actuator=_UA)]
|
|
check("a drain that still stands does not end (its rating binds, no "
|
|
"contradiction, no refusal)", drain_ends(_rating) == set())
|
|
check("a drain whose limits contradict ends: A is sent 0 W (SAFETY-91's "
|
|
"pin as a floor against a closed pack)",
|
|
drain_ends(_rating + [
|
|
Claim.limit("maintenance", P_MAINTENANCE, lo=2000.0, actuator=_UA),
|
|
Claim.limit("pack", P_SAFETY, hi=0.0, actuator=_UA)]) == {A})
|
|
check("a refusal ends a drain only if A will not give (+1); one that will "
|
|
"not take (-1) leaves the drain standing",
|
|
drain_ends(_rating, refused=1) == {A}
|
|
and drain_ends(_rating, refused=-1) == set())
|
|
|
|
_n1c = run(1, 1200.0, store=store(1), dur=4.0)[1]
|
|
_n1c.target, _n1c.i_w, _n1c._drain_ended = 2000.0, 5.0, frozenset({A})
|
|
_n1c._reseed_drain_end([_pin], {A})
|
|
check("N=1: nothing moves", (_n1c.target, _n1c.i_w) == (2000.0, 5.0))
|
|
|
|
_seen = []
|
|
|
|
|
|
def _spy(self, maint_results, live):
|
|
_seen.append(sorted(self._drain_ended))
|
|
return fixed(self, maint_results, live)
|
|
|
|
|
|
Controller._reseed_drain_end = _spy
|
|
try:
|
|
run(2, 1200.0, store=store(2), soc=lambda u, t:
|
|
5.0 if (u == "a" and T <= t < T + 200) else 99.5 if u == "a" and
|
|
t >= T + 200 else 50.0, dur=T + 400.0,
|
|
extra={"maintenance_hold_min": 2})
|
|
finally:
|
|
Controller._reseed_drain_end = fixed
|
|
check("it fires once, at the drain end: not at the charge end, the HOLD "
|
|
"end, or on any other cycle", [x for x in _seen if x] == [[A]]
|
|
and len(_seen) > 300)
|
|
|
|
print("\n drain end | N | house W | site step W head/base | commanded "
|
|
"export W head/base | others' step W head/base | grid import W "
|
|
"head/base | grid step W head/base | export beyond the end's W head/base")
|
|
for k, n, h, a, b in TABLE:
|
|
print(f" {k} | {n} | {h:.0f} | {a['site_step']:.0f}/"
|
|
f"{b['site_step']:.0f} | {a['cmd_export']:.0f}/{b['cmd_export']:.0f}"
|
|
f" | {a['ostep']:.0f}/{b['ostep']:.0f} | {a['imp']:.0f}/"
|
|
f"{b['imp']:.0f} | {a['gstep']:.0f}/{b['gstep']:.0f} | "
|
|
f"{a['exp']:.0f}/{b['exp']:.0f}")
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED: " + ", ".join(fails))
|
|
sys.exit(1)
|
|
print("all SAFETY-95 checks passed")
|