glenn schrooyenandClaude Opus 5.5
4b81387908
The loop's integrator bound counts only on a cycle the loop decided. A battery another owner set (maintenance, the peak guard, a safety limit) alarms only when it gives more than saturation_w less than told for saturation_cycles cycles. At N>=2 the loop is bounded while maintenance holds a battery, so it does not wind and control comes back one slew step a cycle. The status row counts episodes, not cycles. Digests re-pinned where behaviour changed on purpose; DECISIONS/SAFETY-85-no-false-not-keeping-up.md has the list. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019DBUfvndgHwaV339Fvd2gr
312 lines
14 KiB
Python
312 lines
14 KiB
Python
"""Runnable check for SAFETY-79, the margin under the peak limit.
|
|
`python3 test_peak_margin.py`
|
|
|
|
The peak guard aims at `peak_cap_w - peak_margin_w` (default 50 W); the bill
|
|
and every report keep the real `peak_cap_w`. DECISIONS/SAFETY-79-peak-margin.md
|
|
has the record. Drives the REAL Controller through tools/safety79_identity.py.
|
|
|
|
1. `peak_margin_w: 0` is byte-identical to release/1.0 ed739d4.
|
|
2. The guard aims at cap - margin at N=1 and N=2, and with OPT-11 modes on an
|
|
Off battery covers the shortfall to cap - margin.
|
|
3. The bill and the reports still use the real cap.
|
|
4. Each call site in main.py passes the value its side needs.
|
|
5. The setting: default, range, refusals in plain words.
|
|
|
|
Mutations (MUTATION.md, purged copies) are listed in the pull request.
|
|
"""
|
|
|
|
import ast
|
|
import os
|
|
import re
|
|
import sys
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, os.path.join(HERE, "tools"))
|
|
|
|
import safety79_identity as rig # noqa: E402
|
|
from named_absence import nth, present # noqa: E402
|
|
from style_rules import text_problems # noqa: E402
|
|
from app import messages, peakguard # noqa: E402
|
|
from app.control import maintenance_charge_floor # noqa: E402
|
|
from app.peakguard import (PEAK_MARGIN_RANGE_W, # noqa: E402
|
|
PEAK_OPTION_DEFAULTS, quarter_target_w)
|
|
from app.validate import config_notes, validate_startup_config # noqa: E402
|
|
|
|
fails = []
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
CAP = float(rig.CAP_W)
|
|
|
|
# ============================================================================ #
|
|
print("1. peak_margin_w 0: byte-identical to release/1.0 ed739d4")
|
|
# Printed by `python3 tools/safety79_identity.py 0` on the ed739d4 tree (the
|
|
# tool copied in, nothing else changed; that tree ignores the key), in the CI
|
|
# image; re-printed 2026-09-29 (review) after adding the charge headroom, the
|
|
# shaping suspension and the two maintenance scenarios. 540 cycles each.
|
|
ED739D4 = {
|
|
"n1": "82316f7d4362fae4c035a1bed5e7907cf362faba994c634d0b1c356c501f53e5",
|
|
"n2": "0dc24e16a8390ec42edacf735eaf20273847b33b9c67d7cdc536df0ca739e421",
|
|
"n1_off": "eed44f058f3fcc6e7651c5c84c6fa6313e7520035ad7cac282012cf6909668a0",
|
|
"n2_off": "69796c107b9f78892e04d282b754f889590acf9fe1b96f09525f5ea2005e9c0a",
|
|
"n1_maint":
|
|
"bd4a88a10aa840a360ac781ac04e30b040a5aac5426fabba1fb259d71dfeba67",
|
|
"n2_maint":
|
|
"0656bc5a4a2fa3a7455822fdd7da5f68f57fd9593f374e869a0d30ba7e429275",
|
|
}
|
|
_zero = rig.digests(0)
|
|
_default = rig.digests()
|
|
for k in ED739D4:
|
|
print(f" {k:7} margin 0 {'same' if _zero[k] == ED739D4[k] else 'DIFFERS'}"
|
|
f" {_zero[k][:16]}, default {_default[k][:16]}")
|
|
# SAFETY-84 changes n2_maint ON PURPOSE: on 9d5b7d1 its N=2 cheap-window
|
|
# charge was refused on 214 of 540 cycles (site at 0 W); now none is.
|
|
# Printed on the SAFETY-84 tree the same way (margin 0).
|
|
# SAFETY-85: the loop is bounded while A charges (reasons change, commands
|
|
# do not). Was 95722b262f1bcf504493d7b5babc3895799ac56cc15f5d2ccdbe7c9f04adfcbb.
|
|
SAFETY84_N2_MAINT = \
|
|
"ef97ae2e13668b3e0b2d22f3f5f37835ea46cd306c2cc475c97e284f9dcc6545"
|
|
check("⭐ at N=1 and N=2, with and without battery modes and with a "
|
|
"maintenance charge, margin 0 is "
|
|
"byte-identical to ed739d4 (n2_maint apart, below)",
|
|
{k: v for k, v in _zero.items() if k != "n2_maint"}
|
|
== {k: v for k, v in ED739D4.items() if k != "n2_maint"})
|
|
check("...and n2_maint at margin 0 is the SAFETY-84 digest: the N=2 charge "
|
|
"that 9d5b7d1 refused on 214 of 540 cycles now runs (test_safety84.py)",
|
|
_zero["n2_maint"] == SAFETY84_N2_MAINT != ED739D4["n2_maint"])
|
|
check("...and the default (key absent) changes every scenario, so the "
|
|
"scenarios do reach the margin", all(_default[k] != ED739D4[k]
|
|
for k in ED739D4))
|
|
|
|
# ============================================================================ #
|
|
print("\n2. the guard aims at cap - margin")
|
|
_runs = {}
|
|
for name, spec in rig.SCENARIOS.items():
|
|
for m in (None, 0, 120):
|
|
_runs[name, m] = rig.run(**spec, margin=m, dur=3600.0)
|
|
for name in rig.SCENARIOS:
|
|
g = _runs[name, None][1].peak_guard
|
|
print(f" {name}: margin {_runs[name, None][1].peak_margin_w:.0f} W, "
|
|
f"aims at {g.target_w:.0f} W, limit {g.ceiling_w:.0f} W")
|
|
check("test_target_is_cap_minus_margin: with the key absent every scenario "
|
|
"(N=1, N=2, modes on and off) aims at cap - 50 and reports the cap",
|
|
all(_runs[n, None][1].peak_margin_w == 50.0
|
|
and _runs[n, None][1].peak_guard.target_w == CAP - 50
|
|
and _runs[n, None][1].peak_guard.ceiling_w == CAP
|
|
for n in rig.SCENARIOS))
|
|
check("...and a margin of 120 moves the aim to cap - 120, a margin of 0 to "
|
|
"the cap", all(_runs[n, 120][1].peak_guard.target_w == CAP - 120
|
|
and _runs[n, 0][1].peak_guard.target_w == CAP
|
|
for n in rig.SCENARIOS))
|
|
|
|
|
|
def banked(ctl):
|
|
return {b.start.strftime("%H:%M"): round(b.offtake_avg_w, 2)
|
|
for b in ctl.p1.blocks}
|
|
|
|
|
|
_b = {m: banked(_runs["n1_off", m][1]) for m in (0, None, 120)}
|
|
print(f" N=1, the battery Off: quarters banked {_b}")
|
|
check("test_off_battery_n1: at N=1 with the battery Off, the guard makes it "
|
|
"give just enough to bank the 11:30 quarter at cap - margin (2500, "
|
|
"2450, 2380 W for margins 0, 50, 120)",
|
|
[_b[m]["11:30"] for m in (0, None, 120)]
|
|
== [CAP, CAP - 50, CAP - 120])
|
|
|
|
|
|
def meter_at(rows, t):
|
|
row = next((r for r in rows if r[0] == t), None)
|
|
if row is None:
|
|
return None, {}
|
|
return rig.load(t) - sum(w for _u, w in row[2]), dict(row[2])
|
|
|
|
|
|
_g0, _c0 = meter_at(_runs["n2_off", 0][0], 1800.0)
|
|
_g50, _c50 = meter_at(_runs["n2_off", None][0], 1800.0)
|
|
print(f" N=2, B Off, 6000 W house: margin 0 -> {_c0}, meter {_g0} W; "
|
|
f"margin 50 -> {_c50}, meter {_g50} W")
|
|
check("test_off_battery_n2: at N=2 with B Off and A at its rating, B covers "
|
|
"the shortfall to cap - margin, not to the cap",
|
|
_c0["b"] == 1000.0 and _g0 == CAP
|
|
and _c50["b"] == 1050.0 and _g50 == CAP - 50)
|
|
|
|
def _peak(w, observed=True):
|
|
return type("R", (), {"is_observed": observed, "peak_w": w})()
|
|
|
|
|
|
# Review (SAFETY-79 point 2): the margin applies under a month peak already
|
|
# banked, too. Aiming AT it can bank a few W over it, which raises the bill;
|
|
# and a banked peak between cap - margin and the cap (a quarter the old guard
|
|
# held at the cap, say) was aimed at exactly, so one overshoot crossed the cap.
|
|
check("test_margin_under_banked_peak: the aim is the margin under the limit "
|
|
"that holds - the cap, or a month peak already banked above it",
|
|
quarter_target_w(CAP, 50, _peak(2499.0)) == CAP - 50
|
|
and quarter_target_w(CAP, 50, _peak(3000.0)) == 2950.0
|
|
and quarter_target_w(CAP, 50, _peak(3000.0, observed=False)) == CAP - 50
|
|
and quarter_target_w(CAP, 50, None) == CAP - 50
|
|
and quarter_target_w(CAP, 0, _peak(3000.0)) == 3000.0)
|
|
|
|
|
|
def _decide(banked_w, projected_w):
|
|
"""Half a quarter at `projected_w`, battery idle, a month peak banked."""
|
|
from datetime import datetime, timezone
|
|
from app.tariff import Projection
|
|
pr = Projection(projected_w, projected_w, 450.0, 1.0, False,
|
|
datetime(2026, 9, 1, 12, 0, tzinfo=timezone.utc),
|
|
projected_w)
|
|
return peakguard.decide(pr, batt_w=0.0, cap_w=CAP, margin_w=50,
|
|
reading=_peak(banked_w), achievable_w=5000.0,
|
|
soc=60.0, soc_floor=20.0)
|
|
|
|
|
|
_d1, _d2 = _decide(2499.0, 2600.0), _decide(3000.0, 3100.0)
|
|
print(f" banked 2499 W: floor {_d1.floor_w}, limit {_d1.ceiling_w}; "
|
|
f"banked 3000 W: floor {_d2.floor_w}, limit {_d2.ceiling_w}")
|
|
check("...and decide() sizes the floor to land the quarter 50 W under it "
|
|
"(2449 and 2950 W), while the reported limit stays the cap or the peak",
|
|
_d1.floor_w == 2 * (2600.0 - 2450.0) and _d1.ceiling_w == CAP
|
|
and _d2.floor_w == 2 * (3100.0 - 2950.0) and _d2.ceiling_w == 3000.0)
|
|
|
|
# ============================================================================ #
|
|
print("\n3. the bill and the reports keep the real cap")
|
|
_c = _runs["n1_off", None][1]
|
|
_t = rig.rig.BASE # the runs' clock, not the wall clock (a new month)
|
|
_pk = _c.peak.peak(_t).peak_w
|
|
print(f" month peak {_pk:.0f} W, tracker cap {_c.peak.cap_w:.0f} W, "
|
|
f"warning {_c.peak.warning_frac(_t):.3f}, headroom "
|
|
f"{_c.peak.headroom_w(_t):.0f} W")
|
|
check("test_billing_uses_cap: the month-peak tracker (bill, headroom, 80 % "
|
|
"warning) is built on the real cap, not cap - margin",
|
|
_c.peak.cap_w == CAP and _c.peak.warning_frac(_t) == _pk / CAP
|
|
and _c.peak.headroom_w(_t) == CAP - _pk)
|
|
_eng = [g for r in _runs["n1_off", None][0] for g in (r[-1][1],)
|
|
if g and g[0] and g[1]]
|
|
print(f" {len(_eng)} engaged cycles; one: {_eng[0][4] if _eng else None}")
|
|
check("test_reports_use_cap: every engaged sentence names the real limit and "
|
|
"the aim beside it", _eng and all(
|
|
g[2] == CAP and "against a 2500 W limit (peak_cap_w), aiming at "
|
|
"2450 W" in g[4] for g in _eng))
|
|
check("...and the settings note compares the month peak with the real cap: "
|
|
"a 2480 W peak under a 2500 W cap is no warning, 2520 W is",
|
|
config_notes({"peak_cap_w": 2500, "peak_margin_w": 50},
|
|
month_peak_w=2480) == []
|
|
and [n.keys for n in config_notes({"peak_cap_w": 2500,
|
|
"peak_margin_w": 50},
|
|
month_peak_w=2520)]
|
|
== [("peak_cap_w",)])
|
|
_lim = _eng[0][2] if _eng else 0.0
|
|
_notice = messages.peak_saturated(_lim)
|
|
check("...and the user-facing peak notice built from the decision's limit "
|
|
"says the cap", messages.watts(CAP) in _notice.body
|
|
and messages.watts(CAP - 50) not in _notice.body)
|
|
|
|
# ============================================================================ #
|
|
print("\n4. each call site in main.py")
|
|
with open(os.path.join(HERE, "app", "main.py"), encoding="utf-8") as f:
|
|
_src = f.read()
|
|
_tree = ast.parse(_src)
|
|
|
|
|
|
def calls(name):
|
|
return [n for n in ast.walk(_tree) if isinstance(n, ast.Call)
|
|
and isinstance(n.func, (ast.Attribute, ast.Name))
|
|
and getattr(n.func, "attr", getattr(n.func, "id", "")) == name]
|
|
|
|
|
|
def kw(call, key):
|
|
if not present(call):
|
|
return "absent"
|
|
return next((ast.get_source_segment(_src, k.value) for k in call.keywords
|
|
if k.arg == key), "absent")
|
|
|
|
|
|
def arg(call, i):
|
|
if not present(call) or len(call.args) <= i:
|
|
return "absent"
|
|
return ast.get_source_segment(_src, call.args[i])
|
|
|
|
|
|
_dec = [c for c in calls("decide")
|
|
if getattr(c.func, "value", None) is not None
|
|
and getattr(c.func.value, "id", "") == "peakguard"]
|
|
_shape = calls("shaping_suspended")
|
|
_maint = calls("maintenance_charge_floor")
|
|
_tracker = calls("PeakTracker")
|
|
_d, _s, _m, _t = (nth(_dec, 0, "peakguard.decide()"),
|
|
nth(_shape, 0, "shaping_suspended()"),
|
|
nth(_maint, 0, "maintenance_charge_floor()"),
|
|
nth(_tracker, 0, "PeakTracker()"))
|
|
print(f" decide(cap_w={kw(_d, 'cap_w')}, margin_w={kw(_d, 'margin_w')})")
|
|
print(f" shaping_suspended(..., {arg(_s, 2)})")
|
|
print(f" maintenance_charge_floor(..., {arg(_m, 2)})")
|
|
print(f" PeakTracker(cap_w={kw(_t, 'cap_w')})")
|
|
check("test_call_sites: the guard gets the real cap AND the margin; the "
|
|
"forecast checks get cap - margin; the month-peak tracker gets the "
|
|
"real cap",
|
|
len(_dec) == 1 and "peak_cap_w" in kw(_d, "cap_w")
|
|
and kw(_d, "margin_w") == "self.peak_margin_w"
|
|
and len(_shape) == 1 and len(_maint) == 1
|
|
and arg(_s, 2) == "self.peak_aim_w()"
|
|
and arg(_m, 2) == "self.peak_aim_w()"
|
|
and len(_tracker) == 1
|
|
and kw(_t, "cap_w") == 'float(opts.get("peak_cap_w", 3500))')
|
|
check("test_aim_w: `peak_aim_w()` is cap - margin, and the forecast sizing "
|
|
"gives a maintenance charge only what is left under it",
|
|
_c.peak_aim_w() == CAP - 50
|
|
and maintenance_charge_floor(2000.0, 2400.0, _c.peak_aim_w()) == 50.0
|
|
and peakguard.shaping_suspended(None, 2470.0, _c.peak_aim_w())
|
|
and not peakguard.shaping_suspended(None, 2470.0, CAP))
|
|
|
|
# ============================================================================ #
|
|
print("\n5. the setting")
|
|
|
|
|
|
with open(os.path.join(HERE, "config.yaml"), encoding="utf-8") as f:
|
|
_hits = re.findall(r"^ peak_margin_w:\s*(\S+)", f.read(), re.M)
|
|
print(f" config.yaml: {_hits}")
|
|
check("config.yaml ships 50, int(0,500), and the in-code mirror matches",
|
|
_hits == ["50", "int(0,500)"]
|
|
and PEAK_OPTION_DEFAULTS["peak_margin_w"] == 50
|
|
and PEAK_MARGIN_RANGE_W == (0, 500))
|
|
|
|
BASE = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
|
|
"max_w": 2000, "peak_cap_w": 3500}
|
|
_base_problems = validate_startup_config(BASE)
|
|
|
|
|
|
def refused(value, **more):
|
|
opts = dict(BASE, peak_margin_w=value, **more)
|
|
return [p for p in validate_startup_config(opts)
|
|
if p not in _base_problems]
|
|
|
|
|
|
_ok = {v: refused(v) for v in (0, 50, 500)}
|
|
_bad = {v: refused(v) for v in (-1, 501, 12.5, "abc")}
|
|
for v, p in _bad.items():
|
|
print(f" {v!r}: {p}")
|
|
check("test_margin_range: 0, 50 and 500 are accepted, and so is the key "
|
|
"absent", not any(_ok.values()) and _base_problems == [])
|
|
check("...and below 0, above 500, not whole, or not a number is refused, "
|
|
"once, naming the setting and the range",
|
|
all(len(p) == 1 and "“Margin below the peak limit” (peak_margin_w)"
|
|
in p[0] and "from 0 to 500 W" in p[0] for p in _bad.values()))
|
|
_high = refused(500, peak_cap_w=500)
|
|
print(f" 500 under a 500 W limit: {_high}")
|
|
check("...and a margin that leaves nothing under the limit is refused",
|
|
len(_high) == 1 and "not below" in _high[0])
|
|
_style = [text_problems(p[0], keys=("peak_margin_w", "peak_cap_w"),
|
|
entities=False) for p in list(_bad.values()) + [_high]]
|
|
check(f"...in plain words (STYLE.md): {_style}", not any(_style))
|
|
|
|
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED: {', '.join(fails)}")
|
|
sys.exit(1)
|
|
print("all checks passed")
|