第三章：ROS Bag 解析——用 Python 把数据“拆”出来

做机器人数据回放，绕不开 ROS Bag。说白了，它就是 ROS 世界的“黑匣子”，把传感器数据、控制指令、状态信息一股脑全录下来。我刚开始接触时，觉得这玩意儿不就是个压缩包嘛，后来才发现——嗯，它比我想象的讲究多了。

3.1 什么是 ROS Bag？

ROS Bag 是 ROS 系统里专门用来记录和回放消息的文件格式。你可以把它理解成“行车记录仪”，只不过它记录的不是视频，而是话题（Topic）和消息（Message）。

我个人习惯把 Bag 文件比作一个“时间胶囊”：

• 话题（Topic）：就像广播频道，比如 /camera/image 是图像频道，/odom 是里程计频道
• 消息（Message）：每个频道里具体的数据包，比如一张图片、一个坐标
• 时间戳：每条消息都带着精确到纳秒的时间标记

我在项目中遇到过最头疼的事——录了半小时的 Bag，结果发现话题名拼错了，数据全白录。所以，解析 Bag 之前，先搞清楚里面有什么，这是基本功。

3.2 环境准备：装好 rosbag 库

Python 解析 ROS Bag，靠的是 rosbag 这个库。它其实是 ROS 自带的 Python 接口，但如果你只想纯 Python 环境跑，可以单独装：

pip install rosbag

注意：rosbag 依赖 ROS 的 Python 绑定，建议在 ROS 环境（比如 Ubuntu + ROS Noetic）下使用。Windows 用户？嗯，我建议你装个 WSL 或者直接用 Docker，省得折腾。

3.3 核心操作：读取 Bag 文件

读取 Bag 文件其实就三步：打开、遍历、关闭。来看代码：

import rosbag

bag_path = "demo.bag"
bag = rosbag.Bag(bag_path, "r")  # "r" 表示只读模式

for topic, msg, t in bag.read_messages():
    print(f"话题: {topic}")
    print(f"消息类型: {type(msg)}")
    print(f"时间戳: {t.to_sec():.3f} 秒")
    # 这里可以处理具体消息

bag.close()

你想想看，这段代码干了什么？它把 Bag 文件里每一条消息都“拎”出来了。topic 是字符串，msg 是 ROS 消息对象，t 是 rosgraph_msgs.msg.Clock 类型的时间戳。

我个人习惯用 with 语句，省得忘记 close：

with rosbag.Bag("demo.bag", "r") as bag:
    for topic, msg, t in bag.read_messages():
        # 处理逻辑
        pass

3.4 提取特定话题的消息

实际项目中，Bag 文件里可能有几十个话题。你不可能全遍历一遍，太慢了。我一般只关心几个关键话题，比如 /camera/image_raw 和 /odom。

怎么做？用 read_messages() 的 topics 参数：

topics_of_interest = ["/camera/image_raw", "/odom"]

with rosbag.Bag("demo.bag", "r") as bag:
    for topic, msg, t in bag.read_messages(topics=topics_of_interest):
        if topic == "/odom":
            print(f"位置: x={msg.pose.pose.position.x:.2f}, y={msg.pose.pose.position.y:.2f}")
        elif topic == "/camera/image_raw":
            print(f"图像尺寸: {msg.width}x{msg.height}")

这里有个细节：read_messages() 内部做了话题过滤，只返回你指定的话题。效率比全遍历高很多。我记得有一次处理 10GB 的 Bag，用过滤后只花了 3 秒就提取完关键数据。

3.5 获取 Bag 的元信息

有时候你只想看看 Bag 文件里有什么，不想遍历所有消息。可以用 get_type_and_topic_info()：

with rosbag.Bag("demo.bag", "r") as bag:
    info = bag.get_type_and_topic_info()
    
    print("=== Bag 信息 ===")
    print(f"起始时间: {bag.get_start_time():.2f}")
    print(f"结束时间: {bag.get_end_time():.2f}")
    print(f"消息总数: {bag.get_message_count()}")
    
    print("\n=== 话题列表 ===")
    for topic, details in info.topics.items():
        print(f"  {topic}: {details.msg_type} ({details.message_count} 条消息)")

输出效果类似这样：

这个功能特别实用。我每次拿到新 Bag，第一件事就是打印这个信息，看看数据质量怎么样。如果某个话题消息数量为 0，那八成是录制时没启动对应节点。

3.6 实战：提取并保存为 CSV

光看不够，得把数据存下来。下面是一个完整的例子，把 /odom 话题的位姿数据提取出来，保存成 CSV：

import rosbag
import csv

def extract_odom_to_csv(bag_path, csv_path):
    with rosbag.Bag(bag_path, "r") as bag, open(csv_path, "w", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(["timestamp", "x", "y", "z", "qx", "qy", "qz", "qw"])
        
        for topic, msg, t in bag.read_messages(topics=["/odom"]):
            pos = msg.pose.pose.position
            ori = msg.pose.pose.orientation
            writer.writerow([
                t.to_sec(),
                pos.x, pos.y, pos.z,
                ori.x, ori.y, ori.z, ori.w
            ])
    
    print(f"已保存 {csv_path}")

# 使用示例
extract_odom_to_csv("demo.bag", "odom_data.csv")

你想想看，这个 CSV 文件可以直接导入 Excel 或者 Pandas 做分析。我经常用这个方法来快速验证里程计数据有没有跳变。

3.7 知识体系图

下面这张图，帮你理清 ROS Bag 解析的核心流程：

3.8 避坑指南

最后，分享几个我实战中遇到的坑：

• 时间戳单位：ROS 的时间戳是秒+纳秒，t.to_sec() 返回浮点数。别直接用 t.secs，那是整数秒，会丢精度。
• 消息类型不匹配：同一个话题名，不同 Bag 可能用不同消息类型。解析前先用 get_type_and_topic_info() 确认一下。
• 大文件内存问题：别一次性把整个 Bag 读到内存里。read_messages() 是迭代器，逐条处理，内存友好。
• 编码问题：ROS 1 的 Bag 默认是 Python 2 的字符串编码。如果你用 Python 3，遇到中文话题名可能会报错。加个 encoding="utf-8" 参数能解决。

好了，ROS Bag 解析这块就聊到这儿。你手头如果有 Bag 文件，现在就可以试试把它拆开看看——说不定能发现一些有意思的数据规律。