glenn schrooyenandClaude Opus 5.5
7c4c8f5c66
New option peak_margin_w (default 50 W, 0-500 W). The peak guard, the
forecast checks and the maintenance charge sizing aim at
peak_cap_w - peak_margin_w; the month peak, the bill, headroom, warnings
and every notice keep the real cap. Owner decision 2026-09-28.
peak_margin_w 0 is byte-identical to ed739d4 (tools/safety79_identity.py,
test_peak_margin.py part 1). validate.py rule 12 refuses a value outside
0-500, not whole, or not below the peak limit.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UBtqPrG1mDLXzTEBnhEkkV
304 lines
13 KiB
Python
304 lines
13 KiB
Python
#!/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, margin_w=0.0, 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())
|