glenn schrooyenandClaude Opus 5.5
7c4c8f5c66
New option peak_margin_w (default 50 W, 0-500 W). The peak guard, the
forecast checks and the maintenance charge sizing aim at
peak_cap_w - peak_margin_w; the month peak, the bill, headroom, warnings
and every notice keep the real cap. Owner decision 2026-09-28.
peak_margin_w 0 is byte-identical to ed739d4 (tools/safety79_identity.py,
test_peak_margin.py part 1). validate.py rule 12 refuses a value outside
0-500, not whole, or not below the peak limit.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UBtqPrG1mDLXzTEBnhEkkV
115 lines
3.7 KiB
Python
115 lines
3.7 KiB
Python
#!/usr/bin/env python3
|
|
"""SAFETY-79: with `peak_margin_w: 0`, is every command byte-identical to before?
|
|
|
|
python3 tools/safety79_identity.py # prints one digest per scenario
|
|
|
|
tools/opt11_identity.py's harness (the REAL Controller, read_inputs + cycle, in
|
|
simulated time) with a P1 meter, so the peak guard runs, and a peak limit the
|
|
house goes over. Each row hashes opt11_identity's fields plus the guard's floor,
|
|
limit and sentence. It uses only the API release/1.0 ed739d4 already had, so
|
|
the same file runs on the old tree (which ignores `peak_margin_w`) and the new.
|
|
test_peak_margin.py part 1 holds the new tree to the digests printed on ed739d4.
|
|
"""
|
|
|
|
import functools
|
|
import hashlib
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from datetime import timedelta
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
import opt11_identity as rig # noqa: E402
|
|
from app import packmode # noqa: E402
|
|
from app.p1 import make_sample, split_signed # noqa: E402
|
|
from app.store import unit_key # noqa: E402
|
|
|
|
CAP_W = 2500
|
|
|
|
|
|
def load(t):
|
|
"""A house under the limit, over it, a solar surplus, far over, over."""
|
|
if t < 300:
|
|
return 800.0
|
|
if t < 900:
|
|
return 3200.0
|
|
if t < 1500:
|
|
return -2600.0
|
|
if t < 2100:
|
|
return 6000.0
|
|
return 4200.0
|
|
|
|
|
|
def soc(u, t):
|
|
return 60.0 # above the peak reserve: the guard may discharge
|
|
|
|
|
|
def _feed(c, t):
|
|
"""One P1 telegram per cycle: the meter reads what the house and the
|
|
batteries' measured power leave."""
|
|
imp, exp = split_signed(c.hass.grid())
|
|
c.p1.submit(make_sample("ha_signed", imp, exp, phases=1,
|
|
ingest_ts=rig.BASE + timedelta(seconds=t),
|
|
ingest_mono=time.monotonic()))
|
|
|
|
|
|
def store_off(name):
|
|
"""A store where battery `name` is switched Off (OPT-11)."""
|
|
s = rig._Store()
|
|
s.data[unit_key(name, packmode.USE_KEY)] = False
|
|
return s
|
|
|
|
|
|
P1 = {"meter_source": "ha_signed", "p1_net_entity": "sensor.net",
|
|
"meter_phases": 1, "peak_cap_w": CAP_W}
|
|
MODES = {"battery_modes_enabled": True}
|
|
|
|
SCENARIOS = {
|
|
"n1": dict(n=1),
|
|
"n2": dict(n=2),
|
|
# OPT-11: the only battery Off, so only the peak guard makes it give.
|
|
"n1_off": dict(n=1, modes=True, off="inverter-a"),
|
|
# OPT-11: B Off, so B covers what A cannot.
|
|
"n2_off": dict(n=2, modes=True, off="inverter-b"),
|
|
}
|
|
|
|
|
|
def run(n, *, margin=None, modes=False, off=None, dur=2700.0):
|
|
"""Rows for one scenario; `margin=None` leaves the key out."""
|
|
extra = dict(P1, **(MODES if modes else {}))
|
|
if margin is not None:
|
|
extra["peak_margin_w"] = margin
|
|
guard = []
|
|
|
|
def on_cycle(c, t):
|
|
g = c.peak_guard
|
|
guard.append(None if g is None else (g.engaged, g.floor_w, g.ceiling_w,
|
|
g.ceiling_from, g.reason))
|
|
db = os.path.join(tempfile.mkdtemp(prefix="s79-"), "tariff.db")
|
|
real = rig.Controller
|
|
rig.Controller = functools.partial(real, peak_db_path=db)
|
|
try:
|
|
rows, c = rig.run(n, extra=extra, dur=dur, load=load, soc=soc,
|
|
cut=lambda u, t: False, before=_feed,
|
|
on_cycle=on_cycle,
|
|
store=store_off(off) if off else None)
|
|
finally:
|
|
rig.Controller = real
|
|
return [r + (g,) for r, g in zip(rows, guard)], c
|
|
|
|
|
|
def digest(rows) -> str:
|
|
return hashlib.sha256(repr(rows).encode()).hexdigest()
|
|
|
|
|
|
def digests(margin=None) -> dict:
|
|
return {k: digest(run(**v, margin=margin)[0]) for k, v in SCENARIOS.items()}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
m = None if len(sys.argv) < 2 else int(sys.argv[1])
|
|
for k, v in digests(m).items():
|
|
print(f"{k} {v}")
|