Performance for real-time vision

NumPy vectorisation, avoiding copies, ROI views, threading and optimisation flags, GPU and OpenCL modules, and budgeting a frame.

Vectorise instead of looping

import cv2
import numpy as np
import time

image = cv2.imread("frame.jpg")
gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)

# slow: a Python loop over every pixel
start = time.perf_counter()
h, w = gray.shape
slow = np.zeros_like(gray)
for y in range(h):
    for x in range(w):
        v = int(gray[y, x])
        slow[y, x] = 255 if v > 128 else 0
slow_ms = (time.perf_counter() - start) * 1000

# fast: one vectorised operation
start = time.perf_counter()
fast = np.where(gray > 128, 255, 0).astype("uint8")
fast_ms = (time.perf_counter() - start) * 1000

# fastest: OpenCV, which is compiled and threaded
start = time.perf_counter()
_, fastest = cv2.threshold(gray, 128, 255, cv2.THRESH_BINARY)
fastest_ms = (time.perf_counter() - start) * 1000

print(f"loop {slow_ms:.1f} ms  numpy {fast_ms:.2f} ms  opencv {fastest_ms:.2f} ms")

# the same principle for a region of interest
roi = image[200:400, 300:600]              # a view, not a copy
roi_mean = roi.reshape(-1, 3).mean(axis=0)  # one reduction, no Python loop

• Reach for a vectorised expression before writing a loop. If no built-in exists, express it as an array operation over the whole image at once.
• Check that an operation is not already available: a surprising number of hand-written kernels are one line of OpenCV.
• NumPy creates a temporary array for every intermediate result. In a hot loop, passing out= to reuse a buffer avoids most of that allocation.
• Avoid converting dtype in a loop. Convert once, work in float32 where needed, convert back once.

Copies, views and threading

# a slice is a view: writing to it changes the parent
region = image[100:200, 100:200]
region[:] = 0                        # this clears the region in image as well

# an explicit copy detaches it
copy_region = image[100:200, 100:200].copy()
copy_region[:] = 0                   # image is unchanged

# check whether an array owns its data
print(image.flags["OWNDATA"], region.flags["OWNDATA"], copy_region.flags["OWNDATA"])
print(image.flags["C_CONTIGUOUS"])   # negative strides break this

# resize into a preallocated buffer to avoid an allocation per frame
target = np.empty((480, 640, 3), dtype="uint8")
cv2.resize(image, (640, 480), dst=target, interpolation=cv2.INTER_AREA)

# OpenCV parallelises internally: set the thread count deliberately
print("threads", cv2.getNumThreads())
cv2.setNumThreads(4)
cv2.setUseOptimized(True)            # enable the dispatch to optimised CPU code
print("optimised", cv2.useOptimized())

# use IPP and OpenCL only after measuring with them off
if cv2.ocl.haveOpenCL():
    cv2.ocl.setUseOpenCL(False)      # CPU often beats OpenCL for small frames
    print("OpenCL available", cv2.ocl.haveOpenCL())

# pipelining across threads: decode on one thread, process on another
import threading
import queue

frame_queue: "queue.Queue" = queue.Queue(maxsize=4)

def reader(capture, out_queue):
    while True:
        ok, frame = capture.read()
        if not ok:
            out_queue.put(None)
            break
        if out_queue.full():
            try:
                out_queue.get_nowait()      # drop the oldest frame
            except queue.Empty:
                pass
        out_queue.put(frame)

thread = threading.Thread(target=reader, args=(cv2.VideoCapture(0), frame_queue),
                          daemon=True)
thread.start()
print("reader started, queue size", frame_queue.qsize())

• A slice is a view. The most common performance bug is not a slow operation but an accidental copy per frame from a slice that looked free.
• For a camera at 30 fps, drop frames rather than queueing them. A growing queue guarantees increasing latency, and the newest frame is always the most useful.
• Python threads do not give you parallel CPU work because of the GIL, but they do overlap blocking I/O such as camera capture with processing.
• Test cv2.setNumThreads both ways. When several worker processes each spawn many threads, oversubscription makes everything slower.

Measuring a frame budget

import statistics
import time

class FrameProfiler:
    def __init__(self, target_fps=30):
        self.budget_ms = 1000.0 / target_fps
        self.samples = {}

    def time(self, name):
        return _Timer(self, name)

    def add(self, name, milliseconds):
        self.samples.setdefault(name, []).append(milliseconds)

    def report(self):
        rows = []
        for name, values in self.samples.items():
            recent = values[-120:]
            rows.append((name,
                         round(statistics.median(recent), 2),
                         round(max(recent), 2)))
        total = sum(row[1] for row in rows)
        return {
            "stages": rows,
            "median_total_ms": round(total, 2),
            "budget_ms": round(self.budget_ms, 2),
            "headroom_ms": round(self.budget_ms - total, 2),
            "achievable_fps": round(1000.0 / max(total, 1e-6), 1),
        }

class _Timer:
    def __init__(self, profiler, name):
        self.profiler = profiler
        self.name = name

    def __enter__(self):
        self.start = time.perf_counter()
        return self

    def __exit__(self, *exc):
        elapsed = (time.perf_counter() - self.start) * 1000
        self.profiler.add(self.name, elapsed)
        return False

profiler = FrameProfiler(target_fps=30)

for index in range(120):
    frame = cv2.imread("frame.jpg")
    with profiler.time("decode"):
        pass
    with profiler.time("preprocess"):
        small = cv2.resize(frame, (640, 480), interpolation=cv2.INTER_AREA)
        gray = cv2.cvtColor(small, cv2.COLOR_BGR2GRAY)
    with profiler.time("detect"):
        _, mask = cv2.threshold(gray, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)
    with profiler.time("postprocess"):
        contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)

report = profiler.report()
for name, median, worst in report["stages"]:
    print(f"{name:14s} median {median:7.2f} ms  worst {worst:7.2f} ms")
print("total", report["median_total_ms"], "budget", report["budget_ms"],
      "achievable", report["achievable_fps"], "fps")

• Report the median and the worst case. A pipeline with a good median and an occasional 200 ms frame is one that users notice as a stutter.
• Process at the smallest resolution that still works. Halving the width quarters the pixel count and speeds up most stages by close to that factor.
• Draw the budget explicitly: decode, preprocess, detect, postprocess, annotate, display. The remaining headroom tells you how much accuracy you can afford to buy.
• Optimise the largest stage first. A 20% improvement in a stage that is 10% of the frame time is worth far less than finding one that is 60%.

FAQ

Related

Video capture and processing Drawing, annotation and pipeline structure

Last refreshed 2026-09-18.