goodwe-addon/goodwe_controller/tools/replay_ac3.py

#!/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())