goodwe-addon/goodwe_controller/tools/safety79_identity.py

#!/usr/bin/env python3
"""SAFETY-79: with `peak_margin_w: 0`, is every command byte-identical to before?

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

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, 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.
"""

import functools
import hashlib
import os
import sys
import tempfile
import time
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, peakguard                       # noqa: E402
from app.p1 import make_sample, split_signed              # noqa: E402
from app.store import unit_key                            # noqa: E402

CAP_W = 2500


def load(t):
    """A house under the limit, over it, a solar surplus, far over, over."""
    if t < 300:
        return 800.0
    if t < 900:
        return 3200.0
    if t < 1500:
        return -2600.0
    if t < 2100:
        return 6000.0
    return 4200.0


def soc(u, t):
    return 60.0         # above the peak reserve: the guard may discharge


def _feed(c, t):
    """One P1 telegram per cycle: the meter reads what the house and the
    batteries' measured power leave."""
    imp, exp = split_signed(c.hass.grid())
    c.p1.submit(make_sample("ha_signed", imp, exp, phases=1,
                            ingest_ts=rig.BASE + timedelta(seconds=t),
                            ingest_mono=time.monotonic()))


def store_off(name):
    """A store where battery `name` is switched Off (OPT-11)."""
    s = rig._Store()
    s.data[unit_key(name, packmode.USE_KEY)] = False
    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),
    "n2": dict(n=2),
    # OPT-11: the only battery Off, so only the peak guard makes it give.
    "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, maint=False, dur=2700.0):
    """Rows for one scenario; `margin=None` leaves the key out."""
    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((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_maint if maint else _feed,
                          on_cycle=on_cycle,
                          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


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


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


if __name__ == "__main__":
    m = None if len(sys.argv) < 2 else int(sys.argv[1])
    for k, v in digests(m).items():
        print(f"{k} {v}")