goodwe-addon/goodwe_controller/tools/export_p1_history.py

#!/usr/bin/env python3
"""Export a P1 power signal out of Home Assistant's recorder into the committed
scenario format - one gzipped JSON object per line, one file per LOCAL day.

WHY THIS FILE EXISTS AT ALL
===========================
`scenarios/p1-net-2026-08-20.jsonl.gz` and `p1-net-2026-08-23.jsonl.gz` were
produced once, by hand, and the script was thrown away. The consequence was not
theoretical: TARIFF-01's and SAFETY-07's AC 3 ("monthly peak held under the soft
limit in >= 95 % of intervals over a 30-day simulation replay") were both merged
UNMET on the belief that thirty days could not be produced retroactively - a
belief that was simply wrong, and that survived because nobody could re-run the
export to find out. So the export is a committed, runnable artefact now, and the
next person does not have to re-derive it under time pressure while a retention
window rolls.

    python3 tools/export_p1_history.py --days 30 --out-dir /some/where

THE TWO RULES THIS FILE IS BUILT AROUND
=======================================
1. A FAILED READ IS NOT AN OBSERVATION. An HTTP error, a timeout, a truncated
   body, a non-JSON body - none of those are "the meter reported nothing that
   hour". They are "we do not know what the meter reported that hour", and the
   two must never end up looking the same in the output. So:

     * every request is retried with backoff;
     * a day whose read ultimately fails writes NO FILE and is recorded in the
       manifest as `"read": "FAILED"` with the error text, and the process exits
       non-zero;
     * a day that reads back successfully but EMPTY writes an empty file and is
       recorded as `"read": "OK", "samples": 0`. Empty-and-known and
       unknown are therefore distinguishable both in the manifest and by the
       presence or absence of the file;
     * before any day is fetched, the entity is proved to exist through
       /api/states/<entity>. Without that check a typo in --entity would return
       200 + `[]` for all thirty days and produce a beautifully consistent
       dataset of nothing.

2. GAPS ARE VISIBLE, NEVER INTERPOLATED. Nothing here carries a value forward,
   fills, resamples or averages. A recorder outage or an HA restart shows up as
   what it is: consecutive `t` values further apart than the meter's cadence.
   The controller's own ingest already reads that correctly - `meter_max_age_s`
   (30 s shipped) is exactly the age past which a held reading stops counting as
   coverage - so an hour with no telegrams replays as an hour of NO COVERAGE
   rather than as an hour of whatever the last sample said. The manifest lists
   every gap over --gap-s so a reader does not have to diff timestamps to find
   them.

DAY AT A TIME, AND WHY NOT SMALLER
==================================
One request per local day. That is gentle on a live HA (~1 MB of JSON per day)
and it is also the only chunking that reproduces the committed files: the
history API prepends the state as it stood AT the period start
(`include_start_time_state`), stamped with the start instant. Splitting a day
into six-hour chunks would therefore inject three synthetic boundary samples per
day that no meter ever sent. Days are the natural boundary because the committed
files are local days.

VERIFYING THE EXPORTER
======================
    python3 tools/export_p1_history.py --verify scenarios/p1-net-2026-08-20.jsonl.gz

re-fetches that day by this exact route and compares the DECOMPRESSED bytes with
the committed file. Byte equality of the .gz containers is not expected and not
required: a gzip header carries an mtime, and this file writes mtime 0 so its
own output is reproducible. Content equality is the claim that matters.
"""

from __future__ import annotations

import argparse
import gzip
import hashlib
import json
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timedelta, timezone
from pathlib import Path

try:                                    # py3.9+; the add-on image has it
    from zoneinfo import ZoneInfo
except ImportError:                     # pragma: no cover
    ZoneInfo = None                     # type: ignore

DEFAULT_URL = "http://192.168.2.158:8123"
DEFAULT_ENTITY = "sensor.p1_meter_active_power"
DEFAULT_TZ = "Europe/Brussels"
# The shipped meter_max_age_s. A hole longer than this is a hole the controller
# itself would refuse to see through, which makes it the honest gap threshold.
DEFAULT_GAP_S = 30.0


class ReadFailed(RuntimeError):
    """The read did not happen. NOT the same thing as an empty result."""


# ---------------------------------------------------------------------------
# transport


def _get(url: str, token: str, timeout: float):
    req = urllib.request.Request(url, headers={
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
    })
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        if resp.status != 200:
            raise ReadFailed(f"HTTP {resp.status}")
        return resp.read()


def fetch_json(url: str, token: str, *, timeout: float, retries: int,
               backoff: float, log=print):
    """GET and parse, or raise ReadFailed. Never returns a partial answer."""
    last = None
    for attempt in range(1, retries + 1):
        try:
            raw = _get(url, token, timeout)
        except urllib.error.HTTPError as exc:
            last = f"HTTP {exc.code} {exc.reason}"
        except Exception as exc:                     # URLError, timeout, ...
            last = f"{type(exc).__name__}: {exc}"
        else:
            try:
                return json.loads(raw.decode("utf-8")), len(raw), attempt
            except (UnicodeDecodeError, json.JSONDecodeError) as exc:
                # A truncated body looks exactly like this. It is a failure.
                last = f"unparseable body ({len(raw)} bytes): {exc}"
        if attempt < retries:
            wait = backoff * attempt
            log(f"      read failed ({last}); retry {attempt + 1}/{retries} "
                f"in {wait:.0f} s")
            time.sleep(wait)
    raise ReadFailed(last or "unknown")


def assert_entity_exists(base: str, entity: str, token: str, *, timeout: float,
                         retries: int, backoff: float, log=print):
    """Prove the entity is real BEFORE trusting any empty day.

    Without this, `--entity sensor.typo` answers 200 + [] for every day and the
    exporter cheerfully writes thirty empty files.
    """
    url = f"{base}/api/states/{urllib.parse.quote(entity)}"
    doc, _, _ = fetch_json(url, token, timeout=timeout, retries=retries,
                           backoff=backoff, log=log)
    if not isinstance(doc, dict) or doc.get("entity_id") != entity:
        raise ReadFailed(f"{entity} is not a known entity")
    return doc


# ---------------------------------------------------------------------------
# one day


def day_bounds(day: datetime, tz):
    start = datetime(day.year, day.month, day.day, tzinfo=tz)
    return start, start + timedelta(days=1)


def history_url(base: str, entity: str, start: datetime, end: datetime) -> str:
    q = urllib.parse.urlencode({
        "filter_entity_id": entity,
        "end_time": end.isoformat(),
        "minimal_response": "",
    })
    return (f"{base}/api/history/period/"
            f"{urllib.parse.quote(start.isoformat())}?{q}")


def rows_from(doc, entity: str):
    """HA's shape -> [(iso_t, state_str)], plus the states that are not numbers.

    `minimal_response` returns the first record in full and the rest as
    {state, last_changed}. An entity that went unavailable reports the string
    "unavailable" (or "unknown") where a number belongs. Those are dropped from
    the numeric series - there is no float to write and the committed schema has
    no slot for a non-number - but every one is returned here so it lands in the
    manifest. Dropping them is what MAKES the outage visible: the series simply
    has no samples across it, which is the same thing the meter falling silent
    looks like, and is what the controller's freshness rule already handles.
    """
    if not isinstance(doc, list):
        raise ReadFailed(f"expected a list, got {type(doc).__name__}")
    if not doc:
        return [], []
    series = doc[0]
    if not isinstance(series, list):
        raise ReadFailed("history series is not a list")
    good, bad = [], []
    for rec in series:
        eid = rec.get("entity_id")
        if eid is not None and eid != entity:
            raise ReadFailed(f"unexpected entity {eid!r} in the response")
        t = rec.get("last_changed") or rec.get("last_updated")
        s = rec.get("state")
        if t is None:
            raise ReadFailed("a record carries no timestamp")
        try:
            w = float(s)
        except (TypeError, ValueError):
            bad.append({"t": t, "state": s})
            continue
        good.append((t, w))
    return good, bad


def render(rows, *, newline: str) -> bytes:
    """The committed format, exactly: `{"t": <iso>, "w": <float>}` per line.

    json.dumps' default separators (", " / ": ") and repr of a float are what
    the two committed files contain; so is the CRLF, an artefact of the original
    one-off having been run on Windows in text mode. It is reproduced rather
    than tidied because reproducing the committed bytes is this exporter's
    verification, and gzip.open(..., 'rt') strips it either way.
    """
    out = bytearray()
    for t, w in rows:
        out += json.dumps({"t": t, "w": w}).encode("utf-8")
        out += newline.encode("ascii")
    return bytes(out)


def gz(payload: bytes) -> bytes:
    """gzip level 9, mtime 0 - so the same input always gives the same bytes."""
    import io
    buf = io.BytesIO()
    with gzip.GzipFile(fileobj=buf, mode="wb", compresslevel=9, mtime=0) as f:
        f.write(payload)
    return buf.getvalue()


def gaps_in(rows, gap_s: float):
    out = []
    prev = None
    for t, _ in rows:
        cur = datetime.fromisoformat(t)
        if prev is not None:
            d = (cur - prev).total_seconds()
            if d > gap_s:
                out.append({"from": prev.isoformat(), "to": cur.isoformat(),
                            "seconds": round(d, 3)})
        prev = cur
    return out


def export_day(base, entity, token, day, tz, *, timeout, retries, backoff,
               gap_s, newline, log=print):
    """Read one local day. Raises ReadFailed rather than returning a guess."""
    start, end = day_bounds(day, tz)
    doc, nbytes, attempts = fetch_json(
        history_url(base, entity, start, end), token,
        timeout=timeout, retries=retries, backoff=backoff, log=log)
    rows, bad = rows_from(doc, entity)
    entry = {
        "date": day.date().isoformat(),
        "window": [start.isoformat(), end.isoformat()],
        "read": "OK",
        "http_bytes": nbytes,
        "attempts": attempts,
        "samples": len(rows),
        "non_numeric_states": bad,
        "first": rows[0][0] if rows else None,
        "last": rows[-1][0] if rows else None,
        "gaps": gaps_in(rows, gap_s),
    }
    if rows:
        span = (datetime.fromisoformat(rows[-1][0])
                - datetime.fromisoformat(rows[0][0])).total_seconds()
        entry["covered_span_s"] = round(span, 1)
        entry["mean_cadence_s"] = round(span / max(1, len(rows) - 1), 3)
    return rows, entry


# ---------------------------------------------------------------------------
# modes


def do_export(args, token, tz, log=print):
    base = args.url.rstrip("/")
    out_dir = Path(args.out_dir)
    out_dir.mkdir(parents=True, exist_ok=True)
    newline = "\r\n" if args.newline == "crlf" else "\n"

    log(f"    proving {args.entity} exists before trusting any empty day")
    st = assert_entity_exists(base, args.entity, token, timeout=args.timeout,
                              retries=args.retries, backoff=args.backoff,
                              log=log)
    log(f"      {args.entity} = {st.get('state')} "
        f"{st.get('attributes', {}).get('unit_of_measurement', '')}")

    if args.end_date:
        last = datetime.fromisoformat(args.end_date).replace(tzinfo=tz)
    else:                                # yesterday: today is still being logged
        last = datetime.now(tz).replace(hour=0, minute=0, second=0,
                                        microsecond=0) - timedelta(days=1)
    days = [last - timedelta(days=i) for i in range(args.days - 1, -1, -1)]

    manifest = {
        "generated": datetime.now(timezone.utc).isoformat(),
        "generator": "tools/export_p1_history.py",
        "source": f"{base}/api/history/period",
        "entity": args.entity,
        "timezone": args.tz,
        "gap_threshold_s": args.gap_s,
        "format": '{"t": <ISO-8601 UTC>, "w": <float watts>} per line, gzip',
        "days": [],
    }
    failures = []
    for day in days:
        name = f"{args.prefix}{day.date().isoformat()}.jsonl.gz"
        log(f"    {name}")
        try:
            rows, entry = export_day(
                base, args.entity, token, day, tz, timeout=args.timeout,
                retries=args.retries, backoff=args.backoff, gap_s=args.gap_s,
                newline=newline, log=log)
        except ReadFailed as exc:
            # NO FILE. A missing file is the honest record of a read that did
            # not happen; an empty one would claim the meter was silent.
            entry = {"date": day.date().isoformat(), "read": "FAILED",
                     "error": str(exc), "file": None}
            manifest["days"].append(entry)
            failures.append(entry)
            log(f"      READ FAILED: {exc} - no file written")
            continue
        payload = render(rows, newline=newline)
        blob = gz(payload)
        (out_dir / name).write_bytes(blob)
        entry["file"] = name
        entry["bytes_gz"] = len(blob)
        entry["bytes_raw"] = len(payload)
        entry["sha256_raw"] = hashlib.sha256(payload).hexdigest()
        entry["sha256_gz"] = hashlib.sha256(blob).hexdigest()
        manifest["days"].append(entry)
        ng = len(entry.get("gaps", []))
        log(f"      {entry['samples']} samples, {len(blob)} B gz, "
            f"{ng} gap(s) over {args.gap_s:g} s"
            + (f", {len(entry['non_numeric_states'])} non-numeric state(s)"
               if entry["non_numeric_states"] else ""))
        if day is not days[-1]:
            time.sleep(args.delay)

    ok = [d for d in manifest["days"] if d["read"] == "OK"]
    manifest["summary"] = {
        "days_requested": len(days),
        "days_read": len(ok),
        "days_failed": len(failures),
        "samples": sum(d["samples"] for d in ok),
        "bytes_gz": sum(d["bytes_gz"] for d in ok),
        "empty_days": [d["date"] for d in ok if d["samples"] == 0],
    }
    # newline="\n" explicitly: write_text() on Windows emits CRLF, and
    # the manifest would then differ from a Linux run of the same export
    # for no reason visible in a diff.
    with open(args.manifest or (out_dir / "manifest.json"), "w",
              encoding="utf-8", newline="\n") as fh:
        fh.write(json.dumps(manifest, indent=2) + "\n")
    log(f"    {manifest['summary']['days_read']}/{len(days)} days, "
        f"{manifest['summary']['samples']} samples, "
        f"{manifest['summary']['bytes_gz'] / 1e6:.2f} MB gz")
    if failures:
        log(f"    {len(failures)} DAY(S) NOT READ - the dataset is incomplete "
            f"and says so: " + ", ".join(d["date"] for d in failures))
        return 1
    return 0


def do_verify(args, token, tz, log=print):
    """Re-export a committed day by this route and diff it against the file."""
    base = args.url.rstrip("/")
    newline = "\r\n" if args.newline == "crlf" else "\n"
    rc = 0
    for target in args.verify:
        path = Path(target)
        stamp = path.name.split(".")[0].rsplit("-", 3)[-3:]
        day = datetime(int(stamp[0]), int(stamp[1]), int(stamp[2]), tzinfo=tz)
        log(f"    {path.name}  (local day {day.date().isoformat()})")
        committed = gzip.decompress(path.read_bytes())
        rows, entry = export_day(base, args.entity, token, day, tz,
                                 timeout=args.timeout, retries=args.retries,
                                 backoff=args.backoff, gap_s=args.gap_s,
                                 newline=newline, log=log)
        fresh = render(rows, newline=newline)
        if fresh == committed:
            log(f"      IDENTICAL: {len(fresh)} bytes, {entry['samples']} "
                f"samples, sha256 {hashlib.sha256(fresh).hexdigest()[:16]}")
            continue
        rc = 1
        cl = committed.split(newline.encode())
        fl = fresh.split(newline.encode())
        log(f"      DIFFERS: committed {len(committed)} B / {len(cl)} lines, "
            f"re-export {len(fresh)} B / {len(fl)} lines")
        shown = 0
        for i in range(max(len(cl), len(fl))):
            a = cl[i] if i < len(cl) else b"<absent>"
            b = fl[i] if i < len(fl) else b"<absent>"
            if a != b:
                log(f"        line {i}: committed {a!r}")
                log(f"                  re-export {b!r}")
                shown += 1
                if shown >= args.max_diff:
                    log(f"        ... stopping after {shown} differing lines")
                    break
    return rc


def main(argv=None):
    p = argparse.ArgumentParser(description=__doc__.split("\n")[0])
    p.add_argument("--url", default=DEFAULT_URL)
    p.add_argument("--entity", default=DEFAULT_ENTITY)
    p.add_argument("--token", help="bearer token; else $HASS_TOKEN")
    p.add_argument("--tz", default=DEFAULT_TZ,
                   help="local timezone that defines a day boundary")
    p.add_argument("--days", type=int, default=30)
    p.add_argument("--end-date", help="last local day to export (YYYY-MM-DD); "
                                      "default yesterday")
    p.add_argument("--out-dir", default=".")
    p.add_argument("--prefix", default="p1-net-")
    p.add_argument("--manifest", help="path for the manifest JSON")
    p.add_argument("--gap-s", type=float, default=DEFAULT_GAP_S)
    p.add_argument("--newline", choices=("crlf", "lf"), default="crlf",
                   help="line terminator; crlf reproduces the committed files")
    p.add_argument("--timeout", type=float, default=120.0)
    p.add_argument("--retries", type=int, default=4)
    p.add_argument("--backoff", type=float, default=5.0)
    p.add_argument("--delay", type=float, default=2.0,
                   help="pause between days - this is a live HA")
    p.add_argument("--verify", nargs="+", metavar="FILE",
                   help="re-export these committed days and diff them")
    p.add_argument("--max-diff", type=int, default=20)
    args = p.parse_args(argv)

    token = args.token or __import__("os").environ.get("HASS_TOKEN")
    if not token:
        print("no token: pass --token or set HASS_TOKEN", file=sys.stderr)
        return 2
    if ZoneInfo is None:
        print("zoneinfo unavailable", file=sys.stderr)
        return 2
    tz = ZoneInfo(args.tz)
    try:
        if args.verify:
            return do_verify(args, token, tz)
        return do_export(args, token, tz)
    except ReadFailed as exc:
        print(f"read failed: {exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())