Trazio by CerebrixOS

← All guides

Robot and camera video

The situation. A camera covers the forklift lane in a warehouse. You want to be told when a person steps into a robot's operating zone, and to find the exact moment afterwards without scrubbing through hours of footage.

Set the five values first. For robot state without video, see Robotics telemetry.

How it works

Video has three separate lanes, so a slow one never blocks a fast one:

1. Live safety analysis. Frames are analysed in order; the result is a plain, factual detection (for example, "person" and "forklift" in the same frame). The attached rules decide whether it becomes an alert. The model never raises an alert by itself.
2. Scene captions for the viewer. A caption on screen is not a confirmed notification.
3. Durable recording, searchable in plain language afterwards.

If the model's output is inconclusive or contradictory, no alert and no receipt is created.

1. What the pipeline needs

In the guided setup, choose Camera and video or Live video and alerts. Your pipeline is created straight away. Live analysis, the live video channel and scene processing start as approval requests, and staff switch them on once the infrastructure is ready. Until then you can't send live frames. For a pipeline managed by the Trazio team, they enable these for you. Ask them if a step fails.

• Live analysis with an approved model profile (pinned to your pipeline).
• The Rule Packs below attached.
• For recordings: artifact_storage.enabled=true, regional media processing and vector_search.enabled=true.
• Temporal analysis (1, 4 or 8 frames per decision) is chosen by the operator per pipeline. More frames give motion context but cost more inference time.

Your TENANT_JWT needs live:analyze and event:ingest for live frames.

2. Send a live frame

Python

import os
from trazio_telemetry import Client

client = Client(
    base_url=os.environ["PLATFORM_BASE"],
    tenant_id=os.environ["TENANT_ID"],
    pipeline_id=os.environ["PIPELINE_ID"],
    tenant_jwt=os.environ["TENANT_JWT"],
)

detection = client.analyze_live_frame(
    "./frame.jpg",
    stream_session_id="camera-session-17",
    frame_id="frame-42",
    device_id="robot-17",
)
print(detection)

Node.js

import { Client } from "trazio-telemetry";

const client = new Client({
  baseUrl: process.env.PLATFORM_BASE,
  tenantId: process.env.TENANT_ID,
  pipelineId: process.env.PIPELINE_ID,
  tenantJwt: process.env.TENANT_JWT,
});

const detection = await client.analyzeLiveFrame("./frame.jpg", {
  streamSessionId: "camera-session-17",
  frameId: "frame-42",
  deviceId: "robot-17",
});
console.log(detection);

Use a new frame_id for every frame and keep stream_session_id the same for one camera session. The response is a factual detection result. Whether it becomes an alert is decided by the rules in step 4.

3. Upload a recording

Keep recording locally in .partial or .active segments. Close each segment, then upload it with platform-edge (resumable) or the artifact API.

session = client.upload_artifact("./segment-0001.mp4", format="auto",
                                 device_id="robot-17", goal_id="inspect-loading-bay")

Then poll the session until media and vector processing complete (the polling loop is in Robotics telemetry, step 4). Relay success is never archive success: always confirm the recording is complete.

4. Rules that fire

live-generic-object-screening is a demonstration pack. It proves that inference, rules and notifications work end to end ("person observed in a sampled frame"). It is not a certified safety policy.

Before production, test with negative footage from before an incident, positive incident windows, invalid model responses and notification timing.

5. Find the moment

Assistant. Ask:

• "Show the moments a person got close to a forklift."
• "Open the video for the emergency stop on robot 17."
• "Was anyone near the robot in the minute before it stopped?"

curl. A natural-language search within one recording:

export VIDEO_SESSION_ID='<completed-video-session-id>'

curl --fail-with-body -sS -X POST \
  "$PLATFORM_BASE/v1/artifacts/$TENANT_ID/$PIPELINE_ID/search" \
  -H "Authorization: Bearer $TENANT_JWT" \
  -H 'Content-Type: application/json' \
  -d "{\"query\":\"worker near a moving forklift\",\"kind\":\"scene\",\"session_id\":\"$VIDEO_SESSION_ID\",\"top_k\":10}" \
  | jq '{count,matches:[.matches[]? | {score,caption:.metadata.caption,objects:.metadata.objects,start_seconds:.metadata.start_seconds,end_seconds:.metadata.end_seconds,playback}]}'

Open the returned playback.viewer_url in a browser. It plays the whole recording with the matching intervals highlighted. The link is short-lived and does not expose your storage bucket, tenant or pipeline IDs. A weak query returns no_confident_match rather than unrelated scenes.

SDK. Get a playback link for a recording:

view = client.open_video(session_id, title="Loading bay, 02:10")
print(view["viewer_url"])

const view = await client.openVideo(sessionId, { title: "Loading bay, 02:10" });
console.log(view.viewer_url);

Troubleshooting