示例（Python）

rosbag2_examples_py 包提供了一组使用 rosbag2_py Python API 的示例，覆盖录制、读取、数据生成、压缩录制和 bag 转 CSV。

安装该包后（例如 sudo apt install ros-$ROS_DISTRO-rosbag2-examples-py，或通过 colcon 构建），每个示例都会注册一个控制台入口点，可以直接运行。

简单录制器（simple_bag_recorder）

订阅 chatter 话题（std_msgs/msg/String），把收到的消息录制到 sqlite3 格式的 my_bag 中。

ros2 run rosbag2_examples_py simple_bag_recorder

import rclpy
from rclpy.executors import ExternalShutdownException
from rclpy.node import Node
from rclpy.serialization import serialize_message
import rosbag2_py
from std_msgs.msg import String


class SimpleBagRecorder(Node):

    def __init__(self):
        super().__init__('simple_bag_recorder')
        self.writer = rosbag2_py.SequentialWriter()

        storage_options = rosbag2_py.StorageOptions(
            uri='my_bag',
            storage_id='sqlite3')
        converter_options = rosbag2_py.ConverterOptions('', '')
        self.writer.open(storage_options, converter_options)

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

        self.subscription = self.create_subscription(
            String,
            'chatter',
            self.topic_callback,
            10)
        self.subscription

    def topic_callback(self, msg):
        self.writer.write(
            'chatter',
            serialize_message(msg),
            self.get_clock().now().nanoseconds)


def main(args=None):
    try:
        with rclpy.init(args=args):
            sbr = SimpleBagRecorder()
            rclpy.spin(sbr)
    except (KeyboardInterrupt, ExternalShutdownException):
        pass


if __name__ == '__main__':
    main()

简单读取器（simple_bag_reader）

从 sqlite3 格式的 bag 中读取 chatter 消息，并每隔 0.1 秒重新发布一条到 chatter 话题。

ros2 run rosbag2_examples_py simple_bag_reader <bag_directory>

import sys

import rclpy
from rclpy.executors import ExternalShutdownException
from rclpy.node import Node
import rosbag2_py
from std_msgs.msg import String


class SimpleBagReader(Node):

    def __init__(self, bag_filename):
        super().__init__('simple_bag_reader')
        self.reader = rosbag2_py.SequentialReader()
        storage_options = rosbag2_py.StorageOptions(
            uri=bag_filename,
            storage_id='sqlite3')
        converter_options = rosbag2_py.ConverterOptions('', '')
        self.reader.open(storage_options, converter_options)

        self.publisher = self.create_publisher(String, 'chatter', 10)
        self.timer = self.create_timer(0.1, self.timer_callback)

    def timer_callback(self):
        while self.reader.has_next():
            msg = self.reader.read_next()
            if msg[0] != 'chatter':
                continue
            self.publisher.publish(msg[1])
            self.get_logger().info('Publish serialized data to ' + msg[0])
            break


def main(args=None):
    try:
        with rclpy.init(args=args):
            sbr = SimpleBagReader(sys.argv[1])
            rclpy.spin(sbr)
    except (KeyboardInterrupt, ExternalShutdownException):
        pass


if __name__ == '__main__':
    main()

数据生成节点（data_generator_node）

一个 ROS 2 节点，每秒生成一个递增的 Int32 消息并录制到 mcap 格式的 timed_synthetic_bag 中。时间戳来自节点时钟。

ros2 run rosbag2_examples_py data_generator_node

from example_interfaces.msg import Int32
import rclpy
from rclpy.executors import ExternalShutdownException
from rclpy.node import Node
from rclpy.serialization import serialize_message
import rosbag2_py


class DataGeneratorNode(Node):

    def __init__(self):
        super().__init__('data_generator_node')
        self.data = Int32()
        self.data.data = 0
        self.writer = rosbag2_py.SequentialWriter()

        storage_options = rosbag2_py.StorageOptions(
            uri='timed_synthetic_bag',
            storage_id='mcap')
        converter_options = rosbag2_py.ConverterOptions('', '')
        self.writer.open(storage_options, converter_options)

        topic_info = rosbag2_py.TopicMetadata(
            id=0,
            name='synthetic',
            type='example_interfaces/msg/Int32',
            serialization_format='cdr')
        self.writer.create_topic(topic_info)

        self.timer = self.create_timer(1, self.timer_callback)

    def timer_callback(self):
        self.writer.write(
            'synthetic',
            serialize_message(self.data),
            self.get_clock().now().nanoseconds)
        self.data.data += 1


def main(args=None):
    try:
        with rclpy.init(args=args):
            dgn = DataGeneratorNode()
            rclpy.spin(dgn)
    except (KeyboardInterrupt, ExternalShutdownException):
        pass


if __name__ == '__main__':
    main()

数据生成可执行程序（data_generator_executable）

与 data_generator_node 类似的合成数据生成器，但不是一个 ROS 2 节点：它生成 100 条 Int32 消息，时间戳从 0 开始每秒递增，录制到 sqlite3 格式的 big_synthetic_bag 后立即关闭写入器并退出。适合生成确定性的合成 bag 用于测试。

ros2 run rosbag2_examples_py data_generator_executable

from example_interfaces.msg import Int32
from rclpy.clock import Clock
from rclpy.duration import Duration
from rclpy.serialization import serialize_message
import rosbag2_py


def main(args=None):
    writer = rosbag2_py.SequentialWriter()

    storage_options = rosbag2_py.StorageOptions(
        uri='big_synthetic_bag',
        storage_id='sqlite3')
    converter_options = rosbag2_py.ConverterOptions('', '')
    writer.open(storage_options, converter_options)

    topic_info = rosbag2_py.TopicMetadata(
        id=0,
        name='synthetic',
        type='example_interfaces/msg/Int32',
        serialization_format='cdr')
    writer.create_topic(topic_info)

    time_stamp = Clock().now()
    for ii in range(0, 100):
        data = Int32()
        data.data = ii
        writer.write(
            'synthetic',
            serialize_message(data),
            time_stamp.nanoseconds)
        time_stamp += Duration(seconds=1)
    writer.close()
    print("Generated data saved into the '%s'" % storage_options.uri)


if __name__ == '__main__':
    main()

压缩录制器（compressed_bag_recorder）

与 simple_bag_recorder 相同，但使用 SequentialCompressionWriter 和 CompressionMode.MESSAGE 以 zstd 格式对每条消息进行压缩录制。

ros2 run rosbag2_examples_py compressed_bag_recorder

import rclpy
from rclpy.executors import ExternalShutdownException
from rclpy.node import Node
from rclpy.serialization import serialize_message
import rosbag2_py
from std_msgs.msg import String


class CompressedBagRecorder(Node):

    def __init__(self):
        super().__init__('compressed_bag_recorder')

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

        storage_options = rosbag2_py.StorageOptions(
            uri='my_bag',
            storage_id='sqlite3')
        converter_options = rosbag2_py.ConverterOptions('', '')

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

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

        self.subscription = self.create_subscription(
            String,
            'chatter',
            self.topic_callback,
            10)
        self.subscription

    def topic_callback(self, msg):
        self.compressed_writer.write(
            'chatter',
            serialize_message(msg),
            self.get_clock().now().nanoseconds)


def main(args=None):
    try:
        with rclpy.init(args=args):
            cbr = CompressedBagRecorder()
            rclpy.spin(cbr)
    except (KeyboardInterrupt, ExternalShutdownException):
        pass


if __name__ == '__main__':
    main()

bag 转 CSV（rosbag2csv）

读取 bag 中的所有消息，转换为 YAML 字符串后写入 <input_bag>.csv 文件。CSV 包含四列：topic_name、topic_type、timestamp_ns、data。

ros2 run rosbag2_examples_py rosbag2csv -i <bag_directory>

"""Script that reads ROS 2 messages from a bag file and saves them to a csv file."""

import argparse
import csv
import os

from rclpy.serialization import deserialize_message
import rosbag2_py
from rosidl_runtime_py.convert import message_to_yaml
from rosidl_runtime_py.utilities import get_message


def read_messages(input_bag: str):
    reader = rosbag2_py.SequentialReader()
    reader.open(
        rosbag2_py.StorageOptions(uri=input_bag),
        rosbag2_py.ConverterOptions('', ''),
    )

    topic_types = reader.get_all_topics_and_types()

    # Create a map for quicker lookup
    type_map = {topic_types[i].name: topic_types[i].type for i in range(len(topic_types))}

    while reader.has_next():
        topic, data, timestamp = reader.read_next()
        msg_type = get_message(type_map[topic])
        msg = deserialize_message(data, msg_type)
        yield topic, msg, timestamp
    del reader


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        '-i', '--input', help='input bag path (folder or filepath) to read from'
    )

    args = parser.parse_args()

    if 'BUILD_WORKING_DIRECTORY' in os.environ:
        # Workaround for Bazel to support the relative path's for input and output files
        os.chdir(os.environ['BUILD_WORKING_DIRECTORY'])

    with open(f'{args.input}.csv', 'w', newline='') as f_out:
        csv_writer = csv.writer(f_out)
        # Write header
        csv_writer.writerow(['topic_name', 'topic_type', 'timestamp_ns', 'data'])

        for topic_name, msg, timestamp_ns in read_messages(args.input):
            # Convert to YAML
            yaml_str = message_to_yaml(msg)
            # Write the YAML string in the 'data' field
            csv_writer.writerow([topic_name, type(msg).__name__, timestamp_ns, yaml_str])
            # print(f"{topic_name} ({type(msg).__name__}) [{timestamp_ns}]: '{yaml_str}'")


if __name__ == '__main__':
    main()