📼 第13课：rosbag数据录制

📋 课程目标

• 理解rosbag2的作用与架构
• 掌握话题数据的录制与回放
• 学会录制过滤和压缩配置
• 掌握Python API录制回放
• 了解rosbag2的存储格式与工具

🧠 rosbag2是什么？

rosbag2是ROS2的数据录制与回放工具，可以记录ROS2话题消息到文件，之后回放重建当时的通信场景。它是算法调试和数据驱动开发的核心工具。

📼 录制数据

# 录制所有话题
ros2 bag record -a

# 录制指定话题
ros2 bag record /scan /odom /cmd_vel

# 录制并指定输出目录
ros2 bag record -o my_recording /scan /odom

# 录制指定时长（秒）
ros2 bag record --duration 60 /scan /odom

# 录制指定大小限制
ros2 bag record --max-bag-size 100 /scan

# 录制时限制消息数
ros2 bag record --max-cache-size 1000000 /scan

# 使用压缩录制
ros2 bag record --compression-mode file \
  --compression-format zstd /scan /odom

# 排除话题（正则表达式）
ros2 bag record -a --exclude "/camera/.*"

# 录制时包含TF
ros2 bag record /scan /odom /tf /tf_static

▶️ 回放数据

# 回放bag文件
ros2 bag play my_recording

# 以2倍速回放
ros2 bag play my_recording --rate 2.0

# 以0.5倍速回放（慢放）
ros2 bag play my_recording --rate 0.5

# 只回放指定话题
ros2 bag play my_recording --topics /scan /odom

# 循环回放
ros2 bag play my_recording --loop

# 回放指定时间段
ros2 bag play my_recording --start-offset 10 --duration 30

# 不发布时钟
ros2 bag play my_recording --clock

# 调整回放延迟
ros2 bag play my_recording --delay 2.0

📊 查看bag信息

# 查看bag基本信息
ros2 bag info my_recording

# 输出示例:
# Files:             my_recording_0.db3
# Bag size:          125.3 MiB
# Storage id:        sqlite3
# Duration:          60.5s
# Start:             2024-01-15 10:30:00
# End:               2024-01-15 10:31:00
# Messages:          12500
# Topic information:
#   /scan: 600 msgs (sensor_msgs/msg/LaserScan)
#   /odom: 600 msgs (nav_msgs/msg/Odometry)
#   /cmd_vel: 150 msgs (geometry_msgs/msg/Twist)

🐍 Python API：录制与回放

#!/usr/bin/env python3
"""使用Python API进行rosbag2操作"""

import rclpy
from rclpy.node import Node
from rosbag2_py import SequentialWriter, SequentialReader, StorageOptions, ConverterOptions
from rosbag2_py import TopicMetadata
from std_msgs.msg import String
import time


class BagRecorderNode(Node):
    """通过Python API录制bag"""

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

        # 设置存储选项
        storage_options = StorageOptions(
            uri='python_recording',
            storage_id='sqlite3'
        )
        converter_options = ConverterOptions('', '')

        # 创建写入器
        self.writer = SequentialWriter()
        self.writer.open(storage_options, converter_options)

        # 创建话题
        topic = TopicMetadata(
            name='/chatter',
            type='std_msgs/msg/String',
            serialization_format='cdr'
        )
        self.writer.create_topic(topic)

        # 订阅并录制
        self.sub = self.create_subscription(
            String, '/chatter', self.callback, 10)

        self.msg_count = 0
        self.get_logger().info('Bag录制节点启动')

    def callback(self, msg):
        """接收到消息时写入bag"""
        # 写入bag（需要序列化）
        self.writer.write(
            '/chatter',
            msg.get_ros2_type_support_name(),
            msg.serialize(),
            self.get_clock().now().nanoseconds
        )
        self.msg_count += 1
        if self.msg_count % 100 == 0:
            self.get_logger().info(f'已录制 {self.msg_count} 条消息')


def main(args=None):
    rclpy.init(args=args)
    node = BagRecorderNode()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

⚙️ YAML录制配置

# recording_config.yaml
rosbag2_bagfile_configuration:
  storage_id: sqlite3
  max_bagfile_size: 100000000  # 100MB
  max_cache_size: 10000000
  compression_mode: file
  compression_format: zstd
  all_topics: false
  topics:
    - /scan
    - /odom
    - /tf
    - /tf_static
  exclude_topics:
    - "/camera/.*"
  record_qos:
    reliability: best_effort

🎯 练习题

🏆 成就解锁