goodwe-addon/goodwe_controller/test_safety135.py

"""Runnable check for SAFETY-135: a young 15-minute period does not cost the
night charge the whole period.  `python3 test_safety135.py`

The REAL Controller (tools/opt11p3_replay.py, read_inputs + cycle, battery
modes, the peak guard on a P1 meter) with TEST-23-like lag: 1 s cycles,
readings 1 s old, a 5 s old meter in part 2. "Off" is the same run with
every measured stop latching and no wait for a falling give: the tree before
this card.
DECISIONS/SAFETY-135-young-period.md.

1. A peak-bound night charge; a period starts with a 4 s kettle spike. The
   guard cuts at once, and the charge resumes in that period (off: it stays
   off to the period's end). The period stays at or under the aim.
2. A cheap run starts at 01:00 right after a give run, with a 5 s meter: no
   start-then-cut within 60 s (off: one), and no period over the aim.
3. A real over-aim load: the guard and the charge stop on the same cycle as
   off, and the charge stays off while the load is on.
4. The SAFETY-136 rig run C/s5b (house 1 800 W, an oven 2 kW 45 s in every
   105 s, A Leftover solar 5 kW, B Cheap hours 4.6 kW): the charge starts at
   about 01:00:32 and the guard cuts it about 60 s later on the oven. It
   starts again 60 s after that cut (off: never in that period).
5. Repeated spikes in one period: at most one restart, and none sooner than
   RESTART_S after the first stop.

Mutations (MUTATION.md, purged copies) are listed in the pull request.
"""

import logging
import os
import sys
from datetime import datetime, timedelta, timezone

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

import opt11p3_replay as rp                                    # noqa: E402
from app import arbitrage, modetrade, packmode                 # noqa: E402
from named_absence import nth, present                         # noqa: E402

fails = []


def check(name, cond):
    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
    if not cond:
        fails.append(name)


L = timezone(timedelta(hours=1))
D0 = datetime(2026, 11, 3, tzinfo=L)
UTC0 = D0.astimezone(timezone.utc)
# Local quarters: dear 00:00-01:00 (a give run), cheap 01:00-07:00, 0.30 else.
PRICES = [0.40 if q < 4 else 0.10 if q < 28 else 0.30 for q in range(96)] * 2
WPTS = [(UTC0 + timedelta(minutes=15 * i), 900, p)
        for i, p in enumerate(PRICES)]
AIM = rp.PEAK_CAP - 50
CH = {"a": (True, packmode.CHEAP_HOURS, False),
      "b": (True, packmode.CHEAP_HOURS, False)}
T1 = D0 + timedelta(hours=1)                    # 01:00 local, the cheap edge
RESTART = getattr(modetrade, "RESTART_S", 60.0)  # 60.0 on older trees


class _Every:
    """A period that equals every period: `_charge_stop` set to it makes
    every measured stop latch, as before this card."""

    def __eq__(self, other):
        return True

    def __ne__(self, other):
        return False


def _off(c):
    # With RESTART_S 0 (see `off()`) nothing waits; every guard stop latches.
    c._charge_stop = (_Every(), float("-inf"))


def off(fn):
    """`fn()` with SAFETY-135 off: every measured stop latches and no start
    waits for a falling give, as on the tree before this card."""
    names = [n for n in ("YOUNG_S", "RESTART_S") if hasattr(modetrade, n)]
    real = {n: getattr(modetrade, n) for n in names}
    OFF[0] = True
    for n in names:                 # RESTART_S 0: no wait at a period's end
        setattr(modetrade, n, 0.0)
    try:
        return fn()
    finally:
        OFF[0] = False
        for n in names:
            setattr(modetrade, n, real[n])


OFF = [False]


def run(house_w, start, secs, meter_s=1.0, *, choice=None, soc0=None,
        limits=None, rating=None):
    """Rows per cycle and the Controller, for `house_w(seconds since 01:00)`."""
    t0 = start.timestamp()
    house = [(t0 + k, house_w(t0 + k - T1.timestamp()))
             for k in range(int(secs) + 5)]
    rows = []

    def cb(c, h, now, pr):
        g = c.peak_guard
        rows.append(dict(now=now, s=now.timestamp() - T1.timestamp(),
                         grid=h.grid(),
                         charge=getattr(c.arb, "action", None)
                         == arbitrage.CHARGE,
                         engaged=g is not None and g.engaged,
                         reason=getattr(c.arb, "reason", None) or ""))
    was = rp.RATING
    if rating is not None:
        rp.RATING = rating
    try:
        _st, c = rp.run("p3_ch_sell", house, WPTS,
                        start.astimezone(timezone.utc), secs, check=cb,
                        soc0=soc0 or {"a": 30.0, "b": 30.0},
                        choice=choice or CH, limits=limits,
                        n=len(choice or CH),
                        patch=_off if OFF[0] else None,
                        extra={"slew_w": 0, "cycle_cost_threshold_eur": 0.03},
                        lag=rp.Lag(meter_s=meter_s))
    finally:
        rp.RATING = was
    return rows, c


def banked(c, frm):
    """The add-on's own P1 period averages from `frm` (seconds after 01:00)."""
    return [b.offtake_avg_w for b in c.p1.blocks
            if b.start.timestamp() - T1.timestamp() >= frm]


def flaps(rows, frm, to):
    """Charge starts in [frm, to) s that stopped again within 60 s."""
    out = []
    for i in range(1, len(rows)):
        r = rows[i]
        if frm <= r["s"] < to and r["charge"] and not rows[i - 1]["charge"]:
            if any(not x["charge"] for x in rows[i:]
                   if x["s"] - r["s"] <= 60):
                out.append(r["s"])
    return out


# ============================================================================ #
print("1. a 4 s spike at the start of a peak-bound night charge's period")


def spike(s):
    return 2800.0 + (2500.0 if s >= 1800 and (s - 1800) % 900 < 4 else 0.0)


R1, C1 = run(spike, T1 + timedelta(minutes=20), 1560)
R1o, C1o = off(lambda: run(spike, T1 + timedelta(minutes=20), 1560))


def period(rows, frm):
    return [r for r in rows if frm <= r["s"] < frm + 900]


for tag, rows, c in (("on", R1, C1), ("off", R1o, C1o)):
    p = period(rows, 1800)
    cut = next((r["s"] - 1800 for r in p if r["engaged"]), None)
    on = sum(r["charge"] for r in p)
    print(f"    {tag}: guard cut at +{cut} s, charging {on} of {len(p)} s, "
          f"period {banked(c, 1800)[:1]}")
_p = period(R1, 1800)
_cut = next((i for i, r in enumerate(_p) if r["engaged"]), None)
check("the spike trips the guard in the period's first seconds and the "
      "charge stops there",
      _cut is not None and _p[_cut]["s"] - 1800 <= 5
      and not _p[_cut]["charge"])
_back = [r["s"] - 1800 for r in _p[(_cut or 0):] if r["charge"]]
check("the charge resumes in the same period, within 90 s of its start",
      _back and _back[0] <= 90)
_po = period(R1o, 1800)
check("off, the same spike keeps the charge off to the period's end (the "
      "defect this card fixes)",
      any(r["engaged"] for r in _po[:5])
      and not any(r["charge"] for r in _po[10:]))
check("every period with the charge is at or under the aim (lag allows 2 W)",
      banked(C1, 0) and max(banked(C1, 0)) <= AIM + 2.0)

# ============================================================================ #
print("\n2. a cheap run at 01:00 right after a give run, 5 s meter")
for house in (2400.0, 3000.0):
    def flat(s, w=house):
        return w
    R2, C2 = run(flat, T1 - timedelta(minutes=5), 1260, meter_s=5.0)
    R2o, C2o = off(lambda: run(flat, T1 - timedelta(minutes=5), 1260,
                               meter_s=5.0))
    f2, f2o = flaps(R2, 0, 900), flaps(R2o, 0, 900)
    p2, p2o = banked(C2, 0)[:1], banked(C2o, 0)[:1]
    on2 = sum(r["charge"] for r in period(R2, 0))
    on2o = sum(r["charge"] for r in period(R2o, 0))
    print(f"    house {house:.0f} W: on flaps {f2} charging {on2} s period "
          f"{p2}; off flaps {f2o} charging {on2o} s period {p2o}")
    check(f"house {house:.0f} W: no charge start is cut within 60 s in the "
          f"01:00 period", not f2)
    check(f"house {house:.0f} W: the 01:00 period is at or under the aim "
          f"(lag allows 2 W)", p2 and p2[0] <= AIM + 2.0)
    check(f"house {house:.0f} W: the charge still runs in the 01:00 period",
          on2 >= 800)
    if house == 3000.0:
        check("off, the 3 000 W house starts on the young period and is cut "
              "within 60 s (the defect this card fixes)", bool(f2o))

# ============================================================================ #
print("\n3. a real over-aim load still cuts on the same cycle")


def load(s):
    return 2800.0 + (4000.0 if 1820 <= s < 2120 else 0.0)


R3, C3 = run(load, T1 + timedelta(minutes=20), 1500)
R3o, C3o = off(lambda: run(load, T1 + timedelta(minutes=20), 1500))


def first(rows, key, val, frm):
    return next((r["s"] for r in rows if r["s"] >= frm and r[key] == val),
                None)


_g, _go = first(R3, "engaged", True, 1800), first(R3o, "engaged", True, 1800)
_s, _so = first(R3, "charge", False, 1800), first(R3o, "charge", False, 1800)
print(f"    guard cut at {_g} s (off {_go} s), charge stop at {_s} s "
      f"(off {_so} s)")
check("the guard cuts on the same cycle as off, and the charge stops on it",
      _g is not None and _g == _go and _s == _so == _g)
check("the charge stays off while the load is on",
      not any(r["charge"] for r in R3 if _g <= r["s"] < 2120))
check("every period is at or under the aim (lag allows 2 W)",
      banked(C3, 0) and max(banked(C3, 0)) <= AIM + 2.0)

# ============================================================================ #
print("\n4. the SAFETY-136 rig run C/s5b: start at 01:00:32, oven cut a minute "
      "later")
# The rig's house (s5b.ctl.json), in seconds from 01:00: 1 800 W, an oven of
# 2 000 W for 45 s in every 105 s from 00:30 to 01:05:45, 3 000 W 00:56-00:58.


def s5b(s):
    w = 1800.0
    if -1800 <= s < 345 and (s + 1800) % 105 < 45:
        w += 2000.0
    if -240 <= s < -120:
        w += 3000.0
    return w


S5B = dict(choice={"a": (True, packmode.LEFTOVER_SOLAR, False),
                   "b": (True, packmode.CHEAP_HOURS, False)},
           soc0={"a": 53.6, "b": 27.4}, limits={"b": (-4600.0, 4600.0)},
           rating=5000.0)
R4, C4 = run(s5b, T1 - timedelta(minutes=5), 1260, **S5B)
R4o, C4o = off(lambda: run(s5b, T1 - timedelta(minutes=5), 1260, **S5B))


def edges(rows, frm):
    """(start s, stop s) lists of the charge in the period from `frm`."""
    p = period(rows, frm)
    on = [round(b["s"] - frm) for a, b in zip(p, p[1:])
          if b["charge"] and not a["charge"]]
    of = [round(b["s"] - frm) for a, b in zip(p, p[1:])
          if a["charge"] and not b["charge"]]
    return on, of


_on4, _of4 = edges(R4, 0)
_on4o, _of4o = edges(R4o, 0)
print(f"    on: starts {_on4}, stops {_of4}, period {banked(C4, 0)[:1]}; "
      f"off: starts {_on4o}, stops {_of4o}, period {banked(C4o, 0)[:1]}")
check("off, as on the rig: the charge starts about 30 s in, the guard cuts "
      "it about a minute later, and it stays off to the period's end",
      len(_on4o) == 1 and 25 <= _on4o[0] <= 40 and len(_of4o) == 1
      and 80 <= _of4o[0] <= 100)
check("the same start and the same cut as off (the cut is not delayed)",
      _on4[:1] == _on4o[:1] and _of4[:1] == _of4o[:1])
_cut4 = nth(_of4, 0, "the guard's cut in s5b")
check("the charge starts again in that period, 60 s or more after the cut, "
      "once only",
      present(_cut4) and len(_on4) == 2 and _on4[1] - _cut4 >= RESTART - 1)
check("the period is at or under the aim (lag allows 2 W)",
      banked(C4, 0) and banked(C4, 0)[0] <= AIM + 2.0)

# ============================================================================ #
print("\n5. repeated spikes in one period: one restart, not too soon")
_SP = (0, 20, 40, 62, 80, 300, 500)


def spikes(s):
    return 2800.0 + (2500.0 if any(0 <= s - 1800 - k < 4 for k in _SP)
                     else 0.0)


for n_ch in (CH, {"a": CH["a"]}):
    R5, C5 = run(spikes, T1 + timedelta(minutes=20), 1560, choice=n_ch,
                 soc0={k: 30.0 for k in n_ch})
    _on5, _of5 = edges(R5, 1800)
    print(f"    N={len(n_ch)}: starts {_on5}, stops {_of5}, "
          f"period {banked(C5, 1800)[:1]}")
    check(f"N={len(n_ch)}: at most one restart in the period, none sooner "
          f"than RESTART_S after the first stop, and the last stop latches",
          _of5 and len(_on5) <= 1 and len(_of5) == len(_on5) + 1
          and all(t - _of5[0] >= RESTART - 1 for t in _on5))
    check(f"N={len(n_ch)}: the period is at or under the aim (lag allows "
          f"2 W)", banked(C5, 1800) and banked(C5, 1800)[0] <= AIM + 2.0)
    _w = [r["reason"] for r in period(R5, 1800) if not r["charge"]]
    check(f"N={len(n_ch)}: the wait says it in plain words (UX-08)",
          any("tries again in a minute" in x for x in _w)
          and not any("(young)" in x or "(wait)" in x for x in _w))

print()
if fails:
    print(f"FAILED: {len(fails)}")
    sys.exit(1)
print("all SAFETY-135 checks passed")