2. 多路摄像头同步采集与时间戳对齐

任务说明

本案例实现一个多路摄像头同步采集器：为每路摄像头启动独立采集线程，采集到的每一帧都记录同一时钟源下的时间戳，并按“帧序号 + 各路时间戳”对齐输出，保证后续分析时各通道时间基准一致。

核心要点：

• 统一时钟：所有时间戳来自同一个 time.monotonic() 时钟源，避免各线程各自取时导致漂移。
• 同步采集：每路摄像头一个线程，主线程按帧序号收集，凑齐所有通道后输出一组对齐帧。
• 时间戳对齐：以“同一帧序号”为对齐键，记录每路帧的采集时刻，并计算相对基准的偏差。

环境与依赖

• Python 3.8+，仅使用标准库（threading、queue、time、dataclasses）。
• 真实摄像头接入时，把 CameraSource.read() 替换为 OpenCV 的 cap.read() 即可，其余同步逻辑不变。

核心实现

import threading, queue, time
from dataclasses import dataclass, field

@dataclass
class Frame:
    cam_id: int
    seq: int
    ts: float          # 统一时钟下的采集时间戳(秒)
    payload: object    # 真实场景为图像帧，这里用占位数据

class CameraSource:
    """模拟摄像头：按固定帧率产出帧。真实场景替换 read() 为 cap.read()。"""
    def __init__(self, cam_id, fps=30.0):
        self.cam_id = cam_id
        self.interval = 1.0 / fps
        self._next = time.monotonic()

    def read(self, seq):
        now = time.monotonic()
        if now < self._next:
            time.sleep(self._next - now)
        self._next += self.interval
        return Frame(self.cam_id, seq, time.monotonic(), payload=f"cam{self.cam_id}-f{seq}")

class SyncCollector:
    """多路同步采集：每路一个线程，主线程按帧序号对齐。"""
    def __init__(self, cam_ids, fps=30.0, max_frames=10):
        self.cam_ids = cam_ids
        self.max_frames = max_frames
        self.sources = {c: CameraSource(c, fps) for c in cam_ids}
        self.q = queue.Queue()

    def _worker(self, cam_id):
        src = self.sources[cam_id]
        for seq in range(self.max_frames):
            self.q.put(src.read(seq))

    def run(self):
        threads = [threading.Thread(target=self._worker, args=(c,), daemon=True)
                   for c in self.cam_ids]
        for t in threads: t.start()

        # 按帧序号收集，凑齐所有通道后输出一组对齐帧
        buffer = {}
        aligned = []
        total = self.max_frames * len(self.cam_ids)
        for _ in range(total):
            f = self.q.get()
            buffer.setdefault(f.seq, {})[f.cam_id] = f
            if len(buffer[f.seq]) == len(self.cam_ids):
                group = buffer.pop(f.seq)
                base = min(g.ts for g in group.values())   # 以最早帧为基准
                aligned.append((f.seq, {cid: round(g.ts - base, 6)
                                        for cid, g in group.items()}))
        for t in threads: t.join()
        return aligned

if __name__ == "__main__":
    collector = SyncCollector(cam_ids=[0, 1, 2], fps=30.0, max_frames=5)
    result = collector.run()
    for seq, offsets in result:
        print(f"frame {seq}: " + ", ".join(f"cam{cid}={off:+.6f}s"
                                           for cid, off in sorted(offsets.items())))

运行与验证

运行：python sync_collector.py

预期输出（确定性推导，非实测）：共 5 行，每行形如 frame N: cam0=+0.000000s, cam1=+0.0000xx s, cam2=+0.0000xx s。判定通过条件：

• 输出恰好 5 组，帧序号 0~4 连续无缺失，说明三路帧全部对齐成功。
• 每组的 cam0 偏移为 0（基准），cam1/cam2 偏移为毫秒级小量，说明时间戳来自同一时钟源且同步误差在帧间隔内。
• 若某路缺失，该帧序号不会输出，可据此发现采集丢帧。

关键点

• 时间戳在 read() 内取，紧贴数据产出时刻，减少排队延迟污染。
• 对齐键用帧序号而非时间戳，避免浮点比较误差；时间戳只用于计算偏差。
• 真实接入时把 payload 换成图像数组，后续分析按 seq 索引即可保证多路时间一致。