goodwe-addon/goodwe_controller/tools/safety103_plant.py

"""SAFETY-103's plant: test_safety95's lagging battery with the SAFETY-90
register model, driven through the REAL Controller in simulated time.

Each battery ramps at 800 W/s, then follows a first-order lag (`lag` s). Its
power register publishes every `reg_s` s at its own phase (the SAFETY-90/96
rig: A at 0.6, B at 0.16 of the period), and reads `off` W from the truth
(B idles at -100 W). The meter reads every cycle (`tick` s). A drain on A
runs from the first cycle (the others finished theirs); `before(c, t)` ends
it. `run()` has no side effects on import, so tests and probes share it.
"""

import asyncio
import math
import os
import random
import sys
import tempfile
from datetime import timedelta

HERE = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, HERE)
sys.path.insert(0, os.path.join(HERE, "tools"))

import opt11_identity as rig                                   # noqa: E402
from app import main as app_main, packmode                     # noqa: E402
from app.main import Controller                                # noqa: E402
from app.store import unit_key                                 # noqa: E402

A = "inverter-a"
T = 300.0               # A's drain ends here
PHASE = {"a": 0.6, "b": 0.16, "c": 0.38}
OFF = {"b": -100.0}     # the rig's B: idle reading -100 W
M = {"maintenance_enabled": True, "maintenance_start_hour": 11,
     "battery_modes_enabled": True}


def options(n, extra=None):
    o = rig._options(n, dict(M, **(extra or {})))
    if n > 1:
        o["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 "abc"[:n]]
    return o


def store(n):
    """A due for its monthly cycle (it drains from the first cycle)."""
    s = rig._Store()
    for u in "bc"[:n - 1]:
        s.data[unit_key(f"inverter-{u}", "last_completed")] = \
            rig.BASE.isoformat()
    return s


def off(c, t):
    """A switched Off at T (the SAFETY-95 check 5 end)."""
    if t >= T:
        c.store.data[unit_key(A, packmode.USE_KEY)] = False


def run(n, house, *, before=off, dur=T + 90.0, tick=1.0, reg_s=10.0,
        lag=10.0, noise=50.0, seed=103, phase=None, offs=None, extra=None,
        st=None, soc=None, dead=None, reject=None):
    """Rows of one run. Each row is a dict: t, cmd (the add-on's commands),
    target, grid (true), true (each battery's power), read (each register),
    age (s since each register published), phase (maintenance phases).
    `dead` (u, from s): u gives nothing from then on (board at 0 W).
    `reject` {u: from s}: writes to u are refused from then on."""
    units = tuple("abc"[:n])
    phase = PHASE if phase is None else phase
    offs = OFF if offs is None else offs
    rnd = random.Random(seed)
    soc = soc or (lambda u, t: 50.0)

    class Hass(rig._Hass):
        def __init__(self, *a):
            super().__init__(*a)
            self.tgt = {u: 0.0 for u in units}
            self.act = {u: 0.0 for u in units}
            self.reg = {u: 0.0 for u in units}
            self.pub_t = {u: 0.0 for u in units}
            self.slot = {u: None for u in units}
            self.tl = 0.0
            self.hn = 0.0

        def advance(self, t):
            for k in range(int(round((t - self.tl) / 0.1))):
                now = self.tl + (k + 1) * 0.1
                for u in units:
                    c = self.cmd[u]
                    if dead is not None and u == dead[0] and now >= dead[1]:
                        c = 0.0
                    d = c - self.tgt[u]
                    self.tgt[u] += math.copysign(min(abs(d), 80.0), d)
                    self.act[u] += (self.tgt[u] - self.act[u]) * (
                        1 - math.exp(-0.1 / lag))
                    s = math.floor((now - phase.get(u, 0.0) * reg_s) / reg_s)
                    if s != self.slot[u]:
                        self.slot[u] = s
                        self.reg[u] = float(round(self.act[u] + offs.get(u, 0.0)))
                        self.pub_t[u] = now
            self.tl = t
            self.hn = rnd.uniform(-noise, noise) if noise else 0.0

        def power(self, u):
            return self.reg[u]

        def grid(self):
            h = house(self.t) if callable(house) else house
            return float(round(h + self.hn - sum(self.act.values())))

        async def set_number(self, e, v):
            u = e[-1]
            if reject and u in reject and self.t >= reject[u]:
                return False
            self.cmd[u] = float(v)
            return True

    opts = options(n, extra)
    db = os.path.join(tempfile.mkdtemp(prefix="s103-"), "tariff.db")

    async def go():
        hass = Hass(units, soc, lambda u, t: False, lambda t: 0.0)
        c = Controller(opts, hass, st or store(n), rig._Pub(),
                       stop=asyncio.Event(), peak_db_path=db)
        for i in c.inverters:
            i.dev_min, i.dev_max = -5000.0, 5000.0
        c.auto = True
        real_dt, clock = app_main.datetime, [0.0]

        class _DT(real_dt):
            @classmethod
            def now(cls, tz=None):
                return rig.BASE + timedelta(seconds=clock[0])
        app_main.datetime = _DT
        rows = []
        try:
            t = 0.0
            while t < dur:
                hass.advance(t)
                hass.t = clock[0] = t
                for inv in c.inverters:
                    u = inv.name[-1]
                    if reject and u in reject and t >= reject[u] \
                            and inv.rejected_since is None:
                        inv.rejected_since = rig.BASE + timedelta(seconds=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()
                rows.append(dict(
                    t=t, cmd=tuple(i.target for i in c.inverters),
                    target=c.target, grid=hass.grid(),
                    true=tuple(round(hass.act[u]) for u in units),
                    read=tuple(hass.reg[u] for u in units),
                    age=tuple(round(t - hass.pub_t[u], 1) for u in units),
                    phase=tuple(m.phase for m in c.maints),
                    reason=c.loop_reason))
                t += tick
        finally:
            app_main.datetime = real_dt
        return rows, c
    return asyncio.run(go())


def end_at(rows):
    """The first cycle A's command left its drain (None if it never did)."""
    return next((r["t"] for r in rows if r["t"] >= T
                 and r["cmd"][0] < 1990.0), None)


def window(rows, span=23.0):
    """The rows from the end cycle to `span` s after it (the settle window,
    PLANT_SETTLE_DOWN_S)."""
    te = end_at(rows)
    return [r for r in rows if te is not None and te <= r["t"] <= te + span]


def swing(rows, i=1, span=23.0):
    """Battery i's command in the settle window after the end cycle, and
    its largest swing there: a move that is later undone (up then down, or
    down then up), the smaller of its two legs. A steady climb scores 0."""
    w = [r["cmd"][i] for r in window(rows, span)][1:]
    best = 0.0
    for j in range(len(w)):
        lo_b = min(w[:j + 1])
        hi_b = max(w[:j + 1])
        lo_a = min(w[j:])
        hi_a = max(w[j:])
        best = max(best, min(w[j] - lo_b, w[j] - lo_a),
                   min(hi_b - w[j], hi_a - w[j]))
    return best, w