Events & Real-time

import videodb

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

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

# Wire to index with delivery method
index.create_alert(
    event_id=event_id,
    callback_url="https://your-backend.com/alerts",  # Webhook
    ws_connection_id=ws.connection_id                 # Real-time
)

# Alerts fire in <1 second
async for event in ws.stream():
    if event.get("channel") == "alert":
        print(f"ALERT: {event['data']['text']}")

# Events are reusable across streams
safety_event = conn.create_event(
    event_prompt="Detect safety violation",
    label="safety"
)

# Alerts wire events to specific indexes
for rtstream in streams:
    index = rtstream.get_scene_index(index_id)
    index.create_alert(event_id=safety_event)

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

async for event in ws.stream():
    channel = event.get("channel")
    # transcript, scene_index, audio_index, alert

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

{
  "event_label": "intrusion",
  "rtstream_id": "rts-xxx",
  "timestamp": 1710000012340,
  "data": {"text": "Person in restricted area"}
}

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