111 lines
4.2 KiB
Python
Executable File
111 lines
4.2 KiB
Python
Executable File
#!/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")
|
|
args = ap.parse_args()
|
|
|
|
token = os.environ.get("HA_TOKEN")
|
|
if not token:
|
|
sys.exit("HA_TOKEN is not set. See tools/ha_discover.py for how to make one.")
|
|
|
|
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())
|