第三讲：ROS Bag文件解析——从结构到实战
做自动驾驶感知的人，每天打交道最多的数据格式，除了PCD、BIN，就是ROS Bag了。我个人觉得，Bag文件就像是一个时间胶囊——它把传感器数据、车辆状态、算法输出，全都按时间顺序打包存好。你随时可以打开它，回到那个采集时刻。
这一讲，我们就来彻底搞懂ROS Bag。从文件结构到底层原理，从命令行工具到Python读取，最后把点云话题数据提取出来。嗯，都是干货。
3.1 Bag文件结构——它到底长什么样？
ROS Bag本质上是一个二进制文件。但它不是乱糟糟的二进制流，而是有严格结构的。我当年第一次用十六进制编辑器打开Bag文件时，发现它其实分成了三个区域：
- Header（文件头）：记录版本号、索引位置、chunk数量等元信息
- Chunks（数据块）：真正存储消息的地方，每条消息都带时间戳
- Index（索引区）：记录每个话题在chunk中的位置，方便快速查找
说白了，Bag文件就是「索引 + 数据块」的组合。你想想看，如果没有索引，要找一个话题的数据就得从头扫到尾，那效率得多低？
核心概念：Bag文件中的每条消息都包含三个要素——话题名称、消息类型、时间戳。这三者缺一不可。
下面这张图，是我自己画的Bag文件结构示意，帮你快速建立直观印象：
3.2 rosbag命令行工具——你每天都在用的瑞士军刀
ROS自带的rosbag命令行工具，说实话，功能非常强大。我平时80%的数据处理工作，靠它就能搞定。
3.2.1 常用命令一览
| 命令 | 作用 | 我常用的场景 |
|---|---|---|
rosbag info |
查看Bag文件信息 | 拿到一个新bag，先info一下看看有哪些话题 |
rosbag play |
回放Bag数据 | 测试算法时，模拟传感器数据流 |
rosbag record |
录制话题数据 | 路测时采集数据，只录需要的几个话题 |
rosbag filter |
按条件过滤 | 提取某个时间段的数据，或者只保留特定话题 |
rosbag compress |
压缩Bag文件 | 大文件存档前压缩，能省60%空间 |
3.2.2 实战：用info命令查看Bag
# 查看bag文件基本信息
rosbag info 2024-03-15-14-30-00.bag
# 输出示例
path: 2024-03-15-14-30-00.bag
version: 2.0
duration: 180.2s
start: Mar 15 2024 14:30:00.12
end: Mar 15 2024 14:33:00.32
size: 2.4 GB
messages: 125000
compression: none [1/1 chunks]
types: sensor_msgs/PointCloud2 [1158d4...]
sensor_msgs/Imu [6a62c6...]
topics: /points_raw 10000 msgs : sensor_msgs/PointCloud2
/imu_data 50000 msgs : sensor_msgs/Imu
/vehicle_status 65000 msgs : std_msgs/String我的小技巧：看到compression: none时，我一般会顺手压缩一下。用rosbag compress --lz4，压缩速度快，解压也快，适合日常使用。
3.2.3 实战：用filter提取点云话题
# 只保留点云话题，其他数据全部扔掉
rosbag filter input.bag output.bag "topic == '/points_raw'"
# 也可以按时间过滤：只取前60秒的数据
rosbag filter input.bag output.bag "t.to_sec() < 1645000060.0"这里要注意，filter命令的表达式是Python语法。你可以在条件里写topic、t（时间戳）这些变量。我曾经踩过一个坑——时间戳比较时忘了转成秒，结果过滤出来的数据全是空的。
3.3 Python读取Bag文件——用代码掌控数据
命令行工具虽然方便，但要做精细化处理，还得上Python。ROS提供了rosbag Python包，接口设计得很简洁。
3.3.1 安装与导入
# 安装（如果你已经装了ROS，这个包默认就有）
pip install rosbag
# 导入
import rosbag
from sensor_msgs.msg import PointCloud2
import sensor_msgs.point_cloud2 as pc23.3.2 读取Bag文件的基本流程
bag = rosbag.Bag('2024-03-15-14-30-00.bag', 'r')
# 遍历所有消息
for topic, msg, t in bag.read_messages():
print(f"话题: {topic}, 时间戳: {t.to_sec():.3f}")
bag.close()这段代码很简单，但有个细节要注意——bag.read_messages()返回的是一个生成器，不会一次性把所有消息加载到内存。你想想看，一个2.4GB的bag，如果全读到内存里，机器不炸才怪。
3.3.3 只读取特定话题
# 只读取点云话题，效率高很多
bag = rosbag.Bag('2024-03-15-14-30-00.bag', 'r')
for topic, msg, t in bag.read_messages(topics=['/points_raw']):
# 这里只处理点云数据
print(f"收到一帧点云，时间: {t.to_sec():.3f}")
bag.close()性能提示：指定topics参数后，rosbag内部会跳过不相关的话题数据，读取速度能提升5-10倍。我处理大文件时一定会加这个参数。
3.4 提取点云话题数据——从Bag到NumPy数组
好了，到了最核心的部分。我们要把Bag里的点云数据提取出来，转成NumPy数组，这样才能做后续处理。
3.4.1 PointCloud2消息结构
ROS里的点云消息类型是sensor_msgs/PointCloud2。它包含以下关键字段：
| 字段 | 类型 | 说明 |
|---|---|---|
header |
std_msgs/Header | 包含时间戳、坐标系ID（如velodyne） |
height |
uint32 | 点云高度（非结构化点云通常为1） |
width |
uint32 | 点云宽度（即点的数量） |
fields |
PointField[] | 每个点的字段定义（x, y, z, intensity, ring等） |
data |
uint8[] | 原始二进制数据 |
3.4.2 完整提取代码
import rosbag
import sensor_msgs.point_cloud2 as pc2
import numpy as np
def extract_pointcloud_from_bag(bag_path, topic_name):
"""
从bag文件中提取点云数据
返回: 点云列表，每个元素是 (timestamp, points_np)
"""
bag = rosbag.Bag(bag_path, 'r')
cloud_data = []
for topic, msg, t in bag.read_messages(topics=[topic_name]):
# 将PointCloud2消息转为点列表
points_list = list(pc2.read_points(
msg,
field_names=('x', 'y', 'z', 'intensity'),
skip_nans=True
))
# 转为NumPy数组，形状为 (N, 4)
points_np = np.array(points_list, dtype=np.float32)
# 记录时间戳和点云
cloud_data.append((t.to_sec(), points_np))
# 打印一些信息
if len(cloud_data) <= 3:
print(f"第{len(cloud_data)}帧: {points_np.shape[0]}个点, "
f"时间戳: {t.to_sec():.3f}")
bag.close()
print(f"共提取 {len(cloud_data)} 帧点云")
return cloud_data
# 使用示例
clouds = extract_pointcloud_from_bag(
'2024-03-15-14-30-00.bag',
'/points_raw'
)3.4.3 避坑指南
我曾经踩过的坑：
- 内存爆炸：有一次我直接
list(bag.read_messages())，结果一个8GB的bag把服务器内存撑爆了。记住，read_messages()是生成器，别转成list。 - 字段名不匹配：不同激光雷达的字段名不一样。Velodyne用
intensity，Ouster用signal，RoboSense用reflectivity。读取前先用msg.fields确认一下。 - 时间戳精度：ROS的时间戳是秒+纳秒，转成浮点数时注意精度。我一般用
t.to_sec()，够用了。
3.4.4 进阶：批量提取并保存为PCD文件
import open3d as o3d
def save_clouds_to_pcd(cloud_data, output_dir):
"""将提取的点云保存为PCD文件"""
import os
os.makedirs(output_dir, exist_ok=True)
for i, (timestamp, points_np) in enumerate(cloud_data):
# 只保留前100帧做测试
if i >= 100:
break
# 创建Open3D点云对象
pcd = o3d.geometry.PointCloud()
pcd.points = o3d.utility.Vector3dVector(points_np[:, :3])
# 如果有强度信息，可以赋颜色
if points_np.shape[1] >= 4:
intensity = points_np[:, 3]
# 将强度归一化到0-1，映射为灰度颜色
colors = np.zeros((len(intensity), 3))
colors[:, 0] = intensity / 255.0
colors[:, 1] = intensity / 255.0
colors[:, 2] = intensity / 255.0
pcd.colors = o3d.utility.Vector3dVector(colors)
# 保存文件
filename = f"{output_dir}/cloud_{timestamp:.3f}.pcd"
o3d.io.write_point_cloud(filename, pcd)
if i % 20 == 0:
print(f"已保存 {i+1} 帧...")
# 执行保存
save_clouds_to_pcd(clouds, './extracted_clouds')我的建议：实际项目中，我一般不会把所有帧都保存为PCD文件——太占磁盘了。通常的做法是：先提取关键帧（比如每10帧取1帧），或者只保存感兴趣时间段的点云。这样既节省空间，又不影响算法验证。
3.5 本章小结
这一讲我们完整走通了ROS Bag的解析流程。从文件结构到命令行工具，再到Python代码实现，最后把点云数据提取出来并保存为PCD文件。说白了，Bag文件就是一个带时间戳的数据容器，掌握了它的读写方法，你就能自由地处理自动驾驶数据了。
嗯，内容不少，但都是实战中一定会用到的。下次拿到一个新的Bag文件，你应该知道怎么下手了吧？