3. 多路视频流同步采集

任务目标

在边缘设备上同时接入多路USB摄像头或RTSP视频流时，各通道独立解码导致帧到达时刻不一致。本案例实现一个基于OpenCV的同步采集器，以主时钟为基准为每帧打上时间戳，并通过缓冲队列将时间差小于阈值的帧组打包输出，确保后续联合分析（如多视角拼接、深度估计）输入帧在时间上对齐。

核心实现

import threading
import time
from collections import deque
from dataclasses import dataclass
from typing import Dict, List, Optional

import cv2
import numpy as np

@dataclass
class FramePacket:
    """一帧图像及其采集时间（秒，单调时钟）"""
    stream_id: str
    timestamp: float
    image: np.ndarray

class SyncFrameGrabber:
    """
    多路视频流同步采集器。
    每路视频在独立线程中读取帧并打时间戳，主线程按时间窗对齐打包。
    """

    def __init__(self, sources: Dict[str, str], sync_window_ms: float = 30.0):
        """
        :param sources: 流名称 -> 视频源（摄像头索引或RTSP URL）
        :param sync_window_ms: 最大允许时间差（毫秒），超过则丢弃较早帧
        """
        self.sources = sources
        self.sync_window_s = sync_window_ms / 1000.0
        self._caps: Dict[str, cv2.VideoCapture] = {}
        self._queues: Dict[str, deque] = {name: deque(maxlen=30) for name in sources}
        self._locks = {name: threading.Lock() for name in sources}
        self._stop_event = threading.Event()
        self._threads: List[threading.Thread] = []

    def start(self) -> None:
        """打开所有视频源并启动采集线程"""
        for name, src in self.sources.items():
            cap = cv2.VideoCapture(src)
            if not cap.isOpened():
                raise RuntimeError(f"无法打开视频源: {name} ({src})")
            # 降低延迟：关闭OpenCV内部缓冲，只保留最新帧
            cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
            self._caps[name] = cap

            t = threading.Thread(target=self._capture_loop, args=(name,), daemon=True)
            t.start()
            self._threads.append(t)

    def _capture_loop(self, stream_id: str) -> None:
        """单路采集循环：读取帧并记录单调时钟时间戳"""
        cap = self._caps[stream_id]
        while not self._stop_event.is_set():
            ret, frame = cap.read()
            if not ret:
                break
            ts = time.monotonic()
            with self._locks[stream_id]:
                self._queues[stream_id].append(FramePacket(stream_id, ts, frame))

    def get_synced_frames(self, timeout_s: float = 1.0) -> Optional[Dict[str, FramePacket]]:
        """
        取一组时间对齐的帧。
        以最早到达的帧为基准，收集各通道时间差在窗口内的最新帧。
        若某通道在超时时间内无新帧，则返回None。
        """
        # 等待所有通道至少有一帧
        deadline = time.monotonic() + timeout_s
        while time.monotonic() < deadline:
            all_ready = all(len(q) > 0 for q in self._queues.values())
            if all_ready:
                break
            time.sleep(0.001)
        else:
            return None

        # 取各通道队首帧（最早帧）作为基准
        with self._locks:
            pass  # 占位，实际逐通道加锁

        # 基准帧：所有通道队首中时间戳最小的
        heads = {}
        for name in self.sources:
            with self._locks[name]:
                heads[name] = self._queues[name][0]

        base_ts = min(p.timestamp for p in heads.values())

        # 从各队列中取出与基准时间差在窗口内的最新帧
        result = {}
        for name in self.sources:
            with self._locks[name]:
                q = self._queues[name]
                # 丢弃时间戳早于基准-窗口的过期帧
                while q and q[0].timestamp < base_ts - self.sync_window_s:
                    q.popleft()
                # 从队首开始找第一个在窗口内的帧
                chosen = None
                for pkt in q:
                    if abs(pkt.timestamp - base_ts) <= self.sync_window_s:
                        chosen = pkt
                        break
                if chosen is None:
                    return None  # 该通道无对齐帧
                result[name] = chosen
        return result

    def stop(self) -> None:
        """停止所有采集线程并释放资源"""
        self._stop_event.set()
        for t in self._threads:
            t.join(timeout=1.0)
        for cap in self._caps.values():
            cap.release()

# ---------- 最小可运行示例 ----------
if __name__ == "__main__":
    # 使用两个USB摄像头（索引0和1）；若只有单摄像头，可改用RTSP流或视频文件
    sources = {
        "cam_left": 0,
        "cam_right": 1,
    }

    grabber = SyncFrameGrabber(sources, sync_window_ms=30)
    grabber.start()

    try:
        for i in range(50):  # 采集50组同步帧
            packet_group = grabber.get_synced_frames(timeout_s=2.0)
            if packet_group is None:
                print(f"[{i}] 超时未获取到同步帧组")
                continue

            # 验证时间对齐效果
            ts_list = [p.timestamp for p in packet_group.values()]
            spread_ms = (max(ts_list) - min(ts_list)) * 1000
            print(f"[{i}] 帧组时间差: {spread_ms:.1f} ms | "
                  f"通道: {list(packet_group.keys())}")

            # 此处可对packet_group中的帧做联合处理（如拼接、深度估计）
            # 示例：将两帧水平拼接显示
            left = packet_group["cam_left"].image
            right = packet_group["cam_right"].image
            h = min(left.shape[0], right.shape[0])
            combined = np.hstack([left[:h], right[:h]])
            cv2.imshow("Synced Frames", combined)
            if cv2.waitKey(1) & 0xFF == ord('q'):
                break
    finally:
        grabber.stop()
        cv2.destroyAllWindows()

运行与验证

依赖安装：pip install opencv-python numpy
硬件准备：连接至少两路视频源（USB摄像头或RTSP流）。若只有一路物理摄像头，可将第二个源改为视频文件路径（如"test.mp4"）或RTSP测试流地址。
运行：python sync_grabber.py
通过标准：程序持续输出帧组时间差，且绝大多数帧组的spread_ms小于设定的sync_window_ms（30ms）。若某通道掉帧或延迟过大，会打印“超时未获取到同步帧组”，此时应检查视频源稳定性或增大同步窗口。

关键设计说明

时间基准：使用time.monotonic()而非系统时钟，避免系统时间跳变影响同步精度。
缓冲策略：每路通道维护一个容量为30的队列，get_synced_frames()以最早帧为基准，丢弃过期帧并选取窗口内的最新帧，兼顾同步性与实时性。
延迟控制：设置CAP_PROP_BUFFERSIZE=1使OpenCV只保留最新帧，避免内部缓冲引入额外延迟。
线程安全：每路队列独立加锁，主线程取帧时逐通道锁操作，避免竞争条件。

上图展示了双通道采集时的时间对齐逻辑：通道1与通道2各自独立到达，同步器以最早帧为基准，将时间差在30ms内的帧归为一组输出；超出窗口的旧帧被丢弃，避免累积延迟。

📚 课程目录

3. 多路视频流同步采集

任务目标

核心实现

运行与验证

关键设计说明