Python+OpenCV实战:用质心追踪算法打造多目标行人计数器(附完整代码)
Python+OpenCV实战:用质心追踪算法打造多目标行人计数器(附完整代码)
最近在做一个智慧零售门店的客流分析项目,客户需要在入口处部署一个摄像头,实时统计进出人数。最初尝试用YOLO等重型检测模型在每一帧都做识别,结果在边缘计算设备上FPS直接掉到个位数,画面卡顿严重,用户体验极差。后来转向了质心追踪算法,核心思路是只在目标首次出现时做一次“昂贵”的识别,后续帧只做快速的检测和关联,FPS瞬间提升到25以上,完美满足了实时性要求。今天,我就把这个项目中打磨成熟的多目标行人计数方案,从原理到代码,毫无保留地分享给你。
这套方案特别适合安防监控、门店客流统计、公共场所人流量监测等场景。如果你正被实时视频分析的计算资源问题困扰,或者想为你的应用增加一个轻量级、高可用的目标跟踪与计数功能,那么这篇文章正是为你准备的。我们将绕过繁琐的数学公式,直接从工程落地角度,手把手教你如何用Python和OpenCV,构建一个鲁棒的行人计数器。
1. 质心追踪:为何它是轻量级多目标跟踪的利器
在计算机视觉领域,目标跟踪算法五花八门,从简单的光流法到复杂的深度学习跟踪器(如SORT、DeepSORT)。但对于很多需要实时处理、部署在资源受限设备上的应用来说,算法的复杂度和计算开销必须严格控制。质心追踪算法正是在这种需求下脱颖而出的一种经典方法。
它的核心思想异常直观:为每个检测到的目标计算一个“质心”(通常是边界框的中心点),然后在连续的帧之间,通过比较这些质心点的欧氏距离,来关联同一个目标,并为其维持一个唯一的ID。
听起来很简单,对吧?但它的巧妙之处在于,将复杂的“目标再识别”问题,转化为了一个相对简单的“点对点距离匹配”问题。这意味着,只要你的目标检测器能稳定地输出边界框,质心追踪算法就能以极低的计算成本,完成ID关联的任务。
为了让你更清晰地理解质心追踪与其他常见跟踪算法的定位,我整理了下面这个对比表格:
| 特性维度 | 质心追踪 (Centroid Tracking) | 相关滤波跟踪器 (如KCF) | 深度学习跟踪器 (如DeepSORT) |
|---|---|---|---|
| 核心原理 | 基于边界框质心的欧氏距离匹配 | 在频域学习目标的外观模板并进行相关滤波 | 结合检测、运动预测(卡尔曼滤波)和外观特征(ReID网络) |
| 计算开销 | 极低,仅需距离计算 | 中等,需进行傅里叶变换与滤波 | 很高,需要运行神经网络进行特征提取 |
| 跟踪精度 | 中等,依赖检测质量,易受遮挡影响 | 较高,对形变和旋转有一定鲁棒性 | 很高,能较好处理遮挡和ID切换 |
| 是否需要每帧检测 | 是 | 否(可基于第一帧初始化) | 是(通常与检测器耦合) |
| 多目标支持 | 原生支持 | 通常为单目标,多目标需并行实例 | 原生支持 |
| 适用场景 | 实时监控、客流统计、对硬件要求低的场景 | 单目标精准跟踪,如无人机跟踪特定物体 | 对跟踪精度要求极高的场景,如自动驾驶、体育分析 |
提示:选择跟踪算法没有绝对的“最好”,只有“最合适”。质心追踪的优势不在于它有多强大,而在于它在简单性、效率和实用性之间取得了极佳的平衡,特别适合作为入门和轻量级应用的首选。
那么,质心追踪具体是如何一步步工作的呢?它主要包含五个关键步骤,这构成了我们后续实现算法的骨架:
- 接受检测结果并计算质心:从目标检测器获取边界框,计算每个框的中心点作为质心。
- 计算新旧质心距离:对于新一帧的每个质心,计算它与上一帧所有已知目标质心之间的欧氏距离。
- 关联与更新:基于“最小距离”原则,将新质心与最可能的旧目标ID关联,并更新该ID的质心位置。
- 注册新目标:如果出现无法与任何旧目标匹配的新质心,则为它分配一个新的唯一ID。
- 注销消失目标:如果一个目标在连续若干帧中都没有再出现,则移除它的ID,避免内存泄漏和错误计数。
接下来,我们就将围绕这五个步骤,用代码将其具象化。
2. 构建核心:可复用的质心追踪器类
一个良好的工程实践是将核心算法逻辑封装成独立的、可复用的类。这样,我们在不同的项目(比如跟踪行人、车辆、甚至动物)中,只需要更换前端检测器,而追踪器本身可以保持不变。下面,我们就来实现这个 CentroidTracker 类。
首先,创建文件 centroid_tracker.py,并导入必要的库:
# centroid_tracker.py
from scipy.spatial import distance as dist
from collections import OrderedDict
import numpy as np
class CentroidTracker:
def __init__(self, max_disappeared=50, max_distance=50):
"""
初始化质心追踪器。
参数:
max_disappeared (int): 一个目标被移除前,允许连续丢失的最大帧数。
max_distance (int): 关联新旧目标时,可接受的最大欧氏距离阈值。
"""
self.next_object_id = 0
# 使用OrderedDict保持ID插入顺序,便于调试
self.objects = OrderedDict() # key: objectID, value: (cX, cY) centroid
self.disappeared = OrderedDict() # key: objectID, value: 连续丢失的帧数
self.max_disappeared = max_disappeared
self.max_distance = max_distance # 新增:距离阈值
在构造函数中,我们引入了两个重要参数:
max_disappeared:这是算法的“记忆”时长。一个人走出摄像头视野后,不会立刻被遗忘,而是会保留一段时间(比如50帧)。如果在这段时间内他重新出现,ID可以保持不变;如果超过这个时间,则注销该ID。这有效避免了因短暂遮挡或出画导致的ID频繁切换。max_distance:这是一个我在实战中加入的关键改进。原始的质心追踪只认“最小距离”,但如果两个目标交叉走过,他们的质心在某个瞬间可能非常接近,导致ID错误交换。通过设置一个最大距离阈值,只有当最小距离同时小于这个阈值时,我们才进行关联。否则,宁愿将其视为新目标或丢失目标,这大大提升了在拥挤场景下的鲁棒性。
接下来,我们实现注册和注销的辅助方法:
def register(self, centroid):
"""为新检测到的目标注册一个唯一ID。"""
self.objects[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.disappeared[object_id]
现在,来到最核心的 update 方法。它将接收当前帧的所有检测框(矩形坐标),并返回更新后的目标字典(ID到质心的映射)。
def update(self, rects):
"""
更新追踪器状态。
参数:
rects: 一个列表,每个元素为 (startX, startY, endX, endY) 格式的边界框。
返回:
self.objects: 更新后的目标字典 {objectID: (cX, cY)}。
"""
# 情况1: 当前帧没有检测到任何目标
if len(rects) == 0:
# 所有现有目标的“丢失”计数器加1
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
# 情况2: 当前帧有检测目标
# 步骤1: 计算输入边界框的质心
input_centroids = np.zeros((len(rects), 2), dtype="int")
for i, (start_x, start_y, end_x, end_y) in enumerate(rects):
cX = int((start_x + end_x) / 2.0)
cY = int((start_y + end_y) / 2.0)
input_centroids[i] = (cX, cY)
# 步骤2: 如果当前没有追踪任何目标,直接注册所有新检测到的目标
if len(self.objects) == 0:
for i in range(0, len(input_centroids)):
self.register(input_centroids[i])
else:
# 步骤3: 获取现有目标的ID和质心
object_ids = list(self.objects.keys())
object_centroids = list(self.objects.values())
# 步骤4: 计算所有现有质心与输入质心之间的成对欧氏距离
# D 的形状为 (现有目标数, 输入目标数)
D = dist.cdist(np.array(object_centroids), input_centroids)
# 步骤5: 找到最小距离的匹配
# 首先,对每一行(每个现有目标),找到与其距离最小的输入目标列索引
rows = D.min(axis=1).argsort() # 按最小距离排序后的行索引
cols = D.argmin(axis=1)[rows] # 对应行的最小距离列索引
# 用于记录已经匹配过的行和列,避免重复匹配
used_rows = set()
used_cols = set()
# 步骤6: 遍历排序后的(行,列)对,进行匹配
for (row, col) in zip(rows, cols):
# 如果该行或该列已经被使用过,跳过
if row in used_rows or col in used_cols:
continue
# 关键改进:检查最小距离是否在可接受的阈值内
if D[row, col] > self.max_distance:
# 距离太远,不进行关联,留给后续步骤处理(注册新目标或标记丢失)
continue
# 找到匹配!更新该现有目标的质心位置,并重置其“丢失”计数器
object_id = object_ids[row]
self.objects[object_id] = input_centroids[col]
self.disappeared[object_id] = 0
# 标记该行和该列为已使用
used_rows.add(row)
used_cols.add(col)
# 步骤7: 处理未匹配的行和列
unused_rows = set(range(0, D.shape[0])).difference(used_rows)
unused_cols = set(range(0, D.shape[1])).difference(used_cols)
# 情况A: 现有目标数 >= 输入目标数,部分现有目标可能丢失了
if D.shape[0] >= D.shape[1]:
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)
# 情况B: 输入目标数 > 现有目标数,有新的目标出现
else:
for col in unused_cols:
self.register(input_centroids[col])
return self.objects
这个 update 方法虽然代码量稍大,但逻辑完全遵循了我们之前阐述的五个步骤,并加入了 max_distance 阈值判断。通过这样的封装,我们得到了一个健壮、可配置的质心追踪器核心。接下来,我们需要一个“眼睛”来为它提供边界框。
3. 选择与集成:为追踪器配上高效的目标检测器
质心追踪器本身不负责“看”,它只负责“关联”。因此,我们需要一个前端的目标检测器来提供每一帧中行人的边界框。在实时应用中,检测器的速度至关重要。这里我提供两种经过验证的方案供你选择。
方案一:轻量级深度学习模型(MobileNet SSD) 这是平衡精度和速度的优选。我们可以使用OpenCV的DNN模块加载一个预训练的Caffe模型。
首先,确保你下载了模型文件(deploy.prototxt 和 MobileNetSSD_deploy.caffemodel)。将它们放在项目目录下。
方案二:OpenCV内置的HOG行人检测器 这是最轻量的方案,无需下载额外模型,但检测精度和鲁棒性相对较低,更适合背景简单的场景。
为了演示的完整性,我将以更强大的MobileNet SSD为例,展示如何将其与我们的追踪器集成。在 people_counter.py 主脚本中,我们将进行如下操作:
# people_counter.py
import argparse
import time
import cv2
import numpy as np
from centroid_tracker import CentroidTracker
# 定义MobileNet SSD能检测的20个类别(+背景)
CLASSES = ["background", "aeroplane", "bicycle", "bird", "boat",
"bottle", "bus", "car", "cat", "chair", "cow", "diningtable",
"dog", "horse", "motorbike", "person", "pottedplant", "sheep",
"sofa", "train", "tvmonitor"]
# 置信度阈值,过滤掉不可靠的检测
CONFIDENCE_THRESHOLD = 0.5
def main():
# 解析命令行参数
ap = argparse.ArgumentParser()
ap.add_argument("-p", "--prototxt", required=True, help="Caffe模型结构文件路径")
ap.add_argument("-m", "--model", required=True, help="Caffe模型权重文件路径")
ap.add_argument("-v", "--video", help="输入视频文件路径(缺省则使用摄像头)")
ap.add_argument("-o", "--output", help="输出视频文件路径(可选)")
args = vars(ap.parse_args())
# 初始化质心追踪器
# 根据你的视频帧率调整max_disappeared,例如30FPS视频,2秒约为60帧
ct = CentroidTracker(max_disappeared=60, max_distance=80)
# 加载预训练的MobileNet SSD模型
print("[INFO] 加载检测模型...")
net = cv2.dnn.readNetFromCaffe(args["prototxt"], args["model"])
# 初始化视频流
if not args.get("video", False):
print("[INFO] 启动摄像头视频流...")
vs = cv2.VideoCapture(0) # 使用默认摄像头
time.sleep(2.0) # 让摄像头预热
else:
print("[INFO] 打开视频文件...")
vs = cv2.VideoCapture(args["video"])
# 初始化视频写入器(如果需要保存结果)
writer = None
(H, W) = (None, None)
# 用于计数的变量(示例:统计总出现过的不同ID数)
total_count = 0
counted_ids = set()
print("[INFO] 开始处理视频流...")
while True:
(grabbed, frame) = vs.read()
if not grabbed:
break
# 调整帧大小以加快处理速度(保持宽高比)
frame = cv2.resize(frame, (800, 600))
if W is None or H is None:
(H, W) = frame.shape[:2]
# 为DNN检测器准备输入图像(Blob)
blob = cv2.dnn.blobFromImage(frame, 0.007843, (W, H), 127.5)
net.setInput(blob)
detections = net.forward()
rects = [] # 存储当前帧检测到的行人边界框
# 遍历所有检测结果
for i in range(0, detections.shape[2]):
confidence = detections[0, 0, i, 2]
# 过滤掉低置信度的检测,并且只保留“人”这个类别
if confidence > CONFIDENCE_THRESHOLD:
idx = int(detections[0, 0, i, 1])
if CLASSES[idx] != "person":
continue
# 计算边界框坐标(相对于帧尺寸)
box = detections[0, 0, i, 3:7] * np.array([W, H, W, H])
(startX, startY, endX, endY) = box.astype("int")
# 确保边界框在图像范围内
startX = max(0, startX)
startY = max(0, startY)
endX = min(W, endX)
endY = min(H, endY)
rects.append((startX, startY, endX, endY))
# 可选:在帧上绘制检测框(绿色)
cv2.rectangle(frame, (startX, startY), (endX, endY), (0, 255, 0), 2)
# 将当前帧的检测框传递给追踪器进行更新
objects = ct.update(rects)
# 在帧上绘制追踪结果(ID和质心)
for (object_id, centroid) in objects.items():
# 绘制质心(红色圆点)
cv2.circle(frame, (centroid[0], centroid[1]), 4, (0, 0, 255), -1)
# 绘制对象ID文本
text = f"ID {object_id}"
cv2.putText(frame, text, (centroid[0] - 10, centroid[1] - 10),
cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 255), 2)
# 简单的计数逻辑:如果该ID是第一次出现,则计数+1
if object_id not in counted_ids:
counted_ids.add(object_id)
total_count += 1
# 在画面上显示总计数
count_text = f"Total Count: {total_count}"
cv2.putText(frame, count_text, (10, H - 20),
cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2)
# 显示结果帧
cv2.imshow("People Counter", frame)
key = cv2.waitKey(1) & 0xFF
if key == ord("q"):
break
# 清理资源
vs.release()
cv2.destroyAllWindows()
if __name__ == "__main__":
main()
这段主程序完成了从视频流读取、目标检测、追踪更新到可视化显示的完整流水线。你可以通过命令行运行它,例如:
python people_counter.py --prototxt MobileNetSSD_deploy.prototxt --model MobileNetSSD_deploy.caffemodel --video entrance.mp4
4. 从追踪到计数:实现精准的进出方向判断
基本的追踪和ID计数已经实现,但在客流统计等场景中,我们往往更关心进和出的数量。这就需要引入方向判断逻辑。一个常见且有效的方法是设置虚拟计数线。
假设我们的摄像头俯视一个入口,画面底部是门外,顶部是门内。我们可以定义一条水平线(y = line_pos)作为计数线。当一个人的质心从上往下穿过这条线时(direction = "down"),我们认为是“进入”;从下往上穿过(direction = "up")时,则认为是“离开”。
我们需要扩展追踪器,使其能记录每个目标的历史位置,并判断其穿越行为。下面是对 CentroidTracker 类的增强,以及主程序中相应的计数逻辑修改。
首先,修改 CentroidTracker 的 __init__ 和 register 方法,增加历史位置存储:
# 在 __init__ 中增加
self.track_history = OrderedDict() # key: objectID, value: list of recent centroids
# 修改 register 方法
def register(self, centroid):
self.objects[self.next_object_id] = centroid
self.disappeared[self.next_object_id] = 0
self.track_history[self.next_object_id] = [centroid] # 初始化历史记录
self.next_object_id += 1
然后,在 update 方法中成功关联并更新一个目标质心后,将其新质心追加到历史记录中,并保持固定长度(例如最近10个点):
# 在 update 方法的匹配成功部分(更新self.objects[object_id]之后)增加:
if object_id in self.track_history:
self.track_history[object_id].append(input_centroids[col])
# 只保留最近N个点,避免列表无限增长
if len(self.track_history[object_id]) > 10:
self.track_history[object_id].pop(0)
else:
self.track_history[object_id] = [input_centroids[col]]
同时,在 deregister 方法中,记得清理历史记录:
def deregister(self, object_id):
del self.objects[object_id]
del self.disappeared[object_id]
if object_id in self.track_history:
del self.track_history[object_id]
现在,追踪器具备了记忆每个目标短暂运动轨迹的能力。接下来,在主程序 people_counter.py 的循环中,在绘制完追踪结果后,添加方向判断和计数逻辑:
# 定义计数线的y坐标(例如,在画面高度60%的位置画一条线)
line_pos = int(H * 0.6)
cv2.line(frame, (0, line_pos), (W, line_pos), (255, 0, 0), 2) # 蓝色计数线
cv2.putText(frame, "Counting Line", (10, line_pos - 10),
cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 0, 0), 2)
# 初始化进出计数
in_count = 0
out_count = 0
# 用于记录已经完成计数的ID,避免重复计数
counted_ids = set()
for (object_id, centroid) in objects.items():
# ... 原有的绘制质心和ID的代码 ...
# 方向判断与计数
if object_id in ct.track_history:
history = ct.track_history[object_id]
if len(history) >= 2: # 至少有前后两个点才能判断方向
prev_y = history[-2][1] # 上一个质心的y坐标
curr_y = history[-1][1] # 当前质心的y坐标
# 判断是否穿越计数线
# 情况1: 从上往下穿越 (prev_y < line_pos and curr_y >= line_pos) -> 进入
# 情况2: 从下往上穿越 (prev_y > line_pos and curr_y <= line_pos) -> 离开
if prev_y < line_pos <= curr_y and object_id not in counted_ids:
direction = "down"
in_count += 1
counted_ids.add(object_id)
print(f"[INFO] 对象 ID {object_id} 进入区域。")
elif prev_y > line_pos >= curr_y and object_id not in counted_ids:
direction = "up"
out_count += 1
counted_ids.add(object_id)
print(f"[INFO] 对象 ID {object_id} 离开区域。")
else:
direction = None
# 可选:在画面上显示方向箭头
if direction:
arrow_start = (centroid[0], centroid[1] - 20)
arrow_end = (centroid[0], centroid[1] - 40) if direction == "up" else (centroid[0], centroid[1] + 40)
cv2.arrowedLine(frame, arrow_start, arrow_end, (255, 255, 0), 2, tipLength=0.3)
# 在画面上更新进出计数
cv2.putText(frame, f"In: {in_count}", (10, 30),
cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2)
cv2.putText(frame, f"Out: {out_count}", (10, 60),
cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 0, 255), 2)
cv2.putText(frame, f"Total In Frame: {len(objects)}", (10, 90),
cv2.FONT_HERSHEY_SIMPLEX, 0.6, (255, 255, 255), 2)
通过引入虚拟计数线和简单的运动方向分析,我们的系统就从“追踪器”升级为了一个功能完整的“双向客流计数器”。你可以根据实际场景调整计数线的位置和方向判断逻辑(例如使用斜线或区域判断)。
5. 性能优化与实战调试技巧
将代码跑起来只是第一步,要让它在真实环境中稳定可靠地工作,还需要一系列的优化和调试。以下是我在多个项目中总结出的关键点:
1. 调整检测器参数
- 置信度阈值 (
CONFIDENCE_THRESHOLD):太高会漏检,太低会引入大量噪声。通常从0.4到0.6之间调整,并在你的实际视频上观察效果。 - 非极大值抑制 (NMS):MobileNet SSD的输出可能包含大量重叠框。虽然OpenCV的DNN前向传播不直接输出NMS,但你可以手动实现或使用
cv2.dnn.NMSBoxes。这对于减少重复检测、提升追踪稳定性至关重要。# 在主循环的detections处理部分,收集boxes和confidences后 boxes = [] confidences = [] for i in range(0, detections.shape[2]): confidence = detections[0, 0, i, 2] if confidence > CONFIDENCE_THRESHOLD: idx = int(detections[0, 0, i, 1]) if CLASSES[idx] == "person": box = detections[0, 0, i, 3:7] * np.array([W, H, W, H]) (startX, startY, endX, endY) = box.astype("int") boxes.append([startX, startY, endX - startX, endY - startY]) # 需要(x, y, w, h)格式 confidences.append(float(confidence)) # 应用NMS indices = cv2.dnn.NMSBoxes(boxes, confidences, CONFIDENCE_THRESHOLD, 0.4) # 0.4是NMS阈值 rects = [] if len(indices) > 0: for i in indices.flatten(): (x, y, w, h) = boxes[i] rects.append((x, y, x + w, y + h)) # 绘制框...
2. 微调追踪器参数
max_disappeared:这个值需要根据视频帧率(FPS)和目标可能被遮挡的时长来设定。例如,30FPS的视频,如果一个人被柱子遮挡大约1秒,那么设置成30是合理的。你可以通过打印日志观察ID切换情况来调整。max_distance:这是防止ID错误交换的防火墙。它的值应该与目标在帧间移动的最大像素距离相关。你可以通过计算目标在高速行走时,连续两帧间质心的最大像素差来估算。通常设置为50-150之间。设置太小会导致频繁创建新ID,太大则失去防交换作用。
3. 处理边界情况与提升鲁棒性
- 闪烁的检测框:检测器可能在某些帧漏检目标,导致追踪器认为目标消失。适当增加
max_disappeared可以缓解这个问题。 - 拥挤与遮挡:这是质心追踪的天然弱点。除了设置
max_distance,还可以考虑:- 使用更稳定的检测器(如YOLOv5s,虽然稍慢但更准)。
- 引入简单的运动预测,如基于速度的质心位置预测,在匹配时使用预测位置而非上一帧位置。
- 如果场景固定,可以定义“禁止匹配区域”,当两个目标质心都进入该区域时,暂停ID更新,直到它们分开。
- 光照变化:确保检测模型是在与你的部署环境光照条件相似的数据上训练的,或者对输入帧进行直方图均衡化等预处理。
4. 部署与性能监控
- 多线程处理:对于高分辨率视频,可以将视频帧读取、目标检测、追踪与计数、结果显示/保存放在不同的线程中,用队列连接,充分利用多核CPU,显著提升整体吞吐量。
- 资源监控:在循环中定期打印或记录FPS、内存使用情况。这有助于你发现性能瓶颈(通常是检测步骤)。
import time # 在循环开始前 fps_start_time = time.time() fps_frame_count = 0 # 在循环内部,每处理N帧后 fps_frame_count += 1 if fps_frame_count >= 30: elapsed = time.time() - fps_start_time fps = fps_frame_count / elapsed print(f"[INFO] 当前FPS: {fps:.2f}") fps_start_time = time.time() fps_frame_count = 0
将上述优化点应用到你的项目中,反复在真实或模拟数据上进行测试和调整,你就能得到一个在特定场景下表现相当出色的行人计数系统。记住,没有一劳永逸的参数,最好的参数来自于你对具体场景的深入理解和不断调试。
更多推荐


所有评论(0)