SAFETY-103: no swing on old battery readings after a drain ends (#180)

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/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