diff --git a/goodwe_controller/CHANGELOG.md b/goodwe_controller/CHANGELOG.md index 653ee1b..158514e 100644 --- a/goodwe_controller/CHANGELOG.md +++ b/goodwe_controller/CHANGELOG.md @@ -81,6 +81,59 @@ UTC and rendered at the offset that applies to it. The checks drive both 2026 transitions - the 23-hour and 25-hour days - rather than a normal day, where a fixed 24-slot implementation and a correct one agree. +**The capacity-tariff peak is now measured, banked and survives a restart +(TARIFF-01).** New `app/tariff.py`. The 15-minute block average itself was +already built and unwired - `p1.QuarterAverager` computes clock-aligned 900 s +blocks, time-weighted over irregular telegram spacing, offtake-only and +per-phase - and `P1Ingest.blocks` was a 96-entry list a restart erased. This +adds the SQLite behind it (`/data/tariff.db`): one row per completed quarter in +the indefinite 15-min tier, one row per calendar month in the permanent peak +tier, and no `DELETE` against either anywhere in the module. + +**Blocks, not a sliding window.** The card says "rolling"; 9.1 bills "the +highest 15-minute average offtake", stated per QUARTER. "Rolling" here means +continuously updated, not a trailing-15-minutes window - a trailing window +reports a peak the meter never bills, and reports it HIGH, because a burst +straddling :15 fills it completely while contributing to two block averages at +partial weight. + +**A restart mid-block resumes it, and a restart that outlives the block banks +it.** `QuarterAverager` grew `snapshot()`/`restore()`; the tracker checkpoints +the open block every 15 s (worst case 1.7 % of one quarter) and clears the +checkpoint in the same breath as a block close. The held sample keeps its +ORIGINAL timestamp across the restart, so the outage is integrated under the +same `max_hold_s` rule a running process would apply and uncovered seconds drag +the closed block DOWN rather than being held forward. `quarter_start` is the +primary key and a re-bank may only ever RAISE a figure, so a replay cannot +double-count and a partial re-run cannot lower a quarter already measured. + +**An unreadable store is not a zero peak.** `peak()` answers `OBSERVED` / +`NO_RECORD` / `NOT_OBSERVED` and carries `peak_w = None` unless observed - the +`riglock.holder()` fix applied to something that spends money. A peak read as +0 W hands SAFETY-07 maximum headroom and silently disarms the guard at the +moment its own storage breaks, so there is no accessor that defaults it: a +caller who forgets gets a `TypeError`, and the status page prints a different +row for "no quarter yet" than for "the database did not answer". + +**Calendar months are LOCAL, blocks are UTC.** A quarter starting 22:15 UTC on +31 August bills in September. Block alignment still needs no timezone. + +**No new option, and no second peak entity.** The soft limit is the existing +`peak_cap_w`; `config.yaml` and `validate.py` are untouched, so CFG-03 is not +widened. `peak_forecast_entity` stays the input to `maintenance_charge_floor()` +- repointing that consumer changes what moves energy and belongs to SAFETY-07. + +**It measures and nothing else.** No claim, no clamp, no command; graded on the +parsed syntax in `test_tariff.py`, which also proves the same grader lights up +on `main.py`. Per SAFETY-20, the quantity is DELIVERED power - `tariff.py` only +ever sees `P1Sample`, the meter's own reading, never a setpoint. + +**AC 3 is not met and the check says so.** It asks for 30 days of replay; two +days of this house's real meter exist (2026-08-20, 2026-08-23), now committed as +`scenarios/p1-net-*.jsonl.gz`. `test_tariff.py` part 9 replays them, reports +the percentage under the shipped limit next to the sample size in days, and +asserts nothing about the 95 % figure. + **A computed setpoint over the rating now alarms instead of being clipped in silence (SAFETY-03).** The clamp was never the missing half - specs.md 10.3 records an unclamped integrator climbing ~130 W per 4 s past 10 kW and reporting diff --git a/goodwe_controller/app/main.py b/goodwe_controller/app/main.py index 98db6d1..d7fd7ef 100644 --- a/goodwe_controller/app/main.py +++ b/goodwe_controller/app/main.py @@ -45,6 +45,7 @@ from .hass import HomeAssistant from .maintenance import IDLE, MaintConfig, Maintenance from .mqtt import MqttPublisher from .p1 import P1Ingest, build_source, is_enabled +from .tariff import PeakStore, PeakTracker from .validate import ConfigRefused, refuse, validate_startup_config from .watchdog import SignWatchdog, SignWatchdogConfig, loop_is_following from . import web @@ -63,7 +64,8 @@ def load_options() -> dict: class Controller: - def __init__(self, opts: dict, hass: HomeAssistant, store, mqtt_pub, *, stop): + def __init__(self, opts: dict, hass: HomeAssistant, store, mqtt_pub, *, stop, + peak_db_path: str = "/data/tariff.db"): self.o = opts self.hass = hass self.store = store @@ -163,6 +165,29 @@ class Controller: self.p1 = P1Ingest(phases=int(opts.get("meter_phases", 1)), max_age_s=float(opts.get("meter_max_age_s", 30))) self.p1_enabled = is_enabled(opts) + + # TARIFF-01. The capacity-tariff peak: measured from the meter, banked + # in SQLite, restored across a restart. It is an INSTRUMENT - it files no + # claim and clamps nothing. SAFETY-07 is what will read it. + # + # ⚠️ NO NEW OPTION. The soft limit is `peak_cap_w`, already shipped, + # already validated, already read by control.peak_at_risk(). CFG-03 + # tracks three places option defaults live; a fourth, for a limit that + # already exists, is the mistake the clamp watchdog declined to make ~30 + # lines up. config.yaml and validate.py are untouched by this ticket. + # + # ⚠️ AND NO SECOND PEAK ENTITY. `peak_forecast_entity` is the existing + # HA-side delegation and it stays the input to maintenance_charge_floor(): + # repointing that consumer at this measurement CHANGES WHAT MOVES ENERGY + # during a maintenance charge, which belongs to SAFETY-07 with its own + # acceptance, not to a measurement ticket. What this adds is the MEASURED + # MONTH PEAK - a different quantity from a forecast entity, published + # nowhere as a duplicate of it. + self.peak_store = PeakStore(peak_db_path) + self.peak = PeakTracker(self.peak_store, + cap_w=float(opts.get("peak_cap_w", 3500))) + self.peak.restore(self.p1.averager) + _LOG.info("capacity-tariff peak store: %s", self.peak.restore_state) # Set by amain() once the transport is built, so the status page can show # what only the transport knows (homewizard_local's unchanged_s). Stays # None when P1 is off, or under a transport that has no such counter. @@ -500,6 +525,14 @@ class Controller: # is not, and test_freeze.py part 5 fails on exactly that. await self.check_meter_frozen() + # --- TARIFF-01, and it is LAST on purpose ----------------------------- + # Banking a quarter is bookkeeping about a moment that has already + # happened; a slow SD card must never sit between the meter and the + # setpoint. Its position is unconstrained in the way SAFETY-03's and + # SAFETY-18's are not - it grades neither `target` nor `standing` and + # reads only `self.p1`, which no line in this method writes. + self.peak.sync(self.p1) + async def check_output_clamp(self) -> None: """SAFETY-03. Alarm whenever a clamp actually engaged this cycle. @@ -837,6 +870,27 @@ class Controller: f"{self.freeze_wd.cfg.hold_s:.0f} s unchanged " f"while the command is over " f"{rail_w(self.o.get('max_w', 2000)):.0f} W)"}) + + # TARIFF-01. The month's billed peak, and - the load-bearing half - + # a DIFFERENT row when the store could not answer. "no peak yet" is + # a normal state on the 1st; "unreadable" is a broken instrument, and + # a status page that spelled them the same way would let a 0 W read + # look like a quiet month right up until SAFETY-07 shaved nothing. + _pk = self.peak.peak() + if _pk.is_unreadable: + out.append({"ok": False, "warn": True, + "text": "capacity-tariff peak: " + _pk.explain()}) + else: + # §9.1's "warning entity at 80 %", as a status row rather than + # a second entity - see the constructor on not doubling up. + _frac = self.peak.warning_frac() or 0.0 + _ok = _frac < 0.8 + out.append({"ok": _ok, "warn": not _ok, + "text": _pk.explain() + + f" - {_frac * 100:.0f} % of the " + f"{self.peak.cap_w:.0f} W soft limit" + + f"; {self.peak.recorded} quarter(s) banked" + + f"; on start: {self.peak.restore_state}"}) rows = [("battery SoC", self.soc, o.get("soc_entity")), ("battery power", self.batt, o.get("batt_entity"))] if not self.p1_enabled: diff --git a/goodwe_controller/app/p1.py b/goodwe_controller/app/p1.py index 0687b32..c6ee380 100644 --- a/goodwe_controller/app/p1.py +++ b/goodwe_controller/app/p1.py @@ -353,6 +353,54 @@ class QuarterAverager: self._last_pp = sample.per_phase_w return closed + # -- surviving a restart (TARIFF-01) ------------------------------------ + def snapshot(self) -> dict: + """The open block's whole state, as JSON-safe primitives. + + ⚠️ `last_t` is the HELD SAMPLE'S OWN timestamp and is stored as it is, + not refreshed to "now". That is what makes a restart integrate the gap + under exactly the `max_hold_s` rule a running process would have used: + a short restart is a plateau, a long one leaves uncovered seconds that + drag the closed block's average down. Rewriting it to the restore time + would silently credit the whole outage at the last value seen before it. + """ + return { + "block": self._block, + "acc": self._acc, + "pp_acc": list(self._pp_acc), + "elapsed": self._elapsed, + "last_t": self._last_t, + "last_net": self._last_net, + "last_pp": list(self._last_pp) if self._last_pp is not None else None, + "phases": self.phases, + } + + def restore(self, state: dict) -> bool: + """Put a `snapshot()` back. False if it does not fit this averager. + + A phase-count change is a reconfiguration, not a resumption: the + per-phase accumulator would be reshaped mid-block and the two halves + would not be the same measurement. Refusing is the honest answer, and + the caller reports starting fresh rather than pretending it resumed. + """ + try: + if int(state.get("phases", self.phases)) != self.phases: + return False + pp_acc = list(state.get("pp_acc") or []) + if len(pp_acc) != self.phases: + return False + self._block = None if state["block"] is None else int(state["block"]) + self._acc = float(state["acc"]) + self._pp_acc = [float(v) for v in pp_acc] + self._elapsed = float(state["elapsed"]) + self._last_t = None if state["last_t"] is None else float(state["last_t"]) + self._last_net = float(state["last_net"]) + lp = state.get("last_pp") + self._last_pp = tuple(float(v) for v in lp) if lp is not None else None + return True + except (KeyError, TypeError, ValueError): + return False + # --------------------------------------------------------------------------- # # what the rest of the add-on talks to diff --git a/goodwe_controller/app/tariff.py b/goodwe_controller/app/tariff.py new file mode 100644 index 0000000..7893da5 --- /dev/null +++ b/goodwe_controller/app/tariff.py @@ -0,0 +1,682 @@ +"""TARIFF-01 - the monthly capacity-tariff peak, and the record that outlives it. + +WHAT THIS IS, AND WHAT IT DELIBERATELY IS NOT. This module is an INSTRUMENT. It +measures the number the Belgian capaciteitstarief bills on - the highest +15-minute average offtake of the calendar month (Fluvius / Resa / Ores) - and it +writes that number down somewhere a restart cannot reach. It makes no claim, +applies no clamp, and commands nothing. The guard that will read it is SAFETY-07, +and it belongs in `arbiter.py` as a `P_PEAK` claim when it is built. Keeping the +measurement and the actuation apart is the whole reason this file is small. + +WHAT WAS ALREADY BUILT, AND WHY THIS CARD IS NOT 8 SP OF NEW ARITHMETIC. +`p1.QuarterAverager` already computes AC 1 in full: clock-aligned 900 s blocks, +time-weighted over irregular telegram spacing, offtake-only, per-phase, with a +`max_hold_s` gap rule. It was unwired and unpersisted - `P1Ingest.blocks` is a +96-entry list in memory that a restart erases. So this file is the persistence +and the monthly reduction, not a second accumulator. Adding one would be adding +a second source of truth for the exact quantity the money depends on. + + BLOCKS, NOT A SLIDING WINDOW. The Description says "rolling"; AC 1 says + "15-min block average". They are different numbers and the utility bills the + second one. §9.1 calls the billed unit the "highest 15-minute average offtake", + and §9.1's own worked case - "one 4998 W quarter moved a monthly peak from + 4832 -> 5013 W" - is stated per QUARTER, an object a sliding window does not + have. "Rolling" here means CONTINUOUSLY UPDATED (the open block's partial + average is readable at every sample), not a trailing-15-minutes window. A + trailing window would report a peak the meter never bills, and it reports it + HIGHER than the real one: a load burst straddling :15 fills a trailing window + completely while contributing to two block averages at partial weight. Guarding + that would spend battery defending an interval nobody charges for. + + DELIVERED POWER, NOT INTENDED POWER. SAFETY-20 has an open question about + whether guards should grade `Decision.target` (pre-clamp intent) or `standing` + (what actually reached the wire). For a capacity tariff there is no question to + ask: the utility's meter integrates the current that flowed. This module never + sees a setpoint at all - it consumes `P1Sample`, which is the meter's own + reading - and that is on purpose. A peak computed from intent would be wrong in + the expensive direction (a clamped 9 kW request that delivered 5 kW would + record 9 kW, and every consumer would shave against a peak that never existed). + + AN UNREADABLE STORE IS NOT A ZERO PEAK. This is the defect this repo spent a + day removing from `sim/riglock.py`, where `holder()` was rewritten to answer + `observed / not_observed / gone` instead of a partial dict every caller + `.get()`s a default out of. The same trap is here and it is worse, because the + degenerate value is not merely wrong, it is DISARMING: a peak read as 0 W gives + SAFETY-07 the maximum possible headroom and silently switches the guard off at + the moment its own storage is broken. So `peak()` returns a `PeakReading` whose + `state` is one of OBSERVED / NO_RECORD / NOT_OBSERVED, `peak_w` is `None` + unless the state is OBSERVED, and there is no `.get(..., 0)` shaped accessor + anywhere on it. NO_RECORD ("this month has no completed block yet", a normal + state on the 1st) and NOT_OBSERVED ("the database did not answer") are + different types of thing and are never spelled the same way. + +CONFIG: NONE ADDED, ON PURPOSE. CFG-03 is open because the option defaults +already live in three places held apart by a tripwire in `test_validate.py`. The +soft limit this measurement is graded against already exists as `peak_cap_w` +(config.yaml, default 3500), already validated, already read by +`control.peak_at_risk()`. §14 names a `tariffs.capacity_peak_limit_kw: 2.5` - in +KILOwatts, against watts everywhere else - and adding it would have made a fourth +defaults site AND imported the unit mismatch §14's own warning box says must be +caught at startup. So this module takes its limit as a plain constructor +argument in WATTS and config.yaml is untouched by this ticket. + +THE DATABASE. SQLite in /data, the add-on's only durable volume, alongside +`store.py`'s state.json. Two tiers from §15, and their retention is the point: + + quarter_blocks one row per completed 900 s block INDEFINITE + monthly_peak one row per calendar month PERMANENT + +Neither is ever pruned here, and no sweep in this file can prune them - there is +no DELETE statement against either table anywhere in this module. §15's asymmetry +is why: raw samples vanish after 7 days, so a 15-minute aggregate that was not +written AS IT HAPPENED can never be recomputed. `monthly_peak` is maintained as a +derived record - upserted on every block close, and `recompute_month()` re-derives +it from `quarter_blocks` so the two can be PROVEN not to have diverged rather +than assumed not to have. + +RESTART MID-BLOCK. The open block's accumulator is memory. A restart at minute 12 +that dropped it would under-read the rest of the quarter, and SAFETY-07's clamp +would release exactly when it was most needed. So the accumulator is checkpointed +(`QuarterAverager.snapshot()`) into a single row, at most `CHECKPOINT_S` stale, +and restored on startup. Three cases, all of them exercised in `test_tariff.py`: + + * the checkpoint's block is the block containing `now` -> restore it verbatim, + including the held sample's ORIGINAL timestamp, so the first sample after the + restart integrates the gap under the same `max_hold_s` rule a running process + would have applied. A restart longer than `max_hold_s` therefore leaves + uncovered seconds, and a closed block is always divided by the full 900 s, so + the gap DRAGS THE BILLED AVERAGE DOWN instead of being averaged away; + * the checkpoint's block ended while we were down -> that block really happened + and its measured part is real, so it is closed at `acc / 900` and recorded, + then a fresh block is opened. Discarding it would lose measured energy; + * no checkpoint, or one that cannot be read -> start clean and SAY SO. The + return value of `restore()` names which of the three happened; it is never + silently the first one. + +Double counting is answered structurally rather than by care: `quarter_start` is +the PRIMARY KEY, and the recorded average may only ever be RAISED by a later +write (`ON CONFLICT ... WHERE excluded.offtake_avg_w > ...`). Replaying the same +block twice is therefore a no-op, and a restart that reprocesses a block with +LESS coverage than the first pass cannot lower a figure already banked. + +MONTH BOUNDARY. Calendar months are LOCAL (Europe/Brussels); blocks are stored in +UTC. In August those differ by two hours, so a block starting 2026-08-31 22:15 +UTC belongs to SEPTEMBER and putting it in August would move a month's bill. +Block alignment itself needs no timezone - every Belgian offset is a whole number +of hours, so the 900 s UTC grid lands on :00/:15/:30/:45 local on both sides of a +DST transition, which is the argument `QuarterAverager` already carries. +""" + +import json +import logging +import os +import sqlite3 +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone, tzinfo + +_LOG = logging.getLogger("goodwe.tariff") + +QUARTER_S = 900 + +# How stale the partial-block checkpoint may be. This is the ONLY thing a crash +# can lose from an open block, and the loss is bounded and one-directional: the +# lost seconds were already integrated into `_acc`, so dropping them makes the +# restored block read LOW, never high. At 15 s against a 900 s block the +# worst-case under-read of one quarter's average is 15/900 = 1.7 % of that +# quarter's own offtake - see test_tariff.py, which computes that bound from +# these two constants rather than quoting it. +CHECKPOINT_S = 15.0 + +# -- reading states. Three, and never fewer. -------------------------------- # +OBSERVED = "observed" # the store answered and there is a peak +NO_RECORD = "no_record" # the store answered and this month has none yet +NOT_OBSERVED = "not_observed" # the store did not answer. NOT a zero peak. + + +# --------------------------------------------------------------------------- # +# local time, without a hard dependency on the tz database +# --------------------------------------------------------------------------- # +class _EUTimezone(tzinfo): + """Europe/Brussels by the EU harmonised rule, for hosts with no tzdata. + + ⚠️ THIS IS A FALLBACK AND IT IS NOT SILENT. `zoneinfo` is tried first and is + what runs on the add-on's container. The fallback exists because the + alternative - quietly using UTC when tzdata is missing - misfiles every block + in the last two hours of a month into the wrong month's bill, and does it + without a symptom. Encoding the rule (CET, CEST from 01:00 UTC on the last + Sunday of March to 01:00 UTC on the last Sunday of October) is 20 lines and + is checked against zoneinfo, date by date, wherever zoneinfo is available. + """ + + # The rule, in both frames. `fromutc` is overridden rather than inherited + # because the base implementation calls back into `utcoffset` with a + # datetime whose tzinfo is self - and the first version of this class + # answered that by calling `astimezone`, which calls `utcoffset`, which + # recursed until the stack ran out. A tzinfo must answer from the value it + # is handed, never by converting it. + @staticmethod + def _last_sunday(year: int, month: int, hour: int) -> datetime: + d = datetime(year, month + 1, 1) - timedelta(days=1) + return (d - timedelta(days=(d.weekday() + 1) % 7)).replace(hour=hour) + + def _dst_local(self, naive: datetime) -> bool: + """From a LOCAL wall clock: 02:00 CET last Sunday March -> 03:00 CEST + last Sunday October. The one ambiguous hour each autumn resolves to + standard time, which cannot move a month boundary.""" + y = naive.year + return self._last_sunday(y, 3, 2) <= naive < self._last_sunday(y, 10, 3) + + def _dst_utc(self, naive_utc: datetime) -> bool: + """From UTC: both EU transitions happen at 01:00 UTC, everywhere.""" + y = naive_utc.year + return self._last_sunday(y, 3, 1) <= naive_utc < self._last_sunday(y, 10, 1) + + def fromutc(self, dt: datetime) -> datetime: + return dt + timedelta(hours=2 if self._dst_utc(dt.replace(tzinfo=None)) else 1) + + def utcoffset(self, dt): + if dt is None: + return timedelta(hours=1) + return timedelta(hours=2 if self._dst_local(dt.replace(tzinfo=None)) else 1) + + def dst(self, dt): + if dt is None: + return timedelta(0) + return timedelta(hours=1 if self._dst_local(dt.replace(tzinfo=None)) else 0) + + def tzname(self, dt): + return "CEST" if dt is not None and self._dst_local(dt.replace(tzinfo=None)) else "CET" + + +def _billing_tz(): + try: + from zoneinfo import ZoneInfo + tz = ZoneInfo("Europe/Brussels") + # Prove it resolves before trusting it: a ZoneInfo built from an empty + # tzdata answers, and answering wrongly is the failure being avoided. + if datetime(2026, 7, 1, tzinfo=timezone.utc).astimezone(tz).hour == 2: + return tz + _LOG.warning("zoneinfo Europe/Brussels resolved but gave the wrong " + "summer offset - falling back to the encoded EU rule") + except Exception as err: # noqa: BLE001 - any failure + _LOG.info("no usable tz database (%s) - using the encoded EU rule", err) + return _EUTimezone() + + +BILLING_TZ = _billing_tz() + + +def month_key(when: datetime) -> str: + """The calendar month a UTC instant is billed in, as 'YYYY-MM' LOCAL time.""" + if when.tzinfo is None: + when = when.replace(tzinfo=timezone.utc) + local = when.astimezone(BILLING_TZ) + return f"{local.year:04d}-{local.month:02d}" + + +# --------------------------------------------------------------------------- # +# the per-phase reduction policy - one function, deliberately +# --------------------------------------------------------------------------- # +# §9.1 requires the average be computed per-phase-aware, but does NOT say whether +# the BILLED figure on a three-phase connection is the sum of the phases or the +# worst single phase. §5.2 requires the per-phase data; nothing states the +# reduction. So it is one function with one argument, and correcting it later is +# a one-line change plus a test parameter rather than a change to the +# accumulator. §17's "Three-phase unbalanced load" scenario is the case that +# tells the three apart - on a balanced load all three agree, which is exactly +# why a balanced fixture must not be what this is tested on. +POLICY_NET = "net" # default: the connection's net offtake, what the meter bills +POLICY_SUM = "sum" # sum of the per-phase offtakes (differs when a phase exports) +POLICY_MAX = "max" # the worst single phase + + +def billed_w(block, policy: str = POLICY_NET) -> float: + """Reduce one closed block to the single figure the month is billed on.""" + pp = getattr(block, "per_phase_offtake_avg_w", None) + if policy == POLICY_NET or not pp: + return float(block.offtake_avg_w) + if policy == POLICY_SUM: + return float(sum(pp)) + if policy == POLICY_MAX: + return float(max(pp)) + raise ValueError(f"unknown per-phase reduction policy: {policy!r}") + + +# --------------------------------------------------------------------------- # +# what a read of the store answers +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class PeakReading: + """An answer about a month's peak, or an honest refusal to give one. + + ⚠️ There is no `peak_w_or(default)` and there must never be one. Every + accessor that lets a caller turn "the database did not answer" into a number + re-creates the defect this type exists to remove. Callers ask `is_observed` + first, and a caller that forgets gets `None` and a TypeError, which is loud. + """ + + state: str + month: str + peak_w: float | None = None + quarter_start: datetime | None = None + per_phase_w: tuple | None = None + detail: str = "" + + @property + def is_observed(self) -> bool: + return self.state == OBSERVED + + @property + def is_unreadable(self) -> bool: + return self.state == NOT_OBSERVED + + def explain(self) -> str: + if self.state == OBSERVED: + at = self.quarter_start.isoformat() if self.quarter_start else "?" + return f"{self.month}: peak {self.peak_w:.0f} W, set by the quarter at {at}" + if self.state == NO_RECORD: + return f"{self.month}: no completed quarter recorded yet" + return (f"{self.month}: PEAK UNREADABLE ({self.detail}) - this is NOT a " + f"zero peak and must not be used as headroom") + + +# --------------------------------------------------------------------------- # +# the store +# --------------------------------------------------------------------------- # +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS quarter_blocks ( + quarter_start TEXT PRIMARY KEY, -- ISO-8601 UTC, aligned to 900 s + month TEXT NOT NULL, -- 'YYYY-MM', LOCAL calendar month + offtake_avg_w REAL NOT NULL, + per_phase_json TEXT +); +CREATE INDEX IF NOT EXISTS ix_quarter_month ON quarter_blocks(month); +CREATE TABLE IF NOT EXISTS monthly_peak ( + month TEXT PRIMARY KEY, + peak_w REAL NOT NULL, + quarter_start TEXT NOT NULL, + per_phase_json TEXT +); +CREATE TABLE IF NOT EXISTS partial_block ( + id INTEGER PRIMARY KEY CHECK (id = 1), + state_json TEXT NOT NULL +); +""" + + +class PeakStore: + """SQLite behind /data. Every read answers with a state, never with a default.""" + + def __init__(self, path: str = "/data/tariff.db"): + self.path = path + self.ok = False + self.last_error: str | None = None + self._db: sqlite3.Connection | None = None + self._open() + + def _open(self) -> None: + parent = os.path.dirname(self.path) + # ⚠️ THE DIRECTORY IS NOT CREATED. /data is mounted by the Supervisor and + # always exists on a real install, so `makedirs` here could only ever + # succeed where the durable volume is MISSING - manufacturing a database + # on the container's ephemeral layer that looks durable, reads fine, and + # is thrown away on the next restart. That is a peak store that silently + # forgets, which is worse than one that says it cannot answer. + if parent and not os.path.isdir(parent): + self.last_error = f"no durable volume at {parent}" + # INFO, not ERROR: on a dev host or in a test container there is no + # /data and nothing is wrong. A store that OPENED and then broke is + # the error case, and it is logged as one below. + _LOG.info("no peak store: %s - the monthly peak will read " + "NOT_OBSERVED", self.last_error) + return + try: + self._db = sqlite3.connect(self.path, timeout=5.0, + check_same_thread=False) + # WAL: a power cut mid-write must not leave a torn database that + # reads as "no peak this month" - the exact shape of the defect. + self._db.execute("PRAGMA journal_mode=WAL") + self._db.executescript(_SCHEMA) + self._db.commit() + self.ok = True + self.last_error = None + except sqlite3.Error as err: + self._db = None + self.ok = False + self.last_error = str(err) + # ⚠️ Deliberately NOT raised. §6's fail-open rule: an aggregation + # fault must not stop a controller that is regulating a battery. The + # cost of continuing is a peak that reads NOT_OBSERVED, and every + # reader is built to treat that as "do not use me", not as 0 W. + _LOG.error("peak store unusable at %s (%s) - the monthly peak will " + "read NOT_OBSERVED, which is not a zero peak", self.path, err) + + def close(self) -> None: + if self._db is not None: + try: + self._db.close() + finally: + self._db = None + self.ok = False + + # -- writing ------------------------------------------------------------ + def record_block(self, block, policy: str = POLICY_NET) -> bool: + """Persist one completed block and fold it into its month's peak. + + Both tables move in ONE transaction. A crash between them would leave a + banked quarter that the monthly peak does not know about, and the peak is + the record the money is read off. + """ + if self._db is None: + return False + w = billed_w(block, policy) + start = block.start.astimezone(timezone.utc).isoformat() + month = month_key(block.start) + pp = getattr(block, "per_phase_offtake_avg_w", None) + pp_json = json.dumps(list(pp)) if pp else None + try: + with self._db: + # A block may only ever be RAISED. A restart that reprocesses a + # quarter it had already banked will have LESS coverage than the + # first pass (its first seconds are gone), so an unconditional + # REPLACE would silently lower a recorded figure. + self._db.execute( + "INSERT INTO quarter_blocks(quarter_start, month, offtake_avg_w, per_phase_json)" + " VALUES(?,?,?,?)" + " ON CONFLICT(quarter_start) DO UPDATE SET" + " offtake_avg_w=excluded.offtake_avg_w," + " per_phase_json=excluded.per_phase_json" + " WHERE excluded.offtake_avg_w > quarter_blocks.offtake_avg_w", + (start, month, w, pp_json)) + self._db.execute( + "INSERT INTO monthly_peak(month, peak_w, quarter_start, per_phase_json)" + " VALUES(?,?,?,?)" + " ON CONFLICT(month) DO UPDATE SET" + " peak_w=excluded.peak_w," + " quarter_start=excluded.quarter_start," + " per_phase_json=excluded.per_phase_json" + " WHERE excluded.peak_w > monthly_peak.peak_w", + (month, w, start, pp_json)) + return True + except sqlite3.Error as err: + self.last_error = str(err) + _LOG.error("could not record quarter %s (%s)", start, err) + return False + + # -- reading ------------------------------------------------------------ + def peak(self, when: datetime | None = None, month: str | None = None) -> PeakReading: + """The billed peak for a month. Three possible answers, never a default.""" + if month is None: + month = month_key(when or datetime.now(timezone.utc)) + if self._db is None: + return PeakReading(NOT_OBSERVED, month, + detail=self.last_error or "store never opened") + try: + row = self._db.execute( + "SELECT peak_w, quarter_start, per_phase_json FROM monthly_peak" + " WHERE month=?", (month,)).fetchone() + except sqlite3.Error as err: + self.last_error = str(err) + return PeakReading(NOT_OBSERVED, month, detail=str(err)) + if row is None: + return PeakReading(NO_RECORD, month) + pp = tuple(json.loads(row[2])) if row[2] else None + return PeakReading(OBSERVED, month, peak_w=float(row[0]), + quarter_start=datetime.fromisoformat(row[1]), + per_phase_w=pp) + + def recompute_month(self, month: str) -> PeakReading: + """Re-derive the peak from `quarter_blocks` - the divergence check. + + `monthly_peak` is a derived record, so it can be PROVEN equal to the + tier it derives from instead of trusted to be. Read-only. + """ + if self._db is None: + return PeakReading(NOT_OBSERVED, month, + detail=self.last_error or "store never opened") + try: + row = self._db.execute( + "SELECT offtake_avg_w, quarter_start, per_phase_json FROM quarter_blocks" + " WHERE month=? ORDER BY offtake_avg_w DESC, quarter_start ASC LIMIT 1", + (month,)).fetchone() + except sqlite3.Error as err: + self.last_error = str(err) + return PeakReading(NOT_OBSERVED, month, detail=str(err)) + if row is None: + return PeakReading(NO_RECORD, month) + pp = tuple(json.loads(row[2])) if row[2] else None + return PeakReading(OBSERVED, month, peak_w=float(row[0]), + quarter_start=datetime.fromisoformat(row[1]), + per_phase_w=pp) + + def blocks_in(self, month: str) -> list: + """(quarter_start, offtake_avg_w) for a month, oldest first. [] on error + is safe here ONLY because no caller reads emptiness as a peak - AC 3's + counter reports the sample size it actually got.""" + if self._db is None: + return [] + try: + return [(datetime.fromisoformat(a), float(b)) for a, b in self._db.execute( + "SELECT quarter_start, offtake_avg_w FROM quarter_blocks" + " WHERE month=? ORDER BY quarter_start", (month,))] + except sqlite3.Error as err: + self.last_error = str(err) + return [] + + # -- the open block's checkpoint ---------------------------------------- + def save_partial(self, state: dict) -> bool: + if self._db is None: + return False + try: + with self._db: + self._db.execute( + "INSERT INTO partial_block(id, state_json) VALUES(1,?)" + " ON CONFLICT(id) DO UPDATE SET state_json=excluded.state_json", + (json.dumps(state),)) + return True + except (sqlite3.Error, TypeError, ValueError) as err: + self.last_error = str(err) + return False + + def load_partial(self) -> dict | None: + if self._db is None: + return None + try: + row = self._db.execute( + "SELECT state_json FROM partial_block WHERE id=1").fetchone() + return json.loads(row[0]) if row else None + except (sqlite3.Error, ValueError) as err: + self.last_error = str(err) + return None + + def clear_partial(self) -> None: + # The one DELETE in this module, and it is against the checkpoint - a + # scratch row, not a retention tier. Neither quarter_blocks nor + # monthly_peak is deleted from anywhere in this file. + if self._db is None: + return + try: + with self._db: + self._db.execute("DELETE FROM partial_block WHERE id=1") + except sqlite3.Error as err: + self.last_error = str(err) + + +# --------------------------------------------------------------------------- # +# the tracker: what main.py holds +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class Projection: + """The open block's end-of-quarter estimate, with the reasons to distrust it. + + `elapsed_s` travels WITH the number rather than beside it, because early in a + quarter a projection is a 10-second extrapolation over 900 s and a consumer + that cannot see that will react to noise. `covered_frac` is separately useful: + it is below 1.0 when the meter had a gap, which is the "unreliable quarter" + the functional note asks be marked rather than reported low. + """ + + projected_w: float + elapsed_w: float + elapsed_s: float + covered_frac: float + complete: bool + block_start: datetime | None + + def explain(self) -> str: + return (f"quarter projects {self.projected_w:.0f} W " + f"({self.elapsed_s:.0f} s elapsed, {self.covered_frac * 100:.0f} % covered" + + ("" if self.complete else ", INCOMPLETE") + ")") + + +class PeakTracker: + """Drives the store from `P1Ingest`. Measures; never commands.""" + + def __init__(self, store: PeakStore, cap_w: float = 3500.0, + policy: str = POLICY_NET, checkpoint_s: float = CHECKPOINT_S): + self.store = store + self.cap_w = float(cap_w) + self.policy = policy + self.checkpoint_s = float(checkpoint_s) + self.recorded = 0 + self.restore_state = "not attempted" + self._seen: set = set() + self._last_ckpt = 0.0 + + # -- startup ------------------------------------------------------------ + def restore(self, averager, now: datetime | None = None) -> str: + """Put the open block back. Returns which of the three cases happened.""" + now = now or datetime.now(timezone.utc) + snap = self.store.load_partial() + if not snap: + self.restore_state = ("fresh: no checkpoint" if self.store.ok + else "fresh: store unreadable") + return self.restore_state + block = snap.get("block") + if block is None: + self.restore_state = "fresh: checkpoint had no open block" + return self.restore_state + + current = int(now.timestamp() // QUARTER_S) * QUARTER_S + if block == current: + if not averager.restore(snap): + # A reshaped accumulator (meter_phases changed) is a different + # measurement, not a resumable one. Say so; do not half-resume. + self.store.clear_partial() + self.restore_state = "fresh: checkpoint does not fit this averager" + return self.restore_state + self.restore_state = ( + f"resumed the open block at {snap.get('elapsed', 0.0):.0f} s " + f"of {QUARTER_S} s ({snap.get('acc', 0.0) / QUARTER_S:.0f} W banked)") + return self.restore_state + + # The block ended while the add-on was down. Its measured part really + # happened, so bank it at acc/900 rather than discard measured energy. + # It reads LOW because the seconds after the shutdown were never covered, + # and that is the honest figure: coverage a closed block did not get. + closed = _ClosedBlock( + start=datetime.fromtimestamp(block, timezone.utc), + offtake_avg_w=float(snap.get("acc", 0.0)) / QUARTER_S, + per_phase_offtake_avg_w=(tuple(v / QUARTER_S for v in snap["pp_acc"]) + if snap.get("pp_acc") and snap.get("last_pp") else None)) + banked = self.store.record_block(closed, self.policy) + self.store.clear_partial() + self.restore_state = ( + f"closed a stale block from {closed.start.isoformat()} at " + f"{closed.offtake_avg_w:.0f} W (partial coverage), " + + ("banked" if banked else "NOT banked - store unwritable")) + return self.restore_state + + # -- steady state ------------------------------------------------------- + def sync(self, ingest, now_mono: float | None = None) -> int: + """Bank any blocks that closed, and refresh the checkpoint. Returns the + number of blocks newly recorded on this call.""" + import time as _t + n = 0 + for block in getattr(ingest, "blocks", ()): + key = block.start.timestamp() + if key in self._seen: + continue + if self.store.record_block(block, self.policy): + self._seen.add(key) + self.recorded += 1 + n += 1 + if n: + # A block just closed, so the checkpoint's block is gone: clearing it + # in the same breath is what stops a crash here from replaying it. + self.store.clear_partial() + self._last_ckpt = 0.0 + self.checkpoint(ingest, now_mono or _t.monotonic()) + return n + + def checkpoint(self, ingest, now_mono: float, force: bool = False) -> bool: + if not force and (now_mono - self._last_ckpt) < self.checkpoint_s: + return False + avg = getattr(ingest, "averager", None) + if avg is None: + return False + self._last_ckpt = now_mono + return self.store.save_partial(avg.snapshot()) + + # -- reading ------------------------------------------------------------ + def peak(self, when: datetime | None = None) -> PeakReading: + return self.store.peak(when) + + def headroom_w(self, when: datetime | None = None) -> float | None: + """Watts left under the soft limit, or None when it cannot be known. + + ⚠️ `None`, not `self.cap_w`. An unreadable peak returning full headroom + is the disarming default this whole module is shaped against. + """ + r = self.peak(when) + if r.state == NO_RECORD: + return self.cap_w + if not r.is_observed: + return None + return max(0.0, self.cap_w - r.peak_w) + + def warning_frac(self, when: datetime | None = None) -> float | None: + """Peak as a fraction of the soft limit - §9.1's 80 % warning entity.""" + r = self.peak(when) + if r.state == NO_RECORD: + return 0.0 + if not r.is_observed or self.cap_w <= 0: + return None + return r.peak_w / self.cap_w + + @staticmethod + def project(averager, now: datetime) -> Projection: + """End-of-quarter average if the last sample held for the rest of it. + + ⚠️ The two divisors are DIFFERENT and only coincide at the end of the + block. `elapsed_w` divides by the seconds actually covered (what the + meter has shown so far); `projected_w` divides by the full 900 s and + credits the held value over what remains. At `elapsed_s == 900` and full + coverage they are equal - which is precisely why nothing may be tested + only at that point. + """ + start = averager.block_start + if start is None: + return Projection(0.0, 0.0, 0.0, 0.0, False, None) + elapsed_wall = max(0.0, (now - start).total_seconds()) + elapsed = averager.elapsed_s + acc = averager.partial_ws + remaining = max(0.0, QUARTER_S - elapsed_wall) + held = max(getattr(averager, "_last_net", 0.0), 0.0) + return Projection( + projected_w=(acc + held * remaining) / QUARTER_S, + elapsed_w=(acc / elapsed) if elapsed > 0 else 0.0, + elapsed_s=elapsed_wall, + covered_frac=(elapsed / elapsed_wall) if elapsed_wall > 0 else 0.0, + complete=elapsed_wall <= 0 or elapsed >= elapsed_wall - 1e-9, + block_start=start) + + +@dataclass(frozen=True) +class _ClosedBlock: + """A block reconstructed from a checkpoint. Shaped like `p1.QuarterBlock` so + `record_block` has one input type; kept private so nothing else makes one.""" + + start: datetime + offtake_avg_w: float + per_phase_offtake_avg_w: tuple | None diff --git a/goodwe_controller/scenarios/p1-net-2026-08-20.jsonl.gz b/goodwe_controller/scenarios/p1-net-2026-08-20.jsonl.gz new file mode 100644 index 0000000..1b2689e Binary files /dev/null and b/goodwe_controller/scenarios/p1-net-2026-08-20.jsonl.gz differ diff --git a/goodwe_controller/scenarios/p1-net-2026-08-23.jsonl.gz b/goodwe_controller/scenarios/p1-net-2026-08-23.jsonl.gz new file mode 100644 index 0000000..ee770cf Binary files /dev/null and b/goodwe_controller/scenarios/p1-net-2026-08-23.jsonl.gz differ diff --git a/goodwe_controller/test_tariff.py b/goodwe_controller/test_tariff.py new file mode 100644 index 0000000..709ed8a --- /dev/null +++ b/goodwe_controller/test_tariff.py @@ -0,0 +1,910 @@ +"""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-.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")