48 lines
1.7 KiB
Python
48 lines
1.7 KiB
Python
import json
|
|
import pathlib
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
from websockets.sync.client import connect
|
|
|
|
token = pathlib.Path("/run/secrets/api-token").read_text().strip()
|
|
headers = {"Authorization": "Bearer " + token}
|
|
|
|
|
|
def get(path):
|
|
req = urllib.request.Request("http://127.0.0.1:18120" + path, headers=headers)
|
|
with urllib.request.urlopen(req, timeout=10) as response:
|
|
return json.load(response)
|
|
|
|
|
|
health = get("/v1/health")
|
|
capabilities = get("/v1/capabilities")
|
|
messages = []
|
|
ws_error = None
|
|
try:
|
|
with connect("ws://127.0.0.1:18120/v1/poses", additional_headers=headers, open_timeout=10) as ws:
|
|
messages = [json.loads(ws.recv(timeout=20)) for _ in range(10)]
|
|
except Exception as exc:
|
|
ws_error = type(exc).__name__ + ": " + str(exc)[:240]
|
|
|
|
frame_ids = [m["frame_id"] for m in messages]
|
|
summary = {
|
|
"tested_at_unix_ns": time.time_ns(),
|
|
"health": health,
|
|
"capabilities": capabilities,
|
|
"websocket": {
|
|
"messages": len(messages),
|
|
"error": ws_error,
|
|
"frame_ids": frame_ids,
|
|
"strictly_increasing": all(b > a for a, b in zip(frame_ids, frame_ids[1:])),
|
|
"source_alive_all": all(m["source_alive"] for m in messages),
|
|
"people_counts": [len(m["people"]) for m in messages],
|
|
"coordinate_space_all": all(m["frame"]["coordinate_space"] == "ndi_buffer_unmodified" for m in messages),
|
|
"horizontal_flip_all_false": all(not m["frame"]["horizontal_flip_applied"] for m in messages),
|
|
"source_to_sent_ms": [round(m["timing"]["source_to_sent_ms"], 3) for m in messages],
|
|
"sample_messages": messages[:2],
|
|
},
|
|
}
|
|
print(json.dumps(summary, ensure_ascii=False, indent=2))
|