Merge release/1.0 (SAFETY-78, #154) into opt11-p4-solar-forecast

Conflicts: modetrade.decide() takes both the day plan (targets) and
SAFETY-78's draining flag; the Why line keeps SAFETY-78's other_cycle
ahead of the plan's word (test added).

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

goodwe_controller/tools/safety78_identity.py

@@ -0,0 +1,110 @@
+#!/usr/bin/env python3
+"""SAFETY-78: at one inverter, is a maintenance drain byte-identical to before?
+
+    python3 tools/safety78_identity.py          # prints one digest per scenario
+
+tools/safety79_identity.py's harness (the REAL Controller on a P1 meter, so
+the peak guard runs) with the battery draining for its monthly cycle from the
+first cycle, under a house below, at and above the drain rate, a solar
+surplus and a load over the peak limit. Each row hashes safety79's fields,
+the modes' roles and the trade, the action log's newest line, and every claim
+main.py hands `resolve()`, so a claim or a log line added at N=1 that changes
+no watt still moves the digest. The `_modes` scenarios run battery modes on,
+Cheap hours only, over opt11p3_identity's prices. Uses only the API release/1.0 5a4cdc2 has, so the same file runs on
+both trees; test_safety78.py part 5 holds this tree to the digests printed
+there.
+"""
+
+import functools
+import hashlib
+import os
+import sys
+import tempfile
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+import opt11_identity as rig                               # noqa: E402
+import opt11p3_identity as p3                              # noqa: E402
+import safety79_identity as s79                            # noqa: E402
+import app.main as m                                       # noqa: E402
+from app import packmode                                  # noqa: E402
+from app.store import unit_key                            # noqa: E402
+
+DUR = 2400.0
+DRAIN = {"maintenance_enabled": True, "maintenance_start_hour": 11}
+
+
+def load(t):
+    """Below the 2000 W drain, at it, above it, a surplus, over the limit."""
+    k = int(t // 480)
+    return (1200.0, 2000.0, 3000.0, -1500.0, 6000.0)[min(k, 4)]
+
+
+def store_cheap(n):
+    s = rig._Store()
+    for u in ("inverter-a", "inverter-b")[:n]:
+        s.data[unit_key(u, packmode.MODE_KEY)] = packmode.CHEAP_HOURS
+    return s
+
+
+SCENARIOS = {
+    "n1_drain": dict(n=1),
+    "n1_drain_modes": dict(n=1, modes=True),
+}
+
+
+def run(n, *, modes=False, dur=DUR):
+    extra = dict(s79.P1, **DRAIN)
+    if modes:
+        extra.update(p3.ARB, battery_modes_enabled=True)
+    seen, extra_rows = [], []
+    real_resolve = m.resolve
+
+    def spy(claims, **kw):
+        seen.append(tuple((c.source, c.priority, c.kind, c.value, c.lo, c.hi,
+                           c.reason, c.actuator) for c in claims))
+        return real_resolve(claims, **kw)
+
+    def on_cycle(c, t):
+        g, tick, a = c.peak_guard, c.mode_tick, c.arb
+        extra_rows.append((
+            seen[-1] if seen else None,
+            None if g is None else (g.engaged, g.floor_w, g.reason),
+            None if tick is None else (tuple(sorted(tick.roles.items())),
+                                       tick.action, tick.why),
+            None if a is None else (a.action, a.power_w, a.reason),
+            [mm.phase for mm in c.maints],
+            len(c.events), c.events[0] if c.events else None))
+    db = os.path.join(tempfile.mkdtemp(prefix="s78-"), "tariff.db")
+    real = rig.Controller
+    rig.Controller = functools.partial(real, peak_db_path=db)
+    m.resolve = spy
+    try:
+        rows, c = rig.run(n, extra=extra, dur=dur, load=load, soc=s79.soc,
+                          cut=lambda u, t: False,
+                          before=p3._feed if modes else s79._feed,
+                          on_cycle=on_cycle,
+                          store=store_cheap(n) if modes else None)
+    finally:
+        rig.Controller = real
+        m.resolve = real_resolve
+    return [r + (x,) for r, x in zip(rows, extra_rows)], c
+
+
+def digest(rows) -> str:
+    return hashlib.sha256(repr(rows).encode()).hexdigest()
+
+
+def digests() -> dict:
+    return {k: digest(run(**v)[0]) for k, v in SCENARIOS.items()}
+
+
+if __name__ == "__main__":
+    if "--summary" in sys.argv:
+        for k, v in SCENARIOS.items():
+            rows = run(**v)[0]
+            print(f"{k}: {len(rows)} cycles, phases "
+                  f"{sorted({str(r[-1][-1]) for r in rows})}, targets "
+                  f"{sorted({r[1] for r in rows})[:6]}")
+    for k, v in digests().items():
+        print(f"{k} {v}")