Add a standard dashboard build/install/monitor-logs tool

Replaces ad-hoc /tmp WS scripts (each its own permission prompt) with one
checked-in CLI: get-config, set-model (patches the model: line in place),
install (compile+OTA, follows the job, confirms deployed_config_hash
actually changed rather than trusting the job result alone), logs, and
run (set-model + install + logs in one call).

tools/gxepd2_dashboard/README.md

@@ -0,0 +1,21 @@
+# gxepd2_dashboard
+
+Standard build/install/monitor-logs tool for the ESPHome dashboard
+(esp.sea.zi3.fi), replacing ad-hoc `/tmp` WS scripts written per test run.
+See `claude/ESPHOME_DASHBOARD_API.md` for the protocol this wraps.
+
+```
+uv run --with websockets python3 dashboard.py run --model GxEPD2_290c_GDEY029F51
+uv run --with websockets python3 dashboard.py install
+uv run --with websockets python3 dashboard.py logs --seconds 40
+uv run --with websockets python3 dashboard.py set-model GxEPD2_583_GDEQ0583T31
+uv run --with websockets python3 dashboard.py get-config
+```
+
+`--config` (default `gxepd2-test.yaml`) selects which dashboard device
+config to act on. `run` is set-model (if `--model` given) + install + logs
+in one call - the common "switch panel, flash, watch what it does" case.
+
+`install` confirms via `GET /devices` that `deployed_config_hash` actually
+changed, not just that the job reported `exit_code: 0` - a completed
+install job has been observed to precede the hash update by a beat.

tools/gxepd2_dashboard/dashboard.py

@@ -0,0 +1,220 @@
+#!/usr/bin/env python3
+"""Drive the ESPHome dashboard (esp.sea.zi3.fi) for one device: patch its
+model:, compile+install, and stream its boot/update logs - as one command
+instead of ad-hoc WS scripts written into /tmp per run (each of which was
+its own permission prompt). See claude/ESPHOME_DASHBOARD_API.md for the
+underlying protocol this wraps.
+
+Usage:
+    uv run --with websockets python3 dashboard.py run --model GxEPD2_290c_GDEY029F51
+    uv run --with websockets python3 dashboard.py install
+    uv run --with websockets python3 dashboard.py logs --seconds 40
+    uv run --with websockets python3 dashboard.py set-model GxEPD2_583_GDEQ0583T31
+    uv run --with websockets python3 dashboard.py get-config
+
+`run` = set-model (if --model given) + install + logs, the common case of
+"switch panel, flash, watch what it does" in one call.
+"""
+
+import argparse
+import asyncio
+import json
+import re
+import sys
+import time
+import urllib.request
+
+import websockets
+
+WS_URL = "wss://esp.sea.zi3.fi/ws"
+DEVICES_URL = "https://esp.sea.zi3.fi/devices"
+DEFAULT_CONFIG = "gxepd2-test.yaml"
+
+# The dashboard sends ANSI color codes as the literal 4-character text
+# "\033" (backslash, 0, 3, 3) followed by "[...m", not real ESC (0x1B)
+# bytes - so this matches that literal text, and separately real ESC bytes
+# too in case a future endpoint sends those instead.
+_LITERAL_ANSI_RE = re.compile(r"\\033\[[0-9;]*m")
+_REAL_ANSI_RE = re.compile("\033\\[[0-9;]*m")
+
+
+def clean(s: str) -> str:
+    """Strip both forms of ANSI color code and trailing newlines - for
+    readable terminal output instead of raw escape junk."""
+    s = _LITERAL_ANSI_RE.sub("", s)
+    s = _REAL_ANSI_RE.sub("", s)
+    return s.rstrip("\n")
+
+
+async def _connect():
+    ws = await websockets.connect(WS_URL, max_size=None)
+    await ws.recv()  # hello frame
+    return ws
+
+
+async def _send(ws, msg_id, command, args=None):
+    payload = {"command": command, "message_id": str(msg_id)}
+    if args:
+        payload["args"] = args
+    await ws.send(json.dumps(payload))
+    while True:
+        data = json.loads(await ws.recv())
+        if str(data.get("message_id")) == str(msg_id):
+            return data
+
+
+async def get_config(configuration: str) -> str:
+    async with await _connect() as ws:
+        res = await _send(ws, 1, "devices/get_config", {"configuration": configuration})
+        return res["result"]
+
+
+async def set_model(configuration: str, model: str) -> str:
+    """Rewrites the `model:` line under `display:` in place, keeping every
+    other line (lambda, pins, comments, ...) untouched."""
+    content = await get_config(configuration)
+    out = []
+    changed = False
+    for line in content.splitlines():
+        stripped = line.strip()
+        if stripped.startswith("model:"):
+            indent = line[: len(line) - len(line.lstrip())]
+            out.append(f"{indent}model: {model}")
+            changed = True
+        else:
+            out.append(line)
+    if not changed:
+        raise SystemExit(f"No `model:` line found in {configuration} - not touching it blind.")
+    new_content = "\n".join(out) + "\n"
+    async with await _connect() as ws:
+        await _send(ws, 1, "devices/update_config", {"configuration": configuration, "content": new_content})
+    return new_content
+
+
+def device_status(configuration: str) -> dict:
+    with urllib.request.urlopen(DEVICES_URL, timeout=15) as resp:
+        data = json.load(resp)
+    for dev in data.get("configured", []):
+        if dev.get("configuration") == configuration:
+            return dev
+    raise SystemExit(f"No configured device found for {configuration}")
+
+
+async def install(configuration: str, timeout: float = 480) -> bool:
+    """Triggers firmware/install, follows the job to completion, and
+    confirms via GET /devices that deployed_config_hash actually updated -
+    a completed job with exit_code 0 has still been observed to precede a
+    same-hash re-flash by a beat (see esphome_build_info_signal memory),
+    so poll briefly rather than trusting the job result alone."""
+    async with await _connect() as ws:
+        res = await _send(ws, 1, "firmware/install", {"configuration": configuration, "port": "OTA"})
+        job = res["result"]
+        job_id = job["job_id"]
+        print(f"install job {job_id} started ({job['job_type']})", flush=True)
+
+        await ws.send(json.dumps({"command": "firmware/follow_job", "message_id": "2", "args": {"job_id": job_id}}))
+        deadline = time.monotonic() + timeout
+        result = None
+        while time.monotonic() < deadline:
+            data = json.loads(await asyncio.wait_for(ws.recv(), timeout=deadline - time.monotonic()))
+            if str(data.get("message_id")) != "2":
+                continue
+            ev = data.get("event")
+            if ev == "line":
+                d = data.get("data")
+                if isinstance(d, str):
+                    line = clean(d)
+                    if line:
+                        print(line, flush=True)
+            elif ev == "result":
+                result = data.get("data")
+                break
+        if result is None:
+            print("install job timed out following output", flush=True)
+            return False
+        print(f"job result: {result}", flush=True)
+        if result.get("exit_code") != 0:
+            return False
+
+    expected = device_status(configuration).get("expected_config_hash")
+    for _ in range(10):
+        dev = device_status(configuration)
+        if dev.get("deployed_config_hash") == expected:
+            print(f"deployed_config_hash confirmed: {expected}", flush=True)
+            return True
+        await asyncio.sleep(1)
+    print(f"WARNING: deployed_config_hash never matched expected ({expected}) after install", flush=True)
+    return False
+
+
+async def stream_logs(configuration: str, seconds: float):
+    async with await _connect() as ws:
+        await ws.send(json.dumps({"command": "devices/logs", "message_id": "3",
+                                   "args": {"configuration": configuration, "port": "OTA"}}))
+        end = time.monotonic() + seconds
+        try:
+            while True:
+                remaining = end - time.monotonic()
+                if remaining <= 0:
+                    break
+                data = json.loads(await asyncio.wait_for(ws.recv(), timeout=remaining))
+                if str(data.get("message_id")) != "3":
+                    continue
+                d = data.get("data")
+                if isinstance(d, str):
+                    line = clean(d)
+                    if line:
+                        print(line, flush=True)
+        except asyncio.TimeoutError:
+            pass
+
+
+async def main_async(args):
+    if args.cmd == "get-config":
+        print(await get_config(args.config))
+    elif args.cmd == "set-model":
+        new_content = await set_model(args.config, args.model)
+        print(f"model set to {args.model}\n---\n{new_content}")
+    elif args.cmd == "install":
+        ok = await install(args.config, timeout=args.timeout)
+        sys.exit(0 if ok else 1)
+    elif args.cmd == "logs":
+        await stream_logs(args.config, args.seconds)
+    elif args.cmd == "run":
+        if args.model:
+            await set_model(args.config, args.model)
+            print(f"model set to {args.model}", flush=True)
+        ok = await install(args.config, timeout=args.timeout)
+        if not ok:
+            print("install did not succeed - skipping log stream", flush=True)
+            sys.exit(1)
+        await stream_logs(args.config, args.seconds)
+
+
+def main():
+    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    p.add_argument("--config", default=DEFAULT_CONFIG, help=f"dashboard configuration filename (default: {DEFAULT_CONFIG})")
+    sub = p.add_subparsers(dest="cmd", required=True)
+
+    sub.add_parser("get-config", help="print the raw YAML source")
+
+    sm = sub.add_parser("set-model", help="patch the display platform's model: line")
+    sm.add_argument("model")
+
+    inst = sub.add_parser("install", help="compile+OTA install, following the job to completion")
+    inst.add_argument("--timeout", type=float, default=480)
+
+    lg = sub.add_parser("logs", help="stream device logs for a fixed duration")
+    lg.add_argument("--seconds", type=float, default=35)
+
+    run = sub.add_parser("run", help="set-model (optional) + install + logs")
+    run.add_argument("--model", default=None)
+    run.add_argument("--timeout", type=float, default=480)
+    run.add_argument("--seconds", type=float, default=35)
+
+    args = p.parse_args()
+    asyncio.run(main_async(args))
+
+
+if __name__ == "__main__":
+    main()