goodwe-addon/goodwe_controller/test_tariff.py

"""Runnable check for TARIFF-01, the capacity-tariff measurement.

    python3 test_tariff.py

Ten parts. What each is for, and - where it matters - what it is deliberately
NOT allowed to be tested at:

  1. THE BLOCK ARITHMETIC, against a worked example written out segment by
     segment. The fixture is chosen to be non-degenerate on three axes at once:
     irregular sample spacing, a sample that STRADDLES the :15 boundary, and a
     gap longer than `max_hold_s` so the block's coverage is short of 900 s.
  2. THE 900 s DEGENERACY, named and then avoided. `acc/elapsed`, `acc/900` and
     the end-of-quarter projection are THREE DIFFERENT FORMULAS that all agree
     at exactly full coverage of a complete block. A check run only there proves
     nothing about any of them, so part 2 first shows them agreeing at that
     point (so the trap is documented rather than asserted around), then shows
     part 1's fixture separating all three - and every later part uses fixtures
     that are away from it.
  3. RESTART MID-BLOCK, through the real `QuarterAverager.snapshot()`/`restore()`
     and a real SQLite file: a restarted run must land on the same closed block
     as an uninterrupted one, and a run that DOES NOT restore must land lower.
     Without that second half, "restore works" is satisfied by a fixture where
     restoring changes nothing.
  4. MONTH ROLLOVER, in LOCAL time, including a restart that straddles it.
  5. AN UNREADABLE STORE IS NOT A ZERO PEAK - the acceptance argument. Three
     states, three different answers, and no accessor that turns "the database
     did not answer" into a number.
  6. THE TIMEZONE FALLBACK, graded against `zoneinfo` day by day.
  7. THE PER-PHASE REDUCTION on an UNBALANCED load. §17's three-phase case. On a
     balanced load net/sum/max all coincide, which is why a balanced fixture is
     worth nothing here.
  8. `monthly_peak` vs a recompute from `quarter_blocks` - the derived record
     proven equal to the tier it derives from, and proven never to be lowered.
  9. AC 3, ON THE REPLAY DATA THAT ACTUALLY EXISTS. See the honesty note below.
 10. THE WIRING, through the real `Controller` and its real status page.

------------------------------------------------------------------------------
⚠️ AC 3 IS NOT MET, AND PART 9 DOES NOT PRETEND OTHERWISE.

AC 3 asks for "the monthly peak held under the soft limit in >= 95 % of
intervals over a 30-day simulation replay". THERE IS NO 30-DAY DATASET. The
infra repo's `sim/scenarios/` holds exactly two days of this house's real meter
- 2026-08-20 (the Illusmart failure day) and 2026-08-23 (the healthy reference)
- exported on 2026-08-25 because they were about to age out of HA's recorder,
and nothing longer was ever captured. `sim/scenarios/README.md` lists the whole
inventory; the rest is RS485 bus frames, OTA witness logs and a 578-row rollback
poll, none of which carries grid power over calendar time.

So part 9 replays the two days that exist, reports the percentage over them, and
states the sample size in days next to it. It is a MEASUREMENT, not a verdict:
nothing in it exits non-zero on the 95 % figure, because a 2-day run cannot pass
or fail a 30-day criterion and a check that claimed it could would be the more
expensive of the two mistakes. Both days are also *uncontrolled* - the add-on
was not shaving peaks on either - so the figure they give is the BASELINE the
guard has to improve on, which is the useful thing a 2-day replay can say.

The two days are committed here as `scenarios/p1-net-<date>.jsonl.gz`
(`{"t": iso8601, "w": signed_watts}` per line, derived from the HA recorder
export) for the reason `scenarios/p1-age-2026-08-26.jsonl.gz` already is: the
infra repo is private, CI's token is minted for this repository only, and a
regression fixture that CI cannot see is not a regression fixture.

Closing AC 3 needs 30 days of capture. That is TEST-01's replay harness, and it
cannot start before somebody starts recording.
"""

import asyncio
import gzip
import json
import shutil
import sys
import tempfile
from datetime import datetime, timedelta, timezone
from pathlib import Path

from app.main import Controller
from app.p1 import P1Ingest, QuarterAverager, QUARTER_S, make_sample, split_signed
from app.tariff import (
    BILLING_TZ, CHECKPOINT_S, NO_RECORD, NOT_OBSERVED, POLICY_MAX,
    POLICY_NET, POLICY_SUM, PeakStore, PeakTracker, _EUTimezone, billed_w,
    month_key,
)

fails = []
HERE = Path(__file__).resolve().parent
TMP = Path(tempfile.mkdtemp(prefix="goodwe-tariff-"))

# The live install's numbers: NOTES.md and freeze.py both record max_w 5000
# there. `peak_cap_w` ships at 3500 in config.yaml and is the SOFT LIMIT AC 3
# grades against - read out of the shipped file in part 10 rather than typed.
LIVE_MAX_W = 5000.0


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


def near(a, b, tol=1e-6):
    return a is not None and b is not None and abs(a - b) <= tol


def sample(t: datetime, net_w: float, phases=1, phase_net=None):
    """One P1 telegram, through the SHIPPED derivation.

    `split_signed` and `make_sample` are what the real transports call, so the
    sign convention under test is the one that ships rather than a local
    re-derivation of it.
    """
    imp, exp = split_signed(net_w)
    pi = pe = None
    if phase_net is not None:
        pairs = [split_signed(v) for v in phase_net]
        pi = [a for a, _ in pairs]
        pe = [b for _, b in pairs]
    return make_sample("ha_signed", imp, exp, phases=phases,
                       phase_import_w=pi, phase_export_w=pe,
                       ingest_ts=t, ingest_mono=t.timestamp())


def fresh_store(name: str) -> PeakStore:
    return PeakStore(str(TMP / f"{name}.db"))


# B is a clock-aligned quarter boundary: proved, not assumed, because every
# other number in part 1 rests on it.
B = datetime(2026, 8, 30, 9, 0, 0, tzinfo=timezone.utc)


# ==============================================================================
print("1. the block arithmetic, on a worked example")

check(f"the fixture's start {B.isoformat()} really is 900 s aligned",
      int(B.timestamp()) % QUARTER_S == 0)

# (t seconds after B, net W). Deliberately irregular, with an export, a gap
# longer than max_hold_s, and a last sample that straddles the boundary.
MAX_HOLD_S = 30.0
FIXTURE = [(0, 1000.0), (20, 2000.0), (30, -500.0), (50, 3000.0),
           (250, 4000.0), (880, 1200.0), (1000, 900.0)]

# The worked example, written as the segments a person would compute by hand:
# each HELD value and how many seconds of the FIRST block it is evidence for.
# Two rules do all the work: a sample is held forward until the next one, capped
# at MAX_HOLD_S; and export is 0 W of offtake, never a negative contribution.
SEGMENTS = [
    (1000.0, 20),    # 0 -> 20, a normal telegram interval, held in full
    (2000.0, 10),    # 20 -> 30, likewise
    (0.0, 20),       # 30 -> 50: -500 W is EXPORT, so 0 W of offtake for 20 s
    (3000.0, 30),    # 50 -> 250 is a 200 s GAP; only the first 30 s is evidence
    (4000.0, 30),    # 250 -> 880 is a 630 s gap; again only 30 s is evidence
    (1200.0, 20),    # 880 -> 900: the STRADDLE's first half, up to the boundary
]
# The straddle's second half: the same held 1200 W, from the boundary until its
# hold expires at 880 + 30. It must be credited to the NEXT block, not to this
# one and not to neither.
SPLIT_TAIL_W, SPLIT_TAIL_S = 1200.0, 10
EXP_WS = sum(w * s for w, s in SEGMENTS)
EXP_COVERED_S = sum(s for _, s in SEGMENTS)
EXP_BLOCK_AVG = EXP_WS / QUARTER_S              # what the utility bills
EXP_ELAPSED_AVG = EXP_WS / EXP_COVERED_S        # what a /elapsed divisor gives
NAIVE_MEAN = sum(max(w, 0.0) for _, w in FIXTURE[:-1]) / (len(FIXTURE) - 1)

avg = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
closed = []
for dt_s, w in FIXTURE:
    closed += avg.add(sample(B + timedelta(seconds=dt_s), w))

print(f"    {len(SEGMENTS)} held segments -> {EXP_WS:,.0f} Ws over "
      f"{EXP_COVERED_S:.0f} s of the {QUARTER_S} s block")
check(f"exactly one block closed", len(closed) == 1)
check(f"...starting at the fixture's own boundary {B.isoformat()}",
      bool(closed) and closed[0].start == B)
check(f"...averaging {EXP_BLOCK_AVG:.2f} W - the worked example's "
      f"{EXP_WS:,.0f} Ws over the FULL {QUARTER_S} s",
      bool(closed) and near(closed[0].offtake_avg_w, EXP_BLOCK_AVG, 1e-9))

# The three ways of being wrong, each of which produces a plausible number.
print(f"    the same data read three other ways: /elapsed = {EXP_ELAPSED_AVG:.2f} W, "
      f"naive mean of samples = {NAIVE_MEAN:.2f} W, "
      f"billed = {EXP_BLOCK_AVG:.2f} W")
check("...and /elapsed is NOT the billed figure - a gap must drag the average "
      "down, not be averaged away",
      abs(EXP_ELAPSED_AVG - EXP_BLOCK_AVG) > 1.0)
check("...and a naive mean over samples is NOT the billed figure either - it "
      "weights a burst of fast telegrams like a slow one",
      abs(NAIVE_MEAN - EXP_BLOCK_AVG) > 1.0)

# The straddle: the 4000 W sample at t=400 is held ACROSS 900. Its evidence
# expires at 430, so the second block gets none of it - and the second block
# must still have been OPENED at 900, not at 1000.
check(f"the block after the straddle opens at the boundary, not at the sample",
      avg.block_start == B + timedelta(seconds=QUARTER_S))
print(f"    the straddling sample is SPLIT at the boundary: "
      f"{SEGMENTS[-1][0]:.0f} W x {SEGMENTS[-1][1]} s into the first block, "
      f"x {SPLIT_TAIL_S} s into the second")
check(f"...and the second block is credited the tail "
      f"({SPLIT_TAIL_W * SPLIT_TAIL_S:,.0f} Ws), so a sample spanning :15 is "
      f"attributed to BOTH blocks in proportion and wholly to neither",
      near(avg.partial_ws, SPLIT_TAIL_W * SPLIT_TAIL_S, 1e-9))


# ==============================================================================
print()
print("2. the 900 s degeneracy - where three different formulas agree")

# A complete block at a constant power, fully covered. Here acc/elapsed,
# acc/900 and the projection are ALL the same number. Anything checked only
# here is checked against nothing.
HELD_W = 2400.0
deg = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
deg_closed = []
for i in range(0, QUARTER_S + 1, 5):
    deg_closed += deg.add(sample(B + timedelta(seconds=i), HELD_W))
d_block = deg_closed[0].offtake_avg_w
deg2 = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
LAST_S = QUARTER_S - 5
for i in range(0, LAST_S + 1, 5):
    deg2.add(sample(B + timedelta(seconds=i), HELD_W))
d_partial = deg2.offtake_avg_w
# Graded AT the last sample: full coverage behind it, a held value in front.
d_proj = PeakTracker.project(deg2, B + timedelta(seconds=LAST_S)).projected_w
print(f"    at full coverage of a whole block: acc/900 = {d_block:.2f} W, "
      f"acc/elapsed = {d_partial:.2f} W, projection = {d_proj:.2f} W")
check("all three formulas coincide at exactly 900 s of full coverage - so a "
      "check placed only there distinguishes none of them",
      near(d_block, HELD_W, 1e-6) and near(d_partial, HELD_W, 1e-6)
      and near(d_proj, HELD_W, 1e-6))

# Away from it, on part 1's fixture at a point mid-block, they must separate.
mid = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
for dt_s, w in FIXTURE[:-1]:
    mid.add(sample(B + timedelta(seconds=dt_s), w))
NOW = B + timedelta(seconds=500)
p = PeakTracker.project(mid, NOW)
print(f"    at t=500 s on part 1's fixture: acc/elapsed = {p.elapsed_w:.2f} W, "
      f"projection = {p.projected_w:.2f} W, coverage = {p.covered_frac * 100:.1f} %")
check("mid-block, with a gap in it, the elapsed average and the projection are "
      "DIFFERENT numbers", abs(p.elapsed_w - p.projected_w) > 1.0)
check("...and the projection carries elapsed_s, so a consumer can see it is an "
      f"extrapolation over {QUARTER_S - p.elapsed_s:.0f} s it has not seen",
      near(p.elapsed_s, 500.0, 1e-6))
check("...and a gap shows up as coverage below 1.0 rather than as a confident "
      "low number", p.covered_frac < 1.0 and not p.complete)


# ==============================================================================
print()
print("3. restart mid-block: the same block, or a lower one, never a lost one")

RUN_W = 2000.0
STEP_S = 1
RESTART_AT_S = 600      # minute 10 of the quarter - the functional note's case


def feed(avg, t0_s, t1_s, w=RUN_W):
    out = []
    for i in range(t0_s, t1_s + 1, STEP_S):
        out += avg.add(sample(B + timedelta(seconds=i), w))
    return out


# (a) the control: no restart at all.
ctrl = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
ctrl_closed = feed(ctrl, 0, QUARTER_S)
CTRL_W = ctrl_closed[0].offtake_avg_w

# (b) restarted at minute 10, checkpoint restored through a real SQLite file.
st = fresh_store("restart")
tr = PeakTracker(st, cap_w=LIVE_MAX_W)
a1 = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
feed(a1, 0, RESTART_AT_S)
check("the checkpoint round-trips through SQLite", st.save_partial(a1.snapshot()))
a2 = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)   # a brand-new process
restored = tr.restore(a2, B + timedelta(seconds=RESTART_AT_S + 1))
print(f"    restore() said: {restored}")
check("...and restore() reports RESUMING rather than starting fresh",
      restored.startswith("resumed"))
r_closed = feed(a2, RESTART_AT_S + STEP_S, QUARTER_S)
RESTART_W = r_closed[0].offtake_avg_w if r_closed else None

# (c) the mutation guard: the SAME restart with no restore.
a3 = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
n_closed = feed(a3, RESTART_AT_S + STEP_S, QUARTER_S)
NORESTORE_W = n_closed[0].offtake_avg_w if n_closed else None

print(f"    uninterrupted {CTRL_W:.2f} W | restored {RESTART_W:.2f} W | "
      f"restart with no restore {NORESTORE_W:.2f} W")
check(f"a restart at {RESTART_AT_S} s lands on the SAME billed block as an "
      f"uninterrupted run", near(RESTART_W, CTRL_W, 1e-6))
check("...and without the restore it lands LOWER - so 'the restore works' is "
      "not satisfied by a fixture where restoring changes nothing",
      NORESTORE_W is not None and NORESTORE_W < CTRL_W - 1.0)
# Derived, not asserted against a literal: the lost part is exactly the seconds
# before the restart, at the power that was flowing.
DROPPED_S = RESTART_AT_S + STEP_S     # the resumed feed starts one step later
check(f"...and it is lower by exactly the {DROPPED_S} s it dropped "
      f"({RUN_W * DROPPED_S / QUARTER_S:.2f} W)",
      near(CTRL_W - NORESTORE_W, RUN_W * DROPPED_S / QUARTER_S, 1e-6))

# (d) a LONG outage: restored, but the meter was away past max_hold_s. The
# uncovered seconds must drag the block DOWN, not be held forward.
OUTAGE_S = 200
a4 = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
st4 = fresh_store("outage")
st4.save_partial(a1.snapshot())
PeakTracker(st4, cap_w=LIVE_MAX_W).restore(a4, B + timedelta(seconds=RESTART_AT_S + OUTAGE_S))
o_closed = feed(a4, RESTART_AT_S + OUTAGE_S, QUARTER_S)
OUTAGE_UNCOVERED_S = OUTAGE_S - MAX_HOLD_S
print(f"    a {OUTAGE_S} s outage inside the quarter, of which "
      f"{OUTAGE_UNCOVERED_S:.0f} s is past the {MAX_HOLD_S:.0f} s hold: "
      f"{o_closed[0].offtake_avg_w:.2f} W")
check(f"...reads lower than the uninterrupted block by exactly the uncovered "
      f"seconds ({RUN_W * OUTAGE_UNCOVERED_S / QUARTER_S:.2f} W) - a gap is "
      f"never held forward across the whole outage",
      near(CTRL_W - o_closed[0].offtake_avg_w,
           RUN_W * OUTAGE_UNCOVERED_S / QUARTER_S, 1e-6))

# (e) double counting is answered by the primary key, not by care.
st5 = fresh_store("dedupe")
st5.record_block(ctrl_closed[0])
st5.record_block(ctrl_closed[0])
st5.record_block(ctrl_closed[0])
rows = st5.blocks_in(month_key(B))
check("banking the same quarter three times leaves ONE row - quarter_start is "
      "the primary key, so a replay or a restart cannot double-count",
      len(rows) == 1 and near(rows[0][1], CTRL_W, 1e-9))

# ...and a later pass with LESS coverage may not lower a banked figure.
from app.tariff import _ClosedBlock as _CB
st5.record_block(_CB(start=B, offtake_avg_w=CTRL_W / 2, per_phase_offtake_avg_w=None))
after = st5.blocks_in(month_key(B))
check(f"...and re-banking the same quarter at half the figure leaves it at "
      f"{CTRL_W:.0f} W - a restart's partial re-run must never lower a "
      f"quarter that was already measured in full",
      len(after) == 1 and near(after[0][1], CTRL_W, 1e-9))

# (f) the checkpoint's own staleness bound, derived from the constant.
print(f"    CHECKPOINT_S = {CHECKPOINT_S:.0f} s, so a crash can lose at most "
      f"{CHECKPOINT_S / QUARTER_S * 100:.1f} % of one quarter's average")
check("the checkpoint bound is small enough that its worst case is under 2 % "
      "of a quarter", CHECKPOINT_S / QUARTER_S < 0.02)


# ==============================================================================
print()
print("4. month rollover, in LOCAL time, and across a restart that straddles it")

# CEST is UTC+2, so the last two hours of 31 August UTC are already September.
AUG_LATE = datetime(2026, 8, 31, 21, 45, tzinfo=timezone.utc)
SEP_EARLY = datetime(2026, 8, 31, 22, 15, tzinfo=timezone.utc)
print(f"    {AUG_LATE.isoformat()} is local "
      f"{AUG_LATE.astimezone(BILLING_TZ).isoformat()} -> {month_key(AUG_LATE)}")
print(f"    {SEP_EARLY.isoformat()} is local "
      f"{SEP_EARLY.astimezone(BILLING_TZ).isoformat()} -> {month_key(SEP_EARLY)}")
check("a quarter starting 21:45 UTC on 31 Aug bills in AUGUST",
      month_key(AUG_LATE) == "2026-08")
check("a quarter starting 22:15 UTC on 31 Aug bills in SEPTEMBER - filing it "
      "under August would move a whole month's bill",
      month_key(SEP_EARLY) == "2026-09")
check("...and the two are genuinely on opposite sides of the boundary, not "
      "the same answer twice", month_key(AUG_LATE) != month_key(SEP_EARLY))
# Winter, where the offset is +1 and the boundary sits an hour later.
JAN_LATE = datetime(2026, 1, 31, 22, 45, tzinfo=timezone.utc)
check("in WINTER the same trap is an hour later: 22:45 UTC on 31 Jan is still "
      "January, and 23:15 UTC is February",
      month_key(JAN_LATE) == "2026-01"
      and month_key(JAN_LATE + timedelta(minutes=30)) == "2026-02")

st6 = fresh_store("months")
AUG_PEAK_W, SEP_PEAK_W = 4998.0, 900.0
st6.record_block(_CB(AUG_LATE, AUG_PEAK_W, None))
st6.record_block(_CB(SEP_EARLY, SEP_PEAK_W, None))
r_aug, r_sep = st6.peak(month="2026-08"), st6.peak(month="2026-09")
print(f"    {r_aug.explain()}")
print(f"    {r_sep.explain()}")
check(f"August's peak is the {AUG_PEAK_W:.0f} W quarter (specs.md:583's own "
      f"measured case) and September's is not",
      r_aug.is_observed and near(r_aug.peak_w, AUG_PEAK_W)
      and r_sep.is_observed and near(r_sep.peak_w, SEP_PEAK_W))
check("...so a new month starts from its OWN quarters, not from the previous "
      "month's peak", r_sep.peak_w < r_aug.peak_w)

# The restart that straddles the boundary: a checkpoint whose open block is in
# August, restored when `now` is already September.
st7 = fresh_store("straddle")
tr7 = PeakTracker(st7, cap_w=LIVE_MAX_W)
straddler = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
for i in range(0, 400, STEP_S):
    straddler.add(sample(AUG_LATE + timedelta(seconds=i), RUN_W))
st7.save_partial(straddler.snapshot())
after_boot = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
said = tr7.restore(after_boot, SEP_EARLY + timedelta(minutes=3))
print(f"    restore() across the boundary said: {said}")
aug7, sep7 = st7.peak(month="2026-08"), st7.peak(month="2026-09")
check("a restart that straddles the month boundary BANKS the dead August "
      "quarter rather than dropping measured energy",
      said.startswith("closed a stale block") and aug7.is_observed)
check(f"...at its partial-coverage figure ({aug7.peak_w:.2f} W = "
      f"{RUN_W * 399 / QUARTER_S:.2f} W, the seconds it actually covered)",
      near(aug7.peak_w, RUN_W * 399 / QUARTER_S, 1e-6))
check("...and September reads NO_RECORD, which is a different answer from "
      "August's peak AND from a zero",
      sep7.state == NO_RECORD and sep7.peak_w is None)
check("...and the checkpoint is cleared, so a second boot cannot re-bank it",
      st7.load_partial() is None)


# ==============================================================================
print()
print("5. AN UNREADABLE STORE IS NOT A ZERO PEAK  (the acceptance argument)")

# Three stores, three states, and the three must not answer alike.
good = fresh_store("states")
good_tr = PeakTracker(good, cap_w=LIVE_MAX_W)
empty_r = good_tr.peak(B)

gone = fresh_store("gone")
gone_tr = PeakTracker(gone, cap_w=LIVE_MAX_W)
gone.record_block(_CB(B, 3000.0, None))
gone.close()                       # the database goes away under a live reader
gone_r = gone_tr.peak(B)

never = PeakStore(str(TMP / "no-such-dir" / "x.db"))
never_tr = PeakTracker(never, cap_w=LIVE_MAX_W)
never_r = never_tr.peak(B)

# A store that is OPEN and whose query FAILS is a third code path - a closed
# handle short-circuits before the SQL ever runs, so without this the
# `except sqlite3.Error` arm of peak() is never executed by anything.
broken = fresh_store("broken-table")
broken.record_block(_CB(B, 3000.0, None))
broken._db.execute("DROP TABLE monthly_peak")
broken._db.commit()
broken_r = PeakTracker(broken, cap_w=LIVE_MAX_W).peak(B)

for label, r in (("fresh store, no quarter yet", empty_r),
                 ("store that broke under a live reader", gone_r),
                 ("store that never opened", never_r),
                 ("store open, query failed", broken_r)):
    print(f"    {label:38s} -> {r.state:13s} {r.explain()}")

check("a fresh store with no quarter yet says NO_RECORD - a normal state on "
      "the 1st of the month", empty_r.state == NO_RECORD)
check("a store that broke says NOT_OBSERVED", gone_r.state == NOT_OBSERVED)
check("a store that never opened says NOT_OBSERVED", never_r.state == NOT_OBSERVED)
check("a store whose QUERY fails says NOT_OBSERVED too - a different code path "
      "from a closed handle, and the one a half-corrupt database takes",
      broken_r.state == NOT_OBSERVED and broken_r.peak_w is None)
# ...and losing the DERIVED record does not lose the peak, which is the whole
# argument for keeping monthly_peak recomputable from the 15-min tier.
_rec = broken.recompute_month(month_key(B))
print(f"    ...but with the 15-min tier intact, recompute_month() still "
      f"answers: {_rec.explain()}")
check("...while recompute_month() STILL answers from the surviving 15-min "
      "tier - the permanent record is derived, so losing it is recoverable",
      _rec.is_observed and near(_rec.peak_w, 3000.0))
broken._db.execute("DROP TABLE quarter_blocks")
broken._db.commit()
check("...and once BOTH tiers are gone it says NOT_OBSERVED rather than "
      "raising into the control loop",
      broken.recompute_month(month_key(B)).state == NOT_OBSERVED)
check("...and NO_RECORD and NOT_OBSERVED are DIFFERENT states, not two "
      "spellings of the same one", empty_r.state != gone_r.state)
check("...and they read differently to a human, too",
      empty_r.explain() != gone_r.explain()
      and "NOT" in gone_r.explain() and "zero peak" in gone_r.explain())

check("an unreadable peak is None, not 0.0 - there is no number to mistake",
      gone_r.peak_w is None and never_r.peak_w is None)
check("...and NO_RECORD carries no number either", empty_r.peak_w is None)
_typed = False
try:
    float(gone_r.peak_w)
except TypeError:
    _typed = True
check("...and a caller who forgets to ask gets a TypeError rather than a "
      "silently disarming zero", _typed)
check("PeakReading exposes no accessor that defaults a missing peak to a "
      "number - no .get(), no _or() ",
      not [a for a in dir(gone_r)
           if a.startswith(("get", "value")) or a.endswith("_or")])

# The consequence, which is the reason any of this matters.
print(f"    headroom under a {LIVE_MAX_W:.0f} W limit: "
      f"no quarter yet -> {good_tr.headroom_w(B)}, "
      f"unreadable -> {gone_tr.headroom_w(B)}")
check("an unreadable store yields NO headroom figure - the disarming default "
      "(full headroom, guard silently off) is unreachable",
      gone_tr.headroom_w(B) is None and never_tr.headroom_w(B) is None)
check(f"...where a genuinely empty month yields the full {LIVE_MAX_W:.0f} W, "
      f"which is correct and is a different answer",
      near(good_tr.headroom_w(B), LIVE_MAX_W))
check("the 80 % warning fraction is likewise None when nothing was observed",
      gone_tr.warning_frac(B) is None and good_tr.warning_frac(B) == 0.0)

# And a real peak reduces headroom, so "always None" does not pass this part.
good.record_block(_CB(B, LIVE_MAX_W * 0.9, None))
print(f"    after one {LIVE_MAX_W * 0.9:.0f} W quarter: "
      f"headroom {good_tr.headroom_w(B):.0f} W, "
      f"warning {good_tr.warning_frac(B) * 100:.0f} % of the limit")
check("a real observed peak gives a real headroom, so this part is not passed "
      "by a store that answers None to everything",
      near(good_tr.headroom_w(B), LIVE_MAX_W * 0.1, 1e-6))
check(f"...and it is over §9.1's 80 % warning threshold",
      good_tr.warning_frac(B) > 0.8)

# A CORRUPT database file is the third way a store fails, and it must land in
# NOT_OBSERVED too rather than raising into the control loop.
corrupt_path = TMP / "corrupt.db"
corrupt_path.write_bytes(b"this is not a database" * 64)
corrupt = PeakStore(str(corrupt_path))
c_r = PeakTracker(corrupt, cap_w=LIVE_MAX_W).peak(B)
print(f"    a corrupt database file -> {c_r.state}")
check("a corrupt database reads NOT_OBSERVED and does not raise into the "
      "control loop (§6 fail-open)", c_r.state == NOT_OBSERVED)


# ==============================================================================
print()
print("6. the timezone fallback, graded against zoneinfo")

_zi = None
try:
    from zoneinfo import ZoneInfo
    _zi = ZoneInfo("Europe/Brussels")
except Exception as err:                      # noqa: BLE001
    print(f"    zoneinfo unavailable here ({err}) - the fallback IS the tz")

if _zi is None:
    check("zoneinfo is absent, so the encoded rule cannot be graded here - "
          "reported, not silently passed", True)
else:
    fb = _EUTimezone()
    # ⚠️ HOURLY, NOT 3-HOURLY, ACROSS THE TRANSITION DAYS. Both EU transitions
    # happen at 01:00 UTC, so a rule that is wrong BY AN HOUR differs from the
    # right one only between 01:00 and 02:00 UTC. A 3-hourly sweep steps over
    # that window entirely and grades nothing - measured: a mutation moving the
    # UTC threshold from 01:00 to 02:00 survived the whole file at 3-hourly.
    probes = [datetime(2026, 1, 1, 12, tzinfo=timezone.utc)]
    probes += [datetime(2026, m, d, h, mi, tzinfo=timezone.utc)
               for m in (3, 10) for d in (28, 29, 30, 31)
               for h in range(24) for mi in (0, 30)]
    probes += [datetime(2026, m, 1, 0, 30, tzinfo=timezone.utc) for m in range(1, 13)]
    # ⚠️ COMPARE THE CONVERTED WALL CLOCK, NOT ITS utcoffset(). `astimezone`
    # goes through `fromutc`, but the RESULT carries local wall values, so
    # asking it for `.utcoffset()` routes straight back into the LOCAL arm and
    # never grades `fromutc` at all. Measured: a mutation moving the UTC
    # threshold by an hour survived 397 offset-comparing probes and dies
    # immediately on the wall clock they produce.
    def _wallclock(t, tz):
        return t.astimezone(tz).replace(tzinfo=None)

    bad = [t for t in probes if _wallclock(t, fb) != _wallclock(t, _zi)]
    print(f"    {len(probes)} probes across both DST transitions and every "
          f"month boundary: {len(bad)} disagreement(s) in the LOCAL WALL CLOCK "
          f"each one converts to")
    check("the encoded EU rule converts every probe to the same local wall "
          "clock as zoneinfo, including the two transition weekends", not bad)
    # ...and the probes are not all the same offset, or the check is vacuous.
    offsets = {t.astimezone(_zi).utcoffset() for t in probes}
    check(f"...and the probe set spans BOTH offsets ({sorted(str(o) for o in offsets)}), "
          f"so agreeing is not free", len(offsets) == 2)
    check(f"BILLING_TZ resolved to zoneinfo on this host, not the fallback",
          not isinstance(BILLING_TZ, _EUTimezone))

    # ⚠️ THE PROBES ABOVE ONLY EXERCISE `fromutc()`. `astimezone` on a UTC
    # instant never calls `utcoffset()` with a local wall time, so the local
    # arm of the rule was ungraded - a mutation that moved it by an hour
    # survived the whole file. These probes drive it directly. Noon is used
    # because it is unambiguous on every day of the year, including the two
    # transition days where 02:00-03:00 local is missing or repeated.
    #
    # ⚠️ AND THE LOCAL PROBES MUST INCLUDE 03:00-03:59 ON THE OCTOBER SUNDAY.
    # That hour is where a local rule that ends DST an hour late still claims
    # CEST while zoneinfo has already returned to CET; at noon both answers
    # agree and the probe is worth nothing. Measured: at noon-only, a mutation
    # moving the local thresholds from 02:00/03:00 to 03:00/04:00 survived.
    #
    # The MARCH gap (02:00-02:59 local, which does not exist) is deliberately
    # NOT probed: Python hands a nonexistent local time to `utcoffset` as an
    # ordinary value and any answer is defensible, so grading it would pin an
    # arbitrary choice. It cannot move a month boundary - it is the last Sunday
    # of March, never the 1st of a month - which is the only thing this tz is
    # used for.
    wall = [datetime(2026, m, d, 12) for m, d in
            [(3, 27), (3, 28), (3, 29), (3, 30), (10, 24), (10, 25),
             (10, 26), (10, 27)]]
    wall += [datetime(2026, 10, 25, h, mi) for h in range(0, 6) for mi in (0, 30)
             if h != 2]                       # 02:00 local is the ambiguous fold
    wall += [datetime(2026, 3, 29, h, mi) for h in (0, 1, 3, 4, 5) for mi in (0, 30)]
    wall += [datetime(2026, m, 1, 12) for m in range(1, 13)]
    bad_w = [w for w in wall
             if w.replace(tzinfo=fb).utcoffset() != w.replace(tzinfo=_zi).utcoffset()]
    w_offsets = {w.replace(tzinfo=_zi).utcoffset() for w in wall}
    print(f"    {len(wall)} LOCAL-wall probes through utcoffset(): "
          f"{len(bad_w)} disagreement(s), spanning "
          f"{sorted(str(o) for o in w_offsets)}")
    check("utcoffset() on local wall time matches zoneinfo across both "
          "transition weekends - the arm astimezone() never reaches", not bad_w)
    check("...and those probes span both offsets too", len(w_offsets) == 2)
    bad_n = [w for w in wall
             if w.replace(tzinfo=fb).tzname() != w.replace(tzinfo=_zi).tzname()]
    check("...and tzname() agrees with zoneinfo at every one of them (CET/CEST)",
          not bad_n)


# ==============================================================================
print()
print("7. the per-phase reduction, on an UNBALANCED three-phase load (§17)")

# One phase exporting is what separates the three policies. On a balanced load
# they all give the same number, which is why this fixture is not balanced.
PH = (3000.0, 500.0, -200.0)
NET = sum(PH)
u = QuarterAverager(phases=3, max_hold_s=MAX_HOLD_S)
u_closed = []
for i in range(0, QUARTER_S + 1, 5):
    u_closed += u.add(sample(B + timedelta(seconds=i), NET, phases=3, phase_net=PH))
blk = u_closed[0]
vals = {p: billed_w(blk, p) for p in (POLICY_NET, POLICY_SUM, POLICY_MAX)}
print(f"    phases {PH} (net {NET:+.0f} W) -> "
      + ", ".join(f"{k}={v:.0f} W" for k, v in vals.items()))
check(f"the NET policy bills the connection's net offtake ({NET:.0f} W)",
      near(vals[POLICY_NET], NET, 1e-6))
check(f"the SUM policy clips each phase first, so the exporting phase does not "
      f"net off ({sum(max(v, 0.0) for v in PH):.0f} W)",
      near(vals[POLICY_SUM], sum(max(v, 0.0) for v in PH), 1e-6))
check(f"the MAX policy bills the worst single phase ({max(PH):.0f} W)",
      near(vals[POLICY_MAX], max(PH), 1e-6))
check("...and on THIS fixture all three are different numbers - which is the "
      "whole reason the reduction is a named policy and not an assumption",
      len({round(v, 6) for v in vals.values()}) == 3)
check("the shipped default is NET, the meter's own billed register",
      billed_w(blk) == vals[POLICY_NET])

bal = QuarterAverager(phases=3, max_hold_s=MAX_HOLD_S)
b_closed = []
for i in range(0, QUARTER_S + 1, 5):
    b_closed += bal.add(sample(B + timedelta(seconds=i), 3000.0, phases=3,
                               phase_net=(1000.0, 1000.0, 1000.0)))
b_vals = {p: billed_w(b_closed[0], p) for p in (POLICY_NET, POLICY_SUM, POLICY_MAX)}
check(f"...and on a BALANCED load NET and SUM coincide ({b_vals[POLICY_NET]:.0f} W), "
      f"so a balanced fixture could not have told them apart",
      near(b_vals[POLICY_NET], b_vals[POLICY_SUM], 1e-6))


# ==============================================================================
print()
print("8. monthly_peak is DERIVED, and provably has not diverged")

st8 = fresh_store("derive")
QUARTERS = [(0, 1200.0), (1, 4998.0), (2, 3357.0), (3, 800.0), (4, 4832.0)]
for q, w in QUARTERS:
    st8.record_block(_CB(B + timedelta(seconds=q * QUARTER_S), w, None))
MAXQ = max(w for _, w in QUARTERS)
stored = st8.peak(month=month_key(B))
derived = st8.recompute_month(month_key(B))
print(f"    {len(QUARTERS)} quarters banked; stored peak {stored.peak_w:.0f} W, "
      f"recomputed from quarter_blocks {derived.peak_w:.0f} W")
check(f"the stored monthly peak is the largest quarter ({MAXQ:.0f} W - "
      f"specs.md:583's own figure)",
      stored.is_observed and near(stored.peak_w, MAXQ))
check("...and re-deriving it from the 15-min tier gives the same number and "
      "the same quarter, so the two records are proven equal rather than "
      "assumed to be",
      derived.is_observed and near(derived.peak_w, stored.peak_w)
      and derived.quarter_start == stored.quarter_start)
check("...and the two measured regressions from §9.1 land on DIFFERENT peaks - "
      f"the sustained {MAXQ:.0f} W quarter, not the oscillating 3357 W one",
      not near(stored.peak_w, 3357.0))
# A quarter arriving LATER and lower must not lower the month.
st8.record_block(_CB(B + timedelta(seconds=9 * QUARTER_S), 100.0, None))
check("a later, smaller quarter does not lower the month's peak",
      near(st8.peak(month=month_key(B)).peak_w, MAXQ))
check("...and the recompute agrees with it after that write, too",
      near(st8.recompute_month(month_key(B)).peak_w, MAXQ))
# No pruning: §15 says these two tiers are indefinite / permanent.
src = (HERE / "app" / "tariff.py").read_text(encoding="utf-8")
check("§15's retention is structural: no DELETE against quarter_blocks or "
      "monthly_peak exists anywhere in tariff.py",
      "DELETE FROM quarter_blocks" not in src
      and "DELETE FROM monthly_peak" not in src
      and "DROP TABLE" not in src)


# ==============================================================================
print()
print("9. AC 3, on the replay data that actually exists")

REPLAY = sorted((HERE / "scenarios").glob("p1-net-*.jsonl.gz"))
print(f"    replay fixtures found: {len(REPLAY)} "
      f"({', '.join(p.name for p in REPLAY) or 'none'})")
check("the committed replay fixtures are present - a regression fixture CI "
      "cannot see is not a regression fixture", len(REPLAY) >= 2)

# ⚠️ `peak_cap_w:` appears TWICE in config.yaml - once under `options:` with the
# default, once under `schema:` with `int(500,15000)`. Taking the last match
# would blow up on the schema line, and taking any match that parses would be
# luck; the FIRST is the options block, which is the number an install lands on.
SHIPPED_CAP_W = None
for line in (HERE / "config.yaml").read_text(encoding="utf-8").splitlines():
    if line.strip().startswith("peak_cap_w:"):
        try:
            SHIPPED_CAP_W = float(line.split(":", 1)[1].strip())
        except ValueError:
            continue
        break
check("config.yaml's shipped peak_cap_w was read out of the file rather than "
      "typed here", SHIPPED_CAP_W is not None)

replay_days = 0
total_q = 0
under = 0
for path in REPLAY:
    st9 = fresh_store("replay-" + path.stem)
    ing = P1Ingest(phases=1, max_age_s=MAX_HOLD_S)
    tr9 = PeakTracker(st9, cap_w=SHIPPED_CAP_W or LIVE_MAX_W)
    first = last = None
    with gzip.open(path, "rt", encoding="utf-8") as fh:
        for line in fh:
            row = json.loads(line)
            t = datetime.fromisoformat(row["t"])
            first = first or t
            last = t
            ing.submit(sample(t, float(row["w"])))
    tr9.sync(ing)
    months = {month_key(b.start) for b in ing.blocks}
    span_h = (last - first).total_seconds() / 3600.0
    qs = [w for _, w in st9.blocks_in(sorted(months)[0])] if months else []
    peak = max(qs) if qs else 0.0
    ok_q = sum(1 for w in qs if w <= (SHIPPED_CAP_W or LIVE_MAX_W))
    total_q += len(qs)
    under += ok_q
    replay_days += span_h / 24.0
    print(f"    {path.name}: {span_h:.1f} h, {len(qs)} closed quarters, "
          f"month peak {peak:.0f} W, "
          f"{ok_q}/{len(qs)} = {100.0 * ok_q / max(len(qs), 1):.1f} % under the "
          f"{SHIPPED_CAP_W:.0f} W soft limit")
    check(f"{path.name}: the month's peak equals its own largest quarter",
          near(st9.peak(month=sorted(months)[0]).peak_w, peak, 1e-9))
    check(f"{path.name}: a full day of real telegrams produces ~96 quarters, "
          f"got {len(qs)}", 90 <= len(qs) <= 96)

PCT = 100.0 * under / max(total_q, 1)
print()
print(f"    AC 3 MEASURED: {under}/{total_q} quarters = {PCT:.1f} % under the "
      f"{SHIPPED_CAP_W:.0f} W soft limit, over {replay_days:.1f} DAYS of "
      f"replay, UNCONTROLLED (no peak shaving was running on either day).")
print(f"    AC 3 ASKS FOR 30 DAYS. {replay_days:.1f} days exist. "
      f"AC 3 IS NOT MET and this run does not claim it is - see the module "
      f"docstring. The figure above is the BASELINE SAFETY-07 has to improve.")
check("the replay produced a measurable number over a stated sample size, "
      "rather than a criterion quietly redefined to fit the data",
      total_q > 100 and replay_days < 30)


# ==============================================================================
print()
print("10. the wiring: the real Controller, and what its status page says")


class _Store:
    data = {}

    def set(self, *a):
        pass

    def get_time(self, *a):
        return None


class _Pub:
    enabled = False

    def publish(self, *a, **k):
        pass

    def close(self):
        pass


class _Hass:
    async def limits(self, entity):
        return (-LIVE_MAX_W, LIVE_MAX_W, 1.0)

    async def number(self, entity, invert=False):
        return None

    async def set_number(self, entity, value):
        return True

    async def call(self, domain, service, data):
        return True


def built(db_path, opts=None):
    return Controller(dict(
        {"setpoint_entity": "number.sp", "meter_source": "ha_signed",
         "p1_net_entity": "sensor.net", "meter_phases": 1,
         "max_w": LIVE_MAX_W, "gain": 0.6, "slew_w": 1000, "deadband_w": 15,
         "target_grid_w": 0, "step_w": 10, "heartbeat_s": 10,
         "stale_input_s": 15}, **(opts or {})),
        _Hass(), _Store(), _Pub(), stop=asyncio.Event(),
        peak_db_path=str(db_path))


c = built(TMP / "wired.db")
check(f"a Controller built with peak_cap_w ABSENT takes the SHIPPED "
      f"{SHIPPED_CAP_W:.0f} W, so config.yaml and main.py's fallback cannot "
      f"drift apart unnoticed", c.peak.cap_w == SHIPPED_CAP_W)

# Drive one real quarter through the real ingest, then the real cycle's sync.
for i in range(0, QUARTER_S + 1, 5):
    c.p1.submit(sample(B + timedelta(seconds=i), 4998.0))
n = c.peak.sync(c.p1)
wr = c.peak.peak(B)
print(f"    after one 4998 W quarter through the real ingest: "
      f"{n} banked, {wr.explain()}")
check("the closed quarter reached SQLite through the tracker the cycle calls",
      n == 1 and wr.is_observed and near(wr.peak_w, 4998.0, 1e-6))

# ⚠️ `cycle()` calls sync() ONCE A SECOND against a `blocks` list that keeps a
# closed quarter for a DAY. Without a dedupe that is ~86 000 re-banks of the
# same row per block, and `recorded` - the number the status page prints as
# "quarters banked" - would be a count of loop iterations rather than of
# quarters. The database stays correct either way, which is exactly why this
# needs its own check instead of being caught by the peak.
again = c.peak.sync(c.p1)
third = c.peak.sync(c.p1)
print(f"    sync() called three times over the same closed quarter: "
      f"{n}, {again}, {third} newly banked; tracker reports "
      f"{c.peak.recorded} quarter(s)")
check("...and syncing the same quarter again banks nothing new, so the count "
      "the status page shows is quarters and not cycles",
      again == 0 and third == 0 and c.peak.recorded == 1)
check("...and the peak is unchanged by the repeat",
      near(c.peak.peak(B).peak_w, 4998.0, 1e-6))

rows = [r for r in c.checks() if "capacity-tariff" in r["text"] or "peak" in r["text"]]
peak_rows = [r for r in rows if "peak" in r["text"] and "quarter(s) banked" in r["text"]]
print("    status row: " + (peak_rows[0]["text"] if peak_rows else "(none)"))
check("the status page carries exactly one capacity-tariff peak row",
      len(peak_rows) == 1)
check("...naming the peak, the soft limit and the quarters banked, all derived "
      "from the controller that was actually built",
      bool(peak_rows) and "4998 W" in peak_rows[0]["text"]
      and f"{SHIPPED_CAP_W:.0f} W soft limit" in peak_rows[0]["text"]
      and "1 quarter(s) banked" in peak_rows[0]["text"])
check("...and it is FLAGGED, because 4998 W is over §9.1's 80 % warning point "
      f"on a {SHIPPED_CAP_W:.0f} W limit",
      bool(peak_rows) and not peak_rows[0]["ok"] and peak_rows[0]["warn"])

# The half that matters: the same page with a store that cannot answer.
c2 = built(TMP / "no-such-dir" / "nope.db")
u_rows = [r for r in c2.checks() if "capacity-tariff peak" in r["text"]]
print("    status row (unreadable store): "
      + (u_rows[0]["text"] if u_rows else "(none)"))
check("an unreadable store gets its OWN row, saying UNREADABLE", len(u_rows) == 1)
check("...and it does NOT read as a quiet month at 0 W - without this, a broken "
      "instrument looks exactly like a month with no peak",
      bool(u_rows) and "UNREADABLE" in u_rows[0]["text"]
      and "0 %" not in u_rows[0]["text"] and u_rows[0]["warn"])
check("...and the healthy row and the unreadable row are different text",
      bool(u_rows) and bool(peak_rows)
      and u_rows[0]["text"] != peak_rows[0]["text"])

# THE MEASUREMENT IS OF DELIVERED POWER (SAFETY-20). Graded on the parsed
# SYNTAX, not on the text: the module docstring argues this at length and names
# every forbidden word while doing so, so a grep over the source would be
# satisfied by deleting the explanation and would fail on writing one. What must
# be true is that no NAME the code evaluates refers to a command.
import ast as _ast

_tree = _ast.parse(src)
_names = {n.id for n in _ast.walk(_tree) if isinstance(n, _ast.Name)}
_names |= {n.attr for n in _ast.walk(_tree) if isinstance(n, _ast.Attribute)}
_names |= {n.name for n in _ast.walk(_tree)
           if isinstance(n, (_ast.FunctionDef, _ast.ClassDef))}
_FORBIDDEN = {"setpoint", "write_setpoint", "target", "Decision", "i_clamped",
              "clamped", "Claim", "resolve", "P_PEAK", "arbiter"}
_found = sorted(_names & _FORBIDDEN)
_imports = sorted(m.module or "" for m in _ast.walk(_tree)
                  if isinstance(m, _ast.ImportFrom))
print(f"    tariff.py evaluates {len(_names)} distinct names and imports "
      f"from {_imports or ['(stdlib only)']}")
check(f"none of them is a command-side name {sorted(_FORBIDDEN)} - the utility "
      f"bills what FLOWED, so this reads the meter and nothing else"
      + (f"; found {_found}" if _found else ""), not _found)
check("...and it imports nothing from arbiter, clamp or control: it files no "
      "claim, applies no clamp, and commands nothing - SAFETY-07's job",
      not [m for m in _imports if m in ("arbiter", "clamp", "control",
                                        ".arbiter", ".clamp", ".control")])
# ...and the guard is not vacuous: the same grader run over main.py, which DOES
# command, must light up.
_main = {n.attr for n in _ast.walk(
    _ast.parse((HERE / "app" / "main.py").read_text(encoding="utf-8")))
    if isinstance(n, _ast.Attribute)}
check(f"the same grader over main.py finds {sorted(_main & _FORBIDDEN)}, so it "
      f"is capable of failing", bool(_main & _FORBIDDEN))


# ==============================================================================
print()
shutil.rmtree(TMP, ignore_errors=True)
if fails:
    print(f"{len(fails)} FAILED: {', '.join(fails)}")
    sys.exit(1)
print("all checks passed")