Events & Real-time

import videodb

conn = videodb.connect()
coll = conn.get_collection()
rtstream = coll.get_rtstream("rts-xxx")
ws = conn.connect_websocket()
await ws.connect()

# Continuously create and store scene output
scene_understanding = rtstream.understand(
    segmentation={"type": "time", "window": "5s"},
    analyzers=[
        {
            "type": "vlm",
            "name": "scene",
            "sampling": {"frame_count": 2},
            "config": {"prompt": "Describe activity and detect unusual behavior"},
        }
    ],
    store=True,
)

# Index the stored scene output before attaching alerts
scene_index = rtstream.index(
    name="security_monitor",
    source=scene_understanding.outputs["scene"],
    use_for=["semantic"],
)

# Create detection rule (reusable)
event_id = conn.create_event(
    event_prompt="Detect intruder",
    label="security_alert"
)

# Wire to the V2 scene index; callback delivery remains required with WebSocket
scene_index.create_alert(
    event_id=event_id,
    callback_url="https://your-backend.com/alerts",  # Webhook
    ws_connection_id=ws.connection_id                 # Real-time
)

# Inspect WebSocket messages before accessing fields; their decoded payload shape is pipeline-specific.
async for event in ws.receive():
    print(event)

# Reuse the same event with the V2 scene index
safety_event = conn.create_event(
    event_prompt="Detect safety violation",
    label="safety"
)

# Alerts wire events to specific indexes
scene_index.create_alert(
    event_id=safety_event,
    callback_url="https://your-backend.com/alerts",
)

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

async for event in ws.receive():
    print(event)  # Inspect the decoded payload before accessing fields.

scene_index.create_alert(
    event_id=event_id,
    callback_url="https://your-backend.com/alerts"
)

{
  "channel": "alert",
  "rtstream_id": "rts-xxx",
  "data": {
    "event_id": "event-xxx",
    "label": "intrusion",
    "triggered": true,
    "confidence": 0.92,
    "start": 1710000012340,
    "end": 1710000018900
  }
}

# Webhook idempotency
@app.post("/webhooks")
async def handle(request):
    event = await request.json()
    if already_processed(event["event_id"]):
        return {"status": "ok"}
    process(event)