OpenCV-循环读取视频帧，对每一帧进行处理

原型代码

• 内存模型：
  • 核心变量：frame，Numpy ndarray，每次会被覆盖，大小保持恒定

import cv2

video_path = 'your_video.mp4'

cap = cv2.VideoCapture(video_path)

if not cap.isOpened():
	print("Can't open Video")
	exit()

while True:
	ret, frame = cap.read()

	if not ret:
		break

	# 
	# 在这里对每一帧进行处理
	# cv2.imshow('Frame', frame)

	if cv2.waitKey(25) & OxFF == ord('q'): # 25ms内，有按钮按下，且按下的按钮为q
		break

cap.release()
cv2.destroyAllWindows()

函数封装

对其进行一个函数封装，传入一个process函数，其对每个frame进行统一的处理，然后输出一个处理后的视频。

import cv2 
import numpy as np # 用于帧的类型提示 
from typing import Callable, Optional # 用于类型提示

def process_video_frames( video_file_path: str, frame_processor: Callable[[np.ndarray], np.ndarray], output_path=Optional[str] = None ) -> None:
	cap = cv2.VideoCaputer(video_path)
	if not cap.isOpened():
		print("Can't open Video")
		exit()

	# [[OpenCV-获取视频属性]]
	frame_width, frame_height, fps = get_video_properties(video_capture: cv2.VideoCapture)

	video_writer = None
	if output_path:
		fourcc = cv2.VideoWriter_fourcc(*'mp4v') # 定义视频编码器
		video_writer = cv2.VideoWriter(output_path, fourcc, fps, (frame_width, frame_height))
		if not video_writer.isOpened():
			print("Init Video Writer Failed.Processed Video will not be saved")
			video_writer = None
	
	while True:
		ret, frame = cap.read()
	
		if not ret:
			break
	
		# 
		# 在这里对使用process函数对每一帧进行进行处理
		processed_frame = frame_processor(current_frame)
	
		# 如果输出视频写入器已成功初始化，则写入处理后的帧
		if video_writer:
			# 确保处理后的帧尺寸与原始视频一致，否则可能无法写入
			if processed_frame.shape[:2] != (frame_height, frame_width):
				print(f"警告: 处理后的帧尺寸 {processed_frame.shape[:2]} 与原始视频尺寸 "
					  f"{(frame_height, frame_width)} 不匹配。可能导致写入问题。")
			video_writer.write(processed_frame)	

	
	cap.release()
	if video_writer:
		video_writer.release()
	cv2.destroyAllWindows()