adminandClaude Opus 5.5
f967073cfe
- The alarm must hold for peakguard.ALARM_HOLD_S (10 s) on the projection's
clock. A load that tops cap + battery for ~5 s every 30 s alarmed at every
quarter start while the quarter landed under the aim (sine4500_fast).
- A cycle with no battery reading leaves the hold alone.
- Notified once per quarter, but the ERROR line is written per episode.
- New sim scenarios (sine4500_fast, twice7000, step6500_gap) with 5a4cdc2
digests; control still byte-identical. Unit checks for export and the
hold across a missing reading. test_peakguard's saturated fixture moves
the hold start back instead of waiting it out.
- DECISIONS and DOCS updated: costs of the hold, tolerance reasoning, the
double count in control left to a follow-up.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
172 lines
8.0 KiB
Python
172 lines
8.0 KiB
Python
"""Runnable check for SAFETY-81, no false "Load too high for battery" alarms.
|
|
`python3 test_safety81.py`
|
|
|
|
Drives the REAL Controller through tools/safety81_sim.py (a 1 s cycle on a
|
|
plant shaped like the SAFETY-79 rig, one sample in ten late).
|
|
DECISIONS/SAFETY-81-load-too-high.md has the record.
|
|
|
|
1. Every command is byte-identical to release/1.0 5a4cdc2.
|
|
2. No alarm on the quarter-edge spike, the last seconds, noise (margin 50
|
|
and 0) or a burst the quarter absorbs.
|
|
3. Real overloads still alarm, ALARM_HOLD_S after 5a4cdc2.
|
|
4. peakguard.load_too_high() on its own.
|
|
5. The hold across a cycle with no battery reading (review).
|
|
|
|
Mutations (MUTATION.md, purged copies) are listed in the pull request.
|
|
"""
|
|
|
|
import asyncio
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timedelta, timezone
|
|
from types import SimpleNamespace
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, os.path.join(HERE, "tools"))
|
|
|
|
import safety81_sim as sim # noqa: E402
|
|
from app.main import Controller # noqa: E402
|
|
from app.peakguard import load_too_high # noqa: E402
|
|
|
|
fails = []
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
# Printed by `python3 tools/safety81_sim.py` on release/1.0 5a4cdc2 (the tool
|
|
# copied in, nothing else changed), in the CI image: (alarm times, digest).
|
|
BASE = {
|
|
"const4500": ([2699.0], "6ba6b8634f0df9526adc04f03c82db7d083973268deda0d1f07c5519bc2e5e9d"),
|
|
"const4500_m0": ([2699.0], "2bede28f3ea6ba9294be134c9b9733e43d22a7c24deac3bf8210bbbbe09b2039"),
|
|
"sine4500": ([2699.0], "bf70c2c23a0eede40c34478d1354b7daa1858e0a0f920693e2dd15fb5046120f"),
|
|
"edge4600": ([2.0, 902.0, 1802.0, 2699.0], "3cc0c6fec0c70f138530955babc2b416fe7307373037c1b2e927b6ccdaad3a4d"),
|
|
"edge4600_m0": ([2.0, 902.0, 1802.0, 2699.0], "ba9ba2c042ee42b043315b996ece38b3478dc9866aaf833256c9c29c02f0fe9f"),
|
|
"burst_end": ([881.0, 1781.0, 2681.0], "5f70e1eb4244820f1d44b494cb80e51f1a4486dee03357c9a8e8f7d371c5d9af"),
|
|
"noise5000_m50": ([292.0, 2551.0, 2698.0], "48b446e1a78e6d6032b31cb68a4eadb0f74e3aec4cc1e184162e9c65449bfdc5"),
|
|
"noise5000_m0": ([292.0, 2552.0, 2698.0], "1a0527dee6c2b03dc95a90b4682858c7bf5edf7205c2aced249112f04328c960"),
|
|
"step6500": ([1201.0, 1801.0], "95e5c90e4a6814526cb4f3bea76286063b449d86a420f888c014531fb6e913a3"),
|
|
"miss5560": ([1201.0, 1801.0], "cd991cfe1685750598236701a5a2f2f0342d66645609c536d42ab7ba07216e6c"),
|
|
"ramp6500": ([1296.0, 1801.0], "8db7675535f1a01abf38bd6bfb996a6fe57711d3ecca53541bb1c0870748c46b"),
|
|
"step7500_reserve": ([], "21eb9c07a4582f47f347084d1c75a28a5d767bc068bc3f74f32433d3239e9ccb"),
|
|
# Added in review, printed the same way.
|
|
"sine4500_fast": (["112, about one per 30 s"], "710008fd8f2d33a0e2cbb9a626bedafd6d87bc23712bbc5b74a4af9b3ec21897"),
|
|
"twice7000": ([1001.0, 1502.0, 2699.0], "1cddda8cf35859efb2ad503f1683f744cb812a6325a390c4fa3df19d717af043"),
|
|
"step6500_gap": ([1201.0, 1801.0], "95e5c90e4a6814526cb4f3bea76286063b449d86a420f888c014531fb6e913a3"),
|
|
}
|
|
OUT = {k: sim.run(**v) for k, v in sim.SCENARIOS.items()}
|
|
RUN = {k: v[:2] for k, v in OUT.items()}
|
|
ERR = {k: v[4] for k, v in OUT.items()}
|
|
|
|
# ============================================================================ #
|
|
print("1. every command byte-identical to release/1.0 5a4cdc2")
|
|
check("the same scenarios as 5a4cdc2", set(RUN) == set(BASE))
|
|
for k, (alarms, digest) in RUN.items():
|
|
check(f"{k}: digest {digest[:12]}", digest == BASE[k][1])
|
|
|
|
# ============================================================================ #
|
|
print("2. no false alarm (5a4cdc2's are listed)")
|
|
for k in ("edge4600", "edge4600_m0", "const4500", "const4500_m0", "sine4500",
|
|
"noise5000_m50", "noise5000_m0", "burst_end", "sine4500_fast"):
|
|
check(f"{k}: none (5a4cdc2: {BASE[k][0]})", RUN[k][0] == [])
|
|
|
|
# ============================================================================ #
|
|
print("3. real overloads still alarm, ALARM_HOLD_S after 5a4cdc2")
|
|
# ALARM_HOLD_S (10 s) plus up to 2 s of late meter samples; at 1800 s the
|
|
# quarter-edge tick restarts the hold.
|
|
for k, want in (("step6500", [1211.0, 1813.0]),
|
|
("miss5560", [1213.0, 1813.0])):
|
|
print(f" {k}: load steps up at 1200 s; alarms at {RUN[k][0]}, "
|
|
f"5a4cdc2 {BASE[k][0]}")
|
|
check(f"{k}: after the hold, and again only at the next quarter",
|
|
RUN[k][0] == want)
|
|
check("step6500_gap: cycles stopped by a lost battery reading do not "
|
|
"restart the hold",
|
|
RUN["step6500_gap"][0] == [1215.0, 1813.0])
|
|
check("twice7000: notified once in the quarter (5a4cdc2: "
|
|
f"{BASE['twice7000'][0]})", RUN["twice7000"][0] == [1011.0])
|
|
check("twice7000: the ERROR line for both overloads",
|
|
ERR["twice7000"] == [1011.0, 1512.0])
|
|
# By design: until 1306 s (then the hold) this quarter would end at most
|
|
# ALARM_MISS_W over the real 3500 W cap; 5a4cdc2 alarmed at the aim (cap - 50).
|
|
print(f" ramp6500: alarms at {RUN['ramp6500'][0]}, "
|
|
f"5a4cdc2 {BASE['ramp6500'][0]}")
|
|
check("ramp6500: alarms once the quarter will end over the real cap by "
|
|
"more than ALARM_MISS_W",
|
|
RUN["ramp6500"][0] == [1316.0, 1813.0])
|
|
check("step7500 under the peak reserve: not this alarm (the guard is not "
|
|
"allowed to discharge)", RUN["step7500_reserve"][0] == [])
|
|
|
|
# ============================================================================ #
|
|
print("4. load_too_high()")
|
|
|
|
|
|
def proj(elapsed, done_avg, held):
|
|
"""A projection `elapsed` s in, averaging `done_avg` so far."""
|
|
r = 900.0 - elapsed
|
|
return SimpleNamespace(elapsed_s=elapsed, held_w=held,
|
|
projected_w=(done_avg * elapsed + max(held, 0) * r)
|
|
/ 900.0)
|
|
|
|
|
|
def ltl(p, batt, cap=3500.0, rail=2000.0):
|
|
return load_too_high(p, batt_w=batt, ceiling_w=cap, achievable_w=rail)
|
|
|
|
|
|
check("load 6000 W vs 3500 + 2000: yes", ltl(proj(300, 3450, 4000), 2000))
|
|
check("already over, the battery can carry the load but not catch up: yes",
|
|
ltl(proj(450, 7200, 7200), 0, rail=5000))
|
|
check("the last second, a 7 W miss (under ALARM_MISS_W): no",
|
|
not ltl(proj(899, 3509, 2500), 1000))
|
|
check("a miss of exactly ALARM_MISS_W: no",
|
|
not ltl(proj(450, 3520, 3500), 2000))
|
|
check("...and 1 W more: yes", ltl(proj(450, 3522, 3500), 2000))
|
|
check("over cap + battery, but the quarter lands under the cap: no",
|
|
not ltl(proj(880, 3450, 4000), 2000))
|
|
check("...and when it does not: yes", ltl(proj(700, 3450, 4000), 2000))
|
|
check("no battery reading: no", not ltl(proj(300, 3450, 5000), None))
|
|
check("no seconds left: no", not ltl(proj(900, 4000, 5000), 2000))
|
|
check("a quarter already over is not won back by export: yes",
|
|
ltl(proj(800, 3950, 1000), 0))
|
|
|
|
# ============================================================================ #
|
|
print("5. the hold across a cycle with no battery reading")
|
|
# A reading the stale detector drops (STALE_READING) reaches the alarm as None
|
|
# while the cycle still runs. Unknown is not "fine": it must not restart the
|
|
# hold. Driven on Controller.check_peak_saturated() itself, with stubs.
|
|
Q0 = datetime(2026, 9, 20, 11, 0, tzinfo=timezone.utc)
|
|
notes = []
|
|
|
|
|
|
async def _call(domain, service, data):
|
|
notes.append(data.get("notification_id"))
|
|
|
|
|
|
stub = SimpleNamespace(
|
|
peak_guard=SimpleNamespace(saturated=True, ceiling_w=3500.0,
|
|
reason="PEAK_GUARD: fixture"),
|
|
site_rating_w=2000.0, shadow=False, events=[],
|
|
_peak_batt_prev=None, _peak_alarm_block=None, _peak_high_since=None,
|
|
_peak_saturated=False, hass=SimpleNamespace(call=_call),
|
|
log_event=lambda msg, hint=None: None)
|
|
first = None
|
|
for sec in range(0, 30):
|
|
stub.site_batt_w = None if 4 <= sec < 8 else 2000.0
|
|
p = proj(300 + sec, 3450, 6000)
|
|
p.block_start = Q0
|
|
stub._peak_proj = p
|
|
asyncio.run(Controller.check_peak_saturated(stub))
|
|
if notes and first is None:
|
|
first = sec
|
|
check(f"a 4 s gap in the battery reading: alarm at {first} s, not 18 s",
|
|
first == 10)
|
|
|
|
print()
|
|
if fails:
|
|
print(f"FAILED: {len(fails)} check(s)")
|
|
sys.exit(1)
|
|
print("all checks passed")
|