#!/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=(), modes=True): """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`. `modes=False` (OPT-15) runs with battery modes off.""" 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": modes, "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}")