企业官网建设流程全解析

076、视频流推理：cv2.VideoCapture到逐帧推理到结果叠加到cv2.VideoWriter 的完整工程代码

一、一个让我熬夜到凌晨三点的bug

去年给某安防项目做实时检测，现场摄像头推流RTSP，我自信满满地写了个循环：cap.read()→ 模型推理 →writer.write()。结果部署上去，视频文件打不开，或者打开后画面卡成PPT，帧率从25fps掉到3fps。更离谱的是，有一次推理结果框全画歪了，坐标偏移了半个画面。

排查了一整夜，最后发现三个问题：cv2.VideoWriter的编码器参数没对齐、cap.read()返回的帧格式和模型输入尺寸不一致、以及推理结果坐标映射时忘了缩放。这些坑，今天一次性给你填平。

二、cv2.VideoCapture：别以为只是打开摄像头

很多人写视频读取，上来就cap = cv2.VideoCapture(0)，然后直接while True。但实际工程中，视频源可能是文件、RTSP流、或者USB摄像头，每种源的属性获取方式有细微差别。

importcv2# 视频源：0表示默认摄像头，也可以是文件路径或RTSP地址video_source="test.mp4"# 或者 "rtsp://192.168.1.100:554/stream1"cap=cv2.VideoCapture(video_source)# 这里踩过坑：一定要检查是否成功打开ifnotcap.isOpened():raiseIOError(f"无法打开视频源:{video_source}")# 获取视频属性，后面写VideoWriter要用fps=cap.get(cv2.CAP_PROP_FPS)# 原始帧率width=int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))# 原始宽度height=int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))# 原始高度total_frames=int(cap.get(cv2.CAP_PROP_FRAME_COUNT))# 总帧数，仅对文件有效print(f"视频信息:{width}x{height}@{fps}fps, 共{total_frames}帧")

别这样写：直接硬编码width=640, height=480。RTSP流的分辨率可能和文件不同，摄像头也可能支持不同分辨率。用cap.get()动态获取才是正道。

三、逐帧推理：模型输入尺寸和原始帧的映射关系

YOLO模型通常要求固定输入尺寸（比如640x640），但视频帧可能是1920x1080。这里有两个关键操作：保持宽高比的resize和坐标缩放。

importtorchimportnumpyasnpdefpreprocess_frame(frame,input_size=640):""" 预处理帧：保持宽高比resize，填充到正方形 别这样写：直接cv2.resize(frame, (input_size, input_size))，会拉伸变形 """h,w=frame.shape[:2]scale=min(input_size/h,input_size/w)new_h,new_w=int(h*scale),int(w*scale)# 先resize到等比例尺寸resized=cv2.resize(frame,(new_w,new_h),interpolation=cv2.INTER_LINEAR)# 创建画布，填充灰色（128,128,128）canvas=np.full((input_size,input_size,3),128,dtype=np.uint8)# 计算偏移量，居中放置dw=(input_size-new_w)//2dh=(input_size-new_h)//2canvas[dh:dh+new_h,dw:dw+new_w]=resized# 归一化并转tensorinput_tensor=torch.from_numpy(canvas.transpose(2,0,1)).float()/255.0input_tensor=input_tensor.unsqueeze(0)# 添加batch维度# 返回预处理后的tensor和缩放/偏移信息，后面坐标映射要用returninput_tensor,scale,dw,dh,w,hdefpostprocess_boxes(boxes,scale,dw,dh,orig_w,orig_h,input_size=640):""" 将模型输出的归一化坐标映射回原始帧 这里踩过坑：YOLO输出的是相对于输入尺寸的归一化坐标，需要反向映射 """# boxes格式: [x1, y1, x2, y2, conf, cls] 归一化到[0,1]# 先还原到input_size坐标系boxes[:,[0,2]]=boxes[:,[0,2]]*input_size boxes[:,[1,3]]=boxes[:,[1,3]]*input_size# 减去填充偏移boxes[:,[0,2]]-=dw boxes[:,[1,3]]-=dh# 除以缩放比例，得到原始帧坐标boxes[:,[0,2]]/=scale boxes[:,[1,3]]/=scale# 裁剪到原始帧边界，防止越界boxes[:,[0,2]]=np.clip(boxes[:,[0,2]],0,orig_w)boxes[:,[1,3]]=np.clip(boxes[:,[1,3]],0,orig_h)returnboxes.astype(np.int32)

别这样写：直接对模型输出坐标乘以原始宽高。因为模型输入经过了填充和缩放，坐标必须经过逆变换才能正确映射。

四、cv2.VideoWriter：编码器参数是最大的坑

写视频文件时，cv2.VideoWriter的构造函数有四个参数：文件名、编码器、帧率、尺寸。编码器参数fourcc在不同系统上表现不同，这是最容易出问题的地方。

defcreate_video_writer(output_path,fps,width,height,codec='mp4v'):""" 创建视频写入器 这里踩过坑：不同系统支持的编码器不同，Windows上'DIVX'更稳定 """# fourcc编码器选择fourcc_map={'mp4v':cv2.VideoWriter_fourcc(*'mp4v'),# MP4格式'avc1':cv2.VideoWriter_fourcc(*'avc1'),# H.264，macOS常用'xvid':cv2.VideoWriter_fourcc(*'XVID'),# AVI格式'divx':cv2.VideoWriter_fourcc(*'DIVX'),# 兼容性较好}fourcc=fourcc_map.get(codec,cv2.VideoWriter_fourcc(*'mp4v'))# 注意：输出尺寸必须和写入的帧尺寸一致，否则会报错或生成损坏文件writer=cv2.VideoWriter(output_path,fourcc,fps,(width,height))ifnotwriter.isOpened():# 尝试备用编码器print(f"编码器{codec}失败，尝试DIVX...")writer=cv2.VideoWriter(output_path,cv2.VideoWriter_fourcc(*'DIVX'),fps,(width,height))returnwriter

别这样写：直接cv2.VideoWriter('output.mp4', fourcc, 30, (640,480))。帧率和尺寸必须和原始视频匹配，否则播放器会报错。另外，writer.write()写入的帧尺寸必须和构造函数中指定的尺寸一致。

五、完整工程代码：从读取到推理到写入

把上面所有模块串起来，写一个完整的视频推理函数。这里我习惯用tqdm显示进度，因为处理长视频时，没有进度条会让人焦虑。

importcv2importtorchimportnumpyasnpfromtqdmimporttqdmdefvideo_inference(model,video_source,output_path,device='cuda',input_size=640,conf_thres=0.5,iou_thres=0.45):""" 完整视频推理流程 model: YOLO模型，支持forward返回boxes """# 1. 打开视频cap=cv2.VideoCapture(video_source)ifnotcap.isOpened():raiseIOError(f"无法打开视频:{video_source}")# 获取原始视频属性fps=cap.get(cv2.CAP_PROP_FPS)width=int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))height=int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))total_frames=int(cap.get(cv2.CAP_PROP_FRAME_COUNT))# 2. 创建视频写入器，保持原始尺寸writer=create_video_writer(output_path,fps,width,height)# 3. 逐帧处理frame_count=0pbar=tqdm(total=total_frames,desc="推理进度")whileTrue:ret,frame=cap.read()ifnotret:break# 预处理input_tensor,scale,dw,dh,orig_w,orig_h=preprocess_frame(frame,input_size)input_tensor=input_tensor.to(device)# 模型推理withtorch.no_grad():# 假设模型返回的boxes格式: [N, 6] 其中6=[x1,y1,x2,y2,conf,cls]# 这里踩过坑：不同YOLO版本的输出格式不同，需要根据实际模型调整outputs=model(input_tensor)[0]# 取batch中第一张# 后处理：NMS和阈值过滤boxes=non_max_suppression(outputs,conf_thres,iou_thres)# 如果检测到目标，映射坐标并绘制ifboxesisnotNoneandlen(boxes)>0:# 坐标映射回原始帧boxes=postprocess_boxes(boxes,scale,dw,dh,orig_w,orig_h,input_size)# 绘制检测框forboxinboxes:x1,y1,x2,y2,conf,cls_id=box# 这里可以自定义颜色和标签cv2.rectangle(frame,(x1,y1),(x2,y2),(0,255,0),2)label=f"class_{int(cls_id)}:{conf:.2f}"cv2.putText(frame,label,(x1,y1-10),cv2.FONT_HERSHEY_SIMPLEX,0.5,(0,255,0),2)# 写入帧writer.write(frame)frame_count+=1pbar.update(1)# 可选：显示实时画面（调试用）# cv2.imshow('Inference', frame)# if cv2.waitKey(1) & 0xFF == ord('q'):# break# 清理资源cap.release()writer.release()cv2.destroyAllWindows()pbar.close()print(f"处理完成:{frame_count}帧, 输出文件:{output_path}")

六、NMS函数：别用循环，用向量化

很多初学者自己写NMS用for循环，速度慢得离谱。这里给出一个向量化版本，用PyTorch或NumPy实现。

defnon_max_suppression(prediction,conf_thres=0.5,iou_thres=0.45):""" 向量化NMS，避免Python循环 prediction: [N, 6] 其中6=[x1,y1,x2,y2,conf,cls] """# 过滤低置信度mask=prediction[:,4]>conf_thres prediction=prediction[mask]iflen(prediction)==0:returnNone# 按置信度排序order=prediction[:,4].argsort(descending=True)prediction=prediction[order]boxes=prediction[:,:4]scores=prediction[:,4]# 计算所有框的面积area=(boxes[:,2]-boxes[:,0])*(boxes[:,3]-boxes[:,1])keep=[]whilelen(boxes)>0:# 取当前最高分的框i=0keep.append(i)# 计算与其余框的IoUxx1=torch.max(boxes[i,0],boxes[:,0])yy1=torch.max(boxes[i,1],boxes[:,1])xx2=torch.min(boxes[i,2],boxes[:,2])yy2=torch.min(boxes[i,3],boxes[:,3])w=torch.clamp(xx2-xx1,min=0)h=torch.clamp(yy2-yy1,min=0)inter=w*h iou=inter/(area[i]+area-inter)# 保留IoU小于阈值的框mask=iou<=iou_thres boxes=boxes[mask]area=area[mask]scores=scores[mask]returnprediction[keep]

别这样写：用for i in range(len(boxes))逐框比较，处理1000个框时速度会慢10倍以上。

七、性能优化：别让推理成为瓶颈

视频推理最怕帧率下降。这里分享几个实战优化技巧：

1. 异步读取：用多线程预读取帧，避免cap.read()阻塞推理。可以用queue.Queue实现生产者-消费者模式。

2. 跳过帧：如果模型推理速度跟不上视频帧率，可以设置skip_frame参数，每N帧推理一次，中间帧用上一帧的结果。对于监控场景，目标运动缓慢时效果不错。

3. 半精度推理：如果GPU支持，用model.half()将模型转为FP16，推理速度提升30%-50%，精度损失可忽略。

4. 批量推理：如果视频帧率很高，可以累积多帧组成batch一起推理，利用GPU并行计算。但要注意batch中帧的时间顺序。

# 异步读取示例（简化版）importthreadingimportqueueclassVideoReaderThread:def__init__(self,cap,queue_size=30):self.cap=cap self.queue=queue.Queue(maxsize=queue_size)self.running=Trueself.thread=threading.Thread(target=self._read_frames)self.thread.daemon=Trueself.thread.start()def_read_frames(self):whileself.running:ret,frame=self.cap.read()ifnotret:self.running=Falsebreakself.queue.put((ret,frame))defget_frame(self):returnself.queue.get()defstop(self):self.running=Falseself.thread.join()

八、个人经验性建议

1. 编码器选择：跨平台部署时，优先用'mp4v'或'XVID'。'avc1'在macOS上表现好，但Windows上可能不支持。如果遇到视频打不开，先检查writer.isOpened()的返回值。

2. 帧率匹配：不要随意修改输出视频的帧率。如果推理速度慢，可以降低输出帧率（比如原始30fps，输出15fps），但要在VideoWriter构造函数中明确指定。

3. 内存管理：处理长视频时，注意释放不再使用的tensor。用del和torch.cuda.empty_cache()定期清理显存，避免OOM。

4. 调试技巧：先处理10帧测试，确认坐标映射正确、视频能正常播放，再跑全量。我见过太多人直接跑10000帧，最后发现框画歪了。

5. 日志记录：在关键步骤加print或logging，记录每帧的推理耗时、检测目标数。这些数据对后续优化很有价值。

6. 异常处理：cap.read()可能返回空帧（网络流中断），writer.write()可能失败（磁盘空间不足）。加try-except并优雅退出，别让程序直接崩溃。

最后，记住一个原则：视频推理的代码，80%的时间花在数据预处理和后处理上，只有20%是模型推理。把前两个环节优化好，你的视频推理管线才能真正跑起来。