ARTICLE DETAIL

建站实战干货

来自一线的建站与推广经验沉淀,每一条都经过真实交付验证。

不装ROS用Python解析bag文件:提取图像与IMU数据全攻略

2026/10/4 6:12:58 拓冰建站 浏览量
不装ROS用Python解析bag文件:提取图像与IMU数据全攻略 每次拿到.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的官方支持重心在UbuntuWindows上跑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_typestoretypestore) 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_okTrue) video_writer None frame_idx 0 with AnyReader([vins.bag], default_typestoretypestore) 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(connectionsconnections): msg reader.deserialize(rawdata, connection.msgtype) # 根据编码方式转换图像 if msg.encoding rgb8: img np.frombuffer(msg.data, dtypenp.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, dtypenp.uint8).reshape(msg.height, msg.width, 3) elif msg.encoding mono8: img_bgr np.frombuffer(msg.data, dtypenp.uint8).reshape(msg.height, msg.width, 1) else: print(f跳过不支持的编码: {msg.encoding}) continue cv2.imwrite(str(out_dir / fframe_{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/CompressedImageJPEG或PNG压缩后的字节流。上面脚本针对的是前者。如果你拿到的是CompressedImage解析方式会简单很多直接用cv2.imdecode就可以不需要关心encoding。实际处理前先确认bag里是哪种类型代码路径完全不同。3.3 IMU消息导出为CSVIMU数据相比图像简单很多它只是一组带时间戳的线性加速度和角速度。下面的代码会把/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_typestoretypestore) 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(connectionsconnections): 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, dtypenp.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, dtypedtype, countmsg.width * msg.height) typestore get_typestore(Stores.ROS1_NOETIC) with AnyReader([vins.bag], default_typestoretypestore) 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(connectionsconnections): 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(xt_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里的数据做分析不妨按上面的思路自己搭一条轻量解析链路。