Streaming

from vlmrun.client import VLMRun
from vlmrun.common.video import VideoReader

client = VLMRun(api_key="<VLMRUN_API_KEY>")

QUESTIONS = [{"id": "has_people", "type": "noul", "instructions": "Are people visible?"}]

with client.gateway.systemone.stream(
    questions=QUESTIONS,
    state="Answer about the frame.",
    model="google/diffusiongemma-26b-a4b-it",
    detail="low",
    concurrency=8,          # requested as the session's max_inflight
    transport="ws",         # one socket; "http" sends a request per frame
) as stream, VideoReader("clip.mp4") as video:
    for decision in stream.map(video.frames(fps=1)):
        print(decision.timestamp_s, decision.response.nouls["has_people"].noul)

vlmrun gw s1 clip.mp4 --fps 1 --ws -c 8 \
  -m google/diffusiongemma-26b-a4b-it --detail low \
  --noul has_people="Are people visible?"

import asyncio, json, pathlib
import websockets

URL = "wss://gateway.vlm.run/typesafe/ws"

async def main(frames: list[bytes]) -> None:
    async with websockets.connect(
        URL, additional_headers={"Authorization": "Bearer <VLMRUN_API_KEY>"}
    ) as ws:
        await ws.send(json.dumps({
            "type": "session.create",
            "model": "google/diffusiongemma-26b-a4b-it",
            "state": "Answer about the frame.",
            "questions": {
                "has_people": {"type": "noul", "instructions": "Are people visible?"}
            },
            "detail": "low",
        }))
        created = json.loads(await ws.recv())
        print(created["session_id"], created["max_frame_bytes"])

        for frame in frames:          # raw JPEG/PNG/WebP/GIF bytes, no base64
            await ws.send(frame)

        seen = 0
        while seen < len(frames):
            event = json.loads(await ws.recv())
            if event["type"] == "decision":
                print(event["frame"], event["answers"]["has_people"]["noul"])
                seen += 1
            elif event["type"] == "frame.dropped":
                seen += 1             # arrived with every read slot busy
            elif event["type"] == "error":
                raise RuntimeError(event["message"])

        await ws.send(json.dumps({"type": "session.close"}))
        print(json.loads(await ws.recv()))   # session.stats

asyncio.run(main([pathlib.Path("frame1.jpg").read_bytes()]))

import WebSocket from "ws";
import { readFileSync } from "node:fs";

const URL = "wss://gateway.vlm.run/typesafe/ws";
const frames = [readFileSync("frame1.jpg")];   // raw bytes, no base64

const ws = new WebSocket(URL, {
  headers: { Authorization: `Bearer ${process.env.VLMRUN_API_KEY}` },
});

let seen = 0;

ws.on("open", () => {
  ws.send(JSON.stringify({
    type: "session.create",
    model: "google/diffusiongemma-26b-a4b-it",
    state: "Answer about the frame.",
    questions: {
      has_people: { type: "noul", instructions: "Are people visible?" },
    },
    detail: "low",
  }));
});

ws.on("message", (data) => {
  const event = JSON.parse(data.toString());

  if (event.type === "session.created") {
    console.log(event.session_id, event.max_frame_bytes);
    for (const frame of frames) ws.send(frame);
    return;
  }
  if (event.type === "decision") {
    console.log(event.frame, event.answers.has_people.noul);
    seen++;
  } else if (event.type === "frame.dropped") {
    seen++;                                    // every read slot was busy
  } else if (event.type === "error") {
    throw new Error(event.message);
  } else if (event.type === "session.stats") {
    console.log(event);
    ws.close();
    return;
  }

  if (seen === frames.length) ws.send(JSON.stringify({ type: "session.close" }));
});