Video And Streaming Vision

• Choose frame sampling that preserves key moments while meeting compute budgets.
• Stabilize streams with buffering, timestamps, and jitter handling.
• Build tracking across frames and aggregate predictions over time.
• Optimize real-time pipelines with queues, batching, and asynchronous I/O.
• Evaluate with video metrics (latency, FPS, MOTA/IDF1) and iterate.

• Implement uniform, strided, and motion-adaptive sampling.
• Measure effective FPS and CPU/GPU usage.

• Read presentation timestamps (PTS), maintain a small buffer, and re-order out-of-order frames.
• Detect and log dropped frames; pace playback to real time.

• Track detections with a simple tracker (e.g., CSRT/Kalman + IoU matching).
• Persist identities; handle short occlusions.

• Smooth noisy predictions with majority vote or exponential moving average (EMA).
• Aggregate per-frame detections into events.

• Decouple capture, infer, and output with queues/threads or async.
• Measure end-to-end latency; set backpressure policy (drop oldest vs. block).

• Apply denoising, sharpening, and adaptive thresholds.
• Tune shutter/exposure (if you can) and use temporal filtering.

• Track FPS, latency, drop rate, MOTA/IDF1, and event-level precision/recall.
• Create small labeled clips for regression testing.

Start simple; switch to motion-aware sampling to capture key moments without overloading compute.

import cv2, numpy as np

cap = cv2.VideoCapture("input.mp4")
stride = 3  # process every 3rd frame
prev = None
alpha = 0.2  # motion-adaptive threshold scaling

frame_idx = 0
proc_idx = 0
while True:
    ok, frame = cap.read()
    if not ok:
        break
    frame_idx += 1

    # Uniform sampling
    if frame_idx % stride == 0:
        proc_idx += 1
        # run_model(frame)

    # Motion-adaptive sampling (alternate path)
    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
    if prev is not None:
        diff = cv2.absdiff(gray, prev)
        motion = np.mean(diff)
        dynamic_stride = 4 if motion < 5 else (2 if motion > 20 else 3)
        if frame_idx % dynamic_stride == 0:
            pass  # run_model(frame)
    prev = gray

Tip: Log effective FPS and motion score to justify your choice.

import time, collections
import cv2

cap = cv2.VideoCapture(0)  # webcam or RTSP
buf = collections.deque(maxlen=8)
last_pts = None

def now_ms():
    return int(time.time() * 1000)

while True:
    ok, frame = cap.read()
    if not ok:
        break
    pts = cap.get(cv2.CAP_PROP_POS_MSEC)  # ms; may be 0 for some sources
    wall = now_ms()

    if last_pts is not None and pts > 0 and pts < last_pts:
        # Out-of-order; skip or reorder
        continue
    last_pts = pts

    buf.append((pts, wall, frame))

    # Pace to real-time (example):
    if len(buf) >= 2:
        (p0, w0, _), (p1, w1, _) = buf[-2], buf[-1]
        if p0 > 0 and p1 > 0:
            target_delay = (p1 - p0) / 1000.0
            elapsed = (w1 - w0) / 1000.0
            if 0 < target_delay - elapsed < 0.05:
                time.sleep(target_delay - elapsed)

Record dropped frames as gaps in PTS or sequence numbers.

# Pseudocode for clarity
tracks = []  # each: {id, bbox, kf, last_seen}
next_id = 0

while True:
    frame = get_frame()
    dets = run_detector(frame)  # list of [x1,y1,x2,y2,score]

    # Predict step
    for t in tracks:
        t["bbox"] = t["kf"].predict()

    # Associate by IoU
    matches, unmatched_tracks, unmatched_dets = iou_match(tracks, dets, iou_thr=0.3)

    # Update matched tracks
    for ti, di in matches:
        tracks[ti]["kf"].update(dets[di])
        tracks[ti]["bbox"] = dets[di]
        tracks[ti]["last_seen"] = frame.index

    # Create new tracks
    for di in unmatched_dets:
        tracks.append(new_track(next_id, dets[di]))
        next_id += 1

    # Remove stale tracks
    tracks = [t for t in tracks if frame.index - t["last_seen"] <= 30]

Keep association simple first. Add appearance embeddings later if IDs swap too often.

import cv2
import numpy as np

img = cv2.imread("frame.png")
# Gamma correction (brighten dark scenes)
gamma = 1.8
lut = np.array([((i / 255.0) ** (1.0 / gamma)) * 255 for i in range(256)]).astype("uint8")
img_gamma = cv2.LUT(img, lut)

# Denoise (temporal preferred, here single-frame as example)
img_dn = cv2.fastNlMeansDenoisingColored(img_gamma, None, 5, 5, 7, 21)

# Sharpen (unsharp mask)
blur = cv2.GaussianBlur(img_dn, (0,0), 1.2)
img_sharp = cv2.addWeighted(img_dn, 1.5, blur, -0.5, 0)

# Now run model on img_sharp

When possible, reduce exposure time at capture to curb blur, then raise gain carefully. Combine with temporal voting to stabilize detections.

import numpy as np

ema = None
alpha = 0.4  # higher = more responsive

for frame in stream():
    p = model(frame)  # class probabilities or confidence per label, shape [K]
    if ema is None:
        ema = p
    else:
        ema = alpha * p + (1 - alpha) * ema

    final = (ema > 0.6).astype(int)  # threshold after smoothing
    # Optional: require persistence for 3 consecutive frames before alerting

Use per-track EMA for detection confidences to stabilize object presence across frames.

• Overlay frame index, PTS, queue sizes, and per-stage timings on video.
• Record CPU/GPU utilization; correlate spikes to drops.
• Save short clips around mis-detections and replay offline deterministically.
• Seed RNGs; version your model and preprocessing.

1. Capture and sampling: start at stride 2; switch to stride 1 when motion score > threshold.
2. Tracking: assign IDs; remove tracks after 1 second unseen.
3. Aggregation: maintain set of active IDs each minute; emit counts and a "crowding" boolean if > N.
4. Evaluation: report FPS, latency, drop rate, and IDF1 before/after EMA.

import time

latencies = []
for frame in stream():
    t0 = time.time()
    # capture -> preprocess -> model -> postprocess
    _ = pipeline(frame)
    latencies.append((time.time() - t0) * 1000)

print(f"median: {np.median(latencies):.1f} ms, p95: {np.percentile(latencies,95):.1f} ms")