建议:

  • 🎯 读取视频: 使用 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}")

注意事项:

  1. 颜色空间: OpenCV 使用 BGR,记得转换

  2. 数据类型: 确保是 uint8 (0-255)

  3. 文件路径: 确保有写入权限

  4. 资源释放: 记得调用 release()

Logo

中国智能体开发者社区,聚焦智能体与大模型开发,提供前沿资讯、实用工具链、开源项目及行业案例。通过技术沙龙、开发者大赛等活动,促进经验交流与协作,助力开发者快速构建创新智能应用。

更多推荐