ROS bag包解析实战：命令行与Python脚本全攻略

1. 为什么你必须学会解析bag包

做机器人、自动驾驶、SLAM、传感器融合的兄弟们，每天最离不开的东西就是bag包。不管是自己跑一版新算法想回放数据验证，还是拿到别人分享的现场采集数据想分析问题，第一步永远是先把bag包解析清楚。我见过不少新同学，手上捏着一堆bag包，却只会用rosbag play放一下，看个热闹，真要分析某个topic的数值变化、提取图像帧、检查tf树、定位异常时间戳，就卡住了。这篇文章我就把自己这些年解析bag包踩过的坑、沉淀下来的思路，从命令行到Python脚本再到可视化排查，完整梳理一遍，你可以当操作手册直接抄。

这内容适合谁？刚接触ROS的初学者，在实验室里负责数据采集和算法验证的同学，以及需要从海量bag数据中快速定位问题的工程师。读完你至少能独立完成：查看bag信息、提取任意topic、转换图像和视频、批量处理上千个bag、排查时间戳错位和tf缺失这类经典疑难杂症。

先强调一个核心认知：bag包本质上就是一个容器文件，里面按时间顺序存储了一堆ROS消息。理解这一点之后，所有解析操作都会变得非常顺——你做的无非就是“读容器、按需取消息、把消息转成可用格式”这三件事。下面我按实操顺序一步步展开。

2. 动手前的信息侦察：bag包里到底装了什么

拿到一个bag包，别一上来就rosbag play，先把“家底”摸清楚。这一步的效率直接决定你后面代码怎么写。

2.1 用rosbag info快速看懂目录结构

rosbag info是最基本的侦察命令，输出包含duration、topics、types、messages数量等核心信息。我强烈建议你加-y参数，以YAML格式输出，结构清爽，脚本处理也方便：

rosbag info -y 2024-05-12-15-30-00.bag

输出大致长这样：

path: 2024-05-12-15-30-00.bag version: 2.0 duration: 431.26 start: May 12 2024 15:30:01.20 end: May 12 2024 15:37:12.46 size: 5.3GB compression: bz2 messages: 238412 types: - type: sensor_msgs/Image count: 12000 - type: sensor_msgs/Imu count: 24000 topics: - topic: /camera/left/image_raw type: sensor_msgs/Image count: 12000 frequency: 27.8 - topic: /imu/data_raw type: sensor_msgs/Imu count: 24000 frequency: 55.6

重点看三个东西：duration、topics列表、compression。duration告诉你数据总时长，topics列表告诉你有没有你关心的数据，compression则直接决定后续解析时的资源开销。这里提一句，很多我们采集的高帧率图像bag会被压缩成bz2格式，处理时CPU占用会高不少，后面我会专门讲怎么应对。

2.2 查看 message 类型的真实定义

有时候你拿到一个陌生类型，比如custom_msgs/PointCloudArray，光看名字不知道里面字段结构，用rosmsg show查看：

rosmsg show custom_msgs/PointCloudArray

输出字段名和类型，比如：

std_msgs/Header header uint32 seq time stamp string frame_id int32 width custom_msgs/Point3D[] points float32 x float32 y float32 z

这里有个高阶技巧：如果bag包使用的消息类型和你当前ROS环境下的定义不一致（常见于跨机器、跨版本拷数据），解析时可能报rosbag.bag.ROSBagException之类的错误。遇到这种情况不要慌，先检查rosmsg version，必要时用rosbag reindex重建索引再试。

2.3 快速统计每个topic的消息频率和帧数

rosbag info已经给出了频率，但频率是平均值的概念。真实采集场景中，因为CPU调度、网络抖动、传感器丢包，瞬时频率可能忽高忽低。想了解真实分布，我常用一个小命令组合，把每个topic的时间戳间隔统计出来：

rosbag info bagfile.bag | grep topics -A 100

这只是粗看。更靠谱的是用Python遍历，统计标准差和p99值，这个我在第4章会给完整脚本。这里先记着结论：如果某个topic的平均频率只有标称频率的一半，那大概率是采集时存在丢帧，后续做算法评测要小心，别把丢帧当算法问题。

3. 命令行直接抽取：最常用的5个场景

很多需求根本不用写代码，ROS自带的命令行工具就能解决。这里我按频率从高到低列一下实际工作中最常用的操作。

3.1 按时间范围裁剪bag包

采集的bag包经常一录就是十几分钟，但实际有效段落只有几十秒。用rosbag filter就能裁剪出目标时间窗口：

rosbag filter input.bag output.bag "t.to_sec() >= 1715502601.5 and t.to_sec() <= 1715502663.2"

这里的t是rospy.Time对象，to_sec()返回浮点秒数。如果你不想算绝对时间戳，可以用相对时间：

rosbag filter input.bag output.bag "t.to_sec() >= 100.0 and t.to_sec() <= 163.0"

第二种写法里的t.to_sec()，注意依然是绝对时间戳，只是数值恰好从100开始。如果你想用相对bag起始时间，更稳妥的方式是先看rosbag info -y里的start时间，然后手动换算成绝对秒数。我经常看到有人想当然地用“相对秒数”直接写入条件，结果过滤出来是空包，排查半天发现是时间基准理解错了。这里再提醒一句：bag内的时间戳是自1970年1月1日以来的Unix时间戳（秒），filter里的t就是这个绝对时间，跟bag里start字段显示的时间是同一个基准。

3.2 用rosbag filter精确过滤指定topic

不光能按时间过滤，还能按topic名过滤，这是提取子数据的常用手段：

rosbag filter input.bag image_only.bag "topic == '/camera/left/image_raw'"

多个topic用Python的布尔表达式组合：

rosbag filter input.bag imu_odom.bag "topic in ['/imu/data_raw', '/odom']"

这里注意topic是_rosbag内部提供的变量，直接可用。支持and、or、not等运算符，完全可以写出复杂过滤条件，比如“只保留图像topic中时间处于前200秒的数据”：

rosbag filter input.bag part.bag "topic in ['/camera/left/image_raw'] and t.to_sec() >= start_sec and t.to_sec() <= start_sec + 200.0"

其中start_sec需要你从info输出里拿到具体数值写死。这种组合过滤在数据清洗阶段非常常用，建议熟练掌握。

3.3 压缩与解压bag包

压缩通常是为了节省磁盘空间，尤其图像数据量动辄几十个GB。命令行压缩：

rosbag compress --bz2 input.bag

会生成input.bag.bz2，原文件保留。解压则用：

rosbag decompress input.bag.bz2

这里有个关键决策点：如果你只是做顺序回放，用压缩格式完全没毛病。但如果要随机跳转解析多个topic（尤其是组合着读图像和IMU做时间同步），压缩后的bag解压开销非常可观，我实测一个5.3GB的bz2 bag，每次rosbag play启动时CPU占用直接拉满，加载都要好几分钟。所以我的习惯是：采集完先归档为压缩格式，但做算法分析前一律解压，或者用Python直接读时指定read()参数关闭压缩解压的重复劳动。

这里补充一个 Python 读取压缩bag的正确姿势，后面章节会详细讲，先给结论：

open(bag_path, 'r', compression='none')

compression='none'只是告诉rosbag你不需要自动将消息解压成原始序列化对象，而实际上读取时内部仍会自动解压。如果你真的想避免解压开销，最彻底的做法还是先rosbag decompress。

3.4 批量操作多个bag文件

日常巡检或做数据集整理时，经常要对几十个bag执行同样操作。用shell的for循环可以轻松搞定：

for f in *.bag; do echo "processing $f" rosbag info "$f" | grep -E 'duration|topics' done

更复杂的需求建议用Python的glob批量处理，后面有脚本。

3.5 用rosbag play回放时控制速率和循环

虽然回放不算严格意义的解析，但调试时需要反复确认行为，几个参数很实用：

rosbag play --rate 0.5 --loop input.bag

--rate 0.5以半倍速回放，--loop循环播放。还可以指定只播放某些topic：

rosbag play --topics /imu/data_raw /odom input.bag

注意，--topics参数在老版本和新版本中写法略有不同，有的版本是-t紧跟topic列表，有的需要逗号分隔。建议先rosbag play --help确认当前版本语法。

4. Python脚本解析bag包：灵活提取与转换

命令行适合快速调度，遇到复杂数据处理还是要写Python。ROS提供了rosbag这个库，配合genpy的消息对象，你几乎能做任何事。

4.1 安装与基础读取

确认环境已经安装ros-<distro>-rosbag（比如ros-noetic-rosbag、ros-humble-rosbag）。Python侧使用：

import rosbag bag = rosbag.Bag('input.bag', 'r') for topic, msg, t in bag.read_messages(topics=['/imu/data_raw']): print(t.to_sec(), msg.linear_acceleration.x, msg.linear_acceleration.y, msg.linear_acceleration.z) bag.close()

read_messages()返回三要素：topic（字符串）、msg（消息对象）、t（时间戳）。时间戳t是rospy.Time，注意它和msg.header.stamp是两个概念：t是bag记录该消息的到达时间，msg.header.stamp是传感器数据本身的时间。二者在正常情况下应该非常接近，如果差异过大，说明传感器时间同步有问题，这个观察在分析多传感器数据时特别有用。

4.2 提取全部topic统计信息

写个通用函数，输出每个topic的消息数、起始时间、终止时间和平均频率：

import rosbag from collections import defaultdict def bag_stat(bag_path): stats = defaultdict(lambda: {'count': 0, 'first': None, 'last': None, 'durations': []}) with rosbag.Bag(bag_path, 'r') as bag: for topic, msg, t in bag.read_messages(): s = stats[topic] s['count'] += 1 if s['first'] is None: s['first'] = t s['last'] = t print(f"{'topic':<50} {'count':>8} {'duration':>12} {'freq':>8}") for topic, s in sorted(stats.items()): if s['count'] > 1: duration_ms = (s['last'] - s['first']).to_sec() * 1000.0 freq_hz = (s['count'] - 1) / max((s['last'] - s['first']).to_sec(), 1e-9) print(f"{topic:<50} {s['count']:>8} {duration_ms:>10.1f}ms {freq_hz:>8.1f}") else: print(f"{topic:<50} {s['count']:>8} {'N/A':>12} {'N/A':>8}") bag_stat('input.bag')

这个脚本迭代所有topic信息，特别适合大bag总览，比命令行输出的信息更结构化。

4.3 按时间戳提取数据段并另存为新bag

如果只是想提取某个时间段内的某几个topic，但又想保留ROS消息格式，可以用Python写新bag：

import rosbag with rosbag.Bag('input.bag', 'r') as source: with rosbag.Bag('sub.bag', 'w', compression='bz2') as out: start = 10.0 # 相对起始时间的秒数 end = 20.0 first_ts = None for topic, msg, t in source.read_messages(topics=['/camera/right/image_raw', '/imu/data_raw']): if first_ts is None: first_ts = t cur_ms = (t - first_ts).to_sec() if start <= cur_ms <= end: out.write(topic, msg, t)

注意：新bag写入时，write(topic, msg, t)的参数顺序是topic、msg、time，别写错。压缩参数compression='bz2'可选，推荐在数据量大时使用。

4.4 提取图像序列并生成视频

这是最高频的需求之一。从bag里抠出图像帧，保存为图片序列，再转成视频。流程分两步走。

先用Python批量导出图片，这里用OpenCV帮助保存：

import rosbag import cv2 from sensor_msgs.msg import Image as SensorImage from cv_bridge import CvBridge import os bridge = CvBridge() os.makedirs('frames_left', exist_ok=True) with rosbag.Bag('input.bag', 'r') as bag: idx = 0 for topic, msg, t in bag.read_messages(topics=['/camera/left/image_raw']): try: cv_img = bridge.imgmsg_to_cv2(msg, desired_encoding='bgr8') filename = f'frames_left/{idx:06d}_{msg.header.stamp.secs}_{msg.header.stamp.nsecs}.jpg' cv2.imwrite(filename, cv_img) idx += 1 except Exception as e: print(f"frame {idx} error: {e}")

这段代码有几个值得注意的点：

• msg.header.stamp.secs和msg.header.stamp.nsecs组成原始时间戳，保存到文件名里，方便后续抽取特定时刻的帧。
• CvBridge依赖cv_bridge包，在ROS环境中一般自带。如果你在的发行版是ROS 2（Humble以后），可以改用image_transport和cv_bridge的对应版本，接口稍有差别，但思路一致。
• 图像topic如果是压缩传输类型（比如sensor_msgs/CompressedImage），不能直接转bgr8，需要先解码：

from sensor_msgs.msg import CompressedImage if topic.endswith('/compressed'): np_arr = np.frombuffer(msg.data, np.uint8) cv_img = cv2.imdecode(np_arr, cv2.IMREAD_COLOR)

如果不想额外装解码库，也可以直接用image_transport的解码方法，更省心。

导出图片后，用FFmpeg合成视频，这是我最常用的命令：

ffmpeg -framerate 30 -i frames_left/%06d.jpg -c:v libx264 -pix_fmt yuv420p left_video.mp4

-framerate要和图像topic实际频率匹配，否则视频快慢速度不对。我们有时候为了省事，直接设成与topic频率一致即可。如果想无损保存，可以改用-c:v ffv1或直接输出成图像序列，别压缩。视频codec选择libx264是通用兼容性最好的。

4.5 将激光雷达点云数据转为PCD文件

这个需求常见于激光SLAM的数据处理。从sensor_msgs/PointCloud2中导出为PCD：

import rosbag import numpy as np import open3d as o3d with rosbag.Bag('lidar.bag', 'r') as bag: for topic, msg, t in bag.read_messages(topics=['/points']): points = [] for p in msg.points: # 如果用sensor_msgs/PointCloud，可直接用points列表 points.append([p.x, p.y, p.z]) if len(points) == 0: continue pcd = o3d.geometry.PointCloud() pcd.points = o3d.utility.Vector3dVector(np.array(points)) o3d.io.write_point_cloud(f"points_{t.to_nsec()}.pcd", pcd)

注意：sensor_msgs/PointCloud2是扁平化二进制结构，直接用msg.points会把所有字段解析为二维点数组，一般不推荐。正确做法是解析里面的fields定义，提取x、y、z。这里为了演示给sensor_msgs/PointCloud（旧版）用，实际现在大多用PointCloud2，就需要用到point_cloud2模块，类似：

from sensor_msgs import point_cloud2 for p in point_cloud2.read_points(msg, field_names=['x','y','z'], skip_nans=True): points.append(list(p))

如果包体积大，内存不够，采用流式写PCD，分块写入，尽量避免一次性把所有点读入内存。

4.6 遍历bag时处理异常消息

bag包数据难免损坏，比如在设备异常断电后拷贝出来的。Python读取时，read_messages()默认遇到坏数据会抛出异常导致中止。想要跳过坏消息继续遍历，可以用r.read_messages(return_unknown=True)，或者手工捕获异常。更稳妥的是用rosbag reindex在读取前重建索引：

rosbag reindex bad.bag

这会生成bad.bag.active之类的临时文件，如果异常严重可能重建失败。我的建议是：先把坏bag单独隔离，能修则修，不能修就直接放弃该bag，别浪费太多时间。

5. 核心场景深度实战：图像与IMU的时间同步问题

很多真实应用需要融合图像和IMU（比如视觉惯性里程计），这时就面临一个经典难题：bag包中不同topic的时间戳是对齐的吗，频率是否匹配？这里给一套实战排查与处理方案。

5.1 时间对齐的检查方法

用Python读取两个topic，打印时间戳差值：

import rosbag with rosbag.Bag('data.bag', 'r') as bag: imu_ts = None img_ts = None diff_list = [] for topic, msg, t in bag.read_messages(topics=['/imu/data_raw', '/camera/left/image_raw']): if topic == '/imu/data_raw': imu_ts = msg.header.stamp elif topic == '/camera/left/image_raw': img_ts = msg.header.stamp if imu_ts is not None: diff = (img_ts - imu_ts).to_sec() diff_list.append(diff) if len(diff_list) % 100 == 0: print(f"subject {len(diff_list)} diffs: {diff*1e3:.3f} ms")

如果diff_list的均值在几个毫秒内且波动稳定，说明时间同步基本正常。如果出现规律性几百毫秒的漂移，就要检查传感器硬件授时或驱动是否有问题。

5.2 提取指定时间戳前后各一帧图像

比如需要取某时刻t0前后的图像，用于时间戳插值或者对齐。思路是记录每个图像时间戳，然后二分离散查找：

import bisect image_ts = [] image_data = [] # 保存消息对象或路径 with rosbag.Bag('data.bag', 'r') as bag: for topic, msg, t in bag.read_messages(topics=['/camera/left/image_raw']): image_ts.append(t.to_nsec()) image_data.append(msg) def get_frames_at(t_nsec): idx = bisect.bisect_left(image_ts, t_nsec) before = image_data[idx-1] if idx > 0 else None after = image_data[idx] if idx < len(image_data) else None return before, after

注意内存压力：如果bag有几万帧图像，直接image_data.append(msg)会把消息都保留在内存，可能占用几个GB。更好的方案是只在内存里保留时间戳列表，真正的图像从bag里二次读取：

with rosbag.Bag('data.bag', 'r') as bag: for topic, msg, t in bag.read_messages(topics=['/camera/left/image_raw']): if t.to_nsec() >= start_nsec and t.to_nsec() <= target_nsec + margin: if abs(t.to_nsec() - target_nsec) < best_diff: best_diff = abs(t.to_nsec() - target_nsec) best_msg = msg

这种是全量线性扫描，对于只需要少量关键帧的场景足够快，反正大多数情况下我们只是拿个别帧做可视化检查。

5.3 生成对齐后的数据文件

做时间同步后，我们经常把两个topic的数据按照同步时间输出成CSV，方便离线算法使用：

import rosbag import csv from collections import deque deq_imu = deque() deq_img = deque() # 先读IMU，再读图像并不现实，因为bag是按时间排序的，所以使用两个deque分别缓存

对这个问题，更简单的方案是利用read_messages()的顺序特性：bag内部是按时间戳排序的，所以全部消息是按全局时间顺序排列的。因此我们可以维护一个滑动窗口：

import rosbag, csv window = [] # (topic,t,msg) with rosbag.Bag('data.bag', 'r') as bag: for topic, msg, t in bag.read_messages(topics=['/imu/data_raw', '/camera/left/image_raw']): window.append((topic, t, msg)) # 当窗口里有图像时，寻找最近的IMU消息，生成一行 while len(window) > 1 and window[0][0] != '/camera/left/image_raw': window.popleft() # 丢弃太旧的IMU if topic == '/camera/left/image_raw': cur_img = (topic, t, msg) best_imu = min([x for x in window if x[0] == '/imu/data_raw'], key=lambda x: abs((x[1]-t).to_sec())) # 写入csv

这种窗口方法逻辑清晰，内存占用小，是工程中最常用的做法。

6. 可视化排查：用工具快速定位数据问题

除了代码，图形化工具能让很多问题一目了然。

6.1 使用 rqt_bag 可视化

直接运行：

rqt_bag bagfile.bag

可以浏览topic列表，点击绘图插件能查看数值变化曲线。比如检查IMU的加速度计数值是否有跳变，遥感数据是否有毛刺，非常直观。我经常用它快速判断采集数据质量：如果加速度曲线出现大量脉冲式跳动，多半是震动或者传感器故障。

6.2 用 rosbag_fox 或 plotjuggler 分析

现在的社区工具越来越丰富，我特别推荐plotjuggler，它可以直接导入bag包，把任意topic的曲线绘制出来，并且支持时间轴对齐、缩放、数学运算。安装方式各发行版不同，通常：

sudo apt install ros-$ROS_DISTRO-plotjuggler

启动后，选择“ROS”标签，加载bag文件，然后拖拽topic字段到绘图窗口。最方便的用途是检查时间同步是否有延迟差：把两个topic的header.stamp时间绘制在一起，肉眼可见它们的对齐情况。

6.3 检查TF树

在很多机器人系统中，坐标变换（TF）是最容易出问题的一环。从bag中直接回放TF消息并检查变换：

rosbag play bagfile.bag --topics /tf /tf_static & rosrun tf view_frames

或者生成PDF视图列表。如果你想在解析时直接读TF消息并打印变换内容，可以用以下代码：

import rosbag with rosbag.Bag('tf.bag', 'r') as bag: for topic, msg, t in bag.read_messages(topics=['/tf']): for transform in msg.transforms: frame_from = transform.header.frame_id frame_to = transform.child_frame_id trans = transform.transform.translation print(f"{t.to_sec():.3f}: {frame_from} -> {frame_to}, x={trans.x:.3f}, y={trans.y:.3f}, z={trans.z:.3f}")

检查要点：看是否有非连续的跳变，帧ID是否合理，平移和旋转数值是否在合理范围。经常遇到的情况是：bag包中frame_id全是空字符串，这会导致后续SLAM初始化失败。所以解析TF时顺便打印frame_id是很有必要的。

7. 常见问题与避坑技巧速查

从命令行到脚本，我在实际中踩过不少坑，这里整理成一张速查表，方便你直接对照。

8. 进阶思路：解析效率优化与批量处理框架

当你处理成百上千个bag包时，逐包循环 + 逐topic遍历的方式会成为瓶颈。我的优化思路是：先建立一个索引文件，把所有bag包的topic、时间范围、消息数、存储路径存成一张表，之后按需求直接读取对应切片。

8.1 建立索引表

import rosbag, json, glob index_list = [] for bag_file in sorted(glob.glob('/data/*.bag')): try: with rosbag.Bag(bag_file, 'r') as bag: info = { 'path': bag_file, 'start_time': bag.get_start_time(), 'end_time': bag.get_end_time(), 'topics': list(bag.get_type_and_topic_info().first[0].keys()), 'message_counts': {k: v[1] for k, v in bag.get_type_and_topic_info().first[1].items()}, } index_list.append(info) except Exception as e: print(f"failed to index {bag_file}: {e}") with open('bag_index.json', 'w') as f: json.dump(index_list, f, indent=2, default=str)

建立后，你就能用Python直接查“哪个bag包含 /camera/left/image_raw 且时间范围在 X 到 Y 之间”，无需逐个open解析。

8.2 并行提取不同bag的相同topic

由于不同bag文件之间互不依赖，可以用multiprocessing或者简单的线程池并行解析：

from concurrent.futures import ProcessPoolExecutor def process_one(bag_path): # 从bag提取图像并保存 ... return bag_path, success with ProcessPoolExecutor(max_workers=4) as ex: futures = [ex.submit(process_one, p) for p in bag_paths]

注意：进程池在ROS消息对象传递上有坑——消息对象一般不是pickle友好的，所以不要在任务间传递消息对象，而是让每个进程自己打开bag文件，这完全可行。

8.3 使用rosbag read时限制内存占用的技巧

如果遍历一个超大bag时内存上涨很快，很可能是你无意识中把所有消息的引用保存在列表中了。解决方案是：要么只保留时间戳和需要的小字段（提取后释放消息对象），要么直接流式处理，避免在循环体内用全局列表累积。用del手动释放是一种笨办法，最好的办法是拆分处理：利用read_messages(start_time=..., end_time=...)只读取需要的片段。

read_messages本身支持切片参数：

for topic, msg, t in bag.read_messages(topics=['/image_raw'], start_time=rospy.Time.from_sec(start_sec), end_time=rospy.Time.from_sec(end_sec)):

这个切片操作并不是真正在磁盘上做二分查找，而是内部扫描并跳过不在区间内的消息，但仍然比全量解析快很多，也省内存。

9. 送给新手的三条实践心得

最后说几句经验之谈，也是我踩过不少坑之后才体会到的。

第一，解析bag包永远不要只依赖一种工具。命令行适合快速验证，Python脚本适合深度定制，rqt_bag和plotjuggler适合可视化检查，三者结合使用，效率远高于单条路走到底。

第二，拿到陌生bag的第一件事不是直接play，而是info。我见过太多同学一上来就播放，结果播放半天发现不是要找的topic，白费功夫。先花一分钟看信息，能省半小时。

第三，平时做数据采集时养成好习惯：用rosbag record -a记录all topic，最好同时记录/tf和/tf_static，这样才能保证后续分析时有完整的坐标变换链。另外，文件名里加上日期时间，归档时写注释文件，三年后再翻出来还能知道这包数据是什么场景，比什么都强。

我自己的习惯是，每个数据目录放一个README.md，写明采集时间、里程、传感器型号、bag包含的topic列表，以及已知的问题。这样做最直接的好处是：两个月后别的同学来问你某个bag里有什么时，你可以直接甩给他README，而不是重新解析一遍。

解析bag包这个技能，看起来基础，实际上贯穿整个机器人开发周期。数据采集、算法测试、问题复现、技术分享，哪一个环节都离不开它。你现在花半小时把这些命令和脚本跑一遍，后面会节省你无数个加班夜。