Merge release/1.0 (e3e4ba6) into opt15-price-run-fixes

Conflicts (5 files, 9 hunks) resolved keeping both sides:
- p1.py: held_t (OPT-15) beside held_since (SAFETY-90).
- modetrade.py: chargers() takes targets (P4) and probe (OPT-15); decide()
  clears the chargers, a probing refuser included, while a battery drains
  (SAFETY-78); the docstring lists both rules.
- main.py: the roll goes into the SAFETY-90/91 projection chain
  (`_peak_proj`); the SAFETY-95 drain-tail grid_w keeps the run-edge hold on
  its other branch; the P4 forecast state and the hold state both stay.
- DOCS.md, DECISIONS: both lines kept, OPT-15 item refreshed.

Adaptations the merge needs:
- SAFETY-81 x F1: a rolled projection is not graded by the saturation alarm
  until the quarter's first sample, as the old edge tick was not, so a
  standing overload is still notified once per quarter.
- test_safety81, test_modes_p4: modes-on pins re-pinned (with a note), modes
  off pinned to e3e4ba6; test_price_run_fixes: identity pins on e3e4ba6,
  F1 reproduced at one inverter (SAFETY-91 already holds two), tests for the
  drain and the day plan against the probe.

Co-Authored-By: Claude <noreply@anthropic.com>

goodwe_controller/tools/opt11p4_identity.py

@@ -0,0 +1,112 @@
+#!/usr/bin/env python3
+"""OPT-11 P4: with no usable solar forecast, is every command byte-identical
+to release/1.0 5a4cdc2 (P3, owner option A)?
+
+    python3 tools/opt11p4_identity.py          # one digest per scenario
+
+tools/opt11p3_identity.py's harness (the REAL Controller on a P1 meter, a
+cheap slot, a dear slot, a house over the limit, `arbitrage_enabled`), now
+with battery modes ON and each battery's mode set in the store, so P3's run
+code runs on every cycle. Each row adds the trade and the roles. Scenarios:
+* `ls`, `ch`, `ls_ls`, `ch_ls`: no `pv_forecast_entity` (the day plan never
+  runs); N=1 and N=2.
+* `*_fc_down`: `pv_forecast_entity` set but the sensor reads `unavailable`,
+  with a ready house profile in the store (the plan falls back); 5a4cdc2
+  ignores both keys, so the digest must match the scenario without them.
+Uses only the API 5a4cdc2 has; test_modes_p4.py part 1 holds this tree to the
+digests printed on 5a4cdc2.
+"""
+
+import functools
+import os
+import sys
+import tempfile
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+import opt11_identity as rig                               # noqa: E402
+import opt11p3_identity as p3                              # noqa: E402
+import safety79_identity as s79                            # noqa: E402
+from app.store import unit_key                             # noqa: E402
+
+FC = {"pv_forecast_entity": "sensor.energy_production_today_remaining",
+      "pv_forecast_tomorrow_entity": "sensor.energy_production_tomorrow"}
+
+
+async def _unavailable(self, e):
+    return {"state": "unavailable", "attributes": {},
+            "last_updated": "2026-09-20T11:00:00+00:00"}
+
+
+rig._Hass.state = _unavailable
+
+SCENARIOS = {
+    "n1_ls": dict(n=1, modes={"inverter-a": "leftover_solar"}),
+    "n1_ch": dict(n=1, modes={"inverter-a": "cheap_hours"}),
+    "n2_sf_ls": dict(n=2, modes={"inverter-a": "solar_first",
+                                 "inverter-b": "leftover_solar"}),
+    "n2_ch_ls": dict(n=2, modes={"inverter-a": "cheap_hours",
+                                 "inverter-b": "leftover_solar"}),
+    "n2_ls_ls": dict(n=2, modes={"inverter-a": "leftover_solar",
+                                 "inverter-b": "leftover_solar"}),
+    "n2_sf_ch": dict(n=2, modes={"inverter-a": "solar_first",
+                                 "inverter-b": "cheap_hours"}),
+}
+for _k, _v in list(SCENARIOS.items()):
+    SCENARIOS[_k + "_fc_down"] = dict(_v, extra=FC)
+
+
+def run(n, *, modes, extra=None, dur=p3.DUR):
+    opts = dict(s79.P1, **p3.ARB, battery_modes_enabled=True,
+                **(extra or {}))
+    if n == 2:
+        # A capacity on each battery, so a day plan could size a level;
+        # small, so the cheap run is the one cheap slot.
+        opts["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, "capacity_kwh": 1.0}
+            for u in "ab"]
+    store = rig._Store()
+    # At N=1 the one battery is named after the site; the key is per name.
+    for name, mode in modes.items():
+        store.data[unit_key(name, "pack_mode")] = mode
+    if extra:
+        # A ready house profile (a day, 600 W, 3 kW 17:00-22:00), so only
+        # the missing forecast stops a plan here. 5a4cdc2 ignores the key.
+        store.data["house_profile"] = {"2026-09-19": {
+            str(h): (3000.0 if 17 <= h < 22 else 600.0) for h in range(24)}}
+    trade = []
+
+    def on_cycle(c, t):
+        a, tick = c.arb, getattr(c, "mode_tick", None)
+        trade.append((None if a is None else (a.action, a.power_w, a.reason),
+                      None if tick is None else
+                      tuple(sorted(tick.roles.items()))))
+    db = os.path.join(tempfile.mkdtemp(prefix="p4id-"), "tariff.db")
+    real = rig.Controller
+    rig.Controller = functools.partial(real, peak_db_path=db)
+    try:
+        rows, c = rig.run(n, extra=opts, dur=dur, load=p3.load, soc=p3.soc,
+                          cut=lambda u, t: False, before=p3._feed,
+                          on_cycle=on_cycle, store=store)
+    finally:
+        rig.Controller = real
+    return [r + (x,) for r, x in zip(rows, trade)], c
+
+
+def digests() -> dict:
+    return {k: p3.digest(run(**v)[0]) for k, v in SCENARIOS.items()}
+
+
+if __name__ == "__main__":
+    if "--summary" in sys.argv:
+        for k, v in SCENARIOS.items():
+            rows = run(**v)[0]
+            acts = sorted({str(r[-1][0][0]) if r[-1][0] else "None"
+                           for r in rows})
+            roles = sorted({x for r in rows if r[-1][1]
+                            for x in (y for _n, y in r[-1][1])})
+            print(f"{k}: {len(rows)} cycles, trade {acts}, roles {roles}")
+    for k, v in digests().items():
+        print(f"{k} {v}")

goodwe_controller/tools/opt11p4_replay.py

@@ -0,0 +1,246 @@
+#!/usr/bin/env python3
+"""OPT-11 P4: does the solar forecast pay? Replays recorded days through the
+REAL Controller with tools/opt11p3_replay.py's engine (house, batteries,
+prices, peak guard), plus a solar forecast sensor in Home Assistant.
+
+    python3 tools/opt11p4_replay.py --variant p4_ls --fc recorded
+    python3 tools/opt11p4_replay.py --variant p3_ls --prices real2026
+
+Runs unchanged on release/1.0 7a00f12 (P2) and 5a4cdc2 (P3): an older tree
+ignores `pv_forecast_entity` and never reads the sensor.
+
+HOW THE FORECAST IS MADE (the site's own solar was not recorded for these
+days; the 2026-09-24 PV recording is only in the live Home Assistant):
+* The site's solar, hour by hour, is rebuilt from the recording: the house's
+  net use (grid + battery) is house - solar, so solar = max(0, H - net),
+  where H is the median net use of the night hours (22:00-06:00) over all
+  days (`--fc perfect` feeds exactly this).
+* `recorded`: that solar times the day's error of the real Belgian day-ahead
+  solar forecast (Elia, via energy-charts): forecast / actual, per day.
+* `national`: the Belgian day-ahead forecast itself, scaled to the site
+  (one factor: the site's total over the national total, all days). It
+  carries local weather differences too, so it is the harder test.
+* `zero`: a forecast of no sun at all (is the gain the forecast's, or only
+  Leftover solar buying the evening's shortfall?).
+* `high`: `national` x 1.5. `missing`: the sensor is configured but reads
+  `unavailable`. The sensor has Open-Meteo's shape (state = kWh left today,
+  `wh_period` = Wh per hour); `--shape state` drops `wh_period`
+  (Forecast.Solar's shape).
+
+MONEY: as opt11p3_replay.py (all-in import (wholesale + 0.12) x 1.06, export
+at wholesale), plus `stored_eur`: the change in usable energy (above the
+10 % cut-out) from start to end, x 0.95, at the median all-in import price
+of the run. `net_after_eur` = net - stored_eur.
+"""
+
+import argparse
+import bisect
+import json
+import os
+import statistics
+import sys
+import time
+from datetime import datetime, timedelta, timezone
+
+HERE = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, HERE)
+sys.path.insert(0, os.path.join(HERE, "tools"))
+
+import opt11p3_replay as rp                               # noqa: E402
+
+EC = os.path.join(HERE, "scenarios",
+                  "energy-charts-be-solar-price-2026-09-01_2026-09-17.json")
+FC_ENTITY = "sensor.energy_production_today_remaining"
+FC_TOMORROW = "sensor.energy_production_tomorrow"
+
+rp.VARIANTS.update({
+    "p4_ls": (True, False, {"a": (True, "solar_first", False),
+                            "b": (True, "leftover_solar", False)}),
+    "p4_ch": (True, False, {"a": (True, "solar_first", False),
+                            "b": (True, "cheap_hours", False)}),
+})
+LOCAL = timezone(timedelta(hours=2))        # CEST for every replayed day
+
+
+def ec():
+    return json.load(open(EC))
+
+
+def real2026_points():
+    """The real day-ahead prices of the replayed days (15-minute)."""
+    d = ec()["price"]
+    return [(datetime.fromtimestamp(t, timezone.utc), 900, p / 1000.0)
+            for t, p in zip(d["unix_seconds"], d["eur_mwh"]) if p is not None]
+
+
+def site_solar(house):
+    """{hour start (UTC epoch): site solar kWh}, and H (W)."""
+    by_h = {}
+    for t, w in house:
+        by_h.setdefault(int(t // 3600) * 3600, []).append(w)
+    net = {h: statistics.fmean(v) for h, v in by_h.items() if len(v) > 20}
+    night = [w for h, w in net.items()
+             if datetime.fromtimestamp(h, LOCAL).hour in
+             (22, 23, 0, 1, 2, 3, 4, 5)]
+    base = statistics.median(night)
+    return {h: max(0.0, base - w) / 1000.0 for h, w in net.items()}, base
+
+
+def national():
+    s = ec()["solar"]
+    act = dict(zip(s["unix_seconds"], s["actual_mw"]))
+    fc = dict(zip(s["dayahead_unix_seconds"], s["dayahead_mw"]))
+    return act, fc
+
+
+def forecast_table(kind, house, days):
+    """{hour start epoch: forecast kWh}, and a report of its accuracy."""
+    pv, base = site_solar(house)
+    act, fc = national()
+    day_of = {h: datetime.fromtimestamp(h, LOCAL).date() for h in pv}
+    if kind in ("perfect", "missing"):
+        out = dict(pv)
+    elif kind == "zero":
+        out = {h: 0.0 for h in pv}
+    elif kind == "recorded":
+        ratio = {}
+        for d in set(day_of.values()):
+            hs = [h for h in pv if day_of[h] == d]
+            a = sum(act.get(h) or 0.0 for h in hs)
+            f = sum(fc.get(h) or 0.0 for h in hs)
+            ratio[d] = f / a if a > 0 else 1.0
+        out = {h: pv[h] * ratio[day_of[h]] for h in pv}
+    else:
+        k = sum(pv.values()) / max(1e-9, sum(act.get(h) or 0.0 for h in pv))
+        out = {h: (fc.get(h) or 0.0) * k * (1.5 if kind == "high" else 1.0)
+               for h in pv}
+    rep = []
+    for d in sorted(set(day_of.values())):
+        a = sum(pv[h] for h in pv if day_of[h] == d)
+        f = sum(out[h] for h in out if day_of[h] == d)
+        rep.append((d.isoformat(), round(a, 1), round(f, 1)))
+    return out, base, rep
+
+
+def install_sensor(table, kind, shape):
+    """Give opt11p3_replay's Hass a `state()` for the forecast sensors."""
+    hours = sorted(table)
+    t0 = rp_start[0]
+
+    def day_bounds(now):
+        loc = now.astimezone(LOCAL)
+        d0 = loc.replace(hour=0, minute=0, second=0, microsecond=0)
+        return d0, d0 + timedelta(days=1)
+
+    def sensor(now, tomorrow):
+        d0, d1 = day_bounds(now)
+        if tomorrow:
+            d0, d1 = d1, d1 + timedelta(days=1)
+        a, b = d0.timestamp(), d1.timestamp()
+        i, j = bisect.bisect_left(hours, a), bisect.bisect_left(hours, b)
+        rows = [(h, table[h]) for h in hours[i:j]]
+        if not rows:
+            return None
+        n = now.timestamp()
+        left = 0.0
+        for h, kwh in rows:
+            if tomorrow or h >= n:
+                left += kwh
+            elif h + 3600 > n:
+                left += kwh * (h + 3600 - n) / 3600.0
+        attrs = {"unit_of_measurement": "kWh"}
+        if shape == "periods":
+            attrs["wh_period"] = {
+                datetime.fromtimestamp(h, timezone.utc).isoformat():
+                round(kwh * 1000.0, 1) for h, kwh in rows}
+        # The forecast refreshes every hour.
+        rep = datetime.fromtimestamp(int(n // 3600) * 3600, timezone.utc)
+        return {"state": f"{left:.3f}", "attributes": attrs,
+                "last_updated": rep.isoformat(),
+                "last_reported": rep.isoformat()}
+
+    async def state(self, e):
+        now = t0 + timedelta(seconds=self.t)
+        if kind == "missing":
+            return {"state": "unavailable", "attributes": {},
+                    "last_updated": now.isoformat()}
+        if e == FC_ENTITY:
+            return sensor(now, False)
+        if e == FC_TOMORROW:
+            return sensor(now, True)
+        return None
+    rp.Hass.state = state
+
+
+rp_start = [None]
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--variant", default="p4_ls")
+    ap.add_argument("--fc", default="recorded",
+                    choices=("none", "recorded", "national", "perfect",
+                             "high", "missing", "zero"))
+    ap.add_argument("--prices", default="epex2025",
+                    choices=("epex2025", "real2026"))
+    ap.add_argument("--shape", default="periods",
+                    choices=("periods", "state"))
+    ap.add_argument("--first", type=int, default=2)
+    ap.add_argument("--days", type=int, default=16)
+    ap.add_argument("--trust", type=int, default=None)
+    ap.add_argument("--report", action="store_true",
+                    help="print the forecast's accuracy and exit")
+    a = ap.parse_args()
+    days = [f"2026-09-{d:02d}" for d in range(a.first, a.first + a.days)]
+    house = rp.load_house(days)
+    pts = rp.epex_points() if a.prices == "epex2025" else real2026_points()
+    start = datetime.fromisoformat(days[0] + "T00:00:00+02:00") \
+        .astimezone(timezone.utc)
+    rp_start[0] = start
+    extra = {}
+    if a.fc != "none":
+        table, base, rep = forecast_table(a.fc, house, days)
+        if a.report:
+            print(json.dumps({"fc": a.fc, "night_base_w": round(base),
+                              "days": rep}))
+            return
+        install_sensor(table, a.fc, a.shape)
+        extra = {"pv_forecast_entity": FC_ENTITY,
+                 "pv_forecast_tomorrow_entity": FC_TOMORROW}
+        if a.trust is not None:
+            extra["solar_trust_pct"] = a.trust
+    socs = {}
+
+    def check(c, hass, now, pr):
+        if "start" not in socs:
+            socs["start"] = {u: p.soc for u, p in hass.plants.items()}
+        socs["end"] = {u: p.soc for u, p in hass.plants.items()}
+        dp = getattr(c, "day_plan", None)
+        if dp is not None:
+            socs.setdefault("verdicts", {})[
+                now.astimezone(LOCAL).date().isoformat()] = dp.verdict
+    t0 = time.time()
+    stats, _c = rp.run(a.variant, house, pts, start, 86400.0 * len(days),
+                       extra=extra, check=check)
+    out = rp.summary(stats)
+    prices = sorted(rp.all_in(p) for _s, _n, p in pts
+                    if start <= _s < start + timedelta(days=len(days)))
+    med = prices[len(prices) // 2]
+
+    def usable(soc):
+        return sum(max(0.0, v - 10.0) / 100.0 * rp.CAP_KWH
+                   for v in soc.values())
+    d_kwh = usable(socs["end"]) - usable(socs["start"])
+    out["stored_change_kwh"] = round(d_kwh, 2)
+    out["stored_eur"] = round(d_kwh * 0.95 * med, 2)
+    out["net_after_eur"] = round(out["net_eur"] - out["stored_eur"], 2)
+    out["end_soc"] = {u: round(v, 1) for u, v in socs["end"].items()}
+    out.update(variant=a.variant, fc=a.fc, prices=a.prices, shape=a.shape,
+               first=days[0], days=len(days),
+               verdicts=socs.get("verdicts", {}),
+               wall_s=round(time.time() - t0, 1))
+    print(json.dumps(out))
+
+
+if __name__ == "__main__":
+    main()

goodwe_controller/tools/safety103_plant.py

@@ -0,0 +1,195 @@
+"""SAFETY-103's plant: test_safety95's lagging battery with the SAFETY-90
+register model, driven through the REAL Controller in simulated time.
+
+Each battery ramps at 800 W/s, then follows a first-order lag (`lag` s). Its
+power register publishes every `reg_s` s at its own phase (the SAFETY-90/96
+rig: A at 0.6, B at 0.16 of the period), and reads `off` W from the truth
+(B idles at -100 W). The meter reads every cycle (`tick` s). A drain on A
+runs from the first cycle (the others finished theirs); `before(c, t)` ends
+it. `run()` has no side effects on import, so tests and probes share it.
+"""
+
+import asyncio
+import math
+import os
+import random
+import sys
+import tempfile
+from datetime import timedelta
+
+HERE = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, HERE)
+sys.path.insert(0, os.path.join(HERE, "tools"))
+
+import opt11_identity as rig                                   # noqa: E402
+from app import main as app_main, packmode                     # noqa: E402
+from app.main import Controller                                # noqa: E402
+from app.store import unit_key                                 # noqa: E402
+
+A = "inverter-a"
+T = 300.0               # A's drain ends here
+PHASE = {"a": 0.6, "b": 0.16, "c": 0.38}
+OFF = {"b": -100.0}     # the rig's B: idle reading -100 W
+M = {"maintenance_enabled": True, "maintenance_start_hour": 11,
+     "battery_modes_enabled": True}
+
+
+def options(n, extra=None):
+    o = rig._options(n, dict(M, **(extra or {})))
+    if n > 1:
+        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 "abc"[:n]]
+    return o
+
+
+def store(n):
+    """A due for its monthly cycle (it drains from the first cycle)."""
+    s = rig._Store()
+    for u in "bc"[:n - 1]:
+        s.data[unit_key(f"inverter-{u}", "last_completed")] = \
+            rig.BASE.isoformat()
+    return s
+
+
+def off(c, t):
+    """A switched Off at T (the SAFETY-95 check 5 end)."""
+    if t >= T:
+        c.store.data[unit_key(A, packmode.USE_KEY)] = False
+
+
+def run(n, house, *, before=off, dur=T + 90.0, tick=1.0, reg_s=10.0,
+        lag=10.0, noise=50.0, seed=103, phase=None, offs=None, extra=None,
+        st=None, soc=None, dead=None, reject=None):
+    """Rows of one run. Each row is a dict: t, cmd (the add-on's commands),
+    target, grid (true), true (each battery's power), read (each register),
+    age (s since each register published), phase (maintenance phases).
+    `dead` (u, from s): u gives nothing from then on (board at 0 W).
+    `reject` {u: from s}: writes to u are refused from then on."""
+    units = tuple("abc"[:n])
+    phase = PHASE if phase is None else phase
+    offs = OFF if offs is None else offs
+    rnd = random.Random(seed)
+    soc = soc or (lambda u, t: 50.0)
+
+    class Hass(rig._Hass):
+        def __init__(self, *a):
+            super().__init__(*a)
+            self.tgt = {u: 0.0 for u in units}
+            self.act = {u: 0.0 for u in units}
+            self.reg = {u: 0.0 for u in units}
+            self.pub_t = {u: 0.0 for u in units}
+            self.slot = {u: None for u in units}
+            self.tl = 0.0
+            self.hn = 0.0
+
+        def advance(self, t):
+            for k in range(int(round((t - self.tl) / 0.1))):
+                now = self.tl + (k + 1) * 0.1
+                for u in units:
+                    c = self.cmd[u]
+                    if dead is not None and u == dead[0] and now >= dead[1]:
+                        c = 0.0
+                    d = c - self.tgt[u]
+                    self.tgt[u] += math.copysign(min(abs(d), 80.0), d)
+                    self.act[u] += (self.tgt[u] - self.act[u]) * (
+                        1 - math.exp(-0.1 / lag))
+                    s = math.floor((now - phase.get(u, 0.0) * reg_s) / reg_s)
+                    if s != self.slot[u]:
+                        self.slot[u] = s
+                        self.reg[u] = float(round(self.act[u] + offs.get(u, 0.0)))
+                        self.pub_t[u] = now
+            self.tl = t
+            self.hn = rnd.uniform(-noise, noise) if noise else 0.0
+
+        def power(self, u):
+            return self.reg[u]
+
+        def grid(self):
+            h = house(self.t) if callable(house) else house
+            return float(round(h + self.hn - sum(self.act.values())))
+
+        async def set_number(self, e, v):
+            u = e[-1]
+            if reject and u in reject and self.t >= reject[u]:
+                return False
+            self.cmd[u] = float(v)
+            return True
+
+    opts = options(n, extra)
+    db = os.path.join(tempfile.mkdtemp(prefix="s103-"), "tariff.db")
+
+    async def go():
+        hass = Hass(units, soc, lambda u, t: False, lambda t: 0.0)
+        c = Controller(opts, hass, st or store(n), 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 = app_main.datetime, [0.0]
+
+        class _DT(real_dt):
+            @classmethod
+            def now(cls, tz=None):
+                return rig.BASE + timedelta(seconds=clock[0])
+        app_main.datetime = _DT
+        rows = []
+        try:
+            t = 0.0
+            while t < dur:
+                hass.advance(t)
+                hass.t = clock[0] = t
+                for inv in c.inverters:
+                    u = inv.name[-1]
+                    if reject and u in reject and t >= reject[u] \
+                            and inv.rejected_since is None:
+                        inv.rejected_since = rig.BASE + timedelta(seconds=t)
+                if before is not None:
+                    r = before(c, t)
+                    if asyncio.iscoroutine(r):
+                        await r
+                await c.read_inputs(now_mono=t)
+                await c.cycle()
+                rows.append(dict(
+                    t=t, cmd=tuple(i.target for i in c.inverters),
+                    target=c.target, grid=hass.grid(),
+                    true=tuple(round(hass.act[u]) for u in units),
+                    read=tuple(hass.reg[u] for u in units),
+                    age=tuple(round(t - hass.pub_t[u], 1) for u in units),
+                    phase=tuple(m.phase for m in c.maints),
+                    reason=c.loop_reason))
+                t += tick
+        finally:
+            app_main.datetime = real_dt
+        return rows, c
+    return asyncio.run(go())
+
+
+def end_at(rows):
+    """The first cycle A's command left its drain (None if it never did)."""
+    return next((r["t"] for r in rows if r["t"] >= T
+                 and r["cmd"][0] < 1990.0), None)
+
+
+def window(rows, span=23.0):
+    """The rows from the end cycle to `span` s after it (the settle window,
+    PLANT_SETTLE_DOWN_S)."""
+    te = end_at(rows)
+    return [r for r in rows if te is not None and te <= r["t"] <= te + span]
+
+
+def swing(rows, i=1, span=23.0):
+    """Battery i's command in the settle window after the end cycle, and
+    its largest swing there: a move that is later undone (up then down, or
+    down then up), the smaller of its two legs. A steady climb scores 0."""
+    w = [r["cmd"][i] for r in window(rows, span)][1:]
+    best = 0.0
+    for j in range(len(w)):
+        lo_b = min(w[:j + 1])
+        hi_b = max(w[:j + 1])
+        lo_a = min(w[j:])
+        hi_a = max(w[j:])
+        best = max(best, min(w[j] - lo_b, w[j] - lo_a),
+                   min(hi_b - w[j], hi_a - w[j]))
+    return best, w

goodwe_controller/tools/safety78_identity.py

@@ -0,0 +1,186 @@
+#!/usr/bin/env python3
+"""SAFETY-78: at one inverter, is a maintenance drain byte-identical to before?
+
+    python3 tools/safety78_identity.py          # prints one digest per scenario
+
+tools/safety79_identity.py's harness (the REAL Controller on a P1 meter, so
+the peak guard runs) with the battery draining for its monthly cycle from the
+first cycle, under a house below, at and above the drain rate, a solar
+surplus and a load over the peak limit. Each row hashes safety79's fields,
+the modes' roles and the trade, the action log's newest line, and every claim
+main.py hands `resolve()`, so a claim or a log line added at N=1 that changes
+no watt still moves the digest. The `_modes` scenarios run battery modes on,
+Cheap hours only, over opt11p3_identity's prices. Uses only the API release/1.0 5a4cdc2 has, so the same file runs on
+both trees; test_safety78.py part 5 holds this tree to the digests printed
+there.
+
+SAFETY-83 adds the owner's settings (limit 3500, drain 2500, margin 50) with a
+house that goes over drain + aim and back, and N=2 runs of both. At N=1 the
+peak floor now outranks the drain where it asks for more, so the N=1 digests
+moved; `--dump FILE` on the base tree and `--diff FILE` here list every
+changed cycle and whether the peak guard wanted more than the drain on it.
+"""
+
+import functools
+import hashlib
+import os
+import pickle
+import sys
+import tempfile
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+import opt11_identity as rig                               # noqa: E402
+import opt11p3_identity as p3                              # noqa: E402
+import safety79_identity as s79                            # noqa: E402
+import app.main as m                                       # noqa: E402
+from app import packmode                                  # noqa: E402
+from app.store import unit_key                            # noqa: E402
+
+DUR = 2400.0
+DRAIN = {"maintenance_enabled": True, "maintenance_start_hour": 11}
+
+
+def load(t):
+    """Below the 2000 W drain, at it, above it, a surplus, over the limit."""
+    k = int(t // 480)
+    return (1200.0, 2000.0, 3000.0, -1500.0, 6000.0)[min(k, 4)]
+
+
+def store_cheap(n):
+    s = rig._Store()
+    for u in ("inverter-a", "inverter-b")[:n]:
+        s.data[unit_key(u, packmode.MODE_KEY)] = packmode.CHEAP_HOURS
+    return s
+
+
+def load_owner(t):
+    """Below the 2500 W drain, above it, over drain + aim (5950 W), back."""
+    return (1200.0, 3000.0, 6500.0, 3000.0, -1500.0)[min(int(t // 480), 4)]
+
+
+OWNER = {"peak_cap_w": 3500, "peak_margin_w": 50,
+         "maintenance_discharge_w": 2500}
+
+SCENARIOS = {
+    "n1_drain": dict(n=1),
+    "n1_drain_modes": dict(n=1, modes=True),
+    "n1_drain_owner": dict(n=1, load=load_owner, extra=OWNER),
+    "n2_drain": dict(n=2),
+    "n2_drain_owner": dict(n=2, load=load_owner, extra=OWNER),
+}
+
+
+def run(n, *, modes=False, dur=DUR, load=load, extra=None):
+    extra = dict(s79.P1, **DRAIN, **(extra or {}))
+    if modes:
+        extra.update(p3.ARB, battery_modes_enabled=True)
+    seen, extra_rows = [], []
+    real_resolve = m.resolve
+
+    def spy(claims, **kw):
+        seen.append(tuple((c.source, c.priority, c.kind, c.value, c.lo, c.hi,
+                           c.reason, c.actuator) for c in claims))
+        return real_resolve(claims, **kw)
+
+    def on_cycle(c, t):
+        g, tick, a = c.peak_guard, c.mode_tick, c.arb
+        extra_rows.append((
+            seen[-1] if seen else None,
+            None if g is None else (g.engaged, g.floor_w, g.reason),
+            None if tick is None else (tuple(sorted(tick.roles.items())),
+                                       tick.action, tick.why),
+            None if a is None else (a.action, a.power_w, a.reason),
+            [mm.phase for mm in c.maints],
+            len(c.events), c.events[0] if c.events else None))
+    db = os.path.join(tempfile.mkdtemp(prefix="s78-"), "tariff.db")
+    real = rig.Controller
+    rig.Controller = functools.partial(real, peak_db_path=db)
+    m.resolve = spy
+    try:
+        rows, c = rig.run(n, extra=extra, dur=dur, load=load, soc=s79.soc,
+                          cut=lambda u, t: False,
+                          before=p3._feed if modes else s79._feed,
+                          on_cycle=on_cycle,
+                          store=store_cheap(n) if modes else None)
+    finally:
+        rig.Controller = real
+        m.resolve = real_resolve
+    return [r + (x,) for r, x in zip(rows, extra_rows)], c
+
+
+def digest(rows) -> str:
+    return hashlib.sha256(repr(rows).encode()).hexdigest()
+
+
+def digests() -> dict:
+    return {k: digest(run(**v)[0]) for k, v in SCENARIOS.items()}
+
+
+FIELDS = ("t", "target", "commands", "loop reason", "integrator",
+          "saturation", "why", "allocation")
+XFIELDS = ("claims", "guard", "roles", "trade", "phase", "log length",
+           "newest log line")
+
+
+def wanted_more(row) -> bool:
+    """Did the peak guard ask for more than the drain rate on this cycle?"""
+    claims, guard = row[-1][0] or (), row[-1][1]
+    rate = [c[3] for c in claims if c[0] == "maintenance" and c[2] == "set"]
+    return (bool(rate) and guard is not None and bool(guard[0])
+            and guard[1] is not None and guard[1] > max(rate))
+
+
+def diff(base: dict, now: dict) -> dict:
+    """Per scenario: (cycles, changed, changed where the guard wanted more on
+    either tree, changed cycles whose commands differ without that)."""
+    out = {}
+    for k, rows in now.items():
+        old = base.get(k)
+        if old is None:
+            continue
+        ch = [i for i, (a, b) in enumerate(zip(old, rows)) if a != b]
+        ch += list(range(min(len(old), len(rows)), max(len(old), len(rows))))
+        more = [i for i in ch if i < min(len(old), len(rows))
+                and (wanted_more(old[i]) or wanted_more(rows[i]))]
+        cmd = [i for i in ch if i not in more
+               and (i >= min(len(old), len(rows))
+                    or old[i][1:3] != rows[i][1:3])]
+        fields = set()
+        for i in set(ch) - set(more) - set(cmd):
+            fields |= {FIELDS[j] for j in range(8) if old[i][j] != rows[i][j]}
+            fields |= {XFIELDS[j] for j in range(7)
+                       if old[i][8][j] != rows[i][8][j]}
+        out[k] = (len(rows), ch, more, cmd, sorted(fields))
+    return out
+
+
+def all_rows() -> dict:
+    return {k: run(**v)[0] for k, v in SCENARIOS.items()}
+
+
+if __name__ == "__main__":
+    if "--dump" in sys.argv:
+        with open(sys.argv[sys.argv.index("--dump") + 1], "wb") as f:
+            pickle.dump(all_rows(), f)
+        sys.exit(0)
+    if "--diff" in sys.argv:
+        with open(sys.argv[sys.argv.index("--diff") + 1], "rb") as f:
+            base = pickle.load(f)
+        for k, (n, ch, more, cmd, fields) in diff(base, all_rows()).items():
+            rest = sorted(set(ch) - set(more))
+            print(f"{k}: {n} cycles, {len(ch)} changed, {len(more)} with the "
+                  f"guard above the drain, {len(rest)} others "
+                  f"({len(cmd)} with a different target or command"
+                  + (f"; the rest differ only in {', '.join(fields)}"
+                     if fields else "") + ")"
+                  + (f", first changed t={ch[0] * rig.TICK:.0f} s" if ch else ""))
+        sys.exit(0)
+    if "--summary" in sys.argv:
+        for k, v in SCENARIOS.items():
+            rows = run(**v)[0]
+            print(f"{k}: {len(rows)} cycles, phases "
+                  f"{sorted({str(r[-1][-1]) for r in rows})}, targets "
+                  f"{sorted({r[1] for r in rows})[:6]}")
+    for k, v in digests().items():
+        print(f"{k} {v}")

goodwe_controller/tools/safety81_sim.py

@@ -0,0 +1,229 @@
+#!/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}")

goodwe_controller/tools/safety84_identity.py

@@ -0,0 +1,119 @@
+#!/usr/bin/env python3
+"""SAFETY-84: at one inverter, is a maintenance charge byte-identical to before?
+
+    python3 tools/safety84_identity.py          # prints one digest per scenario
+
+safety78_identity's harness (the REAL Controller on a P1 meter, every claim
+hashed) with the battery in its maintenance CHARGE phase from the first cycle,
+on the owner's settings (limit 3500, margin 50, charge 2000 W). The house steps
+below the charge rate, at it, under the aim, over it and into a solar surplus,
+in a cheap window, outside one, and under a month peak banked at 4000 W. Uses
+only the API release/1.0 9d5b7d1 has, so the same file runs on both trees;
+test_safety84.py holds this tree to the digests printed there (N=1) and to the
+new N=2 ones.
+"""
+
+import functools
+import hashlib
+import os
+import sys
+import tempfile
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+import opt11_identity as rig                               # noqa: E402
+import safety79_identity as s79                            # noqa: E402
+import app.main as m                                       # noqa: E402
+from app import tariff                                     # noqa: E402
+
+DUR = 2400.0
+OPT = {"peak_cap_w": 3500, "peak_margin_w": 50, "maintenance_enabled": True,
+       "maintenance_start_hour": 11, "maintenance_charge_w": 2000,
+       "price_now_entity": "sensor.pnow", "price_avg_entity": "sensor.pavg"}
+
+
+def load(t):
+    """Below the 2000 W charge, at it, under the aim, over it, a surplus."""
+    return (800.0, 2000.0, 3000.0, 5000.0, -1500.0)[min(int(t // 480), 4)]
+
+
+def feed(cheap):
+    """s79's P1 feed plus a price now under (cheap) or over the average."""
+    now = 0.05 if cheap else 0.30
+
+    def before(c, t):
+        if not getattr(c.hass, "_s84", False):
+            read = c.hass.number
+
+            async def number(e, invert=False):
+                if e == "sensor.pnow":
+                    return now
+                if e == "sensor.pavg":
+                    return 0.20
+                return await read(e, invert)
+            c.hass.number, c.hass._s84 = number, True
+        s79._feed(c, t)
+    return before
+
+
+SCENARIOS = {
+    "n1_charge": dict(n=1),
+    "n1_charge_dear": dict(n=1, cheap=False),
+    "n1_charge_banked": dict(n=1, month_peak=4000.0),
+    "n2_charge": dict(n=2),
+    "n2_charge_dear": dict(n=2, cheap=False),
+    "n2_charge_banked": dict(n=2, month_peak=4000.0),
+}
+
+
+def run(n, *, cheap=True, month_peak=None, dur=DUR, load=load, soc=None,
+        before=None, cut=None, extra=None, store=None):
+    """Rows (opt11_identity's fields plus every claim, the guard and the
+    phases) and the controller. `month_peak` banks that peak this month."""
+    seen, extra_rows = [], []
+    real_resolve = m.resolve
+
+    def spy(claims, **kw):
+        seen.append(tuple((c.source, c.priority, c.kind, c.value, c.lo, c.hi,
+                           c.reason, c.actuator) for c in claims))
+        return real_resolve(claims, **kw)
+
+    def on_cycle(c, t):
+        g = c.peak_guard
+        extra_rows.append((
+            seen[-1] if seen else None,
+            None if g is None else (g.engaged, g.floor_w, g.reason),
+            c._maint_shape_suspended, [mm.phase for mm in c.maints],
+            len(c.events), c.events[0] if c.events else None))
+    db = os.path.join(tempfile.mkdtemp(prefix="s84-"), "tariff.db")
+    if month_peak is not None:
+        tariff.PeakStore(db)._db.execute(
+            "INSERT INTO monthly_peak VALUES (?, ?, ?, NULL)",
+            (tariff.month_key(rig.BASE), month_peak,
+             "2026-09-02T10:00:00+00:00")).connection.commit()
+    real = rig.Controller
+    rig.Controller = functools.partial(real, peak_db_path=db)
+    m.resolve = spy
+    try:
+        rows, c = rig.run(n, extra=dict(s79.P1, **OPT, **(extra or {})),
+                          dur=dur, load=load, soc=soc or s79.soc,
+                          cut=cut or (lambda u, t: False),
+                          before=before or feed(cheap), on_cycle=on_cycle,
+                          store=store or s79.store_charging(n))
+    finally:
+        rig.Controller = real
+        m.resolve = real_resolve
+    return [r + (x,) for r, x in zip(rows, extra_rows)], c
+
+
+def digest(rows) -> str:
+    return hashlib.sha256(repr(rows).encode()).hexdigest()
+
+
+def digests() -> dict:
+    return {k: digest(run(**v)[0]) for k, v in SCENARIOS.items()}
+
+
+if __name__ == "__main__":
+    for k, v in digests().items():
+        print(f"{k} {v}")

goodwe_controller/tools/safety87_identity.py

@@ -0,0 +1,66 @@
+#!/usr/bin/env python3
+"""SAFETY-87: at one inverter, is the peak guard byte-identical to before?
+
+    python3 tools/safety87_identity.py          # prints one digest per scenario
+
+safety84_identity's harness (the REAL Controller on a P1 meter, every claim
+and the guard's decision hashed) with the battery under, crossing and above
+the peak reserve, the house over the aim, in a maintenance charge and HOLD,
+and under a month peak banked at 4000 W. Uses only the API release/1.0
+b8ddfd7 has, so the same file runs on both trees; test_safety87.py holds this
+tree to the digests printed there.
+"""
+
+import os
+import sys
+from datetime import timedelta
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+import opt11_identity as rig                               # noqa: E402
+import safety84_identity as s84                            # noqa: E402
+
+DUR = 1800.0
+
+
+def crossing(u, t):
+    """30 % falling 1 % a minute to 12 %: crosses the 20 % reserve."""
+    return max(12.0, 30.0 - t / 60.0)
+
+
+def store(phase=None):
+    """In `phase` from the first cycle, or idle with a cycle done yesterday."""
+    s = rig._Store()
+    if phase is None:
+        s.data["last_completed"] = (rig.BASE - timedelta(days=1)).isoformat()
+    else:
+        s.data["phase"], s.data["phase_started"] = phase, rig.BASE.isoformat()
+    return s
+
+
+SCENARIOS = {
+    "n1_under": dict(soc=lambda u, t: 15.0, load=lambda t: 5500.0),
+    "n1_crossing": dict(soc=crossing, load=lambda t: 5000.0),
+    "n1_above": dict(soc=lambda u, t: 60.0, load=lambda t: 5500.0),
+    "n1_under_banked": dict(soc=lambda u, t: 15.0, load=lambda t: 6000.0,
+                            month_peak=4000.0),
+    "n1_charge_low": dict(soc=lambda u, t: 11.0, load=lambda t: 5000.0,
+                          store="charge"),
+    "n1_hold": dict(soc=lambda u, t: 99.0, load=lambda t: 5500.0,
+                    store="hold", cheap=False,
+                    extra={"maintenance_hold_min": 600}),
+}
+
+
+def digests() -> dict:
+    out = {}
+    for k, v in SCENARIOS.items():
+        v = dict(v)
+        st = store(v.pop("store", None))
+        out[k] = s84.digest(s84.run(1, dur=DUR, store=st, **v)[0])
+    return out
+
+
+if __name__ == "__main__":
+    for k, v in digests().items():
+        print(f"{k} {v}")

goodwe_controller/tools/safety94_rig.py

@@ -0,0 +1,271 @@
+#!/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 plus its offset (B: -104 W), with `noise` W of
+seeded jitter (SAFETY-97). 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 random
+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, data=None):
+        from app.store import DEFAULTS
+        self.data = dict(DEFAULTS, **(data or {}))
+
+    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, noise=0.0, seed=0):
+        self.units, self.load, self.tau, self.tick = units, load, tau, tick
+        self.noise, self.rng = noise, random.Random(seed)
+        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] = (
+                        float(round(CUT_W + OFFSET[u] + self.rng.uniform(
+                            -self.noise, self.noise)))
+                        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, noise=0.0, seed=0, store=None):
+    """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. `store`: saved state to start from."""
+    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, noise, seed)
+
+    async def go():
+        c = Controller(options(n, extra), _Hass(plant), _Store(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]}")

goodwe_controller/tools/supervisor_shape_probe.py

@@ -749,6 +749,97 @@ show("(d) GLOBAL `pv_invert: bool` validates too - it loses on SEMANTICS (one "
      "sign for a mixed roster), not on the schema", ok_d, "" if ok_d else out_d)
 
 
+print("\n== I. MAINT-10: `maintenance_start_hour` now defaults to 19 ==")
+_mk = "maintenance_start_hour"
+ok_i, out_i = validate(defaults)
+show("the shipped `options:` block validates and carries 19",
+     ok_i and out_i.get(_mk) == 19, "" if ok_i else out_i)
+# What the Supervisor hands the add-on is deepmerge(options:, stored).
+_stored10 = dict(defaults, **{_mk: 10})
+ok_i2, out_i2 = validate(MERGER.merge(copy.deepcopy(defaults), _stored10))
+show("a stored explicit 10 (what a blob saved before MAINT-10 holds) stays 10",
+     ok_i2 and out_i2.get(_mk) == 10, "" if ok_i2 else out_i2)
+_absent = {k: v for k, v in defaults.items() if k != _mk}
+ok_i3, out_i3 = validate(MERGER.merge(copy.deepcopy(defaults), _absent))
+show("a stored blob WITHOUT the key gets the new default, 19",
+     ok_i3 and out_i3.get(_mk) == 19, "" if ok_i3 else out_i3)
+ok_i4, out_i4 = validate(dict(defaults, **{_mk: 24}))
+show("the range is unchanged: 24 is refused (`at most 23`)",
+     (not ok_i4) and "at most 23" in str(out_i4), out_i4)
+
+print("\n== J. OPT-17: several solar forecast entities per day ==")
+# The same question as §G, for `pv_forecast_entity` /
+# `pv_forecast_tomorrow_entity` (`str?` since 0.15.0). Each refusal is
+# varied from the accepted baseline and asserted on the key's name.
+# DECISIONS/OPT-17-several-forecast-entities.md.
+FC_STORED = {"pv_forecast_entity": "sensor.energy_production_today_remaining",
+             "pv_forecast_tomorrow_entity": "sensor.energy_production_tomorrow"}
+_fc_old = {k: v for k, v in defaults.items()
+           if k not in ("pv_forecast_entities", "pv_forecast_tomorrow_entities")}
+_fc_base = MERGER.merge(copy.deepcopy(defaults),
+                        dict(_fc_old, **FC_STORED))
+ok_j, out_j = validate(_fc_base)
+show("a 0.17.0 blob (single forecast keys set, no list keys) validates "
+     "unchanged, and both new keys default to []",
+     ok_j and out_j.get("pv_forecast_entity") == FC_STORED["pv_forecast_entity"]
+     and out_j.get("pv_forecast_entities") == []
+     and out_j.get("pv_forecast_tomorrow_entities") == [],
+     "" if ok_j else out_j)
+for _k in ("pv_forecast_entity", "pv_forecast_tomorrow_entity"):
+    _s1 = dict(schema, **{_k: ["str"]})
+    ok_j1, out_j1 = validate_with(_s1, _fc_base)
+    show(f"`{_k}` declared as a LIST refuses the string an install stores",
+         (not ok_j1) and f"Invalid list for option '{_k}'" in str(out_j1),
+         out_j1)
+    _m1 = MERGER.merge(copy.deepcopy(dict(defaults, **{_k: []})),
+                       copy.deepcopy(FC_STORED))
+    ok_j2, out_j2 = validate_with(_s1, _m1)
+    show("...and a `[]` default does not rescue it (list merge is override)",
+         _m1[_k] == FC_STORED[_k] and not ok_j2, out_j2)
+    _sm = dict(schema, **{_k: "match(.*)?"})
+    _ok_s, _ = validate_with(_sm, _fc_base)
+    _ok_l, _ = validate_with(_sm, dict(_fc_base, **{_k: ["sensor.a",
+                                                         "sensor.b"]}))
+    _ok_n, _ = validate_with(_sm, dict(_fc_base, **{_k: 417}))
+    _ui_m = _ui_of(_sm, _k)
+    show(f"...`match(.*)?` on `{_k}` takes a string AND a list, but also the "
+         "integer 417, and renders as a plain text box (no second row)",
+         _ok_s and _ok_l and _ok_n and _ui_m
+         and _ui_m[0].get("multiple") is not True,
+         f"string={_ok_s} list={_ok_l} integer-417={_ok_n} "
+         f"ui={json.dumps(_ui_m)}")
+_two = ["sensor.fc_east", "sensor.fc_west"]
+ok_j3, out_j3 = validate(MERGER.merge(copy.deepcopy(defaults), dict(
+    _fc_old, pv_forecast_entities=_two,
+    pv_forecast_tomorrow_entities=_two)))
+show("⭐ SHIPPED: two-entity lists for today and tomorrow validate",
+     ok_j3 and out_j3.get("pv_forecast_entities") == _two
+     and out_j3.get("pv_forecast_tomorrow_entities") == _two,
+     "" if ok_j3 else out_j3)
+for _k in ("pv_forecast_entities", "pv_forecast_tomorrow_entities"):
+    ok_j4, out_j4 = validate({k: v for k, v in defaults.items() if k != _k})
+    show(f"...and the `options:` block with ONLY `{_k}` removed is refused "
+         "by name, so its `[]` default is load-bearing",
+         (not ok_j4) and f"Missing option '{_k}' in root" in str(out_j4),
+         out_j4)
+    _ui_j = _ui_of(schema, _k)
+    show(f"...and `{_k}` renders as a repeatable string field",
+         len(_ui_j) == 1 and _ui_j[0].get("multiple") is True
+         and _ui_j[0].get("type") == "string", json.dumps(_ui_j))
+_flat_j = {k: v for k, v in schema.items()
+           if k not in ("pv_forecast_entities", "pv_forecast_tomorrow_entities")}
+try:
+    _rj = vol.Schema(vol.All(dict, AppOptions(
+        None, _flat_j, cfg["name"], "local_goodwe_controller")))(
+            MERGER.merge(copy.deepcopy(defaults), dict(
+                FC_STORED, pv_forecast_entities=_two)))
+    show("a DOWNGRADE to 0.17.0's schema drops the list keys and keeps the "
+         "single ones",
+         "pv_forecast_entities" not in _rj
+         and _rj.get("pv_forecast_entity") == FC_STORED["pv_forecast_entity"])
+except vol.Invalid as e:
+    show("a DOWNGRADE to 0.17.0's schema drops the list keys", False, str(e))
+
 # ⚠️ THE EXIT CODE IS PART OF THE INSTRUMENT. Without it the probe prints FAIL
 # and still succeeds, and a transcript is only evidence if a wrong answer is
 # distinguishable from a right one by something other than reading it.