Video capture and processing

VideoCapture and VideoWriter, frame loops and timing, background subtraction, optical flow, and a simple tracker that survives occlusion.

Reading and writing video

import cv2
import time

capture = cv2.VideoCapture(0)                  # 0 is the first camera
capture.set(cv2.CAP_PROP_FRAME_WIDTH, 1280)
capture.set(cv2.CAP_PROP_FRAME_HEIGHT, 720)
capture.set(cv2.CAP_PROP_FPS, 30)

info = {
    "width": capture.get(cv2.CAP_PROP_FRAME_WIDTH),
    "height": capture.get(cv2.CAP_PROP_FRAME_HEIGHT),
    "fps": capture.get(cv2.CAP_PROP_FPS),
    "frames": capture.get(cv2.CAP_PROP_FRAME_COUNT),
    "fourcc": int(capture.get(cv2.CAP_PROP_FOURCC)),
}
print(info)

# a file source has a known frame count; a camera reports -1
file_capture = cv2.VideoCapture("clip.mp4")
print("frames", file_capture.get(cv2.CAP_PROP_FRAME_COUNT),
      "fps", file_capture.get(cv2.CAP_PROP_FPS))

writer = None
frame_index = 0
start = time.perf_counter()

while True:
    ok, frame = capture.read()
    if not ok:
        break                                   # end of file, or the camera failed

    if writer is None:
        h, w = frame.shape[:2]
        writer = cv2.VideoWriter("out.mp4", cv2.VideoWriter_fourcc(*"mp4v"), 30, (w, h))

    writer.write(frame)
    frame_index += 1
    if frame_index % 30 == 0:
        elapsed = time.perf_counter() - start
        print(f"frame {frame_index}  {frame_index / elapsed:.1f} fps")

capture.release()
file_capture.release()
if writer is not None:
    writer.release()
print("processed", frame_index, "frames")

• Always check the boolean returned by read(). Reading past the end of a file returns False with an empty frame, and using it produces a confusing error much later.
• VideoWriter needs the exact frame size of the frames you write. A mismatch produces a file that plays as a black rectangle with no error message.
• The codec determines whether the file is playable. mp4v is broadly supported; XVID into an .avi container is the most portable fallback.
• Measure real throughput rather than assuming the configured FPS. Camera capture often returns faster or slower than requested, and processing usually sets the actual rate.

Background subtraction and optical flow

# background subtraction: model the static scene, everything else is foreground
subtractor = cv2.createBackgroundSubtractorMOG2(
    history=500, varThreshold=25, detectShadows=True)

kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5))

def foreground_mask(frame):
    mask = subtractor.apply(frame)
    mask[mask == 127] = 0                      # shadow pixels are marked 127
    mask = cv2.morphologyEx(mask, cv2.MORPH_OPEN, kernel, iterations=1)
    mask = cv2.morphologyEx(mask, cv2.MORPH_CLOSE, kernel, iterations=2)
    mask = cv2.dilate(mask, kernel, iterations=1)
    return mask

# an improved version that handles shadows better (KNN)
knn = cv2.createBackgroundSubtractorKNN(history=500, dist2Threshold=400,
                                        detectShadows=True)

# optical flow: dense motion between two frames
frame_a = cv2.cvtColor(cv2.imread("frame_a.png"), cv2.COLOR_BGR2GRAY)
frame_b = cv2.cvtColor(cv2.imread("frame_b.png"), cv2.COLOR_BGR2GRAY)
flow = cv2.calcOpticalFlowFarneback(
    frame_a, frame_b, None,
    pyr_scale=0.5, levels=3, winsize=15,
    iterations=3, poly_n=5, poly_sigma=1.2, flags=0)

magnitude, angle = cv2.cartToPolar(flow[:, :, 0], flow[:, :, 1])
print("mean flow magnitude", round(float(magnitude.mean()), 2))

hsv_flow = np.zeros((*frame_a.shape, 3), dtype="uint8")
hsv_flow[:, :, 0] = (angle * 180 / np.pi / 2).astype("uint8")
hsv_flow[:, :, 1] = 255
hsv_flow[:, :, 2] = cv2.normalize(magnitude, None, 0, 255, cv2.NORM_MINMAX).astype("uint8")
visualised = cv2.cvtColor(hsv_flow, cv2.COLOR_HSV2BGR)

# sparse flow: track a set of good corners, cheaper and easier to interpret
corners = cv2.goodFeaturesToTrack(frame_a, maxCorners=100, qualityLevel=0.01,
                                  minDistance=10)
next_points, status, error = cv2.calcOpticalFlowPyrLK(frame_a, frame_b, corners, None)
good_new = next_points[status == 1]
good_old = corners[status == 1]
print("tracked points", len(good_new), "of", len(corners))

• Background subtraction needs a warm-up period. The first thirty frames are used to learn the background, so foreground detection is meaningless before that.
• MOG2 marks shadows as 127, not 255. If you do not zero them, shadows appear as solid foreground objects.
• A camera that moves invalidates every background model. Use optical flow or a feature-based tracker instead when the camera is not fixed.
• Dense Farneback flow is expensive. Downscale the frames for the flow computation and scale the resulting vectors back up if you need real-time performance.

Tracking and timing

# the built-in trackers: CSRT is accurate, KCF is fast
tracker_factories = {
    "csrt": cv2.TrackerCSRT_create,
    "kcf": cv2.TrackerKCF_create,
    "mosse": getattr(cv2, "legacy", cv2).TrackerMOSSE_create,
}

def track_in_video(path, initial_box, tracker_name="csrt", max_frames=500):
    capture = cv2.VideoCapture(path)
    ok, frame = capture.read()
    if not ok:
        return []

    tracker = tracker_factories[tracker_name]()
    tracker.init(frame, initial_box)

    trajectory = []
    frame_index = 0
    while frame_index < max_frames:
        ok, frame = capture.read()
        if not ok:
            break
        success, box = tracker.update(frame)
        if success:
            x, y, w, h = [int(v) for v in box]
            trajectory.append((frame_index, x, y, w, h))
        else:
            # on failure the tracker does not recover: reinitialise, or stop
            break
        frame_index += 1

    capture.release()
    return trajectory

def track_with_timing(capture, frames=120):
    """Report the per-stage cost against the frame budget."""
    budget_ms = 1000.0 / 30
    timings = {"decode": 0.0, "process": 0.0}
    for _ in range(frames):
        t0 = time.perf_counter()
        ok, frame = capture.read()
        t1 = time.perf_counter()
        if not ok:
            break
        mask = foreground_mask(frame)
        t2 = time.perf_counter()
        timings["decode"] += (t1 - t0) * 1000
        timings["process"] += (t2 - t1) * 1000
    return {k: round(v / frames, 2) for k, v in timings.items()}, round(budget_ms, 2)

print("per-stage milliseconds and budget:", track_with_timing(capture))

FAQ

Related

Detection with cascades and DNN modules Performance for real-time vision

Last refreshed 2026-09-18.