高效读写视频(decord读,cv2写)
·
建议:
-
🎯 读取视频: 使用
decord(最快) -
💾 保存图像: 使用
PIL或OpenCV -
🎥 保存视频: 使用
OpenCV或imageio
使用decord来加载视频
使用cv2来保存图像和视频
1. 保存单张图像
基础保存
import cv2
import numpy as np
# 创建示例图像 (OpenCV 使用 BGR 格式)
image_bgr = np.random.randint(0, 255, (224, 224, 3), dtype=np.uint8)
# 保存图像
cv2.imwrite('output_image.jpg', image_bgr)
print("图像保存成功")
支持的不同格式
# JPEG (有损压缩,文件小)
cv2.imwrite('image.jpg', image_bgr, [cv2.IMWRITE_JPEG_QUALITY, 95])
# PNG (无损压缩,支持透明度)
cv2.imwrite('image.png', image_bgr, [cv2.IMWRITE_PNG_COMPRESSION, 9])
# BMP (无压缩,文件大)
cv2.imwrite('image.bmp', image_bgr)
# TIFF (高质量)
cv2.imwrite('image.tiff', image_bgr)
质量参数设置
# JPEG 质量 (0-100, 默认95)
cv2.imwrite('high_quality.jpg', image_bgr, [cv2.IMWRITE_JPEG_QUALITY, 100])
cv2.imwrite('low_quality.jpg', image_bgr, [cv2.IMWRITE_JPEG_QUALITY, 10])
# PNG 压缩级别 (0-9, 默认1)
cv2.imwrite('high_compression.png', image_bgr, [cv2.IMWRITE_PNG_COMPRESSION, 9])
cv2.imwrite('low_compression.png', image_bgr, [cv2.IMWRITE_PNG_COMPRESSION, 0])
2. 保存视频
基础视频保存
fps = 25
width, height = 224, 224
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
out = cv2.VideoWriter("output.mp4", fourcc, fps, (width, height))
for frame in video:
out.write(frame)
out.release()
不同视频编码格式
def save_with_different_codecs(frames, output_base_path):
"""
使用不同编码器保存视频
"""
height, width = frames[0].shape[:2]
fps = 30
# 各种编码器
codecs = {
'mp4v': 'output_mp4v.mp4', # MPEG-4 编码
'avc1': 'output_avc1.mp4', # H.264 编码
'XVID': 'output_xvid.avi', # AVI 格式
'MJPG': 'output_mjpg.avi', # Motion JPEG
}
for codec, output_path in codecs.items():
try:
fourcc = cv2.VideoWriter_fourcc(*codec)
out = cv2.VideoWriter(output_path, fourcc, fps, (width, height))
for frame in frames:
out.write(frame)
out.release()
print(f"使用 {codec} 编码器保存成功: {output_path}")
except Exception as e:
print(f"编码器 {codec} 失败: {e}")
# 使用示例
sample_frames = [np.random.randint(0, 255, (480, 640, 3), dtype=np.uint8)
for _ in range(50)]
save_with_different_codecs(sample_frames, 'output')
3. 完整工作流程:decord读取 + OpenCV保存
import decord
import cv2
import numpy as np
def video_processing_pipeline(input_path, output_path, target_size=None):
"""
完整视频处理流程:读取 → 处理 → 保存
"""
# 1. 使用 decord 读取视频
vr = decord.VideoReader(input_path)
original_fps = vr.get_avg_fps()
frames = vr.get_batch(range(len(vr))).asnumpy()
print(f"原始视频: {len(frames)} 帧, 分辨率: {frames[0].shape[:2]}")
# 2. 处理帧 (示例:调整大小和颜色增强)
processed_frames = []
for frame in frames:
# 转换为 BGR (OpenCV 格式)
frame_bgr = cv2.cvtColor(frame, cv2.COLOR_RGB2BGR)
# 调整大小 (如果指定)
if target_size:
frame_bgr = cv2.resize(frame_bgr, target_size)
# 颜色增强示例
hsv = cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2HSV)
hsv[:, :, 1] = cv2.multiply(hsv[:, :, 1], 1.2) # 增加饱和度
frame_bgr = cv2.cvtColor(hsv, cv2.COLOR_HSV2BGR)
processed_frames.append(frame_bgr)
# 3. 使用 OpenCV 保存视频
height, width = processed_frames[0].shape[:2]
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
out = cv2.VideoWriter(output_path, fourcc, original_fps, (width, height))
for frame in processed_frames:
out.write(frame)
out.release()
print(f"处理完成: {output_path}")
# 使用
video_processing_pipeline('input_video.mp4', 'output_processed.mp4', target_size=(640, 480))
4. 高级功能
保存带音频的视频(需要额外处理)
def extract_and_reinsert_audio(input_video, processed_video, output_video):
"""
提取原视频音频并重新插入到处理后的视频中
需要安装 moviepy
"""
from moviepy.editor import VideoFileClip, AudioFileClip
# 提取音频
video_clip = VideoFileClip(input_video)
audio = video_clip.audio
# 将音频添加到处理后的视频
processed_clip = VideoFileClip(processed_video)
final_clip = processed_clip.set_audio(audio)
# 保存带音频的视频
final_clip.write_videofile(output_video, codec='libx264')
video_clip.close()
processed_clip.close()
final_clip.close()
# 使用
# video_processing_pipeline('input.mp4', 'processed_no_audio.mp4')
# extract_and_reinsert_audio('input.mp4', 'processed_no_audio.mp4', 'final_with_audio.mp4')
批量处理图像
def save_image_batch(images, output_dir, base_name='image'):
"""
批量保存图像
"""
import os
os.makedirs(output_dir, exist_ok=True)
for i, image in enumerate(images):
# 确保是 BGR 格式
if len(image.shape) == 3 and image.shape[2] == 3:
# 假设输入是 RGB,转换为 BGR
image_bgr = cv2.cvtColor(image, cv2.COLOR_RGB2BGR)
else:
image_bgr = image
filename = f"{base_name}_{i:04d}.jpg"
filepath = os.path.join(output_dir, filename)
cv2.imwrite(filepath, image_bgr)
print(f"保存了 {len(images)} 张图像到 {output_dir}")
# 使用示例
images = [np.random.randint(0, 255, (224, 224, 3), dtype=np.uint8) for _ in range(10)]
save_image_batch(images, 'output_images')
5. 实用工具函数
视频信息检查
def get_video_info(video_path):
"""
获取视频信息
"""
cap = cv2.VideoCapture(video_path)
info = {
'width': int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)),
'height': int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)),
'fps': cap.get(cv2.CAP_PROP_FPS),
'total_frames': int(cap.get(cv2.CAP_PROP_FRAME_COUNT)),
'duration': int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) / cap.get(cv2.CAP_PROP_FPS)
}
cap.release()
return info
# 使用
info = get_video_info('input_video.mp4')
print(f"视频信息: {info}")
保存进度显示
def save_video_with_progress(frames, output_path, fps=30):
"""
带进度显示的视频保存
"""
if len(frames) == 0:
print("没有帧可保存")
return
height, width = frames[0].shape[:2]
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
out = cv2.VideoWriter(output_path, fourcc, fps, (width, height))
total_frames = len(frames)
for i, frame in enumerate(frames):
out.write(frame)
# 显示进度
if i % 10 == 0 or i == total_frames - 1:
progress = (i + 1) / total_frames * 100
print(f"保存进度: {progress:.1f}% ({i+1}/{total_frames})")
out.release()
print(f"视频保存完成: {output_path}")
注意事项:
-
颜色空间: OpenCV 使用 BGR,记得转换
-
数据类型: 确保是
uint8(0-255) -
文件路径: 确保有写入权限
-
资源释放: 记得调用
release()
更多推荐



所有评论(0)