Claude (SAFETY-125 review)
6eebd4882b
- The over-the-limit rows ran over STYLE's 15 words (EN 16/19, NL 16/21) and two Dutch Home sentences over 12 words; they were not in the samples, so test_messages/test_ux46 never saw them. Now: "Peak guard: grid charge limited, at this rate 5 814 W, over 5 413 W" / "Piekbewaking: laden beperkt, aan dit tempo 5 814 W, boven 5 413 W", "...idle, at this rate ..., over ..." / "...in rust, aan dit tempo ...", NL Home "Laadt minder van het net om onder 5,4 kW te blijven." and "Zou minder van het net laden om onder 5,4 kW te blijven." The rows and the line are added to the samples. The under row is unchanged. - CHANGELOG known issues and the DECISIONS note: at N=1 the maintenance charge is not sized to the room, so SAFETY-99 still applies to it. - DOCS: the peak guard row's new words. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
409 lines
18 KiB
Python
409 lines
18 KiB
Python
"""Runnable check for SAFETY-125. `python3 test_safety125.py`
|
|
|
|
A meter reading that is late near a quarter's end left seconds the
|
|
projection did not count (SAFETY-99), and `charge_headroom_w()` read them as
|
|
room: a grid charge went back to full power in the quarter's last seconds and
|
|
the billed quarter went over the aim, above a banked month peak or the cap.
|
|
And the peak guard's row said "under" while the period was over the limit at
|
|
the current rate. DECISIONS/SAFETY-125-grid-charge-under-the-month-peak.md.
|
|
|
|
Drives the REAL Controller (read_inputs + cycle, the tariff task on a
|
|
day/night profile, a P1 meter on a simulated clock) in the live shape: modes
|
|
on, inverter-a Cheap hours at 5000 W charging harder than asked (as it did
|
|
live), inverter-b Solar first, a house of 1 kW, cap 3500 W, margin 50 W.
|
|
|
|
1. `charge_room_w()` on its own.
|
|
2. A reading 30 s late at each quarter's end: every quarter at or under the
|
|
aim, with a month peak banked above the cap and with none, with the P1
|
|
projection and with a `peak_forecast_entity`; the charge is cut late in
|
|
the quarter and back in the next one.
|
|
3. The same for a price trade with modes off, and a maintenance charge at N=2.
|
|
4. A healthy meter: the quarters as before, and the row and the month-peak
|
|
line say what is true.
|
|
5. The words on their own.
|
|
|
|
Mutations (MUTATION.md, purged copies) are listed in the pull request.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import types
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
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 opt11p3_replay as rp # noqa: E402
|
|
import safety84_identity as s84 # noqa: E402
|
|
import app.main as m # noqa: E402
|
|
import app.p1 as p1mod # noqa: E402
|
|
from named_absence import first, present # noqa: E402
|
|
from app import homeview, messages, peakguard, tariff # noqa: E402
|
|
from app import price as _price, static_tariff as _st # noqa: E402
|
|
from app.main import Controller # noqa: E402
|
|
from app.p1 import make_sample, split_signed # noqa: E402
|
|
from app.store import unit_key # noqa: E402
|
|
from app.tariff import QUARTER_S, Projection # noqa: E402
|
|
|
|
fails = []
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
PEAK = 5413.17 # October's banked peak on the live site
|
|
AIM_PEAK = PEAK - 50.0
|
|
AIM_CAP = 3500.0 - 50.0
|
|
TOL = 10.0 # W over the aim a quarter may land (SAFETY-81)
|
|
FC = "sensor.piek_prognose_kwartier"
|
|
L = timezone(timedelta(hours=1))
|
|
START = datetime(2026, 10, 29, 0, 0, tzinfo=L).astimezone(timezone.utc)
|
|
PROFILE = ('{"periods": ['
|
|
'{"name": "night", "days": "all", "from": "22:00", "to": "07:00",'
|
|
' "eur_per_kwh": 0.2333},'
|
|
'{"name": "day", "days": "all", "from": "07:00", "to": "22:00",'
|
|
' "eur_per_kwh": 0.2801}]}')
|
|
|
|
|
|
def opts(*, fc=False, modes=True, arb=False):
|
|
o = dict(rp.options(modes, arb, 2))
|
|
o.update(cycle_cost_threshold_eur=0.03, static_profile=PROFILE,
|
|
peak_cap_w=3500, peak_margin_w=50)
|
|
o["inverters"][0].update(max_w=5000, capacity_kwh=15.36)
|
|
o["inverters"][1].update(max_w=4600, capacity_kwh=20.48)
|
|
if fc:
|
|
o["peak_forecast_entity"] = FC
|
|
return o
|
|
|
|
|
|
class Harder(rp.Plant):
|
|
"""Charges harder than asked, as inverter-a did live (asked -4271 W,
|
|
measured about -4750 W; asked -5000 W, about -5210 W)."""
|
|
|
|
def step(self, dt, floor):
|
|
c = self.cmd
|
|
if c < 0:
|
|
self.cmd = max(c * 1.11, -5216.0)
|
|
try:
|
|
super().step(dt, floor)
|
|
finally:
|
|
self.cmd = c
|
|
|
|
|
|
async def _no_fetch(*a, **k):
|
|
raise AssertionError("the tariff task reached for a price feed")
|
|
|
|
|
|
def drive(o, *, peak=None, late=0.0, minutes=31, every=0):
|
|
"""The real Controller from 00:00 local for `minutes` at a 1 s cycle
|
|
(OPT-15's lag). `late`: no P1 telegram in each quarter's last `late` s.
|
|
-> (controller, rows, {quarter start: true grid average}, samples), the
|
|
samples taken every `every` cycles from the second quarter (the first
|
|
starts before the first reading): (projection, ceiling, engaged,
|
|
peak_guard row, month-peak line)."""
|
|
store = rp._Store()
|
|
for u, mode in (("a", "cheap_hours"), ("b", "solar_first")):
|
|
store.data[unit_key(f"inverter-{u}", "pack_use")] = True
|
|
store.data[unit_key(f"inverter-{u}", "pack_mode")] = mode
|
|
store.data[unit_key(f"inverter-{u}", "pack_sell")] = False
|
|
rows, samples, sums = [], [], {}
|
|
|
|
async def go():
|
|
lag = rp.Lag()
|
|
hass = rp.Hass(("a", "b"), lambda t: 1000.0, {"a": 11.0, "b": 10.0},
|
|
limits={"a": (-5000.0, 5000.0),
|
|
"b": (-4600.0, 4600.0)}, lag=lag)
|
|
hass.plants["a"].__class__ = Harder
|
|
fc_kw = [0.0]
|
|
|
|
async def state(e):
|
|
# The HA quarter projection, in kW like the live template.
|
|
return None if e != FC else {
|
|
"state": f"{fc_kw[0]:.3f}",
|
|
"attributes": {"unit_of_measurement": "kW"}}
|
|
hass.state = state
|
|
tmp = tempfile.mkdtemp(prefix="s125-")
|
|
db = os.path.join(tmp, "tariff.db")
|
|
if peak is not None:
|
|
tariff.PeakStore(db)._db.execute(
|
|
"INSERT INTO monthly_peak VALUES (?, ?, ?, NULL)",
|
|
(tariff.month_key(START), peak,
|
|
"2026-10-21T20:15:00+00:00")).connection.commit()
|
|
c = Controller(dict(o), hass, store, rp._Pub(), stop=asyncio.Event(),
|
|
peak_db_path=db,
|
|
price_history_path=os.path.join(tmp, "ph.json"))
|
|
c.inverters[0].dev_min, c.inverters[0].dev_max = -5000.0, 5000.0
|
|
c.inverters[1].dev_min, c.inverters[1].dev_max = -4600.0, 4600.0
|
|
c.auto = True
|
|
real = m.datetime, homeview.datetime, p1mod.time
|
|
clock, simt = [START], [0.0]
|
|
|
|
class _DT(real[0]):
|
|
@classmethod
|
|
def now(cls, tz=None):
|
|
return clock[0] if tz is None else clock[0].astimezone(tz)
|
|
m.datetime = homeview.datetime = _DT
|
|
p1mod.time = types.SimpleNamespace(monotonic=lambda: 1e6 + simt[0])
|
|
real_load = _price.load_prices, _st.load_prices
|
|
_price.load_prices = _st.load_prices = _no_fetch
|
|
task = asyncio.create_task(c.run_tariff_refresh())
|
|
try:
|
|
hass.advance(0.0, 10.0)
|
|
t, n = 0.0, 0
|
|
while t < minutes * 60:
|
|
now = START + timedelta(seconds=t)
|
|
hass.t, simt[0], clock[0] = t, t, now
|
|
if t == 0.0:
|
|
for _ in range(500):
|
|
if c.tariff_plan is not None:
|
|
break
|
|
await asyncio.sleep(0)
|
|
q = int(now.timestamp() // QUARTER_S)
|
|
acc = sums.setdefault(q, [0.0, 0])
|
|
g_now = hass.grid()
|
|
fc_kw[0] = (acc[0] + g_now * (QUARTER_S - acc[1])) \
|
|
/ QUARTER_S / 1000.0
|
|
if now.timestamp() % QUARTER_S < QUARTER_S - late \
|
|
and t >= lag.p1_s:
|
|
imp, exp = split_signed(hass._past(lag.meter_s
|
|
+ lag.p1_s)[0])
|
|
c.p1.submit(make_sample(
|
|
"ha_signed", imp, exp, phases=1,
|
|
ingest_ts=now - timedelta(seconds=lag.p1_s),
|
|
ingest_mono=1e6 + t))
|
|
await c.read_inputs(now_mono=t)
|
|
await c.cycle()
|
|
hass.advance(lag.tick, 10.0)
|
|
acc[0] += max(0.0, hass.grid())
|
|
acc[1] += 1
|
|
rows.append((now, c.target, hass.plants["a"].p,
|
|
getattr(c.arb, "action", None)))
|
|
pg = c.peak_guard
|
|
if every and n % every == 0 and t >= QUARTER_S \
|
|
and pg is not None and pg.projected_w is not None:
|
|
row = first((r["text"] for r in c.checks(now)
|
|
if r["id"] == "peak_guard"), "guard row")
|
|
samples.append((pg.projected_w, pg.ceiling_w, pg.engaged,
|
|
row, homeview.month_peak(c, now)["line"]))
|
|
t += lag.tick
|
|
n += 1
|
|
finally:
|
|
task.cancel()
|
|
try:
|
|
await task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
m.datetime, homeview.datetime, p1mod.time = real
|
|
_price.load_prices, _st.load_prices = real_load
|
|
return c
|
|
c = asyncio.run(go())
|
|
full = {datetime.fromtimestamp(q * QUARTER_S, L).strftime("%H:%M"):
|
|
s / k for q, (s, k) in sorted(sums.items()) if k >= QUARTER_S}
|
|
return c, rows, full, samples
|
|
|
|
|
|
def into(now):
|
|
return now.timestamp() % QUARTER_S
|
|
|
|
|
|
# ============================================================================ #
|
|
print("1. charge_room_w(): a late reading's seconds count at that reading")
|
|
_dec = peakguard.decide(
|
|
Projection(5800.0, 5800.0, 450.0, 1.0, True, START, 5800.0),
|
|
batt_w=-4800.0, cap_w=3500, margin_w=50,
|
|
reading=types.SimpleNamespace(is_observed=True, peak_w=PEAK),
|
|
achievable_w=9600.0, soc=50.0, soc_floor=20.0)
|
|
_room = peakguard.charge_headroom_w(_dec)
|
|
_late = Projection(5800.0, 5800.0, 450.0, 430.0 / 450.0, False, START,
|
|
5800.0)
|
|
print(f" headroom {_room:.0f} W; 20 s late, 450 s left: "
|
|
f"{peakguard.charge_room_w(_dec, _late):.0f} W")
|
|
check("with every second counted it is charge_headroom_w() exactly",
|
|
peakguard.charge_room_w(
|
|
_dec, Projection(5800.0, 5800.0, 450.0, 1.0, True, START, 5800.0))
|
|
== _room and _room > 0.0)
|
|
check("20 s late at a 5800 W reading with 450 s left: 5800 x 20 / 450 W "
|
|
"less room",
|
|
abs(peakguard.charge_room_w(_dec, _late)
|
|
- (_room - 5800.0 * 20.0 / 450.0)) < 1e-6)
|
|
check("never below 0 W, and an export held counts nothing",
|
|
peakguard.charge_room_w(_dec, Projection(
|
|
5800.0, 5800.0, 880.0, 850.0 / 880.0, False, START, 5800.0)) == 0.0
|
|
and peakguard.charge_room_w(_dec, Projection(
|
|
5800.0, 5800.0, 450.0, 430.0 / 450.0, False, START, -900.0))
|
|
== _room)
|
|
check("None, blind (0 W) and no projection pass through",
|
|
peakguard.charge_room_w(None, _late) is None
|
|
and peakguard.charge_room_w(_dec, None) == _room
|
|
and peakguard.charge_room_w(
|
|
peakguard.PeakDecision(engaged=True, floor_w=0.0, ceiling_w=3500.0,
|
|
ceiling_from="cap", blind="frozen"),
|
|
_late) == 0.0)
|
|
|
|
|
|
# ============================================================================ #
|
|
def graded(label, full, aim):
|
|
print(f" {label}: quarters "
|
|
f"{ {k: round(v, 1) for k, v in full.items()} } W, aim {aim:.0f}")
|
|
return len(full) >= 2 and all(v <= aim + TOL for v in full.values())
|
|
|
|
|
|
def cut_and_back(rows):
|
|
"""The charge is off in each quarter's last 5 s, and back (A charging
|
|
over 2 kW) within the next quarter's first minute."""
|
|
ends = [r for r in rows if into(r[0]) >= QUARTER_S - 5]
|
|
starts = {}
|
|
for r in rows:
|
|
if into(r[0]) < 60:
|
|
q = int(r[0].timestamp() // QUARTER_S)
|
|
starts[q] = starts.get(q, False) or r[2] < -2000.0
|
|
return (present(first(iter(ends), "a quarter end"))
|
|
and all(r[1] > -1.0 for r in ends)
|
|
and len(starts) >= 2 and all(starts.values()))
|
|
|
|
|
|
print("\n2. a reading 30 s late at each quarter's end, the live shape")
|
|
GAP = {}
|
|
for name, kw, aim in (
|
|
("month peak 5413 W banked, P1 projection",
|
|
dict(peak=PEAK), AIM_PEAK),
|
|
("month peak banked, peak_forecast_entity (kW)",
|
|
dict(peak=PEAK, fc=True), AIM_PEAK),
|
|
("no peak banked (aim = cap - margin), P1 projection",
|
|
dict(), AIM_CAP),
|
|
("no peak banked, peak_forecast_entity", dict(fc=True), AIM_CAP)):
|
|
fc = kw.pop("fc", False)
|
|
_c, _rows, _full, _s = drive(opts(fc=fc), late=30.0, **kw)
|
|
GAP[name] = (_c, _rows, _full)
|
|
check(f"{name}: every quarter at or under the aim "
|
|
f"({aim:.0f} W, +{TOL:.0f} W landing)", graded(name, _full, aim))
|
|
check(f"{name}: the charge is off in the quarter's last seconds and "
|
|
f"back in the next quarter", cut_and_back(_rows))
|
|
_c_fc = GAP["month peak banked, peak_forecast_entity (kW)"][0]
|
|
check("the forecast entity is read, in kW, as the live template (SAFETY-118), "
|
|
"and the cheap run's charge does not depend on it",
|
|
_c_fc.peak_fc is not None and _c_fc.peak_fc > 1000.0
|
|
and [r[1] for r in GAP["month peak banked, peak_forecast_entity (kW)"]
|
|
[1]] == [r[1] for r in GAP[
|
|
"month peak 5413 W banked, P1 projection"][1]])
|
|
|
|
|
|
# ============================================================================ #
|
|
print("\n3. the same for a price trade (modes off) and a maintenance charge")
|
|
_ca, _ra, _fa = drive(opts(modes=False, arb=True), peak=PEAK, late=30.0)[:3]
|
|
check("modes off, price trading on, a reading 30 s late: every quarter at "
|
|
"or under the aim", graded("trade", _fa, AIM_PEAK)
|
|
and any(r[3] == "ARBITRAGE_CHARGE" for r in _ra))
|
|
|
|
|
|
def late_feed(gap):
|
|
"""safety84's cheap-window feed, no telegram in each quarter's last
|
|
`gap` s."""
|
|
feed = s84.feed(True)
|
|
|
|
def before(c, t):
|
|
if (rig.BASE.timestamp() + t) % QUARTER_S < QUARTER_S - gap \
|
|
or not getattr(c.hass, "_s84", False):
|
|
feed(c, t)
|
|
return before
|
|
|
|
|
|
for peak, house, aim in ((5413.0, 4000.0, 5363.0), (None, 2500.0, AIM_CAP)):
|
|
_mr, _mc = s84.run(2, load=lambda t, h=house: h, dur=2700.0,
|
|
month_peak=peak, before=late_feed(30.0))
|
|
_qs = {}
|
|
for r in _mr:
|
|
q = int((rig.BASE.timestamp() + r[0]) // QUARTER_S)
|
|
a = _qs.setdefault(q, [0.0, 0])
|
|
a[0] += max(0.0, house - sum(v for _u, v in r[2]))
|
|
a[1] += 1
|
|
_mf = {datetime.fromtimestamp(q * QUARTER_S, L).strftime("%H:%M"): s / k
|
|
for q, (s, k) in sorted(_qs.items())
|
|
if k * rig.TICK >= QUARTER_S}
|
|
check(f"N=2 maintenance charge, house {house:.0f} W, "
|
|
f"{'peak banked' if peak else 'no peak banked'}, a reading 30 s "
|
|
f"late: every quarter at or under the aim",
|
|
graded("maintenance", _mf, aim)
|
|
and any(mm.phase == "charge" for mm in _mc.maints))
|
|
|
|
|
|
# ============================================================================ #
|
|
print("\n4. a healthy meter: the quarters, and the words")
|
|
for name, kw, aim, kw_text in (("month peak banked", dict(peak=PEAK),
|
|
AIM_PEAK, "5.4 kW"),
|
|
("no peak banked", dict(), AIM_CAP, "3.5 kW")):
|
|
_c, _rows, _full, _s = drive(opts(), every=10, **kw)
|
|
check(f"{name}: every quarter at or under the aim",
|
|
graded(name, _full, aim))
|
|
over = [s for s in _s if not s[2] and s[0] > s[1]]
|
|
under = [s for s in _s if not s[2] and s[0] <= s[1]]
|
|
_o, _u = first(iter(over), "an over sample"), \
|
|
first(iter(under), "an under sample")
|
|
print(f" over the limit at this rate ({len(over)} samples): "
|
|
f"{_o[3] if present(_o) else _o!s} / "
|
|
f"{_o[4] if present(_o) else _o!s}")
|
|
print(f" under it ({len(under)}): "
|
|
f"{_u[3] if present(_u) else _u!s} / "
|
|
f"{_u[4] if present(_u) else _u!s}")
|
|
check(f"{name}: over the limit at this rate the guard row never says "
|
|
f"'under'; it says the grid charge is limited",
|
|
present(_o) and all(
|
|
"under" not in s[3] and "Peak guard: grid charge limited, "
|
|
"at this rate" in s[3] and " over " in s[3] for s in over))
|
|
check(f"{name}: ...and the month-peak line says the rate, and that the "
|
|
f"grid charge is cut back to stay under {kw_text}",
|
|
present(_o) and all(
|
|
s[4].startswith("At this rate this 15-minute period reaches")
|
|
and s[4].endswith("The grid charge is cut back to stay "
|
|
f"under {kw_text}.") for s in over))
|
|
check(f"{name}: under the limit both read as before",
|
|
present(_u) and all(
|
|
s[3].startswith("Peak guard: idle, heading for")
|
|
and " under " in s[3]
|
|
and s[4].startswith("This 15-minute period is heading for")
|
|
and s[4].endswith(f"The batteries step in above {kw_text}.")
|
|
for s in under))
|
|
|
|
|
|
# ============================================================================ #
|
|
print("\n5. the words on their own")
|
|
_en = messages.row_peak_guard_idle(5814.0, 5413.0, month_peak=True)
|
|
_cut = messages.row_peak_guard_idle(5814.0, 5413.0, month_peak=True,
|
|
cut=True)
|
|
_ok = messages.row_peak_guard_idle(5200.0, 5413.0, month_peak=True)
|
|
print(f" {_en}\n {_cut}\n {_ok}")
|
|
check("over, nothing cut: 'idle, at this rate ..., over ...'",
|
|
_en == "Peak guard: idle, at this rate 5 814 W, over 5 413 W")
|
|
check("over, a grid charge cut",
|
|
_cut == "Peak guard: grid charge limited, at this rate 5 814 W, "
|
|
"over 5 413 W")
|
|
check("under: the 0.19.1 sentence, unchanged",
|
|
_ok == "Peak guard: idle, heading for 5 200 W, under this month's "
|
|
"5 413 W peak")
|
|
check("the Dutch forms say the same",
|
|
_cut.nl == "Piekbewaking: laden beperkt, aan dit tempo 5 814 W, "
|
|
"boven 5 413 W"
|
|
and _en.nl.startswith("Piekbewaking: in rust, aan dit tempo"))
|
|
check("the month-peak line in shadow mode says 'would'",
|
|
homeview.line_period(5814.0, 5413.0, True, cut=True).endswith(
|
|
"The grid charge would be cut back to stay under 5.4 kW.")
|
|
and homeview.line_period(5814.0, 5413.0, False).endswith(
|
|
"The batteries step in above 5.4 kW."))
|
|
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED:")
|
|
for f in fails:
|
|
print(f" - {f}")
|
|
sys.exit(1)
|
|
print("all checks passed")
|