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
`modetrade.YOUNG_S = 0`: 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.

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

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


def off(fn):
    """`fn()` with SAFETY-135 off: no period is young."""
    real = getattr(modetrade, "YOUNG_S", None)
    if real is None:
        return fn()
    modetrade.YOUNG_S = 0.0
    try:
        return fn()
    finally:
        modetrade.YOUNG_S = real


def run(house_w, start, secs, meter_s=1.0):
    """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))
    _st, c = rp.run("p3_ch_sell", house, WPTS, start.astimezone(timezone.utc),
                    secs, check=cb, soc0={"a": 30.0, "b": 30.0}, choice=CH,
                    extra={"slew_w": 0, "cycle_cost_threshold_eur": 0.03},
                    lag=rp.Lag(meter_s=meter_s))
    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()
if fails:
    print(f"FAILED: {len(fails)}")
    sys.exit(1)
print("all SAFETY-135 checks passed")