视频流处理指对连续的视频帧进行实时或近实时的处理，OpenCV作为Python生态中常用的计算机视觉库，提供了完善的视频读写和帧处理接口，能够高效完成各类视频流处理任务。
环境准备
首先需要安装OpenCV的Python库，使用pip命令即可完成安装：
pip install opencv-python
安装完成后可以通过导入库验证是否安装成功：
import cv2 print(cv2.__version__)
视频流读取基础
OpenCV通过VideoCapture类来读取视频流，支持本地视频文件、摄像头采集流、网络视频流三种常见来源。
不同来源的视频流初始化
- 读取本地视频文件：传入视频文件的路径即可
- 读取摄像头流：传入摄像头设备索引，笔记本自带摄像头一般为0
- 读取网络视频流：传入网络视频流的地址，比如RTSP地址
基础的视频流读取并播放的代码如下：
import cv2
# 初始化视频流，这里以摄像头为例，参数0表示默认摄像头
cap = cv2.VideoCapture(0)
# 检查视频流是否成功打开
if not cap.isOpened():
print("无法打开视频流")
exit()
# 循环读取每一帧
while True:
# 读取一帧，ret是读取是否成功的标志，frame是帧数据
ret, frame = cap.read()
if not ret:
print("无法读取帧，视频流可能已结束")
break
# 显示当前帧
cv2.imshow("Video Stream", frame)
# 按q键退出循环
if cv2.waitKey(1) & 0xFF == ord('q'):
break
# 释放资源
cap.release()
cv2.destroyAllWindows()
实时分析常见场景实现
场景一：实时人脸检测
结合OpenCV自带的人脸检测模型，可以对视频流中的每一帧进行人脸检测，实时标记出人脸位置。
import cv2
# 加载人脸检测级联模型
face_cascade = cv2.CascadeClassifier(cv2.data.haarcascades + "haarcascade_frontalface_default.xml")
# 初始化摄像头视频流
cap = cv2.VideoCapture(0)
while True:
ret, frame = cap.read()
if not ret:
break
# 转换为灰度图，提升检测效率
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
# 检测人脸，返回人脸区域的左上角坐标和宽高
faces = face_cascade.detectMultiScale(gray, scaleFactor=1.1, minNeighbors=5, minSize=(30, 30))
# 遍历检测到的人脸，绘制矩形框
for (x, y, w, h) in faces:
cv2.rectangle(frame, (x, y), (x+w, y+h), (0, 255, 0), 2)
cv2.imshow("Face Detection", frame)
if cv2.waitKey(1) & 0xFF == ord('q'):
break
cap.release()
cv2.destroyAllWindows()
场景二：实时运动检测
通过计算连续帧之间的差异，可以检测视频流中的运动物体，常用于监控场景。
import cv2
# 初始化视频流
cap = cv2.VideoCapture(0)
# 读取第一帧作为背景帧
ret, prev_frame = cap.read()
prev_gray = cv2.cvtColor(prev_frame, cv2.COLOR_BGR2GRAY)
# 对背景帧进行高斯模糊，减少噪声干扰
prev_gray = cv2.GaussianBlur(prev_gray, (21, 21), 0)
while True:
ret, frame = cap.read()
if not ret:
break
# 当前帧转灰度并模糊
curr_gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
curr_gray = cv2.GaussianBlur(curr_gray, (21, 21), 0)
# 计算当前帧和背景帧的差值
frame_diff = cv2.absdiff(prev_gray, curr_gray)
# 差值二值化，阈值设为25
_, thresh = cv2.threshold(frame_diff, 25, 255, cv2.THRESH_BINARY)
# 膨胀操作，填充小的空洞
thresh = cv2.dilate(thresh, None, iterations=2)
# 查找轮廓
contours, _ = cv2.findContours(thresh, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
# 遍历轮廓，过滤掉面积过小的区域
for contour in contours:
if cv2.contourArea(contour) < 500:
continue
# 绘制运动区域的矩形框
x, y, w, h = cv2.boundingRect(contour)
cv2.rectangle(frame, (x, y), (x+w, y+h), (0, 0, 255), 2)
cv2.imshow("Motion Detection", frame)
# 更新背景帧为当前帧
prev_gray = curr_gray
if cv2.waitKey(1) & 0xFF == ord('q'):
break
cap.release()
cv2.destroyAllWindows()
注意事项
- 处理高分辨率视频流时，可以适当降低帧的分辨率，减少计算量，提升处理速度
- 实时处理场景下，如果单帧处理时间过长，会出现画面卡顿，需要优化处理逻辑或者降低帧率
- 使用完视频流后一定要调用
release方法释放资源，避免占用摄像头或者文件句柄 - 网络视频流读取时可能会因为网络波动出现断连，需要添加重连逻辑保证程序稳定性