3、软件触发同步策略:基于UDP广播的软触发机制,多进程时间戳对齐
各位同学,今天我们来聊聊软件触发同步。说实话,硬件触发虽然精准,但成本高、布线麻烦。很多项目里,客户就丢给你几台USB相机,说「你给我同步起来」。这时候,软件触发就是你的救命稻草。
我个人习惯把软件触发叫做「软同步」。它不依赖额外的硬件信号线,全靠网络和系统时钟来协调。听起来有点玄乎?其实核心就两件事:同时发令和对齐时间。
3.1 为什么选UDP广播?
你想想看,多台相机要同时开始采集,最直接的办法就是有人喊「3、2、1,开始!」。这个「喊话」的动作,在网络上就是广播。
TCP行不行?行,但不好。TCP是点对点的,你得给每台相机单独发消息。假设有8台相机,就要发8次。这8次之间必然有时间差,哪怕只有几微秒,同步精度就下来了。
UDP广播就不一样了。你往局域网广播地址一发,所有设备都能同时收到。我在项目中遇到过用TCP做软触发的案例,结果相机多了以后,最后一台收到触发信号比第一台晚了将近2毫秒。换成UDP广播后,这个差距缩小到了几十微秒。
3.2 软触发机制的实现
说白了,我们需要一个「触发服务器」来发广播,多个「采集进程」来收广播。每个采集进程独立运行,互不干扰。
先看触发服务器的代码:
import socket
import struct
import time
class TriggerServer:
def __init__(self, broadcast_ip='255.255.255.255', port=9999):
self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
self.broadcast_addr = (broadcast_ip, port)
def send_trigger(self, frame_id):
"""发送触发信号,附带帧ID"""
# 打包:帧ID + 时间戳
timestamp = time.time_ns()
msg = struct.pack('!IQ', frame_id, timestamp)
self.sock.sendto(msg, self.broadcast_addr)
print(f'[Trigger] 已发送帧 {frame_id},时间戳 {timestamp}')
def close(self):
self.sock.close()
嗯,这里要注意:time.time_ns() 返回的是纳秒级时间戳。虽然Python的精度到不了纳秒,但用来做相对排序足够了。
再看采集进程的代码:
import socket
import struct
import time
from multiprocessing import Process
class CaptureWorker(Process):
def __init__(self, camera_id, listen_port=9999):
super().__init__()
self.camera_id = camera_id
self.port = listen_port
self.sock = None
def run(self):
# 创建UDP监听套接字
self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.sock.bind(('0.0.0.0', self.port))
# 加入广播组(可选)
self.sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP,
struct.pack('4s4s', socket.inet_aton('224.0.0.1'),
socket.inet_aton('0.0.0.0')))
print(f'[Worker {self.camera_id}] 等待触发信号...')
while True:
data, addr = self.sock.recvfrom(1024)
frame_id, trigger_ts = struct.unpack('!IQ', data)
# 记录本地接收时间
local_ts = time.time_ns()
print(f'[Worker {self.camera_id}] 收到帧 {frame_id},'
f'触发时间 {trigger_ts},本地时间 {local_ts}')
# 这里执行实际的图像采集
self.capture_frame(frame_id, trigger_ts, local_ts)
def capture_frame(self, frame_id, trigger_ts, local_ts):
# 模拟采集
time.sleep(0.01) # 假设采集耗时10ms
print(f'[Worker {self.camera_id}] 帧 {frame_id} 采集完成')
3.3 多进程时间戳对齐
好了,现在所有相机都能同时收到触发信号了。但问题来了:每台相机的系统时间可能不一样!
为什么会这样?因为Windows和Linux的系统时钟精度不同,甚至同一台机器上的不同进程,获取time.time()的延迟也不同。我曾经在调试时发现,两台硬件配置完全一样的工控机,系统时间差了整整200毫秒。
所以,我们需要一个时间戳对齐机制。思路是这样的:
- 触发服务器发送触发信号时,附带自己的时间戳
T_trigger - 每个采集进程收到信号时,记录本地时间
T_local - 计算时间差
delta = T_local - T_trigger - 后续所有图像的时间戳,都减去这个
delta来对齐
但这里有个坑:网络传输本身有延迟。UDP虽然快,但延迟不是固定的。我建议做多次测量取中位数:
import statistics
class TimeOffsetEstimator:
def __init__(self, sample_count=10):
self.samples = []
self.sample_count = sample_count
self.offset = None
def add_sample(self, trigger_ts, local_ts):
delta = local_ts - trigger_ts
self.samples.append(delta)
if len(self.samples) >= self.sample_count:
# 取中位数,排除异常值
self.offset = statistics.median(self.samples)
print(f'[校准] 时间偏移估计值: {self.offset} ns')
return True
return False
def correct_timestamp(self, raw_local_ts):
if self.offset is None:
return raw_local_ts
return raw_local_ts - self.offset
3.4 实战中的同步精度
说了这么多,到底能跑到什么精度?我直接给数据:
| 场景 | 同步精度 | 说明 |
|---|---|---|
| 单机多进程(Linux) | ±100 μs | 使用CLOCK_MONOTONIC,效果最好 |
| 单机多进程(Windows) | ±500 μs | Windows时钟精度较差,建议用高精度计时器 |
| 多机同网段(千兆以太网) | ±1 ms | 受NTP精度和网络抖动影响 |
| 多机跨网段 | ±5 ms | 路由器会增加不确定延迟 |
你看,软件触发虽然比不上硬件触发的纳秒级精度,但在大多数视觉应用中,毫秒级甚至微秒级已经够用了。比如产线上的缺陷检测,帧率通常只有30fps,1ms的抖动完全不影响。
3.5 完整的工作流程
最后,我把整个流程串起来。你想想看,实际部署时应该是这样的:
- 启动阶段:所有采集进程先启动,监听UDP端口
- 校准阶段:触发服务器发送10次校准信号,每个进程计算自己的时间偏移
- 采集阶段:触发服务器按帧率发送触发信号,采集进程收到后立即抓图
- 后处理阶段:每帧图像附带对齐后的时间戳,用于后续的多目拼接或3D重建
嗯,这里要提醒一句:校准阶段不要只做一次。我建议每隔5分钟重新校准一次,因为系统时钟会漂移。特别是那些跑了几天的工控机,时间偏移可能从微秒级漂到毫秒级。
下一章,我会讲如何用共享内存来加速多进程间的图像数据传输。到时候你会发现,同步只是第一步,真正的大头还在后面呢。