Python API（rosbag2_py）

rosbag2_py 是 rosbag2 的 Python 绑定包，基于 pybind11 实现。它提供 SequentialWriter、SequentialReader 等核心类，让 Python 程序可以直接录制和读取 rosbag。

常用 API 概览

StorageOptions

描述 bag 存储位置和存储后端的选项：

storage_options = rosbag2_py.StorageOptions(
    uri='my_bag',          # bag 输出目录（或文件）路径
    storage_id='sqlite3')  # 存储后端：'sqlite3' 或 'mcap'，默认 'mcap'

ConverterOptions

指定消息序列化格式的转换选项。通常使用默认值（''、''），即不转换：

converter_options = rosbag2_py.ConverterOptions('', '')

TopicMetadata

描述要录制的话题元信息（id、名称、类型、序列化格式）：

topic_info = rosbag2_py.TopicMetadata(
    id=0,
    name='chatter',
    type='std_msgs/msg/String',
    serialization_format='cdr')

SequentialWriter

顺序写入器。典型用法：

writer = rosbag2_py.SequentialWriter()
writer.open(storage_options, converter_options)
writer.create_topic(topic_info)

# 写入一条消息：话题名、序列化后的消息字节、时间戳（纳秒）
writer.write('chatter', serialize_message(msg), self.get_clock().now().nanoseconds)

SequentialReader

顺序读取器。典型用法：

reader = rosbag2_py.SequentialReader()
reader.open(storage_options, converter_options)

topic_types = reader.get_all_topics_and_types()   # 获取所有话题及类型
while reader.has_next():
    topic, data, timestamp = reader.read_next()   # 读取下一条消息

CompressionOptions 与 SequentialCompressionWriter

录制时启用压缩的写入器：

compression_options = rosbag2_py.CompressionOptions(
    compression_format='zstd',
    compression_mode=rosbag2_py.CompressionMode.MESSAGE)

compressed_writer = rosbag2_py.SequentialCompressionWriter(compression_options)
compressed_writer.open(storage_options, converter_options)

重新生成 Python stub 文件（.pyi）

Python stub 文件为二进制 Python 模块（例如基于 pybind11 的模块）提供类型提示信息。

在 rosbag2_py 中，stub 文件是用名为 stubgen 的工具生成的。

重新生成 stub 文件的步骤：

cd <workspace>
colcon build --packages-up-to rosbag2_py
source install/setup.sh
# Make sure rosbag2_py can be imported
python3 -c 'import rosbag2_py'

# Ubuntu 24.04
sudo apt update && sudo apt install mypy

# Older Ubuntu
# pip3 install -U mypy==1.9

cd <rosbag2 git repo>
stubgen -p rosbag2_py -o rosbag2_py