Multicam Quickstart

pip install videodb

import videodb

conn = videodb.connect(api_key="your_api_key")
coll = conn.get_collection()

CAMERA_CONFIG = {
    "cam1": ("Plaza Overview", "rtsp://samples.rts.videodb.io:8554/pub-cam1"),
    "cam2": ("Main Walkway", "rtsp://samples.rts.videodb.io:8554/pub-cam2"),
    "cam3": ("Stairway Junction", "rtsp://samples.rts.videodb.io:8554/pub-cam3"),
    "cam4": ("Central Plaza", "rtsp://samples.rts.videodb.io:8554/pub-cam4"),
}

streams = {}
for camera_id, (_, rtsp_url) in CAMERA_CONFIG.items():
    streams[camera_id] = coll.connect_rtstream(
        name=f"Surveillance_{camera_id}",
        url=rtsp_url,
        store=True,
    )

analysis_prompt = """Analyze this surveillance footage. Describe:
1. People with trolley bags, backpacks, or luggage
2. Crowd behavior and groups gathering
3. Unusual or unattended objects
Be specific about location and appearance."""

scene_understandings = {}
for camera_id, stream in streams.items():
    understanding = stream.understand(
        segmentation={"type": "time", "window": "10s"},
        analyzers=[{"type": "vlm", "name": "scene",
                    "sampling": {"frame_count": 1},
                    "config": {"prompt": analysis_prompt}}],
        store=True,
    )
    scene_understandings[camera_id] = {
        "understanding": understanding,
        "output": understanding.outputs.get("scene"),
    }

scene_indexes = {}
for camera_id, stream in streams.items():
    scene = scene_understandings[camera_id]
    index = stream.index(
        source=scene["output"],
        name=f"Surveillance_{camera_id}_Index",
        use_for=["semantic"],
    )
    scene_indexes[camera_id] = {
        "understanding": scene["understanding"],
        "output": scene["output"],
        "index": index,
        "index_id": index.id,
    }

EVENTS_CONFIG = [
    ("person_with_trolley", "Detect any person with a trolley bag or rolling suitcase."),
    ("large_crowd_formation", "Identify 5+ people gathering quickly."),
    ("unattended_luggage", "Detect luggage left unattended for 1+ minute."),
]

events = {
    label: conn.create_event(event_prompt=prompt, label=label)
    for label, prompt in EVENTS_CONFIG
}

import asyncio

ws_wrapper = conn.connect_websocket()
ws = await ws_wrapper.connect()

alerts = {}
for camera_id, scene in scene_indexes.items():
    alerts[camera_id] = {}
    for label, event_id in events.items():
        alerts[camera_id][label] = scene["index"].create_alert(
            event_id,
            callback_url="https://example.com",
            ws_connection_id=ws.connection_id,
        )

stream_to_camera = {stream.id: camera_id for camera_id, stream in streams.items()}
received_alerts = []

async def listen_for_alerts(timeout=30):
    try:
        async with asyncio.timeout(timeout):
            async for message in ws.receive():
                if message.get("channel") == "alert":
                    received_alerts.append(message)
                    camera_id = stream_to_camera.get(message.get("rtstream_id"))
                    data = message.get("data", {})
                    print(camera_id, data.get("label"), data.get("confidence"))
    except asyncio.TimeoutError:
        print(f"Received {len(received_alerts)} alert(s)")

await listen_for_alerts()

stream_urls = {}

selected = next(
    (
        alert for alert in received_alerts
        if isinstance(alert, dict)
        and isinstance(alert.get("data"), dict)
        and alert["data"].get("label") == "person_with_trolley"
    ),
    None,
)

if selected:
    data = selected["data"]
    start, end = data.get("start"), data.get("end")
    if isinstance(start, (int, float)) and isinstance(end, (int, float)) and start < end:
        padding = 10
        clip_start = int(start - padding)
        clip_end = int(end + padding)
        for camera_id, stream in streams.items():
            stream.generate_stream(clip_start, clip_end)
            stream_urls[camera_id] = stream.stream_url
    else:
        print("The selected alert does not include a valid data.start/data.end interval.")
else:
    print("No trolley alerts received yet; run the listener longer or choose another event.")

import subprocess

uploaded_videos = {}
if stream_urls:
    for camera_id, stream_url in stream_urls.items():
        output_file = f"{camera_id}_clip.mp4"
        subprocess.run(
            ["ffmpeg", "-y", "-i", stream_url, "-c", "copy", output_file],
            check=True,
        )
        uploaded_videos[camera_id] = coll.upload(file_path=output_file)

    final_duration = min(video.length for video in uploaded_videos.values())
else:
    print("No synchronized clips are available yet; skip composition.")

if uploaded_videos:
    from videodb.editor import Timeline, Track, Clip, VideoAsset, Position, Offset, Fit

    timeline = Timeline(conn)
    timeline.resolution = "1280x720"
    timeline.background = "#404040"

    layout = [
        ("cam1", Position.top_left, Offset(x=0.03, y=0.025)),
        ("cam2", Position.top_right, Offset(x=-0.03, y=0.025)),
        ("cam3", Position.bottom_left, Offset(x=0.03, y=-0.025)),
        ("cam4", Position.bottom_right, Offset(x=-0.03, y=-0.025)),
    ]

    for camera_id, position, offset in layout:
        track = Track()
        track.add_clip(0, Clip(
            asset=VideoAsset(id=uploaded_videos[camera_id].id),
            duration=final_duration,
            fit=Fit.crop,
            position=position,
            offset=offset,
            scale=0.45,
        ))
        timeline.add_track(track)

    multicam_url = timeline.generate_stream()
else:
    print("No synchronized clips are available yet; skip composition.")

for camera_id, camera_alerts in alerts.items():
    for alert_id in camera_alerts.values():
        scene_indexes[camera_id]["index"].disable_alert(alert_id)

for scene in scene_indexes.values():
    scene["index"].stop()
    scene["understanding"].stop()

for stream in streams.values():
    stream.stop()

await ws_wrapper.close()