Lesson 13: Real-World Project — Real-Time Object Tracker

Frame N:    [YOLO26 detect] → end-to-end boxes → [init trackers]
Frame N+1:  [tracker.update()] → updated positions
Frame N+2:  [tracker.update()] → updated positions
...
Frame 2N:   [YOLO26 detect] → [reinit trackers]

import cv2

# Create individual trackers
tracker_csrt = cv2.legacy.TrackerCSRT_create()
tracker_kcf = cv2.legacy.TrackerKCF_create()
tracker_mosse = cv2.legacy.TrackerMOSSE_create()

from ultralytics import YOLO
import urllib.request
import cv2

# Download the standard Ultralytics test image
url = "https://ultralytics.com/images/bus.jpg"
urllib.request.urlretrieve(url, "bus.jpg")

model = YOLO("yolo26n.pt")
results = model("bus.jpg")
result = results[0]

img = cv2.imread("bus.jpg")
for box in result.boxes:
    cls_id = int(box.cls[0])
    conf = float(box.conf[0])
    x1, y1, x2, y2 = map(int, box.xyxy[0])
    label = f"{result.names[cls_id]}: {conf:.2f}"
    cv2.rectangle(img, (x1, y1), (x2, y2), (0, 255, 0), 2)
    cv2.putText(img, label, (x1, y1 - 8),
                cv2.FONT_HERSHEY_SIMPLEX, 0.55, (0, 255, 0), 2)

cv2.imshow("YOLO26 — bus.jpg test", img)
cv2.waitKey(0)
cv2.destroyAllWindows()

# Expected output:
# 6 persons detected (pedestrians on sidewalk)
# 1 bus (large, center-left of frame)
# 2 cars (partial, right side)
# All boxes tight and accurate, no duplicates (NMS-free)

import cv2
import numpy as np
import time
import urllib.request
from ultralytics import YOLO

# ─── Configuration ────────────────────────────────────────────────────────────
YOLO_MODEL      = "yolo26n.pt"         # Change to yolo26s/m/l/x for more accuracy
CONFIDENCE_THRESH = 0.5
DETECT_EVERY_N  = 10                   # Re-detect every N frames
TARGET_CLASSES  = {0: "person", 2: "car", 3: "motorcycle", 5: "bus", 7: "truck"}

COLORS = np.random.default_rng(42).integers(60, 230, size=(200, 3), dtype=np.uint8)


# ─── Detection ────────────────────────────────────────────────────────────────
class YOLO26Detector:
    """
    Wraps the Ultralytics YOLO26 model.
    YOLO26 is NMS-free: results are end-to-end predictions, ready to use.
    """
    def __init__(self, model_path: str):
        self.model = YOLO(model_path)

    def detect(self, frame: np.ndarray) -> list[tuple]:
        """
        Returns list of (bbox_xywh, confidence, class_id) for TARGET_CLASSES.
        bbox_xywh = (x, y, width, height) in pixels.
        No NMS post-processing needed — YOLO26 output is already end-to-end.
        """
        results = self.model(frame, verbose=False)
        result = results[0]
        detections = []
        for box in result.boxes:
            cls_id = int(box.cls[0])
            if cls_id not in TARGET_CLASSES:
                continue
            conf = float(box.conf[0])
            if conf < CONFIDENCE_THRESH:
                continue
            x1, y1, x2, y2 = map(int, box.xyxy[0])
            detections.append(([x1, y1, x2 - x1, y2 - y1], conf, cls_id))
        return detections


# ─── Tracker Manager ─────────────────────────────────────────────────────────
class TrackerManager:
    """
    Manages a pool of CSRT trackers, one per tracked object.
    Each object gets a persistent numeric ID.
    """
    def __init__(self):
        self.trackers: dict[int, cv2.legacy.TrackerCSRT] = {}
        self.classes:  dict[int, int] = {}
        self.next_id = 0

    def reinit(self, frame: np.ndarray, detections: list[tuple]) -> None:
        """Replace all trackers with fresh ones from new detections."""
        self.trackers.clear()
        self.classes.clear()
        for (x, y, w, h), conf, cls_id in detections:
            tracker = cv2.legacy.TrackerCSRT_create()
            tracker.init(frame, (x, y, w, h))
            self.trackers[self.next_id] = tracker
            self.classes[self.next_id] = cls_id
            self.next_id += 1

    def update(self, frame: np.ndarray) -> dict[int, tuple]:
        """
        Update all trackers. Remove failed ones.
        Returns: {obj_id: ((x, y, w, h), cls_id)}
        """
        active = {}
        failed = []
        for obj_id, tracker in self.trackers.items():
            success, bbox = tracker.update(frame)
            if success:
                active[obj_id] = (bbox, self.classes[obj_id])
            else:
                failed.append(obj_id)
        for obj_id in failed:
            del self.trackers[obj_id]
            del self.classes[obj_id]
        return active


# ─── Annotation ───────────────────────────────────────────────────────────────
def draw_tracked_objects(
    frame: np.ndarray,
    tracked: dict[int, tuple],
    class_names: dict[int, str]
) -> None:
    for obj_id, ((x, y, bw, bh), cls_id) in tracked.items():
        x, y, bw, bh = int(x), int(y), int(bw), int(bh)
        color = COLORS[obj_id % 200].tolist()
        name = class_names.get(cls_id, f"cls{cls_id}")
        label = f"#{obj_id} {name}"

        cv2.rectangle(frame, (x, y), (x + bw, y + bh), color, 2)
        (tw, th), _ = cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, 0.6, 1)
        cv2.rectangle(frame, (x, y - th - 8), (x + tw + 6, y), color, -1)
        cv2.putText(frame, label, (x + 3, y - 4),
                    cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 1)


# ─── Main Loop ────────────────────────────────────────────────────────────────
def main():
    detector    = YOLO26Detector(YOLO_MODEL)
    tracker_mgr = TrackerManager()

    # Use webcam (0) or replace with a file path: cv2.VideoCapture("video.mp4")
    cap = cv2.VideoCapture(0)
    if not cap.isOpened():
        print("Error: cannot open video source")
        return

    src_fps = cap.get(cv2.CAP_PROP_FPS) or 30
    src_w   = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    src_h   = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
    out     = cv2.VideoWriter(
        "tracked_output.mp4",
        cv2.VideoWriter_fourcc(*"mp4v"),
        src_fps, (src_w, src_h)
    )

    frame_count   = 0
    prev_time     = time.perf_counter()
    tracked_objs: dict[int, tuple] = {}

    print("Starting tracker. Press 'q' to quit.")

    while True:
        ret, frame = cap.read()
        if not ret:
            break
        frame_count += 1

        # ── Detection phase (every N frames) ─────────────────────────────
        if frame_count % DETECT_EVERY_N == 1:
            detections = detector.detect(frame)
            # YOLO26: detections are already final — no NMS needed
            tracker_mgr.reinit(frame, detections)

        # ── Tracking phase ────────────────────────────────────────────────
        tracked_objs = tracker_mgr.update(frame)

        # ── Annotation ────────────────────────────────────────────────────
        draw_tracked_objects(frame, tracked_objs, TARGET_CLASSES)

        # ── FPS / info overlay ────────────────────────────────────────────
        curr_time   = time.perf_counter()
        fps_display = 1.0 / (curr_time - prev_time + 1e-9)
        prev_time   = curr_time
        mode        = "DETECT" if frame_count % DETECT_EVERY_N == 1 else "TRACK"
        info        = f"FPS: {fps_display:.1f} | Objects: {len(tracked_objs)} | {mode}"
        cv2.putText(frame, info, (10, 30),
                    cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 255), 2)

        out.write(frame)
        cv2.imshow("YOLO26 Real-Time Tracker", frame)
        if cv2.waitKey(1) & 0xFF == ord("q"):
            break

    cap.release()
    out.release()
    cv2.destroyAllWindows()
    print(f"Done. Processed {frame_count} frames. Output: tracked_output.mp4")


if __name__ == "__main__":
    main()

small = cv2.resize(frame, (640, 360))
detections = detector.detect(small)
# Scale boxes back: multiply x/y/w/h by (orig_w/640, orig_h/360, ...)

# Ultralytics YOLO26 on GPU
model = YOLO("yolo26n.pt")
results = model(frame, device="cuda", verbose=False)

yolo export model=yolo26n.pt format=engine device=0  # TensorRT on Jetson

import cProfile
cProfile.run("main()", sort="cumulative")