SAFETY-103 review: correct the swing after a refusal or lost link in DECISIONS

The test table gives 1.5 kW at N=2 and 3.9 kW summed at N=3 under a
surplus, and about 1 kW after a lost link without one; not 1.9 kW.

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

goodwe_controller/CHANGELOG.md

@@ -1,5 +1,49 @@
 # Changelog
 
+## 0.16.0
+
+**A minor bump: changed behaviour, no option added or removed, and one default
+and its meaning changed.** There are still 83 options. `maintenance_start_hour`
+used to be read as UTC and its default was 10. It is now a local hour (in Home
+Assistant's time zone) and its default is 19. The other changes below apply
+with one inverter or two, unless a line says "Two inverters".
+
+**Before you update (MAINT-10).** A saved `maintenance_start_hour` is kept, and
+it now means local time. If you set it in UTC to get an evening start (for
+example 17 for 19:00 in summer), set it back to the local hour (19). A saved
+10 now starts at 10:00, not 12:00 in summer. The new default 19 only reaches
+installs that have no hour saved.
+
+Contents of this release, in merge order since 0.15.0 (`3ed3302`):
+
+| | |
+|---|---|
+| `7ef7347` | After a gap in the P1 meter, the peak guard counts the missed time as high use and keeps the 15-minute period near the limit; before, it was blind until the next period. In the test, a 5 min meter silence at one inverter ended the period at 3.3 kW, against 4.8 kW on 0.15.0 ([[SAFETY-90]]) |
+| `1bf6b7f` | The maintenance start hour is now a local hour, and its default is 19 (was 10, read as UTC). The log says which time zone it used; with two inverters, both use that hour and still cycle one at a time ([[MAINT-10]]) |
+| `b2f5573` | Two inverters: when one battery's link drops, the other takes over in small steps, and control no longer stalls. The battery that comes back rejoins within 2 s, without a false "battery refuses" ([[SAFETY-96]]) |
+| `340c7ee` | Two inverters: when a drain ends (done, switched Off, timeout, lost link), the other battery covers the house right away, without jumping to 1.6 kW, and then moves in small steps. It no longer gives into a solar surplus ([[SAFETY-95]]) |
+
+### Known issues
+
+- A false "Battery is not keeping up" can still show when the peak guard pauses a maintenance charge ([[SAFETY-92]]).
+- A battery capped at a low limit with a perfectly flat reading can still trip the stale check ([[SAFETY-74]]).
+- A battery whose reading at rest is beyond about ±170 W is still not caught when it refuses, as in 0.14.0 ([[SAFETY-97]]).
+- The peak guard can still back off near the end of a 15-minute period when a meter reading is late, by less than 30 s ([[SAFETY-99]]).
+- A charge to 100 % can time out at 98 %. The add-on then counts the cycle as failed and drains the battery again the next day ([[MAINT-13]]).
+- Two inverters: the maintenance charge can switch on and off while the house is above the peak aim ([[SAFETY-88]]).
+- Two inverters: the "circulating" figure counts a battery's reading at rest ([[SAFETY-98]]).
+- Two inverters: after a drain ends, the other battery can swing by about 800 W for about 5 s, because some readings are old ([[SAFETY-103]]).
+- Two inverters: a drain that starts during a big step in house use can send the other battery to full power, and about 1 kW goes to the grid for about 10 s ([[SAFETY-104]]).
+- Two inverters: when a battery comes back during a maintenance charge, the grid can step by several kW at once ([[SAFETY-101]]).
+- One inverter: after a link drop, the battery's own safety stop to 0 W can be taken as a refusal, and the site stays at 0 W for up to 15 min ([[SAFETY-102]]).
+
+### If you are crossing from a version older than 0.10.0
+
+`## 0.10.0` below can refuse an old config: `slew_w` above its limit, and the
+withdrawn `price_provider` ids `engie_dynamic`, `totalenergies_dynamic` and
+`eneco_eddy`. Read it before you update. Run `tools/predeploy_check.py` with the
+bare options (`jq '.data.options'`) first.
+
 ## 0.15.0
 
 **A minor bump: changed behaviour, and no option changes.** No option was

goodwe_controller/DECISIONS/SAFETY-103-old-readings-after-a-drain.md

@@ -0,0 +1,80 @@
+# SAFETY-103 - no swing on old battery readings after a drain ends
+
+**Status: implemented 2026-10-08, for review.** Tests: `test_safety103.py`.
+
+**The defect** (SAFETY-95 rig test, check 5, F1). While a battery settles
+from its ended drain, SAFETY-95 steers on the house's point: grid + every
+battery's reading - aim. The meter is 1 s old; a battery register can be
+10 s old (the SAFETY-90 register model). At E+13 A's reading said 1238 W
+(6.8 s old) while A was at 616 W, so the house looked ~620 W bigger: B went
+860 -> 1650 -> 870 W. B's own old reading does the same the other way.
+
+**The rule** (`main.py`, `_track_reads()`, `_drain_grid_w()`, N>=2, only
+while a drain's tail settles):
+
+1. Each cycle, per battery: the range its power can be in. That is its last
+   changed reading, widened by every command on its wire since (a lagging
+   battery moves from one toward the others, never beyond). A new reading
+   starts the range again.
+2. A reading that changed this cycle is taken as it reads. So is the side a
+   battery refuses (SAFETY-73): it is not heading that way.
+3. The house point is held (first cycle: the loop's start from SAFETY-95).
+   It moves only as far as the meter proves: to the nearest value of grid +
+   the batteries' ranges - aim.
+
+So the loop no longer moves on a reading that has not caught up. A real
+house step still moves it, as soon as the meter shows more than the
+batteries can explain.
+
+**Why not the tester's rule** ("a battery whose reading is older than its
+last command counts at its command"). B's command changes almost every
+cycle, so B is almost always at its command. That is the plain meter, which
+chases A's tail: B swings 1420 W. A lag model needs the plant's time
+constant, which is not known for a client's inverter. Measured on the plant
+below (N=2, 1200 W, A Off; swing = a move of the others later undone):
+
+| rule | 10 s lag | 5 s lag | 20 s lag | surplus, 5 s lag |
+|---|---|---|---|---|
+| release/1.0 (readings) | 760 | 840 | 150 | 800 |
+| tester's: at the command | 1420 | 1020 | 1600 | 800 |
+| lag model, tau 10 s | 70 | 140 | 80 | 580 |
+| lag model, tau 5 s, plant 20 s | | | 600 | |
+| hold all window | 0 | 0 | 0 | 0 |
+| **this rule** | **0** | **0** | **0** | **130** |
+
+Holding all window ignores a house step: a 2 kW kettle at E+5 s imported
+14.4 Wh over the next 30 s (release 11.5, this rule 11.3); 2 kW more sun
+exported 16.9 Wh (release 8.5, this rule 8.9).
+
+**Before -> after** (`test_safety103.py`, `tools/safety103_plant.py`: 800
+W/s ramp, 10 s lag, meter 1 s, registers 10 s at the rig's phases, B idles
+at -100 W; summed swing of the others in the 23 s after the end):
+
+| case | release/1.0 | after |
+|---|---|---|
+| rig check 5: N=2, 1200 W, A Off | B 1140 -> 850 -> 1630 -> 870, swing 760 W | B holds 1140 W |
+| N=3, 1200 W / 4000 W, A Off | 680 / 630 W | 0 / 0 W |
+| solar surplus, N=2 / N=3 | 890 / 740 W | 130 / 130 W |
+| lag 5 or 20 s, registers 5 or 10 s, N=2 and N=3 | | at most 10 W |
+| registers as fresh as the meter (1 s) | | rows byte-identical |
+| refusal end | | rows byte-identical |
+| N=1, every end | | rows byte-identical (also against the release/1.0 tree) |
+
+**Cost.** In the window, a house change smaller than what the batteries
+could still be doing is not followed until a reading catches up (at most a
+register period) or the window ends (23 s). Under a surplus the others
+still swing up to 130 W, a third of a slew step.
+
+**Not covered.**
+- Outside the drain tail the loop steers on the meter; readings only reach
+  the saturation freeze (SAFETY-96), unchanged.
+- A lost link or a refusal ends the tail at once (A is out of the split, or
+  reads ~0 W). The swing after them is the loop's own on a lagging plant,
+  the same with 1 s registers and on release/1.0: under a surplus up to
+  1.5 kW at N=2 and 3.9 kW summed at N=3; after a lost link about 1 kW
+  even without one. Not this card.
+- A battery that overshoots its command, or a reading offset beyond the
+  range, can put the true power outside the range; the meter then moves the
+  point, as before.
+- The end cycle's start (SAFETY-95 step 2) still uses the readings; the
+  batteries were steady during the drain, so they are current there.

goodwe_controller/DOCS.md

@@ -1704,7 +1704,8 @@ When a battery's link drops, the others take over one slew step a cycle,
 and control does not stall on the dropped battery's power (SAFETY-96).
 When a drain ends (done, Off, timeout, abort, lost link, refused), the
 other batteries cover the house at once but do not export its drain; then
-they move one slew step a cycle (SAFETY-95).
+they move one slew step a cycle (SAFETY-95). While that battery settles,
+an old battery reading does not move them (SAFETY-103).
 
 #### Capacity-tariff exposure of the grid-following loop
 

goodwe_controller/app/main.py

@@ -1031,6 +1031,11 @@ class Controller:
         # SAFETY-95: batteries settling after their drain ended; meanwhile
         # the loop's error counts every battery at its command.
         self._drain_tail = set()
+        # SAFETY-103: per battery, the range its power can be in since its
+        # reading last changed (that reading and every command since), and
+        # the house point held while a drain's tail settles.
+        self._read_span = {}
+        self._drain_house = None
         # SAFETY-96: last cycle's commandable batteries, this cycle's start
         # point per battery where a reseed moved it, and batteries settling
         # after a step the loop did not make ({name: since, monotonic}).
@@ -2527,13 +2532,56 @@ class Controller:
         most PLANT_SETTLE_DOWN_S), the meter still shows the drain's tail and
         the others' climb, so the loop would chase them (B down, then up
         again). Meanwhile the batteries count at the site command: the error
-        is the house's own point minus the loop's start."""
+        is the house's own point minus the loop's start.
+
+        SAFETY-103: a reading can be up to a register period (10 s) older
+        than the meter, so the house point is held, and moved only as far as
+        the meter proves: to the nearest value the batteries can be at (each
+        between its last changed reading and its commands since; a reading
+        that changed this cycle, or the side a battery refuses, is taken as
+        it reads). DECISIONS/SAFETY-103-old-readings-after-a-drain.md.
+        """
         grid = float(self.grid)
         self._drain_tail &= set(self._settle) & set(live)
         if not self._drain_tail:
+            self._drain_house = None
             return grid
-        return grid - float(self.target) + sum(
-            float(inv.batt) for inv in self.inverters if inv.name in live)
+        aim = self.tuning.target_grid_w
+        lo = hi = grid - aim
+        for inv in self.inverters:
+            if inv.name not in live:
+                continue
+            r = float(inv.batt)
+            span = self._read_span.get(inv.name)
+            if (span is None or span[2] != inv.batt_fresh.changed_mono
+                    or span[2] == self._now_mono):
+                span = (r, r)
+            # A battery that refuses one way (SAFETY-73) is not heading
+            # that way, whatever it was asked: that side is its reading.
+            if (inv.refused or 0) > 0:
+                span = (min(span[0], r), r)
+            elif (inv.refused or 0) < 0:
+                span = (r, max(span[1], r))
+            lo, hi = lo + span[0], hi + span[1]
+        held = (float(self.target) if self._drain_house is None
+                else self._drain_house)
+        self._drain_house = min(hi, max(lo, held))
+        return self._drain_house - float(self.target) + aim
+
+    def _track_reads(self) -> None:
+        """SAFETY-103, N>=2: per battery, the range its power can be in now:
+        its last changed reading and every command on its wire since (a
+        lagging battery moves from one toward the others, never beyond)."""
+        if len(self.inverters) < 2:
+            return
+        for inv in self.inverters:
+            if inv.batt is None:
+                continue
+            cmd, since = inv.standing_w(), inv.batt_fresh.changed_mono
+            lo, hi, was = self._read_span.get(inv.name, (0.0, 0.0, None))
+            if was is None or since != was:
+                lo = hi = float(inv.batt)
+            self._read_span[inv.name] = (min(lo, cmd), max(hi, cmd), since)
 
     def _reseed_drop(self, live) -> None:
         """SAFETY-96, N>=2: a battery that drops out of the split (stale
@@ -4914,6 +4962,9 @@ class Controller:
             # level: the two halves of §6.2's identity break in opposite
             # directions. See `site_batt_w` and test_alloc.py part 8.
             site_batt = self.site_batt_w
+            # SAFETY-103: what each battery can be doing (its reading may be
+            # older than the meter).
+            self._track_reads()
             # SAFETY-96: a battery the loop cannot move counts at its command,
             # so its own power does not freeze the loop (the meter has it).
             site_batt += self._held_out_w(_live96)

goodwe_controller/config.yaml

@@ -1,5 +1,5 @@
 name: GoodWe RS485 Controller
-version: "0.15.0"
+version: "0.16.0"
 slug: goodwe_controller
 description: >-
   Drives a GoodWe ES/BP battery inverter over RS485 by emulating its smart

goodwe_controller/test_safety103.py

@@ -0,0 +1,401 @@
+"""Runnable check for SAFETY-103: no swing on old battery readings after a
+drain ends. `python3 test_safety103.py`
+
+While a battery settles from its ended drain (SAFETY-95), the loop works
+from the grid plus each battery's reading. A reading can be one register
+period (10 s) older than the meter: on the rig (SAFETY-95 test, check 5) A's
+reading still said 1238 W while A was at 616 W, and B went 860 -> 1650 ->
+870 W. Now the house point is held, and moved only as far as the meter
+proves (each battery between its last changed reading and its commands
+since). Drives the REAL Controller on tools/safety103_plant.py: a lagging
+battery, the meter every 1 s, registers every 10 s at the rig's phases, B
+idling at -100 W. DECISIONS/SAFETY-103-old-readings-after-a-drain.md.
+
+1. The rig case and every drain end at N=2 and N=3, 10 s registers: no
+   swing beyond one slew step; release/1.0 swings (non-vacuous).
+2. SAFETY-95/96's gains stay: no jump at the end, no give into a surplus,
+   one slew step a cycle, refusal and stale link still learned.
+3. A house step inside the window is still followed.
+4. With readings as fresh as the meter, every row is release/1.0's; N=1 is
+   untouched.
+5. The rule's parts, one by one.
+
+Mutations (MUTATION.md, purged copies) are listed in the pull request.
+"""
+
+import hashlib
+import logging
+import math
+import os
+import sys
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, HERE)
+sys.path.insert(0, os.path.join(HERE, "tools"))
+logging.disable(logging.WARNING)
+
+import safety103_plant as p                                    # noqa: E402
+from app.main import Controller                                # noqa: E402
+from app.maintenance import DRAIN                              # noqa: E402
+
+fails = []
+SLEW = 400.0            # the rig's slew_w
+AIM = -10.0             # the rig's target_grid_w
+T = p.T
+A = p.A
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+fixed = Controller._drain_grid_w
+track = getattr(Controller, "_track_reads", None)
+
+
+def _release(self, live):
+    """release/1.0's `_drain_grid_w` (SAFETY-95 as merged): every battery at
+    its reading."""
+    grid = float(self.grid)
+    self._drain_tail &= set(self._settle) & set(live)
+    if not self._drain_tail:
+        return grid
+    return grid - float(self.target) + sum(
+        float(inv.batt) for inv in self.inverters if inv.name in live)
+
+
+def without(fn):
+    """Run `fn` with SAFETY-103 taken out (release/1.0's rule)."""
+    Controller._drain_grid_w = _release
+    try:
+        return fn()
+    finally:
+        Controller._drain_grid_w = fixed
+
+
+def soc_done(u, t):
+    return 5.0 if (u == "a" and t >= T) else 50.0
+
+
+# A's drain ends at T: switched Off (the rig case), completed on SoC, its
+# battery refusing (gives nothing from T; SAFETY-73 learns it later), or its
+# link going stale (writes refused from T, its board to 0 W 25 s later).
+ENDS = {
+    "Off": dict(),
+    "complete": dict(before=None, soc=soc_done),
+    "refusal": dict(before=None, dead=("a", T), dur=T + 150.0),
+    "stale link": dict(before=None, reject={"a": T}, dead=("a", T + 25.0),
+                       dur=T + 150.0),
+}
+
+
+def end_of(rows):
+    """The first cycle A's drain is over (its command left 2000 W, or its
+    phase left DRAIN); None if never."""
+    return next((r["t"] for r in rows if r["t"] >= T and (
+        r["cmd"][0] < 1990.0 or r["phase"][0] != DRAIN)), None)
+
+
+def measure(rows, n, house):
+    """From the end on: the largest swing of the others' summed command in
+    the 23 s window (a move later undone; the slew step is the site's); their
+    largest command on the end cycle; the site target's largest step after
+    the end cycle; the others' largest command after the end (for a
+    surplus); the grid's range in the window."""
+    te = end_of(rows)
+    if te is None:
+        return dict(end=math.nan, swing=math.nan, first=math.nan,
+                    step=math.nan, give=math.nan, gmin=math.nan,
+                    gmax=math.nan, rows=rows)
+    win = [r for r in rows if te <= r["t"] <= te + 23.0]
+    sw = 0.0
+    w = [sum(r["cmd"][1:]) for r in win][1:]
+    for j in range(len(w)):
+        sw = max(sw, min(w[j] - min(w[:j + 1]), w[j] - min(w[j:])),
+                 min(max(w[:j + 1]) - w[j], max(w[j:]) - w[j]))
+    seg = [r["target"] for r in rows if r["t"] >= te]
+    return dict(
+        end=te, swing=sw,
+        first=max(win[0]["cmd"][1:]) if win else math.nan,
+        step=max(abs(b - a) for a, b in zip(seg, seg[1:])),
+        give=max(max(r["cmd"][1:]) for r in rows if r["t"] >= te),
+        gmin=min(r["grid"] for r in win), gmax=max(r["grid"] for r in win),
+        rows=rows)
+
+
+def run(kind, n, house, **k):
+    return p.run(n, house, **dict(ENDS[kind], **k))[0]
+
+
+TABLE = []
+# A settles from its drain: the tail this rule is about. A refusing battery
+# already reads ~0 W, and a dropped one is out of the split (the meter has
+# it), so their tails last at most a cycle.
+SETTLE = ("Off", "complete")
+
+
+def case(kind, n, house, **k):
+    now = measure(run(kind, n, house, **k), n, house)
+    old = measure(without(lambda: run(kind, n, house, **k)), n, house)
+    fresh = measure(run(kind, n, house, reg_s=1.0, **k), n, house)
+    TABLE.append((kind, n, house, now, old, fresh))
+    return now, old, fresh
+
+
+# ============================================================================ #
+print("1. no swing on an old reading: every drain end, N=2 and N=3, 10 s "
+      "registers")
+for kind in ENDS:
+    for n in (2, 3):
+        for house in (1200.0, 4000.0, -1500.0):
+            now, old, fresh = case(kind, n, house)
+            own = kind in SETTLE
+            check(f"{'⭐' if kind == 'Off' else '  '} {kind}, N={n}, house "
+                  f"{house:.0f} W: old readings add at most one slew step "
+                  f"to the others' swing in the 23 s after the end"
+                  f"{', and it is at most one slew step' if own else ''} "
+                  f"({now['swing']:.0f} W; 1 s registers "
+                  f"{fresh['swing']:.0f}; release/1.0 {old['swing']:.0f})",
+                  now["swing"] - fresh["swing"] <= SLEW
+                  and (not own or now["swing"] <= SLEW))
+_by = {(k, n, h): (a, b) for k, n, h, a, b, f in TABLE}
+_now, _old = _by[("Off", 2, 1200.0)]
+_b = [r["cmd"][1] for r in _now["rows"] if 0 <= r["t"] - _now["end"] <= 23]
+_bo = [r["cmd"][1] for r in _old["rows"] if 0 <= r["t"] - _old["end"] <= 23]
+print(f"    rig check 5 (N=2, 1200 W, A Off): B {_b}\n"
+      f"    release/1.0: B {_bo}")
+check("⭐ the rig case (SAFETY-95 check 5): release/1.0 swings B by more "
+      "than a slew step on A's old reading; now B holds within 50 W",
+      _old["swing"] > SLEW and max(_b) - min(_b) <= 50.0)
+for n in (2, 3):
+    check(f"non-vacuous (N={n}): on release/1.0 old readings add more than "
+          f"a slew step in at least one of these cases",
+          any(b["swing"] - f["swing"] > SLEW
+              for k, nn, h, a, b, f in TABLE if nn == n))
+_s5 = []
+for lag in (5.0, 20.0):
+    for reg in (5.0, 10.0):
+        for n in (2, 3):
+            _s5.append(measure(run("Off", n, 1200.0, lag=lag, reg_s=reg), n,
+                               1200.0)["swing"])
+check(f"no plant constant is assumed: battery lag 5 s and 20 s, registers "
+      f"every 5 s and 10 s, N=2 and N=3: largest swing {max(_s5):.0f} W",
+      max(_s5) <= SLEW)
+_ph = [measure(run("Off", 2, 1200.0, phase={"a": a, "b": b}), 2, 1200.0)
+       ["swing"] for a, b in ((0.0, 0.5), (0.9, 0.1), (0.3, 0.3))]
+check(f"other register phases: largest swing {max(_ph):.0f} W",
+      max(_ph) <= SLEW)
+
+# ============================================================================ #
+print("\n2. SAFETY-95/96's gains stay")
+for kind in ENDS:
+    for n in (2, 3):
+        now, old = _by[(kind, n, 1200.0)]
+        check(f"{kind}, N={n}, 1200 W: no jump on the end cycle (others "
+              f"<= the house, {now['first']:.0f} W) and the site moves at "
+              f"most one slew step a cycle after ({now['step']:.0f} W)",
+              now["first"] <= 1200.0 - AIM + 50.0 and now["step"] <= SLEW)
+        now, old = _by[(kind, n, -1500.0)]
+        check(f"{kind}, N={n}, solar surplus: no battery gives into it "
+              f"(largest command after the end {now['give']:.0f} W"
+              + (f"; a lost link as release/1.0, {old['give']:.0f} W)"
+                 if kind == "stale link" else ")"),
+              now["give"] <= (max(50.0, old["give"]) if kind == "stale link"
+                              else 50.0))
+_now, _old = _by[("Off", 2, 1200.0)]
+check(f"the grid in the window stays inside release/1.0's range "
+      f"([{_now['gmin']:.0f}, {_now['gmax']:.0f}] W, release/1.0 "
+      f"[{_old['gmin']:.0f}, {_old['gmax']:.0f}])",
+      _now["gmin"] >= _old["gmin"] - 50.0 and _now["gmax"] <= _old["gmax"])
+
+
+_ref = {}
+for name, f in (("now", lambda: p.run(2, 1200.0, **ENDS["refusal"])),
+                ("old", lambda: without(lambda: p.run(
+                    2, 1200.0, **ENDS["refusal"])))):
+    rows, c = f()
+    _ref[name] = (c.inverters[0].refused, c.inverters[0].refused_at)
+check(f"a refusal is still learned, at the same instant as release/1.0 "
+      f"({_ref['now']})", _ref["now"] == _ref["old"]
+      and _ref["now"][0] == 1)
+_st = {}
+for name, f in (("now", lambda: p.run(2, 1200.0, **ENDS["stale link"])),
+                ("old", lambda: without(lambda: p.run(
+                    2, 1200.0, **ENDS["stale link"])))):
+    rows, c = f()
+    _st[name] = sorted(c.live_units())
+check(f"a stale link is still dropped, as in release/1.0 "
+      f"(live: {_st['now']})", _st["now"] == _st["old"]
+      and len(_st["now"]) == 1)
+
+# ============================================================================ #
+print("\n3. a house step inside the window is still followed")
+
+
+def step(sign):
+    return lambda t: 1200.0 + (sign * 2000.0 if t >= T + 5.0 else 0.0)
+
+
+for sign, what in ((1, "kettle +2000 W"), (-1, "solar -2000 W")):
+    for n in (2, 3):
+        rows = p.run(n, step(sign))[0]
+        rows_o = without(lambda: p.run(n, step(sign)))[0]
+        te = end_of(rows)
+
+        def wh(rs):
+            return sum(abs(r["grid"] - AIM) for r in rs
+                       if te + 5.0 <= r["t"] <= te + 35.0) / 3600.0
+        sw = measure(rows, n, 1200.0)["swing"]
+        check(f"{what} at E+5 s, N={n}: the grid's distance from its aim "
+              f"over the next 30 s is within 1.5 Wh of release/1.0's "
+              f"({wh(rows):.1f} Wh, release/1.0 {wh(rows_o):.1f}); swing "
+              f"{sw:.0f} W", wh(rows) <= wh(rows_o) + 1.5 and sw <= SLEW)
+
+# ============================================================================ #
+print("\n4. fresh readings and N=1: release/1.0's rows")
+
+
+def digest(rows):
+    return hashlib.sha256(repr(rows).encode()).hexdigest()
+
+
+_fresh = [(n, h) for n in (2, 3) for h in (1200.0, 4000.0, -1500.0)]
+check("⭐ registers as fresh as the meter (1 s), N=2 and N=3, three houses: "
+      "every row byte-identical with release/1.0 (a reading that changed "
+      "this cycle is taken as it is)",
+      [digest(p.run(n, h, reg_s=1.0)[0]) for n, h in _fresh]
+      == [digest(without(lambda: p.run(n, h, reg_s=1.0))[0])
+          for n, h in _fresh])
+_rf = [(n, h) for n in (2, 3) for h in (1200.0, -1500.0)]
+check("⭐ a refusal end, 10 s registers, N=2 and N=3, 1200 W and a surplus: "
+      "every row byte-identical with release/1.0 (A reads ~0 W and will not "
+      "give, so its side is its reading)",
+      [digest(run("refusal", n, h)) for n, h in _rf]
+      == [digest(without(lambda: run("refusal", n, h))) for n, h in _rf])
+_n1 = [dict(), dict(before=None, soc=soc_done)]
+check("⭐ N=1, A switched Off and A's drain completing, 10 s registers: "
+      "every row byte-identical with release/1.0",
+      [digest(p.run(1, 1200.0, **k)[0]) for k in _n1]
+      == [digest(without(lambda: p.run(1, 1200.0, **k))[0]) for k in _n1])
+
+# ============================================================================ #
+print("\n5. the rule's parts")
+_c = p.run(2, 1200.0, dur=4.0)[1]
+AB = {"inverter-a", "inverter-b"}
+ia, ib = _c.inverters
+
+
+def at(now, batts, cmds, changed):
+    """Set the clock, each battery's reading, its standing command and the
+    instant its reading last changed; then track."""
+    _c._now_mono = now
+    for inv, b, cmd, ch in zip(_c.inverters, batts, cmds, changed):
+        inv.batt, inv.last_write = b, cmd
+        inv.batt_fresh.changed_mono = ch
+    _c._track_reads()
+
+
+_c._read_span = {}
+at(10.0, (2000.0, -100.0), (2000.0, 0.0), (10.0, 10.0))
+check("a reading that changed is the start of its range: (R, R) joined "
+      "with the command on its wire",
+      _c._read_span[A] == (2000.0, 2000.0, 10.0)
+      and _c._read_span["inverter-b"] == (-100.0, 0.0, 10.0))
+at(11.0, (2000.0, -100.0), (0.0, 1140.0), (10.0, 10.0))
+at(12.0, (2000.0, -100.0), (0.0, 800.0), (10.0, 10.0))
+check("every command since the reading widens the range (a lagging "
+      "battery moves toward each in turn): A [0, 2000], B [-100, 1140]",
+      _c._read_span[A][:2] == (0.0, 2000.0)
+      and _c._read_span["inverter-b"][:2] == (-100.0, 1140.0))
+at(13.0, (1241.0, -100.0), (0.0, 800.0), (13.0, 10.0))
+check("a new reading starts the range again: A [0, 1241]",
+      _c._read_span[A][:2] == (0.0, 1241.0))
+_c._read_span = {}
+_c.inverters = _c.inverters[:1]
+at(14.0, (2000.0,), (2000.0,), (14.0,))
+_c.inverters = [ia, ib]
+check("N=1: nothing is tracked", _c._read_span == {})
+
+
+def tail(now, grid, target, held=None, live=AB):
+    """`_drain_grid_w` with A settling: the error it hands the loop (grid
+    minus aim) and the house point it holds."""
+    _c._now_mono, _c.grid, _c.target = now, grid, target
+    _c._drain_tail, _c._settle = {A}, {A: 0.0}
+    _c._drain_house = held
+    g = _c._drain_grid_w(live)
+    return round(g - AIM, 6), _c._drain_house
+
+
+# A: reading 1241 W from 13 s, range [0, 1241]; B: reading -100 W from 10 s,
+# range [-100, 1140]. Grid -330 W: the house is in
+# [-330 + 10 + 0 - 100, -330 + 10 + 1241 + 1140] = [-420, 2061].
+_c._read_span = {A: (0.0, 1241.0, 13.0), "inverter-b": (-100.0, 1140.0, 10.0)}
+ia.batt, ib.batt = 1241.0, -100.0
+ia.batt_fresh.changed_mono, ib.batt_fresh.changed_mono = 13.0, 10.0
+check("⭐ held: the first window cycle holds the loop's start (1140 W) while "
+      "the meter allows it (error 0; release/1.0's would be -330 + 1241 - "
+      "100 + 10 - 1140 = -319 W)", tail(14.0, -330.0, 1140.0) == (0.0, 1140.0)
+      and round(_release(_c, AB) - AIM) == -319)
+check("⭐ held, then kept: a later cycle keeps the held point, not the "
+      "loop's target", tail(15.0, -330.0, 1000.0, held=1140.0)
+      == (140.0, 1140.0))
+check("⭐ moved as far as the meter proves: a grid of +2000 W puts the "
+      "house at least at 2000 + 10 + 0 - 100 = 1910 W",
+      tail(15.0, 2000.0, 1140.0, held=1140.0) == (770.0, 1910.0))
+check("...and down: a grid of -2500 W puts it at most at -2500 + 10 + 1241 "
+      "+ 1140 = -109 W", tail(15.0, -2500.0, 1140.0, held=1140.0)
+      == (-1249.0, -109.0))
+ia.batt_fresh.changed_mono = 15.0
+check("⭐ a reading that changed this cycle is taken as it is: A at 1241 W, "
+      "not [0, 1241], so a grid of +2000 W puts the house at least at "
+      "2000 + 10 + 1241 - 100 = 3151 W", tail(15.0, 2000.0, 1140.0,
+                                              held=1140.0) == (2011.0, 3151.0))
+ia.batt_fresh.changed_mono = 13.0
+# A refuses to give (SAFETY-73): it reads 8 W, though it was asked 2000 W
+# since that reading. Grid -1500 W: the house is at most -1500 + 10 + 8 +
+# 1140 = -342 W, not -1500 + 10 + 2000 + 1140 = 1650 W.
+_c._read_span[A] = (8.0, 2000.0, 13.0)
+ia.batt = 8.0
+ia.refused = 1
+_ref1 = tail(15.0, -1500.0, 1140.0, held=1140.0)
+ia.refused = -1
+_c._read_span[A] = (-2000.0, 8.0, 13.0)
+_ref2 = tail(15.0, 1000.0, -500.0, held=-500.0)
+ia.refused = 0
+check("⭐ a battery that refuses one way is taken at its reading on that "
+      "side: will not give, asked 2000 W, reads 8 W -> the house at most "
+      "-342 W; will not take, asked -2000 W -> at least 1000 + 10 + 8 - 100 "
+      "= 918 W", _ref1 == (-1482.0, -342.0) and _ref2 == (1418.0, 918.0))
+_c._read_span[A], ia.batt = (0.0, 1241.0, 10.0), 1241.0
+check("a range from an older reading than the one in hand (a cycle the "
+      "loop did not run) is not used: A at its reading, 1241 W",
+      tail(15.0, 2000.0, 1140.0, held=1140.0) == (2011.0, 3151.0))
+_c._read_span[A] = (0.0, 1241.0, 13.0)
+check("⭐ a battery out of the split is not counted (the meter has its "
+      "power): B dropped, a grid of +2000 W puts the house at least at "
+      "2000 + 10 + 0 = 2010 W, not 1910 W",
+      tail(15.0, 2000.0, 1140.0, held=1140.0, live={A}) == (870.0, 2010.0))
+_c._drain_tail, _c._settle, _c._drain_house = {A}, {}, 1140.0
+_c.grid = -330.0
+check("⭐ the tail's end lets the hold go: the meter again, and the next "
+      "tail starts from the loop's start",
+      _c._drain_grid_w(AB) == -330.0 and _c._drain_house is None)
+
+# ============================================================================ #
+print("\n    drain end | N | house W | swing W now/release/1 s registers | "
+      "end-cycle others W now/release | site step W now/release | grid in "
+      "window W now | release")
+for k, n, h, a, b, f in TABLE:
+    print(f"    {k} | {n} | {h:.0f} | {a['swing']:.0f}/{b['swing']:.0f}/"
+          f"{f['swing']:.0f} | "
+          f"{a['first']:.0f}/{b['first']:.0f} | {a['step']:.0f}/"
+          f"{b['step']:.0f} | [{a['gmin']:.0f}, {a['gmax']:.0f}] | "
+          f"[{b['gmin']:.0f}, {b['gmax']:.0f}]")
+print()
+if fails:
+    print(f"{len(fails)} FAILED: " + ", ".join(fails))
+    sys.exit(1)
+print("all SAFETY-103 checks passed")

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