Real-Time Detection: Making It Fast on a Laptop

capture -> detect -> annotate -> display
 camera    model     draw boxes   window

import os
import threading
import time
from collections import defaultdict
 
import cv2
import numpy as np
 
# ---------------- SETTINGS (change one at a time and measure!) ----------------
SOURCE = "demo_clip.mp4"        # a video file, or 0 for the webcam
DETECT_EVERY = 1                # run the detector on every Nth frame (3 = skip two of every three)
SCALE = 1.0                     # shrink the frame before detecting (0.5 = half width and height)
DETECTOR = "colour"             # "colour" (lesson 8, instant) | "slow" (simulates a heavy model) | "yolo" (needs ultralytics)
SIMULATED_MS = 60               # how long the pretend heavy model takes per frame
SHOW = True                     # False = no window (benchmark only)
MAX_FRAMES = 150                # stop after this many frames when benchmarking (0 = run until the end or 'q')
# ------------------------------------------------------------------------------
 
 
class ColourBall:
    """The lesson 8 detector: finds an orange ball by colour. Fast, but only a stand-in for YOLO."""
    LOWER, UPPER = (8, 150, 120), (25, 255, 255)
 
    def __call__(self, frame):
        hsv = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV)
        mask = cv2.inRange(hsv, self.LOWER, self.UPPER)
        contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
        boxes = []
        for c in contours:
            if cv2.contourArea(c) >= 40:
                x, y, w, h = cv2.boundingRect(c)
                boxes.append((x, y, x + w, y + h, "ball", 0.9))
        return boxes
 
 
class SlowDetector:
    """Wraps another detector and adds a delay, to practise speed-ups without a GPU.
    Real models take roughly time proportional to the number of pixels, so the pretend delay does too."""
 
    def __init__(self, inner, milliseconds, reference_pixels=640 * 360):
        self.inner, self.delay, self.ref = inner, milliseconds / 1000, reference_pixels
 
    def __call__(self, frame):
        time.sleep(self.delay * frame.shape[0] * frame.shape[1] / self.ref)
        return self.inner(frame)
 
 
class YoloDetector:
    def __init__(self, weights="best.pt", conf=0.25):
        from ultralytics import YOLO
        self.model, self.conf = YOLO(weights), conf
 
    def __call__(self, frame):
        result = self.model(frame, conf=self.conf, verbose=False)[0]
        out = []
        for b in result.boxes:
            x1, y1, x2, y2 = (int(v) for v in b.xyxy[0].tolist())
            out.append((x1, y1, x2, y2, self.model.names[int(b.cls[0])], float(b.conf[0])))
        return out
 
 
def make_detector():
    if DETECTOR == "yolo":
        return YoloDetector()
    if DETECTOR == "slow":
        return SlowDetector(ColourBall(), SIMULATED_MS)
    return ColourBall()
 
 
timers = defaultdict(list)
 
 
def timed(name, fn, *args):
    t = time.perf_counter()
    out = fn(*args)
    timers[name].append((time.perf_counter() - t) * 1000)
    return out
 
 
def draw(frame, boxes):
    for (x1, y1, x2, y2, label, conf) in boxes:
        cv2.rectangle(frame, (x1, y1), (x2, y2), (0, 255, 0), 2)
        cv2.putText(frame, f"{label} {conf:.2f}", (x1, max(12, y1 - 4)), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 1, cv2.LINE_AA)
    return frame
 
 
def detect_scaled(detector, frame):
    """Detect on a smaller copy, then scale the boxes back up to the full-size frame."""
    if SCALE == 1.0:
        return detector(frame)
    small = cv2.resize(frame, None, fx=SCALE, fy=SCALE)
    return [(int(x1 / SCALE), int(y1 / SCALE), int(x2 / SCALE), int(y2 / SCALE), label, conf)
            for (x1, y1, x2, y2, label, conf) in detector(small)]
 
 
def main():
    if isinstance(SOURCE, str) and not os.path.exists(SOURCE):
        from demo_clip import make_demo_clip            # the lesson 5 helper
        make_demo_clip(SOURCE)
    cap = cv2.VideoCapture(SOURCE)
    detector = make_detector()
    boxes, frame_no = [], 0
    start = time.perf_counter()
 
    while True:
        ret, frame = timed("1 capture", cap.read)
        if not ret:
            break
        if frame_no % DETECT_EVERY == 0:                             # frame skipping: reuse the last boxes in between
            boxes = timed("2 detect", detect_scaled, detector, frame)
        frame = timed("3 annotate", draw, frame, boxes)
        if SHOW:
            timed("4 display", cv2.imshow, "pipeline", frame)
            if cv2.waitKey(1) & 0xFF == ord("q"):
                break
        frame_no += 1
        if MAX_FRAMES and frame_no >= MAX_FRAMES:
            break
 
    seconds = time.perf_counter() - start
    cap.release()
    cv2.destroyAllWindows()
 
    print(f"\n{frame_no} frames in {seconds:.2f}s  ->  {frame_no / seconds:.1f} FPS   "
          f"(detector={DETECTOR}, every {DETECT_EVERY}, scale {SCALE})")
    print(f"{'stage':<12}{'calls':>7}{'avg ms':>9}{'share of time':>15}")
    total = sum(sum(v) for v in timers.values())
    for name in sorted(timers):
        v = timers[name]
        print(f"{name:<12}{len(v):>7}{np.mean(v):>9.2f}{100 * sum(v) / total:>14.0f}%")
    slowest = max(timers, key=lambda k: sum(timers[k]))
    print("Slowest stage overall:", slowest, " <- speed this one up first")
 
 
if __name__ == "__main__":
    main()

import threading
import time
 
import cv2
 
 
class LatestFrame:
    """Reads the camera in a background thread and always keeps only the NEWEST frame.
    The slow detector never works on old frames, and the camera buffer never fills up (no growing lag)."""
 
    def __init__(self, source=0, capture=None):
        self.cap = capture if capture is not None else cv2.VideoCapture(source)
        self.frame, self.ok, self.running = None, False, True
        self.lock = threading.Lock()
        threading.Thread(target=self._run, daemon=True).start()
 
    def _run(self):
        while self.running:
            ok, frame = self.cap.read()
            with self.lock:
                self.ok = ok
                if ok:
                    self.frame = frame              # keep the last good frame if the camera stops
            if not ok:
                break
 
    def read(self):
        with self.lock:
            return self.ok, (None if self.frame is None else self.frame.copy())
 
    def release(self):
        self.running = False
        self.cap.release()
 
 
# Try it with a pretend 30 FPS camera (a real webcam is slow like this; a video file would be read instantly)
class PretendCamera:
    def __init__(self, path):
        self.inner = cv2.VideoCapture(path)
 
    def read(self):
        time.sleep(1 / 30)
        return self.inner.read()
 
    def release(self):
        self.inner.release()
 
 
grabber = LatestFrame(capture=PretendCamera("demo_clip.mp4"))
time.sleep(0.2)
 
seen = set()
for _ in range(20):
    ok, frame = grabber.read()
    if ok and frame is not None:
        seen.add(int(frame.sum()))         # a rough fingerprint of the frame, to count distinct frames
    time.sleep(0.05)                       # pretend the detector takes 50 ms per frame
grabber.release()
print("distinct frames processed in 1 second:", len(seen), "(the camera made ~30; we kept up with the newest, not the oldest)")

import time
 
import cv2
import numpy as np
from ultralytics import YOLO
 
# 1. Export once (writes best.onnx next to best.pt). Other formats exist, for example "openvino" for Intel CPUs.
YOLO("best.pt").export(format="onnx", imgsz=640)
 
# 2. Compare speed: same frame, same settings
frame = cv2.imread("test_frame.jpg")
for name in ("best.pt", "best.onnx"):
    model = YOLO(name)
    model(frame, verbose=False)                                  # warm-up run (the first call is always slow)
    times = []
    for _ in range(20):
        t = time.perf_counter()
        model(frame, imgsz=640, verbose=False)
        times.append((time.perf_counter() - t) * 1000)
    print(f"{name:12s} {np.mean(times):6.1f} ms per frame  ({1000 / np.mean(times):.1f} FPS)")