Apollo自动驾驶平台.record文件数据解析：从传感器数据到消息格式的完整指南

共计 2408 个字符，预计需要花费 7 分钟才能阅读完成。

核心概念：.record 文件的二进制结构

Apollo 平台的.record 文件本质上是一个序列化的消息存储容器，采用自定义二进制格式组织数据。它的设计遵循了三个核心原则：

• 时间有序性：所有消息按采集时间戳严格排序
• 多通道支持：通过 Channel 区分不同传感器数据流
• 自描述格式：包含消息类型和序列化元数据

文件结构可以理解为由三部分组成：

1. 文件头：包含魔数标识和版本信息
2. 消息块序列：每个块包含 Header（时间戳、消息类型）和 Payload
3. 索引区（可选）：加速随机访问的元数据

开发者常见痛点分析

在实际开发中，处理.record 文件常遇到这些问题：

• 数据吞吐瓶颈：单文件常达数十 GB，直接全量加载导致内存溢出
• 时间同步困难：相机 / 激光雷达等异构传感器时钟基准不同
• 解析效率低下：传统逐字节解析方式 CPU 利用率不足 30%

Cyber RT Python API 实战

基础解析流程

以下是使用 Apollo 7.0 Cyber RT 解析文件的标准流程：

from cyber_py import record

# 1. 创建 Reader 实例
reader = record.RecordReader('example.record')

# 2. 获取文件元信息
print(f"文件包含 {reader.get_messagenumber()} 条消息")
print(f"支持的 Channel: {reader.get_channellist()}")

# 3. 按顺序读取消息
for channel_name, message, datatype, timestamp in reader.read_messages():
    if channel_name == '/apollo/sensor/camera/front_6mm':
        # 处理相机数据
        process_image(message)
    elif channel_name == '/apollo/sensor/lidar32/compensator/PointCloud2':
        # 处理点云
        process_pointcloud(message)

关键数据类型解析

激光雷达点云解析示例：

import numpy as np
from modules.drivers.proto.pointcloud_pb2 import PointCloud

def process_pointcloud(proto_data):
    pc = PointCloud()
    pc.ParseFromString(proto_data)

    # 转换为 Numpy 数组
    points = np.array([[p.x, p.y, p.z] for p in pc.point])
    intensities = np.array([p.intensity for p in pc.point])

    print(f"获取到 {len(points)} 个点，强度范围: {intensities.min()}-{intensities.max()}")

相机图像解析示例：

import cv2
from modules.drivers.proto.sensor_image_pb2 import Image

def process_image(proto_data):
    img = Image()
    img.ParseFromString(proto_data)

    # 转换 BGR 格式
    mat = cv2.imdecode(np.frombuffer(img.data, dtype=np.uint8), 
        cv2.IMREAD_COLOR
    )
    cv2.imwrite(f"output/{img.header.timestamp_sec}.jpg", mat)

性能优化技巧

索引加速方案

对于大型文件，建议先构建索引：

1. 创建索引文件

   cyber_recorder index -f input.record -o output.index

2. Python 中使用索引

   reader = record.RecordReader(
       'input.record', 
       index_file='output.index'
   )

实测对比（1.2GB 文件）：

内存管理建议

• 使用迭代器而非全量加载：reader.read_messages()本身就是生成器
• 及时释放 ProtoBuf 对象：处理完立即 del 大型消息对象
• 分批处理：对于超长记录，按时间窗口分段处理

时间同步避坑指南

常见问题场景：
– 相机和 IMU 数据时间戳相差几百毫秒
– 不同激光雷达扫描周期不完全对齐

解决方案：

1. 检查传感器硬件时钟同步
2. 软件层对齐方法：

   from cyber_py import cyber_time
   
   # 创建时间对齐器
   aligner = cyber_time.MessageTimeAlign(max_time_diff=0.1  # 允许 100ms 误差)
   
   # 添加各通道消息
   aligner.add_channel('/camera')
   aligner.add_channel('/lidar')
   
   # 获取同步消息组
   sync_msgs = aligner.get_synced_messages()

进阶调试技巧

当遇到解析异常时，可以：

1. 检查文件完整性

   cyber_recorder info -f broken.record

2. 提取特定时间段的记录

   cyber_recorder cut -f input.record -b 1625000000 -e 1625000100 -o segment.record

3. 使用可视化工具检查

   cyber_visualizer --record input.record

经验总结

经过多个实际项目验证，高效处理.record 文件的关键在于：

• 预处理阶段：务必先检查文件完整性并建立索引
• 解析阶段：按需加载，避免全量数据驻留内存
• 后处理阶段：合理使用 Cyber RT 内置的时间同步工具

最后建议开发者定期使用 cyber_recorder repair 修复异常记录文件，这个隐藏功能能解决 90% 的文件损坏问题。