goodwe-addon/goodwe_controller/tools/opt20_scen.py

"""OPT-20: the refusal-probe scenarios, on the REAL Controller in the lagged
replay (tools/opt11p3_replay.py). Uses only the API bf9019e has, so the same
file runs on both trees (test_opt20.py pins the digests printed on bf9019e).

Battery B is Cheap hours and refuses to charge; it is probed every 15 minutes
inside the cheap run (OPT-15). A is, per scenario, in its maintenance charge
phase, idle (Solar first), or a normal charger in the same run.
"""

import hashlib
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.dirname(HERE))
logging.disable(logging.WARNING)

import opt11p3_replay as rp                                    # noqa: E402
from app import packmode                                       # noqa: E402
from app.maintenance import CHARGE, DRAIN, MaintResult         # noqa: E402

L = timezone(timedelta(hours=2))
A, B = "inverter-a", "inverter-b"
LAG = rp.Lag(tau_s=0.05, ramp_w_s=1e9)
SF_CH = {"a": (True, packmode.SOLAR_FIRST, False),
         "b": (True, packmode.CHEAP_HOURS, False)}
CH_CH = {"a": (True, packmode.CHEAP_HOURS, False),
         "b": (True, packmode.CHEAP_HOURS, False)}

# A day of the owner's kind: cheap 02:00-04:00 local, flat otherwise.
D0 = datetime(2026, 9, 9, 0, 0, tzinfo=L)
UTC0 = D0.astimezone(timezone.utc)
_PR = [0.30] * 192
for _i in range(8, 16):
    _PR[_i] = 0.10
WPTS = [(UTC0 + timedelta(minutes=15 * i), 900, p) for i, p in enumerate(_PR)]

# OPT-15 rig g7: prices 0 from 17:45 to 18:45 local, flat 0.06 otherwise.
G0 = datetime(2026, 10, 13, 0, 0, tzinfo=L)
GUTC0 = G0.astimezone(timezone.utc)
_GP = [0.06] * 192
for _i in range(71, 75):
    _GP[_i] = 0.0
for _i in range(76, 78):
    _GP[_i] = 0.5
G7PTS = [(GUTC0 + timedelta(minutes=15 * i), 900, p)
         for i, p in enumerate(_GP)]


def at(h, base=UTC0):
    return base + timedelta(hours=h)


def _rows(rows):
    def cb(c, h, now, pr):
        sh = dict(c.allocation.shares) if c.allocation is not None \
            and c.allocation.shares else {}
        rows.append(dict(
            now=now, grid=h.grid(), shares=sh,
            action=getattr(c.arb, "action", None),
            reason=getattr(c.arb, "reason", None) or "",
            probing=c.inverters[1].probe_from is not None,
            roles=dict(c.mode_tick.roles) if c.mode_tick else {},
            p={u: p.p for u, p in h.plants.items()}))
    return cb


def day(choice, *, a_maint=None, shaped=False, refuse=("b",), hours=0.5,
        house_w=800.0):
    """02:00..: a cheap run, `house_w` house. `a_maint` is A's forced maintenance
    result from 5 minutes in (B's refusal stands by then); `shaped` closes the
    maintenance charge's own watts (charge-only shaping, hi = 0)."""
    rows = []

    def patch(c):
        if shaped:
            c._charge_headroom = lambda floor, claims: 0.0
        if a_maint is None:
            return
        m, real = c.maints[0], c.maints[0].tick

        def tick(now, soc, blocked_by=None, gate_open=True):
            if now >= at(1.9 + 5 / 60):
                return a_maint
            return real(now, soc, blocked_by=blocked_by, gate_open=gate_open)
        m.tick = tick
    rp.run("p3_ch_sell", [(UTC0.timestamp() - 10, house_w)], WPTS, at(1.9),
           3600 * hours, check=_rows(rows), lag=LAG, n=2,
           soc0={"a": 50.0, "b": 30.0}, choice=choice, refuse=refuse,
           patch=patch, extra={"cycle_cost_threshold_eur": 0.10})
    return rows


def g7():
    """The OPT-15 rig's g7: B refuses from the start of the cheap run, A
    drains 18:00 to 18:17:04 and then sits in its maintenance charge phase
    (charge-only) at 0 W. The probe after the drain used to put the trade's
    2500 W on A."""
    rows = []
    t_end = at(18 + 17 / 60 + 4 / 3600, GUTC0)

    def patch(c):
        m, real = c.maints[0], c.maints[0].tick

        def tick(now, soc, blocked_by=None, gate_open=True):
            if at(18.0, GUTC0) <= now < t_end:
                return MaintResult(DRAIN, True, 2000.0, False, [])
            if now >= t_end:
                return MaintResult(CHARGE, False, None, True, [])
            return real(now, soc, blocked_by=blocked_by, gate_open=gate_open)
        m.tick = tick
    rp.run("p3_ch_sell", [(GUTC0.timestamp() - 10, 800.0)], G7PTS,
           at(17.6, GUTC0), 3600 * 1.1, check=_rows(rows), lag=LAG, n=2,
           soc0={"a": 17.0, "b": 73.0}, choice=SF_CH, refuse=("b",),
           patch=patch, extra={"cycle_cost_threshold_eur": 0.10})
    return rows


MAINT_CHARGE = MaintResult(CHARGE, False, None, True, [])


def windows(rows):
    """[(first, last)] row indexes of each probe window."""
    out, i = [], 0
    while i < len(rows):
        if rows[i]["probing"]:
            j = i
            while j + 1 < len(rows) and rows[j + 1]["probing"]:
                j += 1
            out.append((i, j))
            i = j + 1
        else:
            i += 1
    return out


def digest(rows) -> str:
    keep = [(r["now"].isoformat(), sorted(r["shares"].items()),
             round(r["grid"], 3), r["action"], r["reason"],
             sorted(r["roles"].items()), sorted(r["p"].items()))
            for r in rows]
    return hashlib.sha256(repr(keep).encode()).hexdigest()


def no_probe_digests() -> dict:
    """Battery modes on, no refusal, so no probe: a scripted cheap run."""
    return {
        "sf_ch": digest(day(SF_CH, refuse=(), hours=1.2)),
        "ch_ch": digest(day(CH_CH, refuse=(), hours=1.2)),
        "ch_ch_maint": digest(day(CH_CH, a_maint=MAINT_CHARGE, refuse=(),
                                  hours=1.2)),
    }


if __name__ == "__main__":
    for k, v in no_probe_digests().items():
        print(k, v)