docs: showpiece README — mermaid architecture, sensor/alert tables [#762]
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
This commit is contained in:
@@ -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())
|
||||
Reference in New Issue
Block a user