Streaming over LAN

import asyncio
import json
from urllib.parse import urlencode

import requests
import websockets  # pip install websockets

API_HEADERS = {
    "x-auth-scheme": "api-token",
    "x-auth-apikey": "YOUR_API_KEY",
    "Content-Type": "application/json",
}
CAMERA_UUID = "YOUR_CAMERA_UUID"


def mint_device_token(device_uuid: str) -> str:
    response = requests.post(
        "https://api2.rhombussystems.com/api/org/generateFederatedSessionToken",
        headers=API_HEADERS,
        json={"durationSec": 3600, "deviceUUid": device_uuid},
    )
    response.raise_for_status()
    return response.json()["federatedSessionToken"]


def lan_h264_uri(camera_uuid: str) -> str:
    response = requests.post(
        "https://api2.rhombussystems.com/api/camera/getMediaUris",
        headers=API_HEADERS,
        json={"cameraUuid": camera_uuid},
    )
    response.raise_for_status()
    uris = response.json().get("lanLiveH264Uris") or []
    if not uris:
        raise SystemExit("The camera reported no LAN address")
    return uris[0]


async def watch(uri: str, token: str, max_frames: int = 30) -> None:
    auth = urlencode({"x-auth-scheme": "federated-token", "x-auth-ft": token})
    async with websockets.connect(f"{uri}?{auth}") as ws:
        frames = 0
        async for message in ws:
            if isinstance(message, str):  # one text init message comes first
                print("init:", json.loads(message))
                continue
            timestamp_ms, offset = None, 0
            while offset + 4 <= len(message):
                tlv_type = message[offset]
                length = int.from_bytes(message[offset + 1 : offset + 4], "big")
                value = message[offset + 4 : offset + 4 + length]
                offset += 4 + length
                if tlv_type == 0x02:
                    timestamp_ms = int.from_bytes(value, "big")
                elif tlv_type in (0x00, 0x01):  # frame data is always last
                    kind = "keyframe" if tlv_type == 0x00 else "delta"
                    print(f"{kind}: {len(value)} bytes at {timestamp_ms}")
                    frames += 1
                    break
            if frames >= max_frames:
                break


asyncio.run(watch(lan_h264_uri(CAMERA_UUID), mint_device_token(CAMERA_UUID)))

// federatedToken comes from your server (Step 1), never your API key
const url = new URL(lanLiveH264Uri);
url.searchParams.set('x-auth-scheme', 'federated-token');
url.searchParams.set('x-auth-ft', federatedToken);

const ws = new WebSocket(url);
ws.binaryType = 'arraybuffer';
ws.onmessage = (event) => {
  if (typeof event.data === 'string') {
    console.log('init', JSON.parse(event.data));
    return;
  }
  handleH264Message(event.data); // see the TLV format linked below
};