SAFETY-79 review: margin under a banked month peak too; identity harness reaches headroom and maintenance

- quarter_target_w() is now the margin under quarter_ceiling_w(): a peak
  banked between cap - margin and the cap was aimed at exactly, so one
  overshoot crossed the cap; above the cap the same overshoot raised the bill.
- safety79_identity.py hashes charge_headroom_w() and the shaping suspension,
  and adds n1_maint/n2_maint (cheap window, forecast across cap - 50 .. cap).
  Digests re-printed on ed739d4 and pinned; margin 0 still identical.
- test_margin_under_banked_peak; DECISIONS and DOCS updated.

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

goodwe_controller/tools/safety79_identity.py

@@ -6,7 +6,11 @@
 tools/opt11_identity.py's harness (the REAL Controller, read_inputs + cycle, in
 simulated time) with a P1 meter, so the peak guard runs, and a peak limit the
 house goes over. Each row hashes opt11_identity's fields plus the guard's floor,
-limit and sentence. It uses only the API release/1.0 ed739d4 already had, so
+limit and sentence, its charge headroom (what arbitrage is sized to) and the
+maintenance shaping's suspension. The two `_maint` scenarios run a maintenance
+charge in a cheap window with a demand forecast that crosses cap - 50 .. cap,
+so `shaping_suspended()` and `maintenance_charge_floor()` are both reached.
+It uses only the API release/1.0 ed739d4 already had, so
 the same file runs on the old tree (which ignores `peak_margin_w`) and the new.
 test_peak_margin.py part 1 holds the new tree to the digests printed on ed739d4.
 """
@@ -22,7 +26,7 @@ from datetime import timedelta
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 
 import opt11_identity as rig                               # noqa: E402
-from app import packmode                                  # noqa: E402
+from app import packmode, peakguard                       # noqa: E402
 from app.p1 import make_sample, split_signed              # noqa: E402
 from app.store import unit_key                            # noqa: E402
 
@@ -62,9 +66,43 @@ def store_off(name):
     return s
 
 
+def forecast(t):
+    """A demand forecast that sweeps 2300-2600 W every 10 minutes: across
+    cap - 50 and the cap, so the margin moves the forecast checks."""
+    return 2300.0 + (t % 600) / 600 * 300
+
+
+def _feed_maint(c, t):
+    """`_feed`, plus the forecast and a cheap price (now under average)."""
+    if not getattr(c.hass, "_s79", False):
+        read = c.hass.number
+        fixed = {"sensor.pnow": 0.05, "sensor.pavg": 0.20}
+
+        async def number(e, invert=False):
+            if e == "sensor.fc":
+                return forecast(c.hass.t)
+            if e in fixed:
+                return fixed[e]
+            return await read(e, invert)
+        c.hass.number, c.hass._s79 = number, True
+    _feed(c, t)
+
+
+def store_charging(n):
+    """A store where the first battery is in its maintenance CHARGE phase."""
+    s = rig._Store()
+    for k in ("phase", "phase_started"):
+        s.data[k if n == 1 else unit_key("inverter-a", k)] = (
+            "charge" if k == "phase" else rig.BASE.isoformat())
+    return s
+
+
 P1 = {"meter_source": "ha_signed", "p1_net_entity": "sensor.net",
       "meter_phases": 1, "peak_cap_w": CAP_W}
 MODES = {"battery_modes_enabled": True}
+MAINT = {"maintenance_enabled": True, "maintenance_start_hour": 11,
+         "peak_forecast_entity": "sensor.fc", "price_now_entity": "sensor.pnow",
+         "price_avg_entity": "sensor.pavg"}
 
 SCENARIOS = {
     "n1": dict(n=1),
@@ -73,28 +111,34 @@ SCENARIOS = {
     "n1_off": dict(n=1, modes=True, off="inverter-a"),
     # OPT-11: B Off, so B covers what A cannot.
     "n2_off": dict(n=2, modes=True, off="inverter-b"),
+    # A maintenance charge in a cheap window, with a demand forecast.
+    "n1_maint": dict(n=1, maint=True),
+    "n2_maint": dict(n=2, maint=True),
 }
 
 
-def run(n, *, margin=None, modes=False, off=None, dur=2700.0):
+def run(n, *, margin=None, modes=False, off=None, maint=False, dur=2700.0):
     """Rows for one scenario; `margin=None` leaves the key out."""
-    extra = dict(P1, **(MODES if modes else {}))
+    extra = dict(P1, **(MODES if modes else {}), **(MAINT if maint else {}))
     if margin is not None:
         extra["peak_margin_w"] = margin
     guard = []
 
     def on_cycle(c, t):
         g = c.peak_guard
-        guard.append(None if g is None else (g.engaged, g.floor_w, g.ceiling_w,
-                                             g.ceiling_from, g.reason))
+        guard.append((c._maint_shape_suspended, None if g is None else (
+            g.engaged, g.floor_w, g.ceiling_w, g.ceiling_from, g.reason,
+            peakguard.charge_headroom_w(g))))
     db = os.path.join(tempfile.mkdtemp(prefix="s79-"), "tariff.db")
     real = rig.Controller
     rig.Controller = functools.partial(real, peak_db_path=db)
     try:
         rows, c = rig.run(n, extra=extra, dur=dur, load=load, soc=soc,
-                          cut=lambda u, t: False, before=_feed,
+                          cut=lambda u, t: False,
+                          before=_feed_maint if maint else _feed,
                           on_cycle=on_cycle,
-                          store=store_off(off) if off else None)
+                          store=(store_off(off) if off else
+                                 store_charging(n) if maint else None))
     finally:
         rig.Controller = real
     return [r + (g,) for r, g in zip(rows, guard)], c