"""Runnable check for SAFETY-84: a maintenance charge at N=2 no longer stalls.
`python3 test_safety84.py`

In a cheap window the charging battery's own `hi` makes it charge. At two
inverters the loop's site target sat above it, the other battery had to give,
the allocator refused the opposite directions and the site sat at 0 W. Now a
site `hi` (`charge`, P_MAINT_SHAPE) keeps the site target at or below the
charge, and the charge is sized to the quarter's headroom so it yields to the
peak guard. Drives the REAL Controller (tools/safety84_identity.py's harness).
DECISIONS/SAFETY-84-charge-at-two.md has the record.

1. N=2 in a cheap window, house below, at and above the charge rate and a
   solar surplus: the charge runs, no opposite split, no refusal, and it ends.
2. The peak guard: the charge yields just enough to hold the aim (cap -
   margin, and a banked month peak - margin); over the aim the guard wins,
   the charging battery gives too, and nothing parks; no false 'Load too
   high'. Without P1 the charge is not sized, as at N=1.
3. Outside a cheap window nothing changes.
4. A safety stop parks both batteries without a refusal, and the charge
   resumes after it.
5. One battery refusing, or dropped, and B Off with battery modes on.
6. A trade outranked by the charge (P_ARBITRAGE < P_MAINT_SHAPE).
7. A seeded sweep of the ceiling against the loop, trades, refusals, packs
   and modes.
8. The N=1 digests equal release/1.0 9d5b7d1, and the N=2 ones are pinned.

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

import logging
import os
import random
import sys

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 opt11_identity as rig                                   # noqa: E402
import opt11p3_identity as p3                                  # noqa: E402
import safety84_identity as s84                                # noqa: E402
from app import arbitrage, packmode                            # noqa: E402
from app.arbiter import (Claim, P_ARBITRAGE, P_LOOP, P_MAINT_SHAPE,  # noqa
                         P_MODE, P_SAFETY, resolve)
from app.main import Controller                                # noqa: E402
from app.maintenance import CHARGE, HOLD                       # noqa: E402
from app.store import unit_key                                 # noqa: E402

fails = []


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


A, B = "inverter-a", "inverter-b"
AIM = 3450.0            # peak_cap_w 3500 - peak_margin_w 50
TOL = 10.0              # W over the aim a banked quarter may show (SAFETY-81)


def cmd(r):
    return dict(r[2])


def refused(rows):
    return sum(1 for r in rows if r[7] and "REFUSED" in r[7])


def opposite(rows):
    return sum(1 for r in rows if cmd(r)["a"] * cmd(r)["b"] < 0)


def charge_claims(r):
    """The site `charge` claims main.py handed resolve() on row `r`."""
    return [c for c in r[-1][0] if c[0] == "charge"]


def banked(c):
    return [b.offtake_avg_w for b in c.p1.blocks]


def alarmed(c):
    """Did SAFETY-81's 'Load too high' alarm fire on this run? Its quarter
    stamp too, since ALLOCATOR rows can push its row out of the 40-row log
    (SAFETY-85)."""
    return c._peak_alarm_block is not None or any(
        "PEAK_GUARD_SATURATED" in e for e in c.events)


def run(house, n=2, **kw):
    return s84.run(n, load=house if callable(house) else lambda t: house,
                   **kw)


def rise(u, t):
    """A fills from 60 % (1 % a cycle); B stays at 50 %."""
    return min(100.0, 60.0 + t / 5.0) if u == "a" else 50.0


# ============================================================================ #
print("1. N=2 in a cheap window: the charge runs whatever the house uses")
RUNS = {h: run(h, dur=900.0) for h in (800.0, 2000.0, 3000.0, -1500.0)}
for h, (rows, c) in RUNS.items():
    print(f"    house {h:+.0f} W: last A {cmd(rows[-1])['a']:+.0f} W, B "
          f"{cmd(rows[-1])['b']:+.0f} W; {rows[-1][6]}; refusals "
          f"{refused(rows)}")
check("⭐ below, at and above the charge rate and under a solar surplus, "
      "battery A charges on EVERY cycle (it used to sit at 0 W from the "
      "first cycle)",
      all(cmd(r)["a"] < 0.0 for rows, _c in RUNS.values() for r in rows)
      and all(c.maints[0].phase == CHARGE for _r, c in RUNS.values()))
check("⭐ no cycle splits the site both ways and the allocator refuses "
      "nothing: battery B never gives while A charges",
      all(opposite(rows) == 0 and refused(rows) == 0
          and all(cmd(r)["b"] <= 0.0 for r in rows)
          for rows, _c in RUNS.values()))
check("...and nothing is logged per cycle: no ARBITER line in the action log",
      not any("ARBITER" in e for _r, c in RUNS.values() for e in c.events))
check("with the quarter's headroom above the charge rate, A charges at the "
      "full 2000 W and the Why names the charge",
      all(cmd(r)["a"] == -2000.0 for h in (800.0, -1500.0)
          for r in RUNS[h][0])
      and "limited by charge" in RUNS[800.0][0][-1][6])
print(f"    Why: {RUNS[800.0][1].why_words}")
check("...and the page's Why reads as at one battery: 'held back by battery "
      "maintenance'",
      RUNS[800.0][1].why_words.endswith("held back by battery maintenance"))

_end_rows, _end = run(800.0, dur=600.0, soc=rise)
check("⭐ and the charge ENDS: A reaches its 99 % target and the cycle moves "
      "on to its hold, with no refusal on the way",
      _end.maints[0].phase == HOLD and refused(_end_rows) == 0
      and any("charge complete" in e for e in _end.events))

_real = Controller._charge_ceiling
Controller._charge_ceiling = lambda self, claims: None
try:
    _bite, _bc = run(800.0, dur=120.0)
finally:
    Controller._charge_ceiling = _real
print(f"    without the ceiling: refusals {refused(_bite)}/{len(_bite)}, "
      f"last A {cmd(_bite[-1])['a']:+.0f} W")
check("non-vacuous: with the ceiling taken out, the same fixture refuses "
      "every cycle and parks A at 0 W (the reported defect)",
      refused(_bite) == len(_bite) and cmd(_bite[-1])["a"] == 0.0)

# ============================================================================ #
print("\n2. the peak guard outranks the charge")
_q_rows, _q = run(3000.0, dur=2700.0)
print(f"    house 3000: banked {[round(v) for v in banked(_q)]} W, last A "
      f"{cmd(_q_rows[-1])['a']:+.0f} W")
check("⭐ house 3000 W, aim 3450 W: the charge yields just enough - every "
      "banked quarter lands on the aim, not on house + 2000 W",
      len(banked(_q)) >= 2
      and all(AIM - TOL <= v <= AIM + TOL for v in banked(_q)))
check("...and it yields by charging less, not by stopping: A charges and B "
      "is 0 W on every cycle, nothing is refused, no 'Load too high'",
      all(cmd(r)["a"] < 0.0 and cmd(r)["b"] == 0.0 for r in _q_rows)
      and refused(_q_rows) == 0 and not alarmed(_q))
check("...and the peak guard never has to step in (the charge stays under "
      "what the quarter can afford)",
      not any(r[-1][1] and r[-1][1][0] for r in _q_rows))

_o_rows, _o = run(5000.0, dur=2700.0)
_both = [r for r in _o_rows
         if charge_claims(r) and any(c[0] == "peak" for c in r[-1][0])]
print(f"    house 5000: banked {[round(v) for v in banked(_o)]} W, last "
      f"{cmd(_o_rows[-1])}, {_o_rows[-1][6]}")
check("⭐ house 5000 W, over the aim alone: the peak guard wins - every "
      "banked quarter is at or under the aim, both batteries give in the "
      "same direction, nothing is refused and nothing parks at 0 W",
      all(v <= AIM + TOL for v in banked(_o)) and opposite(_o_rows) == 0
      and refused(_o_rows) == 0
      and sum(cmd(r)["a"] + cmd(r)["b"] > 0.0 for r in _o_rows) > 100)
check("...and the guard's `lo` never meets the charge's `hi` in one cycle "
      "(the charge is suspended while the guard holds)", not _both)
check("...with no false 'Load too high' alarm", not alarmed(_o))

_k_rows, _k = run(3000.0, dur=2700.0, month_peak=4000.0)
print(f"    month peak 4000 banked, house 3000: banked "
      f"{[round(v) for v in banked(_k)]} W")
check("⭐ under a month peak banked at 4000 W the aim is 3950 W: the charge "
      "takes the extra headroom and every banked quarter lands on 3950 W",
      len(banked(_k)) >= 2
      and all(3950.0 - TOL <= v <= 3950.0 + TOL for v in banked(_k))
      and refused(_k_rows) == 0 and not alarmed(_k))

_real = Controller._charge_headroom
Controller._charge_headroom = lambda self, floor, claims: floor
try:
    _nh_rows, _nh = run(3000.0, dur=2700.0)
finally:
    Controller._charge_headroom = _real
_gave = sum(1 for r in _nh_rows if cmd(r)["a"] > 0.0 or cmd(r)["b"] > 0.0)
print(f"    without the headroom: cycles a battery gives {_gave}, banked "
      f"{[round(v) for v in banked(_nh)]} W")
check("non-vacuous: charging at the full rate instead, the guard has to "
      "step in and the batteries give back on many cycles",
      _gave > 50)


def peak_run(house):
    rows, c = run(house, dur=1800.0)
    return rows, c, [r for r in rows if r[-1][1] and r[-1][1][0]]


# A 6500 W house needs 3050 W off the meter: more than B's 2500 W alone.
_p_rows, _p, _p_eng = peak_run(6500.0)
print(f"    house 6500: banked {[round(v) for v in banked(_p)]} W, A gives "
      f"on {sum(cmd(r)['a'] > 0.0 for r in _p_eng)} of {len(_p_eng)} "
      f"engaged cycles")
check("⭐ house 6500 W, more than B alone can cover: the charge is suspended "
      "and A, the battery being filled, gives too; the quarter lands on the "
      "aim, nothing is refused, no 'Load too high'",
      len(_p_eng) > 100 and all(cmd(r)["a"] > 0.0 for r in _p_eng)
      and all(v <= AIM + TOL for v in banked(_p))
      and refused(_p_rows) == 0 and not alarmed(_p))
_real = s84.m.peakguard.shaping_suspended
s84.m.peakguard.shaping_suspended = lambda *a: False
try:
    _ns_rows, _ns, _ns_eng = peak_run(6500.0)
finally:
    s84.m.peakguard.shaping_suspended = _real
check("non-vacuous: with the suspension taken out, A is held at 0 W, the "
      "quarter goes over the cap and 'Load too high' fires",
      not any(cmd(r)["a"] > 0.0 for r in _ns_eng)
      and max(banked(_ns)) > 3500.0 and alarmed(_ns))


def prices_only(c, t):
    """A cheap price and no P1 telegram (`meter_source: off`)."""
    if not getattr(c.hass, "_px", False):
        read, px = c.hass.number, {"sensor.pnow": 0.05, "sensor.pavg": 0.20}

        async def number(e, invert=False):
            return px[e] if e in px else await read(e, invert)
        c.hass.number, c.hass._px = number, True


_p1 = s84.s79.P1
s84.s79.P1 = {"meter_source": "off", "peak_cap_w": 3500}
try:
    _off_rows, _off = run(800.0, dur=600.0, before=prices_only)
finally:
    s84.s79.P1 = _p1
check("⭐ with no P1 meter nothing measures the quarter and the charge is not "
      "sized, as at N=1: A charges at 2000 W on every cycle, nothing refused",
      _off.peak_guard is None and all(cmd(r)["a"] == -2000.0 for r in _off_rows)
      and refused(_off_rows) == 0)

# ============================================================================ #
print("\n3. outside a cheap window nothing changes")
_d = {h: run(h, cheap=False, dur=600.0) for h in (800.0, -1500.0)}
check("⭐ no `charge` claim is filed outside a cheap window: B covers the "
      "house, the charge takes solar only, nothing is refused",
      not any(charge_claims(r) for rows, _c in _d.values() for r in rows)
      and cmd(_d[800.0][0][-1]) == {"a": 0.0, "b": 800.0}
      and cmd(_d[-1500.0][0][-1])["a"] < 0.0
      and all(refused(rows) == 0 for rows, _c in _d.values()))


# ============================================================================ #
print("\n4. a safety stop outranks the charge")


def blind_at(t0, t1):
    """A's power reading is missing from `t0` to `t1`: a safety SET."""
    def before(c, t):
        s84.feed(True)(c, t)
        if not getattr(c.hass, "_blind", False):
            real = c.hass.number

            async def number(e, invert=False):
                if e == "sensor.batt_a" and t0 <= c.hass.t < t1:
                    return None
                return await real(e, invert)
            c.hass.number, c.hass._blind = number, True
    return before


_s_rows, _s = run(800.0, dur=600.0, before=blind_at(150.0, 400.0))
_stop = [r for r in _s_rows if 175.0 <= r[0] < 400.0]
_last = _s_rows[79]                     # t = 395 s, still without the reading
print(f"    A's reading lost 150-400 s: {_last[2]}, {_last[6]}; after: "
      f"{_s_rows[-1][2]}")
check("⭐ with a reading gone the safety SET parks both batteries at 0 W "
      "with no refusal and no ARBITER line (it used to refuse every cycle)",
      all(cmd(r) == {"a": 0.0, "b": 0.0} for r in _stop)
      and _last[0] == 395.0 and _last[6].startswith("safety -> 0 W")
      and refused(_s_rows) == 0 and not any("ARBITER" in e for e in _s.events))
check("...and the charge ran before and runs again after",
      cmd(_s_rows[20])["a"] == -2000.0 and cmd(_s_rows[-1])["a"] == -2000.0)


_full = _s._charge_headroom(2000.0, [])
check("the stop test reads only SITE SETs: a unit's own safety SET (on B) "
      "or a site limit does not cut A's charge, a site safety SET does",
      _s._charge_headroom(2000.0, [Claim.set("failsafe", P_SAFETY, 0.0,
                                             actuator=B)]) == _full > 0.0
      and _s._charge_headroom(2000.0, [Claim.limit("peak", P_SAFETY,
                                                   lo=0.0)]) == _full
      and _s._charge_headroom(2000.0, [Claim.set("safety", P_SAFETY,
                                                 0.0)]) == 0.0)


def stop_at(t0):
    def before(c, t):
        s84.feed(True)(c, t)
        if t >= t0:
            c.auto = False
    return before


_st_rows, _st = run(800.0, dur=400.0, before=stop_at(200.0))
check("control switched off mid-charge: both batteries at 0 W, no refusal",
      cmd(_st_rows[-1]) == {"a": 0.0, "b": 0.0} and refused(_st_rows) == 0)

# ============================================================================ #
print("\n5. one battery refusing, dropped, or Off")


def a_takes_nothing(c, t):
    """A reads 0 W whenever it is asked to charge (a battery that refuses)."""
    s84.feed(True)(c, t)
    if not getattr(c.hass, "_rt", False):
        real = c.hass.power
        c.hass.power = lambda u: (0.0 if u == "a" and c.hass.cmd["a"] < 0.0
                                  else real(u))
        c.hass._rt = True


_r_rows, _r = run(800.0, dur=900.0, before=a_takes_nothing)
print(f"    A refuses to charge: {_r_rows[-1][2]}, {_r_rows[-1][6]}")
check("⭐ A refuses to charge (SAFETY-73): no ceiling is filed for it, B "
      "covers the 800 W house, and no cycle is split both ways or refused",
      _r.inverters[0].refused == -1 and cmd(_r_rows[-1]) == {"a": 0.0,
                                                           "b": 800.0}
      and not charge_claims(_r_rows[-1])
      and opposite(_r_rows) == 0 and refused(_r_rows) == 0)

_g_rows, _g = run(3000.0, dur=600.0, cut=lambda u, t: u == "b")
check("B refuses to give: A still charges and nothing is refused",
      all(cmd(r)["a"] < 0.0 for r in _g_rows) and refused(_g_rows) == 0)


def reject_a(c, t):
    """Every write to A is rejected, so A's link goes stale (not live)."""
    s84.feed(True)(c, t)
    if not getattr(c.hass, "_rej", False):
        real = c.hass.set_number

        async def set_number(e, v):
            return False if e == "number.sp_a" else await real(e, v)
        c.hass.set_number, c.hass._rej = set_number, True


_l_rows, _l = run(1500.0, dur=400.0, before=reject_a)
print(f"    A's writes rejected: {_l_rows[-1][2]}, {_l_rows[-1][7][-40:]}")
check("⭐ a charging battery that is dropped (stale link) adds no ceiling: "
      "B covers the 1500 W house instead of charging for A",
      "dropped inverter-a" in _l_rows[-1][7]
      and 1490.0 <= cmd(_l_rows[-1])["b"] <= 1530.0
      and not charge_claims(_l_rows[-1]))


def modes_store(b_mode=None, b_off=False):
    s = rig._Store()
    for k in ("phase", "phase_started"):
        s.data[unit_key(A, k)] = ("charge" if k == "phase"
                                  else rig.BASE.isoformat())
    if b_mode:
        s.data[unit_key(B, packmode.MODE_KEY)] = b_mode
    if b_off:
        s.data[unit_key(B, packmode.USE_KEY)] = False
    return s


_m = {h: run(h, dur=600.0, extra={"battery_modes_enabled": True},
             store=modes_store(b_off=True)) for h in (800.0, -3000.0)}
print(f"    B Off: {[(h, cmd(rows[-1])) for h, (rows, _c) in _m.items()]}")
check("⭐ battery modes on, B Off: A charges at 2000 W over an 800 W house "
      "and takes 2500 W of a 3000 W surplus, B stays at 0 W, nothing is "
      "refused",
      cmd(_m[800.0][0][-1]) == {"a": -2000.0, "b": 0.0}
      and cmd(_m[-3000.0][0][-1]) == {"a": -2500.0, "b": 0.0}
      and all(refused(rows) == 0 and opposite(rows) == 0
              for rows, _c in _m.values()))

# ============================================================================ #
print("\n6. a trade does not outrank the charge")


def sell_feed(c, t):
    p3._feed(c, t)
    s84.feed(True)(c, t)


_trades = []
_t_rows, _t = run(800.0, dur=5400.0, before=sell_feed,
                  extra=dict(p3.ARB),
                  soc=lambda u, t: 60.0)
print(f"    modes off, a sell at 75-90 min: last {_t_rows[-1][2]}")
_sells = [r for r in _t_rows if any(c[0] == "arbitrage" and c[3] > 0.0
                                    for c in r[-1][0])]
check("⭐ battery modes off: the site trade asks to SELL while A charges; "
      "the charge (P_MAINT_SHAPE) outranks it, so A keeps charging and "
      "nothing is refused",
      len(_sells) > 50
      and all(cmd(r)["a"] < 0.0 and cmd(r)["b"] <= 0.0 for r in _sells)
      and refused(_t_rows) == 0)

# ============================================================================ #
print("\n7. a seeded sweep of the ceiling")
_c = run(800.0, dur=30.0)[1]
_rnd = random.Random(84)
_bad = []
for _i in range(20000):
    ra, rb = _rnd.choice((2000.0, 2500.0)), _rnd.choice((1000.0, 2500.0))
    f = _rnd.uniform(1.0, ra)
    cl = [Claim.limit("supervised", P_SAFETY, -ra, ra, actuator=A),
          Claim.limit("supervised", P_SAFETY, -rb, rb, actuator=B),
          Claim.limit("maintenance", P_MAINT_SHAPE, hi=-f, actuator=A)]
    x = _rnd.random()
    if x < 0.1:
        cl.append(Claim.limit("refused", P_SAFETY, 0.0, actuator=A))
    elif x < 0.2:
        cl.append(Claim.limit("pack", P_SAFETY, 0.0, actuator=A))
    y = _rnd.random()
    if y < 0.15:
        cl.append(Claim.limit("refused", P_SAFETY, *_rnd.choice(
            ((0.0, float("inf")), (float("-inf"), 0.0))), actuator=B))
    elif y < 0.3:
        cl.append(Claim.limit("mode", P_MODE, *_rnd.choice(
            ((0.0, 0.0), (0.0, rb), (-rb, 0.0))), actuator=B))
    cl.append(Claim.set("loop", P_LOOP, _rnd.uniform(-6000, 6000)))
    if _rnd.random() < 0.4:
        cl.append(Claim.set("arbitrage", P_ARBITRAGE,
                            _rnd.uniform(-5000, 5000)))
    ceil = _c._charge_ceiling(cl)
    if ceil is not None:
        cl.append(Claim.limit("charge", P_MAINT_SHAPE, hi=ceil))
    res = resolve(cl, units=(A, B), live={A, B})
    sh = res.shares or {}
    why = []
    if not res.allocated or res.contradiction:
        why.append("refused or contradicted")
    if sh.get(A, 0.0) * sh.get(B, 0.0) < 0.0:
        why.append("opposite")
    a_free = x >= 0.2
    if a_free and (ceil is None or sh.get(A, 0.0) > -f + 1.0):
        why.append("A does not charge")
    if not a_free and ceil is not None:
        why.append("ceiling filed for a pack that cannot charge")
    if why:
        _bad.append((_i, why))
print(f"    20000 claim lists, bad {len(_bad)}: {_bad[:3]}")
check("⭐ 20000 random N=2 claim lists (a charge on A, refusals, a full "
      "pack, B's mode, the loop and trades in both directions): never "
      "refused, never both ways, A always charges at least its rate, and "
      "no ceiling for a pack that cannot charge", not _bad)

# ============================================================================ #
print("\n8. the N=1 and N=2 digests")
# Printed by `python3 tools/safety84_identity.py` in goodwe-addon-checks:1,
# 2026-09-30, 480 cycles each, every claim hashed. N=1 and n2_charge_dear:
# the same on release/1.0 9d5b7d1 (the tool copied in). N=2 in a cheap
# window: new, the charge that 9d5b7d1 refused on every cycle.
# SAFETY-85 re-pinned the three N=2 ones: the loop is bounded while A's
# charge pin stands. n2_charge and _banked change only the loop's reason and
# claim on 2 cycles; n2_charge_dear changes commands on 98 cycles, where A's
# pin lifts (t=1800) and B no longer jumps 1390 W, and after the surplus step.
PINNED = {
    "n1_charge":
        "0ed25c856456b97acb07a39299a8cb7b1a762c4d0005b88a0823c03057a80be5",
    "n1_charge_dear":
        "c25340b56060d4581f7204e200b07e7c8618711be2baca3b1667b7ebeb600dc3",
    "n1_charge_banked":
        "2619acdd06b04e56c000b168d64e1d044346c6ba3f929c94258d6ad8189ec609",
    "n2_charge":
        "b16d692128f14bcce3d87b142d290e0d6b13724289680e44a05215c9eaa24a7c",
    "n2_charge_dear":
        "00a197d89a5dd1e8b3ae6a5d1af66329a4dc4850476569c5931c125fa4d9eaee",
    "n2_charge_banked":
        "a6973faea916e4c88ca1770047f94478346eeb1efa4889b154fad0747f91267f",
}
_now = s84.digests()
for k in PINNED:
    print(f"    {k} {'same' if _now[k] == PINNED[k] else 'DIFFERS'} "
          f"{_now[k][:16]}")
check("⭐ at N=1 a charge in and out of a cheap window and under a banked "
      "month peak is byte-identical to 9d5b7d1 (commands, reasons and every "
      "claim), and N=2 outside a cheap window matches its SAFETY-85 pin",
      all(_now[k] == PINNED[k] for k in ("n1_charge", "n1_charge_dear",
                                         "n1_charge_banked",
                                         "n2_charge_dear")))
check("...and the N=2 cheap-window digests match the pins",
      _now["n2_charge"] == PINNED["n2_charge"]
      and _now["n2_charge_banked"] == PINNED["n2_charge_banked"])
_n1 = s84.run(1, dur=600.0)[0]
check("...and no N=1 cycle files a `charge` claim",
      not any(charge_claims(r) for r in _n1))

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