ROS2机器人数据录制与回放实战

可视化演示

展示自注意力与交叉注意力中 Query-Key 权重矩阵的动态分布，通过温度参数调节注意力聚焦程度。

初识rosbag2：传感器的“多路记录仪”

你车上的行车记录仪通常有好几路摄像头：前路拍路况，后路防追尾，车内防纠纷。每路摄像头各拍各的，互不干扰。ROS2 里的“话题”（topic）就相当于这些摄像头——激光雷达把点云发到 /scan 话题，相机把图像发到 /camera 话题，IMU 把角速度和加速度发到 /imu 话题。谁想用这些数据，谁就去“订阅”对应的话题。

rosbag2 就像一台多路记录仪，能把多路话题的信号全部“录”下来。它做的事情很简单：当一个话题上有新消息发布时，它把消息内容、话题名、以及消息被发布那一刻的时间戳（timestamp）按顺序写进磁盘。于是 bag 文件本质上就是一条“带时间标签的消息流水账”。

在 ROS2 中，bag 不是单个文件，而是一个目录，里面通常有一个 SQLite3 数据库文件（比如 xxx_0.db3）和一份 metadata.yaml 元数据。前者存消息本体，后者记录话题列表、消息数量、录制时间等信息。用 ros2 bag info 可以查看目录内有哪些话题、各有多少条消息。

$$\text{bag} = \left\{ (t_i, \text{topic}_i, m_i) \;\middle|\; i = 1, 2, \ldots, N \right\}$$

1. bag 集合符号：左边 \text{bag} 代表整个数据包，相当于记录仪的 SD 卡目录。

$$\text{bag} = \{ \dots \}$$

2. 有序三元组：每条记录用括号里的三要素表示，类似“时间 + 摄像头编号 + 画面”。

$$(t_i, \text{topic}_i, m_i)$$

3. 时间戳 t_i：第 i 条消息被发布的时刻，回放时按它排序。

$$t_i$$

4. 话题名 topic_i：这条消息来自哪个话题，比如 /scan 表示激光雷达。

$$\text{topic}_i$$

5. 消息体 m_i：消息的具体内容，比如一帧激光点云或一张图像。

$$m_i$$

6. 总条数 N：bag 中所有消息的数量，i 从 1 数到 N。

$$i = 1, 2, \ldots, N$$

理解了 bag 的“三要素”存储方式，再看后面的录制、回放、裁剪，就会发现它们都围绕同一件事：怎么把“带时间的话题消息”管好。就像行车记录仪，先知道它存什么，才知道怎么选路录像、怎么翻看回放。

录制启动：决定录哪几路“摄像头”

装行车记录仪时，师傅会问你前路要不要、后路要不要、车内要不要。全都要当然“安全”，但存储卡就那么大，录太多反而拖垮画质。机器人录制同理，第一步是想清楚：这次调试需要哪几路数据？

rosbag2 的录制命令非常直白：ros2 bag record。不指定话题时它会提示你加上 --topics；想录全部话题可以加 -a（--all）参数；想指定输出目录用 -o。比如只录激光雷达和 IMU，可以这样写：

录制前最好估一下数据量。每个话题的数据速率 λ 不同：激光雷达每秒几十万点，相机一帧可能就几百 KB。录制时长 T 乘以速率总和，就是预计占用的磁盘空间。现场磁盘不够时，宁可少录一路不紧要的话题，也别把系统“录”到卡死。

$$\displaystyle S = \sum_{j \in \mathcal{T}} \lambda_j \times T_{\text{rec}}$$

1. 空间总和 S：预计占用的磁盘空间，单位是字节。

$$S$$

2. 话题集合 \mathcal{T}：你选了哪几路“摄像头”，就放进这个集合。

$$\mathcal{T}$$

3. 速率 \lambda_j：话题 j 每秒钟产生多少字节数据。

$$\lambda_j$$

4. 时长 T_{\text{rec}}：计划录制多少秒。

$$T_{\text{rec}}$$

5. 累加符号 \sum：把每一路话题的“速率 × 时长”都加起来。

$$\sum_{j \in \mathcal{T}}$$

所以录制不是“多多益善”，而是像老司机调记录仪一样——根据今天要处理的事故类型，选对摄像头角度。想清楚录什么，后面回放和分析才轻松。

ros2 bag record -o nav_bag /scan /imu

回放调试：按下“重演”键

事故发生后，交警最常做的动作是：调出记录仪，把时间拨回事发前五分钟，放慢看。rosbag2 的回放功能就是这个“重演”键，让你把现场任意时刻的数据“再演一遍”。

回放命令也简单：ros2 bag play bag目录。它会读取 bag 里的消息，按录制时的时间戳顺序，重新发布到对应话题上。此时任何订阅这些话题的节点都能收到数据——rviz2、导航算法、感知模型，全都“以为”自己又回到了现场。

回放还支持几个实用参数：-r 2.0 表示 2 倍速回放；-l 循环播放；--topics 只回放部分话题；--wait-for-all-acknowledged 等所有订阅端都准备好再发。倍率与耗时的关系如下：

$$\displaystyle T_{\text{real}} = \frac{T_{\text{recorded}}}{r_{\text{rate}}}$$

1. 实际耗时 T_{\text{real}}：从开始到放完，真实过去了多少秒。

$$T_{\text{real}}$$

2. 录制时长 T_{\text{recorded}}：bag 里消息跨越的时间长度。

$$T_{\text{recorded}}$$

3. 倍率 r_{\text{rate}}：播放速度倍率，2.0 就是两倍速。

$$r_{\text{rate}}$$

4. 除法关系：倍率越大，实际耗时越短，消息之间的时间间隔被压缩。

$$\frac{T_{\text{recorded}}}{r_{\text{rate}}}$$

有了“重演”键，bug 就不再是“现场限定”了。哪怕客户在千里之外，只要他发来一个 bag，你就能在办公室里把案发现场重演一百遍，直到看穿它。

过滤裁剪：剪出“高光时刻”

记录仪录了 5 个小时，事故只发生在 30 秒内。你当然不会把 5 小时录像原封不动发出去，而是剪出关键片段。bag 也一样，现场一录就是几 GB，真正有价值的可能只有几十秒。

过滤裁剪的本质，是从旧 bag 中挑出“时间段落在 [t_a, t_b] 内”的消息，写成一个新 bag。这个话题留、那个话题不要，也可以顺手完成，这就是话题筛选重组。ROS2 较新版本提供 ros2 bag filter 命令，更通用的是用 Python 的 rosbag2_py 库程序化处理；需要拆包时按时间或话题切分，需要合包时依次读入多个旧包写入同一个新包。必要时开启 --compression-format zstd 压缩，把 bag 体积再压一压。

过滤时按消息的时间戳判断：只保留时间戳落在目标区间内的消息。这样新 bag 既保留了现场的关键数据，又大幅缩小体积，方便拷贝分享和反复加载。

$$\displaystyle M_{\text{new}} = \left\{\, m \in M_{\text{old}} \;\middle|\; t_a \le t_m \le t_b \,\right\}$$

1. 旧消息集 M_{\text{old}}：原始 bag 里的全部消息。

$$M_{\text{old}}$$

2. 新消息集 M_{\text{new}}：过滤后生成的新 bag 的消息。

$$M_{\text{new}}$$

3. 消息时间戳 t_m：某条消息 m 被记录的时刻。

$$t_m$$

4. 区间端点 t_a、t_b：裁剪的起始与结束时刻。

$$t_a \le t_m \le t_b$$

5. 集合条件式：竖线后面是筛选条件，只有满足条件的 m 才被保留。

$$\{ m \in M_{\text{old}} \mid \dots \}$$

裁剪就像视频剪辑软件里的“剪刀”：把 5 小时的录像剪成 30 秒的“高光时刻”，证据更聚焦，传给别人也更方便。记住，时间戳就是记录仪上的时间轴。

可视化分析：让数据“看得见”

光看记录仪画面还不够直观，老司机更信任“画面 + 轨迹”叠加的视图：车在哪、朝哪开、四周有什么障碍，一目了然。机器人的“画面 + 轨迹”就是 rviz2 里的话题消息 + TF 坐标变换。

回放 bag 的同时打开 rviz2，添加对应的话题显示（比如 /scan 的 LaserScan、/camera 的 Image），就能看到激光点云和图像像“现场直播”一样动起来。为了让回放数据驱动 rviz2 的时间轴，需要在 rviz2 里勾选 Use Sim Time（使用仿真时间），让消息的时间戳统一参考 bag 里的时钟。

多传感器能否同步，直接影响融合算法的正确性。检查时间对齐的方法是看消息时间戳与基准时间的差值 Δt 是否在容差 τ 内。比如相机与激光雷达要求 5ms 内对齐，超过就要检查硬件时钟或触发模式。

$$\displaystyle \Delta t = \left| t_{\text{sensor}} - t_{\text{master}} \right|, \quad \Delta t \le \tau$$

1. 传感器时间 t_{\text{sensor}}：某条传感器消息自带的时间戳。

$$t_{\text{sensor}}$$

2. 基准时间 t_{\text{master}}：作为对齐基准的时间源，如 TF 或主传感器。

$$t_{\text{master}}$$

3. 绝对值符号：不管谁大谁小，只关心差距有多大。

$$\left| \dots \right|$$

4. 容差 \tau：允许的最大偏差，例如 5ms。

$$\tau$$

5. 不等式判断：偏差在容差内，两个传感器才算“合拍”。

$$\Delta t \le \tau$$

当记录仪画面和轨迹叠加显示，事故原因往往一眼就看穿：可能是某个传感器掉线，可能是 TF 树断了。可视化让“看不见的数据”变成“看得见的现场”，这正是 rviz2 配合 rosbag2 的威力。

实战技巧：老司机的“维护心得”

老司机用记录仪有几个习惯：定期格式化存储卡、高温天别连续录太久、重要片段加锁防止被覆盖。玩转 rosbag2 也有类似的心法：管好磁盘、盯住丢帧、把现场问题变成离线任务。

现场录制最怕“录了个寂寞”——以为全录了，回放才发现丢帧。丢帧的常见原因有三个：磁盘写入速度跟不上、CPU 负载过高导致回调积压、订阅方 QoS 与发布方不匹配导致消息被丢弃。判断丢帧可以用录制完整度 C 来衡量。

完整度 C 表示 bag 实际写入的消息数占发布端发出消息数的比例。录制时盯着 C，若明显低于 100%，优先排查磁盘和 CPU；若发布端本身在丢，还要检查 QoS。日常维护建议：录制前用 df -h 看剩余空间，给 bag 目录单独挂大容量盘，重要数据录完后及时用 ros2 bag info 核验。

$$\displaystyle C = \frac{N_{\text{saved}}}{N_{\text{published}}} \times 100\%$$

1. 已存消息数 N_{\text{saved}}：bag 里实际写入的消息条数。

$$N_{\text{saved}}$$

2. 发布消息数 N_{\text{published}}：录制期间发布端一共发出的消息条数。

$$N_{\text{published}}$$

3. 取比值：两个数相除，得到“存下来的比例”。

$$\frac{N_{\text{saved}}}{N_{\text{published}}}$$

4. 乘 100%：把比例换成百分比，方便直观判断。

$$\times 100\%$$

5. 完整度 C：越接近 100% 越健康，明显偏低就按三步排查。

$$C$$

掌握了这些心法，你就不再是“会按录制键”的新手，而是懂得什么时候录、怎么不丢帧、如何把现场问题一键搬运回办公室的“老司机”。机器人的每一段关键时刻，都有你的记录仪在守护。

AI 应用场景

Python 演示

"""rosbag2 数据工具箱: 扫描话题/统计消息/按时间过滤生成新 bag""" import sysfrom pathlib import Path from rosbag2_py import (    SequentialReader,    SequentialWriter,    StorageOptions,    ConverterOptions,    TopicMetadata,)  class BagToolkit:    """封装 rosbag2 读写能力的工具类"""     def __init__(self, bag_dir):        self.bag_dir = Path(bag_dir)        self.counts = {}     def _open_reader(self):        reader = SequentialReader()        storage = StorageOptions(uri=str(self.bag_dir), storage_id="sqlite3")        converter = ConverterOptions(            input_serialization_format="cdr",            output_serialization_format="cdr",        )        reader.open(storage, converter)        return reader     def scan_topics(self):        """列出 bag 中所有话题及消息类型"""        reader = self._open_reader()        topics = reader.get_all_topics_and_types()        print(f"发现 {len(topics)} 个话题:")        for t in topics:            print(f"  {t.name}  [{t.type}]")        return topics     def count_messages(self):        """逐条读取消息, 统计每个话题的消息数"""        reader = self._open_reader()        self.counts.clear()        while reader.has_next():            topic_name, _data, _stamp = reader.read_next()            self.counts[topic_name] = self.counts.get(topic_name, 0) + 1        total = sum(self.counts.values())        print(f"消息总数: {total}")        for name, n in sorted(self.counts.items()):            print(f"  {name}: {n} 条")        return total     def filter_by_time(self, start_ns, end_ns, output_dir):        """按时间戳区间过滤并写出新 bag"""        reader = self._open_reader()        writer = SequentialWriter()        out_storage = StorageOptions(uri=str(output_dir), storage_id="sqlite3")        out_converter = ConverterOptions(            input_serialization_format="cdr",            output_serialization_format="cdr",        )        writer.open(out_storage, out_converter)        for t in reader.get_all_topics_and_types():            writer.create_topic(                TopicMetadata(name=t.name, type=t.type, serialization_format="cdr")            )        kept = 0        while reader.has_next():            name, data, stamp = reader.read_next()            if start_ns <= stamp <= end_ns:                writer.write(name, data, stamp)                kept += 1        print(f"时间过滤完成, 保留 {kept} 条消息 -> {output_dir}")        return kept  def main():    if len(sys.argv) < 2:        print("用法: python bag_tool.py <bag目录> [起始ns] [结束ns] [输出目录]")        sys.exit(1)    tool = BagToolkit(sys.argv[1])    tool.scan_topics()    if len(sys.argv) == 5:        start_ns, end_ns = int(sys.argv[2]), int(sys.argv[3])        tool.filter_by_time(start_ns, end_ns, sys.argv[4])    else:        tool.count_messages()  if __name__ == "__main__":    main()

1. reader = SequentialReader()创建顺序读取器，就像打开录像带准备逐帧观看。
2. StorageOptions(uri=str(self.bag_dir), storage_id="sqlite3")指定 bag 目录路径和存储引擎。
3. reader.get_all_topics_and_types()读取 bag 的“目录”，列出所有话题及其消息类型。
4. reader.has_next()判断是否还有下一条消息，相当于检查录像带是否放完。
5. reader.read_next()取出一条消息，得到话题名、数据和时间戳三要素。
6. self.counts.get(topic_name, 0) + 1字典计数技巧：没有就按 0 起步再加 1。
7. writer.create_topic(...)给新 bag 声明话题结构，否则写入会失败。
8. writer.write(name, data, stamp)把消息的三要素原样写入新 bag。
9. start_ns <= stamp <= end_ns时间戳区间判断，只保留“案发时刻”的消息。
10. tool.filter_by_time(...)一条命令完成“剪出高光时刻”的完整流程。

常见误区

面试题