New panel design
This commit is contained in:
1 parent
face2432e4
commit
b5d6b4a897
30 files changed
+4540
-171
No files matched your search
Executable
+110
@@ -0,0 +1,110 @@
|
||||
#!/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())
|
||||
Reference in new issue
Block a user