Stream analysis

const WS_URL = "wss://api.interhuman.ai/v2/stream/analyze";
// A short-lived client token minted by your server with POST /v1/client_tokens.
// Never send your API key to the browser.
const clientToken = "YOUR_CLIENT_TOKEN";

const ws = new WebSocket(WS_URL, ["access_token", clientToken]);
ws.binaryType = "arraybuffer";

ws.addEventListener("open", () => {
  const sessionConfig = {
    include: [
      "conversation_quality_overall",
      "conversation_quality_timeline",
    ],
  };
  ws.send(JSON.stringify(sessionConfig));
});

ws.addEventListener("message", (event) => {
  if (typeof event.data !== "string") return;
  const payload = JSON.parse(event.data);
  console.log(payload.type, payload);
});

import asyncio
import json
import os

import websockets

WS_URL = "wss://api.interhuman.ai/v2/stream/analyze"


async def connect():
    api_key = os.environ["API_KEY"]
    headers = {"Authorization": f"Bearer {api_key}"}

    ws = await websockets.connect(
        WS_URL,
        additional_headers=headers,
        max_size=None,
    )

    session_config = {
        "include": [
            "conversation_quality_overall",
            "conversation_quality_timeline",
        ],
    }
    await ws.send(json.dumps(session_config))
    return ws


# Call connect() from the async main() in step 3.

const mediaStream = await navigator.mediaDevices.getUserMedia({
  video: true,
  audio: true,
});

const preview = document.querySelector("#preview");
preview.srcObject = mediaStream;
preview.play();

import platform

# ffmpeg input arguments for the default camera and microphone.
# Install ffmpeg first (for example `brew install ffmpeg` or `apt install ffmpeg`).
system = platform.system()
if system == "Darwin":
    CAPTURE_INPUT = ["-f", "avfoundation", "-framerate", "30", "-i", "0:0"]
elif system == "Linux":
    CAPTURE_INPUT = [
        "-f", "v4l2", "-i", "/dev/video0",
        "-f", "pulse", "-i", "default",
    ]
else:
    # Windows: list device names with
    #   ffmpeg -list_devices true -f dshow -i dummy
    CAPTURE_INPUT = ["-f", "dshow", "-i", "video=YOUR_CAMERA:audio=YOUR_MICROPHONE"]

const SEGMENT_MS = 3000;

const mimeType =
  ["video/webm;codecs=vp9,opus", "video/webm;codecs=vp8,opus", "video/webm"].find(
    (m) => MediaRecorder.isTypeSupported(m)
  ) || "";

const recorder = new MediaRecorder(
  mediaStream,
  mimeType ? { mimeType } : undefined
);

// Keep this handler synchronous: WebSocket.send() accepts the Blob as is and
// queues frames in order, so session.close can't overtake the last blob.
recorder.addEventListener("dataavailable", (event) => {
  if (ws.readyState !== WebSocket.OPEN) return;
  if (event.data && event.data.size > 0) {
    ws.send(event.data);
  }
  // After recorder.stop(), the last blob arrives with the recorder inactive:
  // end the session only once that blob has been sent.
  if (recorder.state === "inactive") {
    ws.send(JSON.stringify({ type: "session.close" }));
  }
});

// Call once the WebSocket is open and session config is sent:
recorder.start(SEGMENT_MS);

import asyncio
import json

FRAGMENT_SECONDS = 3


async def read_box(stdout):
    # Read one top-level MP4 box: a 4-byte size, a 4-byte type, then the body.
    header = await stdout.readexactly(8)
    size = int.from_bytes(header[:4], "big")
    return header[4:8], header + await stdout.readexactly(size - 8)


async def stream_camera(ws, stop: asyncio.Event) -> None:
    # One long-lived ffmpeg writes camera and microphone as fragmented MP4
    # (H.264 video + AAC audio) to stdout.
    proc = await asyncio.create_subprocess_exec(
        "ffmpeg", "-loglevel", "error",
        *CAPTURE_INPUT,
        "-c:v", "libx264", "-preset", "veryfast", "-tune", "zerolatency",
        "-pix_fmt", "yuv420p", "-g", str(30 * FRAGMENT_SECONDS),
        "-c:a", "aac",
        "-f", "mp4", "-movflags", "frag_keyframe+empty_moov+default_base_moof",
        "-frag_duration", str(FRAGMENT_SECONDS * 1_000_000),
        "pipe:1",
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
    )

    async def stop_when_asked():
        await stop.wait()
        proc.stdin.write(b"q")  # ffmpeg flushes the last fragment and exits
        await proc.stdin.drain()

    stopper = asyncio.create_task(stop_when_asked())
    pending = b""
    try:
        while True:
            box_type, box = await read_box(proc.stdout)
            pending += box
            # Send the header (ftyp + moov) as the first frame, then one
            # frame per fragment (moof + mdat).
            if box_type in (b"moov", b"mdat"):
                await ws.send(pending)
                pending = b""
    except asyncio.IncompleteReadError:
        pass  # ffmpeg exited: every fragment has been sent
    stopper.cancel()
    await proc.wait()


async def receive_events(ws) -> None:
    # Events are not one-per-segment: read them in their own loop.
    async for msg in ws:
        print(msg)


async def main() -> None:
    ws = await connect()
    receiver = asyncio.create_task(receive_events(ws))

    stop = asyncio.Event()
    # Call stop.set() when the user stops; this example stops after 30 seconds.
    asyncio.get_running_loop().call_later(30, stop.set)
    await stream_camera(ws, stop)

    # End the session (see below), then wait for the server to finish.
    await ws.send(json.dumps({"type": "session.close"}))
    await receiver


asyncio.run(main())

// The dataavailable handler above sends session.close after the last blob.
if (recorder && recorder.state !== "inactive") {
  recorder.stop();
}

// The server closes the socket after session.ended.
ws.addEventListener("close", () => {
  mediaStream.getTracks().forEach((track) => track.stop());
});

# In main() above: stop.set() stops ffmpeg, stream_camera() returns once the
# last fragment is sent, and only then is session.close sent.
await ws.send(json.dumps({"type": "session.close"}))
await receiver  # receive_events returns when the server closes after session.ended