#!/usr/bin/env python3 """SAFETY-128: the REAL Controller with Home Assistant readings arriving late. python3 tools/safety128_rig.py # prints a few runs' summaries The plant is TEST-23's (SAFETY-127 rig, t127_plant.py): two batteries (A 5 000 W, B 4 600 W) that ramp at 5 000 W/s and then follow a 1 s lag; each battery reading is published every 10 s (A at :x6, B at :x1.6); the meter every 2 s; the house is `house` W plus seeded +-`noise` W, new each second. `p1_s` sets the meter's period, `dead_s` a command dead time (SAFETY-75's laggy battery), `units` one battery ("a") or two. Home Assistant runs `lag_s` seconds behind (a number, or a function of t) between `lag_on` and `lag_off`: every reading the add-on gets (meter, battery power, battery level) is the one published `lag_s` seconds earlier. When the lag starts, nothing new comes for `lag_s` seconds, then the backlog; when it stops, the readings jump to now. Commands reach the plant at once (t127_shim.py's lag mode). The add-on reads every 1 s and runs a cycle on a new meter value or after `cycle_floor_s`, as `run_control()` does. Uses only the Controller API that release/1.0 77fe31b has, so the same file runs on the tree before SAFETY-128. """ import asyncio import math import os import random import sys import tempfile import time from datetime import datetime, timedelta, timezone sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import main_files # noqa: E402 import app.main as m # noqa: E402 from app.main import Controller # noqa: E402 from app.p1 import make_sample, split_signed # noqa: E402 # 18:00 local (CET) on a weekday: the evening give run. BASE = datetime(2026, 11, 4, 17, 0, tzinfo=timezone.utc) UNITS = ("a", "b") MAXW = {"a": 5000.0, "b": 4600.0} CAPWH = {"a": 15360.0, "b": 20480.0} PHASE = {"a": 6.0, "b": 1.6} BPER = 10.0 P1PER = 2.0 RAMP = 5000.0 LAG = 1.0 def options(units=UNITS): o = dict(OPTIONS) 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": int(MAXW[u]), "bms_soc_floor": 10, "capacity_kwh": CAPWH[u] / 1000.0} for u in units] return o OPTIONS = { "meter_entity": "sensor.grid", "meter_source": "ha_signed", "p1_net_entity": "sensor.grid", "meter_phases": 1, "meter_max_age_s": 30, "peak_cap_w": 3500, "peak_margin_w": 50, "peak_soc_floor": 20, "gain": 0.6, "slew_w": 0, "target_grid_w": -10, "deadband_w": 15, "step_w": 10, "heartbeat_s": 10, "stale_input_s": 15, "cycle_floor_s": 15, "saturation_w": 500, "saturation_cycles": 3, "auto_start": True, "battery_modes_enabled": False, "arbitrage_enabled": False, "bms_soc_floor": 10, } 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 0.1 s steps.""" def __init__(self, house, noise, seed, soc, cut=None, units=UNITS, p1_s=P1PER, dead_s=0.0): self.house_fn, self.noise, self.seed = house, noise, seed self.units, self.p1_s, self.dead_s = units, p1_s, dead_s self.cut = cut or (lambda u, t: False) self.cmd = {u: 0.0 for u in units} self.queue = [] # (t, unit, command), dead time self.tgt = {u: 0.0 for u in units} self.act = {u: 0.0 for u in units} self.soc = {u: float(soc) for u in units} self.reg = {u: 0.0 for u in units} self.slot = {u: None for u in units} self.p1 = None self.p1_n = 0 self.t = 0.0 self.house = float(house(0.0)) self.hist = [] # (t, p1 value, p1 index, {reg}, {soc}) def grid(self): return self.house - sum(self.act.values()) def advance(self, t): while self.t < t - 1e-9: self.t = round(self.t + 0.1, 6) sec = int(math.floor(self.t + 1e-9)) rs = random.Random(self.seed * 1000003 + sec) self.house = (float(self.house_fn(self.t)) + rs.uniform(-self.noise, self.noise)) while self.queue and self.queue[0][0] <= self.t + 1e-9: _t, u, v = self.queue.pop(0) self.cmd[u] = v for u in self.units: want = max(-MAXW[u], min(MAXW[u], self.cmd[u])) if self.cut(u, self.t): want = 0.0 if self.soc[u] <= 10.0 and want > 0: want = 0.0 if self.soc[u] >= 100.0 and want < 0: want = 0.0 d = want - self.tgt[u] self.tgt[u] += math.copysign(min(abs(d), RAMP * 0.1), d) self.act[u] += ((self.tgt[u] - self.act[u]) * (1 - math.exp(-0.1 / LAG))) self.soc[u] = max(0.0, min(100.0, self.soc[u] - self.act[u] * 0.1 / 3600.0 / CAPWH[u] * 100.0)) slot = math.floor((self.t - PHASE[u]) / BPER + 1e-9) if slot != self.slot[u]: self.slot[u] = slot self.reg[u] = float(round(self.act[u])) if abs(self.t / self.p1_s - round(self.t / self.p1_s)) < 1e-6: self.p1 = round(self.grid(), 1) self.p1_n += 1 self.hist.append((self.t, self.p1, self.p1_n, dict(self.reg), {u: round(v, 2) for u, v in self.soc.items()})) del self.hist[:-3000] def view(self, t, lag_s, lag_on, lag_off): """What Home Assistant shows at `t`: (p1, p1 index, regs, socs).""" lag_s = lag_s(t) if callable(lag_s) else lag_s if lag_s and lag_on <= t < lag_off: want = max(lag_on, t - lag_s) for row in reversed(self.hist): if row[0] <= want + 1e-9: return row[1:] return self.hist[0][1:] return (self.p1, self.p1_n, dict(self.reg), {u: round(v, 2) for u, v in self.soc.items()}) class _Hass: def __init__(self, rig): self.rig = rig async def limits(self, e): return (-MAXW[e[-1]], MAXW[e[-1]], 1.0) async def call(self, *a): return True async def state(self, e): return None async def number(self, e, invert=False): _p1, _n, regs, socs = self.rig.seen u = e[-1] if e.startswith("sensor.soc"): return socs[u] if e.startswith("sensor.batt"): return regs[u] return None async def set_number(self, e, v): p = self.rig.plant p.queue.append((p.t + p.dead_s, e[-1], float(v))) self.rig.writes.append((self.rig.plant.t, e[-1], float(v))) return True class Rig: def __init__(self, house=1800.0, noise=50.0, seed=90, soc=60.0, lag_s=0.0, lag_on=0.0, lag_off=0.0, extra=None, cut=None, start=BASE, units=UNITS, p1_s=P1PER, dead_s=0.0): house_fn = house if callable(house) else (lambda t, h=house: h) self.plant = Plant(house_fn, noise, seed, soc, cut, units, p1_s, dead_s) self.lag = (lag_s, lag_on, lag_off) self.o = dict(options(units), **(extra or {})) self.start = start self.writes = [] self.rows = [] self.seen = None def run(self, dur, sample=None): """Run `dur` s. Rows: (t, true grid, site command, site actual, loop reason, controller reason). `sample(rig, c, t)` per second.""" async def go(): db = os.path.join(tempfile.mkdtemp(prefix="s128-"), "tariff.db") c = Controller(self.o, _Hass(self), _Store(), _Pub(), stop=asyncio.Event(), peak_db_path=db) for i in c.inverters: i.dev_min, i.dev_max = -MAXW[i.name[-1]], MAXW[i.name[-1]] c.auto = True self.c = c real_dt = main_files.live.datetime clock = [self.start] class _DT(real_dt): @classmethod def now(cls, tz=None): return clock[0] if tz is None else clock[0].astimezone(tz) main_files.live.datetime = _DT floor = float(self.o.get("cycle_floor_s", 15)) last_grid, last_cycle, last_n = object(), None, None try: t = 0.0 while t < dur: self.plant.advance(t) now = self.start + timedelta(seconds=t) clock[0] = now self.seen = self.plant.view(t, *self.lag) p1, n = self.seen[0], self.seen[1] if p1 is not None and n != last_n: last_n = n imp, exp = split_signed(p1) c.p1.submit(make_sample( "ha_signed", imp, exp, phases=1, ingest_ts=now, ingest_mono=time.monotonic())) await c.read_inputs(now_mono=t) if (c.grid != last_grid or last_cycle is None or t - last_cycle >= floor): last_grid, last_cycle = c.grid, t await c.cycle() self.rows.append(( t, self.plant.grid(), sum(self.plant.cmd.values()), sum(self.plant.act.values()), c.loop_reason, c.reason)) if sample is not None: sample(self, c, t) t += 1.0 finally: main_files.live.datetime = real_dt return c return asyncio.run(go()) def summary(rows, t0, t1): """Over [t0, t1): (min cmd, max cmd, min grid, max grid, s of export below -200 W, s below -1 000 W).""" sel = [r for r in rows if t0 <= r[0] < t1] return (min(r[2] for r in sel), max(r[2] for r in sel), min(r[1] for r in sel), max(r[1] for r in sel), sum(1 for r in sel if r[1] < -200), sum(1 for r in sel if r[1] < -1000)) if __name__ == "__main__": for lag in (0.0, 10.0, 30.0): r = Rig(lag_s=lag, lag_on=300.0, lag_off=450.0) r.run(600.0) print(f"lag {lag:4.0f} s: before {summary(r.rows, 200, 300)}") print(f" during {summary(r.rows, 300, 450)}") print(f" after {summary(r.rows, 480, 600)}")