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; the same with both batteries due for maintenance; and the arbiter
alone on 6000 seeded claim lists (peak floor, trades, maintenance, refusals).
"""

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 state(self, e):
        """SAFETY-118: the peak forecast is read as a state object; every
        number this rig serves is in watts."""
        v = await self.number(e)
        return None if v is None else {
            "state": str(v), "attributes": {"unit_of_measurement": "W"}}

    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()


# Both batteries due for their monthly cycle at the run's first cycle, so the
# maintenance slot, its pins and its shaping are in every row (PR #149 review:
# the scenarios above never reach maintenance).
_MAINT = {"maintenance_enabled": True, "maintenance_start_hour": 11}

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}),
    "n1_maint": dict(n=1, extra=_MAINT),
    "n2_maint": dict(n=2, extra=_MAINT),
}


def resolve_rows(cases=6000, seed=1149):
    """The arbiter alone, on seeded random claim lists with no mode claim:
    the peak floor, a trade, the loop, maintenance pins and shaping, SAFETY-73
    refusal limits and probe pins, packs, ratings and safety stops, at one to
    three inverters. The peak guard needs a P1 meter the cycle runs above do
    not have, so its claim is graded here. Uses only a5dddad's API."""
    import random
    from math import inf
    from app import arbiter as a
    rnd = random.Random(seed)
    rows = []
    for _ in range(cases):
        n = rnd.choice((1, 2, 3))
        units = ("inverter-a", "inverter-b", "inverter-c")[:n]

        def act(src, u):
            return a.actuator_for(src, u, units)
        cl = []
        for u in units:
            r = rnd.choice((1000.0, 2500.0, 5000.0))
            cl.append(a.Claim.limit("supervised", a.P_SAFETY, -r, r, "max_w",
                                    actuator=act("supervised", u)))
            x = rnd.random()
            if x < 0.15:
                cl.append(a.Claim.limit("pack", a.P_SAFETY, *rnd.choice(
                    ((0.0, inf), (-inf, 0.0))), actuator=act("pack", u)))
            elif x < 0.25:
                cl.append(a.Claim.limit("refused", a.P_SAFETY, *rnd.choice(
                    ((-inf, 0.0), (0.0, inf))), "refused",
                    actuator=act("refused", u)))
            elif x < 0.32:
                cl.append(a.Claim.set("refused", a.P_SAFETY,
                                      rnd.choice((200.0, -200.0)), "probing",
                                      actuator=act("refused", u)))
            elif x < 0.42:
                cl.append(a.Claim.set("maintenance", a.P_MAINTENANCE,
                                      rnd.choice((2000.0, -2000.0)), "drain",
                                      actuator=act("maintenance", u)))
            elif x < 0.5:
                cl.append(a.Claim.limit("maintenance", a.P_MAINT_SHAPE,
                                        hi=-rnd.choice((500.0, 2000.0)),
                                        actuator=act("maintenance", u)))
        y = rnd.random()
        if y < 0.8:
            cl.append(a.Claim.set("loop", a.P_LOOP, rnd.uniform(-8000, 8000)))
        if 0.5 < y < 0.9:
            cl.append(a.Claim.set("arbitrage", a.P_ARBITRAGE,
                                  rnd.uniform(-6000, 6000)))
        if rnd.random() < 0.6:
            cl.append(a.Claim.limit("peak", a.P_PEAK,
                                    lo=rnd.uniform(-4000, 6000)))
        if rnd.random() < 0.05:
            cl.append(a.Claim.set("safety", a.P_SAFETY, 0.0, "stopped"))
        rnd.shuffle(cl)
        r = a.resolve(cl, units=units if n > 1 else None,
                      weights=(None if rnd.random() < 0.5 else
                               {u: rnd.choice((0.0, 0.5, 1.0)) for u in units}))
        rows.append((r.target_w, r.winner, r.bound_by, r.contradiction,
                     tuple(sorted((r.shares or {}).items())), r.allocated,
                     r.allocation.explain() if r.allocation else None))
    return rows


def digests() -> dict:
    out = {k: digest(run(**v)[0]) for k, v in SCENARIOS.items()}
    out["resolve"] = digest(resolve_rows())
    return out


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})}")
        rr = resolve_rows()
        print(f"resolve: {len(rr)} claim lists, winners "
              f"{sorted({r[1] for r in rr})}, bound by "
              f"{sorted({str(r[2]).split('(')[0] for r in rr})}")
    for k, v in digests().items():
        print(f"{k} {v}")