Dưới đây là quy trình chi tiết từng bước xây dựng một sản phẩm MVP (Minimum Viable Product) thực tế: Hệ thống giám sát và đếm đối tượng cắt qua vạch ảo trên máy tính cá nhân (chạy trực tiếp trên môi trường local, tận dụng webcam hoặc video file, xuất giao diện Web tương tác thời gian thực).
Một hệ thống thị giác máy tính tiêu chuẩn gồm 4 tầng dữ liệu vận hành theo mô hình đường ống (Pipeline):
┌─────────────────┐
│ INPUT LAYER │ Camera RTSP / Webcam USB / Video File MP4 (cv2.VideoCapture)
└────────┬────────┘
│ (Frame thô: BGR NumPy array)
▼
┌─────────────────┐
│ PROCESSING CORE │ YOLOv8 Inference ──> Bounding Boxes [x1, y1, x2, y2, conf, cls]
│ (AI & Tracking) │ │
│ │ Centroid Tracker ──> Gán ID ổn định qua thời gian: {ID: (cx, cy)}
└────────┬────────┘
│ (Tọa độ tâm + ID qua các frame)
▼
┌─────────────────┐
│ BACKEND LOGIC │ Virtual Tripwire (Vector giao cắt vạch kẻ ảo)
│ & STORAGE │ ├─ Phân loại chiều di chuyển: IN (Vào) / OUT (Ra)
│ │ └─ Ghi log sự kiện vào SQLite Database: (id, timestamp, direction)
└────────┬────────┘
│ (Frame đã vẽ HUD + Số liệu thống kê)
▼
┌─────────────────┐
│ PRESENTATION │ Streamlit Dashboard (Hiển thị luồng video, KPI đếm, bảng log)
└─────────────────┘
Mở Terminal (macOS/Linux) hoặc PowerShell (Windows), tạo thư mục dự án và cài đặt các thư viện cần thiết:
mkdir cv_tripwire_mvp
cd cv_tripwire_mvp
# Tạo và kích hoạt môi trường ảo Python
python -m venv venv
# Trên Windows:
venv\Scripts\activate
# Trên Linux/macOS:
source venv/bin/activate
# Cài đặt các thư viện
pip install opencv-python ultralytics streamlit numpy pandas
Cấu trúc dự án sẽ tổ chức thành các module tách biệt:
cv_tripwire_mvp/
├── tracker.py # Thuật toán gán ID và theo dõi vết tâm vật thể
├── db.py # Khởi tạo và ghi log vào SQLite
├── core.py # Lõi xử lý: Đọc video, YOLOv8, kiểm tra cắt vạch
└── app.py # Giao diện Web Dashboard bằng Streamlit
tracker.py)Khi mô hình YOLO chạy trên từng frame, nó chỉ biết "ở đây có xe/sản phẩm" nhưng không biết vật thể ở frame hiện tại có phải là vật thể ở frame trước hay không. Ta cần Centroid Tracker dùng khoảng cách Euclidean để gán một ID cố định cho mỗi vật thể khi nó di chuyển.
Tạo file tracker.py:
import math
class CentroidTracker:
def __init__(self, max_disappeared=15, max_distance=60):
# Lưu trữ tọa độ tâm hiện tại: {object_id: (cx, cy)}
self.objects = {}
# Lưu vết tọa độ trước đó để tính hướng đi: {object_id: (prev_cx, prev_cy)}
self.previous_positions = {}
# Đếm số frame vật thể bị mất dấu
self.disappeared = {}
self.next_object_id = 0
self.max_disappeared = max_disappeared
self.max_distance = max_distance
def update(self, rects):
# rects: danh sách bounding box [[x1, y1, x2, y2], ...]
if len(rects) == 0:
for obj_id in list(self.disappeared.keys()):
self.disappeared[obj_id] += 1
if self.disappeared[obj_id] > self.max_disappeared:
self._deregister(obj_id)
return self.objects
# Tính toán trọng tâm (Centroid) của các box mới phát hiện
input_centroids = []
for (x1, y1, x2, y2) in rects:
cx = int((x1 + x2) / 2.0)
cy = int((y1 + y2) / 2.0)
input_centroids.append((cx, cy))
# Nếu chưa theo dõi vật thể nào, đăng ký mới toàn bộ
if len(self.objects) == 0:
for centroid in input_centroids:
self._register(centroid)
else:
object_ids = list(self.objects.keys())
object_centroids = list(self.objects.values())
# Ghép cặp tâm mới với tâm cũ dựa trên khoảng cách Euclidean nhỏ nhất
used_rows = set()
used_cols = set()
for r, (obj_id, old_centroid) in enumerate(zip(object_ids, object_centroids)):
min_dist = float("inf")
best_c = -1
for c, new_centroid in enumerate(input_centroids):
if c in used_cols:
continue
dist = math.hypot(old_centroid[0] - new_centroid[0], old_centroid[1] - new_centroid[1])
if dist < min_dist:
min_dist = dist
best_c = c
# Nếu khoảng cách nằm trong ngưỡng di chuyển hợp lý
if min_dist < self.max_distance and best_c != -1:
self.previous_positions[obj_id] = self.objects[obj_id]
self.objects[obj_id] = input_centroids[best_c]
self.disappeared[obj_id] = 0
used_rows.add(r)
used_cols.add(best_c)
# Các tâm mới chưa được gán -> vật thể mới xuất hiện
for c in range(len(input_centroids)):
if c not in used_cols:
self._register(input_centroids[c])
# Các vật thể cũ không tìm thấy tâm mới tương ứng
for r in range(len(object_ids)):
if r not in used_rows:
obj_id = object_ids[r]
self.disappeared[obj_id] += 1
if self.disappeared[obj_id] > self.max_disappeared:
self._deregister(obj_id)
return self.objects
def _register(self, centroid):
self.objects[self.next_object_id] = centroid
self.previous_positions[self.next_object_id] = centroid
self.disappeared[self.next_object_id] = 0
self.next_object_id += 1
def _deregister(self, object_id):
del self.objects[object_id]
del self.previous_positions[object_id]
del self.disappeared[object_id]db.py)Mỗi khi có đối tượng cắt qua vạch, thông tin phải được lưu lại tức thì để phục vụ báo cáo.
Tạo file db.py:
import sqlite3
from datetime import datetime
DB_NAME = "events.db"
def init_db():
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS counts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
object_id INTEGER NOT NULL,
class_name TEXT NOT NULL,
direction TEXT NOT NULL
)
""")
conn.commit()
conn.close()
def log_event(object_id, class_name, direction):
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
cursor.execute("""
INSERT INTO counts (timestamp, object_id, class_name, direction)
VALUES (?, ?, ?, ?)
""", (now_str, object_id, class_name, direction))
conn.commit()
conn.close()
def get_recent_events(limit=10):
conn = sqlite3.connect(DB_NAME)
cursor = conn.cursor()
cursor.execute("SELECT timestamp, object_id, class_name, direction FROM counts ORDER BY id DESC LIMIT ?", (limit,))
rows = cursor.fetchall()
conn.close()
return rowscore.py)Module này thực hiện luồng xử lý chính:
Đọc frame từ camera/video.
Dùng mô hình YOLOv8 Nano (yolov8n.pt) để bóc tách bounding box.
Cập nhật Tracker để giữ ID.
Kiểm tra sự kiện cắt vạch (Tripwire Intersection) và xác định hướng IN hoặc OUT.
Tạo file core.py:
import cv2
import numpy as np
from ultralytics import YOLO
from tracker import CentroidTracker
from db import log_event
class VideoProcessor:
def __init__(self, source=0, line_y=350, target_classes=[2, 7]):
"""
source: 0 (webcam), đường dẫn file MP4, hoặc link RTSP
line_y: Tọa độ trục Y của vạch ảo nằm ngang
target_classes: ID nhãn theo tập COCO (2: car, 7: truck, 0: person,...)
"""
self.cap = cv2.VideoCapture(source)
# Nạp YOLOv8 Nano pre-trained (tự động tải về máy trong lần chạy đầu)
self.model = YOLO("yolov8n.pt")
self.tracker = CentroidTracker(max_disappeared=10, max_distance=50)
self.line_y = line_y
self.target_classes = target_classes
self.count_in = 0
self.count_out = 0
# Tập hợp lưu các object_id đã được đếm để tránh đếm trùng lặp
self.counted_ids = set()
def process_frame(self):
ret, frame = self.cap.read()
if not ret:
return False, None
h, w = frame.shape[:2]
# 1. Suy luận YOLOv8 (Inference)
results = self.model.predict(frame, conf=0.4, verbose=False)[0]
detections = []
for box in results.boxes:
cls_id = int(box.cls[0])
if cls_id in self.target_classes:
x1, y1, x2, y2 = map(int, box.xyxy[0])
detections.append([x1, y1, x2, y2])
# 2. Cập nhật vị trí vật thể qua Tracker
tracked_objects = self.tracker.update(detections)
# 3. Vẽ vạch ảo Tripwire (Vạch ngang màu đỏ)
cv2.line(frame, (0, self.line_y), (w, self.line_y), (0, 0, 255), 2)
cv2.putText(frame, "TRIPWIRE LINE", (10, self.line_y - 10),
cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 0, 255), 2)
# 4. Kiểm tra điều kiện cắt vạch
for obj_id, (cx, cy) in tracked_objects.items():
prev_cx, prev_cy = self.tracker.previous_positions.get(obj_id, (cx, cy))
# Vẽ tâm vật thể và ID
cv2.circle(frame, (cx, cy), 5, (0, 255, 0), -1)
cv2.putText(frame, f"ID: {obj_id}", (cx - 10, cy - 10),
cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 255), 2)
if obj_id not in self.counted_ids:
# Điều kiện đi từ TRÊN xuống DƯỚI (HƯỚNG IN)
if prev_cy < self.line_y and cy >= self.line_y:
self.count_in += 1
self.counted_ids.add(obj_id)
log_event(obj_id, "vehicle", "IN")
# Hiệu ứng đổi màu vạch báo hiệu
cv2.line(frame, (0, self.line_y), (w, self.line_y), (0, 255, 0), 4)
# Điều kiện đi từ DƯỚI lên TRÊN (HƯỚNG OUT)
elif prev_cy > self.line_y and cy <= self.line_y:
self.count_out += 1
self.counted_ids.add(obj_id)
log_event(obj_id, "vehicle", "OUT")
cv2.line(frame, (0, self.line_y), (w, self.line_y), (255, 255, 0), 4)
# 5. Vẽ HUD thông tin lên góc trên màn hình
cv2.rectangle(frame, (10, 10), (220, 90), (0, 0, 0), -1)
cv2.putText(frame, f"IN : {self.count_in}", (20, 45),
cv2.FONT_HERSHEY_SIMPLEX, 0.9, (0, 255, 0), 2)
cv2.putText(frame, f"OUT: {self.count_out}", (20, 80),
cv2.FONT_HERSHEY_SIMPLEX, 0.9, (0, 255, 255), 2)
return True, frame
def release(self):
self.cap.release()app.py)Sử dụng Streamlit để tạo một Dashboard tương tác hiển thị thời gian thực gồm luồng video trực tiếp, thẻ KPI tổng số đếm, thanh điều chỉnh vị trí vạch kẻ ảo và bảng log sự kiện từ database.
Tạo file app.py:
import streamlit as st
import cv2
import pandas as pd
from core import VideoProcessor
from db import init_db, get_recent_events
# Cấu hình giao diện Streamlit
st.set_page_config(page_title="AI Tripwire Counter", layout="wide")
st.title("🚦 Hệ Thống Giám Sát & Đếm Đối Tượng Qua Vạch Ảo (MVP)")
# Khởi tạo cơ sở dữ liệu
init_db()
# Thanh Sidebar cấu hình
st.sidebar.header("Cấu Hình Hệ Thống")
source_option = st.sidebar.radio("Nguồn video:", ("Webcam", "File Video có sẵn"))
line_position = st.sidebar.slider("Vị trí vạch cắt (Trục Y):", min_value=100, max_value=600, value=300, step=10)
if source_option == "Webcam":
video_source = 0
else:
video_source = "test_traffic.mp4" # Đường dẫn tới video mẫu trên máy bạn
start_btn = st.sidebar.button("Bắt đầu giám sát")
stop_btn = st.sidebar.button("Dừng lại")
# Khu vực hiển thị kết quả chính
kpi1, kpi2, kpi3 = st.columns(3)
in_metric = kpi1.empty()
out_metric = kpi2.empty()
total_metric = kpi3.empty()
col_video, col_logs = st.columns([2, 1])
video_placeholder = col_video.empty()
log_placeholder = col_logs.empty()
if start_btn:
processor = VideoProcessor(source=video_source, line_y=line_position)
while processor.cap.isOpened():
ret, frame = processor.process_frame()
if not ret or stop_btn:
break
# Cập nhật số liệu KPI
in_metric.metric(label="Tổng Lượt Vào (IN)", value=processor.count_in)
out_metric.metric(label="Tổng Lượt Ra (OUT)", value=processor.count_out)
total_metric.metric(label="Tổng Lưu Lượng", value=processor.count_in + processor.count_out)
# Chuyển BGR sang RGB để Streamlit hiển thị đúng màu
frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
video_placeholder.image(frame_rgb, channels="RGB", use_container_width=True)
# Cập nhật bảng nhật ký sự kiện từ SQLite
events = get_recent_events(limit=8)
if events:
df = pd.DataFrame(events, columns=["Thời gian", "ID", "Loại", "Chiều"])
log_placeholder.dataframe(df, use_container_width=True)
processor.release()
st.info("Hệ thống đã dừng.")Khởi chạy ứng dụng: Mở terminal tại thư mục dự án và gõ lệnh:
streamlit run app.py
Tương tác trên giao diện:
Trình duyệt sẽ tự động mở địa chỉ: http://localhost:8501.
Nếu không có video mẫu, chọn chế độ Webcam (chọn target_classes=[0] trong core.py để đếm người đi qua vạch).
Dùng thanh trượt điều chỉnh vị trí đường thẳng vạch ảo phù hợp với góc quan sát.
Bấm Bắt đầu giám sát để theo dõi các đối tượng được đóng khung bounding box, gán ID liên tục và chứng kiến bảng số liệu cùng cơ sở dữ liệu SQLite tự động ghi nhận khi có lượt cắt vạch.
Bạn có muốn tối ưu hệ thống này bằng cách đưa luồng đọc camera vào một Thread riêng biệt (threading.Thread) để giao diện web không bị giảm tốc độ khung hình (drop FPS) khi mô hình AI đang suy luận không?