glenn schrooyenandClaude
8bf25696cb
TEST-23 F-1: on a 5 s meter a steady overload warned once in 70 min. The
loop was at its bound 4-5 cycles a quarter, all inside SAFETY-119's settle
of 12 cycles (60 s) and under its budget of 8 per 900 s. The simulation
fed the meter a cycle behind the battery reading, so the guard and the
charge chased each other and gave ~20 bound cycles a quarter.
The settle is now 4 cycles and 22 s (HANDBACK_CYCLES, SHORT_SPAN_S), and
the rolling budget one number, 11 per 900 s, at every cycle length. The
cycle length is no longer tracked. Commands unchanged; the re-pinned
digests of test_safety84/87 differ from bee7431 in OUTPUT_CLAMP rows only.
test_safety119 runs the rig's shape (meter in the same cycle) at 1-15 s
and checks a stop keeps the budget (review F2, X3).
Co-Authored-By: Claude <noreply@anthropic.com>
738 lines
31 KiB
Python
738 lines
31 KiB
Python
"""Runnable check for SAFETY-119: no "Battery is not keeping up" when the
|
|
loop takes the wire back from another owner. `python3 test_safety119.py`
|
|
|
|
During a maintenance charge with no usable peak forecast, the measured peak
|
|
guard steps in during a quarter's last seconds and hands back. On the cycle
|
|
the loop gets the wire back, its integrator is re-seeded from the other
|
|
owner's command and kicked by a meter reading taken before the battery moved.
|
|
It crossed its rail for one or two cycles, the loop's ask beat the one-sided
|
|
limit, so the cycle counted as the loop's own and raised
|
|
`goodwe_integrator_bound`. The battery gave what it was told. Now the loop's
|
|
bound does not count until the plant's settle (SHORT_CYCLES cycles and
|
|
SHORT_SPAN_S) has passed since the hand-back. SAFETY-121: the settle is
|
|
HANDBACK_CYCLES (4) cycles and SHORT_SPAN_S (22 s), and across all
|
|
hand-backs the bound is hidden on at most ROLLING_BOUND_CYCLES (11) cycles in
|
|
any ROLLING_S (900 s), at every cycle length.
|
|
DECISIONS/SAFETY-119-bound-after-hand-back.md,
|
|
DECISIONS/SAFETY-121-bound-every-quarter.md.
|
|
|
|
1. The rig's shape (1 s meter that lags the battery, 1 s battery lag, house
|
|
800 W, rating 2500 W, limit 2500 W, margin 50, charge 2000 W, no
|
|
forecast), 2 h, three seeds: no notification, the cap held, every command
|
|
the same as without the fix; without the fix it warns.
|
|
2. A genuine integrator bound still warns: the loop in charge with the house
|
|
over the rating (at once, as before), and the loop at its rail after a
|
|
hand-back (after the settle).
|
|
3. With a usable forecast: the same commands and the same alarms as without
|
|
the fix.
|
|
4. Repeated hand-backs (N=2, the guard and the charge take turns), 2, 5, 10
|
|
and 15 s meters: it warns in every quarter, both with the meter a cycle
|
|
behind the battery reading and in the rig's shape (SAFETY-121, F-1: the
|
|
meter read in the same cycle, 1 to 5 bound cycles a quarter), where
|
|
SAFETY-119's settle of 12 cycles hid every quarter but one; the
|
|
quarter-edge case stays silent at 2, 5, 10 and 15 s; the budget's margins
|
|
(8 warns at the quarter edge, 14 misses quarters); a start or a blind
|
|
cycle is not a hand-back; the delay after one hand-back.
|
|
5. The rule at its boundaries (`Controller._settling`): the settle, the
|
|
rolling budget, and a stop that does not give the budget back.
|
|
6. test_safety84's and test_safety87's re-pinned digests: the same commands
|
|
and claims; only OUTPUT_CLAMP log rows differ.
|
|
|
|
Mutations (MUTATION.md, purged copies) are listed in the pull request.
|
|
"""
|
|
|
|
import logging
|
|
import math
|
|
import os
|
|
import random
|
|
import sys
|
|
import time
|
|
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
|
|
import app.main as app_main # noqa: E402
|
|
from app import messages # noqa: E402
|
|
from app.main import Controller # noqa: E402
|
|
from app.p1 import make_sample, split_signed # noqa: E402
|
|
from named_absence import first, nth, present # noqa: E402
|
|
|
|
fails = []
|
|
OPT0 = dict(s84.OPT)
|
|
CAP = 2500.0
|
|
QUARTER = 900.0
|
|
|
|
|
|
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 (time, id)."""
|
|
if len(a) >= 3 and a[1] == "create" and isinstance(a[2], dict):
|
|
self.__dict__.setdefault("notes", []).append(
|
|
(self.t, a[2].get("notification_id")))
|
|
return True
|
|
|
|
|
|
def _power(self, u, lag=1.0):
|
|
"""The TEST-23 plant's battery: 800 W/s ramp, 1 s first-order lag."""
|
|
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)
|
|
if self.cut(u, self.t) and self.cmd[u] >= -4.0:
|
|
return -4.0
|
|
return float(round(a))
|
|
|
|
|
|
_real_number, _real_power = rig._Hass.number, rig._Hass.power
|
|
_real_feed = s79._feed
|
|
|
|
|
|
async def _number(self, e, invert=False):
|
|
"""sensor.fc: a usable demand forecast of 800 W (part 3)."""
|
|
if e == "sensor.fc":
|
|
return 800.0
|
|
return await _real_number(self, e, invert)
|
|
|
|
|
|
rig._Hass.call, rig._Hass.power, rig._Hass.number = _call, _power, _number
|
|
rig.TICK = 1.0
|
|
DELAY = [1]
|
|
|
|
|
|
def _feed(c, t):
|
|
"""The P1 meter `DELAY` ticks behind the battery, as on the rig (the
|
|
meter reached HA after the battery reading). Records the true grid."""
|
|
h = c.hass.__dict__.setdefault("grid_true", [])
|
|
h.append((t, c.hass.grid()))
|
|
g = h[max(0, len(h) - 1 - DELAY[0])][1]
|
|
imp, exp = split_signed(g)
|
|
c.p1.submit(make_sample("ha_signed", imp, exp, phases=1,
|
|
ingest_ts=rig.BASE + timedelta(seconds=t),
|
|
ingest_mono=time.monotonic()))
|
|
|
|
|
|
s79._feed = _feed
|
|
|
|
|
|
def house(seed, base=800.0, step=None):
|
|
"""800 W +- 30 W, a new value each second; `step` = (t, W) from then."""
|
|
rnd = random.Random(seed)
|
|
noise = [rnd.uniform(-30.0, 30.0) for _ in range(4000)]
|
|
|
|
def load(t):
|
|
if step is not None and t >= step[0]:
|
|
return step[1]
|
|
return base + noise[int(t) % 4000]
|
|
return load
|
|
|
|
|
|
def run(load, *, n=1, dur=7200.0, delay=1, extra=None, cheap=True, cut=None,
|
|
charging=True, blind=0.0):
|
|
"""One run of `n` inverters of 2500 W; `charging=False`: no maintenance,
|
|
the loop in charge; `blind`: no meter reading until that second."""
|
|
extra = dict(extra or {})
|
|
s84.OPT = dict(OPT0, peak_cap_w=int(CAP * n), peak_margin_w=50,
|
|
maintenance_charge_w=2000)
|
|
if not charging:
|
|
extra.update(maintenance_enabled=False, peak_cap_w=20000)
|
|
for k in [k for k in extra if k in s84.OPT]:
|
|
s84.OPT[k] = extra.pop(k)
|
|
DELAY[0] = delay
|
|
try:
|
|
feed = s84.feed(cheap)
|
|
|
|
def before(c, t):
|
|
if t >= blind: # before that: no meter reading
|
|
return feed(c, t)
|
|
return s84.run(n, load=load, cheap=cheap,
|
|
store=(s79.store_charging(n) if charging
|
|
else rig._Store()),
|
|
soc=lambda u, t: 60.0, dur=dur, cut=cut,
|
|
before=before if blind else None,
|
|
extra=dict({"max_w": 2500, "slew_w": 0}, **extra))
|
|
finally:
|
|
s84.OPT = OPT0
|
|
|
|
|
|
def patched(obj, name, value):
|
|
def wrap(fn):
|
|
real = getattr(obj, name)
|
|
setattr(obj, name, value)
|
|
try:
|
|
return fn()
|
|
finally:
|
|
setattr(obj, name, real)
|
|
return wrap
|
|
|
|
|
|
no_fix = patched(Controller, "_settling", lambda self, owns, now: False)
|
|
|
|
|
|
def notes(c):
|
|
return [t for t, i in getattr(c.hass, "notes", [])
|
|
if i == messages.ID_LOOP_BOUND]
|
|
|
|
|
|
def quarters(c, tick=1.0):
|
|
"""Each whole quarter's true average import (W)."""
|
|
q = {}
|
|
for t, g in c.hass.grid_true:
|
|
q.setdefault(int(t // QUARTER), []).append(g)
|
|
return [sum(v) / len(v) for v in q.values()
|
|
if len(v) == QUARTER / tick]
|
|
|
|
|
|
def commands(rows):
|
|
return [(r[0], r[1], r[2]) for r in rows]
|
|
|
|
|
|
def guard_steps(c):
|
|
return sum(1 for e in c.events if "PEAK_GUARD:" in e)
|
|
|
|
|
|
# ============================================================================ #
|
|
print("1. a maintenance charge with no forecast, the guard steps in at each "
|
|
"quarter's end, 2 h")
|
|
NEW, OLD = {}, {}
|
|
for seed, delay in ((1, 1), (2, 1), (3, 2)):
|
|
NEW[seed] = run(house(seed), delay=delay)
|
|
OLD[seed] = no_fix(lambda: run(house(seed), delay=delay))
|
|
print(f" seed {seed} meter {delay} s behind: notifications "
|
|
f"{notes(NEW[seed][1])} (without the fix {notes(OLD[seed][1])}), "
|
|
f"highest quarter {max(quarters(NEW[seed][1])):.0f} W, guard "
|
|
f"step-ins in the last 40 log rows {guard_steps(NEW[seed][1])}")
|
|
check("⭐ no 'Battery is not keeping up' notification, cycle or episode",
|
|
all(notes(c) == [] and c.clamp_wd.bound == 0
|
|
and c.clamp_wd.bound_episodes == 0 for _r, c in NEW.values()))
|
|
check("⭐ the peak cap is held: every quarter's average is under 2500 W",
|
|
all(len(quarters(c)) == 8 and max(quarters(c)) < CAP
|
|
for _r, c in NEW.values()))
|
|
check("⭐ control is unchanged: every command the same as without the fix",
|
|
all(commands(NEW[k][0]) == commands(OLD[k][0]) for k in NEW))
|
|
check("...the charge ran and the guard stepped in near each quarter's end",
|
|
all(any(r[1] == -2000.0 for r in rows) and any(
|
|
r[-1][1] and r[-1][1][0] and r[0] % QUARTER >= QUARTER - 40
|
|
for r in rows) for rows, _c in NEW.values()))
|
|
check("non-vacuous: without the fix every run warns, at a quarter's edge",
|
|
all(len(notes(c)) >= 1 for _r, c in OLD.values())
|
|
and all(min(t % QUARTER, QUARTER - t % QUARTER) <= 40
|
|
for _r, c in OLD.values() for t in notes(c)))
|
|
|
|
# ============================================================================ #
|
|
print("\n2. a genuine integrator bound still warns")
|
|
_rail = run(house(1, base=4000.0), dur=600.0,
|
|
charging=False)
|
|
_rail_old = no_fix(lambda: run(house(1, base=4000.0), dur=600.0,
|
|
charging=False))
|
|
print(f" loop in charge, house 4000 W over the 2500 W rating: "
|
|
f"{notes(_rail[1])}, without the fix {notes(_rail_old[1])}")
|
|
check("⭐ the loop in charge at its rail warns (log and notification), at "
|
|
"the same cycle as before",
|
|
len(notes(_rail[1])) >= 1 and notes(_rail[1]) == notes(_rail_old[1])
|
|
and _rail[1].clamp_wd.bound == _rail_old[1].clamp_wd.bound
|
|
and any("OUTPUT_CLAMP episode" in e for e in _rail[1].events))
|
|
|
|
# The charge sets the wire until 600 s, then a solar surplus of 6000 W: the
|
|
# loop takes the wire back and stands at its -2500 W rail.
|
|
_hb = run(house(1, step=(600.0, -6000.0)), dur=1200.0,
|
|
extra={"peak_cap_w": 20000})
|
|
_hb_old = no_fix(lambda: run(house(1, step=(600.0, -6000.0)), dur=1200.0,
|
|
extra={"peak_cap_w": 20000}))
|
|
_first = notes(_hb[1])[:1]
|
|
print(f" hand-back to a loop at its rail at 600 s: first notification "
|
|
f"{_first}, without the fix {notes(_hb_old[1])[:1]}; command at the "
|
|
f"end {_hb[0][-1][1]}")
|
|
MB = 11 # the budget (at 1 s it ends the settle)
|
|
check(f"⭐ the loop at its rail after a hand-back still warns, once "
|
|
f"{MB} of its bound cycles are hidden (1 s meter: {MB} s)",
|
|
_first and 600.0 + MB <= _first[0] <= 600.0 + MB + 4.0
|
|
and _hb[0][-1][1] == -2500.0
|
|
and _hb[1].clamp_wd.bound > 100)
|
|
check("...without the fix it warned at once (non-vacuous)",
|
|
notes(_hb_old[1])[:1] and notes(_hb_old[1])[0] < 600.0 + MB)
|
|
# The same charge with no forecast; at 1795 s, under the guard's step-in at
|
|
# the quarter's end, the house goes to 4000 W, over the 2500 W rating.
|
|
_up = run(house(1, step=(1795.0, 4000.0)), dur=1900.0)
|
|
_up_old = no_fix(lambda: run(house(1, step=(1795.0, 4000.0)), dur=1900.0))
|
|
print(f" guard step-in, then the house over the rating from 1795 s: "
|
|
f"{notes(_up[1])}, without the fix {notes(_up_old[1])}")
|
|
check("⭐ a genuine bound in the rig's own scenario still warns, within the "
|
|
"settle of the guard's hand-back",
|
|
notes(_up[1]) and 1795.0 <= notes(_up[1])[0] <= 1795.0 + 40.0
|
|
and any("OUTPUT_CLAMP episode" in e for e in _up[1].events))
|
|
|
|
# ============================================================================ #
|
|
print("\n3. with a usable forecast (800 W): unchanged")
|
|
FC = {"peak_forecast_entity": "sensor.fc"}
|
|
_fc = run(house(1), extra=FC)
|
|
_fc_old = no_fix(lambda: run(house(1), extra=FC))
|
|
print(f" notifications {notes(_fc[1])} (without the fix "
|
|
f"{notes(_fc_old[1])}), highest quarter "
|
|
f"{max(quarters(_fc[1])):.0f} W, charge "
|
|
f"{min(r[1] for r in _fc[0][:300]):.0f} W")
|
|
check("⭐ the same commands and the same alarms as without the fix, and "
|
|
"the cap held",
|
|
commands(_fc[0]) == commands(_fc_old[0])
|
|
and notes(_fc[1]) == notes(_fc_old[1])
|
|
and _fc[1].clamp_wd.engagements == _fc_old[1].clamp_wd.engagements
|
|
and max(quarters(_fc[1])) < CAP)
|
|
check("...the forecast was used (the charge is shaped under -2000 W)",
|
|
-2000.0 < min(r[1] for r in _fc[0][:300]) < 0.0)
|
|
|
|
# ============================================================================ #
|
|
print("\n4. repeated hand-backs: the guard and the charge take turns")
|
|
|
|
|
|
def round1(self, owns, now):
|
|
"""Round 1's rule, kept here to show what it hid: every hand-back
|
|
restarts the window, so a loss within the settle never lets it end."""
|
|
d = self.__dict__
|
|
if not owns:
|
|
d["_r1"], d["_r1_n"] = True, None
|
|
return False
|
|
if d.get("_r1"):
|
|
d["_r1"], d["_r1_n"] = False, (now, 0)
|
|
if d.get("_r1_n") is None:
|
|
return False
|
|
t0, n = d["_r1_n"]
|
|
d["_r1_n"] = (t0, n + 1)
|
|
if n + 1 >= app_main.SHORT_CYCLES and (
|
|
now is None or t0 is None or now - t0 >= app_main.SHORT_SPAN_S):
|
|
d["_r1_n"] = None
|
|
return False
|
|
return True
|
|
|
|
|
|
# N=2 (2 x 2500 W), a cheap-window charge, no usable forecast, the house 50 W
|
|
# over the 5000 W rating from 300 s: the loop never held the wire for more
|
|
# than 10 cycles. Round 1: no notification in 2 h; round 2: none at 10 s or
|
|
# 15 s meters.
|
|
def never(tick, hw=5050.0, fix=True, seed=0, delay=1):
|
|
"""`seed` 0: a constant house; else 800 W +- 30 W, `hw` +- 30 W from
|
|
300 s. `delay` 0 is the rig's shape: the meter read in the same cycle
|
|
as the battery (SAFETY-121)."""
|
|
rig.TICK = tick
|
|
if seed:
|
|
load = (lambda t, h=house(seed):
|
|
h(t) + (hw - 800.0 if t >= 300.0 else 0.0))
|
|
else:
|
|
load = house(1, step=(300.0, hw))
|
|
try:
|
|
f = lambda: run(load, n=2, delay=delay)
|
|
return f() if fix else no_fix(f)
|
|
finally:
|
|
rig.TICK = 1.0
|
|
|
|
|
|
def per_quarter(c):
|
|
return [sum(1 for t in notes(c) if int(t // QUARTER) == q)
|
|
for q in range(8)]
|
|
|
|
|
|
def every_quarter(c):
|
|
return all(n >= 1 for n in per_quarter(c)[1:])
|
|
|
|
|
|
_nv = {t: never(t) for t in (2.0, 5.0, 10.0, 15.0)}
|
|
_nv_old = {t: never(t, fix=False) for t in (2.0, 5.0, 10.0, 15.0)}
|
|
_nv_r1 = patched(Controller, "_settling", round1)(lambda: never(5.0))
|
|
for _t in _nv:
|
|
print(f" N=2, {_t:.0f} s meter, house 5050 W: notifications "
|
|
f"{len(notes(_nv[_t][1]))} (per quarter "
|
|
f"{per_quarter(_nv[_t][1])}), without the fix "
|
|
f"{len(notes(_nv_old[_t][1]))}, highest quarter "
|
|
f"{max(quarters(_nv[_t][1], _t)):.0f} W")
|
|
print(f" round 1's rule at 5 s: {len(notes(_nv_r1[1]))}")
|
|
check("⭐ 2, 5, 10 and 15 s meters: it warns in every quarter after the "
|
|
"first (round 2 gave none at 10 and 15 s)",
|
|
all(every_quarter(_nv[t][1]) for t in _nv))
|
|
check("non-vacuous: round 1's rule (a window per hand-back) never warned "
|
|
"here, and the fix still hides some of what was raised without it",
|
|
notes(_nv_r1[1]) == [] and all(
|
|
_nv[t][1].clamp_wd.bound < _nv_old[t][1].clamp_wd.bound
|
|
for t in _nv))
|
|
check("...every command is the same as without the fix, at 2, 5, 10 and "
|
|
"15 s",
|
|
all(commands(_nv[t][0]) == commands(_nv_old[t][0]) for t in _nv))
|
|
check("...and at 5 s the cap held",
|
|
len(quarters(_nv[5.0][1], 5.0)) == 8
|
|
and max(quarters(_nv[5.0][1], 5.0)) < 5000.0)
|
|
|
|
# SAFETY-121, the rig's shape (TEST-23, F-1): the meter is read in the same
|
|
# cycle as the battery, so the guard's charge headroom is right after it
|
|
# hands back and the guard and the charge do not chase each other. The loop
|
|
# then holds the wire once a quarter, at its start, and is at its bound on
|
|
# 1 to 5 cycles (5 s), not 20. House 5100 W +- 30 W (5300 W at 15 s, where
|
|
# 5100 W is over the rating in one quarter only, without the fix too).
|
|
RIG = {1.0: 5100.0, 2.0: 5100.0, 5.0: 5100.0, 10.0: 5100.0, 15.0: 5300.0}
|
|
_rg = {t: never(t, hw, seed=1, delay=0) for t, hw in RIG.items()}
|
|
_rg_old = {t: never(t, hw, fix=False, seed=1, delay=0)
|
|
for t, hw in RIG.items()}
|
|
# SAFETY-119 as merged (bee7431) at 5 s: a settle of 12 cycles, budget 8.
|
|
_rg_119 = patched(app_main, "HANDBACK_CYCLES", app_main.SHORT_CYCLES)(
|
|
lambda: patched(app_main, "ROLLING_BOUND_CYCLES", 8)(
|
|
lambda: never(5.0, 5100.0, seed=1, delay=0)))
|
|
for _t in _rg:
|
|
print(f" rig shape, {_t:.0f} s meter, house {RIG[_t]:.0f} W: per "
|
|
f"quarter {per_quarter(_rg[_t][1])}, without the fix "
|
|
f"{per_quarter(_rg_old[_t][1])}, highest quarter "
|
|
f"{max(quarters(_rg[_t][1], _t)):.0f} W")
|
|
print(f" rig shape, 5 s, SAFETY-119's settle and budget: per quarter "
|
|
f"{per_quarter(_rg_119[1])}")
|
|
check("⭐ the rig's shape at 1, 2, 5, 10 and 15 s meters: it warns in every "
|
|
"quarter after the first, as without the fix",
|
|
all(every_quarter(_rg[t][1]) and every_quarter(_rg_old[t][1])
|
|
for t in _rg))
|
|
check("non-vacuous (F-1): with SAFETY-119's settle of 12 cycles and budget "
|
|
"of 8, the rig's shape at 5 s warns in at most one quarter",
|
|
sum(1 for n in per_quarter(_rg_119[1])[1:] if n) <= 1)
|
|
check("...every command the same as without the fix, and the cap held at "
|
|
"1 to 10 s",
|
|
all(commands(_rg[t][0]) == commands(_rg_old[t][0]) for t in _rg)
|
|
and all(max(quarters(_rg[t][1], t)) < 5000.0
|
|
for t in (1.0, 2.0, 5.0, 10.0)))
|
|
|
|
# The quarter-edge case (part 1) at the slower meters: still no warning.
|
|
# The meter's own lag limits the cap there, the same with and without the fix.
|
|
_qe = {}
|
|
for _t in (2.0, 5.0, 10.0, 15.0): # with part 1: 1, 2, 5, 10, 15 s
|
|
rig.TICK = _t
|
|
for _seed, _dl in ((1, 1), (2, 1), (3, 2)):
|
|
_qe[_t, _seed] = (run(house(_seed), delay=_dl),
|
|
no_fix(lambda: run(house(_seed), delay=_dl)))
|
|
rig.TICK = 1.0
|
|
print(f" {_t:.0f} s meter, quarter edge: notifications "
|
|
f"{[len(notes(r[0][1])) for k, r in _qe.items() if k[0] == _t]} "
|
|
f"(without the fix "
|
|
f"{[len(notes(r[1][1])) for k, r in _qe.items() if k[0] == _t]})")
|
|
check("⭐ the quarter-edge case at 2, 5, 10 and 15 s meters, seeds 1 to 3: "
|
|
"no notification",
|
|
all(notes(r[0][1]) == [] and r[0][1].clamp_wd.bound == 0
|
|
for r in _qe.values()))
|
|
check("...every command the same as without the fix; without it the false "
|
|
"alarm is there at each meter (non-vacuous)",
|
|
all(commands(r[0][0]) == commands(r[1][0]) for r in _qe.values())
|
|
and all(any(len(notes(_qe[_t, k][1][1])) >= 1 for k in (1, 2, 3))
|
|
for _t in (2.0, 5.0, 10.0, 15.0)))
|
|
|
|
# The budget's margins (DECISIONS/SAFETY-121): 9 to 13 met both cases in the
|
|
# sweep. 8 warns at the quarter edge (2 s, seed 25); 14 misses quarters in
|
|
# the rig's shape (1 s, house 5050 W, seed 1).
|
|
rig.TICK = 2.0
|
|
_q25 = (run(house(25), delay=2),
|
|
patched(app_main, "ROLLING_BOUND_CYCLES", 8)(
|
|
lambda: run(house(25), delay=2)))
|
|
rig.TICK = 1.0
|
|
_m14 = patched(app_main, "ROLLING_BOUND_CYCLES", 14)(
|
|
lambda: never(1.0, 5050.0, seed=1, delay=0))
|
|
_m11 = never(1.0, 5050.0, seed=1, delay=0)
|
|
print(f" 2 s, quarter edge, seed 25: {notes(_q25[0][1])}, budget 8 "
|
|
f"{notes(_q25[1][1])}; rig shape 1 s 5050 W per quarter "
|
|
f"{per_quarter(_m11[1])}, budget 14 {per_quarter(_m14[1])}")
|
|
check("...the budget's margins: 11 is silent at 2 s seed 25 and warns every "
|
|
"quarter in the rig's shape at 1 s and 5050 W; 8 warns there, 14 "
|
|
"misses a quarter there",
|
|
notes(_q25[0][1]) == [] and len(notes(_q25[1][1])) >= 1
|
|
and every_quarter(_m11[1]) and not every_quarter(_m14[1]))
|
|
check("the rule is the one the sweep chose. Changing it means re-running "
|
|
"the sweep in DECISIONS/SAFETY-121-bound-every-quarter.md",
|
|
app_main.HANDBACK_CYCLES == 4 and app_main.ROLLING_BOUND_CYCLES == 11
|
|
and app_main.ROLLING_S == 900.0 and app_main.SHORT_SPAN_S == 22.0)
|
|
|
|
# A start, or a blind cycle, is not a hand-back: the loop at its rail after
|
|
# 20 s of not running warns at once (R2: a restart counted as a hand-back).
|
|
_bl = run(house(1, base=4000.0), dur=120.0, charging=False, blind=20.0)
|
|
_bl_old = no_fix(lambda: run(house(1, base=4000.0), dur=120.0,
|
|
charging=False, blind=20.0))
|
|
print(f" the loop not running until 20 s, then at its rail: "
|
|
f"{notes(_bl[1])}, without the fix {notes(_bl_old[1])}")
|
|
check("⭐ a start after a blind spell is not a hand-back: it warns at the "
|
|
"same cycle as without the fix",
|
|
notes(_bl[1])[:1] and 20.0 <= notes(_bl[1])[0] <= 26.0
|
|
and notes(_bl[1]) == notes(_bl_old[1]))
|
|
|
|
# One hand-back to a loop at its rail at 600 s: the first warning comes when
|
|
# the settle ends, 22 s and 4 cycles after it (SAFETY-119: 12 cycles, 40 s at
|
|
# 5 s and 80 s at 10 s).
|
|
_sl = {}
|
|
for _t in (5.0, 10.0):
|
|
rig.TICK = _t
|
|
_sl[_t] = run(house(1, step=(600.0, -6000.0)), dur=900.0,
|
|
extra={"peak_cap_w": 20000})
|
|
rig.TICK = 1.0
|
|
_sl1 = {t: [x for x in notes(r[1]) if x >= 600.0][:1] for t, r in _sl.items()}
|
|
print(f" hand-back to a loop at its rail at 600 s, first notification: "
|
|
f"5 s {_sl1[5.0]}, 10 s {_sl1[10.0]}")
|
|
check("⭐ one hand-back (the loop has the wire from the cycle after the "
|
|
"step): the first warning comes 25 s after it at a 5 s meter (the "
|
|
"22 s span), 30 s after it at 10 s (4 cycles)",
|
|
_sl1[5.0] == [630.0] and _sl1[10.0] == [640.0])
|
|
|
|
# ============================================================================ #
|
|
print("\n5. the rule at its boundaries")
|
|
|
|
|
|
class _C:
|
|
_settling = Controller._settling
|
|
|
|
def __init__(self):
|
|
self._wire_lost, self._handback, self._hidden = False, None, ()
|
|
self.loop_clamped = (False, "", False)
|
|
|
|
|
|
def seq(c, steps):
|
|
"""(owned, time[, the loop's bound that cycle]) -> does it count not."""
|
|
out = []
|
|
for st in steps:
|
|
c.loop_clamped = (len(st) > 2 and st[2], "", False)
|
|
out.append(c._settling(st[0], st[1]))
|
|
return out
|
|
|
|
|
|
SPAN = app_main.SHORT_SPAN_S
|
|
K, M, ROLL = 4, 11, 900.0 # the chosen rule, written out on purpose
|
|
_c = _C()
|
|
check("from the start, owned every cycle: never settling (no hand-back)",
|
|
seq(_c, [(True, float(i)) for i in range(30)]) == [False] * 30)
|
|
_c = _C()
|
|
_out = seq(_c, [(False, 0.0)] + [(True, 2.0 * i) for i in range(1, 15)])
|
|
check(f"after another owner at 2 s: settling for 11 owned cycles, then not "
|
|
f"(the span, {SPAN:.0f} s, decides)",
|
|
_out == [False] + [True] * 11 + [False] * 3)
|
|
_c = _C()
|
|
_out = seq(_c, [(False, 0.0)] + [(True, 0.5 * i) for i in range(1, 60)])
|
|
_end = _out.index(False, 1) if False in _out[1:] else -1
|
|
check(f"at a 0.5 s meter the span decides: settling until {SPAN:.0f} s",
|
|
_end == 1 + int(SPAN / 0.5) and all(_out[1:_end]))
|
|
_set = {}
|
|
for _tick in (10.0, 30.0, 60.0):
|
|
_c = _C()
|
|
_set[_tick] = seq(_c, [(False, 0.0)]
|
|
+ [(True, _tick * i) for i in range(1, K + 2)])
|
|
check(f"at 10, 30 and 60 s the cycles decide: settling for {K - 1} owned "
|
|
f"cycles, then not (SAFETY-119: 11)",
|
|
all(o == [False] + [True] * (K - 1) + [False] * 2
|
|
for o in _set.values()))
|
|
_c = _C()
|
|
_out = seq(_c, [(False, 0.0), (True, 2.0), (True, 4.0), (False, 6.0),
|
|
(True, 8.0), (True, 10.0)])
|
|
check("losing the wire again restarts the settle",
|
|
_out == [False, True, True, False, True, True]
|
|
and _c._handback == (8.0, 2))
|
|
_c = _C()
|
|
check("no monotonic clock: the cycle count alone decides",
|
|
seq(_c, [(False, None)] + [(True, None)] * (K + 1))
|
|
== [False] + [True] * (K - 1) + [False] * 2)
|
|
|
|
# One hand-back, the loop's bound on every cycle: the settle (5 s) or the
|
|
# budget (1 s) ends the hiding.
|
|
_c = _C()
|
|
_o5 = seq(_c, [(False, 0.0)] + [(True, 5.0 * i, True) for i in range(1, 9)])
|
|
_c = _C()
|
|
_o1 = seq(_c, [(False, 0.0)] + [(True, 1.0 * i, True) for i in range(1, 21)])
|
|
check("one hand-back, the bound on every cycle: 5 hidden at 5 s (22 s), "
|
|
f"then counted; {M} hidden at 1 s (the budget), then counted",
|
|
_o5 == [False] + [True] * 5 + [False] * 3
|
|
and _o1 == [False] + [True] * M + [False] * (20 - M))
|
|
|
|
|
|
def turns(c, tick, lost, periods, first=0.0, bound=True):
|
|
"""A hand-back, then `periods` x (1 owned cycle with the loop's bound,
|
|
`lost` cycles away). Returns [(time, hidden)] of the owned cycles."""
|
|
t, out = first, []
|
|
for _ in range(lost):
|
|
c._settling(False, t)
|
|
t += tick
|
|
for _ in range(periods):
|
|
c.loop_clamped = (bound, "", False)
|
|
out.append((t, c._settling(True, t)))
|
|
t += tick
|
|
for _ in range(lost):
|
|
c._settling(False, t)
|
|
t += tick
|
|
return out, t
|
|
|
|
|
|
# The rolling budget: the loop's bound is hidden on at most M cycles in any
|
|
# ROLL seconds, across all hand-backs.
|
|
_c = _C()
|
|
_o, _ = turns(_c, 5.0, 3, 60)
|
|
_hid = [t for t, h in _o if h]
|
|
check(f"across hand-backs (one bound cycle per turn, 5 s): the first {M} "
|
|
"are hidden, then every one counts",
|
|
[h for _t, h in _o[:M + 5]] == [True] * M + [False] * 5)
|
|
check(f"...never more than {M} hidden in any {ROLL:.0f} s, and they come "
|
|
f"back as the oldest ages out (exactly {M} in the first {ROLL:.0f} s "
|
|
"after the first)",
|
|
all(sum(1 for x in _hid if t <= x < t + ROLL) <= M for t in _hid)
|
|
and sum(1 for x in _hid if x < _hid[0] + ROLL) == M
|
|
and any(x >= _hid[0] + ROLL for x in _hid))
|
|
_c = _C()
|
|
_c._hidden = (0.0,) * M
|
|
_c._wire_lost, _c._handback = False, (ROLL - 30.0, 0)
|
|
_c.loop_clamped = (True, "", False)
|
|
_a = _c._settling(True, ROLL - 1.0)
|
|
_c.loop_clamped = (True, "", False)
|
|
_b = _c._settling(True, ROLL)
|
|
check(f"the budget refills {ROLL:.0f} s after the oldest hidden cycle: "
|
|
f"counted at {ROLL - 1:.0f} s, hidden again at {ROLL:.0f} s",
|
|
_a is False and _b is True)
|
|
_c = _C()
|
|
_o1, _t = turns(_c, 2.0, 3, 20)
|
|
for _ in range(12): # the wire away a whole settle
|
|
_c._settling(False, _t)
|
|
_t += 2.0
|
|
_o2, _ = turns(_c, 2.0, 3, 5, first=_t)
|
|
check("the budget is not per hand-back or per run: after a whole settle "
|
|
"with the wire away, a new hand-back within the same "
|
|
f"{ROLL:.0f} s hides nothing more",
|
|
[h for _t2, h in _o1[:M]] == [True] * M
|
|
and not any(h for _t2, h in _o2))
|
|
_c = _C()
|
|
_o1, _t = turns(_c, 2.0, 3, 20)
|
|
_ = [_c._settling(True, _t + 2.0 * i) for i in range(13)]
|
|
_c._settling(False, _t + 40.0)
|
|
_c.loop_clamped = (True, "", False)
|
|
check("the loop holding the wire a whole settle ends the window: the next "
|
|
"hand-back hides nothing more while the budget is spent",
|
|
_c._settling(True, _t + 42.0) is False)
|
|
_c = _C()
|
|
_c._wire_lost = True
|
|
_t = 0.0
|
|
for _ in range(M + 4): # an over-rating: always counts
|
|
_c.loop_clamped = (True, "", True)
|
|
_c._settling(True, _t)
|
|
_c._wire_lost = True
|
|
_t += 5.0
|
|
_c._settling(False, _t)
|
|
_c.loop_clamped = (True, "", False)
|
|
check("an over-rating does not use up the budget",
|
|
_c._settling(True, _t + 5.0) is True and len(_c._hidden) == 1)
|
|
|
|
# The budget is one number at every cycle length (SAFETY-119 scaled it).
|
|
_bud = {}
|
|
for _tick in (1.0, 2.0, 3.0, 5.0, 10.0, 15.0, 20.0, 30.0):
|
|
_c = _C()
|
|
_o, _ = turns(_c, _tick, 1, 14)
|
|
_bud[_tick] = sum(1 for _t2, h in _o if h)
|
|
_c = _C()
|
|
_o60, _ = turns(_c, 60.0, 1, 14)
|
|
print(f" hidden before the first counts, one bound cycle per turn: {_bud}; "
|
|
f"at 60 s, turns of 2 cycles: {sum(1 for _t2, h in _o60 if h)} of 14 "
|
|
"hidden (the budget never runs out: DECISIONS limit)")
|
|
check(f"the budget does not follow the cycle length: {M} hidden at 1 to "
|
|
"30 s, then counted",
|
|
set(_bud.values()) == {M})
|
|
|
|
# No monotonic clock: the budget cannot refill, so a bound that stays is
|
|
# counted after the budget however often the wire changes hands.
|
|
_c = _C()
|
|
_nc = []
|
|
for _ in range(40):
|
|
for _ in range(3):
|
|
_c._settling(False, None)
|
|
_c.loop_clamped = (True, "", False)
|
|
_nc.append(_c._settling(True, None))
|
|
check(f"no monotonic clock, repeated hand-backs: the budget holds ({M} "
|
|
"hidden, then every bound counts)", _nc == [True] * M + [False] * (40 - M))
|
|
|
|
# An over-rating (a command over `max_w`) is a bug, never settled away.
|
|
_ctl = _hb[1]
|
|
_ctl.loop_clamped = (True, "x", True)
|
|
_ctl._wire_lost, _ctl._handback, _ctl._hidden = True, None, ()
|
|
|
|
|
|
class _Res:
|
|
winner, bound_by, target_w = "loop", "", 0.0
|
|
unit_bounds = {}
|
|
|
|
|
|
class _D:
|
|
held, target_w = False, 0.0
|
|
|
|
|
|
_ctl._loop_decision, _ctl.allocation = _D(), None
|
|
_ctl._loop_alarm_input(_Res())
|
|
check("an over-rating on a hand-back cycle still counts (bug=True kept)",
|
|
_ctl.loop_clamped == (True, "x", True) and _ctl._handback is not None)
|
|
_ctl.loop_clamped = (True, "y", False)
|
|
_ctl._loop_alarm_input(_Res())
|
|
check("...while the integrator bound on the next cycle is not counted",
|
|
_ctl.loop_clamped[0] is False)
|
|
|
|
# A stop ends the window but does not give the budget back (review F2, X3).
|
|
_now = _ctl._now_mono
|
|
_ctl._hidden = (_now,) * M
|
|
_ctl._wire_lost, _ctl._handback = True, (1.0, 3)
|
|
_ctl._loop_decision = None # stopped or blind
|
|
_ctl._loop_alarm_input(_Res())
|
|
_st, _kept = (_ctl._wire_lost, _ctl._handback), _ctl._hidden
|
|
_ctl._settling(False, _now + 1.0)
|
|
_ctl.loop_clamped = (True, "", False)
|
|
_after = _ctl._settling(True, _now + 2.0)
|
|
check("a stop clears the window (the next start is not a hand-back) and "
|
|
f"keeps the budget: with {M} hidden, the bound after the next "
|
|
"hand-back counts",
|
|
_st == (False, None) and _kept == (_now,) * M and _after is False)
|
|
print(f" after the stop: hidden kept {len(_kept)}, the next hand-back's "
|
|
f"bound hidden {_after}")
|
|
|
|
# ============================================================================ #
|
|
print("\n6. the pinned digests of test_safety84 and test_safety87: only the "
|
|
"log changes")
|
|
# Those harnesses run at the 5 s tick with a battery that follows at once.
|
|
rig._Hass.power, rig.TICK, s79._feed = _real_power, 5.0, _real_feed
|
|
rig._Hass.number = _real_number
|
|
import safety87_identity as s87 # noqa: E402
|
|
|
|
RUNS = dict({f"s84 {k}": (lambda v=v: s84.run(**v))
|
|
for k, v in s84.SCENARIOS.items()},
|
|
**{f"s87 {k}": (lambda v=v: s87_run(v))
|
|
for k, v in s87.SCENARIOS.items()})
|
|
|
|
|
|
def s87_run(v):
|
|
v = dict(v)
|
|
return s84.run(1, dur=s87.DUR, store=s87.store(v.pop("store", None)), **v)
|
|
|
|
|
|
_moved, _gone = [], 0
|
|
for k, fn in RUNS.items():
|
|
new, old = fn()[0], no_fix(fn)[0]
|
|
same = all(a[:-1] == b[:-1] and a[-1][:4] == b[-1][:4]
|
|
for a, b in zip(new, old)) and len(new) == len(old)
|
|
logs = {e for r in old for e in [r[-1][5]] if e} - {
|
|
e for r in new for e in [r[-1][5]] if e}
|
|
_gone += len(logs)
|
|
if not same or any("OUTPUT_CLAMP episode" not in e for e in logs):
|
|
_moved.append(k)
|
|
print(f" {k}: commands and claims "
|
|
f"{'the same' if same else 'DIFFER'}, log rows differ {len(logs)}, "
|
|
f"digest {s84.digest(new)[:16]}")
|
|
check("⭐ in all twelve, every command, loop field, claim, guard decision "
|
|
"and phase is the same as without the fix; the only log rows that "
|
|
"differ are OUTPUT_CLAMP rows (gone, or a lower count)",
|
|
not _moved and _gone > 0)
|
|
|
|
print()
|
|
if fails:
|
|
print(f"FAILED: {len(fails)} check(s)")
|
|
for f in fails:
|
|
print(f" - {f}")
|
|
sys.exit(1)
|
|
print("all SAFETY-119 checks passed")
|