Python视频处理基础

"""
Python OpenCV 视频处理基础详解
包含：VideoCapture 读取、VideoWriter 写入、逐帧处理、FPS 控制
"""
import cv2
import numpy as np
import time
import os


def process_frame(frame, effect='gray'):
"""对单帧图像应用不同的处理效果"""
if effect == 'gray':
# 转为灰度图并转回三通道以保持视频格式
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
return cv2.cvtColor(gray, cv2.COLOR_GRAY2BGR)
elif effect == 'edge':
# Canny 边缘检测效果
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
blurred = cv2.GaussianBlur(gray, (3, 3), 0)
edges = cv2.Canny(blurred, 50, 150)
return cv2.cvtColor(edges, cv2.COLOR_GRAY2BGR)
elif effect == 'negative':
# 颜色取反（底片效果）
return cv2.bitwise_not(frame)
elif effect == 'mosaic':
# 局部马赛克效果（对中心区域做像素化）
h, w = frame.shape[:2]
frame_out = frame.copy()
cx, cy, size = w // 2, h // 2, 40
roi = frame_out[cy - size:cy + size, cx - size:cx + size]
small = cv2.resize(roi, (10, 10), interpolation=cv2.INTER_LINEAR)
mosaic = cv2.resize(small, (size * 2, size * 2),
interpolation=cv2.INTER_NEAREST)
frame_out[cy - size:cy + size, cx - size:cx + size] = mosaic
return frame_out
return frame


# ========== 1. 打开视频文件或摄像头 ==========
# 参数可以是视频文件路径、摄像头 ID（0 表示默认摄像头）或 URL
input_path = 'input_video.mp4' # 如文件不存在，尝试打开摄像头
cap = cv2.VideoCapture(input_path)
if not cap.isOpened():
print(f"无法打开视频文件 {input_path}，尝试打开摄像头...")
cap = cv2.VideoCapture(0) # 0 为默认摄像头

if not cap.isOpened():
print("无法打开任何视频源，生成模拟视频用于演示。")
# 创建一个虚拟帧来源
fps = 30
width, height = 640, 480
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
out = cv2.VideoWriter('output_video.mp4', fourcc, fps, (width, height))
# 生成 150 帧模拟动画
for i in range(150):
frame = np.zeros((height, width, 3), dtype=np.uint8)
cv2.circle(frame, (320, 240), 50 + i % 100,
(0, 255 * i // 150, 255), -1)
out.write(frame)
if i % 30 == 0:
print(f"生成模拟帧: {i // 30 + 1}s")
out.release()
print("模拟视频已生成。重新读取...")
cap = cv2.VideoCapture('output_video.mp4')

# ========== 2. 获取视频属性信息 ==========
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)) # 总帧数
duration = total_frames / fps if fps > 0 else 0 # 视频总时长（秒）

print(f"视频信息: {width}x{height}, {fps:.2f} FPS, "
f"{total_frames} 帧, 时长 {duration:.1f}s")

# ========== 3. 设置 VideoWriter 用于输出 ==========
# fourcc 指定编码格式，'mp4v' 对应 MP4 容器
output_path = 'processed_video.mp4'
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
out = cv2.VideoWriter(output_path, fourcc, fps, (width, height))

# ========== 4. 逐帧读取、处理并写入 ==========
frame_count = 0
start_time = time.time()
effects_cycle = ['gray', 'edge', 'negative', 'mosaic']

while True:
ret, frame = cap.read()
if not ret:
break # 视频结束或读取失败

# 根据帧号循环切换效果
effect = effects_cycle[(frame_count // 60) % len(effects_cycle)]
processed = process_frame(frame, effect)

# 在帧上叠加当前信息（帧号和效果名称）
info_text = f"Frame: {frame_count} Effect: {effect}"
cv2.putText(processed, info_text, (20, 40),
cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 0), 2)

# 写入输出视频
out.write(processed)
frame_count += 1

# 控制处理速度：保持与视频原始 fps 同步
elapsed = time.time() - start_time
expected_frames = int(elapsed * fps)
if frame_count > expected_frames:
time.sleep(1.0 / fps)

# 按帧率显示进度（每 30 帧输出一次）
if frame_count % 30 == 0:
progress = frame_count / total_frames * 100 if total_frames else 0
print(f"处理进度: {frame_count}/{total_frames} ({progress:.1f}%)")

# ========== 5. 释放资源 ==========
cap.release()
out.release()
cv2.destroyAllWindows()

elapsed_total = time.time() - start_time
print(f"\n视频处理完成！")
print(f"处理帧数: {frame_count}")
print(f"实际帧率: {frame_count / elapsed_total:.2f} FPS")
print(f"输出文件: {output_path}")
print(f"视频处理基础演示完成，涵盖读取、处理、写入和 FPS 控制。")