"""Runnable check for SAFETY-83: the peak guard outranks the drain at N=1.
`python3 test_safety83.py`

At one inverter the monthly drain is the site SET at P_MAINTENANCE, above the
peak floor, so the house imported what the drain left and quarters went over
the peak limit. Now a peak floor above the drain rate is refiled at
P_MAINTENANCE and wins. Drives the REAL Controller (tools/opt11_identity.py's
harness on a P1 meter). DECISIONS/SAFETY-83-peak-over-drain.md has the record.

1. A house of 5000, 6000 and 7000 W during a drain, limits 2500 and 3500,
   margins 0 and 50: every quarter banks at or under the aim, the drain keeps
   running, no 'Load too high' alarm, and without the fix the same runs go over.
   Under a month peak banked above the limit the aim is that peak - margin.
2. The drain never makes the banked quarter worse than no drain, below and
   above the drain rate; below it the rest is still exported.
3. The drain still ends, and the order holds: a lost reading parks the
   battery, the rating caps the floor without a clash.
4. Only cycles where the guard wanted more than the drain change, and N=2 is
   untouched.

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

import functools
import logging
import os
import sys
import tempfile
import types

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 safety78_identity as s78                                # noqa: E402
import safety79_identity as s79                                # noqa: E402
from app import tariff                                         # noqa: E402
from app.arbiter import (Claim, P_MAINTENANCE, P_PEAK, P_SAFETY,  # noqa
                         resolve)
from app.main import Controller                                # noqa: E402
from app.maintenance import CHARGE, DRAIN, MaintResult         # noqa: E402

fails = []
TOL = 1.0       # watts over the aim a banked quarter may show


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


def run(house, cap, margin, drain_w, *, maint=True, dur=2700.0, soc=None,
        before=None, store=None, month_peak=None, **extra):
    """N=1 on a P1 meter; returns (rows, controller, banked quarters).
    `month_peak` banks that quarter earlier in the month."""
    opts = dict(s79.P1, peak_cap_w=cap, peak_margin_w=margin, **extra)
    if maint:
        opts.update(maintenance_enabled=True, maintenance_start_hour=11,
                    maintenance_discharge_w=drain_w)
    db = os.path.join(tempfile.mkdtemp(prefix="s83-"), "tariff.db")
    if month_peak is not None:
        tariff.PeakStore(db)._db.execute(
            "INSERT INTO monthly_peak VALUES (?, ?, ?, NULL)",
            (tariff.month_key(rig.BASE), month_peak,
             "2026-09-02T10:00:00+00:00")).connection.commit()
    real = rig.Controller
    rig.Controller = functools.partial(real, peak_db_path=db)
    try:
        rows, c = rig.run(1, extra=opts, dur=dur,
                          load=house if callable(house) else lambda t: house,
                          soc=soc or (lambda u, t: 60.0),
                          cut=lambda u, t: False, before=before or s79._feed,
                          store=store)
    finally:
        rig.Controller = real
    return rows, c, {b.start.strftime("%H:%M"): b.offtake_avg_w
                     for b in c.p1.blocks}


def without_fix(fn):
    real = Controller._peak_over_drain
    Controller._peak_over_drain = lambda self, claims, results: None
    try:
        return fn()
    finally:
        Controller._peak_over_drain = real


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


def alarmed(c):
    """Did SAFETY-81's 'Load too high' alarm fire on this run?"""
    return any("PEAK_GUARD_SATURATED" in e for e in c.events)


# ============================================================================ #
print("1. banked quarters during a drain, N=1")
CASES = [(cap, drain, m, h) for cap, drain in ((2500, 2000), (3500, 2500))
         for m in (0, 50) for h in (5000.0, 6000.0, 7000.0)]
RES, OLD = {}, {}
for k in CASES:
    cap, drain, m, h = k
    RES[k] = run(h, cap, m, drain)
    OLD[k] = without_fix(lambda: run(h, cap, m, drain))
    print(f"    limit {cap} drain {drain} margin {m:>2} house {h:.0f}: "
          f"banked {[round(v) for v in RES[k][2].values()]} W "
          f"(before the fix {[round(v) for v in OLD[k][2].values()]} W)")
check("⭐ every quarter banks at or under the aim (limit - margin) plus 1 W, "
      "for a 5000, 6000 and 7000 W house, limits 2500 and 3500, margins 0 "
      "and 50",
      all(len(b) >= 2 and max(b.values()) <= k[0] - k[2] + TOL
          for k, (_r, _c, b) in RES.items()))
check("⭐ the drain keeps running: the phase is drain on every run and the "
      "battery gives at least the drain rate on every cycle",
      all(c.maints[0].phase == DRAIN and all(cmd(r) >= k[1] for r in rows)
          for k, (rows, c, _b) in RES.items()))
check("...and where the house is over drain + aim, the battery gives more "
      "than the drain rate, bound by the peak floor",
      all(any(cmd(r) > k[1] and "limited by peak" in r[6] for r in rows)
          for k, (rows, _c, _b) in RES.items() if k[3] - k[1] > k[0] - k[2]))
check("non-vacuous: without the fix the same runs bank over the limit "
      "wherever the house is above drain + aim (the reported defect: 3000 W "
      "at a 5000 W house against 2500 W)",
      all(max(b.values()) > k[0] + 100.0 for k, (_r, _c, b) in OLD.items()
          if k[3] - k[1] > k[0])
      and round(OLD[(2500, 2000, 50, 5000.0)][2]["11:15"]) == 3000)
# A hob cycling: one 5 s spike to 11000 W every 150 s on a 6000 W house. Each
# spike saturates the guard for one cycle, shorter than ALARM_HOLD_S.
_sat = []


def _spiky_feed(c, t):
    s79._feed(c, t)
    _sat.append(bool(c.peak_guard is not None and c.peak_guard.saturated))


_k_rows, _k, _kb = run(lambda t: 11000.0 if t % 150 < 5 else 6000.0,
                       3500, 50, 2500, before=_spiky_feed)
check("⭐ a drain the guard raises gives no 'Load too high' alarm (SAFETY-81), "
      "steady or with 5 s spikes that saturate the guard",
      not any(alarmed(c) for _r, c, _b in RES.values())
      and any(_sat) and not alarmed(_k))

# Owner (SAFETY-79): under a month peak banked above the limit, the aim is that
# peak - margin, during a drain too. Owner's settings, peak banked at 4500 W.
BANK = {h: (run(h, 3500, 50, 2500, month_peak=4500.0),
            without_fix(lambda: run(h, 3500, 50, 2500, month_peak=4500.0)))
        for h in (6500.0, 7500.0)}
for h, (new, old) in BANK.items():
    print(f"    month peak 4500, house {h:.0f}: banked "
          f"{[round(v) for v in new[2].values()]} W (before the fix "
          f"{[round(v) for v in old[2].values()]} W)")
check("⭐ under a month peak of 4500 W: a 7500 W house banks at 4450 W "
      "(peak - margin, not the 3450 W cap aim), was 5000 W; a 6500 W house "
      "stays under the drain alone at 4000 W",
      all(4430.0 <= v <= 4450.0 + TOL for v in BANK[7500.0][0][2].values())
      and max(BANK[7500.0][1][2].values()) > 4900.0
      and all(round(v) in (4000, 4014)
              for v in BANK[6500.0][0][2].values()))

# ============================================================================ #
print("\n2. the drain never makes the banked quarter worse, N=1")
CAP, DRAIN_W, MARGIN = 3500, 2500, 50          # the owner's settings
AIM = CAP - MARGIN
WORSE = {}
for h in (1200.0, 3000.0, 5000.0, 6500.0, 7000.0):
    _rows, _c, b = run(h, CAP, MARGIN, DRAIN_W)
    _r0, _c0, b0 = run(h, CAP, MARGIN, DRAIN_W, maint=False)
    WORSE[h] = (b, b0, _rows)
    print(f"    house {h:.0f}: with the drain {[round(v) for v in b.values()]}"
          f" W, without {[round(v) for v in b0.values()]} W")
check("⭐ below and above the drain rate, every quarter with the drain banks "
      "no more than the larger of the aim and the same quarter without it",
      all(b[q] <= max(AIM, b0[q]) + TOL for b, b0, _r in WORSE.values()
          for q in b))
_r12 = WORSE[1200.0][2]
check("below the drain rate the battery gives the drain rate and exports the "
      "rest (owner, 2026-09-29): 2500 W against a 1200 W house",
      all(cmd(r) == DRAIN_W for r in _r12[1:]))

# ============================================================================ #
print("\n3. the drain ends, and the order holds")
_e_rows, _e, _eb = run(6500.0, CAP, MARGIN, DRAIN_W, dur=1200.0,
                       soc=lambda u, t: max(5.0, 60.0 - t / 20.0))
check("⭐ with the guard binding, the drain still reaches its floor and moves "
      "to the charge phase, with no abort",
      _e.maints[0].phase == CHARGE
      and not any("abort" in e.lower() for e in _e.events))


def blind_at(t0):
    """The battery power reading goes missing at `t0`: a safety SET."""
    def before(c, t):
        s79._feed(c, t)
        if t >= t0 and not getattr(c.hass, "_blind", False):
            real = c.hass.number

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


_s_rows, _s, _sb = run(6500.0, CAP, MARGIN, DRAIN_W, dur=600.0,
                       before=blind_at(300.0))
print(f"    reading lost at 300 s: last {cmd(_s_rows[-1]):+.0f} W, "
      f"{_s_rows[-1][6]}")
check("⭐ a safety SET outranks the refiled peak floor: before the reading "
      "goes the battery gives above the drain rate, after it it is at 0 W",
      any(cmd(r) > DRAIN_W for r in _s_rows if r[0] < 300.0)
      and all(cmd(r) == 0.0 for r in _s_rows if r[0] >= 400.0)
      and _s_rows[-1][6].startswith("safety -> 0 W"))
_m_rows, _m, _mb = run(7000.0, 2500, 50, 2000, dur=900.0, max_w=3000)
print(f"    max_w 3000, house 7000: last {cmd(_m_rows[-1]):+.0f} W, "
      f"{_m_rows[-1][6]}")
check("⭐ the rating caps the floor without a clash: at max_w 3000 the "
      "battery gives 3000 W, never 0 W, and nothing contradicts",
      all(cmd(r) >= 2000.0 for r in _m_rows)
      and cmd(_m_rows[-1]) == 3000.0
      and not any("CONTRADICTION" in (r[6] or "") for r in _m_rows))
check("...and there, with the battery at its rating and the house still over, "
      "the 'Load too high' alarm does fire", alarmed(_m))

# ============================================================================ #
print("\n4. only the cycles where the guard wanted more change")


def promoted(row):
    return [c for c in row[-1][0] if c[0] == "peak"]


_id = {k: s78.run(**v)[0] for k, v in s78.SCENARIOS.items()}
_bad = []
for k, rows in _id.items():
    for r in rows:
        claims, guard = r[-1][0], r[-1][1]
        rate = [c[3] for c in claims if c[0] == "maintenance" and c[2] == "set"]
        pk = promoted(r)
        if k.startswith("n2"):
            ok = all(c[1] == P_PEAK for c in pk)
        elif s78.wanted_more(r):
            ok = ([c[1] for c in pk] == [P_MAINTENANCE]
                  and r[1] == guard[1] and "limited by peak" in r[6])
        else:
            ok = (all(c[1] == P_PEAK for c in pk)
                  and (not rate or r[1] == rate[0]))
        if not ok:
            _bad.append((k, r[0], r[1], r[6], pk))
_bad1 = [b for b in _bad if b[0].startswith("n1")]
_bad2 = [b for b in _bad if b[0].startswith("n2")]
_more = {k: sum(map(s78.wanted_more, rows)) for k, rows in _id.items()}
print(f"    cycles with the guard above the drain: {_more}; bad {_bad[:3]}")
check("⭐ at N=1, on every cycle where the guard asks for more than the drain "
      "the peak claim is refiled at P_MAINTENANCE and the target is its floor; "
      "on every other cycle the peak claim stays at P_PEAK and the target is "
      "the drain rate",
      not _bad1 and all(_more[k] > 50 for k in _id if k.startswith("n1")))
check("⭐ at N=2 the peak claim is never refiled (the SAFETY-78 floor does "
      "that job), and test_safety78.py part 5 holds N=2 to the digests of "
      "release/1.0 35cdffe", not _bad2)


# The rule itself, at the boundary: a floor AT the drain rate changes nothing.
_one = types.SimpleNamespace(inverters=[None])
_edge = {}
for lo in (1999.5, 2000.0, 2000.5):
    cl = [Claim.set("maintenance", P_MAINTENANCE, 2000.0, DRAIN),
          Claim.limit("peak", P_PEAK, lo=lo, reason="quarter peak")]
    Controller._peak_over_drain(_one, cl, [MaintResult(DRAIN, True, 2000.0,
                                                       False, [])])
    _edge[lo] = cl[1].priority
check("at the boundary: a floor at or below the 2000 W drain stays at P_PEAK, "
      "a floor 0.5 W above it is refiled at P_MAINTENANCE",
      _edge == {1999.5: P_PEAK, 2000.0: P_PEAK, 2000.5: P_MAINTENANCE})

_stop = [Claim.set("maintenance", P_MAINTENANCE, 2000.0, DRAIN),
         Claim.limit("peak", P_PEAK, lo=4000.0, reason="quarter peak"),
         Claim.set("safety", P_SAFETY, 0.0, "inputs missing")]
Controller._peak_over_drain(_one, _stop, [MaintResult(DRAIN, True, 2000.0,
                                                      False, [])])
check("⭐ and a refiled floor stays under a safety SET: the site resolves "
      "to 0 W", _stop[1].priority == P_MAINTENANCE
      and resolve(_stop).target_w == 0.0)


def store_hold():
    st = rig._Store()
    st.data["phase"], st.data["phase_started"] = "hold", rig.BASE.isoformat()
    return st


_h_rows, _h, _hb = run(6000.0, 2500, 50, 2000, dur=600.0, store=store_hold())
check("a maintenance HOLD at N=1 now gives way to the guard too (SAFETY-86, "
      "test_safety86.py): the battery gives with the house over the limit",
      _h.maints[0].phase == "hold" and _h.peak_guard.engaged
      and cmd(_h_rows[-1]) > 0.0 and "limited by peak" in _h_rows[-1][6])

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