Chuyển đến nội dung chính

Bài 6: Real-time Detection — Camera, Video Stream

Object detection real-time: webcam, RTSP stream, video file. Tracking: SORT, DeepSORT, ByteTrack. Counting objects qua line, zone-based detection. Optimization cho FPS cao.

🧠 AI & ML — Bài 5 Bài 6: Real-time Detection — Camera, Video Stream

Computer Vision với Deep Learning: Từ CNN đến Vision Transformer

Phần 2: Object Detection

xdev.asia

Giới thiệu

Detection trên ảnh tĩnh thì dễ. Nhưng real-world cần real-time: camera giám sát, xe tự lái, đếm khách. Bài này cover: video detection, object tracking (theo dõi đối tượng qua frames), counting (đếm qua đường), và tối ưu FPS.

🎯 Mục tiêu: Xây hệ thống detection real-time 30+ FPS với tracking và counting.


1. YOLO trên Video Stream

1.1 Webcam Real-time

"""YOLO real-time trên webcam"""
import cv2
from ultralytics import YOLO
import time

model = YOLO("yolo11n.pt")

cap = cv2.VideoCapture(0)  # 0 = webcam
cap.set(cv2.CAP_PROP_FRAME_WIDTH, 1280)
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 720)

fps_counter = 0
fps_start = time.time()
fps_display = 0

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

    # Detection
    results = model(frame, verbose=False, conf=0.5)
    annotated = results[0].plot()

    # FPS counter
    fps_counter += 1
    elapsed = time.time() - fps_start
    if elapsed >= 1.0:
        fps_display = fps_counter / elapsed
        fps_counter = 0
        fps_start = time.time()

    cv2.putText(annotated, f"FPS: {fps_display:.0f}", (10, 30),
                cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)

    cv2.imshow("YOLO Real-time", annotated)
    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

cap.release()
cv2.destroyAllWindows()

1.2 RTSP / IP Camera Stream

"""YOLO trên IP camera (RTSP stream)"""
from ultralytics import YOLO

model = YOLO("yolo11n.pt")

# RTSP stream
rtsp_url = "rtsp://admin:[email protected]:554/stream1"

# Stream mode — xử lý frame-by-frame, tiết kiệm RAM
results = model(rtsp_url, stream=True, conf=0.25)

for result in results:
    annotated = result.plot()
    cv2.imshow("RTSP Stream", annotated)
    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

2. Object Tracking

2.1 Tại sao cần Tracking?

Detection only:    Frame 1: person A, person B
                   Frame 2: person ?, person ?  ← KHÔNG BIẾT ai là ai!

Detection + Tracking: Frame 1: person A (ID=1), person B (ID=2)
                      Frame 2: person A (ID=1), person B (ID=2) ← BIẾT!

2.2 Tracking Algorithms

AlgorithmCách hoạt độngTốc độAccuracyUse case
SORTKalman Filter + Hungarian⚡ Rất nhanh⭐⭐⭐Realtime, simple
DeepSORTSORT + appearance features (Re-ID)🔥 Nhanh⭐⭐⭐⭐People tracking
ByteTrackTrack cả low-confidence detections⚡ Rất nhanh⭐⭐⭐⭐⭐SOTA, recommended
BoT-SORTByteTrack + camera motion compensation🔥 Nhanh⭐⭐⭐⭐⭐Moving camera

2.3 YOLO + Tracking trong 3 dòng code

"""YOLO Tracking — cực kỳ đơn giản với Ultralytics"""
from ultralytics import YOLO

model = YOLO("yolo11n.pt")

# Track trên video — 1 dòng!
results = model.track(
    source="people_walking.mp4",
    tracker="bytetrack.yaml",  # Hoặc "botsort.yaml"
    show=True,                 # Hiển thị real-time
    conf=0.3,
    persist=True,              # Giữ tracking IDs giữa frames
)

2.4 Tracking chi tiết — Lấy thông tin

"""Phân tích tracking results"""
import cv2
from ultralytics import YOLO
from collections import defaultdict

model = YOLO("yolo11n.pt")
cap = cv2.VideoCapture("crosswalk.mp4")

# Lưu trajectory (đường đi) của mỗi object
track_history = defaultdict(list)

while cap.isOpened():
    ret, frame = cap.read()
    if not ret:
        break

    results = model.track(frame, persist=True, verbose=False)
    result = results[0]

    if result.boxes.id is not None:
        boxes = result.boxes.xywh.cpu().numpy()
        track_ids = result.boxes.id.int().cpu().numpy()
        classes = result.boxes.cls.int().cpu().numpy()

        for box, track_id, cls in zip(boxes, track_ids, classes):
            x, y, w, h = box
            class_name = result.names[cls]

            # Lưu vị trí center
            track_history[track_id].append((float(x), float(y)))

            # Vẽ trajectory (đường đi)
            track = track_history[track_id]
            if len(track) > 1:
                points = [(int(p[0]), int(p[1])) for p in track[-30:]]
                for i in range(1, len(points)):
                    cv2.line(frame, points[i-1], points[i], (0, 255, 0), 2)

            # Info
            cv2.putText(frame, f"ID:{track_id} {class_name}",
                       (int(x-w/2), int(y-h/2)-10),
                       cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 255, 255), 2)

    cv2.imshow("Tracking", frame)
    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

cap.release()
print(f"Total unique objects tracked: {len(track_history)}")

3. Object Counting

3.1 Line Counting — Đếm qua đường kẻ

"""Đếm objects đi qua 1 đường kẻ"""
import cv2
import numpy as np
from ultralytics import YOLO
from collections import defaultdict

model = YOLO("yolo11n.pt")
cap = cv2.VideoCapture("highway.mp4")

# Đường đếm (ngang giữa frame)
frame_width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
frame_height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
count_line_y = frame_height // 2

# Tracking state
track_history = defaultdict(list)
counted_ids = set()
count_up = 0
count_down = 0

while cap.isOpened():
    ret, frame = cap.read()
    if not ret:
        break

    results = model.track(frame, persist=True, verbose=False, classes=[2])  # class 2 = car

    # Vẽ counting line
    cv2.line(frame, (0, count_line_y), (frame_width, count_line_y),
             (0, 0, 255), 2)

    if results[0].boxes.id is not None:
        for box, track_id in zip(results[0].boxes.xywh.cpu(),
                                  results[0].boxes.id.int().cpu()):
            x, y = float(box[0]), float(box[1])
            tid = int(track_id)

            track_history[tid].append(y)

            # Check nếu đi qua line
            if tid not in counted_ids and len(track_history[tid]) >= 2:
                prev_y = track_history[tid][-2]
                curr_y = track_history[tid][-1]

                if prev_y < count_line_y <= curr_y:
                    count_down += 1
                    counted_ids.add(tid)
                elif prev_y > count_line_y >= curr_y:
                    count_up += 1
                    counted_ids.add(tid)

    # Display counts
    cv2.putText(frame, f"Up: {count_up} | Down: {count_down}",
               (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 0), 2)

    cv2.imshow("Vehicle Counter", frame)
    if cv2.waitKey(1) & 0xFF == ord('q'):
        break

cap.release()
print(f"Final count — Up: {count_up}, Down: {count_down}")

3.2 Zone-based Detection — Đếm trong vùng

"""Đếm objects trong 1 vùng (polygon zone)"""
import numpy as np

# Định nghĩa vùng (polygon)
zone_polygon = np.array([
    [200, 300],
    [600, 300],
    [700, 500],
    [100, 500],
], dtype=np.int32)

def point_in_polygon(point, polygon):
    """Kiểm tra point có nằm trong polygon không"""
    return cv2.pointPolygonTest(polygon, point, False) >= 0

# Trong tracking loop:
objects_in_zone = 0
for box in results[0].boxes.xywh.cpu():
    center = (float(box[0]), float(box[1]))
    if point_in_polygon(center, zone_polygon):
        objects_in_zone += 1

# Vẽ zone
cv2.polylines(frame, [zone_polygon], True, (255, 255, 0), 2)
cv2.putText(frame, f"In Zone: {objects_in_zone}",
           (10, 60), cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 0), 2)

4. Tối ưu FPS

4.1 Tips tối ưu speed

"""Các kỹ thuật tăng FPS"""

# 1. Dùng model nhỏ
model = YOLO("yolo11n.pt")  # nano: nhanh nhất

# 2. Giảm input size
results = model(frame, imgsz=320)  # 320 thay vì 640

# 3. Half precision (FP16)
model = YOLO("yolo11n.pt")
results = model(frame, half=True)  # FP16 trên GPU

# 4. Export sang TensorRT
model.export(format="engine", half=True)
trt_model = YOLO("yolo11n.engine")  # 2-3x faster

# 5. Skip frames — không xử lý mọi frame
frame_skip = 2  # Xử lý 1 frame, skip 1
frame_count = 0
while cap.isOpened():
    ret, frame = cap.read()
    frame_count += 1
    if frame_count % frame_skip != 0:
        continue  # Skip frame
    results = model(frame)

# 6. Chỉ detect classes cần thiết
results = model(frame, classes=[0, 2])  # Chỉ person + car

4.2 Benchmark FPS

Cấu hìnhFPS (RTX 3060)FPS (T4)FPS (CPU)
yolo11n, 64018012025
yolo11n, 320300+20045
yolo11s, 6401208012
yolo11m, 64070455
yolo11n TRT, 640350+250—

Tóm tắt

ConceptGhi nhớ
Video DetectionYOLO + OpenCV, stream=True cho video dài
Object TrackingByteTrack (recommended), gán unique ID mỗi object
Line CountingTrack + check qua line → count
Zone DetectionpointPolygonTest để check object trong vùng
FPS OptimizationModel nhỏ, imgsz nhỏ, TensorRT, skip frames

Bài tập tổng hợp

  1. Webcam Detector: YOLO + webcam + hiển thị FPS.
  2. People Counter: Đếm người đi qua 1 cửa (line counting) từ video.
  3. Speed Compare: Benchmark yolo11n vs yolo11s trên video 1 phút. FPS?
  4. Trajectory Vis: Vẽ đường đi (trajectory) của mỗi person qua 30 frames gần nhất.

Bài tiếp theo: Image Segmentation — phân loại từng pixel, semantic vs instance vs panoptic.