From 184d8288fb17e93a76b544791fb1503df70883f3 Mon Sep 17 00:00:00 2001 From: "stas.bucik" Date: Fri, 15 May 2026 09:31:52 +0200 Subject: [PATCH 1/4] Fix bug in example, to allow multiple cameras on PTP master device Signed-off-by: stas.bucik --- .../cpp/Misc/MultiDevice/multi_device_frame_sync.cpp | 9 ++++++--- .../python/Misc/MultiDevice/multi_device_frame_sync.py | 6 ++++-- tests/include/fsync_ptp_test_utils.hpp | 3 ++- tests/src/onhost_tests/utility/fsync_ptp_test_utils.cpp | 9 ++++++--- 4 files changed, 18 insertions(+), 9 deletions(-) diff --git a/examples/cpp/Misc/MultiDevice/multi_device_frame_sync.cpp b/examples/cpp/Misc/MultiDevice/multi_device_frame_sync.cpp index 9102dc77ba..3077666e13 100644 --- a/examples/cpp/Misc/MultiDevice/multi_device_frame_sync.cpp +++ b/examples/cpp/Misc/MultiDevice/multi_device_frame_sync.cpp @@ -136,7 +136,8 @@ void setUpCameraSocket(std::shared_ptr& pipeline, std::optional role, std::optional>& masterNode, std::map>>& slaveQueues, - std::vector& camSockets) { + std::vector& camSockets, + std::optional& masterName) { auto outNode = createCameraOutputs(pipeline, socket, targetFps, syncType, role); if(syncType == SyncType::EXTERNAL) { @@ -162,6 +163,9 @@ void setUpCameraSocket(std::shared_ptr& pipeline, // Actual PTP master might be different, but it doesn't matter for this example. if(!masterNode.has_value()) { masterNode.emplace(); + masterName = name; + } + if (masterName == name) { masterNode.value().emplace(dai::toString(socket), outNode); } else { if(slaveQueues.find(name) == slaveQueues.end()) { @@ -214,7 +218,7 @@ void setupDevice(dai::DeviceInfo& deviceInfo, std::cout << " Num of cameras: " << device->getConnectedCameras().size() << std::endl; for(auto socket : device->getConnectedCameras()) { - setUpCameraSocket(pipeline, socket, name, targetFps, syncType, role, masterNode, slaveQueues, camSockets); + setUpCameraSocket(pipeline, socket, name, targetFps, syncType, role, masterNode, slaveQueues, camSockets, masterName); } if(syncType == SyncType::EXTERNAL) { @@ -238,7 +242,6 @@ void setupDevice(dai::DeviceInfo& deviceInfo, // Actual PTP master might be different, but it doesn't matter for this example. if(masterPipeline == nullptr) { masterPipeline = pipeline; - masterName = name; } else { slavePipelines[name] = pipeline; } diff --git a/examples/python/Misc/MultiDevice/multi_device_frame_sync.py b/examples/python/Misc/MultiDevice/multi_device_frame_sync.py index 7178b7ae54..728171e0cf 100644 --- a/examples/python/Misc/MultiDevice/multi_device_frame_sync.py +++ b/examples/python/Misc/MultiDevice/multi_device_frame_sync.py @@ -108,7 +108,7 @@ def setUpCameraSocket( deviceName: str, targetFps: float, role: dai.ExternalFrameSyncRole): - global masterNode, slaveQueues, camSockets, syncType + global masterNode, slaveQueues, camSockets, syncType, masterName pipeline, outNode = createCameraOutputs(pipeline, socket, targetFps, role) if syncType == SyncType.EXTERNAL: @@ -131,6 +131,9 @@ def setUpCameraSocket( # Actual PTP master might be different, but it doesn't matter for this example if masterNode is None: masterNode = {} + masterName = deviceName + + if masterName == deviceName: masterNode[socket.name] = outNode else: if slaveQueues.get(deviceName) is None: @@ -196,7 +199,6 @@ def setupDevice( # Actual PTP master might be different, but it doesn't matter for this example if masterPipeline is None: masterPipeline = pipeline - masterName = name else: slavePipelines[name] = pipeline diff --git a/tests/include/fsync_ptp_test_utils.hpp b/tests/include/fsync_ptp_test_utils.hpp index 2d59a3bf52..d58db353c7 100644 --- a/tests/include/fsync_ptp_test_utils.hpp +++ b/tests/include/fsync_ptp_test_utils.hpp @@ -56,7 +56,8 @@ void setUpCameraSocket(std::shared_ptr& pipeline, std::optional role, std::optional>& masterNode, std::map>>& slaveQueues, - std::vector& camSockets); + std::vector& camSockets, + std::optional& masterName); void setUpIrLeds(std::shared_ptr device); diff --git a/tests/src/onhost_tests/utility/fsync_ptp_test_utils.cpp b/tests/src/onhost_tests/utility/fsync_ptp_test_utils.cpp index 2ac951a958..104d631ac0 100644 --- a/tests/src/onhost_tests/utility/fsync_ptp_test_utils.cpp +++ b/tests/src/onhost_tests/utility/fsync_ptp_test_utils.cpp @@ -183,7 +183,8 @@ void setUpCameraSocket(std::shared_ptr& pipeline, std::optional role, std::optional>& masterNode, std::map>>& slaveQueues, - std::vector& camSockets) { + std::vector& camSockets, + std::optional& masterName) { auto outNode = createPipeline(pipeline, socket, targetFps, syncType, role); if(syncType == SyncType::EXTERNAL) { @@ -207,6 +208,9 @@ void setUpCameraSocket(std::shared_ptr& pipeline, // Actual PTP master might be different, but it doesn't matter for this test. if(!masterNode.has_value()) { masterNode.emplace(); + masterName = name; + } + if (masterName == name) { masterNode.value().emplace(dai::toString(socket), outNode); } else { if(slaveQueues.find(name) == slaveQueues.end()) { @@ -274,7 +278,7 @@ void setupDevice(dai::DeviceInfo& deviceInfo, std::cout << " Num of cameras: " << device->getConnectedCameras().size() << std::endl; for(auto socket : device->getConnectedCameras()) { - setUpCameraSocket(pipeline, socket, name, targetFps, syncType, role, masterNode, slaveQueues, camSockets); + setUpCameraSocket(pipeline, socket, name, targetFps, syncType, role, masterNode, slaveQueues, camSockets, masterName); } setUpIrLeds(device); @@ -300,7 +304,6 @@ void setupDevice(dai::DeviceInfo& deviceInfo, // Actual PTP master might be different, but it doesn't matter for this test. if(masterPipeline == nullptr) { masterPipeline = pipeline; - masterName = name; } else { slavePipelines[name] = pipeline; } From 03a58bcc558b2ff605ba16bfe93783f2cf0d5754 Mon Sep 17 00:00:00 2001 From: "stas.bucik" Date: Fri, 15 May 2026 10:31:02 +0200 Subject: [PATCH 2/4] Add minimal examples Signed-off-by: stas.bucik --- .../external_sync_frame_sync_minimal.py | 179 ++++++++++++++++++ .../MultiDevice/ptp_frame_sync_minimal.py | 176 +++++++++++++++++ 2 files changed, 355 insertions(+) create mode 100644 examples/python/Misc/MultiDevice/external_sync_frame_sync_minimal.py create mode 100644 examples/python/Misc/MultiDevice/ptp_frame_sync_minimal.py diff --git a/examples/python/Misc/MultiDevice/external_sync_frame_sync_minimal.py b/examples/python/Misc/MultiDevice/external_sync_frame_sync_minimal.py new file mode 100644 index 0000000000..ae275c8988 --- /dev/null +++ b/examples/python/Misc/MultiDevice/external_sync_frame_sync_minimal.py @@ -0,0 +1,179 @@ +import depthai as dai +import contextlib +from typing import Optional, Dict +from datetime import timedelta, datetime +import signal +import time +import threading +import cv2 + +# This example only works on devices that are connected with M8 cables with FSYNC Y splitters + +deviceInfos = dai.Device.getAllAvailableDevices() +targetFps = 30 +resolution = (640, 480) +syncThresholdSec = 1e-3 # 1ms +running = True + +def interruptHandler(sig, frame): + global running + if running: + print("Interrupted! Exiting...") + running = False + else: + print("Exiting now!") + exit(0) + +signal.signal(signal.SIGINT, interruptHandler) + +def getDeviceName(device : dai.Device) -> str: + info = device.getDeviceInfo() + name = info.deviceId + if info.name is not None and info.name != "": + name += "[" + info.name + "]" + return name + +with contextlib.ExitStack() as stack: + # Variables to keep track of master and slave pipelines and outputs + masterPipeline: Optional[dai.Pipeline] = None + masterNode: Optional[Dict[str, dai.Node.Output]] = None + masterName: Optional[str] = None + + slavePipelines: Dict[str, dai.Pipeline] = {} + slaveQueues: Dict[str, Dict[str, dai.MessageQueue]] = {} + + # keep track of sync node inputs for slaves + inputQueues = {} + + # keep track of all sync node output names + outputNames = [] + + for deviceInfo in deviceInfos: + # Create pipeline for each device + devicePipeline = stack.enter_context(dai.Pipeline(dai.Device(deviceInfo))) + device = devicePipeline.getDefaultDevice() + deviceName = getDeviceName(device) + fsyncRole = device.getExternalFrameSyncRole() + + for socket in device.getConnectedCameras(): + # create a queue for each camera on the device + if fsyncRole == dai.ExternalFrameSyncRole.MASTER: + cam = devicePipeline.create(dai.node.Camera).build(socket, sensorFps=targetFps) + else: + # slaves will lock to the master's FPS + cam = devicePipeline.create(dai.node.Camera).build(socket) + outputNode = cam.requestOutput(resolution, dai.ImgFrame.Type.NV12, dai.ImgResizeMode.CROP) + + # Master cameras will be linked to the sync node directly + if fsyncRole == dai.ExternalFrameSyncRole.MASTER: + if masterNode is None: + masterNode = {} + + masterNode[socket.name] = outputNode + + # Gather all slave camera outputs + elif fsyncRole == dai.ExternalFrameSyncRole.SLAVE: + if slaveQueues.get(deviceName) is None: + slaveQueues[deviceName] = {} + slaveQueues[deviceName][socket.name] = outputNode.createOutputQueue() + + if fsyncRole == dai.ExternalFrameSyncRole.MASTER: + device.setExternalStrobeEnable(True) + print(f"{device.getDeviceId()} is master") + + if masterPipeline is not None: + raise RuntimeError("Only one master pipeline is supported") + + masterPipeline = devicePipeline + masterName = deviceName + elif fsyncRole == dai.ExternalFrameSyncRole.SLAVE: + slavePipelines[deviceName] = devicePipeline + print(f"{device.getDeviceId()} is slave") + + if masterPipeline is None or masterNode is None: + raise RuntimeError("No master detected!") + + if len(slavePipelines) < 1: + raise RuntimeError("No slaves detected!") + + # Create sync node + syncNode = masterPipeline.create(dai.node.Sync) + + # Sync node will run on the host, since it needs to sync multiple devices + syncNode.setRunOnHost(True) + # group frames into pairs that are within 1/2 frame period + syncNode.setSyncThreshold(timedelta(milliseconds=1000 / (2 * targetFps))) + + # Link master camera outputs to the sync node + for socketName, camOutput in masterNode.items(): + name = f"master_{masterName}_{socketName}" + camOutput.link(syncNode.inputs[name]) + outputNames.append(name) + + # For slaves, we must create an input queue for each output + # We will then manually forward the frames from each input queue to the output queue + # This is because slave devices have separate pipelines from the master + for deviceName, sockets in slaveQueues.items(): + for socketName, _ in sockets.items(): + name = f"slave_{deviceName}_{socketName}" + outputNames.append(name) + input_queue = syncNode.inputs[name].createInputQueue() + inputQueues[name] = input_queue + + syncedGroups = syncNode.out.createOutputQueue() + + # thread worker for forwarding slave queues to sync node + def data_collector(deviceName, socketName): + # Send frames from slave output queues to sync node input queues + camOutputQueue = slaveQueues[deviceName][socketName] + while running: + if camOutputQueue.has(): + inputQueues[f"slave_{deviceName}_{socketName}"].send(camOutputQueue.get()) + else: + time.sleep(0.001) + + # Start pipelines + masterPipeline.start() + for _, slavePipeline in slavePipelines.items(): + slavePipeline.start() + + # Start threads + threads = {} + for deviceName, sockets in slaveQueues.items(): + for socketName, camOutputQueue in sockets.items(): + threads[f"slave_{deviceName}_{socketName}"] = threading.Thread(target=data_collector, args=(deviceName, socketName)) + threads[f"slave_{deviceName}_{socketName}"].start() + + # main display loop + latestFrameGroup = None + while running: + # Get frames from sync node output queue + while syncedGroups.has(): + latestFrameGroup = syncedGroups.get() + + if latestFrameGroup is not None and latestFrameGroup.getNumMessages() == len(outputNames): + tsValues = {} + for name in outputNames: + tsValues[name] = latestFrameGroup[name].getTimestamp(dai.CameraExposureOffset.END).total_seconds() + + delta = max(tsValues.values()) - min(tsValues.values()) + syncStatus = abs(delta) < syncThresholdSec + + if not syncStatus: + print(f"Sync error: Sync lost, threshold exceeded {delta * 1e6} us") + continue + + for outputName in outputNames: + msg = latestFrameGroup[outputName] + frame = msg.getCvFrame() + cv2.imshow(f"synced_view_{outputName}", frame) + + latestFrameGroup = None # Wait for next batch + + if cv2.waitKey(1) & 0xFF == ord("q"): + running = False + break + + for t in threads.keys(): + threads[t].join() + cv2.destroyAllWindows() \ No newline at end of file diff --git a/examples/python/Misc/MultiDevice/ptp_frame_sync_minimal.py b/examples/python/Misc/MultiDevice/ptp_frame_sync_minimal.py new file mode 100644 index 0000000000..e10f854759 --- /dev/null +++ b/examples/python/Misc/MultiDevice/ptp_frame_sync_minimal.py @@ -0,0 +1,176 @@ +import depthai as dai +import contextlib +from typing import Optional, Dict +from datetime import timedelta, datetime +import signal +import time +import threading +import cv2 + +# This example only works on devices that have PTP enabled + +deviceInfos = dai.Device.getAllAvailableDevices() +targetFps = 30 +resolution = (640, 480) +syncThresholdSec = 1e-3 # 1ms +running = True + +def interruptHandler(sig, frame): + global running + if running: + print("Interrupted! Exiting...") + running = False + else: + print("Exiting now!") + exit(0) + +signal.signal(signal.SIGINT, interruptHandler) + +def getDeviceName(device : dai.Device) -> str: + info = device.getDeviceInfo() + name = info.deviceId + if info.name is not None and info.name != "": + name += "[" + info.name + "]" + return name + +with contextlib.ExitStack() as stack: + # Variables to keep track of master and slave pipelines and outputs + masterPipeline: Optional[dai.Pipeline] = None + masterNode: Optional[Dict[str, dai.Node.Output]] = None + masterName: Optional[str] = None + + slavePipelines: Dict[str, dai.Pipeline] = {} + slaveQueues: Dict[str, Dict[str, dai.MessageQueue]] = {} + + # keep track of sync node inputs for slaves + inputQueues = {} + + # keep track of all sync node output names + outputNames = [] + + for deviceInfo in deviceInfos: + # Create pipeline for each device + devicePipeline = stack.enter_context(dai.Pipeline(dai.Device(deviceInfo))) + device = devicePipeline.getDefaultDevice() + deviceName = getDeviceName(device) + + for socket in device.getConnectedCameras(): + ######################################################################## + # TODO: remove this when OV9282 is supporter for PTP + sensorName = "" + for sckt, sName in device.getCameraSensorNames().items(): + if sckt == socket: + sensorName = sName + break + if sensorName == "": + raise RuntimeError(f"No sensor name found for {socket.name} on {deviceName}") + if sensorName == "OV9282": + continue + ######################################################################## + + # create a queue for each camera on the device + cam = devicePipeline.create(dai.node.Camera).build(socket, sensorFps=targetFps) + outputNode = cam.requestOutput(resolution, dai.ImgFrame.Type.NV12, dai.ImgResizeMode.CROP) + + # Set sync mode to PTP + cam.initialControl.setFrameSyncMode(dai.CameraControl.FrameSyncMode.TIME_PTP) + + # Put the first camera in master + # Actual PTP master might be different, but it doesn't matter for this example + if masterNode is None: + masterNode = {} + masterName = deviceName + + if masterName == deviceName: + masterNode[socket.name] = outputNode + else: + if slaveQueues.get(deviceName) is None: + slaveQueues[deviceName] = {} + slaveQueues[deviceName][socket.name] = outputNode.createOutputQueue() + + if masterPipeline is None: + masterPipeline = devicePipeline + else: + slavePipelines[deviceName] = devicePipeline + + # Create sync node + syncNode = masterPipeline.create(dai.node.Sync) + + # Sync node will run on the host, since it needs to sync multiple devices + syncNode.setRunOnHost(True) + # group frames into pairs that are within 1/2 frame period + syncNode.setSyncThreshold(timedelta(milliseconds=1000 / (2 * targetFps))) + + # Link master camera outputs to the sync node + for socketName, camOutput in masterNode.items(): + name = f"master_{masterName}_{socketName}" + camOutput.link(syncNode.inputs[name]) + outputNames.append(name) + + # For slaves, we must create an input queue for each output + # We will then manually forward the frames from each input queue to the output queue + # This is because slave devices have separate pipelines from the master + for deviceName, sockets in slaveQueues.items(): + for socketName, _ in sockets.items(): + name = f"slave_{deviceName}_{socketName}" + outputNames.append(name) + input_queue = syncNode.inputs[name].createInputQueue() + inputQueues[name] = input_queue + + syncedGroups = syncNode.out.createOutputQueue() + + # thread worker for forwarding slave queues to sync node + def data_collector(deviceName, socketName): + # Send frames from slave output queues to sync node input queues + camOutputQueue = slaveQueues[deviceName][socketName] + while running: + if camOutputQueue.has(): + inputQueues[f"slave_{deviceName}_{socketName}"].send(camOutputQueue.get()) + else: + time.sleep(0.001) + + # Start pipelines + masterPipeline.start() + for _, slavePipeline in slavePipelines.items(): + slavePipeline.start() + + # Start threads + threads = {} + for deviceName, sockets in slaveQueues.items(): + for socketName, camOutputQueue in sockets.items(): + threads[f"slave_{deviceName}_{socketName}"] = threading.Thread(target=data_collector, args=(deviceName, socketName)) + threads[f"slave_{deviceName}_{socketName}"].start() + + # main display loop + latestFrameGroup = None + while running: + # Get frames from sync node output queue + while syncedGroups.has(): + latestFrameGroup = syncedGroups.get() + + if latestFrameGroup is not None and latestFrameGroup.getNumMessages() == len(outputNames): + tsValues = {} + for name in outputNames: + tsValues[name] = latestFrameGroup[name].getTimestamp(dai.CameraExposureOffset.END).total_seconds() + + delta = max(tsValues.values()) - min(tsValues.values()) + syncStatus = abs(delta) < syncThresholdSec + + if not syncStatus: + print(f"Sync error: Sync lost, threshold exceeded {delta * 1e6} us") + continue + + for outputName in outputNames: + msg = latestFrameGroup[outputName] + frame = msg.getCvFrame() + cv2.imshow(f"synced_view_{outputName}", frame) + + latestFrameGroup = None # Wait for next batch + + if cv2.waitKey(1) & 0xFF == ord("q"): + running = False + break + + for t in threads.keys(): + threads[t].join() + cv2.destroyAllWindows() \ No newline at end of file From 14e36180a780387e1efb8144daead0a9be4ad1b3 Mon Sep 17 00:00:00 2001 From: "stas.bucik" Date: Fri, 15 May 2026 11:11:05 +0200 Subject: [PATCH 3/4] Add C++ minimal examples Signed-off-by: stas.bucik --- examples/cpp/Misc/MultiDevice/CMakeLists.txt | 4 +- .../external_sync_frame_sync_minimal.cpp | 233 ++++++++++++++++++ .../MultiDevice/ptp_frame_sync_minimal.cpp | 228 +++++++++++++++++ .../MultiDevice/ptp_frame_sync_minimal.py | 20 +- 4 files changed, 474 insertions(+), 11 deletions(-) create mode 100644 examples/cpp/Misc/MultiDevice/external_sync_frame_sync_minimal.cpp create mode 100644 examples/cpp/Misc/MultiDevice/ptp_frame_sync_minimal.cpp diff --git a/examples/cpp/Misc/MultiDevice/CMakeLists.txt b/examples/cpp/Misc/MultiDevice/CMakeLists.txt index 68110f894f..b15a5d1dfe 100644 --- a/examples/cpp/Misc/MultiDevice/CMakeLists.txt +++ b/examples/cpp/Misc/MultiDevice/CMakeLists.txt @@ -4,4 +4,6 @@ cmake_minimum_required(VERSION 3.10) ## function: dai_add_example(example_name example_src enable_test use_pcl) ## function: dai_set_example_test_labels(example_name ...) -dai_add_example(multi_device_frame_sync multi_device_frame_sync.cpp OFF OFF) \ No newline at end of file +dai_add_example(multi_device_frame_sync multi_device_frame_sync.cpp OFF OFF) +dai_add_example(ptp_frame_sync_minimal ptp_frame_sync_minimal.cpp OFF OFF) +dai_add_example(external_sync_frame_sync_minimal external_sync_frame_sync_minimal.cpp OFF OFF) diff --git a/examples/cpp/Misc/MultiDevice/external_sync_frame_sync_minimal.cpp b/examples/cpp/Misc/MultiDevice/external_sync_frame_sync_minimal.cpp new file mode 100644 index 0000000000..76dc2f4a82 --- /dev/null +++ b/examples/cpp/Misc/MultiDevice/external_sync_frame_sync_minimal.cpp @@ -0,0 +1,233 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "depthai/depthai.hpp" + +namespace { + +constexpr float TARGET_FPS = 30.0f; +constexpr std::pair RESOLUTION{640, 480}; +constexpr double SYNC_THRESHOLD_SEC = 1e-3; + +std::atomic_bool running{true}; + +void interruptHandler(int) { + if(running.exchange(false)) { + std::cout << "Interrupted! Exiting..." << std::endl; + } else { + std::cout << "Exiting now!" << std::endl; + std::exit(0); + } +} + +std::string getDeviceName(const std::shared_ptr& device) { + const auto info = device->getDeviceInfo(); + auto name = info.deviceId; + if(!info.name.empty()) { + name += "[" + info.name + "]"; + } + return name; +} + +} // namespace + +int main() { + signal(SIGINT, interruptHandler); + + // This example only works on devices that are connected with M8 cables with FSYNC Y splitters. + const auto deviceInfos = dai::Device::getAllAvailableDevices(); + + // Variables to keep track of master and slave pipelines and outputs + std::shared_ptr masterPipeline; + std::optional> masterOutputs; + std::optional masterName; + + std::map> slavePipelines; + std::map>> slaveQueues; + + // keep track of sync node inputs for slaves + std::map> inputQueues; + // keep track of all sync node output names + std::vector outputNames; + + for(const auto& deviceInfo : deviceInfos) { + // Create pipeline for each device + auto pipeline = std::make_shared(std::make_shared(deviceInfo)); + auto device = pipeline->getDefaultDevice(); + const auto deviceName = getDeviceName(device); + const auto fsyncRole = device->getExternalFrameSyncRole(); + + for(const auto socket : device->getConnectedCameras()) { + // create a queue for each camera on the device + std::shared_ptr cam; + if(fsyncRole == dai::ExternalFrameSyncRole::MASTER) { + cam = pipeline->create()->build(socket, std::nullopt, TARGET_FPS); + } else { + // slaves will lock to the master's FPS + cam = pipeline->create()->build(socket); + } + + auto* output = cam->requestOutput(RESOLUTION, dai::ImgFrame::Type::NV12, dai::ImgResizeMode::CROP); + const auto socketName = dai::toString(socket); + + // Master cameras will be linked to the sync node directly + if(fsyncRole == dai::ExternalFrameSyncRole::MASTER) { + if(!masterOutputs.has_value()) { + masterOutputs.emplace(); + } + (*masterOutputs)[socketName] = output; + // Gather all slave camera outputs + } else if(fsyncRole == dai::ExternalFrameSyncRole::SLAVE) { + slaveQueues[deviceName][socketName] = output->createOutputQueue(); + } + } + + if(fsyncRole == dai::ExternalFrameSyncRole::MASTER) { + device->setExternalStrobeEnable(true); + std::cout << device->getDeviceId() << " is master" << std::endl; + + if(masterPipeline != nullptr) { + throw std::runtime_error("Only one master pipeline is supported"); + } + + masterPipeline = pipeline; + masterName = deviceName; + } else if(fsyncRole == dai::ExternalFrameSyncRole::SLAVE) { + slavePipelines[deviceName] = pipeline; + std::cout << device->getDeviceId() << " is slave" << std::endl; + } + } + + if(masterPipeline == nullptr || !masterOutputs.has_value() || !masterName.has_value()) { + throw std::runtime_error("No master detected!"); + } + + if(slavePipelines.empty()) { + throw std::runtime_error("No slaves detected!"); + } + + // Create sync node + auto sync = masterPipeline->create(); + // Sync node will run on the host, since it needs to sync multiple devices + sync->setRunOnHost(true); + // group frames into pairs that are within 1/2 frame period + sync->setSyncThreshold(std::chrono::duration_cast(std::chrono::duration(0.5 / TARGET_FPS))); + + // Link master camera outputs to the sync node + for(const auto& [socketName, output] : *masterOutputs) { + const auto name = "master_" + *masterName + "_" + socketName; + output->link(sync->inputs[name]); + outputNames.push_back(name); + } + + // For slaves, we must create an input queue for each output + // We will then manually forward the frames from each input queue to the output queue + // This is because slave devices have separate pipelines from the master + for(const auto& [deviceName, sockets] : slaveQueues) { + for(const auto& [socketName, queue] : sockets) { + (void)queue; + const auto name = "slave_" + deviceName + "_" + socketName; + outputNames.push_back(name); + inputQueues[name] = sync->inputs[name].createInputQueue(); + } + } + + auto syncedGroups = sync->out.createOutputQueue(); + + // thread worker for forwarding slave queues to sync node + auto dataCollector = [&](std::string deviceName, std::string socketName) { + const auto queueName = "slave_" + deviceName + "_" + socketName; + auto camOutputQueue = slaveQueues.at(deviceName).at(socketName); + auto inputQueue = inputQueues.at(queueName); + + // Send frames from slave output queues to sync node input queues + while(running.load()) { + if(camOutputQueue->has()) { + inputQueue->send(camOutputQueue->get()); + } else { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + } + }; + + // Start pipelines + masterPipeline->start(); + for(const auto& [deviceName, pipeline] : slavePipelines) { + (void)deviceName; + pipeline->start(); + } + + // Start threads + std::vector threads; + for(const auto& [deviceName, sockets] : slaveQueues) { + for(const auto& [socketName, queue] : sockets) { + (void)queue; + threads.emplace_back(dataCollector, deviceName, socketName); + } + } + + std::optional> latestFrameGroup; + + // main display loop + while(running.load()) { + // Get frames from sync node output queue + while(syncedGroups->has()) { + latestFrameGroup = syncedGroups->get(); + } + + if(latestFrameGroup.has_value() && static_cast(latestFrameGroup.value()->getNumMessages()) == outputNames.size()) { + using ts_type = std::chrono::time_point; + std::map tsValues; + for(auto name : outputNames) { + auto frame = latestFrameGroup.value()->get(name); + tsValues.emplace(name, frame->getTimestamp(dai::CameraExposureOffset::END)); + } + auto compFunct = [](const std::pair& p1, const std::pair& p2) -> bool { return p1.second < p2.second; }; + + auto maxElement = std::max_element(tsValues.begin(), tsValues.end(), compFunct); + auto minElement = std::min_element(tsValues.begin(), tsValues.end(), compFunct); + + auto delta = maxElement->second - minElement->second; + auto deltaUs = std::chrono::duration_cast(delta).count(); + + if(deltaUs >= SYNC_THRESHOLD_SEC * 1e6) { + std::cout << "Sync error: Sync lost, threshold exceeded " << deltaUs << " us" << std::endl; + continue; + } + + for(const auto& outputName : outputNames) { + auto frame = latestFrameGroup.value()->get(outputName); + cv::imshow("synced_view_" + outputName, frame->getCvFrame()); + } + + latestFrameGroup.reset(); // Wait for next batch + } + + if((cv::waitKey(1) & 0xFF) == 'q') { + running.store(false); + break; + } + } + + for(auto& thread : threads) { + thread.join(); + } + + cv::destroyAllWindows(); + return 0; +} diff --git a/examples/cpp/Misc/MultiDevice/ptp_frame_sync_minimal.cpp b/examples/cpp/Misc/MultiDevice/ptp_frame_sync_minimal.cpp new file mode 100644 index 0000000000..67de8bb989 --- /dev/null +++ b/examples/cpp/Misc/MultiDevice/ptp_frame_sync_minimal.cpp @@ -0,0 +1,228 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "depthai/depthai.hpp" + +namespace { + +constexpr float TARGET_FPS = 30.0f; +constexpr std::pair RESOLUTION{640, 480}; +constexpr double SYNC_THRESHOLD_SEC = 1e-3; + +std::atomic_bool running{true}; + +void interruptHandler(int) { + if(running.exchange(false)) { + std::cout << "Interrupted! Exiting..." << std::endl; + } else { + std::cout << "Exiting now!" << std::endl; + std::exit(0); + } +} + +std::string getDeviceName(const std::shared_ptr& device) { + const auto info = device->getDeviceInfo(); + auto name = info.deviceId; + if(!info.name.empty()) { + name += "[" + info.name + "]"; + } + return name; +} + +std::string getSensorName(const std::shared_ptr& device, dai::CameraBoardSocket socket) { + const auto sensorNames = device->getCameraSensorNames(); + const auto it = sensorNames.find(socket); + if(it == sensorNames.end()) { + throw std::runtime_error("No sensor name found for " + dai::toString(socket) + " on " + getDeviceName(device)); + } + return it->second; +} + +} // namespace + +int main() { + signal(SIGINT, interruptHandler); + + // This example only works on devices that have PTP enabled. + const auto deviceInfos = dai::Device::getAllAvailableDevices(); + + // Variables to keep track of master and slave pipelines and outputs + std::shared_ptr masterPipeline; + std::optional> masterOutputs; + std::optional masterName; + + std::map> slavePipelines; + std::map>> slaveQueues; + + // keep track of sync node inputs for slaves + std::map> inputQueues; + // keep track of all sync node output names + std::vector outputNames; + + for(const auto& deviceInfo : deviceInfos) { + // Create pipeline for each device + auto pipeline = std::make_shared(std::make_shared(deviceInfo)); + auto device = pipeline->getDefaultDevice(); + const auto deviceName = getDeviceName(device); + + for(const auto socket : device->getConnectedCameras()) { + // TODO: remove this when OV9282 is supported for PTP. + if(getSensorName(device, socket) == "OV9282") { + continue; + } + + // create a queue for each camera on the device + auto cam = pipeline->create()->build(socket, std::nullopt, TARGET_FPS); + auto* output = cam->requestOutput(RESOLUTION, dai::ImgFrame::Type::NV12, dai::ImgResizeMode::CROP); + const auto socketName = dai::toString(socket); + + // Set sync mode to PTP + cam->initialControl.setFrameSyncMode(dai::CameraControl::FrameSyncMode::TIME_PTP); + + // Put the first camera in master + // Actual PTP master might be different, but it doesn't matter for this example + if(!masterOutputs.has_value()) { + masterOutputs.emplace(); + masterName = deviceName; + } + + if(*masterName == deviceName) { + (*masterOutputs)[socketName] = output; + } else { + slaveQueues[deviceName][socketName] = output->createOutputQueue(); + } + } + + if(masterPipeline == nullptr) { + masterPipeline = pipeline; + } else { + slavePipelines[deviceName] = pipeline; + } + } + + // Create sync node + auto sync = masterPipeline->create(); + // Sync node will run on the host, since it needs to sync multiple devices + sync->setRunOnHost(true); + // group frames into pairs that are within 1/2 frame period + sync->setSyncThreshold(std::chrono::duration_cast(std::chrono::duration(0.5 / TARGET_FPS))); + + // Link master camera outputs to the sync node + for(const auto& [socketName, output] : *masterOutputs) { + const auto name = "master_" + *masterName + "_" + socketName; + output->link(sync->inputs[name]); + outputNames.push_back(name); + } + + // For slaves, we must create an input queue for each output + // We will then manually forward the frames from each input queue to the output queue + // This is because slave devices have separate pipelines from the master + for(const auto& [deviceName, sockets] : slaveQueues) { + for(const auto& [socketName, queue] : sockets) { + (void)queue; + const auto name = "slave_" + deviceName + "_" + socketName; + outputNames.push_back(name); + inputQueues[name] = sync->inputs[name].createInputQueue(); + } + } + + auto syncedGroups = sync->out.createOutputQueue(); + + // thread worker for forwarding slave queues to sync node + auto dataCollector = [&](std::string deviceName, std::string socketName) { + const auto queueName = "slave_" + deviceName + "_" + socketName; + auto camOutputQueue = slaveQueues.at(deviceName).at(socketName); + auto inputQueue = inputQueues.at(queueName); + + // Send frames from slave output queues to sync node input queues + while(running.load()) { + if(camOutputQueue->has()) { + inputQueue->send(camOutputQueue->get()); + } else { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + } + }; + + // Start pipelines + masterPipeline->start(); + for(const auto& [deviceName, pipeline] : slavePipelines) { + (void)deviceName; + pipeline->start(); + } + + // Start threads + std::vector threads; + for(const auto& [deviceName, sockets] : slaveQueues) { + for(const auto& [socketName, queue] : sockets) { + (void)queue; + threads.emplace_back(dataCollector, deviceName, socketName); + } + } + + std::optional> latestFrameGroup; + + // main display loop + while(running.load()) { + // Get frames from sync node output queue + while(syncedGroups->has()) { + latestFrameGroup = syncedGroups->get(); + } + + if(latestFrameGroup.has_value() && static_cast(latestFrameGroup.value()->getNumMessages()) == outputNames.size()) { + using ts_type = std::chrono::time_point; + std::map tsValues; + for(auto name : outputNames) { + auto frame = latestFrameGroup.value()->get(name); + tsValues.emplace(name, frame->getTimestamp(dai::CameraExposureOffset::END)); + } + auto compFunct = [](const std::pair& p1, const std::pair& p2) -> bool { return p1.second < p2.second; }; + + auto maxElement = std::max_element(tsValues.begin(), tsValues.end(), compFunct); + auto minElement = std::min_element(tsValues.begin(), tsValues.end(), compFunct); + + auto delta = maxElement->second - minElement->second; + auto deltaUs = std::chrono::duration_cast(delta).count(); + + if(deltaUs >= SYNC_THRESHOLD_SEC * 1e6) { + std::cout << "Sync error: Sync lost, threshold exceeded " << deltaUs << " us" << std::endl; + continue; + } + + for(const auto& outputName : outputNames) { + auto frame = latestFrameGroup.value()->get(outputName); + cv::imshow("synced_view_" + outputName, frame->getCvFrame()); + } + + latestFrameGroup.reset(); // Wait for next batch + } + + if((cv::waitKey(1) & 0xFF) == 'q') { + running.store(false); + break; + } + } + + for(auto& thread : threads) { + thread.join(); + } + + cv::destroyAllWindows(); + return 0; +} diff --git a/examples/python/Misc/MultiDevice/ptp_frame_sync_minimal.py b/examples/python/Misc/MultiDevice/ptp_frame_sync_minimal.py index e10f854759..856a8967d6 100644 --- a/examples/python/Misc/MultiDevice/ptp_frame_sync_minimal.py +++ b/examples/python/Misc/MultiDevice/ptp_frame_sync_minimal.py @@ -24,6 +24,15 @@ def interruptHandler(sig, frame): print("Exiting now!") exit(0) +def getSensorName(device: dai.Device, socket: dai.CameraBoardSocket) -> str: + sensorName = "" + for sckt, sName in device.getCameraSensorNames().items(): + if sckt == socket: + sensorName = sName + break + if sensorName == "": + raise RuntimeError(f"No sensor name found for {socket.name} on {deviceName}") + signal.signal(signal.SIGINT, interruptHandler) def getDeviceName(device : dai.Device) -> str: @@ -55,18 +64,9 @@ def getDeviceName(device : dai.Device) -> str: deviceName = getDeviceName(device) for socket in device.getConnectedCameras(): - ######################################################################## # TODO: remove this when OV9282 is supporter for PTP - sensorName = "" - for sckt, sName in device.getCameraSensorNames().items(): - if sckt == socket: - sensorName = sName - break - if sensorName == "": - raise RuntimeError(f"No sensor name found for {socket.name} on {deviceName}") - if sensorName == "OV9282": + if getSensorName(device, socket) == "OV9282": continue - ######################################################################## # create a queue for each camera on the device cam = devicePipeline.create(dai.node.Camera).build(socket, sensorFps=targetFps) From 2ac2f50f2ecbc143deffe6fa4f2fa664232c3e62 Mon Sep 17 00:00:00 2001 From: "stas.bucik" Date: Wed, 20 May 2026 12:45:02 +0200 Subject: [PATCH 4/4] Reduce the ammount of sync error messages on startup Signed-off-by: stas.bucik --- .../Misc/MultiDevice/ptp_frame_sync_minimal.py | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/examples/python/Misc/MultiDevice/ptp_frame_sync_minimal.py b/examples/python/Misc/MultiDevice/ptp_frame_sync_minimal.py index 856a8967d6..6f1887eb27 100644 --- a/examples/python/Misc/MultiDevice/ptp_frame_sync_minimal.py +++ b/examples/python/Misc/MultiDevice/ptp_frame_sync_minimal.py @@ -143,6 +143,8 @@ def data_collector(deviceName, socketName): # main display loop latestFrameGroup = None + initialSync = False + displayedInitialSyncMessage = False while running: # Get frames from sync node output queue while syncedGroups.has(): @@ -157,9 +159,17 @@ def data_collector(deviceName, socketName): syncStatus = abs(delta) < syncThresholdSec if not syncStatus: - print(f"Sync error: Sync lost, threshold exceeded {delta * 1e6} us") + if not initialSync: + if not displayedInitialSyncMessage: + print("Waiting for initial sync...") + displayedInitialSyncMessage = True + else: + print(f"Sync error: Sync lost, threshold exceeded {delta * 1e6} us") continue + if not initialSync: + initialSync = True + for outputName in outputNames: msg = latestFrameGroup[outputName] frame = msg.getCvFrame()