每次拿到.bag文件,我的第一反应都是“又要装ROS才能看了吗”。尤其是有时候别人发来一个VINS Fusion的bag文件,我只是想提取里面的图像和IMU数据做标定,为这点事去装一个完整的ROS环境实在不划算。如果你也遇到类似的情况——主力机器是Windows,或者手头只有一台没装过ROS的Linux服务器,或者你只是单纯想把bag数据拉出来喂给Python做分析——那这篇文章就是给你准备的。
标题里“python 解析bag文件(不用安装ros系统)”这句话,听起来像是个绕过了什么大坑的偏门技巧,实际上它并不是什么hack,而是bag文件本身的格式并没有绑定ROS。只要搞清楚它的记录结构,配合合适的纯Python库,就能直接把图像、点云、IMU、里程计这些话题数据从bag里读取并落地成jpeg、csv、npy这些日常格式。这篇文章会从bag的文件结构讲起,再对比几种可行的解析方案,最后用一份可以复制的完整代码演示怎么把VINS数据集里的图像和IMU数据提取出来,顺带把我在实际解析中踩过的坑也一并说了。
1. 为什么解析bag文件非得绕开ROS系统
1.1 bag文件不是只能被ROS读的“亲儿子”
很多第一次接触bag文件的人会误以为它是某种加密或私有格式,离开了ROS就没法读。其实ROS bag就是一个带有固定魔数开头的二进制容器文件,类似SQLite或者Zip,它的内容是一段接一段的记录(Record),每条记录里保存着元数据或者消息本体。ROS只是这个格式最常用的生产者,不代表它必须是唯一消费者。
bag文件的物理结构可以粗略看成这样:文件开头固定是#ROSBAG V2.0\n这一行魔法字符串,后面跟着若干条记录。每条记录由header和data组成,header里面会标记当前记录的类型(比如连接信息、消息数据、索引等),data区域则是具体的二进制内容。也就是说,只要你愿意照着格式去解析,完全可以用Python标准库struct逐字节拆开。当然,工程上没必要这么原始,因为社区已经有人把这件事做成了纯Python库。
我之所以强调这一点,是因为很多教程一上来就让你装ROS,导致大家默认“解析bag就必须先有ROS”。但实际上bin格式、hd5、甚至视频流都不存在这种依赖,bag也一样。理解这个底层事实,你才能真正接受“不装ROS也能解析”这件事。
1.2 不装ROS的实际好处:从Windows到云端都可行
先说我个人的实际感受。ROS1的官方支持重心在Ubuntu,Windows上跑ROS要么用WSL,要么用虚拟机,折腾网络和图形界面就得半天;macOS就更不用说了,官方根本不做支持。如果你只是拿到一个bag文件想看看里面有什么,为这个去装一套Ubuntu系统或者引入Docker镜像,产线投入明显不成比例。
再说服务器和云端场景。我做数据处理时经常要把bag文件丢到一台只有Python运行时的容器里跑批,这个容器不能也不应该引入整个ROS系统。一旦引入ROS,不仅镜像体积暴涨,还会带来一堆默认依赖和系统库冲突。用纯Python库解析bag,只需要pip安装几个包,在容器、CI、函数计算这些环境里都能跑,这才是离线数据分析该有的样子。
最后,Python生态本身也很配合。numpy、pandas、matplotlib这些数据处理库,和解析出来的消息数据可以无缝衔接。我从bag里拿到的IMU数据可以直接塞进DataFrame做统计,图像数据可以直接转成numpy array喂给模型,中间不需要任何ROS桥接。这种轻量链路,在实际工程里非常舒服。
1.3 纯Python解析的几种路线
既然决定不装ROS,那实现路径大概有三类:
- 官方
rosbag库:本质上依赖ROS环境,不符合我们的前提,只能在已经有ROS的环境里用Python调用,不能算“不用安装ROS系统”。 - 第三方纯Python库:比如
rosbags、bagpy。它们自己实现了bag格式解析和消息反序列化,pip安装后即可使用,不需要任何ROS程序运行环境。 - 自己写解析器:用
struct或numpy从文件里按记录头解析,适合学习和调试,但工程效率太低,遇到压缩的chunk还要自己处理lz4/bz2解压。
这三条路线里,最符合标题场景、也最适合大多数人用的是第二条。后面我会详细展开为什么rosbags库值得作为主力,以及它和bagpy的区别。
2. 先看清bag文件结构,再选择解析方案
2.1 bag文件的Records结构与Chunk压缩原理
为了选对解析方案,最好先了解bag文件在底层是怎么组织的。bag文件里的记录主要有这么几类:
| 记录类型 | 含义 |
|---|---|
| BagHeader | 文件级元数据,包含索引位置等信息 |
| Chunk | 一段连续的消息数据集合,可能带压缩 |
| Connection | 某个话题的连接信息,包括话题名、消息类型、消息定义 |
| MessageData | 一条具体的序列化消息数据 |
| IndexData | 索引记录,标识某个连接的消息在文件中的位置 |
| ChunkInfo | 某个Chunk内的连接和消息数量统计 |
平时ROS录制数据时,消息不会一条一条平铺在文件里,而是攒成一个个Chunk再写入。Chunk可以选择不压缩,也可以选择bz2或lz4压缩。这就是为什么有些bag文件特别大、有些则相对小,也是后面你会遇到“明明能读消息却报错”的根源之一。
Connection记录是解析时最需要重视的部分。它里面保存着话题名、消息类型,以及消息定义的完整文本。有了消息定义,解析器才能把二进制数据反序列化成Python对象。这也解释了为什么rosbags能脱离ROS环境工作——它并不是魔法,而是把ROS的消息定义文件内置到了库里,按定义去解析原始字节。
2.2 三个可用的纯Python解析方案对比
我实际用过的纯Python方案主要有三个:
rosbags:目前维护最活跃的纯Python bag解析库,支持ROS1和ROS2,官方命名为rosbags。bagpy:也是pip可安装的非ROS库,接口仿照rosbag,读常用的图像和IMU消息没问题,但更新频率低,对压缩bag的支持稍弱。- 自己用
struct解析:不推荐作为日常方案,但可用来理解格式。
下面这张表是它们在几个维度上的直观对比:
| 方案 | 是否需要ROS | 支持ROS1/Ros2 | 压缩支持 | 维护活跃度 | 上手难度 |
|---|---|---|---|---|---|
| rosbags | 不需要 | 都支持 | lz4/bz2 | 高 | 低 |
| bagpy | 不需要 | 仅ROS1 | 有限 | 中 | 低 |
| 自写parser | 不需要 | 取决于自己实现 | 需自己处理 | 无 | 较高 |
从表格看,rosbags几乎在每项上都占优势。这也是我现在优先推荐它的原因。
2.3 为什么我主力推荐rosbags库
只说“维护活跃”太抽象,我说几个具体的点。
第一,rosbags提供了高层API和底层API两层接口。高层API里有个AnyReader,它能自动识别你给的是ROS1 bag还是ROS2 bag,省去手动判断格式的麻烦;底层API则暴露了Reader、Writer等类,适合做格式转换或者深度定制。第二,它默认内置了常见消息类型的类型系统,包括sensor_msgs/Image、sensor_msgs/PointCloud2、tf2_msgs/TFMessage等,不需要你自己维护消息定义文件。第三,它的消息遍历是流式读取的,不是把所有消息一次性载入内存,所以处理几十GB的bag也没问题。第四,它还带命令行工具,可以用来查看bag信息和做bag格式转换。
我实测过几个100GB左右的数据集,用rosbags遍历所有连接的消息,速度完全可以接受,内存占用也稳定在一个很低的水平。这种表现足够支撑日常工作。
3. 实操:用rosbags提取VINS bag中的图像和IMU数据
3.1 环境安装与元数据读取
先安装依赖:
pip install rosbags numpy opencv-pythonrosbags会自动安装lz4这些解压依赖,opencv用于图像保存,numpy用于把图像原始缓冲区转成数组。
写第一个脚本看看bag里到底有哪些话题:
from pathlib import Path from rosbags.highlevel import AnyReader from rosbags.typesys import Stores, get_typestore typestore = get_typestore(Stores.ROS1_NOETIC) with AnyReader(['vins.bag'], default_typestore=typestore) as reader: for connection in reader.connections: print(connection.topic, connection.msgtype, connection.msgcount)这段代码会输出类似下面这样的信息:
/cam0/image_raw sensor_msgs/msg/Image 1217 /cam1/image_raw sensor_msgs/msg/Image 1217 /imu sensor_msgs/msg/Imu 6111 /vins_estimator/odometry nav_msgs/msg/Odometry 1205看到这些输出,你就能判断自己需要哪些话题了。顺便提一句,如果你的bag是ROS2的,只需要把get_typestore(Stores.ROS1_NOETIC)换成对应的ROS2 typestore,比如Stores.ROS2_HUMBLE,但AnyReader也会自动处理。这里显式指定typestore主要为了在反序列化时能正确识别ROS1的消息结构。
3.2 图像消息落地为JPEG/视频
以VINS数据集常见的/cam0/image_raw话题为例,下面的代码会在当前目录创建frames文件夹,把每一帧图像写成JPEG,同时用第一帧的宽高初始化一个视频写入器:
from pathlib import Path import cv2 import numpy as np from rosbags.highlevel import AnyReader from rosbags.typesys import Stores, get_typestore typestore = get_typestore(Stores.ROS1_NOETIC) out_dir = Path('frames') out_dir.mkdir(exist_ok=True) video_writer = None frame_idx = 0 with AnyReader(['vins.bag'], default_typestore=typestore) as reader: connections = [c for c in reader.connections if c.topic == '/cam0/image_raw'] if not connections: raise RuntimeError('没有找到 /cam0/image_raw 话题') for connection, timestamp, rawdata in reader.messages(connections=connections): msg = reader.deserialize(rawdata, connection.msgtype) # 根据编码方式转换图像 if msg.encoding == 'rgb8': img = np.frombuffer(msg.data, dtype=np.uint8).reshape(msg.height, msg.width, 3) img_bgr = cv2.cvtColor(img, cv2.COLOR_RGB2BGR) elif msg.encoding == 'bgr8': img_bgr = np.frombuffer(msg.data, dtype=np.uint8).reshape(msg.height, msg.width, 3) elif msg.encoding == 'mono8': img_bgr = np.frombuffer(msg.data, dtype=np.uint8).reshape(msg.height, msg.width, 1) else: print(f'跳过不支持的编码: {msg.encoding}') continue cv2.imwrite(str(out_dir / f'frame_{frame_idx:06d}.jpg'), img_bgr) if video_writer is None: video_writer = cv2.VideoWriter( 'output_video.avi', cv2.VideoWriter_fourcc(*'MJPG'), 20.0, (msg.width, msg.height) ) video_writer.write(img_bgr) frame_idx += 1 if video_writer is not None: video_writer.release() print(f'已导出 {frame_idx} 帧图像')这里有个细节:ROS里稳定流传的消息有两种,一种是sensor_msgs/Image(未压缩的原始图像),另一种是sensor_msgs/CompressedImage(JPEG或PNG压缩后的字节流)。上面脚本针对的是前者。如果你拿到的是CompressedImage,解析方式会简单很多,直接用cv2.imdecode就可以,不需要关心encoding。实际处理前先确认bag里是哪种类型,代码路径完全不同。
3.3 IMU消息导出为CSV
IMU数据相比图像简单很多,它只是一组带时间戳的线性加速度和角速度。下面的代码会把/imu话题组织成CSV,方便后续用pandas分析:
import csv from rosbags.highlevel import AnyReader from rosbags.typesys import Stores, get_typestore typestore = get_typestore(Stores.ROS1_NOETIC) with AnyReader(['vins.bag'], default_typestore=typestore) as reader: connections = [c for c in reader.connections if c.topic == '/imu'] if not connections: raise RuntimeError('没有找到 /imu 话题') with open('imu.csv', 'w', newline='') as f: writer = csv.writer(f) writer.writerow(['recv_time_ns', 'sec', 'nsec', 'ax', 'ay', 'az', 'gx', 'gy', 'gz']) for connection, timestamp, rawdata in reader.messages(connections=connections): msg = reader.deserialize(rawdata, connection.msgtype) stamp = msg.header.stamp ax = msg.linear_acceleration.x ay = msg.linear_acceleration.y az = msg.linear_acceleration.z gx = msg.angular_velocity.x gy = msg.angular_velocity.y gz = msg.angular_velocity.z writer.writerow([timestamp, stamp.sec, stamp.nanosec, ax, ay, az, gx, gy, gz]) print('IMU数据已写入 imu.csv')这里我顺便把两条时间信息都导出了:recv_time_ns是bag记录时系统接收消息的时间戳,sec/nsec是消息内部header.stamp的采集时间戳。两者用途不同,在第4部分我会专门说明。
3.4 跑通后的验收与自检清单
脚本能跑通不代表数据是正确的。我通常按下面这个清单自检:
- 导出的图像数量与
connection.msgcount一致,并且帧号连续。 - 随机抽几帧JPEG打开检查是否模糊、是否有绿边或花屏。
- CSV文件用
pd.read_csv('imu.csv')能正常加载,行数正确。 - 对比
recv_time_ns和sec/nsec,如果差距很大,要考虑是否有时钟同步问题。 - 如果需要做视觉惯性对齐,检查图像时间戳与IMU时间戳是否有合理的重叠区间。
这一步看似繁琐,但能避免后面拿着坏数据跑流程。
4. 解析过程中最容易踩的5个坑及对应解法
4.1 压缩bag的解压依赖问题
很多公开数据集发布的bag是压缩过的,Chunk区域是lz4或bz2压缩。如果你用早期的bagpy或者自己写解析器,经常会遇到解压失败的问题。rosbags本身会依赖lz4库,正常情况下pip安装就会带上,但如果你在极其精简的环境里,可能仍然缺库,运行时报类似ModuleNotFoundError: No module named 'lz4'的错误。
解决办法很简单,单独执行pip install lz4。真正要注意的不是安装,而是确认你手里的bag到底压缩没有。可以在读取前用十六进制工具打开文件,搜一下lz4或者bz2关键字,也可以直接用rosbags的低层API去读header。如果你要写一个通用工具包给团队用,建议在文档里明确要求安装lz4,避免环境不一致导致现场踩坑。
4.2 header时间戳与接收时间戳的分工
这是我在处理SLAM数据时被坑过最惨的一次。bag里每条消息除了自身带的时间戳,还有一个由录制节点分配的时间戳,也就是reader.messages()返回的timestamp。前者是传感器采集时刻或者算法发布时刻,后者是bag写入时刻。两者在实时系统里一般差距很小,但如果录制时电脑负载高、网络存在延迟,它们的顺序就可能出现错位。
做传感器融合时,千万不要混用这两种时间。图像和IMU的对齐应该用header.stamp,因为它代表数据本身的采集时刻;而如果你需要还原消息在bag里的录制顺序,或者分析回放延迟,才看timestamp。如果你发现同一帧的两种时间戳相差上百毫秒,说明录制环境本身就有问题,需要提前处理。
4.3 图像编码RGB/BGR陷阱
直接用numpy把图像数据reshape成数组并保存,最容易出现颜色错乱。报错往往没有,但存出来的图片红色和蓝色对调了。原因很简单:ROS图像消息的encoding字段告诉我们通道顺序,比如bgr8、rgb8、mono8、16UC1,而OpenCV默认是BGR排列。如果你把rgb8的数据直接当成BGR送入cv2.imwrite,就会看到红蓝互换。
我习惯在解析函数里做一次编码分支处理:遇到rgb8先转成bgr8,遇到bgr8直接用,遇到mono8保持单通道。处理16位深度图时还要额外考虑数据类型是uint16而非uint8,如果直接reshape成uint8,图像亮度会变得一片灰或者有条纹。只要把编码和dtype对应上,这类问题就可以完全避免。
4.4 PointCloud2点云的字段解析
PointCloud2是bag里另一个高频消息,但它不像Image那样有固定的宽高和通道顺序,它把每个点的所有字段(x/y/z/intensity/rgb等)打包在一个字节流里,并用fields列表描述每个字段的名称、偏移量、数据类型。这就意味着你不能想当然地认为“点云就是三个float32排在前面”,必须按字段定义去解析。
实际解析时,我会先遍历msg.fields,找到x、y、z这些字段的offset,然后用np.frombuffer(msg.data, dtype=np.uint8).reshape(height, width, point_step)去按point_step切片。这在下一部分会给出完整代码。如果你忽略offset直接按固定布局读,一旦遇到有人用不同的点云拼接顺序,数据就是乱的。
4.5 大bag文件的内存与IO优化
处理几十GB的bag时,最容易犯的错误是一次性把消息读进内存。比如有人喜欢先收集所有消息再统一处理,结果内存直接爆炸。正确的做法是依赖流式API,也就是上面代码演示的方式:reader.messages()每次返回一条消息,你用一条存一条,读完就释放。这样无论bag多大,内存占用始终稳定。
另外,遍历时尽量用connections参数过滤话题,只读取你关心的连接,避免把雷达、tf、nav_msgs等不相关话题全部反序列化一遍。IO上如果数据要落盘,建议批量写,不要在循环里频繁print或者逐条commit CSV,否则速度会被拖慢不少。
5. 不装ROS还能继续做哪些事
5.1 手动解析PointCloud2的完整代码
这部分直接给出一个可复用的函数:
import numpy as np from rosbags.highlevel import AnyReader from rosbags.typesys import Stores, get_typestore # 根据 sensor_msgs/PointField 的 datatype 映射到 numpy 类型 FIELD_TYPE_MAP = { 1: np.int8, 2: np.uint8, 3: np.int16, 4: np.uint16, 5: np.int32, 6: np.uint32, 7: np.float32, 8: np.float64, } def pointcloud2_to_structured_array(msg): names = [] formats = [] offsets = [] for field in msg.fields: if field.datatype not in FIELD_TYPE_MAP: continue names.append(field.name) formats.append(FIELD_TYPE_MAP[field.datatype]) offsets.append(field.offset) dtype = np.dtype({ 'names': names, 'formats': formats, 'offsets': offsets, 'itemsize': msg.point_step, }) return np.frombuffer(msg.data, dtype=dtype, count=msg.width * msg.height) typestore = get_typestore(Stores.ROS1_NOETIC) with AnyReader(['vins.bag'], default_typestore=typestore) as reader: connections = [c for c in reader.connections if c.topic == '/points'] if not connections: raise RuntimeError('没有找到点云话题') for connection, timestamp, rawdata in reader.messages(connections=connections): msg = reader.deserialize(rawdata, connection.msgtype) points = pointcloud2_to_structured_array(msg) # 提取 xyz xyz = np.stack([points['x'], points['y'], points['z']], axis=-1) # 保存前100个点核对一下 print(xyz[:5]) break这段代码按point_step作为结构化数组的itemsize,保证了即使字段之间有padding也能对齐。如果你还需要颜色或者intensity,直接通过字段名称访问即可。
5.2 TF数据与其他消息类型的扩展
bag里常常还有TF变换树,用于描述各坐标系之间的位姿关系。ROS里TF消息的典型类型是tf2_msgs/msg/TFMessage。解析起来很简单:
if connection.msgtype == 'tf2_msgs/msg/TFMessage': msg = reader.deserialize(rawdata, connection.msgtype) for transform in msg.transforms: print(transform.header.frame_id, '->', transform.child_frame_id)消息里的平移是transform.translation.x/y/z,旋转是transform.rotation.x/y/z/w。拿到这些数据后,你可以自己构建坐标变换链,不需要TF树服务。其实只要理解了rosbags的反序列化机制,任何已知消息类型都能用同样的方式读取。唯一要注意的是自定义消息,这类消息需要额外导入消息定义,或者用rosbags的register_types方法注册,否则反序列化时会报未知类型。
5.3 把解析结果接入Pandas和可视化
解析只是第一步,真正有意思的是把数据用起来。IMU数据导出成CSV后,用pandas加载非常方便:
import pandas as pd df = pd.read_csv('imu.csv') # 把 time 转成相对秒 df['t_sec'] = df['sec'] + df['nsec'] * 1e-9 df['t_sec'] -= df['t_sec'].iloc[0] # 画个加速度曲线看看 df.plot(x='t_sec', y=['ax', 'ay', 'az'])如果你从bag里提取了Odometry消息,还可以直接画出轨迹。Odometry消息的位姿在msg.pose.pose.position.x/y/z,四元数在msg.pose.pose.orientation。把这些字段按时间串联起来,用matplotlib或者plotly画出来,立刻能看到设备是怎么运动的。整个过程都不需要ROS,纯Python就能闭环。
我在实际工作中,已经把这种“不装ROS解析bag”的流程固化成了一套内部工具包,前端用rosbags读取,中间用numpy/pandas清洗,最后统一输出成标准格式。刚开始做的时候确实花了一些时间踩坑,但一旦跑通,效率比在ROS里写rosbag脚本高得多,尤其是换机器、换系统、上云的时候,差别非常明显。如果你也只是想快速拿到bag里的数据做分析,不妨按上面的思路自己搭一条轻量解析链路。