SAFETY-94: a healthy battery that lags or reverses is not flagged as frozen or refusing
The stale-reading gate counted its 35 s from the register's last change, so after a rest the first read after a new command tripped it (SAFETY-58; live 2026-10-03, four trips 1 s after the command). It now counts only the time the command asked the register to move. At N=1 the refusal engaged after 12 frozen cycles if the reading was under 100 W at that moment, which a battery reversing through its lag passes at a 1 s meter. The refusal now needs the N>1 evidence at every roster size: under 100 W, one way, for 12 cycles spanning 22 s, restarting when the reading moves 50 W towards the ask. DECISIONS/SAFETY-94-lagging-battery.md; test_safety94.py. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -0,0 +1,266 @@
|
||||
#!/usr/bin/env python3
|
||||
"""SAFETY-94: the REAL Controller on a lagging battery whose reading is late.
|
||||
|
||||
python3 tools/safety94_rig.py # prints every scenario's counts
|
||||
|
||||
Each battery ramps at 800 W/s, then follows a first-order lag (`tau`, 10 s
|
||||
as on the SAFETY-85 rig). Its reading (register 0x0518) is published every
|
||||
10 s, at a different phase per unit, and repeats its value between publishes,
|
||||
as the live boards do. B reads 100 W below what it gives (its standby draw:
|
||||
-100 W at idle, live 2026-10-03). The meter updates every `tick` s; the add-on
|
||||
reads every 1 s and cycles on a new meter value, as `run_control()` does.
|
||||
|
||||
A frozen register stops publishing at `freeze_at`; a refusing battery (`cut`)
|
||||
gives 0 W and reads -4 W. Uses only the Controller API release/1.0 02ed120
|
||||
has, so the same file runs on the tree before SAFETY-94.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import math
|
||||
import os
|
||||
import sys
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
import app.main as m # noqa: E402
|
||||
from app.main import Controller # noqa: E402
|
||||
|
||||
BASE = datetime(2026, 10, 3, 10, 50, tzinfo=timezone.utc) # drain from 600 s
|
||||
PUBLISH_S = 10.0
|
||||
PHASE = {"a": 6.0, "b": 1.6} # live: A publishes at :x6, B at :x1.6
|
||||
OFFSET = {"a": 0.0, "b": -100.0} # B's standby draw, live 2026-10-03
|
||||
CUT_W = -4.0
|
||||
LIVE = {"gain": 0.6, "slew_w": 400, "target_grid_w": -10, "deadband_w": 15,
|
||||
"step_w": 10, "heartbeat_s": 10, "stale_input_s": 15,
|
||||
"auto_start": True, "meter_entity": "sensor.grid",
|
||||
"meter_source": "off"}
|
||||
MAINT = {"maintenance_enabled": True, "maintenance_start_hour": 11,
|
||||
"maintenance_discharge_w": 2500, "maintenance_hold_min": 30}
|
||||
|
||||
|
||||
class _Store:
|
||||
def __init__(self):
|
||||
from app.store import DEFAULTS
|
||||
self.data = dict(DEFAULTS)
|
||||
|
||||
def set(self, k, v):
|
||||
self.data[k] = v
|
||||
|
||||
def get_time(self, k):
|
||||
v = self.data.get(k)
|
||||
return datetime.fromisoformat(v) if v else None
|
||||
|
||||
def set_time(self, k, when):
|
||||
self.data[k] = when.isoformat() if when else None
|
||||
|
||||
|
||||
class _Pub:
|
||||
enabled = False
|
||||
|
||||
def publish(self, *a, **k):
|
||||
pass
|
||||
|
||||
def close(self):
|
||||
pass
|
||||
|
||||
|
||||
class Plant:
|
||||
"""The batteries, their registers and the meter, in simulated seconds."""
|
||||
|
||||
def __init__(self, units, load, tau, tick, soc, cut, freeze_at,
|
||||
publish_s=PUBLISH_S, dither=False):
|
||||
self.units, self.load, self.tau, self.tick = units, load, tau, tick
|
||||
self.publish_s, self.dither, self.n_pub = publish_s, dither, 0
|
||||
self.soc_fn, self.cut, self.freeze_at = soc, cut, freeze_at
|
||||
self.cmd = {u: 0.0 for u in units}
|
||||
self.ramp = {u: 0.0 for u in units}
|
||||
self.real = {u: 0.0 for u in units}
|
||||
self.reg = {u: round(OFFSET[u]) for u in units}
|
||||
self.grid_v = float(load(0.0))
|
||||
self.t = 0.0
|
||||
|
||||
def advance(self, t):
|
||||
"""Integrate to `t` in 0.1 s steps; publish and sample on schedule."""
|
||||
while self.t < t - 1e-9:
|
||||
self.t = round(self.t + 0.1, 6)
|
||||
for u in self.units:
|
||||
want = 0.0 if self.cut(u, self.t) else self.cmd[u]
|
||||
d = want - self.ramp[u]
|
||||
self.ramp[u] += math.copysign(min(abs(d), 80.0), d)
|
||||
self.real[u] += ((self.ramp[u] - self.real[u])
|
||||
* (1 - math.exp(-0.1 / self.tau)))
|
||||
frozen = (self.freeze_at is not None
|
||||
and self.freeze_at(u) is not None
|
||||
and self.t >= self.freeze_at(u))
|
||||
ph = (self.t - PHASE[u] % self.publish_s) / self.publish_s
|
||||
if not frozen and abs(ph - round(ph)) < 1e-6:
|
||||
# `dither`: the +-1 W a settled register shows (live).
|
||||
self.n_pub += 1
|
||||
jit = (self.n_pub * 7 % 3 - 1) if self.dither else 0
|
||||
self.reg[u] = (CUT_W if self.cut(u, self.t)
|
||||
else float(round(self.real[u] + OFFSET[u])
|
||||
+ jit))
|
||||
if abs(self.t / self.tick - round(self.t / self.tick)) < 1e-6:
|
||||
self.grid_v = float(round(self.load(self.t)
|
||||
- sum(self.real.values())))
|
||||
|
||||
|
||||
class _Hass:
|
||||
def __init__(self, plant):
|
||||
self.p = plant
|
||||
|
||||
async def limits(self, e):
|
||||
return (-5000.0, 5000.0, 1.0)
|
||||
|
||||
async def call(self, *a):
|
||||
return True
|
||||
|
||||
async def number(self, e, invert=False):
|
||||
if e == "sensor.grid":
|
||||
return self.p.grid_v
|
||||
u = e[-1]
|
||||
if e.startswith("sensor.soc"):
|
||||
return self.p.soc_fn(u, self.p.t)
|
||||
if e.startswith("sensor.batt"):
|
||||
return self.p.reg[u]
|
||||
return None
|
||||
|
||||
async def set_number(self, e, v):
|
||||
self.p.cmd[e[-1]] = float(v)
|
||||
return True
|
||||
|
||||
|
||||
def options(n, extra=None):
|
||||
o = dict(LIVE)
|
||||
if n == 1:
|
||||
o.update({"setpoint_entity": "number.sp_a",
|
||||
"soc_entity": "sensor.soc_a", "batt_entity": "sensor.batt_a",
|
||||
"max_w": 4000, "bms_soc_floor": 10})
|
||||
else:
|
||||
o["inverters"] = [
|
||||
{"name": f"inverter-{u}", "setpoint_entity": f"number.sp_{u}",
|
||||
"soc_entity": f"sensor.soc_{u}", "batt_entity": f"sensor.batt_{u}",
|
||||
"max_w": 2500, "bms_soc_floor": 10} for u in "ab"[:n]]
|
||||
o.update(extra or {})
|
||||
return o
|
||||
|
||||
|
||||
def run(n, load, dur, *, tau=10.0, tick=1.0, soc=None, cut=None,
|
||||
freeze_at=None, extra=None, floor_s=15.0, publish_s=PUBLISH_S,
|
||||
dither=False):
|
||||
"""Returns a dict: per-unit stale trips [(t, unit)], refusals
|
||||
[(t, unit or 'site')], rows [(t, grid, {u: cmd}, {u: real}, {u: reg},
|
||||
phases)] and the controller."""
|
||||
units = tuple("ab"[:n])
|
||||
soc = soc or (lambda u, t: 50.0)
|
||||
cut = cut or (lambda u, t: False)
|
||||
plant = Plant(units, load, tau, tick, soc, cut, freeze_at, publish_s,
|
||||
dither)
|
||||
|
||||
async def go():
|
||||
c = Controller(options(n, extra), _Hass(plant), _Store(), _Pub(),
|
||||
stop=asyncio.Event())
|
||||
for i in c.inverters:
|
||||
i.dev_min, i.dev_max = -5000.0, 5000.0
|
||||
c.auto = True
|
||||
real_dt = m.datetime
|
||||
clock = [0.0]
|
||||
|
||||
class _DT(real_dt):
|
||||
@classmethod
|
||||
def now(cls, tz=None):
|
||||
return BASE + timedelta(seconds=clock[0])
|
||||
|
||||
m.datetime = _DT
|
||||
out = {"stale": [], "refused": [], "rows": [], "c": c}
|
||||
was = {i.name: False for i in c.inverters}
|
||||
was_ref = {i.name: 0 for i in c.inverters}
|
||||
was_site = 0
|
||||
last_grid, last_cycle = None, None
|
||||
try:
|
||||
t = 0.0
|
||||
while t < dur:
|
||||
plant.advance(t)
|
||||
clock[0] = t
|
||||
await c.read_inputs(now_mono=t)
|
||||
for i in c.inverters:
|
||||
if i.batt_fresh.stale and not was[i.name]:
|
||||
out["stale"].append((t, i.name))
|
||||
was[i.name] = i.batt_fresh.stale
|
||||
cycled = (c.grid != last_grid or last_cycle is None
|
||||
or t - last_cycle >= floor_s)
|
||||
if cycled:
|
||||
last_grid, last_cycle = c.grid, t
|
||||
await c.cycle()
|
||||
# A refusal can start on a read (a gate trip) or in a cycle.
|
||||
for i in c.inverters:
|
||||
if i.refused and not was_ref[i.name]:
|
||||
out["refused"].append((t, i.name))
|
||||
was_ref[i.name] = i.refused
|
||||
if c.batt_refused and not was_site:
|
||||
out["refused"].append((t, "site"))
|
||||
was_site = c.batt_refused
|
||||
if cycled:
|
||||
out["rows"].append((
|
||||
t, plant.grid_v, dict(plant.cmd),
|
||||
{u: round(v) for u, v in plant.real.items()},
|
||||
dict(plant.reg), [mm.phase for mm in c.maints]))
|
||||
t += 1.0
|
||||
finally:
|
||||
m.datetime = real_dt
|
||||
return out
|
||||
return asyncio.run(go())
|
||||
|
||||
|
||||
# --- the scenarios ---------------------------------------------------------
|
||||
def rest_then(step_at, w):
|
||||
"""A house at 0 W (the batteries rest) until `step_at`, then `w`."""
|
||||
return lambda t: w if t >= step_at else 0.0
|
||||
|
||||
|
||||
def reversal(at_s, charge_w, house_w):
|
||||
"""Solar surplus (the batteries charge), then a house load from `at_s`."""
|
||||
return lambda t: house_w if t >= at_s else charge_w
|
||||
|
||||
|
||||
def drain_soc(floor_at):
|
||||
"""A drains (maintenance) and reaches its floor at `floor_at`; B at 50 %."""
|
||||
def soc(u, t):
|
||||
if u != "a":
|
||||
return 50.0
|
||||
return 50.0 if t < floor_at else 10.0
|
||||
return soc
|
||||
|
||||
|
||||
SCENARIOS = {
|
||||
# first command after a long rest (the live 17:39 shape)
|
||||
"rest_step": dict(load=rest_then(600.0, 2400.0), dur=900.0),
|
||||
# maintenance drain starts after a rest (the live 19:00 shape)
|
||||
"drain_start": dict(load=lambda t: 300.0, dur=900.0,
|
||||
extra=MAINT, soc=drain_soc(1e9)),
|
||||
# A's drain ends and its charge starts (the live 21:21 shape)
|
||||
"drain_charge": dict(load=lambda t: 300.0, dur=1200.0, extra=MAINT,
|
||||
soc=drain_soc(900.0)),
|
||||
# a charge reverses into a discharge through the lag (SAFETY-94)
|
||||
"reversal": dict(load=reversal(300.0, -2500.0, 800.0), dur=600.0),
|
||||
}
|
||||
|
||||
|
||||
def counts(n, name, **kw):
|
||||
sc = dict(SCENARIOS[name], **kw)
|
||||
r = run(n, **sc)
|
||||
return len(r["stale"]), len(r["refused"]), r
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
for n in (1, 2):
|
||||
for name in SCENARIOS:
|
||||
for tick in (1.0, 2.0, 5.0):
|
||||
for pub, dith in ((10.0, False), (1.0, True)):
|
||||
s, f, r = counts(n, name, tick=tick, publish_s=pub,
|
||||
dither=dith)
|
||||
print(f"N={n} {name:13s} tick {tick:.0f} s publish "
|
||||
f"{pub:2.0f} s{' dither' if dith else ' '}: "
|
||||
f"stale {s} {r['stale'][:2]}, refusals {f} "
|
||||
f"{r['refused'][:2]}")
|
||||
Reference in New Issue
Block a user