glenn schrooyenandClaude
065792aad3
Conflicts (5 files, 9 hunks) resolved keeping both sides:
- p1.py: held_t (OPT-15) beside held_since (SAFETY-90).
- modetrade.py: chargers() takes targets (P4) and probe (OPT-15); decide()
clears the chargers, a probing refuser included, while a battery drains
(SAFETY-78); the docstring lists both rules.
- main.py: the roll goes into the SAFETY-90/91 projection chain
(`_peak_proj`); the SAFETY-95 drain-tail grid_w keeps the run-edge hold on
its other branch; the P4 forecast state and the hold state both stay.
- DOCS.md, DECISIONS: both lines kept, OPT-15 item refreshed.
Adaptations the merge needs:
- SAFETY-81 x F1: a rolled projection is not graded by the saturation alarm
until the quarter's first sample, as the old edge tick was not, so a
standing overload is still notified once per quarter.
- test_safety81, test_modes_p4: modes-on pins re-pinned (with a note), modes
off pinned to e3e4ba6; test_price_run_fixes: identity pins on e3e4ba6,
F1 reproduced at one inverter (SAFETY-91 already holds two), tests for the
drain and the day plan against the probe.
Co-Authored-By: Claude <noreply@anthropic.com>
212 lines
11 KiB
Python
212 lines
11 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"),
|
|
}
|
|
# OPT-15: with battery modes ON the guard no longer reads "no seconds left" on
|
|
# a quarter's edge tick (it projects the quarter it is in), so the 1 s 0 W
|
|
# command at each edge is gone and every run differs from 5a4cdc2's. Printed by
|
|
# `python3 tools/safety81_sim.py` on the OPT-15 head (BASE above stays as the
|
|
# reference). Modes OFF stays byte-identical (section 1b).
|
|
OPT15 = {
|
|
"const4500": "ed2f4ef9a74172ada2f00707a09ad839743449642f61dc6583ace44bac33caf4",
|
|
"const4500_m0": "63d4d62643b9d39697b16078603576d180d90090117a497122e77c33f3db2227",
|
|
"sine4500": "a034af0a896d55552915f1e4816055e26c006d4df0a86d1c9d9c3a625faf747f",
|
|
"edge4600": "74e03c7182469715c1517765c2f53f1d68d7342aed4a8ad388878eac24964020",
|
|
"edge4600_m0": "93e4402b6b879a44da5ecc28fb3b51b16c0b0f701396e00a9703803273dbd18e",
|
|
"burst_end": "8dc4001e06ad8fa562caea02d755227cdfa16396708fa986aef7a5704e22a691",
|
|
"noise5000_m50": "71e29a970837211835e56636280c29f70d7c2d3f3062cfac2b940ea30fa66424",
|
|
"noise5000_m0": "a47988b7bbee629a6f8722588f325a4c87d7dcb143bba549b6df452664b30251",
|
|
"step6500": "a7356fe0f8c00ac4d16dd4f60b054af7336ceb2eedcd76911d30138df4b457b8",
|
|
"miss5560": "2b0afb600fe1cfd0bfe4b28fbfeeff73fc7e875e3d5f377ae7c2a489e20abf53",
|
|
"ramp6500": "ae3838aa8f2af9458f73b540eeb76cb5fa25ace0f1ea06df2b4f6d9201907522",
|
|
"step7500_reserve": "21eb9c07a4582f47f347084d1c75a28a5d767bc068bc3f74f32433d3239e9ccb",
|
|
"sine4500_fast": "47ed9d02a4f7afbc50905f44519b14364f262ba793aae80f3cd077eb6f600b8b",
|
|
"twice7000": "b89396022f589bbb90bc592ece121cf2a0bf3a4f04316da36891ac1a2b6ae18e",
|
|
"step6500_gap": "a7356fe0f8c00ac4d16dd4f60b054af7336ceb2eedcd76911d30138df4b457b8",
|
|
}
|
|
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]} (OPT-15 pin; modes off in 1b)",
|
|
digest == OPT15[k])
|
|
|
|
# ============================================================================ #
|
|
print("1b. battery modes off: every command byte-identical to release/1.0 e3e4ba6")
|
|
# Printed on e3e4ba6 (this sim copied in, `modes=False`), in the CI image.
|
|
OFF = {
|
|
"const4500": "719d8bfa450710d9c0c22282d17ea4ab9945b1cb53ce344084e5ff906ca027b3",
|
|
"edge4600": "4f8f943f5a4e9b218c2e8bc6a95d20934ec19f1bebfbdc5748ca52e753e5afa0",
|
|
"step6500": "719d8bfa450710d9c0c22282d17ea4ab9945b1cb53ce344084e5ff906ca027b3",
|
|
"burst_end": "4ca4ba895afe7dacfe3952743e13d21607e690dfec39029fd5c5745fe03af5d7",
|
|
"twice7000": "719d8bfa450710d9c0c22282d17ea4ab9945b1cb53ce344084e5ff906ca027b3",
|
|
"sine4500_fast": "3e41ad417880a4ebb63ae3d6826f413fb779c18dfb677207775b1c980133423f",
|
|
}
|
|
for k, want in OFF.items():
|
|
digest = sim.run(modes=False, **sim.SCENARIOS[k])[1]
|
|
check(f"{k} (modes off): digest {digest[:12]}", digest == want)
|
|
|
|
# ============================================================================ #
|
|
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 (OPT-15: with modes on, by not grading
|
|
# a projection that rolled into the new quarter before its first sample).
|
|
for k, want in (("step6500", [1211.0, 1811.0]),
|
|
("miss5560", [1213.0, 1811.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, 1811.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] == [1317.0, 1811.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_rolled=False,
|
|
_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")
|