当机器人同时装备相机和激光雷达时,Robostral Navigate要接收的不是原始话题,而是经过对齐和构造后的输入张量。前置处理的关键是先把两路消息的格式、坐标系和时间戳对齐,再转成统一的数据张量。这里给出的是通用ROS流程,适用于大多数基于ROS的导航系统整合。
多传感器前置处理的判断方向:先查话题消息类型,确认数据字段;再用tf2把雷达和图像统一到同一坐标系;用ApproximateTimeSynchronizer做时间戳对齐;最后把图像和激光扫描转成numpy数组并归一化拼接成张量。验证时检查shape、值域和NaN。若某一步缺失(如没有tf变换或时间戳偏差过大),融合张量会无效,需先补全硬件驱动或静态变换。
检查各传感器话题消息类型
先确认机器人的话题列表,看看有没有图像和雷达话题。用 rostopic list 列出所有话题,再分别用 rostopic type 查看话题的消息类型,用 rostopic echo -n1 看一条消息的具体字段。
rostopic list
rostopic type /camera/image_raw
rostopic echo -n1 /camera/image_raw
如果输出是 sensor_msgs/Image,需要记下编码、宽高和数据步长;如果是 sensor_msgs/LaserScan,需要记下角度范围和距离数组;如果是 sensor_msgs/PointCloud2,则需要查看点的字段布局。这一步决定了后面如何解析数据,也决定了输入张量的维度。
注意:有些雷达驱动同时发布LaserScan和PointCloud2,优先选择模型能直接消费的格式,否则还得做额外转换。
将传感器数据统一到相同坐标系
相机和雷达通常安装在不同位置,需要把两路数据变换到同一个参考系(如 base_link 或 odom)。在ROS里用tf2做变换。先确认机器人有 camera_link 到 laser_frame 的tf,然后用 lookup_transform 拿到变换矩阵,再用 do_transform_cloud 或 do_transform_pose 应用。
import tf2_ros
import tf2_geometry_msgs
from sensor_msgs.msg import PointCloud2
tf_buffer = tf2_ros.Buffer()
tf_listener = tf2_ros.TransformListener(tf_buffer)
try:
transform = tf_buffer.lookup_transform('base_link', 'laser_frame', rospy.Time(0), rospy.Duration(1.0))
transformed_cloud = tf2_geometry_msgs.do_transform_cloud(cloud_msg, transform)
except (tf2_ros.LookupException, tf2_ros.ExtrapolationException) as e:
rospy.logerr("tf transform failed: %s", e)
如果是图像,由于图像本身是二维信息,不能直接做三维变换。通常做法是把点云变换到相机坐标系下,然后投影到图像平面;或者只用相机外参把图像像素坐标转换为空间射线。对Robostral Navigate输入融合场景,建议把激光点云变换到与图像同一个坐标源头(相机坐标系),再根据内参投影,得到深度图与图像对齐。
如果机器人没有可用的tf,需要先配置静态变换参数,或者改进驱动发布正确的TF。
实现时间戳同步
相机和雷达帧率不同,直接拿各自的最新消息会导致时间错位。用ROS的 message_filters 做时间同步。推荐 ApproximateTimeSynchronizer,它允许消息时间在设定容差内匹配。
from message_filters import ApproximateTimeSynchronizer, Subscriber
from sensor_msgs.msg import Image, LaserScan
image_sub = Subscriber('/camera/image_raw', Image)
scan_sub = Subscriber('/scan', LaserScan)
def callback(img_msg, scan_msg):
# 在这里处理同步后的图像和激光扫描
pass
sync = ApproximateTimeSynchronizer([image_sub, scan_sub], queue_size=10, slop=0.1)
sync.registerCallback(callback)
rospy.spin()
其中 slop 是允许的最大时间差,单位秒。通常根据传感器帧周期来定,例如10Hz的传感器可以设0.1~0.2秒。如果传感器数据经常匹配不上,可以适当提高 queue_size 或放宽 slop,但过大会引入时间误差,导致预测不稳。
构造模型输入张量并归一化
同步后的图像和激光扫描需要转成numpy数组并归一化。以下假设图像编码是 mono8(灰度),激光扫描是 LaserScan。
import numpy as np
import cv2
from sensor_msgs.msg import Image, LaserScan
def image_to_numpy(img_msg):
img_data = np.frombuffer(img_msg.data, dtype=np.uint8).reshape(img_msg.height, img_msg.width, -1)
if img_msg.encoding == 'mono8':
img_np = img_data[..., 0] # (H, W)
elif img_msg.encoding == 'rgb8':
img_np = img_data[..., :3]
else:
raise ValueError("Unsupported encoding: {}".format(img_msg.encoding))
img_np = cv2.resize(img_np, (224, 224)).astype(np.float32) / 255.0
return img_np
def scan_to_numpy(scan_msg, max_range=10.0):
ranges = np.array(scan_msg.ranges, dtype=np.float32)
ranges = np.nan_to_num(ranges, nan=max_range, posinf=max_range, neginf=0.0)
ranges = np.clip(ranges, 0.0, max_range) / max_range
return ranges
def callback(img_msg, scan_msg):
img_np = image_to_numpy(img_msg)
scan_np = scan_to_numpy(scan_msg)
# 示例:将图像展平与扫描拼接成一维向量
img_flat = img_np.flatten()
final_tensor = np.concatenate([img_flat, scan_np]).astype(np.float32)
# 如果模型接受多通道,也可以分别构造
# 这里需要根据Robostral Navigate输入层要求决定具体组合方式
注意归一化:图像除以255,激光距离除以最大距离。如果模型要求特定shape如 (N, H, W, C),需要按模型文档的输入规范重新组织,例如把scan补零或插值到固定长度。
验证融合后输入的有效性
最后一定要验证张量的shape和值域,再送入模型做一次前向日志确认。这样可以快速发现预处理错误,而不是等模型输出不可解释后再回头排查。
print("Final tensor shape:", final_tensor.shape)
print("Final tensor dtype:", final_tensor.dtype)
print("Final tensor min: {}, max: {}".format(np.min(final_tensor), np.max(final_tensor)))
if np.isnan(final_tensor).any():
raise ValueError("NaN in input tensor")
# 模拟前向日志(根据实际模型接口替换)
import logging
logging.debug("Model input prepared: %s", final_tensor.shape)
检查点:shape是否和模型输入层一致?值域是否在预期范围内?有没有NaN或Inf?如果出现NaN,多半是点云转换时缺少变换或距离数组本身有非法值。如果shape不对,检查图像resize尺寸和scan长度是否固定。建议在回调里加一个 rospy.loginfo_once 打印形状,避免刷屏。
这一步通过后,后续融合模型才能稳定运行。