> ## Documentation Index
> Fetch the complete documentation index at: https://docs.videodb.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Multicam Quickstart

> Connect four live cameras, receive WebSocket alerts, and create synchronized multi-angle evidence

<a href="https://colab.research.google.com/github/video-db/videodb-cookbook/blob/main/quickstart/Multicam_Quickstart.ipynb" target="_blank">
  <img src="https://colab.research.google.com/assets/colab-badge.svg" alt="Open In Colab" noZoom />
</a>

## What You'll Build

This quickstart covers the reusable multicam pattern behind VideoDB's live-intelligence examples:

* Connect and store four RTSP feeds
* Run an independent visual index for every camera
* Detect three events on every feed
* Receive and attribute alerts over one WebSocket
* Retrieve the detected interval from all cameras
* Compose a synchronized 2×2 video with the Timeline Editor

The sample uses four public-plaza feeds and detects people with trolley bags, crowd formation, and unattended luggage. Replace the URLs and prompts to adapt the pattern to your own cameras.

## Setup

```bash theme={null}
pip install videodb
```

```python theme={null}
import videodb

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

## 1. Connect the Camera Feeds

```python theme={null}
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,
    )
```

<Note>
  Keep `store=True` enabled. Multicam replay depends on requesting the same historical time range from every camera.
</Note>

## 2. Index Every Camera

```python theme={null}
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,
    }
```

## 3. Create Events and Alerts

```python theme={null}
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
}
```

Open one WebSocket and attach every camera-event combination to it:

```python theme={null}
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,
        )
```

This creates 12 alerts: four cameras multiplied by three event types.

## 4. Receive Alerts from Every Camera

```python theme={null}
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()
```

An alert's top-level `rtstream_id` identifies the camera. The observed `data` envelope can include the event label, confidence, explanation, detected `start` and `end` timestamps, and a clip URL; inspect live messages because optional fields are delivery-dependent.

## 5. Retrieve a Synchronized Time Window

This uses the same Unix timestamp range across every stream, so the camera clocks must already be aligned for frame-level synchronization.

```python theme={null}
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.")
```

<Warning>
  `generate_stream()` returns a player URL. Read `stream.stream_url` after the call when you need the HLS media URL for downloading or composition.
</Warning>

## 6. Compose the Multi-Angle Video

The notebook completes the replay in three stages:

1. Download each HLS URL with FFmpeg.
2. Upload the four MP4 clips with `coll.upload(file_path=...)`.
3. Place each `VideoAsset(id=...)` on its own Timeline Editor track.

```python theme={null}
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.")
```

```python theme={null}
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.")
```

Open the notebook for the complete FFmpeg, upload, duration matching, camera label, and playback cells.

## Cleanup

```python theme={null}
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()
```

<Card icon="notebook" title="Open the Multicam Quickstart" href="https://colab.research.google.com/github/video-db/videodb-cookbook/blob/main/quickstart/Multicam_Quickstart.ipynb">
  Run the complete four-camera workflow in Google Colab.
</Card>

## Next Steps

<CardGroup cols={2}>
  <Card title="Public Safety Surveillance" icon="cctv" href="/examples-and-tutorials/live-intelligence/multicam-public-surveillance">
    Apply the pattern to a public-plaza monitoring use case
  </Card>

  <Card title="Basketball Analytics" icon="activity" href="/examples-and-tutorials/live-intelligence/multicam-basketball-analysis">
    Adapt multicam alerts and composition to live sports
  </Card>
</CardGroup>