#!/usr/bin/env python3 """Run TARIFF-01's and SAFETY-07's AC 3 over a directory of `p1-net-*.jsonl.gz`. AC 3, both cards, same words: *"monthly peak held under the soft limit in >= 95 % of intervals over a 30-day simulation replay."* Both cards merged with it UNMET because two days of capture existed. DATA-01 exported thirty. This is the thing that turns those thirty days into the number. cd goodwe_controller python3 tools/replay_ac3.py scenarios/p1-net-30d-2026-08 WHAT IT MEASURES ================ Two runs over the same telegrams, one file at a time in date order into a SHARED peak store, so the 15-minute blocks accumulate into a real MONTH rather than into 30 disconnected days: UNCONTROLLED the recorded meter as it happened. Nothing was shaving peaks on this house during August 2026, so this is not a simulation of anything - it is what the month did. It is TARIFF-01's AC 3. GUARDED the same month with `peakguard.decide()` in the loop and the battery it asks for modelled, closed-loop: the meter the guard sees is the recorded net MINUS the discharge it obtained. That is SAFETY-07's AC 3. Both numbers are `quarters <= peak_cap_w` over `quarters closed`, with the cap read out of config.yaml rather than typed. Nothing here exits non-zero on 95 %: a measurement below the criterion is a finding about the house and the guard, and this tool's job is to report it, not to be graded by it. FAITHFULNESS TO THE SUITE ========================= The replay loop is `test_peakguard.py` part 6's, with `pack=None` (SoC pinned above the floor, so SAFETY-24's energy floor is not the thing deciding the answer). It is not imported from there because that file runs its whole suite on import. The duplication is checked instead of asserted: pointed at `scenarios/` - the two committed days - this tool must reproduce part 6's published totals, and `--expect` makes that a hard check. python3 tools/replay_ac3.py scenarios --expect 189/190 GAPS ==== Nothing is interpolated. A hole in the export is a hole here: the shipped `meter_max_age_s` is what decides whether a held reading still counts as coverage, so an hour with no telegrams closes no quarters and contributes to neither numerator nor denominator. `--report-gaps` lists them. """ from __future__ import annotations import argparse import gzip import json import sys import tempfile from datetime import datetime, timedelta from pathlib import Path HERE = Path(__file__).resolve().parent.parent sys.path.insert(0, str(HERE)) from app.p1 import P1Ingest, make_sample, split_signed # noqa: E402 from app.peakguard import decide # noqa: E402 from app.tariff import PeakStore, PeakTracker, month_key # noqa: E402 def read_option(key: str): """The shipped default for `key`, out of config.yaml. ⚠️ `peak_cap_w:` appears TWICE in config.yaml - once under `options:` with the default, once under `schema:` as `int(500,15000)`. The FIRST match is the options block, which is the number an install actually lands on; taking the last would try to float() the schema line. Same rule as the suites. """ for line in (HERE / "config.yaml").read_text(encoding="utf-8").splitlines(): if line.strip().startswith(f"{key}:"): try: return float(line.split(":", 1)[1].strip()) except ValueError: continue return None def read_schema_max(key: str): for line in (HERE / "config.yaml").read_text(encoding="utf-8").splitlines(): s = line.strip() if s.startswith(f"{key}:") and "(" in s and ")" in s: try: return float(s[s.index("(") + 1:s.index(")")].split(",")[-1]) except ValueError: continue return None def sample(t: datetime, net_w: float): """One telegram through the SHIPPED sign derivation - the suites' helper.""" imp, exp = split_signed(net_w) return make_sample("ha_signed", imp, exp, phases=1, ingest_ts=t, ingest_mono=t.timestamp()) def rows(path): """Every sample in one file, in order, with the gaps left as gaps.""" with gzip.open(path, "rt", encoding="utf-8") as fh: for line in fh: line = line.strip() if not line: continue row = json.loads(line) yield datetime.fromisoformat(row["t"]), float(row["w"]) def replay(paths, *, guarded: bool, cap_w, max_w, max_hold_s, cycle_s, soc, soc_floor, tmp: Path, gap_s: float): """test_peakguard.py part 6's loop, over an arbitrary number of days. ⚠️ ONE STORE, AND AN INGEST THAT IS RESET ONLY AT A REAL DISCONTINUITY. Neither half of that is a detail. The STORE is shared across the files because "monthly peak" is what AC 3 is about and a per-file store cannot express it. The INGEST is carried across a file boundary when the next file's first telegram follows the previous file's last one within `gap_s`, and rebuilt when it does not. Both failure modes are real and were measured here: * carried across a genuine hole, one averager manufactures quarters out of nothing - streamed straight through the three-day gap between the two committed days this reported 285/286 where the suite reports 189/190; * rebuilt at every file, each day's LAST quarter (21:45-22:00 UTC) never closes, because it closes only when a telegram past the boundary arrives and that telegram is in the next file. That is why the suite's two days give 95 quarters each rather than 96, and it silently discards one quarter per day - 30 of them over a month. So contiguous days are joined and non-contiguous ones are not, which gives 189/190 on the committed pair and a full 96 quarters a day on a run of consecutive days. """ store = PeakStore(str(tmp / f"ac3-{'g' if guarded else 'u'}.db")) tracker = PeakTracker(store, cap_w=cap_w) engaged = blind = saturated = cycles = n = joins = 0 wh = 0.0 gaps = [] months = set() ing = None batt = 0.0 # delivered discharge, W (positive) next_cycle = None prev_t = None for path in paths: for i, (t, w) in enumerate(rows(path)): if i == 0: contiguous = (prev_t is not None and 0 <= (t - prev_t).total_seconds() <= gap_s) if contiguous: joins += 1 else: if ing is not None: months |= {month_key(b.start) for b in ing.blocks} ing = P1Ingest(phases=1, max_age_s=max_hold_s) batt = 0.0 next_cycle = None prev_t = None if prev_t is not None: dt = (t - prev_t).total_seconds() if dt > gap_s: gaps.append((prev_t, t, dt)) wh += batt * dt / 3600.0 prev_t = t n += 1 ing.submit(sample(t, w - batt)) tracker.sync(ing) if next_cycle is not None and t < next_cycle: continue next_cycle = t + timedelta(seconds=cycle_s) cycles += 1 d = decide(PeakTracker.project(ing.averager, t), batt_w=batt, cap_w=cap_w, reading=tracker.peak(t), achievable_w=max_w, max_hold_s=max_hold_s, soc=soc, soc_floor=soc_floor, frozen=False) if d.engaged: blind += 1 if d.blind else 0 engaged += 0 if d.blind else 1 saturated += 1 if d.saturated else 0 if not guarded: continue engaged_floor = d.engaged and d.floor_w is not None batt = max(0.0, d.floor_w) if engaged_floor else 0.0 if ing is not None: months |= {month_key(b.start) for b in ing.blocks} out = {m: dict(store.blocks_in(m)) for m in sorted(months)} return {"months": out, "samples": n, "cycles": cycles, "joins": joins, "engaged": engaged, "blind": blind, "saturated": saturated, "wh": wh, "gaps": gaps} def main(argv=None): p = argparse.ArgumentParser(description=__doc__.split("\n")[0]) p.add_argument("directory", help="a directory of p1-net-*.jsonl.gz") p.add_argument("--glob", default="p1-net-*.jsonl.gz") p.add_argument("--expect", help="UNDER/TOTAL the uncontrolled run must " "reproduce, e.g. 189/190") p.add_argument("--report-gaps", action="store_true") args = p.parse_args(argv) paths = sorted(Path(args.directory).glob(args.glob)) if not paths: print(f"no {args.glob} under {args.directory}", file=sys.stderr) return 2 CAP_W = read_option("peak_cap_w") MAX_HOLD_S = read_option("meter_max_age_s") CYCLE_S = read_option("cycle_floor_s") SOC_FLOOR = read_option("peak_soc_floor") LIVE_MAX_W = read_schema_max("max_w") if None in (CAP_W, MAX_HOLD_S, CYCLE_S, SOC_FLOOR, LIVE_MAX_W): print("could not read the shipped options out of config.yaml", file=sys.stderr) return 2 # The suites' fixture SoC: comfortably above the floor, so SAFETY-24's # energy floor is not what decides an AC 3 that is not about it. SOC_OK = SOC_FLOOR + 30.0 days = len(paths) span_first = span_last = None print(f"AC 3 - {days} file(s) from {args.directory}") print(f" shipped: peak_cap_w {CAP_W:.0f} W, meter_max_age_s " f"{MAX_HOLD_S:.0f} s, cycle_floor_s {CYCLE_S:.0f} s, " f"max_w schema ceiling {LIVE_MAX_W:.0f} W") tmp = Path(tempfile.mkdtemp(prefix="goodwe-ac3-")) kw = dict(cap_w=CAP_W, max_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, cycle_s=CYCLE_S, soc=SOC_OK, soc_floor=SOC_FLOOR, tmp=tmp, gap_s=MAX_HOLD_S) u = replay(paths, guarded=False, **kw) g = replay(paths, guarded=True, **kw) def flat(run): return [w for m in run["months"].values() for w in m.values()] uq, gq = flat(u), flat(g) u_under = sum(1 for w in uq if w <= CAP_W) g_under = sum(1 for w in gq if w <= CAP_W) u_pct = 100.0 * u_under / max(len(uq), 1) g_pct = 100.0 * g_under / max(len(gq), 1) print(f" {u['samples']} samples, {len(uq)} closed quarters " f"({len(uq) / 96.0:.1f} days' worth), {days - u['joins']} contiguous " f"run(s) of days") for m in sorted(u["months"]): mu = u["months"][m] mg = g["months"].get(m, {}) print(f" month {m}: {len(mu)} quarters, peak " f"{max(mu.values()):.1f} W uncontrolled -> " f"{max(mg.values()) if mg else 0.0:.1f} W guarded") print() print(f" TARIFF-01 AC 3 (UNCONTROLLED, what the month did): " f"{u_under}/{len(uq)} = {u_pct:.2f} % of quarters at or under " f"{CAP_W:.0f} W") print(f" SAFETY-07 AC 3 (GUARDED, peakguard.decide in the loop): " f"{g_under}/{len(gq)} = {g_pct:.2f} %") print(f" guard cycles {g['cycles']}, engaged on the measurement " f"{g['engaged']}, blind {g['blind']}, saturated {g['saturated']}, " f"modelled discharge {g['wh']:.0f} Wh") # AC 3'S WORDS SUPPORT TWO READINGS AND BOTH ARE REPORTED, because they # disagree on this data and picking the flattering one is the failure mode # this whole card exists to correct. u_peak = max(uq) if uq else 0.0 g_peak = max(gq) if gq else 0.0 print(f" reading A - PERCENTAGE OF INTERVALS under the cap (the reading " f"test_tariff.py part 9 already uses): uncontrolled {u_pct:.2f} %, " f"guarded {g_pct:.2f} %") print(f" reading B - THE MONTHLY PEAK ITSELF under the cap: " f"{u_peak:.1f} W uncontrolled -> {g_peak:.1f} W guarded, against a " f"{CAP_W:.0f} W soft limit " f"({'under' if g_peak <= CAP_W else 'STILL OVER'})") print(f" verdict at the 95 % criterion: uncontrolled " f"{'MEETS' if u_pct >= 95.0 else 'DOES NOT MEET'}, guarded " f"{'MEETS' if g_pct >= 95.0 else 'DOES NOT MEET'} " f"({len(uq) / 96.0:.1f} days of replay)") if u["gaps"]: tot = sum(d for _, _, d in u["gaps"]) print(f" gaps over {MAX_HOLD_S:.0f} s: {len(u['gaps'])}, " f"{tot:.0f} s total - NOT interpolated; the quarters they fall in " f"simply carry less coverage") if args.report_gaps: for a, b, d in u["gaps"]: print(f" {a.isoformat()} -> {b.isoformat()} {d:.0f} s") if args.expect: want_u, want_t = (int(x) for x in args.expect.split("/")) ok = (u_under, len(uq)) == (want_u, want_t) print(f" reproduces test_peakguard.py part 6's published " f"{want_u}/{want_t}: {'yes' if ok else 'NO'} " f"(got {u_under}/{len(uq)})") return 0 if ok else 1 return 0 if __name__ == "__main__": raise SystemExit(main())