Raspberry Pi + OpenCV: Đếm người ra/vào (Object detection & People Counter)

1. Giới thiệu

Project này dựng một hệ thống đếm người ra/vào (people counter) chạy hoàn toàn trên Raspberry Pi 4, dùng Camera Module v3 làm mắt và OpenCV làm bộ não xử lý ảnh. Thay vì cần cảm biến hồng ngoại hay mạch đếm chuyên dụng, ta chỉ cần một camera nhìn xuống lối đi: bộ dò người HOG (Histogram of Oriented Gradients) của OpenCV phát hiện hình dáng người trong từng khung hình, một bộ theo dõi centroid đơn giản (CentroidTracker) gán ID cho từng người và theo dõi vị trí qua các khung hình liên tiếp, rồi so sánh vị trí đó với một vạch ảo nằm ngang để quyết định người đó đang đi vào hay đi ra.

Toàn bộ pipeline chạy bằng Python thuần trên Raspberry Pi OS, không cần huấn luyện mô hình riêng — HOG People Detector là bộ dò có sẵn trong OpenCV, đủ dùng cho các ứng dụng đếm lưu lượng người ở quy mô nhỏ (cửa hàng, phòng học, phòng lab). Vì đây là bộ dò cổ điển (không phải deep-learning hiện đại như YOLO), độ chính xác sẽ giảm khi nhiều người đứng che khuất nhau hoặc ánh sáng yếu — phần "Lỗi thường gặp" ở cuối bài sẽ nói rõ giới hạn này.

2. Linh kiện cần có

3. Sơ đồ kết nối

Kết nối chính là cáp ribbon CSI 15-chân đi kèm camera — cắm thẳng vào cổng CSI trên Raspberry Pi 4 (nằm giữa cổng HDMI và jack audio).

Lưu ý chiều cáp: mặt có các đường tiếp xúc kim loại (dây đồng) phải hướng về phía cổng HDMI (ngược với hướng cổng USB); cắm ngược chiều là nguyên nhân phổ biến nhất khiến hệ thống báo "no cameras available". Nhớ mở khoá chốt nhựa trên đầu nối CSI trước khi lắp cáp, luồn cáp vào rồi ấn khoá lại chắc chắn.

Bốn dây SDA/SCL/3V3/GND chỉ áp dụng cho các board có route thêm chân I2C điều khiển autofocus của Camera Module v3 ra ngoài đầu nối phụ — nếu module của bạn chỉ có đầu nối ribbon CSI đơn thuần (không có 4 chân rời), bỏ qua phần này, chỉ cần cáp CSI là đủ để camera hoạt động (autofocus vẫn chạy qua chính cáp CSI ở đa số board chính hãng).

4. Code ví dụ #1 — Kiểm tra camera hoạt động (Camera Self-Test)

Chạy script này đầu tiên sau khi lắp camera để xác nhận phần cứng nhận diện đúng trước khi chạy ứng dụng chính. Script in banner khởi động, thử mở camera qua picamera2, chụp thử vài khung hình trong 3 giây và báo FPS ước tính.

Nếu thiếu thư viện hoặc phần cứng lỗi, script báo lỗi rõ ràng thay vì treo máy.

"""
camera_test.py
Kiem tra nhanh Camera Module v3 hoat dong tren Raspberry Pi 4 truoc khi
chay bai toan dem nguoi. In banner khoi dong, thu mo camera qua picamera2,
chup vai khung hinh, in kich thuoc + FPS uoc tinh. Neu khong co camera/
picamera2 (vi du chay tren may tinh thuong de kiem tra code), script se
bao loi ro rang thay vi treo may.
"""

import sys
import time


def main() -> int:
    print("[BOOT] Raspberry Pi People Counter - Camera Self-Test")
    print("[BOOT] Dang khoi tao picamera2...")

    try:
        from picamera2 import Picamera2
    except ImportError:
        print("[ERROR] Khong tim thay thu vien picamera2.")
        print("        Cai dat: sudo apt install -y python3-picamera2")
        return 1

    try:
        picam2 = Picamera2()
        config = picam2.create_preview_configuration(
            main={"size": (640, 480), "format": "RGB888"}
        )
        picam2.configure(config)
        picam2.start()
        time.sleep(1.0)  # cho cam ban do phoi sang tu dong on dinh
    except Exception as exc:  # noqa: BLE001 - can bao loi phan cung ro rang
        print(f"[ERROR] Khong the khoi dong camera: {exc}")
        print("        Kiem tra: cap ribbon CSI cam dung chieu, dung chan,")
        print("        camera duoc bat trong raspi-config (Interface Options).")
        return 1

    frame_count = 0
    start = time.monotonic()
    test_duration_s = 3.0

    while time.monotonic() - start < test_duration_s:
        frame = picam2.capture_array()
        frame_count += 1
        if frame_count == 1:
            print(f"[OK] Khung hinh dau tien: shape={frame.shape}, dtype={frame.dtype}")

    elapsed = time.monotonic() - start
    fps = frame_count / elapsed if elapsed > 0 else 0.0
    print(f"[OK] Da chup {frame_count} khung hinh trong {elapsed:.2f}s (~{fps:.1f} FPS)")

    picam2.stop()
    print("[DONE] Camera hoat dong binh thuong.")
    return 0


if __name__ == "__main__":
    sys.exit(main())

5. Code ví dụ #2 — Ứng dụng chính: đếm người qua vạch ảo (people_counter.py)

Đây là ứng dụng chính, chạy vòng lặp liên tục: đọc khung hình từ camera, chạy HOG People Detector để tìm vị trí (centroid) từng người, đưa qua CentroidTracker để gán ID ổn định qua các khung hình, rồi so sánh tọa độ y hiện tại và trước đó của mỗi ID với vạch đếm nằm ngang (giữa khung hình) để xác định hướng đi vào/đi ra.

Script tuân thủ các quy tắc mô phỏng bắt buộc: banner khởi động ngay sau khi console sẵn sàng, retry có giới hạn (không lặp vô hạn) khi mở camera thất bại, bỏ qua khung hình lỗi thay vì crash, và in telemetry JSON định kỳ mỗi 5 giây.

"""
people_counter.py
Ung dung chinh: dem nguoi ra/vao qua mot vach ao nam ngang, dung Raspberry
Pi 4 + Camera Module v3 + OpenCV HOG People Detector + CentroidTracker
(tracker_utils.py).

Hanh vi mo phong quan sat duoc (bat buoc theo chuan project IoTLabs Maker):
  - In banner ngay sau khi Serial/console san sang.
  - Neu khong mo duoc camera: thu lai theo chu ky (khong while vo han),
    log ro nguyen nhan, roi thoat sau so lan thu quy dinh.
  - Neu mot khung hinh loi/decode hong: bo qua khung do, telemetry dan
    "detections": null cho khung do thay vi crash.
  - In telemetry JSON dinh ky (moi TELEMETRY_INTERVAL_S giay) gom
    total_in/total_out/nguoi dang trong khung.
"""

from __future__ import annotations

import json
import time
from typing import List, Optional, Tuple

from tracker_utils import CentroidTracker, CsvEventLogger

FRAME_WIDTH = 640
FRAME_HEIGHT = 480
COUNT_LINE_Y = FRAME_HEIGHT // 2  # vach ao nam ngang giua khung hinh
TELEMETRY_INTERVAL_S = 5.0
CAMERA_RETRY_INTERVAL_S = 3.0
CAMERA_MAX_RETRIES = 5
CSV_LOG_PATH = "people_counter_events.csv"


def open_camera():
    """Mo Camera Module v3 qua picamera2, retry co gioi han thay vi while vo han."""
    from picamera2 import Picamera2

    for attempt in range(1, CAMERA_MAX_RETRIES + 1):
        try:
            picam2 = Picamera2()
            config = picam2.create_preview_configuration(
                main={"size": (FRAME_WIDTH, FRAME_HEIGHT), "format": "RGB888"}
            )
            picam2.configure(config)
            picam2.start()
            time.sleep(0.5)
            print(f"[BOOT] Camera san sang sau {attempt} lan thu.")
            return picam2
        except Exception as exc:  # noqa: BLE001
            print(f"[WARN] Lan thu {attempt}/{CAMERA_MAX_RETRIES} mo camera that bai: {exc}")
            if attempt < CAMERA_MAX_RETRIES:
                time.sleep(CAMERA_RETRY_INTERVAL_S)

    print("[ERROR] Khong mo duoc camera sau nhieu lan thu, dung chuong trinh.")
    return None


def detect_people_centroids(frame, hog) -> List[Tuple[int, int]]:
    """Chay HOG People Detector tren mot khung hinh, tra ve danh sach centroid.

    frame: ndarray RGB888 tu picamera2. Neu khung hinh None/rong, tra ve [].
    """
    import cv2

    if frame is None or frame.size == 0:
        return []

    gray_ready = cv2.cvtColor(frame, cv2.COLOR_RGB2BGR)
    boxes, _weights = hog.detectMultiScale(
        gray_ready, winStride=(8, 8), padding=(8, 8), scale=1.05
    )

    centroids: List[Tuple[int, int]] = []
    for (x, y, w, h) in boxes:
        cx = x + w // 2
        cy = y + h // 2
        centroids.append((cx, cy))
    return centroids


def build_telemetry(total_in: int, total_out: int, active_objects: int, detections: Optional[int]) -> str:
    payload = {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "total_in": total_in,
        "total_out": total_out,
        "currently_tracked": active_objects,
        "detections_this_frame": detections,
    }
    return json.dumps(payload, ensure_ascii=False)


def main() -> int:
    print("[BOOT] Raspberry Pi People Counter khoi dong...")
    print(f"[BOOT] Vach dem: y={COUNT_LINE_Y}px, khung hinh {FRAME_WIDTH}x{FRAME_HEIGHT}")

    try:
        import cv2
    except ImportError:
        print("[ERROR] Thieu thu vien opencv-python. Cai dat: pip install opencv-python")
        return 1

    picam2 = open_camera()
    if picam2 is None:
        return 1

    hog = cv2.HOGDescriptor()
    hog.setSVMDetector(cv2.HOGDescriptor_getDefaultPeopleDetector())

    tracker = CentroidTracker(max_distance=80.0, max_disappeared=15)
    logger = CsvEventLogger(CSV_LOG_PATH)

    total_in = 0
    total_out = 0
    last_telemetry_at = time.monotonic()
    # Nho vi tri y truoc do cua tung object de biet huong bang qua vach
    previous_y_by_id = {}

    try:
        while True:
            try:
                frame = picam2.capture_array()
            except Exception as exc:  # noqa: BLE001
                print(f"[WARN] Loi doc khung hinh, bo qua: {exc}")
                frame = None

            detections_count: Optional[int] = None
            if frame is not None:
                centroids = detect_people_centroids(frame, hog)
                detections_count = len(centroids)
                tracked = tracker.update(centroids)

                for object_id, obj in tracked.items():
                    cx, cy = obj.centroid
                    prev_y = previous_y_by_id.get(object_id)
                    previous_y_by_id[object_id] = cy

                    if prev_y is None or obj.counted:
                        continue

                    crossed_downward = prev_y < COUNT_LINE_Y <= cy
                    crossed_upward = prev_y > COUNT_LINE_Y >= cy

                    if crossed_downward:
                        total_in += 1
                        obj.counted = True
                        logger.log(object_id, "in", total_in, total_out)
                        print(f"[EVENT] Nguoi #{object_id} di VAO. Tong vao={total_in}")
                    elif crossed_upward:
                        total_out += 1
                        obj.counted = True
                        logger.log(object_id, "out", total_in, total_out)
                        print(f"[EVENT] Nguoi #{object_id} di RA. Tong ra={total_out}")
            else:
                tracker.update([])

            now = time.monotonic()
            if now - last_telemetry_at >= TELEMETRY_INTERVAL_S:
                print(build_telemetry(total_in, total_out, len(tracker.objects), detections_count))
                last_telemetry_at = now

            time.sleep(0.03)  # ~30 khung hinh/giay toi da, giam tai CPU
    except KeyboardInterrupt:
        print("[STOP] Nhan Ctrl+C, dang dung...")
    finally:
        logger.close()
        picam2.stop()
        print(f"[DONE] Tong ket: vao={total_in}, ra={total_out}")

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

6. Code ví dụ #3 — Module hỗ trợ: theo dõi centroid và ghi log (tracker_utils.py)

Module này tách riêng hai tiện ích mà people_counter.py import và sử dụng: CentroidTracker (thuật toán ghép centroid tham lam theo khoảng cách Euclid, tạo/xoá ID theo số khung hình không xuất hiện) và CsvEventLogger (ghi mỗi sự kiện vào/ra ra file CSV kèm timestamp, tự tạo file + header nếu chưa có). Tách thành module riêng giúp dễ unit-test thuật toán tracking độc lập với phần xử lý ảnh/camera.

"""
tracker_utils.py
Tien ich dung chung cho people_counter.py:
  - CentroidTracker: gan ID on dinh cho tung nguoi qua cac khung hinh lien
    tiep, dua tren khoang cach Euclid giua centroid cu va centroid moi.
  - CsvEventLogger: ghi moi su kien vao/ra ra file CSV de xem lai lich su.

Khong phu thuoc OpenCV/picamera2 truc tiep - chi dung math thuan de de
unit-test va compile doc lap.
"""

from __future__ import annotations

import csv
import math
import os
import time
from dataclasses import dataclass, field
from typing import Dict, List, Tuple


Point = Tuple[int, int]


@dataclass
class TrackedObject:
    object_id: int
    centroid: Point
    last_seen_at: float = field(default_factory=time.monotonic)
    counted: bool = False


class CentroidTracker:
    """Bo theo doi centroid don gian (khong dung Kalman/deep-sort).

    Moi khung hinh, goi update(centroids_moi) voi danh sach centroid phat
    hien duoc. Tracker se ghep centroid moi voi object da biet gan nhat
    (trong nguong max_distance), tao object moi cho centroid khong ghep
    duoc, va xoa object khong xuat hien qua max_disappeared khung hinh.
    """

    def __init__(self, max_distance: float = 80.0, max_disappeared: int = 15) -> None:
        self.next_object_id = 0
        self.objects: Dict[int, TrackedObject] = {}
        self._disappeared: Dict[int, int] = {}
        self.max_distance = max_distance
        self.max_disappeared = max_disappeared

    def _register(self, centroid: Point) -> int:
        object_id = self.next_object_id
        self.objects[object_id] = TrackedObject(object_id=object_id, centroid=centroid)
        self._disappeared[object_id] = 0
        self.next_object_id += 1
        return object_id

    def _deregister(self, object_id: int) -> None:
        self.objects.pop(object_id, None)
        self._disappeared.pop(object_id, None)

    @staticmethod
    def _distance(a: Point, b: Point) -> float:
        return math.hypot(a[0] - b[0], a[1] - b[1])

    def update(self, input_centroids: List[Point]) -> Dict[int, TrackedObject]:
        if not input_centroids:
            for object_id in list(self._disappeared.keys()):
                self._disappeared[object_id] += 1
                if self._disappeared[object_id] > self.max_disappeared:
                    self._deregister(object_id)
            return self.objects

        if not self.objects:
            for centroid in input_centroids:
                self._register(centroid)
            return self.objects

        object_ids = list(self.objects.keys())
        object_centroids = [self.objects[oid].centroid for oid in object_ids]

        # Ma tran khoang cach object_cu x centroid_moi, ghep tham lam
        # (greedy) theo khoang cach nho nhat truoc - du dung cho so luong
        # nguoi it trong khung hinh cua bai toan nay.
        unmatched_rows = set(range(len(object_centroids)))
        unmatched_cols = set(range(len(input_centroids)))
        pairs: List[Tuple[int, int, float]] = []
        for row, oc in enumerate(object_centroids):
            for col, ic in enumerate(input_centroids):
                pairs.append((row, col, self._distance(oc, ic)))
        pairs.sort(key=lambda p: p[2])

        for row, col, dist in pairs:
            if row not in unmatched_rows or col not in unmatched_cols:
                continue
            if dist > self.max_distance:
                continue
            object_id = object_ids[row]
            self.objects[object_id].centroid = input_centroids[col]
            self.objects[object_id].last_seen_at = time.monotonic()
            self._disappeared[object_id] = 0
            unmatched_rows.discard(row)
            unmatched_cols.discard(col)

        for row in unmatched_rows:
            object_id = object_ids[row]
            self._disappeared[object_id] += 1
            if self._disappeared[object_id] > self.max_disappeared:
                self._deregister(object_id)

        for col in unmatched_cols:
            self._register(input_centroids[col])

        return self.objects


class CsvEventLogger:
    """Ghi su kien vao/ra ra CSV, tu tao file + header neu chua co."""

    def __init__(self, csv_path: str) -> None:
        self.csv_path = csv_path
        is_new_file = not os.path.exists(csv_path)
        self._file = open(csv_path, "a", newline="", encoding="utf-8")
        self._writer = csv.writer(self._file)
        if is_new_file:
            self._writer.writerow(["timestamp_iso", "object_id", "direction", "total_in", "total_out"])
            self._file.flush()

    def log(self, object_id: int, direction: str, total_in: int, total_out: int) -> None:
        timestamp_iso = time.strftime("%Y-%m-%dT%H:%M:%S")
        self._writer.writerow([timestamp_iso, object_id, direction, total_in, total_out])
        self._file.flush()

    def close(self) -> None:
        self._file.close()

7. Lỗi thường gặp

8. Tổng kết

Với Raspberry Pi 4, Camera Module v3 và một vòng lặp OpenCV không quá 200 dòng code, bạn đã có một hệ thống đếm người ra/vào hoạt động thời gian thực, log lại từng sự kiện ra CSV để phân tích sau.

Đây là nền tảng tốt để mở rộng: thêm cảnh báo khi vượt ngưỡng sức chứa phòng, đẩy dữ liệu lên dashboard MQTT/IoT, hoặc thay HOG bằng mô hình deep-learning nhẹ để tăng độ chính xác trong điều kiện đông người.

Vì đây là bộ dò cổ điển chạy trên CPU, hãy kỳ vọng hiệu năng vừa phải (vài FPS) và độ chính xác giảm trong điều kiện ánh sáng kém hoặc mật độ người cao — phù hợp cho demo/học tập và các không gian lưu lượng thấp-trung bình hơn là triển khai đếm chính xác tuyệt đối ở quy mô lớn.