1. 项目概述：Ubuntu环境下的人流量分析方案

在零售、交通枢纽和公共场所管理中，人流量统计一直是核心需求。传统方案往往依赖专用硬件或商业软件，成本高昂且灵活性不足。最近我在一台老旧的ThinkPad T480上，基于Ubuntu 22.04 LTS搭建了一套基本可用的开源人流量分析系统，整套方案零成本且响应速度达到实时级别。

这个方案的核心价值在于：

• 完全基于开源工具链（OpenCV + TensorFlow Lite）
• 对硬件要求极低（实测4代i5+8G内存流畅运行）
• 支持主流监控摄像头RTSP流接入
• 提供基础计数和热力图生成功能

关键提示：选择Ubuntu而非Windows的主要考虑是其对Python生态的完整支持，以及长期运行时的系统稳定性。实测连续运行72小时内存占用稳定在2.3GB左右。

2. 环境搭建与依赖安装

2.1 基础系统准备

推荐使用Ubuntu 22.04 LTS版本，其长期支持特性和稳定的软件源是关键保障。安装时需注意：

# 安装后立即执行 sudo apt update && sudo apt upgrade -y sudo apt install -y build-essential cmake unzip pkg-config

显卡驱动选择：

• NVIDIA显卡：建议使用官方驱动（版本525以上）
• 核显/AMD显卡：默认开源驱动即可

# NVIDIA驱动安装示例 sudo ubuntu-drivers autoinstall sudo reboot

2.2 Python环境配置

使用Miniconda创建独立环境是避免依赖冲突的最佳实践：

wget https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh bash Miniconda3-latest-Linux-x86_64.sh conda create -n crowd python=3.9 conda activate crowd

核心依赖安装清单：

pip install opencv-python==4.5.5.64 pip install tensorflow-cpu==2.9.1 # 无GPU时使用 pip install imutils==0.5.4 pip install numpy==1.21.6

避坑指南：OpenCV版本过高可能导致与TensorFlow的兼容性问题，实测4.5.5版本最稳定。若需GPU加速，应安装tensorflow-gpu并配置CUDA环境。

3. 核心算法实现

3.1 基于背景减除的移动物体检测

使用OpenCV实现的背景减除算法是最轻量级的解决方案：

import cv2 backSub = cv2.createBackgroundSubtractorMOG2(history=500, varThreshold=16, detectShadows=True) def detect_movement(frame): fg_mask = backSub.apply(frame) _, thresh = cv2.threshold(fg_mask, 244, 255, cv2.THRESH_BINARY) contours, _ = cv2.findContours(thresh, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) return [c for c in contours if cv2.contourArea(c) > 500]

参数调优要点：

• history：值越大对光照变化越鲁棒，但内存占用越高
• varThreshold：值越小灵敏度越高，但噪声也越多
• 面积阈值500可过滤小动物等干扰物

3.2 基于YOLOv4-tiny的人体检测

当需要更精确的识别时，可采用轻量级目标检测模型：

net = cv2.dnn.readNet("yolov4-tiny.weights", "yolov4-tiny.cfg") layer_names = net.getLayerNames() output_layers = [layer_names[i[0] - 1] for i in net.getUnconnectedOutLayers()] def detect_people(frame): blob = cv2.dnn.blobFromImage(frame, 0.00392, (416, 416), (0, 0, 0), True, crop=False) net.setInput(blob) outs = net.forward(output_layers) # 后续处理获取人体检测框...

模型部署技巧：

1. 从Darknet官网下载预训练权重
2. 使用OpenCV的dnn模块加载可避免框架依赖
3. 输入尺寸416x416在精度和速度间取得平衡

4. 系统架构设计与实现

4.1 视频流处理流水线

完整的处理流程包含以下环节：

graph TD A[视频源] --> B[帧提取] B --> C{检测模式} C -->|简单场景| D[背景减除] C -->|复杂场景| E[YOLO检测] D/E --> F[计数逻辑] F --> G[数据存储] G --> H[可视化输出]

关键实现代码：

class PeopleCounter: def __init__(self, rtsp_url): self.cap = cv2.VideoCapture(rtsp_url) self.count = 0 self.line_y = 360 # 虚拟计数线位置 def run(self): while True: ret, frame = self.cap.read() if not ret: break # 根据场景复杂度选择检测方法 if self.simple_scene: contours = detect_movement(frame) self.count += len(contours) else: boxes = detect_people(frame) self.count += len(boxes) # 绘制计数线和结果 cv2.line(frame, (0, self.line_y), (frame.shape[1], self.line_y), (0,255,0), 2) cv2.putText(frame, f"Count: {self.count}", (10,30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0,0,255), 2) cv2.imshow("Monitoring", frame) if cv2.waitKey(1) == 27: break

4.2 数据持久化方案

使用SQLite进行本地数据存储是最轻量级的方案：

import sqlite3 from datetime import datetime def init_db(): conn = sqlite3.connect('crowd.db') c = conn.cursor() c.execute('''CREATE TABLE IF NOT EXISTS counts (timestamp TEXT, count INTEGER, scene TEXT)''') conn.commit() conn.close() def save_count(count, scene="default"): conn = sqlite3.connect('crowd.db') c = conn.cursor() c.execute("INSERT INTO counts VALUES (?,?,?)", (datetime.now().isoformat(), count, scene)) conn.commit() conn.close()

5. 性能优化技巧

5.1 多进程处理框架

使用Python的multiprocessing模块实现生产者-消费者模式：

from multiprocessing import Process, Queue def video_worker(input_q, output_q): while True: frame = input_q.get() # 处理帧... output_q.put(result) if __name__ == '__main__': input_queue = Queue(maxsize=32) output_queue = Queue() processes = [Process(target=video_worker, args=(input_queue, output_queue)) for _ in range(4)] for p in processes: p.daemon = True p.start() while True: ret, frame = cap.read() if input_queue.full(): continue input_queue.put(frame)

5.2 模型量化与加速

对TensorFlow模型进行动态范围量化可提升3倍推理速度：

converter = tf.lite.TFLiteConverter.from_saved_model(saved_model_dir) converter.optimizations = [tf.lite.Optimize.DEFAULT] tflite_quant_model = converter.convert() with open('model_quant.tflite', 'wb') as f: f.write(tflite_quant_model)

实测性能对比：

6. 典型问题排查指南

6.1 视频流断连问题

RTSP流不稳定的常见解决方案：

1. 添加重连机制

def get_video_stream(url, max_retries=5): for i in range(max_retries): cap = cv2.VideoCapture(url) if cap.isOpened(): return cap time.sleep(2**i) # 指数退避 raise ConnectionError(f"无法连接视频流: {url}")

2. 修改OpenCV缓冲区设置

cap.set(cv2.CAP_PROP_BUFFERSIZE, 1) # 减少缓冲帧数 cap.set(cv2.CAP_PROP_POS_FRAMES, 0) # 重置帧位置

6.2 误检过滤策略

通过轨迹分析过滤瞬态误检：

from collections import deque class TrackedObject: def __init__(self, maxlen=10): self.path = deque(maxlen=maxlen) def update(self, center): self.path.append(center) if len(self.path) > 5: # 持续5帧才计入统计 return True return False

7. 可视化与报表生成

7.1 实时热力图生成

使用OpenCV实现的热力图效果：

def generate_heatmap(frames, size=(800,600)): heatmap = np.zeros((*size, 3), dtype=np.float32) for frame in frames: resized = cv2.resize(frame, size) mask = detect_people_mask(resized) heatmap += mask * 0.1 heatmap = cv2.normalize(heatmap, None, 0, 255, cv2.NORM_MINMAX) heatmap = cv2.applyColorMap(heatmap.astype(np.uint8), cv2.COLORMAP_JET) return heatmap

7.2 基于Matplotlib的统计图表

生成日流量趋势图示例：

import matplotlib.pyplot as plt from matplotlib.dates import DateFormatter def plot_daily_trend(): conn = sqlite3.connect('crowd.db') df = pd.read_sql("SELECT * FROM counts", conn) conn.close() df['datetime'] = pd.to_datetime(df['timestamp']) hourly = df.groupby(df['datetime'].dt.hour)['count'].sum() fig, ax = plt.subplots(figsize=(10,6)) ax.bar(hourly.index, hourly.values) ax.xaxis.set_major_formatter(DateFormatter('%H:%M')) plt.savefig('daily_trend.png')

这套系统在商场入口的实测数据显示，在光照条件稳定的情况下，计数准确率可达85%-92%。虽然相比商业系统仍有差距，但零成本和高度可定制的特性使其非常适合预算有限的中小场景。我后续计划加入ReID技术来实现更精确的进出方向判断，当前代码已开源在个人GitHub仓库。