SAFETY-85 review: grade discharge only, over the plant's settle time

Rule 2 graded each battery against last cycle's command and graded charges
too. On plantsim's battery (10 s lag) at the 2 s meter it warned at every
drain or charge start and 15 times an hour under the peak guard at N=2; a
charge tapering near full warned 11 times in 50 min (SAFETY-42 section 7).
Now a discharge is graded against the least it stood at over SHORT_CYCLES /
SHORT_SPAN_S. DOCS lists `mode-bound` under Loop reason. test_safety85 part 7:
peak guard with a dead battery, lagging plant, taper; part 3: the bound ends
with maintenance.

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

goodwe_controller/test_safety85.py

@@ -19,6 +19,8 @@ while maintenance holds a battery. Drives the REAL Controller
    (a battery at its cut-out) does not warn.
 5. The status row counts episodes, not cycles.
 6. The rules at their boundaries.
+7. Review: a battery the peak guard sets that gives nothing warns; a battery
+   that lags (10 s) and a charge that tapers do not.
 
 Mutations (MUTATION.md, purged copies) are listed in the pull request.
 """
@@ -82,13 +84,15 @@ def store(n, phase):
 
 def run(n, phase, house, cap, *, dur=HOUR, soc=None, cut=None, extra=None,
         maint=True):
-    """Rows and controller; `cap` is the peak limit (20000: never binds)."""
+    """Rows and controller; `cap` is the peak limit (20000: never binds);
+    `house` is watts or a function of time."""
     s84.OPT = dict(OPT0, peak_cap_w=cap, maintenance_enabled=maint)
     if soc is None:
         def soc(u, t):
             return 99.0 if phase == "hold" and u == "a" else 60.0
     try:
-        return s84.run(n, load=lambda t: house, cheap=(phase == "charge"),
+        load = house if callable(house) else (lambda t: house)
+        return s84.run(n, load=load, cheap=(phase == "charge"),
                        store=store(n, phase), soc=soc, dur=dur, cut=cut,
                        extra=dict({"maintenance_hold_min": 600,
                                    "maintenance_discharge_w": 2000},
@@ -209,7 +213,20 @@ check("a dip of two cycles is not an alarm", warned(_dip[1]) == (0, 0, 0))
 # ============================================================================ #
 print("\n3. control coming back from maintenance does not jump")
 END = {"maintenance_hold_min": 10}
-_h2 = run(2, "hold", 4500.0, 20000, dur=900.0, extra=END)
+_bounds = []
+_real_compute = app_main.compute
+
+
+def _spy_compute(*a, **kw):
+    _bounds.append(kw.get("bound"))
+    return _real_compute(*a, **kw)
+
+
+app_main.compute = _spy_compute
+try:
+    _h2 = run(2, "hold", 4500.0, 20000, dur=900.0, extra=END)
+finally:
+    app_main.compute = _real_compute
 _h2_old = no_bound(lambda: run(2, "hold", 4500.0, 20000, dur=900.0,
                                extra=END))
 _h1 = run(1, "hold", 4500.0, 20000, dur=900.0, extra=END)
@@ -224,6 +241,14 @@ print(f"    N=2 hold end: largest step {steps(_h2[0], 560, 700):.0f} W "
 check("⭐ N=2, the HOLD ends under a 4500 W house: the batteries move at most "
       "one slew step a cycle",
       _h2[1].maints[0].phase != "hold" and steps(_h2[0], 560, 700) <= SLEW)
+_held = [any(p != "idle" for p in r[-1][3]) for r in _h2[0]]
+print(f"    bound per cycle: {len(_bounds)} calls, {len(_held)} rows, "
+      f"{sum(_held)} in maintenance, "
+      f"{sum(b is not None for b in _bounds)} bounded")
+check("...and the bound is on while A holds and off from the cycle the HOLD "
+      "ends (review)",
+      len(_bounds) == len(_held) and any(not h for h in _held)
+      and all((b is not None) == h for b, h in zip(_bounds, _held)))
 check("⭐ N=2, A's charge pin lifts at a quarter's start: at most one slew "
       "step a cycle", steps(_cd[0], 1780, 1840) <= SLEW)
 check("...and N=1 already moved one slew step at a time (re-seeded each "
@@ -283,12 +308,11 @@ check("⭐ the page row reads the episodes, not the cycles "
 # ============================================================================ #
 print("\n6. the rules at their boundaries")
 DS = [((2000, 1499, 500), True), ((2000, 1500, 500), False),
-      ((2000, 2600, 500), False), ((-2000, -1499, 500), True),
-      ((-2000, -2600, 500), False), ((0, -900, 500), False),
-      ((0, 900, 500), False),
-      ((2000, -4, 500), True), ((-2000, 600, 500), True)]
-check("delivery_short: more than the margin short in the asked direction; "
-      "more than asked, or 0 W asked, is never short",
+      ((2000, 2600, 500), False), ((2000, -4, 500), True),
+      ((0, -900, 500), False), ((0, 900, 500), False),
+      ((-2000, -1499, 500), False), ((-2000, 600, 500), False)]
+check("delivery_short: a discharge more than the margin short; more than "
+      "asked, 0 W asked, or a charge (a full pack tapers) is never short",
       all(clamp.delivery_short(*a) is want for a, want in DS))
 check("the note names each battery, asked before given",
       clamp.shortfall_note([("inverter-a", 2000.0, -4.0)])
@@ -296,6 +320,129 @@ check("the note names each battery, asked before given",
       and clamp.shortfall_note([(None, -2000.0, 0.0)])
       == "OUTPUT_CLAMP: the battery gives +0 W of the -2000 W asked")
 
+# ============================================================================ #
+print("\n7. review: the peak guard, a battery that lags, a charge that tapers")
+import math                                                     # noqa: E402
+
+
+def under_guard(n, cut):
+    """A HOLD the peak guard raises (house 6500, limit 3500), 20 min."""
+    return run(n, "hold", 6500.0, 3500, dur=1200.0, cut=cut)
+
+
+_g1 = under_guard(1, lambda u, t: t >= 600)
+_g2 = under_guard(2, lambda u, t: u == "a" and t >= 600)
+print(f"    under the guard, the battery gives nothing: N=1 {warned(_g1[1])}, "
+      f"N=2 {warned(_g2[1])}")
+check("⭐ the peak guard sets the battery and it gives nothing: it warns at "
+      "N=1, and at N=2 (A, raised for the peak, gives nothing; B gives all "
+      "it is told)",
+      warned(_g1[1])[0] >= 1 and warned(_g2[1])[0] >= 1
+      and all("peak" in r[6] for rows, _c in (_g1, _g2) for r in rows
+              if 600 <= r[0] <= 620))
+check("...and it gives what it is told until then (no warning before 600 s)",
+      all(e.split(" ", 1)[0] >= "11:10:00" for c in (_g1[1], _g2[1])
+          for e in c.events if "OUTPUT_CLAMP" in e))
+
+
+def lagging(fn, lag=10.0):
+    """Run `fn` on plantsim's battery (800 W/s ramp, `lag` s first-order lag)
+    at the live 2 s meter tick."""
+    real_p, real_t = rig._Hass.power, rig.TICK
+
+    def power(self, u):
+        st = self.__dict__.setdefault("_plant", {})
+        a, tgt, tl = st.get(u, (0.0, 0.0, self.t))
+        for _ in range(int(round((self.t - tl) / 0.1))):
+            d = self.cmd[u] - tgt
+            tgt += math.copysign(min(abs(d), 80.0), d)
+            a += (tgt - a) * (1 - math.exp(-0.1 / lag))
+        st[u] = (a, tgt, self.t)
+        if self.cut(u, self.t) and self.cmd[u] >= -4.0:
+            return -4.0
+        return float(round(a))
+    rig._Hass.power, rig.TICK = power, 2.0
+    try:
+        return fn()
+    finally:
+        rig._Hass.power, rig.TICK = real_p, real_t
+
+
+def kettle(t):
+    """800 W, plus 2 kW 3 min in 10, plus 3 kW 1 min in 4."""
+    return (800.0 + (2000.0 if t % 600 < 180 else 0.0)
+            + (3000.0 if t % 240 < 60 else 0.0))
+
+
+def lag_runs():
+    return [run(n, ph, kettle, 3500, dur=1800.0)
+            for n, ph in ((1, "drain"), (2, "drain"), (2, "hold"))]
+
+
+_lag = lagging(lag_runs)
+_lag_old = patched(app_main, "SHORT_CYCLES", 1)(
+    lambda: patched(app_main, "SHORT_SPAN_S", 0.0)(
+        lambda: lagging(lag_runs)))
+_lag_dead = lagging(lambda: run(1, "drain", 1000.0, 20000, dur=900.0,
+                                cut=gone(600, 900)))
+print(f"    lagging battery, house steps under the guard: "
+      f"{[warned(c) for _r, c in _lag]}, graded on last cycle's command "
+      f"{[warned(c) for _r, c in _lag_old]}; gives nothing "
+      f"{warned(_lag_dead[1])}")
+check("⭐ a battery that lags (10 s) behind maintenance and peak-guard steps "
+      "at a 2 s meter does not warn (N=1 drain, N=2 drain and HOLD, 30 min)",
+      all(warned(c) == (0, 0, 0) for _r, c in _lag))
+check("non-vacuous: graded on last cycle's command alone, it warns",
+      sum(warned(c)[0] for _r, c in _lag_old) >= 3)
+check("...and a lagging battery that gives nothing still warns",
+      warned(_lag_dead[1])[0] >= 1)
+
+
+def taper(fn):
+    """Battery A takes the full charge until 600 s, then its acceptance falls
+    to 150 W over 30 min (constant-voltage taper)."""
+    real_p = rig._Hass.power
+
+    def power(self, u):
+        c = self.cmd[u]
+        if c < 0 and u == "a":
+            c = max(c, -max(150.0, 2500.0 - max(0.0, self.t - 600) / 1800
+                            * 2350.0))
+        return float(round(c))
+    rig._Hass.power = power
+    try:
+        return fn()
+    finally:
+        rig._Hass.power = real_p
+
+
+def charge_runs():
+    s84.OPT = dict(OPT0, peak_cap_w=20000)
+    try:
+        return [s84.run(n, load=lambda t: 800.0, cheap=True,
+                        store=s79.store_charging(n), soc=lambda u, t: 60.0,
+                        dur=2400.0) for n in (1, 2)]
+    finally:
+        s84.OPT = OPT0
+
+
+def both_ways(asked, given, margin):
+    asked, given = float(asked), float(given)
+    return asked != 0.0 and (asked - given) * math.copysign(1.0, asked) > margin
+
+
+_tp = taper(charge_runs)
+_tp_old = patched(app_main, "delivery_short", both_ways)(
+    lambda: taper(charge_runs))
+print(f"    a charge that tapers: {[warned(c) for _r, c in _tp]}, grading "
+      f"charges too {[warned(c) for _r, c in _tp_old]}")
+check("⭐ a maintenance charge that tapers near full does not warn (N=1, N=2)",
+      all(warned(c) == (0, 0, 0) for _r, c in _tp)
+      and all("charge" in {p for r in rows for p in r[-1][3]}
+              for rows, _c in _tp))
+check("non-vacuous: grading charges too, the taper warns",
+      all(warned(c)[0] >= 3 for _r, c in _tp_old))
+
 print()
 if fails:
     print(f"{len(fails)} FAILED: " + ", ".join(fails))