Merge release/1.0 (SAFETY-81, #155) into opt11-p4-solar-forecast

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

goodwe_controller/DECISIONS/SAFETY-81-load-too-high.md

@@ -0,0 +1,63 @@
+# SAFETY-81 — no false "Load too high for battery" alarms
+
+**Status: 2026-09-29, owner go for the card. Review kept the 10 W tolerance
+as a constant and added the 10 s hold.**
+
+## Defect
+
+The SAFETY-79 test (SAFETY-79-test-report.md, F1/F2) saw 6 false alarms in 7.
+The alarm fired whenever `decide()` clamped its floor to the rating
+(`saturated`). Three things set that while the quarter landed on target:
+
+- **Double count.** `decide()` sizes from the held meter reading plus the
+  battery reading. When the meter sample predates the battery's last increase,
+  the increase is counted twice. It happens under noise, and at each quarter
+  start, where the battery is back while the held sample is still the spike.
+- **The aim, not the cap.** `saturated` is graded against cap − margin, so at
+  margin 50 noise needs 50 W less to trip it (F2).
+- **Last seconds.** The gain 900 / remaining turns a few W over the aim into
+  thousands of W wanted, with the battery far from its rating.
+
+## Decision
+
+The alarm needs `saturated` and `peakguard.load_too_high()`: with the battery
+at full power from now on, the quarter still ends more than `ALARM_MISS_W`
+(10 W) over the real limit (`ceiling_w`). That is the miss itself, without the
+gain. The battery reading it uses is the lower of this cycle's and the last
+one's. It must hold for `ALARM_HOLD_S` (10 s). At most one notification per
+quarter, because near the threshold the test flickers with the meter; the
+ERROR log line is written for every episode.
+
+**Why 10 W:** at margin 0 the guard itself lands up to +5.7 W over (SAFETY-79
+test, check 5), and those misses are real but tiny; the alarm would fire in
+the last second of most guarded quarters. The margin (default 50) is the fix
+for them. Cost: a real miss of 10 W or less is not alarmed; at a capacity
+tariff of the order of EUR 50 per kW a year, 10 W in one month is cents.
+Review measured +13 W on a 10 kW two-unit site at margin 0, not saturated, so
+no alarm: the tolerance only matters when the battery is at its rating. A
+constant, not an option: `peak_margin_w` is the owner's knob.
+
+**Why the hold (review):** the test assumes the load now lasts the rest of
+the quarter. A load over cap + battery for ~5 s every 30 s (`sine4500_fast`,
+a cycling hob or heater) alarmed at each quarter start while the quarter
+landed under the aim. The hold stops that.
+
+**Cost:** a real alarm comes 10-12 s later than on 5a4cdc2 (`step6500` at
+1211 s, not 1201 s); a slow ramp 20 s later (`ramp6500`), as it waits for the
+real cap. While the battery refuses or its reading is lost, the alarm waits
+until the quarter is lost even at full battery (27-60 s later in review runs);
+those faults have their own alarms. A load over cap + battery for more than
+10 s still alarms even if the quarter later lands under. `ALARM_HOLD_S = 0`
+removes the hold.
+
+## Not changed: the 1 s spike itself
+
+The cycle after a quarter boundary and before the new quarter's first meter
+sample still sees 0 s left, files no claim, and an Off battery drops to 0 W for
+one cycle. That is the same edge tick as OPT-15 F1, so it is left to OPT-15.
+Control is unchanged here: every command in `tools/safety81_sim.py` is
+byte-identical to 5a4cdc2 (test_safety81.py part 1).
+
+The double count also moves control, not only the alarm: on a late sample the
+guard asks for the full rating for one cycle (edge4600 at 902 s: 2000 W where
+1150 W holds the aim). Left to a follow-up card.

goodwe_controller/DOCS.md

@@ -1752,7 +1752,8 @@ control action and goes to the action log and the status page. A *saturated*
 guard — asking for everything the inverter has while the quarter is still
 projected over — raises one notification (`goodwe_peak_saturated`), because that
 is the one state where the money will be spent and shedding load is the only
-thing left that changes the bill.
+thing left that changes the bill. SAFETY-81 keeps a 1 s meter spike, noise and
+a quarter's last seconds from raising it: see `peakguard.load_too_high()`.
 
 **And it sizes a price-driven charge, not only a forced discharge (SAFETY-45).**
 The same arithmetic read the other way answers *"how hard may this site charge
@@ -3224,9 +3225,12 @@ reason code `OUTPUT_CLAMP`.
 **Load too high for battery.** The capacity-peak guard (SAFETY-07) asks for
 everything the inverter can discharge and the running 15-minute average is
 still projected over the peak limit, so the month's billed peak
-(capaciteitstarief) goes up. In shadow mode the guard's projection is made on a
+(capaciteitstarief) goes up. It fires only when, for 10 s, the period would
-plant that did not respond (SAFETY-47), and the text says so. See *The
+still end more than 10 W over the real limit (not the margin's aim) with the
-capacity-peak guard*.
+battery at full power. It notifies at most once per period; later overloads
+in the same period go to the log (SAFETY-81). In shadow mode the
+guard's projection is made on a plant that did not respond (SAFETY-47), and
+the text says so. See *The capacity-peak guard*.
 
 **Battery stopped limiting the peak.** The peak guard may not discharge below
 `peak_soc_floor` (label “Peak reserve”, SAFETY-24). It also stops when the

goodwe_controller/app/main.py

@@ -929,6 +929,11 @@ class Controller:
             "peak_margin_w", peakguard.PEAK_OPTION_DEFAULTS["peak_margin_w"]))
         self._peak_engaged = False
         self._peak_saturated = False
+        # SAFETY-81: what the alarm is graded on, beside the guard's decision.
+        self._peak_proj = None
+        self._peak_batt_prev = None
+        self._peak_alarm_block = None
+        self._peak_high_since = None
         self._peak_floor_limited = False
         self._maint_shape_suspended = False
         # OPT-11. Battery modes, off by default so an install that has not
@@ -3821,8 +3826,10 @@ class Controller:
         # is not measured on that path, and the honest report is NOT COVERED -
         # the same answer, for the same reason, that the status page already
         # gives for SAFETY-18 on a transport that cannot see `unchanged_s`.
+        self._peak_proj = (None if not self.p1_enabled
+                           else self.peak.project(self.p1.averager, now))
         self.peak_guard = None if not self.p1_enabled else peakguard.decide(
-            self.peak.project(self.p1.averager, now),
+            self._peak_proj,
             # ⚠️ THE SITE SUM. The guard sizes a floor from what the battery is
             # ALREADY removing from the meter, and the meter sees every
             # inverter. One unit's power here would size the floor from a
@@ -4925,22 +4932,56 @@ class Controller:
         available to it. That is the one state where the money is going to be
         spent and a human is the only remaining actuator.
 
+        SAFETY-81: `g.saturated` alone is not enough. It is graded against the
+        aim, and a 1 s meter spike, noise or the last seconds of a quarter set
+        it while the quarter lands on target; `peakguard.load_too_high()` must
+        agree too, for `ALARM_HOLD_S`. The battery reading it gets is the lower
+        of this cycle's and the last one's - see there.
+
         Edge-triggered, and de-asserting arms it again - the same shape as
         FreezeWatchdog's `evaluate()`, and for the same reason: one notification
-        per episode, not one per cycle.
+        per episode, not one per cycle. SAFETY-81: the notification and action
+        log line at most once per quarter; the ERROR line once per episode.
         """
         g = self.peak_guard
-        saturated = bool(g is not None and g.saturated)
+        batt, prev = self.site_batt_w, self._peak_batt_prev
+        if batt is None:
+            # Unknown, not "fine": the hold below neither grows nor restarts.
+            return
+        self._peak_batt_prev = batt
+        if prev is not None:
+            batt = min(float(batt), float(prev))
+        p = self._peak_proj
+        high = bool(
+            g is not None and g.saturated and p is not None
+            and p.block_start is not None
+            and peakguard.load_too_high(
+                p, batt_w=batt, ceiling_w=g.ceiling_w,
+                achievable_w=self.site_rating_w))
+        # SAFETY-81: and for ALARM_HOLD_S, on the projection's clock.
+        saturated = False
+        if high:
+            t = p.block_start + timedelta(seconds=float(p.elapsed_s))
+            if self._peak_high_since is None:
+                self._peak_high_since = t
+            saturated = ((t - self._peak_high_since).total_seconds()
+                         >= peakguard.ALARM_HOLD_S)
+        else:
+            self._peak_high_since = None
         was, self._peak_saturated = self._peak_saturated, saturated
         if not saturated or was:
             return
-        self.log_event("PEAK_GUARD_SATURATED - " + g.reason,
-                       hint=messages.hint("peak_saturated"))
         _LOG.error(
             "Switch off large loads, the battery alone cannot keep this "
             "15-minute period under the peak limit: %s.",
             messages.reason_code(g.reason))
         _LOG.debug("Evidence: %s.", g.reason)
+        # SAFETY-81: notified once per quarter; a later episode is logged only.
+        if p.block_start == self._peak_alarm_block:
+            return
+        self._peak_alarm_block = p.block_start
+        self.log_event("PEAK_GUARD_SATURATED - " + g.reason,
+                       hint=messages.hint("peak_saturated"))
         # A fixed notification_id, and a DIFFERENT one from every other alarm
         # here - check_meter_frozen() gives the reasoning. A saturated peak guard
         # is not a frozen meter and is not an out-of-range setpoint.

goodwe_controller/app/peakguard.py

@@ -866,6 +866,40 @@ def decide(projection, *, batt_w, cap_w, margin_w, reading, achievable_w, soc,
                if saturated else "")))
 
 
+# SAFETY-81. A quarter must end this many W over the real limit before
+# "Load too high for battery" fires. Above the guard's own landing error at
+# margin 0 (up to +5.7 W, SAFETY-79 test check 5), which margin 0 accepts.
+ALARM_MISS_W = 10.0
+# ...and for this long without a break, so a load that peaks for a few seconds
+# (a hob or heater cycling) is not read as the rest of the quarter.
+ALARM_HOLD_S = 10.0
+
+
+def load_too_high(projection, *, batt_w, ceiling_w, achievable_w) -> bool:
+    """SAFETY-81. May "Load too high for battery" fire? Only if, with the
+    battery at full power from now on, the quarter still ends more than
+    `ALARM_MISS_W` over the real limit (`ceiling_w`, not the aim).
+
+    This is the miss itself, in W of the quarter average, so it does not have
+    `decide()`'s gain (900 / remaining): near a quarter's end that gain turns
+    a watt over the aim into thousands of W wanted.
+
+    The load is the held meter reading plus `batt_w`. The caller passes the
+    lower of this cycle's and the last cycle's battery reading, so a meter
+    sample taken before the battery's last increase is not counted twice
+    (DECISIONS/SAFETY-81-load-too-high.md). A real overload shows on the meter
+    before the battery moves, so that filter costs it no cycle.
+    """
+    remaining = QUARTER_S - float(projection.elapsed_s)
+    if batt_w is None or remaining <= 0.0:
+        return False
+    rest = float(projection.held_w) + float(batt_w) - float(achievable_w)
+    done_ws = (QUARTER_S * float(projection.projected_w)
+               - max(float(projection.held_w), 0.0) * remaining)
+    ends_w = (done_ws + max(rest, 0.0) * remaining) / QUARTER_S
+    return ends_w - float(ceiling_w) > ALARM_MISS_W
+
+
 def charge_headroom_w(decision) -> float | None:
     """SAFETY-45. The hardest this SITE may charge without putting the billed
     quarter over the ceiling, in W, positive - or None when nothing measured it.

goodwe_controller/test_peakguard.py

@@ -1259,6 +1259,13 @@ check(f"the saturating fixture ({NET2:.0f} W) is a load the real ingest "
       f"accepts, not one it rejects as implausible", NET2 < PLAUSIBLE_MAX_W)
 print(f"    saturated case: {c2.peak_guard.explain()}")
 check("a load larger than the inverter saturates the guard", c2.peak_guard.saturated)
+# SAFETY-81 review: the alarm must hold ALARM_HOLD_S first. The fixture runs on
+# the wall clock, so the start of the hold is moved back instead of waited out.
+check("...but not on its first cycle: it must hold peakguard.ALARM_HOLD_S "
+      "(SAFETY-81)", hass2.notifications.count("goodwe_peak_saturated") == 0
+      and c2._peak_high_since is not None)
+c2._peak_high_since -= timedelta(seconds=peakguard.ALARM_HOLD_S)
+asyncio.get_event_loop_policy().new_event_loop().run_until_complete(c2.cycle())
 check("...and THAT raises exactly one notification, with its own id - the one "
       "state where the money will be spent and a human is the only remaining "
       "actuator", hass2.notifications.count("goodwe_peak_saturated") == 1)

goodwe_controller/test_safety81.py

@@ -0,0 +1,171 @@
+"""Runnable check for SAFETY-81, no false "Load too high for battery" alarms.
+`python3 test_safety81.py`
+
+Drives the REAL Controller through tools/safety81_sim.py (a 1 s cycle on a
+plant shaped like the SAFETY-79 rig, one sample in ten late).
+DECISIONS/SAFETY-81-load-too-high.md has the record.
+
+1. Every command is byte-identical to release/1.0 5a4cdc2.
+2. No alarm on the quarter-edge spike, the last seconds, noise (margin 50
+   and 0) or a burst the quarter absorbs.
+3. Real overloads still alarm, ALARM_HOLD_S after 5a4cdc2.
+4. peakguard.load_too_high() on its own.
+5. The hold across a cycle with no battery reading (review).
+
+Mutations (MUTATION.md, purged copies) are listed in the pull request.
+"""
+
+import asyncio
+import os
+import sys
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, os.path.join(HERE, "tools"))
+
+import safety81_sim as sim                                    # noqa: E402
+from app.main import Controller                               # noqa: E402
+from app.peakguard import load_too_high                       # noqa: E402
+
+fails = []
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+# Printed by `python3 tools/safety81_sim.py` on release/1.0 5a4cdc2 (the tool
+# copied in, nothing else changed), in the CI image: (alarm times, digest).
+BASE = {
+    "const4500": ([2699.0], "6ba6b8634f0df9526adc04f03c82db7d083973268deda0d1f07c5519bc2e5e9d"),
+    "const4500_m0": ([2699.0], "2bede28f3ea6ba9294be134c9b9733e43d22a7c24deac3bf8210bbbbe09b2039"),
+    "sine4500": ([2699.0], "bf70c2c23a0eede40c34478d1354b7daa1858e0a0f920693e2dd15fb5046120f"),
+    "edge4600": ([2.0, 902.0, 1802.0, 2699.0], "3cc0c6fec0c70f138530955babc2b416fe7307373037c1b2e927b6ccdaad3a4d"),
+    "edge4600_m0": ([2.0, 902.0, 1802.0, 2699.0], "ba9ba2c042ee42b043315b996ece38b3478dc9866aaf833256c9c29c02f0fe9f"),
+    "burst_end": ([881.0, 1781.0, 2681.0], "5f70e1eb4244820f1d44b494cb80e51f1a4486dee03357c9a8e8f7d371c5d9af"),
+    "noise5000_m50": ([292.0, 2551.0, 2698.0], "48b446e1a78e6d6032b31cb68a4eadb0f74e3aec4cc1e184162e9c65449bfdc5"),
+    "noise5000_m0": ([292.0, 2552.0, 2698.0], "1a0527dee6c2b03dc95a90b4682858c7bf5edf7205c2aced249112f04328c960"),
+    "step6500": ([1201.0, 1801.0], "95e5c90e4a6814526cb4f3bea76286063b449d86a420f888c014531fb6e913a3"),
+    "miss5560": ([1201.0, 1801.0], "cd991cfe1685750598236701a5a2f2f0342d66645609c536d42ab7ba07216e6c"),
+    "ramp6500": ([1296.0, 1801.0], "8db7675535f1a01abf38bd6bfb996a6fe57711d3ecca53541bb1c0870748c46b"),
+    "step7500_reserve": ([], "21eb9c07a4582f47f347084d1c75a28a5d767bc068bc3f74f32433d3239e9ccb"),
+    # Added in review, printed the same way.
+    "sine4500_fast": (["112, about one per 30 s"], "710008fd8f2d33a0e2cbb9a626bedafd6d87bc23712bbc5b74a4af9b3ec21897"),
+    "twice7000": ([1001.0, 1502.0, 2699.0], "1cddda8cf35859efb2ad503f1683f744cb812a6325a390c4fa3df19d717af043"),
+    "step6500_gap": ([1201.0, 1801.0], "95e5c90e4a6814526cb4f3bea76286063b449d86a420f888c014531fb6e913a3"),
+}
+OUT = {k: sim.run(**v) for k, v in sim.SCENARIOS.items()}
+RUN = {k: v[:2] for k, v in OUT.items()}
+ERR = {k: v[4] for k, v in OUT.items()}
+
+# ============================================================================ #
+print("1. every command byte-identical to release/1.0 5a4cdc2")
+check("the same scenarios as 5a4cdc2", set(RUN) == set(BASE))
+for k, (alarms, digest) in RUN.items():
+    check(f"{k}: digest {digest[:12]}", digest == BASE[k][1])
+
+# ============================================================================ #
+print("2. no false alarm (5a4cdc2's are listed)")
+for k in ("edge4600", "edge4600_m0", "const4500", "const4500_m0", "sine4500",
+          "noise5000_m50", "noise5000_m0", "burst_end", "sine4500_fast"):
+    check(f"{k}: none (5a4cdc2: {BASE[k][0]})", RUN[k][0] == [])
+
+# ============================================================================ #
+print("3. real overloads still alarm, ALARM_HOLD_S after 5a4cdc2")
+# ALARM_HOLD_S (10 s) plus up to 2 s of late meter samples; at 1800 s the
+# quarter-edge tick restarts the hold.
+for k, want in (("step6500", [1211.0, 1813.0]),
+                ("miss5560", [1213.0, 1813.0])):
+    print(f"    {k}: load steps up at 1200 s; alarms at {RUN[k][0]}, "
+          f"5a4cdc2 {BASE[k][0]}")
+    check(f"{k}: after the hold, and again only at the next quarter",
+          RUN[k][0] == want)
+check("step6500_gap: cycles stopped by a lost battery reading do not "
+      "restart the hold",
+      RUN["step6500_gap"][0] == [1215.0, 1813.0])
+check("twice7000: notified once in the quarter (5a4cdc2: "
+      f"{BASE['twice7000'][0]})", RUN["twice7000"][0] == [1011.0])
+check("twice7000: the ERROR line for both overloads",
+      ERR["twice7000"] == [1011.0, 1512.0])
+# By design: until 1306 s (then the hold) this quarter would end at most
+# ALARM_MISS_W over the real 3500 W cap; 5a4cdc2 alarmed at the aim (cap - 50).
+print(f"    ramp6500: alarms at {RUN['ramp6500'][0]}, "
+      f"5a4cdc2 {BASE['ramp6500'][0]}")
+check("ramp6500: alarms once the quarter will end over the real cap by "
+      "more than ALARM_MISS_W",
+      RUN["ramp6500"][0] == [1316.0, 1813.0])
+check("step7500 under the peak reserve: not this alarm (the guard is not "
+      "allowed to discharge)", RUN["step7500_reserve"][0] == [])
+
+# ============================================================================ #
+print("4. load_too_high()")
+
+
+def proj(elapsed, done_avg, held):
+    """A projection `elapsed` s in, averaging `done_avg` so far."""
+    r = 900.0 - elapsed
+    return SimpleNamespace(elapsed_s=elapsed, held_w=held,
+                           projected_w=(done_avg * elapsed + max(held, 0) * r)
+                           / 900.0)
+
+
+def ltl(p, batt, cap=3500.0, rail=2000.0):
+    return load_too_high(p, batt_w=batt, ceiling_w=cap, achievable_w=rail)
+
+
+check("load 6000 W vs 3500 + 2000: yes", ltl(proj(300, 3450, 4000), 2000))
+check("already over, the battery can carry the load but not catch up: yes",
+      ltl(proj(450, 7200, 7200), 0, rail=5000))
+check("the last second, a 7 W miss (under ALARM_MISS_W): no",
+      not ltl(proj(899, 3509, 2500), 1000))
+check("a miss of exactly ALARM_MISS_W: no",
+      not ltl(proj(450, 3520, 3500), 2000))
+check("...and 1 W more: yes", ltl(proj(450, 3522, 3500), 2000))
+check("over cap + battery, but the quarter lands under the cap: no",
+      not ltl(proj(880, 3450, 4000), 2000))
+check("...and when it does not: yes", ltl(proj(700, 3450, 4000), 2000))
+check("no battery reading: no", not ltl(proj(300, 3450, 5000), None))
+check("no seconds left: no", not ltl(proj(900, 4000, 5000), 2000))
+check("a quarter already over is not won back by export: yes",
+      ltl(proj(800, 3950, 1000), 0))
+
+# ============================================================================ #
+print("5. the hold across a cycle with no battery reading")
+# A reading the stale detector drops (STALE_READING) reaches the alarm as None
+# while the cycle still runs. Unknown is not "fine": it must not restart the
+# hold. Driven on Controller.check_peak_saturated() itself, with stubs.
+Q0 = datetime(2026, 9, 20, 11, 0, tzinfo=timezone.utc)
+notes = []
+
+
+async def _call(domain, service, data):
+    notes.append(data.get("notification_id"))
+
+
+stub = SimpleNamespace(
+    peak_guard=SimpleNamespace(saturated=True, ceiling_w=3500.0,
+                               reason="PEAK_GUARD: fixture"),
+    site_rating_w=2000.0, shadow=False, events=[],
+    _peak_batt_prev=None, _peak_alarm_block=None, _peak_high_since=None,
+    _peak_saturated=False, hass=SimpleNamespace(call=_call),
+    log_event=lambda msg, hint=None: None)
+first = None
+for sec in range(0, 30):
+    stub.site_batt_w = None if 4 <= sec < 8 else 2000.0
+    p = proj(300 + sec, 3450, 6000)
+    p.block_start = Q0
+    stub._peak_proj = p
+    asyncio.run(Controller.check_peak_saturated(stub))
+    if notes and first is None:
+        first = sec
+check(f"a 4 s gap in the battery reading: alarm at {first} s, not 18 s",
+      first == 10)
+
+print()
+if fails:
+    print(f"FAILED: {len(fails)} check(s)")
+    sys.exit(1)
+print("all checks passed")

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=()):
+    """Alarm times (s from the start of the first quarter), a digest of every
+    cycle's target and written command, the Controller, the rows, and the
+    times of the ERROR log line. A sample is also late at each `t % 900` in
+    `late_at`."""
+    opts = rig._options(1, {
+        "meter_source": "ha_signed", "p1_net_entity": "sensor.net",
+        "meter_phases": 1, "peak_cap_w": cap, "peak_margin_w": margin,
+        "battery_modes_enabled": True, "max_w": rating})
+    store = rig._Store()
+    store.data[unit_key("inverter-a", packmode.USE_KEY)] = False
+    store.data[packmode.USE_KEY] = False
+    db = os.path.join(tempfile.mkdtemp(prefix="s81-"), "tariff.db")
+
+    async def go():
+        plant = _Plant(load, rating, soc, batt_gap)
+        c = Controller(opts, plant, store, rig._Pub(), stop=asyncio.Event(),
+                       peak_db_path=db)
+        for i in c.inverters:
+            i.dev_min, i.dev_max = -5000.0, 5000.0
+        c.auto = True
+        real_dt, clock = m.datetime, [0.0]
+
+        class _DT(real_dt):
+            @classmethod
+            def now(cls, tz=None):
+                return rig.BASE + timedelta(seconds=clock[0])
+        m.datetime = _DT
+        rows, late = [], random.Random(8100)
+        errors = []
+
+        class _Err(logging.Handler):
+            def emit(self, r):
+                if "battery alone cannot keep" in r.getMessage():
+                    errors.append(plant.t)
+        h = _Err()
+        logging.getLogger("goodwe").addHandler(h)
+        try:
+            t = 0.0
+            while t < dur:
+                plant.t = clock[0] = t
+                ts = t - SAMPLE_LEAD_S
+                if late.random() >= LATE_P and t % 900 not in late_at:
+                    plant.meter = plant.grid(ts)
+                    imp, exp = split_signed(plant.meter)
+                    c.p1.submit(make_sample(
+                        "ha_signed", imp, exp, phases=1,
+                        ingest_ts=rig.BASE + timedelta(seconds=ts),
+                        ingest_mono=time.monotonic()))
+                await c.read_inputs(now_mono=t)
+                await c.cycle()
+                rows.append((t, c.target, round(plant.hist[-1][1], 6)))
+                t += TICK
+        finally:
+            m.datetime = real_dt
+            logging.getLogger("goodwe").removeHandler(h)
+        return plant, c, rows, errors
+    plant, c, rows, errors = asyncio.run(go())
+    alarms = [t for t, nid in plant.notes if nid == ALARM]
+    return (alarms, hashlib.sha256(repr(rows).encode()).hexdigest(), c, rows,
+            errors)
+
+
+# Three quarters each (2700 s); a new quarter starts at 900 s and 1800 s.
+SCENARIOS = {
+    # F1: constant and sine loads the battery carries, across quarter edges.
+    "const4500": dict(load=constant(4500.0)),
+    "const4500_m0": dict(load=constant(4500.0), margin=0),
+    "sine4500": dict(load=sine(4500.0, 400.0)),
+    # F1 at the start of a quarter: the sample after the battery's return
+    # is late, so the held reading is the 1 s spike (the battery at 0 W).
+    "edge4600": dict(load=constant(4600.0), late_at=(2.0,)),
+    "edge4600_m0": dict(load=constant(4600.0), margin=0, late_at=(2.0,)),
+    # 6000 W in a quarter's last 20 s: over cap + battery, but the quarter
+    # still lands between the aim and the cap.
+    "burst_end": dict(load=lambda t: 6000.0 if t % 900 >= 880 else 4500.0),
+    # F2: a noisy load the battery carries, at margin 50 and 0.
+    "noise5000_m50": dict(load=noisy(5000.0, 150.0)),
+    "noise5000_m0": dict(load=noisy(5000.0, 150.0), margin=0),
+    # Real: from 1200 s the house needs 1000 W more than cap + battery.
+    "step6500": dict(load=step(4500.0, 6500.0, 1200.0)),
+    # Real, small: from 1200 s, 60 W (+/- 20 W noise) more than cap + battery.
+    "miss5560": dict(load=lambda t: step(4500.0, 5560.0, 1200.0)(t)
+                     + noisy(0.0, 20.0)(t)),
+    # Real, slow: 4500 -> 6500 W over 200 s from 1200 s, through the band
+    # between the aim (cap - 50) and the real cap.
+    "ramp6500": dict(load=ramp(4500.0, 6500.0, 1200.0, 200.0)),
+    # The pack under its peak reserve (20 %), and a load far over cap +
+    # battery: that is "Battery stopped limiting the peak", not this alarm.
+    "step7500_reserve": dict(load=step(4500.0, 7500.0, 1200.0), soc=15.0),
+    # Review: a load swinging 4500 +/- 1200 W every 30 s (a hob or heater
+    # cycling). It tops cap + battery for about 5 s a swing, and the quarter
+    # still lands under the aim.
+    "sine4500_fast": dict(load=sine(4500.0, 1200.0, period=30.0, noise=200.0)),
+    # Review: two real overloads in one quarter, 1000-1100 s and 1500-1700 s.
+    "twice7000": dict(load=lambda t: 7000.0 if (1000 <= t < 1100
+                                                or 1500 <= t < 1700)
+                      else 4500.0),
+    # Review: step6500, and the battery reading is lost 1203-1215 s.
+    "step6500_gap": dict(load=step(4500.0, 6500.0, 1200.0),
+                         batt_gap=(1203.0, 1215.0)),
+}
+
+
+if __name__ == "__main__":
+    for k, v in SCENARIOS.items():
+        a, d, _, _, _ = run(**v)
+        print(f"{k:16s} alarms at {a} digest {d}")