Building a pedestrian detection system for a warehouse or industrial site is not the same as running a demo on a GPU server. The hardware is fixed, the latency budget is hard, and a missed detection has real safety consequences. This post covers the engineering decisions involved in deploying a real-time pedestrian detection system: model selection, multi-object tracking with Kalman filters, inference optimization for embedded hardware, and the integration patterns that make these systems reliable in production.
Industrial CV deployments run under constraints that change every decision:
For real-time pedestrian detection on embedded hardware, YOLOv8n (nano) or YOLOv8s (small) is the practical starting point. The full YOLOv8m/l/x variants exceed the latency budget on Jetson hardware.
from ultralytics import YOLO
import cv2
import time
# Load a pretrained YOLOv8n model (COCO-pretrained, includes "person" class = class 0)
model = YOLO("yolov8n.pt")
# Fine-tune on your specific environment
# Critical: collect training data in your actual deployment environment
# The COCO "person" class covers standing pedestrians but not workers in full PPE
results = model.train(
data="industrial_pedestrian.yaml", # custom dataset
epochs=50,
imgsz=640,
batch=16,
device=0,
workers=4,
project="runs/detect",
name="pedestrian_industrial",
# Augmentation tuned for industrial environments
degrees=10.0, # rotation (cameras may not be level)
fliplr=0.5,
mosaic=0.5,
mixup=0.1,
hsv_h=0.02,
hsv_s=0.5,
hsv_v=0.5, # lighting variation
)
COCO pedestrian images are not sufficient. Industrial workers wear high-visibility vests, hard hats, and full-body PPE that changes their visual profile significantly. Collect at least 2,000-3,000 images in your actual deployment environment, annotated with bounding boxes. Use tools like Roboflow or Label Studio.
Critical augmentations for industrial environments:
Detection per frame is not enough. You need tracking: a consistent ID for each person across frames, so you can detect if a person enters a restricted zone, compute their velocity, or trigger an alert only after sustained presence in a danger area.
A Kalman filter is a recursive state estimator. Given a sequence of noisy bounding box observations, it maintains an estimate of the "true" position and velocity of a tracked object. When the detector misses a frame (occlusion, motion blur), the Kalman filter predicts where the object should be — allowing tracking to continue across short gaps.
import numpy as np
from filterpy.kalman import KalmanFilter
from scipy.optimize import linear_sum_assignment
class KalmanBoxTracker:
"""
State vector: [cx, cy, s, r, dcx, dcy, ds]
where (cx, cy) = center, s = scale (area), r = aspect ratio, d* = derivatives
"""
count = 0
def __init__(self, bbox):
self.kf = KalmanFilter(dim_x=7, dim_z=4)
# State transition matrix (constant velocity model)
self.kf.F = np.array([
[1, 0, 0, 0, 1, 0, 0],
[0, 1, 0, 0, 0, 1, 0],
[0, 0, 1, 0, 0, 0, 1],
[0, 0, 0, 1, 0, 0, 0],
[0, 0, 0, 0, 1, 0, 0],
[0, 0, 0, 0, 0, 1, 0],
[0, 0, 0, 0, 0, 0, 1],
], dtype=float)
# Measurement function: observe [cx, cy, s, r]
self.kf.H = np.array([
[1, 0, 0, 0, 0, 0, 0],
[0, 1, 0, 0, 0, 0, 0],
[0, 0, 1, 0, 0, 0, 0],
[0, 0, 0, 1, 0, 0, 0],
], dtype=float)
self.kf.R[2:, 2:] *= 10. # measurement noise on scale and ratio
self.kf.P[4:, 4:] *= 1000. # high initial velocity uncertainty
self.kf.P *= 10.
self.kf.Q[-1, -1] *= 0.01
self.kf.Q[4:, 4:] *= 0.01
self.kf.x[:4] = self._bbox_to_z(bbox)
self.time_since_update = 0
self.id = KalmanBoxTracker.count
KalmanBoxTracker.count += 1
self.history = []
self.hits = 0
self.hit_streak = 0
self.age = 0
@staticmethod
def _bbox_to_z(bbox):
"""[x1, y1, x2, y2] → [cx, cy, s, r]"""
w = bbox[2] - bbox[0]
h = bbox[3] - bbox[1]
x = bbox[0] + w / 2.
y = bbox[1] + h / 2.
s = w * h
r = w / float(h) if h > 0 else 1.0
return np.array([x, y, s, r]).reshape((4, 1))
@staticmethod
def _z_to_bbox(z):
"""[cx, cy, s, r] → [x1, y1, x2, y2]"""
w = np.sqrt(z[2] * z[3])
h = z[2] / w if w > 0 else 0
return np.array([
z[0] - w / 2., z[1] - h / 2.,
z[0] + w / 2., z[1] + h / 2.
]).flatten()
def predict(self):
if self.time_since_update > 0:
self.hit_streak = 0
self.time_since_update += 1
self.kf.predict()
self.history.append(self._z_to_bbox(self.kf.x[:4]))
return self.history[-1]
def update(self, bbox):
self.time_since_update = 0
self.hits += 1
self.hit_streak += 1
self.kf.update(self._bbox_to_z(bbox))
def iou(bb_a, bb_b):
"""Intersection over Union for two bounding boxes."""
xa = max(bb_a[0], bb_b[0])
ya = max(bb_a[1], bb_b[1])
xb = min(bb_a[2], bb_b[2])
yb = min(bb_a[3], bb_b[3])
inter = max(0, xb - xa) * max(0, yb - ya)
area_a = (bb_a[2] - bb_a[0]) * (bb_a[3] - bb_a[1])
area_b = (bb_b[2] - bb_b[0]) * (bb_b[3] - bb_b[1])
return inter / float(area_a + area_b - inter + 1e-6)
class SORTTracker:
def __init__(self, max_age: int = 10, min_hits: int = 3, iou_threshold: float = 0.3):
self.max_age = max_age
self.min_hits = min_hits
self.iou_threshold = iou_threshold
self.trackers: list[KalmanBoxTracker] = []
self.frame_count = 0
def update(self, detections: np.ndarray) -> np.ndarray:
"""
detections: np.ndarray of shape (N, 5) — [x1, y1, x2, y2, confidence]
returns: np.ndarray of shape (M, 5) — [x1, y1, x2, y2, track_id]
"""
self.frame_count += 1
predicted = np.array([t.predict() for t in self.trackers])
# Hungarian algorithm for assignment
if len(self.trackers) > 0 and len(detections) > 0:
iou_matrix = np.array([[iou(det[:4], pred) for pred in predicted] for det in detections])
row_ind, col_ind = linear_sum_assignment(-iou_matrix)
matched = {(r, c) for r, c in zip(row_ind, col_ind) if iou_matrix[r, c] >= self.iou_threshold}
else:
matched = set()
# Update matched trackers
matched_dets = {r for r, _ in matched}
matched_trks = {c for _, c in matched}
for r, c in matched:
self.trackers[c].update(detections[r, :4])
# Add new trackers for unmatched detections
for r in range(len(detections)):
if r not in matched_dets:
self.trackers.append(KalmanBoxTracker(detections[r, :4]))
# Remove dead trackers
active = []
results = []
for i, trk in enumerate(self.trackers):
if trk.time_since_update <= self.max_age:
active.append(trk)
if trk.hit_streak >= self.min_hits or self.frame_count <= self.min_hits:
bbox = trk.history[-1]
results.append([*bbox, trk.id])
self.trackers = active
return np.array(results) if results else np.empty((0, 5))
To run at 25+ fps on a Jetson NX or similar embedded device, you need to export the model to TensorRT.
# Export to TensorRT (run on the target device or same architecture)
from ultralytics import YOLO
model = YOLO("runs/detect/pedestrian_industrial/weights/best.pt")
# Export to TensorRT with FP16 precision
model.export(
format="engine",
device=0,
half=True, # FP16 — 2x throughput on Tensor Core GPUs
imgsz=(640, 640),
workspace=4, # GB of GPU memory for TRT optimization
)
# Load and run the TRT engine
trt_model = YOLO("runs/detect/pedestrian_industrial/weights/best.engine")
# Benchmark inference time
import time
import numpy as np
dummy_frame = np.random.randint(0, 255, (640, 640, 3), dtype=np.uint8)
# Warm up
for _ in range(10):
trt_model(dummy_frame, verbose=False)
times = []
for _ in range(100):
t0 = time.perf_counter()
trt_model(dummy_frame, verbose=False)
times.append(time.perf_counter() - t0)
print(f"Mean inference: {np.mean(times)*1000:.1f}ms")
print(f"P99 inference: {np.percentile(times, 99)*1000:.1f}ms")
On a Jetson NX with TensorRT FP16, YOLOv8n runs at ~8ms per frame (125 fps). Adding the SORT tracking step adds ~2ms per frame. Total pipeline latency is comfortably under the 100ms budget.
from dataclasses import dataclass
from typing import Optional
import numpy as np
@dataclass
class SafetyZone:
name: str
polygon: np.ndarray # shape (N, 2), in image pixel coordinates
alert_after_frames: int = 5 # require sustained presence to avoid false alarms
def point_in_polygon(point: np.ndarray, polygon: np.ndarray) -> bool:
"""Ray casting algorithm."""
x, y = point
n = len(polygon)
inside = False
j = n - 1
for i in range(n):
xi, yi = polygon[i]
xj, yj = polygon[j]
if ((yi > y) != (yj > y)) and (x < (xj - xi) * (y - yi) / (yj - yi + 1e-9) + xi):
inside = not inside
j = i
return inside
class SafetyMonitor:
def __init__(self, zones: list[SafetyZone]):
self.zones = zones
self.zone_entry_frames: dict[int, dict[str, int]] = {} # track_id → {zone_name → frames_inside}
def check(self, tracks: np.ndarray) -> list[dict]:
"""
tracks: output from SORTTracker.update() — [x1, y1, x2, y2, track_id]
Returns list of active alerts.
"""
alerts = []
active_ids = set()
for track in tracks:
x1, y1, x2, y2, tid = track
tid = int(tid)
active_ids.add(tid)
center = np.array([(x1 + x2) / 2, (y1 + y2) / 2])
if tid not in self.zone_entry_frames:
self.zone_entry_frames[tid] = {}
for zone in self.zones:
if point_in_polygon(center, zone.polygon):
self.zone_entry_frames[tid][zone.name] = self.zone_entry_frames[tid].get(zone.name, 0) + 1
if self.zone_entry_frames[tid][zone.name] >= zone.alert_after_frames:
alerts.append({
"zone": zone.name,
"track_id": tid,
"frames_in_zone": self.zone_entry_frames[tid][zone.name],
"bbox": [x1, y1, x2, y2],
})
else:
self.zone_entry_frames[tid].pop(zone.name, None)
# Clean up dead tracks
dead_ids = set(self.zone_entry_frames.keys()) - active_ids
for tid in dead_ids:
del self.zone_entry_frames[tid]
return alerts
Industrial CV systems run for months unattended. The application layer needs to handle failures gracefully:
import threading
import queue
import time
import logging
logger = logging.getLogger(__name__)
class RobustVideoProcessor:
def __init__(self, source: str, detector, tracker, monitor, alert_callback):
self.source = source
self.detector = detector
self.tracker = tracker
self.monitor = monitor
self.alert_callback = alert_callback
self._stop_event = threading.Event()
self._frame_queue = queue.Queue(maxsize=5)
def _capture_thread(self):
"""Capture frames in a separate thread — don't block on inference."""
cap = cv2.VideoCapture(self.source)
consecutive_failures = 0
while not self._stop_event.is_set():
ret, frame = cap.read()
if not ret:
consecutive_failures += 1
logger.warning(f"Frame capture failed ({consecutive_failures} consecutive)")
if consecutive_failures >= 30:
# Camera disconnect — attempt reconnect
cap.release()
time.sleep(2)
cap = cv2.VideoCapture(self.source)
consecutive_failures = 0
continue
consecutive_failures = 0
try:
self._frame_queue.put_nowait(frame)
except queue.Full:
pass # Drop frame rather than fall behind
cap.release()
def run(self):
capture = threading.Thread(target=self._capture_thread, daemon=True)
capture.start()
while not self._stop_event.is_set():
try:
frame = self._frame_queue.get(timeout=1.0)
except queue.Empty:
continue
try:
results = self.detector(frame, classes=[0], verbose=False)[0] # class 0 = person
detections = results.boxes.data.cpu().numpy() # [x1, y1, x2, y2, conf, cls]
tracks = self.tracker.update(detections[:, :5])
alerts = self.monitor.check(tracks)
for alert in alerts:
self.alert_callback(alert)
except Exception as e:
logger.error(f"Inference error: {e}", exc_info=True)
def stop(self):
self._stop_event.set()
Real-time pedestrian detection for industrial deployment requires solving three distinct problems: a detection model accurate enough for the specific environment (PPE, occlusion, lighting), a tracking layer that maintains identity across frames and short occlusions, and an inference pipeline fast enough to run at useful frame rates on embedded hardware. TensorRT export of YOLOv8 handles the latency requirement; SORT with Kalman filters handles tracking; zone-based safety logic converts raw detections into actionable alerts. The reliability patterns — separate capture thread, reconnect logic, error isolation — are what make the system usable in production, not just in demo conditions.