Building a Real-Time Video Analytics Dashboard with Flask, PyTorch, and LiveKit
Real-Time Video Analytics Dashboard : A Flask-based Web Application
The demand for intelligent video analytics is surging across industries—from retail stores tracking customer flow to security systems detecting intrusions and smart cities managing traffic. Traditional video monitoring requires human operators watching multiple feeds, a task that is both expensive and prone to errors. The Real-Time Video Analytics Dashboard addresses this challenge by combining deep learning with modern web technologies to create an intelligent, automated monitoring system.
System Architecture Overview
The architecture follows a decoupled design pattern where video ingestion, AI processing, and user presentation operate independently while communicating through well-defined channels:
text:
┌─────────────────────────────────────────────────────────────┐
│ Browser Client │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────┐ │
│ │ Video │ │ Dashboard │ │ Chart.js │ │
│ │ Player │ │ UI │ │ Analytics │ │
│ └─────────────┘ └─────────────┘ └─────────────────────┘ │
└────────────────────────┬────────────────────────────────────┘
│ WebRTC (Video) / WebSockets (Data)
┌────────────────────────▼────────────────────────────────────┐
│ FastAPI/Flask Gateway │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────┐ │
│ │ Camera │ │ Room │ │ WebSocket │ │
│ │ Management │ │ Management │ │ Server │ │
│ └─────────────┘ └─────────────┘ └─────────────────────┘ │
└────────────────────────┬────────────────────────────────────┘
│
┌────────────────────────▼────────────────────────────────────┐
│ LiveKit WebRTC Server │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────┐ │
│ │ Signaling │ │ Media │ │ TURN/STUN │ │
│ │ Handler │ │ Forwarding │ │ Relay │ │
│ └─────────────┘ └─────────────┘ └─────────────────────┘ │
└────────────────────────┬────────────────────────────────────┘
│
┌────────────────────────▼────────────────────────────────────┐
│ GPU Worker Pool (Celery) │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────┐ │
│ │ PyTorch │ │ CUDA │ │ Object │ │
│ │ Models │ │ Acceleration │ │ Tracking │ │
│ └─────────────┘ └─────────────┘ └─────────────────────┘ │
└────────────────────────┬────────────────────────────────────┘
│
┌────────────────────────▼────────────────────────────────────┐
│ Redis (State Management) │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────┐ │
│ │ Detection │ │ Frame Cache │ │ Job Queue │ │
│ │ State │ │ │ │ (Celery) │ │
│ └─────────────┘ └─────────────┘ └─────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
This architecture enables independent scaling of components—API servers for request handling, GPU workers for processing, and LiveKit servers for media streaming.
Setting Up the Flask Backend
Core Dependencies
The backend requires a combination of web framework, video processing, and real-time communication libraries:
bash:
pip install flask flask-socketio livekit-api livekit-protocol redis celery
pip install torch torchvision opencv-python ultralytics
For GPU acceleration, ensure PyTorch is installed with CUDA support matching your hardware configuration.
Flask Application with LiveKit Integration
The Flask backend serves as the orchestration layer, handling camera registration, room management, and token issuance. This pattern follows the Recognizer example from jetson-inference, where a Flask webserver manages WebRTC streaming, inferencing, and training threads :
python:
from flask import Flask, request, jsonify
from livekit import api
import os
import redis
import uuid
app = Flask(__name__)
redis_client = redis.Redis(host="localhost", port=6379, decode_responses=True)
# JWT token generation for LiveKit client authentication
@app.post("/token")
async def get_token():
room_name = request.json.get("room_name")
identity = request.json.get("identity", str(uuid.uuid4()))
grant = api.VideoGrants(
room_join=True,
room=room_name,
can_publish=True,
can_subscribe=True,
)
token = api.AccessToken() \
.with_identity(identity) \
.with_name(f"User {identity}") \
.with_grants(grant) \
.with_ttl(3600)
return {"token": token.to_jwt()}
# Endpoint to register a camera feed
@app.post("/api/cameras")
def register_camera():
data = request.json
camera_id = str(uuid.uuid4())
redis_client.hset(
f"camera:{camera_id}",
mapping={
"name": data.get("name"),
"url": data.get("url"),
"active": "1",
"created_at": str(datetime.utcnow())
}
)
return {"camera_id": camera_id}
WebSocket for Real-Time Data
WebSockets provide the communication channel for sending detection results and alerts to the dashboard. The Real-Time Restricted Area Monitoring System demonstrates this pattern—FastAPI handles real-time WebSocket communication while Streamlit provides the interactive UI for live visualization :
python:
from flask_socketio import SocketIO, emit
socketio = SocketIO(app, cors_allowed_origins="*")
def broadcast_detection(camera_id, detections):
"""Broadcast detection results to all connected dashboard clients."""
socketio.emit(
"detection_update",
{"camera_id": camera_id, "detections": detections}
)
GPU-Accelerated Video Processing
Object Detection with YOLO
YOLO (You Only Look Once) family of models provides state-of-the-art object detection with excellent speed-accuracy trade-offs. YOLOv7 introduces improvements over previous versions, offering faster inference and higher detection accuracy suitable for surveillance and automation . More recent YOLOv11 models achieve 66.67-95.83% counting accuracy in traffic monitoring applications, with high precision for vehicles (cars: 0.97-1.00, trucks: 1.00) and strong recall (cars: 0.82-1.00, trucks: 0.70-1.00).
The processing pipeline integrates the YOLO architecture with PyTorch, using a robust backbone for feature extraction, a feature aggregation neck, and a detection head for predicting object classes and bounding box coordinates :
python:
from ultralytics import YOLO
import torch
import cv2
class VideoAnalyzer:
def __init__(self, model_path="yolo11n.pt"):
self.model = YOLO(model_path)
# Enable GPU if available
self.device = "cuda" if torch.cuda.is_available() else "cpu"
def process_frame(self, frame):
"""Run detection on a single frame."""
results = self.model(frame, device=self.device, verbose=False)[0]
detections = []
if results.boxes is not None:
for box in results.boxes:
x1, y1, x2, y2 = box.xyxy[0].tolist()
detections.append({
"bbox": [x1, y1, x2, y2],
"confidence": float(box.conf[0]),
"class_id": int(box.cls[0]),
"class_name": self.model.names[int(box.cls[0])]
})
return detections
Real-Time Tracking with ByteTrack
For applications requiring consistent object identification across frames—people counting, vehicle tracking, anomaly detection—multi-object tracking algorithms are essential. The combination of YOLO detector with ByteTrack, BoT-SORT, or DeepSORT enables stable tracking with persistent object IDs :
python:
def process_video_stream(self, video_source, callback=None):
"""Process video stream with detection and tracking."""
cap = cv2.VideoCapture(video_source)
frame_count = 0
while cap.isOpened():
ret, frame = cap.read()
if not ret:
break
frame_count += 1
# Run tracking with persistence
results = self.model.track(
frame,
persist=True,
tracker="bytetrack.yaml",
verbose=False
)[0]
if results.boxes is not None and results.boxes.id is not None:
for box, track_id in zip(results.boxes, results.boxes.id):
# Update state for tracking across frames
self.update_tracker_state(track_id, box)
if callback:
callback(frame, results)
Redis for State Management and Caching
Redis serves as the central state store, providing three essential capabilities: caching frame detections for performance, maintaining tracking state across restarts, and managing the Celery job queue for background processing.
Detection Caching for Performance
In video analytics, processing identical frames repeatedly is wasteful. Caching detection results with Redis dramatically improves performance—first inference takes approximately 30-50ms on a good GPU, but subsequent retrievals from Redis complete in about 0.5-1ms :
python:
import hashlib
import pickle
import redis
import cv2
redis_client = redis.Redis(host='localhost', port=6379, db=0)
def frame_hash(frame):
return hashlib.md5(frame.tobytes()).hexdigest()
def get_detections_with_cache(frame, model):
"""Get detections with Redis cache."""
key = f"yolo_det:{frame_hash(frame)}"
cached = redis_client.get(key)
if cached:
return pickle.loads(cached)
# Run inference and cache results
results = model(frame, verbose=False)[0]
detections = []
if results.boxes is not None:
for box in results.boxes:
x1, y1, x2, y2 = box.xyxy[0].tolist()
detections.append({
"bbox": [x1, y1, x2, y2],
"confidence": float(box.conf[0]),
"class_id": int(box.cls[0])
})
# Cache for 30 minutes
redis_client.setex(key, 1800, pickle.dumps(detections))
return detections
Persistent Tracking State
One challenge with tracking is that state (object IDs, trajectories) lives in the process memory and is lost on restart. Redis solves this by persisting tracking data—on restart, the system resumes with consistent IDs :
python:
def save_track_state(track_id, bbox, frame_timestamp):
"""Save tracking state to Redis."""
key = f"track:{track_id}"
redis_client.hset(key, mapping={
"last_seen": frame_timestamp,
"bbox": json.dumps(bbox)
})
redis_client.expire(key, 5) # Clean up stale tracks
def restore_tracking_state():
"""Restore tracking state on system restart."""
# Rebuild tracker state from Redis
# This allows consistent IDs across restarts
Asynchronous Job Queue with Celery
For processing large video files or offloading heavy inference tasks, Celery with Redis as the broker creates an asynchronous background processing pipeline :
python:
from celery import Celery
celery_app = Celery(
"video_worker",
broker="redis://localhost:6379/0",
backend="redis://localhost:6379/1"
)
@celery_app.task(bind=True)
def process_video_async(self, video_id, video_path):
"""Process video in background."""
self.update_state(
state="PROCESSING",
meta={"status": "Loading model...", "progress": 0}
)
# Process video with GPU
results = analyze_video(video_path, callback=self.update_state)
return {"video_id": video_id, "results": results}
WebRTC for Low-Latency Video Delivery
LiveKit provides the WebRTC infrastructure for low-latency video transport. The Python SDK supports connecting to LiveKit rooms as part of request handlers, making it compatible with Flask's synchronous request/response cycle :
python:
import asyncio
from livekit import rtc, api
async def connect_to_livekit_room(room_name, token):
"""Connect to a LiveKit room for video ingestion."""
room = rtc.Room()
@room.on("track_subscribed")
def on_track_subscribed(track, publication, participant):
if track.kind == rtc.TrackKind.KIND_VIDEO:
print(f"Video track from {participant.identity}")
# Process video track as it arrives
await room.connect(
os.environ["LIVEKIT_URL"],
token
)
print(f"Connected to room: {room.name}")
return room
The NVIDIA Video Analytics UI reference implementation demonstrates this approach—the UI communicates with the Media Streaming module using HTTP calls and streams video content over WebRTC channels, with analytics data and application metadata queried via HTTP/WebSocket channels.
The Browser Dashboard
Dashboard UI with React or Streamlit
The dashboard presents live video feeds, detection results, and analytics visualizations. Both React (with Google Maps API for geospatial visualization) and Streamlit (for rapid prototyping) are viable options.
The dashboard typically includes:
- Live video streams with bounding boxes overlaid
- Real-time charts using Chart.js for detection counts and trends
- Alert panels displaying security or anomaly events
- KPI cards showing total detections, violations, and most frequent object classes
WebSocket Integration
The browser dashboard connects to the WebSocket server to receive real-time updates without page refreshes :
javascript:
// Connect to Flask WebSocket server
const socket = io();
socket.on('detection_update', function(data) {
// Update charts
updateChart(data.camera_id, data.detections);
// Update alerts if violations detected
const violations = data.detections.filter(d => d.class_id in restricted_classes);
if (violations.length > 0) {
displayAlert(violations);
}
});
Performance Optimization and Production Considerations
Scaling GPU Workers
For multiple camera feeds, scale GPU workers horizontally based on processing demand. The worker monitoring dashboard should track queue depth, GPU utilization, and processing latency.
UI Delay Compensation
Video analytics introduces processing latency. The UI configuration can specify a uiDelaySeconds parameter—the time the UI lags behind real video to align analytics data with the corresponding video position :
json:
{
"uiDelaySeconds": 20,
"alertQueryDurationInHours": 2,
"alertListLength": 20,
"apiRefreshIntervalSeconds": 2
}
Security and Deployment
Production deployments require JWT authentication, role-based access control, and secure environment variables. Projects like Synth demonstrate single Docker container deployment with all services—Flask backend, React frontend, and LiveKit—using Docker Compose.
Real-World Implementation Examples
Several production systems demonstrate this architecture:
Recognizer (jetson-inference): A Flask-based video tagging and classification webapp with interactive data collection, background training with PyTorch, and TensorRT inference. It supports WebRTC client video input and dynamic model reloading.
Synth: A real-time voice channel app inspired by Discord using React 19, Flask, LiveKit WebRTC, and MongoDB. It demonstrates role-based access control with server-scoped permissions, JWT authentication, and single Docker container deployment.
Real-Time Restricted Area Monitoring System: Integrates FastAPI and Streamlit with YOLO for real-time object detection and violation alerts. Uses WebSockets for live data streaming and CSV-based detection logging.
Intelligent Traffic Monitoring: A real-time system coupling YOLOv11 with BoT-SORT/ByteTrack for vehicle detection and counting, achieving high precision (cars: 0.97-1.00) and robust performance across diverse scenes.
Conclusion
The Real-Time Video Analytics Dashboard demonstrates how modern deep learning and web technologies can work together to create intelligent video monitoring systems. By combining Flask for API orchestration, PyTorch and YOLO for GPU-accelerated detection, LiveKit for WebRTC video transport, and Redis for state management and caching, the architecture delivers a scalable, production-ready solution.
The key to success lies in the decoupled design—each component operates independently while communicating through well-defined channels. This allows independent scaling of API servers, GPU workers, and media streaming infrastructure, making the system suitable for applications ranging from retail analytics and security surveillance to smart city traffic monitoring.