thermostat-hw/tools/ha_snapshot.py

#!/usr/bin/env python3
"""
Capture Home Assistant responses to disk for the simulator to render.

Iterating on layout against live data is awkward: the interesting states --
rain falling, a light dimmed, a sensor dropping out -- do not happen on
demand, and live data changes underneath a before/after comparison. Snapshots
are reproducible and can be hand-edited to produce a state that has not
occurred yet.

    export HA_TOKEN=$(cat ~/.config/thermostat-hw/ha_token)
    python3 tools/ha_snapshot.py                  # -> sim/snapshots/live.json
    python3 tools/ha_snapshot.py --name rainy

    ./sim/build/thermostat_sim --data sim/snapshots/live.json \
                               --forecast sim/snapshots/live.forecast.json

The template is not duplicated here. It is read from the simulator binary via
--print-template, which is compiled from main/api/ha_template.h, so a snapshot
can never be captured with a template the firmware does not use.
"""

import argparse
import json
import os
import subprocess
import sys
import urllib.error
import urllib.request
from pathlib import Path

REPO = Path(__file__).resolve().parent.parent
SIM_BIN = REPO / "sim" / "build" / "thermostat_sim"
SNAP_DIR = REPO / "sim" / "snapshots"


def load_template() -> str:
    if not SIM_BIN.exists():
        sys.exit(f"{SIM_BIN} not built.\n"
                 "  cmake -S sim -B sim/build -G Ninja && cmake --build sim/build")
    out = subprocess.run([str(SIM_BIN), "--print-template"],
                         capture_output=True, text=True, check=True)
    return out.stdout.strip()


def call(host: str, token: str, path: str, body: dict | str):
    data = body.encode() if isinstance(body, str) else json.dumps(body).encode()
    req = urllib.request.Request(
        f"http://{host}:8123{path}",
        data=data,
        headers={"Authorization": f"Bearer {token}",
                 "Content-Type": "application/json"},
    )
    try:
        return urllib.request.urlopen(req, timeout=20).read().decode()
    except urllib.error.HTTPError as e:
        sys.exit(f"HTTP {e.code} on {path}: {e.read().decode()[:400]}")
    except urllib.error.URLError as e:
        sys.exit(f"Cannot reach Home Assistant at {host}:8123 -- {e.reason}")


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--host", default="192.168.1.134")
    ap.add_argument("--name", default="live", help="snapshot basename")
    ap.add_argument("--token-file",
                    default=str(Path.home() / ".config/thermostat-hw/ha_token"),
                    help="file holding the long-lived token; read directly, never printed")
    args = ap.parse_args()

    # Read the token from its file by default. Passing it through the
    # environment means it appears on a command line, and command lines end up
    # in shell history and terminal transcripts; reading the path here keeps
    # the value inside the process that consumes it.
    token = os.environ.get("HA_TOKEN")
    if not token:
        try:
            token = Path(args.token_file).read_text().strip()
        except OSError:
            sys.exit(f"No token: set HA_TOKEN or create {args.token_file}")

    SNAP_DIR.mkdir(parents=True, exist_ok=True)

    state = call(args.host, token, "/api/template", {"template": load_template()})
    # Parse before writing: a template that renders to broken JSON should fail
    # here rather than at the point some future run tries to render it.
    try:
        parsed = json.loads(state)
    except json.JSONDecodeError as e:
        sys.exit(f"Template rendered invalid JSON ({e}):\n{state[:500]}")

    state_path = SNAP_DIR / f"{args.name}.json"
    state_path.write_text(state)

    forecast = call(args.host, token,
                    "/api/services/weather/get_forecasts?return_response",
                    {"entity_id": "weather.forecast_home", "type": "daily"})
    fc_path = SNAP_DIR / f"{args.name}.forecast.json"
    fc_path.write_text(forecast)

    print(f"wrote {state_path.relative_to(REPO)} ({len(state)} bytes)")
    print(f"wrote {fc_path.relative_to(REPO)} ({len(forecast)} bytes)")

    dead = [k for k, v in parsed.items() if v is None]
    if dead:
        print(f"\nunavailable sensors in this snapshot: {', '.join(dead)}")

    lights = parsed.get("lights", [])
    print(f"lights: {sum(1 for l in lights if l.get('s') == 'on')} on, "
          f"{sum(1 for l in lights if l.get('s') == 'off')} off, "
          f"{sum(1 for l in lights if l.get('s') not in ('on', 'off'))} unavailable")

    print(f"\n  ./sim/build/thermostat_sim --data {state_path.relative_to(REPO)} \\")
    print(f"                             --forecast {fc_path.relative_to(REPO)}")
    return 0


if __name__ == "__main__":
    sys.exit(main())