OpenCV + Real-ESRGAN 视频超分实战：老视频 4K 修复完整方案

一、引言

视频超分辨率是比单图超分更复杂的任务——需要在保持帧间一致性的同时提升每帧的分辨率。本文将使用 OpenCV + Real-ESRGAN 构建一个完整的视频超分管线，支持从 480p 到 4K 的视频修复，包含去噪、去模糊、超分和帧间一致性优化。

二、技术方案设计

2.1 整体流程

输入视频 (480p/720p)
    │
    ▼
[OpenCV] 解码视频流 → 逐帧读取 (BGR)
    │
    ▼
[预处理] 色彩空间转换 (BGR → RGB)
    │   去噪 (Fast Non-Local Means)
    │   锐化 (Unsharp Masking)
    │
    ▼
[Real-ESRGAN] 超分辨率 (×4)
    │   分块处理 (Tile-based)
    │   FP16 推理加速
    │
    ▼
[后处理] RGB → BGR 转换
    │   帧间平滑（可选）
    │
    ▼
[OpenCV] 编码输出视频 (H.264/H.265)
    │   保持原帧率
    │
    ▼
输出视频 (4K)

2.2 关键技术挑战

三、完整代码实现

3.1 核心类 VideoSuperRes

import cv2
import torch
import numpy as np
from multiprocessing import Pool, cpu_count
from basicsr.archs.rrdbnet_arch import RRDBNet
from realesrgan import RealESRGANer
import argparse


class VideoSuperRes:
    \"\"\"视频超分辨率处理器\"\"\"

    def __init__(self, model_path='weights/RealESRGAN_x4plus.pth',
                 scale=4, tile=400, use_fp16=True, denoise_strength=3):
        # 初始化 Real-ESRGAN
        model = RRDBNet(num_in_ch=3, num_out_ch=3, num_feat=64,
                        num_block=23, num_grow_ch=32, scale=scale)
        self.upsampler = RealESRGANer(
            scale=scale,
            model_path=model_path,
            model=model,
            tile=tile,
            tile_pad=10,
            pre_pad=0,
            half=use_fp16
        )
        self.scale = scale
        self.denoise_strength = denoise_strength

    def preprocess_frame(self, frame):
        \"\"\"帧预处理：去噪 + 锐化\"\"\"
        # BGR → RGB
        if frame.shape[2] == 3:
            frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)

        # 非局部均值去噪
        if self.denoise_strength > 0:
            frame = cv2.fastNlMeansDenoisingColored(
                frame, None,
                self.denoise_strength,  # h (滤波强度)
                self.denoise_strength,  # hColor
                7, 21  # 模板窗口大小
            )

        # Unsharp Masking 锐化
        gaussian = cv2.GaussianBlur(frame, (0, 0), 2.0)
        frame = cv2.addWeighted(frame, 1.5, gaussian, -0.5, 0)

        return np.clip(frame, 0, 255).astype(np.uint8)

    def super_resolve_frame(self, frame):
        \"\"\"单帧超分辨率\"\"\"
        output, _ = self.upsampler.enhance(frame, outscale=self.scale)
        return output

    def process_video(self, input_path, output_path,
                      start_frame=0, end_frame=None,
                      crf=18):
        \"\"\"处理整个视频\"\"\"
        cap = cv2.VideoCapture(input_path)

        # 获取视频元信息
        fps = cap.get(cv2.CAP_PROP_FPS)
        total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
        width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
        height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))

        print(f"输入: {width}×{height}, {fps:.1f}fps, {total_frames}帧")
        print(f"输出: {width*self.scale}×{height*self.scale}, {fps:.1f}fps")

        # 设置输出编码器
        fourcc = cv2.VideoWriter_fourcc(*'avc1')  # H.264
        out = cv2.VideoWriter(
            output_path,
            fourcc,
            fps,
            (width * self.scale, height * self.scale)
        )

        if end_frame is None:
            end_frame = total_frames

        cap.set(cv2.CAP_PROP_POS_FRAMES, start_frame)

        frame_idx = start_frame
        while frame_idx < end_frame:
            ret, frame = cap.read()
            if not ret:
                break

            # 处理流水线
            processed = self.preprocess_frame(frame)
            sr_frame = self.super_resolve_frame(processed)

            # RGB → BGR 用于 OpenCV 编码
            sr_frame_bgr = cv2.cvtColor(sr_frame, cv2.COLOR_RGB2BGR)
            out.write(sr_frame_bgr)

            frame_idx += 1
            if frame_idx % 10 == 0:
                progress = (frame_idx - start_frame) / (end_frame - start_frame) * 100
                print(f"进度: {progress:.1f}% ({frame_idx}/{end_frame})")

        cap.release()
        out.release()
        print(f"完成! 输出: {output_path}")


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--input', '-i', required=True, help='输入视频路径')
    parser.add_argument('--output', '-o', required=True, help='输出视频路径')
    parser.add_argument('--scale', type=int, default=4, help='放大倍数')
    parser.add_argument('--tile', type=int, default=400, help='分块大小')
    parser.add_argument('--start', type=int, default=0)
    parser.add_argument('--end', type=int, default=None)
    args = parser.parse_args()

    processor = VideoSuperRes(scale=args.scale, tile=args.tile)
    processor.process_video(args.input, args.output,
                            args.start, args.end)

3.2 多进程加速（可选）

对于长视频，可以使用多进程并行处理：

from multiprocessing import Pool, cpu_count
import subprocess
import os

def process_video_segment(args):
    \"\"\"处理视频片段\"\"\"
    segment_path, output_path, start, end, scale = args
    processor = VideoSuperRes(scale=scale, tile=300)
    processor.process_video(segment_path, output_path,
                            start_frame=0, end_frame=end-start)
    return output_path

def parallel_video_super_res(input_path, output_path,
                              num_workers=None, scale=4):
    \"\"\"多进程视频超分\"\"\"
    if num_workers is None:
        num_workers = cpu_count() // 2

    cap = cv2.VideoCapture(input_path)
    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    fps = cap.get(cv2.CAP_PROP_FPS)
    cap.release()

    # 分割视频
    chunk_size = total_frames // num_workers
    temp_dir = 'temp_chunks/'
    os.makedirs(temp_dir, exist_ok=True)

    tasks = []
    for i in range(num_workers):
        start = i * chunk_size
        end = (i + 1) * chunk_size if i < num_workers - 1 else total_frames
        chunk_input = f'{temp_dir}chunk_{i}_input.mp4'
        chunk_output = f'{temp_dir}chunk_{i}_sr.mp4'

        # 使用 FFmpeg 切分视频
        subprocess.run([
            'ffmpeg', '-y', '-i', input_path,
            '-ss', str(start / fps),
            '-to', str((end - start) / fps),
            '-c', 'copy', chunk_input
        ], capture_output=True)

        tasks.append((chunk_input, chunk_output, start, end, scale))

    # 并行处理
    with Pool(num_workers) as pool:
        chunk_outputs = pool.map(process_video_segment, tasks)

    # 合并视频
    concat_file = f'{temp_dir}concat_list.txt'
    with open(concat_file, 'w') as f:
        for chunk in chunk_outputs:
            f.write(f"file '{os.path.abspath(chunk)}'\\n")

    subprocess.run([
        'ffmpeg', '-y', '-f', 'concat', '-safe', '0',
        '-i', concat_file, '-c', 'copy', output_path
    ])

    print(f"多进程处理完成: {output_path}")

四、帧间一致性优化

逐帧独立超分可能引入帧间闪烁。下面是基于光流的时域一致性损失方案：

import cv2

def temporal_consistency(prev_frame, curr_frame, alpha=0.3):
    \"\"\"
    简单的时域一致性平滑
    将当前帧与前帧混合，减少闪烁
    \"\"\"
    if prev_frame is None:
        return curr_frame

    # 计算光流（简化版）
    prev_gray = cv2.cvtColor(prev_frame, cv2.COLOR_BGR2GRAY)
    curr_gray = cv2.cvtColor(curr_frame, cv2.COLOR_BGR2GRAY)

    flow = cv2.calcOpticalFlowFarneback(
        prev_gray, curr_gray, None,
        0.5, 3, 15, 3, 5, 1.2, 0
    )

    # 基于光流 warp 前帧到当前帧
    h, w = flow.shape[:2]
    flow_map = np.column_stack([
        (np.arange(w) + flow[..., 0]).ravel(),
        (np.arange(h)[:, None] + flow[..., 1]).ravel()
    ])

    warped_prev = cv2.remap(
        prev_frame,
        flow_map[..., 0].reshape(h, w).astype(np.float32),
        flow_map[..., 1].reshape(h, w).astype(np.float32),
        cv2.INTER_LINEAR
    )

    # 指数移动平均混合
    result = cv2.addWeighted(curr_frame, 1 - alpha,
                              warped_prev, alpha, 0)
    return result

五、性能基准

在 RTX 3090 (24GB VRAM) 上测试：

六、总结

本文介绍了基于 OpenCV + Real-ESRGAN 的视频超分辨率完整方案，包括帧预处理、分块超分、后处理和帧间一致性优化。对于长视频，多进程并行可以将处理速度提升 3-4 倍。该方案可直接用于老视频修复、监控视频增强等场景。