"""Runnable check for SAFETY-79, the margin under the peak limit.
`python3 test_peak_margin.py`

The peak guard aims at `peak_cap_w - peak_margin_w` (default 50 W); the bill
and every report keep the real `peak_cap_w`. DECISIONS/SAFETY-79-peak-margin.md
has the record. Drives the REAL Controller through tools/safety79_identity.py.

1. `peak_margin_w: 0` is byte-identical to release/1.0 ed739d4.
2. The guard aims at cap - margin at N=1 and N=2, and with OPT-11 modes on an
   Off battery covers the shortfall to cap - margin.
3. The bill and the reports still use the real cap.
4. Each call site in main.py passes the value its side needs.
5. The setting: default, range, refusals in plain words.

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

import ast
import os
import re
import sys

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

import safety79_identity as rig                               # noqa: E402
from named_absence import nth, present                        # noqa: E402
from style_rules import text_problems                         # noqa: E402
from app import messages, peakguard                 # noqa: E402
from app.control import maintenance_charge_floor              # noqa: E402
from app.peakguard import (PEAK_MARGIN_RANGE_W,               # noqa: E402
                           PEAK_OPTION_DEFAULTS, quarter_target_w)
from app.validate import config_notes, validate_startup_config  # noqa: E402

fails = []


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


CAP = float(rig.CAP_W)

# ============================================================================ #
print("1. peak_margin_w 0: byte-identical to release/1.0 ed739d4")
# Printed by `python3 tools/safety79_identity.py 0` on the ed739d4 tree (the
# tool copied in, nothing else changed; that tree ignores the key), in the CI
# image; re-printed 2026-09-29 (review) after adding the charge headroom, the
# shaping suspension and the two maintenance scenarios. 540 cycles each.
ED739D4 = {
    "n1": "82316f7d4362fae4c035a1bed5e7907cf362faba994c634d0b1c356c501f53e5",
    "n2": "0dc24e16a8390ec42edacf735eaf20273847b33b9c67d7cdc536df0ca739e421",
    "n1_off": "eed44f058f3fcc6e7651c5c84c6fa6313e7520035ad7cac282012cf6909668a0",
    "n2_off": "69796c107b9f78892e04d282b754f889590acf9fe1b96f09525f5ea2005e9c0a",
    "n1_maint":
        "bd4a88a10aa840a360ac781ac04e30b040a5aac5426fabba1fb259d71dfeba67",
    "n2_maint":
        "0656bc5a4a2fa3a7455822fdd7da5f68f57fd9593f374e869a0d30ba7e429275",
}
_zero = rig.digests(0)
_default = rig.digests()
for k in ED739D4:
    print(f"    {k:7} margin 0 {'same' if _zero[k] == ED739D4[k] else 'DIFFERS'}"
          f" {_zero[k][:16]}, default {_default[k][:16]}")
# SAFETY-84 changes n2_maint ON PURPOSE: on 9d5b7d1 its N=2 cheap-window
# charge was refused on 214 of 540 cycles (site at 0 W); now none is.
# Printed on the SAFETY-84 tree the same way (margin 0).
# SAFETY-85: the loop is bounded while A charges (reasons change, commands
# do not). Was 95722b262f1bcf504493d7b5babc3895799ac56cc15f5d2ccdbe7c9f04adfcbb.
SAFETY84_N2_MAINT = \
    "ef97ae2e13668b3e0b2d22f3f5f37835ea46cd306c2cc475c97e284f9dcc6545"
check("⭐ at N=1 and N=2, with and without battery modes and with a "
      "maintenance charge, margin 0 is "
      "byte-identical to ed739d4 (n2_maint apart, below)",
      {k: v for k, v in _zero.items() if k != "n2_maint"}
      == {k: v for k, v in ED739D4.items() if k != "n2_maint"})
check("...and n2_maint at margin 0 is the SAFETY-84 digest: the N=2 charge "
      "that 9d5b7d1 refused on 214 of 540 cycles now runs (test_safety84.py)",
      _zero["n2_maint"] == SAFETY84_N2_MAINT != ED739D4["n2_maint"])
check("...and the default (key absent) changes every scenario, so the "
      "scenarios do reach the margin", all(_default[k] != ED739D4[k]
                                          for k in ED739D4))

# ============================================================================ #
print("\n2. the guard aims at cap - margin")
_runs = {}
for name, spec in rig.SCENARIOS.items():
    for m in (None, 0, 120):
        _runs[name, m] = rig.run(**spec, margin=m, dur=3600.0)
for name in rig.SCENARIOS:
    g = _runs[name, None][1].peak_guard
    print(f"    {name}: margin {_runs[name, None][1].peak_margin_w:.0f} W, "
          f"aims at {g.target_w:.0f} W, limit {g.ceiling_w:.0f} W")
check("test_target_is_cap_minus_margin: with the key absent every scenario "
      "(N=1, N=2, modes on and off) aims at cap - 50 and reports the cap",
      all(_runs[n, None][1].peak_margin_w == 50.0
          and _runs[n, None][1].peak_guard.target_w == CAP - 50
          and _runs[n, None][1].peak_guard.ceiling_w == CAP
          for n in rig.SCENARIOS))
check("...and a margin of 120 moves the aim to cap - 120, a margin of 0 to "
      "the cap", all(_runs[n, 120][1].peak_guard.target_w == CAP - 120
                     and _runs[n, 0][1].peak_guard.target_w == CAP
                     for n in rig.SCENARIOS))


def banked(ctl):
    return {b.start.strftime("%H:%M"): round(b.offtake_avg_w, 2)
            for b in ctl.p1.blocks}


_b = {m: banked(_runs["n1_off", m][1]) for m in (0, None, 120)}
print(f"    N=1, the battery Off: quarters banked {_b}")
check("test_off_battery_n1: at N=1 with the battery Off, the guard makes it "
      "give just enough to bank the 11:30 quarter at cap - margin (2500, "
      "2450, 2380 W for margins 0, 50, 120)",
      [_b[m]["11:30"] for m in (0, None, 120)]
      == [CAP, CAP - 50, CAP - 120])


def meter_at(rows, t):
    row = next((r for r in rows if r[0] == t), None)
    if row is None:
        return None, {}
    return rig.load(t) - sum(w for _u, w in row[2]), dict(row[2])


_g0, _c0 = meter_at(_runs["n2_off", 0][0], 1800.0)
_g50, _c50 = meter_at(_runs["n2_off", None][0], 1800.0)
print(f"    N=2, B Off, 6000 W house: margin 0 -> {_c0}, meter {_g0} W; "
      f"margin 50 -> {_c50}, meter {_g50} W")
check("test_off_battery_n2: at N=2 with B Off and A at its rating, B covers "
      "the shortfall to cap - margin, not to the cap",
      _c0["b"] == 1000.0 and _g0 == CAP
      and _c50["b"] == 1050.0 and _g50 == CAP - 50)

def _peak(w, observed=True):
    return type("R", (), {"is_observed": observed, "peak_w": w})()


# Review (SAFETY-79 point 2): the margin applies under a month peak already
# banked, too. Aiming AT it can bank a few W over it, which raises the bill;
# and a banked peak between cap - margin and the cap (a quarter the old guard
# held at the cap, say) was aimed at exactly, so one overshoot crossed the cap.
check("test_margin_under_banked_peak: the aim is the margin under the limit "
      "that holds - the cap, or a month peak already banked above it",
      quarter_target_w(CAP, 50, _peak(2499.0)) == CAP - 50
      and quarter_target_w(CAP, 50, _peak(3000.0)) == 2950.0
      and quarter_target_w(CAP, 50, _peak(3000.0, observed=False)) == CAP - 50
      and quarter_target_w(CAP, 50, None) == CAP - 50
      and quarter_target_w(CAP, 0, _peak(3000.0)) == 3000.0)


def _decide(banked_w, projected_w):
    """Half a quarter at `projected_w`, battery idle, a month peak banked."""
    from datetime import datetime, timezone
    from app.tariff import Projection
    pr = Projection(projected_w, projected_w, 450.0, 1.0, False,
                    datetime(2026, 9, 1, 12, 0, tzinfo=timezone.utc),
                    projected_w)
    return peakguard.decide(pr, batt_w=0.0, cap_w=CAP, margin_w=50,
                            reading=_peak(banked_w), achievable_w=5000.0,
                            soc=60.0, soc_floor=20.0)


_d1, _d2 = _decide(2499.0, 2600.0), _decide(3000.0, 3100.0)
print(f"    banked 2499 W: floor {_d1.floor_w}, limit {_d1.ceiling_w}; "
      f"banked 3000 W: floor {_d2.floor_w}, limit {_d2.ceiling_w}")
check("...and decide() sizes the floor to land the quarter 50 W under it "
      "(2449 and 2950 W), while the reported limit stays the cap or the peak",
      _d1.floor_w == 2 * (2600.0 - 2450.0) and _d1.ceiling_w == CAP
      and _d2.floor_w == 2 * (3100.0 - 2950.0) and _d2.ceiling_w == 3000.0)

# ============================================================================ #
print("\n3. the bill and the reports keep the real cap")
_c = _runs["n1_off", None][1]
_t = rig.rig.BASE       # the runs' clock, not the wall clock (a new month)
_pk = _c.peak.peak(_t).peak_w
print(f"    month peak {_pk:.0f} W, tracker cap {_c.peak.cap_w:.0f} W, "
      f"warning {_c.peak.warning_frac(_t):.3f}, headroom "
      f"{_c.peak.headroom_w(_t):.0f} W")
check("test_billing_uses_cap: the month-peak tracker (bill, headroom, 80 % "
      "warning) is built on the real cap, not cap - margin",
      _c.peak.cap_w == CAP and _c.peak.warning_frac(_t) == _pk / CAP
      and _c.peak.headroom_w(_t) == CAP - _pk)
_eng = [g for r in _runs["n1_off", None][0] for g in (r[-1][1],)
        if g and g[0] and g[1]]
print(f"    {len(_eng)} engaged cycles; one: {_eng[0][4] if _eng else None}")
check("test_reports_use_cap: every engaged sentence names the real limit and "
      "the aim beside it", _eng and all(
          g[2] == CAP and "against a 2500 W limit (peak_cap_w), aiming at "
          "2450 W" in g[4] for g in _eng))
check("...and the settings note compares the month peak with the real cap: "
      "a 2480 W peak under a 2500 W cap is no warning, 2520 W is",
      config_notes({"peak_cap_w": 2500, "peak_margin_w": 50},
                   month_peak_w=2480) == []
      and [n.keys for n in config_notes({"peak_cap_w": 2500,
                                         "peak_margin_w": 50},
                                        month_peak_w=2520)]
      == [("peak_cap_w",)])
_lim = _eng[0][2] if _eng else 0.0
_notice = messages.peak_saturated(_lim)
check("...and the user-facing peak notice built from the decision's limit "
      "says the cap", messages.watts(CAP) in _notice.body
      and messages.watts(CAP - 50) not in _notice.body)

# ============================================================================ #
print("\n4. each call site in main.py")
with open(os.path.join(HERE, "app", "main.py"), encoding="utf-8") as f:
    _src = f.read()
_tree = ast.parse(_src)


def calls(name):
    return [n for n in ast.walk(_tree) if isinstance(n, ast.Call)
            and isinstance(n.func, (ast.Attribute, ast.Name))
            and getattr(n.func, "attr", getattr(n.func, "id", "")) == name]


def kw(call, key):
    if not present(call):
        return "absent"
    return next((ast.get_source_segment(_src, k.value) for k in call.keywords
                 if k.arg == key), "absent")


def arg(call, i):
    if not present(call) or len(call.args) <= i:
        return "absent"
    return ast.get_source_segment(_src, call.args[i])


_dec = [c for c in calls("decide")
        if getattr(c.func, "value", None) is not None
        and getattr(c.func.value, "id", "") == "peakguard"]
_shape = calls("shaping_suspended")
_maint = calls("maintenance_charge_floor")
_tracker = calls("PeakTracker")
_d, _s, _m, _t = (nth(_dec, 0, "peakguard.decide()"),
                  nth(_shape, 0, "shaping_suspended()"),
                  nth(_maint, 0, "maintenance_charge_floor()"),
                  nth(_tracker, 0, "PeakTracker()"))
print(f"    decide(cap_w={kw(_d, 'cap_w')}, margin_w={kw(_d, 'margin_w')})")
print(f"    shaping_suspended(..., {arg(_s, 2)})")
print(f"    maintenance_charge_floor(..., {arg(_m, 2)})")
print(f"    PeakTracker(cap_w={kw(_t, 'cap_w')})")
check("test_call_sites: the guard gets the real cap AND the margin; the "
      "forecast checks get cap - margin; the month-peak tracker gets the "
      "real cap",
      len(_dec) == 1 and "peak_cap_w" in kw(_d, "cap_w")
      and kw(_d, "margin_w") == "self.peak_margin_w"
      and len(_shape) == 1 and len(_maint) == 1
      and arg(_s, 2) == "self.peak_aim_w()"
      and arg(_m, 2) == "self.peak_aim_w()"
      and len(_tracker) == 1
      and kw(_t, "cap_w") == 'float(opts.get("peak_cap_w", 3500))')
check("test_aim_w: `peak_aim_w()` is cap - margin, and the forecast sizing "
      "gives a maintenance charge only what is left under it",
      _c.peak_aim_w() == CAP - 50
      and maintenance_charge_floor(2000.0, 2400.0, _c.peak_aim_w()) == 50.0
      and peakguard.shaping_suspended(None, 2470.0, _c.peak_aim_w())
      and not peakguard.shaping_suspended(None, 2470.0, CAP))

# ============================================================================ #
print("\n5. the setting")


with open(os.path.join(HERE, "config.yaml"), encoding="utf-8") as f:
    _hits = re.findall(r"^  peak_margin_w:\s*(\S+)", f.read(), re.M)
print(f"    config.yaml: {_hits}")
check("config.yaml ships 50, int(0,500), and the in-code mirror matches",
      _hits == ["50", "int(0,500)"]
      and PEAK_OPTION_DEFAULTS["peak_margin_w"] == 50
      and PEAK_MARGIN_RANGE_W == (0, 500))

BASE = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
        "max_w": 2000, "peak_cap_w": 3500}
_base_problems = validate_startup_config(BASE)


def refused(value, **more):
    opts = dict(BASE, peak_margin_w=value, **more)
    return [p for p in validate_startup_config(opts)
            if p not in _base_problems]


_ok = {v: refused(v) for v in (0, 50, 500)}
_bad = {v: refused(v) for v in (-1, 501, 12.5, "abc")}
for v, p in _bad.items():
    print(f"    {v!r}: {p}")
check("test_margin_range: 0, 50 and 500 are accepted, and so is the key "
      "absent", not any(_ok.values()) and _base_problems == [])
check("...and below 0, above 500, not whole, or not a number is refused, "
      "once, naming the setting and the range",
      all(len(p) == 1 and "“Margin below the peak limit” (peak_margin_w)"
          in p[0] and "from 0 to 500 W" in p[0] for p in _bad.values()))
_high = refused(500, peak_cap_w=500)
print(f"    500 under a 500 W limit: {_high}")
check("...and a margin that leaves nothing under the limit is refused",
      len(_high) == 1 and "not below" in _high[0])
_style = [text_problems(p[0], keys=("peak_margin_w", "peak_cap_w"),
                        entities=False) for p in list(_bad.values()) + [_high]]
check(f"...in plain words (STYLE.md): {_style}", not any(_style))


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