多路视频信号整合技术：实现5路摄像头同步采集的完整方案

带宽 (bps) = 宽度 × 高度 × 像素深度 × 帧率 × 格式系数

1920 × 1080 × 10 × 30 × 1.2（格式系数，考虑数据包开销）≈ 746 Mbps

单摄像头带宽 = 分辨率 × 帧率 × 像素深度 × 格式系数
总带宽需求 = Σ(所有摄像头带宽)

总带宽 = 5 × 330 Mbps = 1650 Mbps ≈ 1.65 Gbps

单路带宽 = 1280 × 720 × 24位 × 30fps × 1.2（开销系数）
         = 1280 × 720 × 3字节 × 30 × 1.2
         = 99,532,800 字节/秒
         ≈ 95 MB/s
         ≈ 760 Mbps

总带宽 = 5 × 760 Mbps = 3800 Mbps = 3.8 Gbps

cam_module0: cam_module@0 {
    compatible = "arducam,arducam-pivariety";
    reg = <0x0c>;
    clocks = <&tegra_car TEGRA186_CLK_EXTPERIPH1>;
    clock-names = "clk";
    mclk = "extperiph1";
    reset-gpios = <&tegra_main_gpio CAM0_RST_L GPIO_ACTIVE_HIGH>;
    vana-supply = <&cam_module0_avdd>;
    status = "okay";

    ports {
        #address-cells = <1>;
        #size-cells = <0>;

        port@0 {
            reg = <0>;
            cam_module0_out: endpoint {
                port-index = <0>;
                bus-width = <4>;
                remote-endpoint = <&csi_in0>;
            };
        };
    };
};

gst-launch-1.0 nvarguscamerasrc sensor-id=0 ! \
    'video/x-raw(memory:NVMM),width=1280,height=720,framerate=30/1' ! \
    nvvidconv ! \
    'video/x-raw,format=I420' ! \
    nvoverlaysink

# 摄像头0
gst-launch-1.0 nvarguscamerasrc sensor-id=0 ! \
    'video/x-raw(memory:NVMM),width=1280,height=720,framerate=30/1' ! \
    nvvidconv ! videoconvert ! appsink name=sink0 &

# 摄像头1
gst-launch-1.0 nvarguscamerasrc sensor-id=1 ! \
    'video/x-raw(memory:NVMM),width=1280,height=720,framerate=30/1' ! \
    nvvidconv ! videoconvert ! appsink name=sink1 &

# ... 类似地配置摄像头2、3、4

import cv2
import time
import threading

class SynchronizedCamera:
    def __init__(self, camera_ids):
        self.camera_ids = camera_ids
        self.cameras = []
        self.frames = {}
        self.lock = threading.Lock()

    def initialize_cameras(self):
        for cam_id in self.camera_ids:
            cap = cv2.VideoCapture(cam_id)
            cap.set(cv2.CAP_PROP_FRAME_WIDTH, 1280)
            cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 720)
            cap.set(cv2.CAP_PROP_FPS, 30)
            self.cameras.append(cap)

    def capture_synchronized(self):
        # 触发硬件同步（如果支持）
        # 然后同时读取所有摄像头
        frames = {}
        for i, cam in enumerate(self.cameras):
            ret, frame = cam.read()
            if ret:
                frames[i] = (time.time(), frame)
        return frames

import cv2
import numpy as np
import threading
import time

class MultiCameraCapture:
    def __init__(self, camera_indices):
        self.camera_indices = camera_indices
        self.cameras = []
        self.frames = {}
        self.running = False
        self.lock = threading.Lock()

    def initialize_cameras(self):
        """初始化所有摄像头"""
        for idx in self.camera_indices:
            cap = cv2.VideoCapture(idx)
            if not cap.isOpened():
                raise RuntimeError(f"无法打开摄像头 {idx}")

            # 设置摄像头参数
            cap.set(cv2.CAP_PROP_FRAME_WIDTH, 1280)
            cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 720)
            cap.set(cv2.CAP_PROP_FPS, 30)
            cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)  # 减少缓冲区延迟

            self.cameras.append(cap)
            self.frames[idx] = None

    def capture_frame(self, camera_idx, cap):
        """从单个摄像头捕获帧"""
        while self.running:
            ret, frame = cap.read()
            if ret:
                with self.lock:
                    self.frames[camera_idx] = (time.time(), frame)
            time.sleep(0.001)  # 短暂休眠避免CPU占用过高

    def start_capture(self):
        """启动多线程采集"""
        self.running = True
        self.threads = []

        for i, cap in enumerate(self.cameras):
            thread = threading.Thread(
                target=self.capture_frame,
                args=(self.camera_indices[i], cap)
            )
            thread.daemon = True
            thread.start()
            self.threads.append(thread)

    def get_synchronized_frames(self):
        """获取同步的帧（基于时间戳）"""
        with self.lock:
            if len(self.frames) < len(self.camera_indices):
                return None

            # 获取所有帧的时间戳
            timestamps = {idx: self.frames[idx][0]
                         for idx in self.camera_indices
                         if self.frames[idx] is not None}

            if len(timestamps) < len(self.camera_indices):
                return None

            # 检查时间戳是否接近（同步窗口）
            max_diff = max(timestamps.values()) - min(timestamps.values())
            if max_diff > 0.033:  # 超过一帧的时间（30fps）
                return None

            # 返回所有帧
            return {idx: self.frames[idx][1]
                   for idx in self.camera_indices}

    def stop_capture(self):
        """停止采集"""
        self.running = False
        for thread in self.threads:
            thread.join()
        for cap in self.cameras:
            cap.release()

# 使用示例
if __name__ == "__main__":
    # 假设有5个USB摄像头，设备索引为0-4
    camera_indices = [0, 1, 2, 3, 4]

    multi_cam = MultiCameraCapture(camera_indices)
    multi_cam.initialize_cameras()
    multi_cam.start_capture()

    try:
        while True:
            frames = multi_cam.get_synchronized_frames()
            if frames:
                # 处理同步的帧
                for idx, frame in frames.items():
                    cv2.imshow(f'Camera {idx}', frame)

            if cv2.waitKey(1) & 0xFF == ord('q'):
                break
    finally:
        multi_cam.stop_capture()
        cv2.destroyAllWindows()

import v4l2
import fcntl
import mmap
import struct

class V4L2Camera:
    def __init__(self, device_path):
        self.device_path = device_path
        self.fd = None
        self.buffers = []

    def open_device(self):
        """打开V4L2设备"""
        self.fd = open(self.device_path, 'rb+', buffering=0)

        # 设置格式
        fmt = v4l2.v4l2_format()
        fmt.type = v4l2.V4L2_BUF_TYPE_VIDEO_CAPTURE
        fmt.fmt.pix.width = 1280
        fmt.fmt.pix.height = 720
        fmt.fmt.pix.pixelformat = v4l2.V4L2_PIX_FMT_YUYV
        fmt.fmt.pix.field = v4l2.V4L2_FIELD_INTERLACED

        fcntl.ioctl(self.fd, v4l2.VIDIOC_S_FMT, fmt)

        # 请求缓冲区
        req = v4l2.v4l2_requestbuffers()
        req.count = 4
        req.type = v4l2.V4L2_BUF_TYPE_VIDEO_CAPTURE
        req.memory = v4l2.V4L2_MEMORY_MMAP

        fcntl.ioctl(self.fd, v4l2.VIDIOC_REQBUFS, req)

        # 映射缓冲区
        for i in range(req.count):
            buf = v4l2.v4l2_buffer()
            buf.type = v4l2.V4L2_BUF_TYPE_VIDEO_CAPTURE
            buf.memory = v4l2.V4L2_MEMORY_MMAP
            buf.index = i

            fcntl.ioctl(self.fd, v4l2.VIDIOC_QUERYBUF, buf)

            mm = mmap.mmap(
                self.fd.fileno(),
                buf.length,
                mmap.MAP_SHARED,
                mmap.PROT_READ | mmap.PROT_WRITE,
                offset=buf.m.offset
            )

            self.buffers.append({
                'buffer': buf,
                'mmap': mm
            })

    def start_capture(self):
        """开始采集"""
        for buf_info in self.buffers:
            buf = buf_info['buffer']
            fcntl.ioctl(self.fd, v4l2.VIDIOC_QBUF, buf)

        buf_type = v4l2.v4l2_buf_type(v4l2.V4L2_BUF_TYPE_VIDEO_CAPTURE)
        fcntl.ioctl(self.fd, v4l2.VIDIOC_STREAMON, buf_type)

    def read_frame(self):
        """读取一帧"""
        buf = v4l2.v4l2_buffer()
        buf.type = v4l2.V4L2_BUF_TYPE_VIDEO_CAPTURE
        buf.memory = v4l2.V4L2_MEMORY_MMAP

        fcntl.ioctl(self.fd, v4l2.VIDIOC_DQBUF, buf)

        buf_info = self.buffers[buf.index]
        frame_data = buf_info['mmap'][:buf.bytesused]

        # 将帧重新入队
        fcntl.ioctl(self.fd, v4l2.VIDIOC_QBUF, buf)

        return frame_data, buf.timestamp

import time
from collections import deque

class TimestampSynchronizer:
    def __init__(self, sync_window=0.033):  # 一帧的时间（30fps）
        self.sync_window = sync_window
        self.frame_buffers = {}  # {camera_id: deque of (timestamp, frame)}
        self.max_buffer_size = 10

    def add_frame(self, camera_id, timestamp, frame):
        """添加一帧到缓冲区"""
        if camera_id not in self.frame_buffers:
            self.frame_buffers[camera_id] = deque(maxlen=self.max_buffer_size)

        self.frame_buffers[camera_id].append((timestamp, frame))

    def get_synchronized_frames(self):
        """获取同步的帧"""
        if len(self.frame_buffers) < 5:  # 需要5个摄像头
            return None

        # 获取每个摄像头的最新时间戳
        latest_timestamps = {}
        for cam_id, buffer in self.frame_buffers.items():
            if len(buffer) > 0:
                latest_timestamps[cam_id] = buffer[-1][0]

        if len(latest_timestamps) < 5:
            return None

        # 找到时间戳范围
        timestamps = list(latest_timestamps.values())
        min_ts = min(timestamps)
        max_ts = max(timestamps)

        # 检查是否在同步窗口内
        if max_ts - min_ts > self.sync_window:
            return None

        # 获取每个摄像头最接近目标时间戳的帧
        target_ts = (min_ts + max_ts) / 2
        synchronized_frames = {}

        for cam_id, buffer in self.frame_buffers.items():
            best_frame = None
            best_diff = float('inf')

            for ts, frame in buffer:
                diff = abs(ts - target_ts)
                if diff < best_diff:
                    best_diff = diff
                    best_frame = frame

            if best_frame is not None:
                synchronized_frames[cam_id] = best_frame

        return synchronized_frames if len(synchronized_frames) == 5 else None

# 伪代码示例，实际API取决于具体SDK
import pcie_capture_sdk

class PCIeMultiCamera:
    def __init__(self):
        self.sdk = pcie_capture_sdk.initialize()
        self.cameras = []

    def initialize_cameras(self, num_cameras=5):
        """初始化5个摄像头通道"""
        for i in range(num_cameras):
            camera = self.sdk.open_channel(i)
            camera.set_resolution(1920, 1080)
            camera.set_framerate(30)
            camera.enable_hardware_sync()  # 启用硬件同步
            self.cameras.append(camera)

    def start_capture(self):
        """开始采集"""
        # 硬件同步：所有通道共享时钟
        sync_signal = self.sdk.get_sync_signal()
        for camera in self.cameras:
            camera.set_sync_source(sync_signal)
            camera.start()

    def get_synchronized_frames(self):
        """获取同步的帧"""
        frames = {}
        for i, camera in enumerate(self.cameras):
            frame = camera.get_frame(timeout=1000)  # 1秒超时
            if frame:
                frames[i] = frame
        return frames if len(frames) == 5 else None

    def stop_capture(self):
        """停止采集"""
        for camera in self.cameras:
            camera.stop()
        self.sdk.cleanup()

class SoftwareSynchronizer:
    def __init__(self, sync_threshold=0.033):  # 一帧的时间
        self.sync_threshold = sync_threshold
        self.frame_queues = {}  # {camera_id: deque of (timestamp, frame)}

    def add_frame(self, camera_id, timestamp, frame):
        """添加帧到队列"""
        if camera_id not in self.frame_queues:
            self.frame_queues[camera_id] = deque(maxlen=10)
        self.frame_queues[camera_id].append((timestamp, frame))

    def get_synchronized_frames(self):
        """获取同步的帧"""
        if len(self.frame_queues) < 5:
            return None

        # 获取每个摄像头的最新时间戳
        latest_ts = {}
        for cam_id, queue in self.frame_queues.items():
            if len(queue) > 0:
                latest_ts[cam_id] = queue[-1][0]

        if len(latest_ts) < 5:
            return None

        # 检查时间戳差异
        timestamps = list(latest_ts.values())
        ts_diff = max(timestamps) - min(timestamps)

        if ts_diff > self.sync_threshold:
            return None

        # 返回对齐的帧
        target_ts = sum(timestamps) / len(timestamps)
        synchronized = {}

        for cam_id, queue in self.frame_queues.items():
            best_frame = None
            best_diff = float('inf')

            for ts, frame in queue:
                diff = abs(ts - target_ts)
                if diff < best_diff and diff <= self.sync_threshold:
                    best_diff = diff
                    best_frame = frame

            if best_frame:
                synchronized[cam_id] = best_frame

        return synchronized if len(synchronized) == 5 else None

class FrameBufferSynchronizer:
    def __init__(self, buffer_size=5, sync_window=0.033):
        self.buffer_size = buffer_size
        self.sync_window = sync_window
        self.buffers = {}  # {camera_id: list of (timestamp, frame)}

    def add_frame(self, camera_id, timestamp, frame):
        """添加帧到缓冲区"""
        if camera_id not in self.buffers:
            self.buffers[camera_id] = []

        self.buffers[camera_id].append((timestamp, frame))

        # 保持缓冲区大小
        if len(self.buffers[camera_id]) > self.buffer_size:
            self.buffers[camera_id].pop(0)

    def find_synchronized_set(self):
        """查找同步的帧集"""
        if len(self.buffers) < 5:
            return None

        # 为每个摄像头选择一帧，使得时间戳差异最小
        best_set = None
        best_max_diff = float('inf')

        # 遍历所有可能的组合（简化版本，实际可以使用更高效的算法）
        for frames_0 in self.buffers[0]:
            for frames_1 in self.buffers[1]:
                for frames_2 in self.buffers[2]:
                    for frames_3 in self.buffers[3]:
                        for frames_4 in self.buffers[4]:
                            timestamps = [
                                frames_0[0], frames_1[0], frames_2[0],
                                frames_3[0], frames_4[0]
                            ]
                            max_diff = max(timestamps) - min(timestamps)

                            if max_diff < self.sync_window and max_diff < best_max_diff:
                                best_max_diff = max_diff
                                best_set = {
                                    0: frames_0[1],
                                    1: frames_1[1],
                                    2: frames_2[1],
                                    3: frames_3[1],
                                    4: frames_4[1]
                                }

        return best_set