goodwe-addon/goodwe_controller/tools/opt11_identity.py

#!/usr/bin/env python3
"""OPT-11: with battery modes off, is every command byte-identical to before?

    python3 tools/opt11_identity.py          # prints one digest per scenario

Drives the REAL `Controller` (read_inputs + cycle: the freshness gate, the
arbiter, the allocator, OPT-09's pack claim, SAFETY-73's refusal claims) in
simulated time, and hashes, per cycle, everything a command depends on: the
site target, each inverter's written value, the loop's reason, integrator and
saturation count, and the allocation record. It uses only the Controller API
that release/1.0 a5dddad already had, so the SAME file runs on the old tree and
the new one. test_modes.py part 1 holds the new tree to the digests this file
printed on a5dddad (they are written there, with that provenance).

Scenarios: N=1 and N=2, a house that steps, a surplus of solar, a battery
reaching its cut-out level and, at N=2, a battery that refuses (SAFETY-73) and
is probed.
"""

import asyncio
import hashlib
import os
import sys
from datetime import datetime, timedelta, timezone

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import app.main as m                                      # noqa: E402
from app.main import Controller                           # noqa: E402

TICK = 5.0
BASE = datetime(2026, 9, 20, 11, 0, tzinfo=timezone.utc)


class _Store:
    def __init__(self):
        from app.store import DEFAULTS
        self.data = dict(DEFAULTS)

    def set(self, k, v):
        self.data[k] = v

    def get_time(self, k):
        v = self.data.get(k)
        return datetime.fromisoformat(v) if v else None

    def set_time(self, k, when):
        self.data[k] = when.isoformat() if when else None


class _Pub:
    enabled = False

    def publish(self, *a, **k):
        pass

    def close(self):
        pass


def house(t):
    """A house that steps, then a solar surplus, then an evening load."""
    if t < 300:
        return 800.0
    if t < 900:
        return 3200.0
    if t < 1500:
        return -2600.0          # more solar than the house uses
    if t < 2100:
        return 1400.0
    return 4200.0


class _Hass:
    """One meter, `units` batteries that follow their command unless `cut`."""

    def __init__(self, units, soc, cut, load=house):
        self.units, self.soc, self.cut, self.load = units, soc, cut, load
        self.cmd = {u: 0.0 for u in units}
        self.t = 0.0

    def power(self, u):
        c = self.cmd[u]
        if self.cut(u, self.t) and c >= -4.0:
            return -4.0
        return float(round(c))

    def grid(self):
        return self.load(self.t) - sum(self.power(u) for u in self.units)

    async def limits(self, e):
        return (-5000.0, 5000.0, 1.0)

    async def call(self, *a):
        return True

    async def number(self, e, invert=False):
        if e == "sensor.grid":
            return self.grid()
        u = e[-1]
        if e.startswith("sensor.soc"):
            return self.soc(u, self.t)
        if e.startswith("sensor.batt"):
            return self.power(u)
        return None

    async def set_number(self, e, v):
        self.cmd[e[-1]] = float(v)
        return True


def _options(n, extra):
    common = {"meter_entity": "sensor.grid", "meter_source": "off",
              "gain": 0.6, "slew_w": 400, "target_grid_w": -10,
              "deadband_w": 15, "step_w": 10, "heartbeat_s": 10,
              "stale_input_s": 15, "auto_start": True}
    if n == 1:
        common.update({"setpoint_entity": "number.sp_a",
                       "soc_entity": "sensor.soc_a",
                       "batt_entity": "sensor.batt_a", "max_w": 5000,
                       "bms_soc_floor": 10})
    else:
        common["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 "ab"[:n]]
    common.update(extra or {})
    return common


def _soc(u, t):
    # A empties toward its cut-out level over the run; B stays half full.
    return max(8.0, 60.0 - t / 40.0) if u == "a" else 50.0


def run(n, *, extra=None, dur=2700.0, store=None, on_cycle=None, load=house,
        soc=_soc, cut=None, before=None):
    """Cycle a real Controller; returns (rows, controller). `before(c, t)`
    runs before each cycle, `on_cycle(c, t)` after it."""
    units = tuple("ab"[:n])
    if cut is None:
        def cut(u, t):
            return n > 1 and u == "a" and 1600.0 <= t < 2300.0

    async def go():
        hass = _Hass(units, soc, cut, load)
        c = Controller(_options(n, extra), hass, store or _Store(), _Pub(),
                       stop=asyncio.Event())
        for i in c.inverters:
            i.dev_min, i.dev_max = -5000.0, 5000.0
        c.auto = True
        real_dt = m.datetime
        clock = [0.0]

        class _DT(real_dt):
            @classmethod
            def now(cls, tz=None):
                return BASE + timedelta(seconds=clock[0])

        m.datetime = _DT
        rows = []
        try:
            t = 0.0
            while t < dur:
                hass.t = clock[0] = 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()
                alloc = getattr(c, "allocation", None)
                rows.append((t, c.target, tuple(sorted(hass.cmd.items())),
                             c.loop_reason, round(float(c.i_w), 6),
                             c.sat_count, c.reason,
                             alloc.explain() if alloc is not None else None))
                if on_cycle is not None:
                    on_cycle(c, t)
                t += TICK
        finally:
            m.datetime = real_dt
        return rows, c
    return asyncio.run(go())


def digest(rows) -> str:
    return hashlib.sha256(repr(rows).encode()).hexdigest()


SCENARIOS = {
    "n1": dict(n=1),
    "n2": dict(n=2),
    # The option present and false: what an install that saved the new
    # option without turning it on runs. The old tree ignores the key.
    "n1_off": dict(n=1, extra={"battery_modes_enabled": False}),
    "n2_off": dict(n=2, extra={"battery_modes_enabled": False}),
}


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, loop reasons "
                  f"{sorted({r[3] for r in rows})}, arbiter "
                  f"{sorted({r[6].split(' ->')[0] for r in rows})}")
    for k, v in digests().items():
        print(f"{k} {v}")