高效提取视频帧与Landmarks并跨进程处理的最优方案

你的当前方案速度慢的核心原因有两个：一是把所有帧和Landmark全量加载到内存再序列化，既占用大量内存又拖慢写入速度；二是提取与处理完全串行，没有利用并行计算能力。以下是具体优化方案：

1. 避免全量内存存储，改用增量式写入磁盘 #

不要将所有帧存入列表，而是边提取边写入更高效的存储格式，大幅降低内存占用和序列化开销：

用HDF5格式（推荐） #

HDF5支持分块写入与随机读取，专为大量数值型数据设计，读写效率远高于pickle：

import h5py
import cv2
import numpy as np

def get_landmarks(frame):
    # 你的Landmark提取逻辑
    return ...

path = "your_video_path.mp4"
cap = cv2.VideoCapture(path)

# 创建HDF5文件
with h5py.File("data.h5", "w") as f:
    # 获取视频参数，预分配空间（可选，进一步提升写入效率）
    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    height, width = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)), int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    
    # 创建帧数据集（分块存储）
    frames_ds = f.create_dataset("frames", shape=(total_frames, height, width, 3), dtype=np.uint8, chunks=True)
    # 创建Landmark数据集（根据实际数据形状调整）
    landmarks_ds = f.create_dataset("landmarks", shape=(total_frames, 136), dtype=np.float32, chunks=True)
    
    idx = 0
    while cap.isOpened():
        success, frame = cap.read()
        if not success:
            break
        lm = get_landmarks(frame)
        # 写入当前帧与Landmark
        frames_ds[idx] = frame
        landmarks_ds[idx] = lm
        idx += 1
cap.release()

读取时可直接按索引从HDF5文件中加载数据，无需一次性读取全部内容。

用Numpy分批次存储 #

如果不想依赖HDF5库，也可以用numpy的savez分批次写入：

import numpy as np
import cv2

batch_size = 100
frames_batch = []
landmarks_batch = []
batch_idx = 0

cap = cv2.VideoCapture(path)
while cap.isOpened():
    success, frame = cap.read()
    if not success:
        break
    lm = get_landmarks(frame)
    frames_batch.append(frame)
    landmarks_batch.append(lm)
    
    # 每攒够batch_size就写入磁盘
    if len(frames_batch) == batch_size:
        np.savez(f"batch_{batch_idx}.npz", frames=np.array(frames_batch), landmarks=np.array(landmarks_batch))
        frames_batch.clear()
        landmarks_batch.clear()
        batch_idx += 1

# 写入剩余未完成的批次
if frames_batch:
    np.savez(f"batch_{batch_idx}.npz", frames=np.array(frames_batch), landmarks=np.array(landmarks_batch))
cap.release()

2. 启用跨进程并行处理（无需先写磁盘） #

让提取进程与处理进程同时运行，通过进程间通信（IPC）实时传递数据，彻底消除串行等待的时间损耗：

用多进程队列传递数据 #

适合数据量适中的场景，代码简单易维护：

import cv2
import multiprocessing as mp

def get_landmarks(frame):
    # 你的Landmark提取逻辑
    return ...

def extract_worker(path, queue):
    cap = cv2.VideoCapture(path)
    while cap.isOpened():
        success, frame = cap.read()
        if not success:
            break
        lm = get_landmarks(frame)
        # 将帧与Landmark传入队列
        queue.put((frame, lm))
    cap.release()
    # 放入结束标记
    queue.put(None)

def process_worker(queue):
    while True:
        data = queue.get()
        if data is None:
            break
        frame, lm = data
        # 你的帧与Landmark处理逻辑
        process_frame_and_landmark(frame, lm)

if __name__ == "__main__":
    path = "your_video_path.mp4"
    # 设置队列最大长度，防止提取进程过快导致内存溢出
    queue = mp.Queue(maxsize=10)
    
    # 启动两个进程
    extract_proc = mp.Process(target=extract_worker, args=(path, queue))
    process_proc = mp.Process(target=process_worker, args=(queue,))
    
    extract_proc.start()
    process_proc.start()
    
    extract_proc.join()
    process_proc.join()

用共享内存传递帧（推荐大帧场景） #

队列会对数据进行序列化/反序列化，大帧场景下开销较高。用共享内存可直接传递数组内存指针，避免数据拷贝：

import cv2
import multiprocessing as mp
import numpy as np
from multiprocessing.shared_memory import SharedMemory

def get_landmarks(frame):
    return ...

def extract_worker(path, shm_name, shape, dtype, queue):
    cap = cv2.VideoCapture(path)
    # 连接到共享内存
    shm = SharedMemory(name=shm_name)
    frame_shared = np.ndarray(shape, dtype=dtype, buffer=shm.buf)
    
    while cap.isOpened():
        success, frame = cap.read()
        if not success:
            break
        lm = get_landmarks(frame)
        # 将当前帧复制到共享内存
        frame_shared[:] = frame
        # 通知处理进程读取，同时传递Landmark
        queue.put((lm, True))
    
    cap.release()
    # 发送结束信号
    queue.put((None, False))
    shm.close()

def process_worker(shm_name, shape, dtype, queue):
    shm = SharedMemory(name=shm_name)
    frame_shared = np.ndarray(shape, dtype=dtype, buffer=shm.buf)
    
    while True:
        lm, has_data = queue.get()
        if not has_data:
            break
        # 直接从共享内存读取帧（如需保留副本可调用copy()）
        frame = frame_shared.copy()
        process_frame_and_landmark(frame, lm)
    
    shm.close()
    shm.unlink()  # 释放共享内存

if __name__ == "__main__":
    path = "your_video_path.mp4"
    cap = cv2.VideoCapture(path)
    height, width = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)), int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    cap.release()
    
    # 创建共享内存（单帧大小）
    shape = (height, width, 3)
    dtype = np.uint8
    shm = SharedMemory(create=True, size=np.prod(shape)*dtype.itemsize)
    
    queue = mp.Queue()
    
    extract_proc = mp.Process(target=extract_worker, args=(path, shm.name, shape, dtype, queue))
    process_proc = mp.Process(target=process_worker, args=(shm.name, shape, dtype, queue))
    
    extract_proc.start()
    process_proc.start()
    
    extract_proc.join()
    process_proc.join()
    
    shm.close()

3. 优化序列化格式（如果必须写磁盘） #

若仍需用文件存储替代IPC，替换pickle为更高效的格式：

• Numpy原生格式：np.save/np.savez比pickle快数倍，专门针对数组优化。
• Msgpack：比pickle更快更紧凑，适合结构化数据（如Landmark）。
• 压缩存储：用np.savez_compressed可减少磁盘占用，小数据量下性能损耗可忽略。

4. 优化Landmark提取逻辑 #

如果get_landmarks是性能瓶颈：

• 启用GPU加速：比如使用MediaPipe时设置model_complexity并确保GPU可用，或用TensorRT加速模型推理。
• 减少冗余计算：仅处理感兴趣区域（ROI），或在不影响精度的前提下降低帧分辨率。

内容的提问来源于stack exchange，提问作者connor449