goodwe-addon/goodwe_controller/test_peak_margin.py

"""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.
"""

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, 2026-09-28. 540 cycles each: every command, the loop and arbiter
# sentences, the allocation, and the guard's floor, limit and sentence.
ED739D4 = {
    "n1": "36f2dc6cd82ed63b93125dc3f8ea5f475e494255c614a4d2f353dad06e44f884",
    "n2": "1d388daf1bcfc3e16b1ef696377ae2a01831963374e2a6a4d2f998b5c6d997c2",
    "n1_off": "76e41db45667fffebd3ce99a49873a31d0b86fcc0c520057bbf38cc4f27138ca",
    "n2_off": "196653957e2fdf227daa42447ac170c45292e6c3f4184196dd99c1140359e415",
}
_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]}")
check("⭐ at N=1 and N=2, with and without battery modes, margin 0 is "
      "byte-identical to ed739d4", _zero == ED739D4)
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)

_obs = type("R", (), {"is_observed": True, "peak_w": 2480.0})()
check("test_month_peak_not_lowered: a month peak already paid for is aimed "
      "at as it stands (2480 W with cap 2500, margin 50), and a higher one "
      "too; no peak aims at cap - margin",
      quarter_target_w(CAP, 50, _obs) == 2480.0
      and quarter_target_w(CAP, 50,
                           type("R", (), {"is_observed": True,
                                          "peak_w": 3000.0})()) == 3000.0
      and quarter_target_w(CAP, 50, None) == CAP - 50)

# ============================================================================ #
print("\n3. the bill and the reports keep the real cap")
_c = _runs["n1_off", None][1]
_pk = _c.peak.peak().peak_w
print(f"    month peak {_pk:.0f} W, tracker cap {_c.peak.cap_w:.0f} W, "
      f"warning {_c.peak.warning_frac():.3f}, headroom "
      f"{_c.peak.headroom_w():.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() == _pk / CAP
      and _c.peak.headroom_w() == CAP - _pk)
_eng = [g for r in _runs["n1_off", None][0] for g in (r[-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")


def _cfg_pair(key):
    """(default, schema) lines for `key` at two spaces in config.yaml."""
    with open(os.path.join(HERE, "config.yaml"), encoding="utf-8") as f:
        hits = re.findall(rf"^  {key}:\s*(\S+)", f.read(), re.M)
    return hits


_hits = _cfg_pair("peak_margin_w")
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")