如何通过YOLO异步函数实现ZED相机无丢帧视频录制？

我正在开发基于YOLO和ZED立体相机（60fps）的自动录制系统，设定每秒运行一次YOLO，检测到人物时持续录制视频，连续5次未检测到人物则停止录制。但运行YOLO时会出现丢帧，导致视频无法完整录制。我尝试在代码中使用异步函数解决该问题，但未成功。以下是我的测试伪代码：

import time
from utils.detect import YoloONNX
from common.config import VIDEO_CODEC
from utils.util import *
import cv2
import imutils
import os
import numpy as np
import pyzed.sl as sl
import asyncio

model = YoloONNX("./models/yolov7.onnx")

async def prepare_detect_person(frame):
    try:
        print('prepare_detect_person')
        frame = imutils.resize(frame, width=600)
        loop = asyncio.get_event_loop()
        await loop.run_in_executor(None, detect_person, frame)
    except Exception as e:
        print(f"Error occurred: {str(e)}")
def detect_person(frame):
    res = model.onnx_inference(frame)

async def main():
    zed = sl.Camera()

    init_params = sl.InitParameters()
    init_params.camera_resolution = sl.RESOLUTION.HD720
    init_params.camera_fps = 60

    err = zed.open(init_params)
    if err != sl.ERROR_CODE.SUCCESS:
        print("Camera Open : "+repr(err)+". Exit program.")
        exit()

    fourcc = cv2.VideoWriter_fourcc(*VIDEO_CODEC)
    video_writer = cv2.VideoWriter('./async_test.mp4', fourcc, 60,(1280, 720))
    i = 0
    max_frame = 10000
    image = sl.Mat()
    runtime_parameters = sl.RuntimeParameters()
    while i < max_frame:
        if zed.grab(runtime_parameters) == sl.ERROR_CODE.SUCCESS:
            zed.retrieve_image(image, sl.VIEW.LEFT)
            frame = image.get_data()
            frame = cv2.cvtColor(frame, cv2.COLOR_RGBA2RGB)
            if i % 30 == 0:
                print(f'start detect person')
                # await prepare_detect_person(frame)
                asyncio.create_task(prepare_detect_person(frame))
            i += 1
            video_writer.write(frame)

    video_writer.release()
    zed.close()

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    loop.close()

请问能否在运行YOLO模型的同时实现无丢帧的视频写入？

核心问题是YOLO推理的阻塞导致相机帧读取/写入被打断，异步函数未生效的原因是asyncio更适合IO密集型任务，而YOLO属于CPU/GPU密集型任务，需要用多进程/线程池彻底解耦检测与录制流程。

方案1：多进程分离检测与录制（推荐） #

用独立进程处理YOLO检测，主线程专注帧采集和视频写入，完全避免阻塞：

import time
from utils.detect import YoloONNX
from common.config import VIDEO_CODEC
import cv2
import imutils
import numpy as np
import pyzed.sl as sl
from multiprocessing import Process, Queue

# 进程间通信队列
detect_queue = Queue(maxsize=10)  # 传递待检测帧
result_queue = Queue(maxsize=10)  # 传递检测结果

def detect_worker():
    """独立进程中的YOLO检测逻辑"""
    model = YoloONNX("./models/yolov7.onnx")
    while True:
        frame = detect_queue.get()
        if frame is None:  # 收到结束信号
            break
        # 预处理+推理
        frame_resized = imutils.resize(frame, width=600)
        res = model.onnx_inference(frame_resized)
        # 判断是否检测到人物（根据你的YOLO输出格式调整）
        has_person = any(obj.get("class") == "person" for obj in res)
        result_queue.put(has_person)

def main():
    # 启动检测进程
    detect_process = Process(target=detect_worker)
    detect_process.start()

    # 初始化ZED相机
    zed = sl.Camera()
    init_params = sl.InitParameters()
    init_params.camera_resolution = sl.RESOLUTION.HD720
    init_params.camera_fps = 60

    err = zed.open(init_params)
    if err != sl.ERROR_CODE.SUCCESS:
        print(f"Camera Open : {repr(err)}. Exit program.")
        detect_queue.put(None)
        detect_process.join()
        exit()

    fourcc = cv2.VideoWriter_fourcc(*VIDEO_CODEC)
    video_writer = None
    consecutive_no_person = 0
    record_flag = False
    i = 0
    max_frame = 10000
    image = sl.Mat()
    runtime_parameters = sl.RuntimeParameters()

    while i < max_frame:
        if zed.grab(runtime_parameters) == sl.ERROR_CODE.SUCCESS:
            # 获取并预处理帧
            zed.retrieve_image(image, sl.VIEW.LEFT)
            frame = image.get_data()
            frame = cv2.cvtColor(frame, cv2.COLOR_RGBA2RGB)
            
            # 每秒一次检测（60fps，每60帧触发一次）
            if i % 60 == 0 and not detect_queue.full():
                detect_queue.put(frame.copy())  # 传递帧副本，避免内存冲突
            
            # 处理检测结果，更新录制状态
            while not result_queue.empty():
                has_person = result_queue.get()
                if has_person:
                    consecutive_no_person = 0
                    if not record_flag:
                        # 开始录制，生成带时间戳的视频文件
                        video_writer = cv2.VideoWriter(f'./recording_{int(time.time())}.mp4', fourcc, 60, (1280, 720))
                        record_flag = True
                        print("开始录制")
                else:
                    consecutive_no_person += 1
                    if consecutive_no_person >= 5 and record_flag:
                        video_writer.release()
                        record_flag = False
                        consecutive_no_person = 0
                        print("停止录制")
            
            # 录制状态下写入帧
            if record_flag:
                video_writer.write(frame)
            
            i += 1

    # 清理资源
    if video_writer is not None:
        video_writer.release()
    zed.close()
    detect_queue.put(None)  # 通知检测进程结束
    detect_process.join()

if __name__ == "__main__":
    main()

关键优化点 #

• 多进程解耦：检测任务完全脱离主线程，不会影响帧采集和写入的实时性
• 队列通信：用Queue保证进程间数据传递的线程安全
• 帧副本传递：避免ZED共享内存帧在多进程间的冲突
• 录制逻辑精准实现：严格按照“检测到人持续录制，连续5次未检测到停止”的需求执行

方案2：线程池处理检测任务 #

如果不想用多进程，可使用线程池将检测任务异步化，同时保证主线程的帧写入优先级：

import time
from utils.detect import YoloONNX
from common.config import VIDEO_CODEC
import cv2
import imutils
import numpy as np
import pyzed.sl as sl
from concurrent.futures import ThreadPoolExecutor

model = YoloONNX("./models/yolov7.onnx")
executor = ThreadPoolExecutor(max_workers=1)  # 单线程避免YOLO并发冲突
consecutive_no_person = 0
record_flag = False
video_writer = None
fourcc = cv2.VideoWriter_fourcc(*VIDEO_CODEC)

def detect_person(frame):
    """YOLO检测逻辑"""
    frame_resized = imutils.resize(frame, width=600)
    res = model.onnx_inference(frame_resized)
    return any(obj.get("class") == "person" for obj in res)

def handle_result(future):
    """处理检测结果的回调函数"""
    global consecutive_no_person, record_flag, video_writer
    has_person = future.result()
    if has_person:
        consecutive_no_person = 0
        if not record_flag:
            video_writer = cv2.VideoWriter(f'./recording_{int(time.time())}.mp4', fourcc, 60, (1280, 720))
            record_flag = True
    else:
        consecutive_no_person += 1
        if consecutive_no_person >= 5 and record_flag:
            video_writer.release()
            record_flag = False
            consecutive_no_person = 0

def main():
    # 初始化ZED相机
    zed = sl.Camera()
    init_params = sl.InitParameters()
    init_params.camera_resolution = sl.RESOLUTION.HD720
    init_params.camera_fps = 60

    err = zed.open(init_params)
    if err != sl.ERROR_CODE.SUCCESS:
        print(f"Camera Open : {repr(err)}. Exit program.")
        exit()

    i = 0
    max_frame = 10000
    image = sl.Mat()
    runtime_parameters = sl.RuntimeParameters()

    while i < max_frame:
        if zed.grab(runtime_parameters) == sl.ERROR_CODE.SUCCESS:
            zed.retrieve_image(image, sl.VIEW.LEFT)
            frame = image.get_data()
            frame = cv2.cvtColor(frame, cv2.COLOR_RGBA2RGB)
            
            # 每秒提交一次检测任务
            if i % 60 == 0:
                future = executor.submit(detect_person, frame.copy())
                future.add_done_callback(handle_result)
            
            # 录制状态下写入帧
            if record_flag:
                video_writer.write(frame)
            
            i += 1

    # 清理资源
    if video_writer is not None:
        video_writer.release()
    zed.close()
    executor.shutdown()

if __name__ == "__main__":
    main()

原异步代码失效原因 #

• asyncio的run_in_executor虽能将同步任务放入线程池，但你未处理检测结果，也未确保帧写入的优先级
• YOLO属于CPU/GPU密集型任务，异步IO模型更适合处理文件读写、网络请求这类IO密集型场景，多进程/线程池才是正确选择

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