数据采集系统架构：传感器选型、数据流设计、实时性要求分析
好，咱们进入第二章。上一章聊了整体方案，这一章我打算把数据采集系统这块掰开揉碎了讲。说白了，机械臂控制得好不好，七成功力在数据采集上。你传感器选错了，或者数据流设计有漏洞，后面算法再牛也白搭。
我个人习惯，做任何机器人项目，第一步不是写代码，而是先把「数据怎么来、怎么走、多快到」这三件事想清楚。今天咱们就按这个顺序来。
2.1 传感器选型：别只看参数表
传感器选型这事儿，我踩过不少坑。刚入行那会儿，我盯着数据手册上的精度和采样率看，觉得参数越高越好。结果呢？装上去发现通信带宽不够，数据根本传不回来。所以选型要综合看四个维度。
| 传感器类型 | 核心参数 | ROS2接口方式 | 我踩过的坑 |
|---|---|---|---|
| 关节角度编码器 | 分辨率 ≥ 18bit，更新率 ≥ 1kHz | CAN / EtherCAT → ros2_control | 某次用了12bit编码器，低速抖动根本看不出来 |
| 六维力/力矩传感器 | 采样率 ≥ 500Hz，过载保护 3x | EtherCAT / 串口 → custom msg | 串口波特率设低了，力控延迟飙到20ms |
| IMU（加速度计+陀螺仪） | 输出率 ≥ 400Hz，加速度 ±16g | SPI / I2C → sensor_msgs/Imu | I2C总线挂太多设备，时钟拉伸导致丢帧 |
| 末端视觉（深度相机） | 帧率 ≥ 30fps，深度精度 ≤ 3mm @ 1m | USB3.0 / GigE → image_transport | USB带宽被多个相机抢，图像传输断断续续 |
2.2 数据流设计：从传感器到控制器的管道
传感器选好了，接下来就是数据怎么流。我习惯把数据流分成三层：采集层、传输层、处理层。每一层都有讲究。
2.2.1 采集层：驱动与节点分离
每个传感器对应一个ROS2驱动节点。但我建议驱动节点只做一件事：把原始数据打包成ROS2消息，然后发出去。别在驱动里做滤波、坐标变换这些事。
# 一个简洁的编码器驱动节点示例
class EncoderDriver(Node):
def __init__(self):
super().__init__('encoder_driver')
self.pub = self.create_publisher(JointState, 'joint_states_raw', 10)
self.timer = self.create_timer(0.001, self.read_encoder) # 1kHz
def read_encoder(self):
# 只读原始值，不做处理
raw_position = self.hardware.read_position()
msg = JointState()
msg.position = [raw_position]
self.pub.publish(msg)
为什么这么做？因为一旦你在驱动里加了滤波，延迟就不可控了。我曾经在一个项目里把卡尔曼滤波写进了驱动节点，结果调试时发现滤波参数一改就得重启整个驱动，太痛苦了。
2.2.2 传输层：QoS配置是关键
ROS2的DDS给了我们很大的灵活性，但也容易配置错。我见过太多人直接用默认QoS，结果数据要么丢包严重，要么延迟爆炸。
| 数据类型 | 推荐QoS策略 | 原因 |
|---|---|---|
| 关节状态（高频） | KEEP_LAST + depth=1 + RELIABILITY_BEST_EFFORT | 只关心最新值，丢一帧无所谓 |
| 力传感器（关键） | KEEP_LAST + depth=5 + RELIABILITY_RELIABLE | 力控不能丢数据，但可以容忍少量延迟 |
| 视觉图像（大带宽） | KEEP_LAST + depth=1 + RELIABILITY_BEST_EFFORT + DURABILITY_VOLATILE | 图像量大，必须用best effort，否则网络会堵死 |
| 控制指令（最高优先级） | KEEP_LAST + depth=1 + RELIABILITY_RELIABLE + LIVELINESS_MANUAL | 必须确保指令到达，且能检测节点是否存活 |
2.2.3 处理层：时间戳对齐
数据到了处理层，第一件事就是时间戳对齐。ROS2的Time库提供了很好的工具，但前提是每个传感器消息都得带上正确的时间戳。
# 时间戳对齐的典型做法
class DataFusionNode(Node):
def __init__(self):
super().__init__('data_fusion')
self.joint_sub = self.create_subscription(
JointState, 'joint_states_raw', self.joint_callback, 10)
self.force_sub = self.create_subscription(
WrenchStamped, 'force_raw', self.force_callback, 10)
self.buffer = TimeSynchronizer([JointState, WrenchStamped], 10)
def joint_callback(self, msg):
# 确保时间戳是硬件时间，不是收到消息的时间
if msg.header.stamp.sec == 0:
self.get_logger().warn('时间戳未设置，使用当前时间')
msg.header.stamp = self.get_clock().now().to_msg()
self.buffer.add('joint', msg)
嗯，这里要注意：时间戳最好在驱动层就打好，用硬件触发的时间。如果等到处理层再打时间戳，那已经是「收到消息的时间」了，不是「数据产生的时间」，两者可能差好几毫秒。
2.3 实时性要求分析：多少毫秒才算够？
实时性这事儿，得看你的控制周期。机械臂控制一般分三个层级，每个层级的实时性要求完全不同。
你想想看，关节控制层如果抖动超过0.1ms，电机电流环就会震荡。我调试过一个六轴机器人，编码器数据走的是USB转CAN，结果USB的调度延迟不稳定，有时候2ms才来一次数据，电机直接嗡嗡响。后来换成EtherCAT，问题立刻解决。
2.4 数据采集系统的整体架构
最后，我把上面这些整合成一个完整的数据采集架构。这个架构我在三个项目里验证过，效果都不错。
这个架构的核心思路就一句话：传感器只管采，总线只管传，节点只管发，融合只管对齐。每一层职责清晰，出了问题也好定位。