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>
230 lines
9.1 KiB
Python
230 lines
9.1 KiB
Python
#!/usr/bin/env python3
|
|
"""SAFETY-81: when does "Load too high for battery" fire?
|
|
|
|
python3 tools/safety81_sim.py # alarms and a control digest per run
|
|
|
|
The REAL Controller (read_inputs + cycle) at a 1 s cycle, on a plant shaped
|
|
like the SAFETY-79 test rig: one battery switched Off (so only the peak guard
|
|
makes it give, max 2000 W), a 3500 W cap, and a P1 meter that publishes
|
|
0.3 s before each cycle. The battery reaches a new command 0.5 s after it is
|
|
written. One meter sample in ten arrives late (the add-on still holds the one
|
|
before), so the held meter reading can predate the battery's last change:
|
|
the double count behind the rig's false alarms. The cycle on a quarter
|
|
boundary, before the new quarter's first sample, files no peak claim, which
|
|
is the rig's 1 s grid spike at each quarter change. Each run returns the
|
|
alarm times and a digest of every command, so the same runs show both what
|
|
changed and what did not.
|
|
"""
|
|
|
|
import asyncio
|
|
import hashlib
|
|
import logging
|
|
import os
|
|
import random
|
|
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
|
|
import app.main as m # noqa: E402
|
|
from app import packmode # 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
|
|
|
|
TICK = 1.0
|
|
SAMPLE_LEAD_S = 0.3 # the meter publishes this long before each cycle
|
|
RESPONSE_S = 0.5 # a new command reaches the battery this much later
|
|
LATE_P = 0.1 # share of meter samples that arrive late
|
|
ALARM = "goodwe_peak_saturated"
|
|
|
|
|
|
def constant(w):
|
|
return lambda t: w
|
|
|
|
|
|
def noisy(mean, amp, seed=81):
|
|
"""`mean` +/- `amp`, a new uniform value every second."""
|
|
rnd = random.Random(seed)
|
|
table = {}
|
|
|
|
def f(t):
|
|
k = int(t)
|
|
if k not in table:
|
|
table[k] = mean + rnd.uniform(-amp, amp)
|
|
return table[k]
|
|
return f
|
|
|
|
|
|
def sine(mean, amp, period=120.0, noise=50.0):
|
|
import math
|
|
n = noisy(0.0, noise, seed=79)
|
|
return lambda t: mean + amp * math.sin(2 * math.pi * t / period) + n(t)
|
|
|
|
|
|
def step(before, after, at):
|
|
return lambda t: before if t < at else after
|
|
|
|
|
|
def ramp(before, after, at, secs):
|
|
return lambda t: before + (after - before) * min(max(t - at, 0.0) / secs, 1.0)
|
|
|
|
|
|
class _Plant:
|
|
"""One battery that reaches its command RESPONSE_S after the write."""
|
|
|
|
def __init__(self, load, rating, soc, batt_gap=()):
|
|
self.load, self.rating, self.soc = load, rating, soc
|
|
self.batt_gap = batt_gap # (from, to): the battery reading is lost
|
|
self.t = 0.0
|
|
self.hist = [(-1e9, 0.0)] # (time the command takes effect, W)
|
|
self.notes = []
|
|
self.meter = 0.0 # the last meter reading delivered
|
|
|
|
def batt(self, at):
|
|
w = 0.0
|
|
for when, v in self.hist:
|
|
if when <= at:
|
|
w = v
|
|
return max(-self.rating, min(self.rating, float(round(w))))
|
|
|
|
def grid(self, at):
|
|
return self.load(at) - self.batt(at)
|
|
|
|
async def limits(self, e):
|
|
return (-5000.0, 5000.0, 1.0)
|
|
|
|
async def call(self, domain, service, data):
|
|
if service == "create":
|
|
self.notes.append((self.t, data.get("notification_id")))
|
|
return True
|
|
|
|
async def number(self, e, invert=False):
|
|
if e in ("sensor.grid", "sensor.net"):
|
|
return self.meter
|
|
if e.startswith("sensor.soc"):
|
|
return self.soc
|
|
if e.startswith("sensor.batt"):
|
|
if self.batt_gap and self.batt_gap[0] <= self.t < self.batt_gap[1]:
|
|
return None
|
|
return self.batt(self.t)
|
|
return None
|
|
|
|
async def set_number(self, e, v):
|
|
self.hist.append((self.t + RESPONSE_S, float(v)))
|
|
return True
|
|
|
|
|
|
def run(load, *, margin=50, dur=2700.0, rating=2000.0, cap=3500.0,
|
|
late_at=(), soc=60.0, batt_gap=()):
|
|
"""Alarm times (s from the start of the first quarter), a digest of every
|
|
cycle's target and written command, the Controller, the rows, and the
|
|
times of the ERROR log line. A sample is also late at each `t % 900` in
|
|
`late_at`."""
|
|
opts = rig._options(1, {
|
|
"meter_source": "ha_signed", "p1_net_entity": "sensor.net",
|
|
"meter_phases": 1, "peak_cap_w": cap, "peak_margin_w": margin,
|
|
"battery_modes_enabled": True, "max_w": rating})
|
|
store = rig._Store()
|
|
store.data[unit_key("inverter-a", packmode.USE_KEY)] = False
|
|
store.data[packmode.USE_KEY] = False
|
|
db = os.path.join(tempfile.mkdtemp(prefix="s81-"), "tariff.db")
|
|
|
|
async def go():
|
|
plant = _Plant(load, rating, soc, batt_gap)
|
|
c = Controller(opts, plant, store, rig._Pub(), stop=asyncio.Event(),
|
|
peak_db_path=db)
|
|
for i in c.inverters:
|
|
i.dev_min, i.dev_max = -5000.0, 5000.0
|
|
c.auto = True
|
|
real_dt, clock = m.datetime, [0.0]
|
|
|
|
class _DT(real_dt):
|
|
@classmethod
|
|
def now(cls, tz=None):
|
|
return rig.BASE + timedelta(seconds=clock[0])
|
|
m.datetime = _DT
|
|
rows, late = [], random.Random(8100)
|
|
errors = []
|
|
|
|
class _Err(logging.Handler):
|
|
def emit(self, r):
|
|
if "battery alone cannot keep" in r.getMessage():
|
|
errors.append(plant.t)
|
|
h = _Err()
|
|
logging.getLogger("goodwe").addHandler(h)
|
|
try:
|
|
t = 0.0
|
|
while t < dur:
|
|
plant.t = clock[0] = t
|
|
ts = t - SAMPLE_LEAD_S
|
|
if late.random() >= LATE_P and t % 900 not in late_at:
|
|
plant.meter = plant.grid(ts)
|
|
imp, exp = split_signed(plant.meter)
|
|
c.p1.submit(make_sample(
|
|
"ha_signed", imp, exp, phases=1,
|
|
ingest_ts=rig.BASE + timedelta(seconds=ts),
|
|
ingest_mono=time.monotonic()))
|
|
await c.read_inputs(now_mono=t)
|
|
await c.cycle()
|
|
rows.append((t, c.target, round(plant.hist[-1][1], 6)))
|
|
t += TICK
|
|
finally:
|
|
m.datetime = real_dt
|
|
logging.getLogger("goodwe").removeHandler(h)
|
|
return plant, c, rows, errors
|
|
plant, c, rows, errors = asyncio.run(go())
|
|
alarms = [t for t, nid in plant.notes if nid == ALARM]
|
|
return (alarms, hashlib.sha256(repr(rows).encode()).hexdigest(), c, rows,
|
|
errors)
|
|
|
|
|
|
# Three quarters each (2700 s); a new quarter starts at 900 s and 1800 s.
|
|
SCENARIOS = {
|
|
# F1: constant and sine loads the battery carries, across quarter edges.
|
|
"const4500": dict(load=constant(4500.0)),
|
|
"const4500_m0": dict(load=constant(4500.0), margin=0),
|
|
"sine4500": dict(load=sine(4500.0, 400.0)),
|
|
# F1 at the start of a quarter: the sample after the battery's return
|
|
# is late, so the held reading is the 1 s spike (the battery at 0 W).
|
|
"edge4600": dict(load=constant(4600.0), late_at=(2.0,)),
|
|
"edge4600_m0": dict(load=constant(4600.0), margin=0, late_at=(2.0,)),
|
|
# 6000 W in a quarter's last 20 s: over cap + battery, but the quarter
|
|
# still lands between the aim and the cap.
|
|
"burst_end": dict(load=lambda t: 6000.0 if t % 900 >= 880 else 4500.0),
|
|
# F2: a noisy load the battery carries, at margin 50 and 0.
|
|
"noise5000_m50": dict(load=noisy(5000.0, 150.0)),
|
|
"noise5000_m0": dict(load=noisy(5000.0, 150.0), margin=0),
|
|
# Real: from 1200 s the house needs 1000 W more than cap + battery.
|
|
"step6500": dict(load=step(4500.0, 6500.0, 1200.0)),
|
|
# Real, small: from 1200 s, 60 W (+/- 20 W noise) more than cap + battery.
|
|
"miss5560": dict(load=lambda t: step(4500.0, 5560.0, 1200.0)(t)
|
|
+ noisy(0.0, 20.0)(t)),
|
|
# Real, slow: 4500 -> 6500 W over 200 s from 1200 s, through the band
|
|
# between the aim (cap - 50) and the real cap.
|
|
"ramp6500": dict(load=ramp(4500.0, 6500.0, 1200.0, 200.0)),
|
|
# The pack under its peak reserve (20 %), and a load far over cap +
|
|
# battery: that is "Battery stopped limiting the peak", not this alarm.
|
|
"step7500_reserve": dict(load=step(4500.0, 7500.0, 1200.0), soc=15.0),
|
|
# Review: a load swinging 4500 +/- 1200 W every 30 s (a hob or heater
|
|
# cycling). It tops cap + battery for about 5 s a swing, and the quarter
|
|
# still lands under the aim.
|
|
"sine4500_fast": dict(load=sine(4500.0, 1200.0, period=30.0, noise=200.0)),
|
|
# Review: two real overloads in one quarter, 1000-1100 s and 1500-1700 s.
|
|
"twice7000": dict(load=lambda t: 7000.0 if (1000 <= t < 1100
|
|
or 1500 <= t < 1700)
|
|
else 4500.0),
|
|
# Review: step6500, and the battery reading is lost 1203-1215 s.
|
|
"step6500_gap": dict(load=step(4500.0, 6500.0, 1200.0),
|
|
batt_gap=(1203.0, 1215.0)),
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
for k, v in SCENARIOS.items():
|
|
a, d, _, _, _ = run(**v)
|
|
print(f"{k:16s} alarms at {a} digest {d}")
|