YOLOv8与质心追踪实现智能人流统计系统

1. 项目概述：智能人流统计系统的核心价值

在商业综合体、交通枢纽和景区等公共场所，准确统计人流量一直是运营管理的核心需求。传统人工计数方式效率低下且误差率高，而基于计算机视觉的智能人流统计系统正在成为行业标配。这个项目将使用YOLOv8这一当前最先进的目标检测框架，结合质心追踪算法，构建一个完整的端到端解决方案。

我选择YOLOv8作为基础框架，主要基于三个考量：首先，相比前代YOLOv5，v8版本在保持高速度的同时提升了小目标检测精度；其次，其Python接口友好，便于快速集成到业务系统中；最重要的是，官方提供的预训练模型在COCO数据集上mAP达到53.7%，为我们的迁移学习提供了良好起点。

2. 环境配置与依赖安装

2.1 基础环境搭建

推荐使用Python 3.8-3.10版本，过高版本可能导致部分依赖不兼容。创建隔离环境是避免依赖冲突的关键步骤：

bash复制conda create -n yolov8_counting python=3.9
conda activate yolov8_counting

核心依赖包括：

• torch>=1.8.0 （建议使用CUDA 11.3以上版本以获得GPU加速）
• ultralytics>=8.0.0 （YOLOv8官方库）
• opencv-python>=4.5.4 （视频流处理）
• numpy>=1.20.0 （数值计算）
• scipy>=1.7.0 （质心追踪算法依赖）

安装命令：

bash复制pip install ultralytics opencv-python scipy

注意：如果使用GPU加速，需先手动安装与CUDA版本匹配的torch，再安装其他依赖

2.2 验证环境配置

通过以下代码验证关键组件是否正常工作：

python复制import torch
from ultralytics import YOLO

print(torch.cuda.is_available())  # 应输出True
model = YOLO('yolov8n.pt')  # 测试加载纳米级预训练模型
print(model.info())  # 应显示模型结构信息

3. 数据准备与标注规范

3.1 数据集构建策略

人流统计场景需要特别关注三类数据：

1. 高密度人群（如地铁站早晚高峰）
2. 遮挡场景（如商场扶梯区域）
3. 多角度拍摄（顶视/斜视视角）

建议采集至少2000张包含不同光照条件、人群密度的现场图片。标注时需注意：

• 只标注"person"类别（COCO类别0）
• 对遮挡超过50%的人体仍应标注
• 包含至少20%的小目标（高度<50像素）

3.2 标注工具与格式转换

推荐使用CVAT或LabelImg进行标注，输出YOLO格式的txt文件。每个标注文件对应同名图片，格式示例：

code复制0 0.543 0.612 0.124 0.245  # 类别 x_center y_center width height

使用以下代码验证标注正确性：

python复制import cv2
import glob

for txt_file in glob.glob('labels/*.txt'):
    img = cv2.imread(txt_file.replace('labels','images').replace('.txt','.jpg'))
    h, w = img.shape[:2]
    with open(txt_file) as f:
        for line in f:
            cls, xc, yc, bw, bh = map(float, line.strip().split())
            # 转换为像素坐标
            x1 = int((xc - bw/2) * w)
            y1 = int((yc - bh/2) * h)
            x2 = int((xc + bw/2) * w)
            y2 = int((yc + bh/2) * h)
            cv2.rectangle(img, (x1,y1), (x2,y2), (0,255,0), 2)
    cv2.imshow('verify', img)
    if cv2.waitKey(0) == 27: break

4. 模型训练与调优实战

4.1 基础训练配置

创建dataset.yaml配置文件：

yaml复制path: /path/to/dataset
train: images/train
val: images/val
test: images/test

names:
  0: person

启动训练命令：

bash复制yolo task=detect mode=train model=yolov8n.pt data=dataset.yaml epochs=100 imgsz=640 batch=16

关键参数解析：

• imgsz：根据GPU显存调整（12G显存建议640）
• batch：在OOM错误时降低此值
• patience：早停机制参数，建议设为20

4.2 高级训练技巧

提升小目标检测效果的三种方法：

1. 修改anchors：在data.yaml中添加适配人体比例的anchors

   yaml复制anchors:
     - [5,8, 8,16, 12,32]  # 更适合直立人体的宽高比
   

2. 使用--multi-scale训练：随机缩放图像增强小目标识别

   bash复制yolo train ... --multi-scale
   

3. 添加SPP模块：在models/yolov8.yaml中增加SPPF层

4.3 模型评估与优化

验证集评估命令：

bash复制yolo val model=runs/detect/train/weights/best.pt data=dataset.yaml

重点关注两个指标：

• mAP@0.5:0.95（综合精度）
• mAP@0.5（IOU=0.5时的精度）

如果发现漏检严重，可以：

1. 降低conf阈值（--conf 0.3）
2. 增加--augment参数增强推理时的数据增强
3. 使用TTA（测试时增强）

   bash复制yolo predict ... --augment
   

5. 人流追踪算法实现

5.1 质心追踪器原理

质心追踪算法通过计算连续帧中检测框中心的欧氏距离来关联目标。实现步骤：

1. 为每个新检测分配新ID
2. 计算当前帧与上一帧质心的距离矩阵
3. 使用匈牙利算法进行最优匹配
4. 更新已存在目标的坐标
5. 移除丢失超过N帧的目标

核心代码实现：

python复制from scipy.spatial import distance_matrix
from scipy.optimize import linear_sum_assignment

class CentroidTracker:
    def __init__(self, max_disappeared=30):
        self.next_id = 0
        self.objects = {}
        self.disappeared = {}
        self.max_disappeared = max_disappeared

    def update(self, rects):
        if len(rects) == 0:
            self._update_disappeared()
            return self.objects
        
        # 计算质心
        centroids = np.array([[(x1+x2)/2, (y1+y2)/2] for (x1,y1,x2,y2) in rects])
        
        # 初始化新注册目标
        if len(self.objects) == 0:
            for i in range(len(centroids)):
                self.objects[self.next_id] = centroids[i]
                self.disappeared[self.next_id] = 0
                self.next_id += 1
        else:
            # 计算距离矩阵
            object_ids = list(self.objects.keys())
            object_centroids = list(self.objects.values())
            D = distance_matrix(object_centroids, centroids)
            
            # 匈牙利算法匹配
            row_ind, col_ind = linear_sum_assignment(D)
            
            used_rows = set()
            used_cols = set()
            
            # 更新已有目标
            for (row, col) in zip(row_ind, col_ind):
                object_id = object_ids[row]
                self.objects[object_id] = centroids[col]
                self.disappeared[object_id] = 0
                used_rows.add(row)
                used_cols.add(col)
            
            # 处理未匹配目标
            unused_rows = set(range(len(object_centroids))) - used_rows
            unused_cols = set(range(len(centroids))) - used_cols
            
            # 丢失目标处理
            for row in unused_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 unused_cols:
                self._register(centroids[col])
        
        return self.objects

5.2 跨线计数实现

在入口处设置虚拟计数线，当质心跨越时触发计数：

python复制def is_crossing_line(prev_pos, curr_pos, line):
    """判断是否跨越虚拟线
    line: ((x1,y1), (x2,y2)) 表示计数线
    """
    prev_side = (line[1][0]-line[0][0])*(prev_pos[1]-line[0][1]) - \
                (line[1][1]-line[0][1])*(prev_pos[0]-line[0][0])
    curr_side = (line[1][0]-line[0][0])*(curr_pos[1]-line[0][1]) - \
                (line[1][1]-line[0][1])*(curr_pos[0]-line[0][0])
    
    return prev_side * curr_side < 0  # 两侧异号表示跨越

6. 系统集成与性能优化

6.1 视频流处理架构

采用生产者-消费者模式提高处理效率：

python复制from queue import Queue
from threading import Thread

class VideoStream:
    def __init__(self, src=0):
        self.stream = cv2.VideoCapture(src)
        self.stopped = False
        self.frames = Queue(maxsize=32)  # 缓冲队列
    
    def start(self):
        Thread(target=self.update, args=()).start()
        return self
    
    def update(self):
        while not self.stopped:
            if not self.frames.full():
                ret, frame = self.stream.read()
                if not ret:
                    self.stop()
                    return
                self.frames.put(frame)
    
    def read(self):
        return self.frames.get()
    
    def stop(self):
        self.stopped = True

# 使用示例
vs = VideoStream(src="rtsp://192.168.1.64/stream").start()
while True:
    frame = vs.read()
    # 处理帧...

6.2 多线程推理加速

利用Python的concurrent.futures实现并行处理：

python复制from concurrent.futures import ThreadPoolExecutor

def process_frame(frame, model):
    results = model(frame)
    return results

with ThreadPoolExecutor(max_workers=4) as executor:
    futures = []
    while True:
        frame = vs.read()
        future = executor.submit(process_frame, frame, model)
        futures.append(future)
        
        # 获取最早完成的结果
        if len(futures) >= 4:
            done, _ = concurrent.futures.wait(
                futures, return_when=concurrent.futures.FIRST_COMPLETED)
            for future in done:
                results = future.result()
                # 处理结果...
                futures.remove(future)

7. 部署方案与性能对比

7.1 不同硬件平台性能测试

7.2 ONNX格式导出与优化

导出为ONNX格式实现跨平台部署：

bash复制yolo export model=best.pt format=onnx opset=12 simplify=True

优化技巧：

1. 添加--dynamic参数支持动态输入尺寸
2. 使用onnxruntime进行图优化

   python复制import onnxruntime as ort
   
   sess_options = ort.SessionOptions()
   sess_options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
   sess = ort.InferenceSession("model.onnx", sess_options)
   

8. 常见问题与解决方案

8.1 检测抖动问题

现象：同一目标ID频繁切换
解决方法：

1. 增加追踪器的max_disappeared参数（建议30-50）
2. 添加卡尔曼滤波预测

   python复制class KalmanFilter:
       def __init__(self):
           self.kf = cv2.KalmanFilter(4,2)
           self.kf.measurementMatrix = np.array([[1,0,0,0],[0,1,0,0]], np.float32)
           self.kf.transitionMatrix = np.array([[1,0,1,0],[0,1,0,1],[0,0,1,0],[0,0,0,1]], np.float32)
       
       def predict(self, coord):
           self.kf.predict()
           measurement = np.array([[np.float32(coord[0])], [np.float32(coord[1])]])
           self.kf.correct(measurement)
           return self.kf.statePrediction[:2]
   

8.2 漏检问题排查流程

1. 检查标注质量（特别是小目标和遮挡目标）
2. 调整conf阈值（训练时建议0.01，推理时可提高到0.3）
3. 增加训练时的mosaic数据增强

   bash复制yolo train ... --mosaic 0.5
   

4. 使用更大的模型（如yolov8m或yolov8l）

8.3 计数误差分析

主要误差来源：

1. 目标遮挡导致的ID切换
2. 虚拟线位置设置不合理
3. 目标移动速度过快

优化方案：

1. 在关键区域设置ROI限制检测范围
2. 使用多条计数线投票机制
3. 添加移动方向过滤（只统计进入方向的目标）

9. 完整系统代码结构

code复制yolov8_people_counter/
├── configs/
│   ├── dataset.yaml          # 数据集配置
│   └── tracking.yaml         # 追踪参数配置
├── models/
│   ├── yolov8n_custom.pt     # 自定义训练权重
│   └── utils.py              # 工具函数
├── tracking/
│   ├── centroid_tracker.py   # 质心追踪实现
│   └── counter.py           # 计数逻辑
├── utils/
│   ├── visualization.py      # 可视化工具
│   └── video_utils.py        # 视频处理工具
└── main.py                   # 主入口文件

主程序框架示例：

python复制import cv2
from models import load_model
from tracking import CentroidTracker, LineCounter

def main():
    # 初始化组件
    model = load_model("models/yolov8n_custom.pt")
    tracker = CentroidTracker(max_disappeared=30)
    counter = LineCounter(line=((100,200),(500,200)))
    
    # 视频流处理
    cap = cv2.VideoCapture("input.mp4")
    while cap.isOpened():
        ret, frame = cap.read()
        if not ret: break
        
        # 推理
        results = model(frame)[0]
        boxes = results.boxes.xyxy.cpu().numpy()
        
        # 追踪
        objects = tracker.update(boxes)
        
        # 计数
        counter.update(objects)
        
        # 可视化
        visualize(frame, objects, counter.count)
        
    cap.release()

if __name__ == "__main__":
    main()

10. 项目扩展方向

1. 多摄像头协同计数：通过RTSP协议接入多个摄像头，使用Redis共享计数状态
2. 人群密度分析：基于检测框覆盖面积计算区域密度热力图
3. 异常行为检测：结合姿态估计识别跌倒、聚集等异常行为
4. 年龄性别识别：在YOLOv8基础上添加多任务头
5. 边缘设备部署：使用TensorRT加速在Jetson系列设备上的推理

在实际部署中发现，将conf阈值设置为0.35-0.4能取得最佳平衡。对于1080P视频流，建议将推理分辨率设置为1280x1280，这样既能保证小目标检测效果，又能维持实时性能。