feat(ha): websocket driver tool; Govee integration live end-to-end [#620]

💘 Generated with Crush

Assisted-by: Crush:glm-5.2
This commit is contained in:
2026-09-02 12:59:36 -05:00
parent a2ff60908e
commit 53b847985e
2 changed files with 65 additions and 4 deletions
+53
View File
@@ -0,0 +1,53 @@
#!/usr/bin/env python3
"""ha-ws-call: send websocket commands to Home Assistant, print JSON replies.
Env: HASS_URL (http(s)://host[:port]), HASS_TOKEN (long-lived token).
Args: one or more JSON command objects — argv literals or file paths.
Replies printed one-per-command, JSON indented. Stdlib + websockets only.
Usage example (docker, from repo root):
docker run --rm --network host --env-file ~/.creds/homeassistant.env \
-v "$PWD/HomeAssistant:/w" python:3.12-alpine \
sh -c 'pip install -q websockets && python /w/ha-ws-call.py "{...json...}"'
"""
import asyncio
import json
import os
import sys
try:
import websockets
except ImportError:
sys.exit("missing dependency: pip install websockets")
def load_cmd(arg):
if arg.startswith("{"):
return json.loads(arg)
with open(arg, encoding="latin1") as handle:
return json.load(handle)
async def main():
base = os.environ["HASS_URL"]
url = base.replace("http", "ws", 1).rstrip("/") + "/api/websocket"
token = os.environ["HASS_TOKEN"]
cmds = [load_cmd(arg) for arg in sys.argv[1:]]
async with websockets.connect(url, max_size=16 * 1024 * 1024) as ws:
hello = json.loads(await ws.recv())
if hello.get("type") != "auth_required":
sys.exit(f"unexpected hello: {hello}")
await ws.send(json.dumps({"type": "auth", "access_token": token}))
authed = json.loads(await ws.recv())
if authed.get("type") != "auth_ok":
sys.exit(f"auth failed: {authed}")
for message_id, cmd in enumerate(cmds, start=1):
cmd = dict(cmd)
cmd["id"] = message_id
await ws.send(json.dumps(cmd))
reply = json.loads(await ws.recv())
print(json.dumps(reply, indent=1))
if __name__ == "__main__":
asyncio.run(main())