mrcharles
6c5efbb609
Also absorbs PFVCluster operator tooling: tools/ha-ws-call.py and docs/reference-packs/ (historical, NOT loaded by HA — clearly marked). No live-config changes in this commit. Detail: https://projects.knownelement.com/issues/762
54 lines
1.8 KiB
Python
54 lines
1.8 KiB
Python
#!/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())
|