rosbag2 Recording Guide 2026Record · Playback · Filter · Python API

# Record ALL topics
ros2 bag record -a

# Record specific topics only
ros2 bag record /camera/image_raw /scan /odom /imu

# Record with output directory name
ros2 bag record -o my_dataset /scan /odom /cmd_vel

# Record with compression (LZ4 — fast, medium ratio)
ros2 bag record -a --compression-mode file --compression-format lz4

# Record with Zstandard (slower, higher ratio — best for long runs)
ros2 bag record -a --compression-mode message --compression-format zstd

# Record for a fixed duration (seconds)
ros2 bag record -a --duration 60   # record 60 seconds then stop

# Record up to a max file size (bytes → auto-split)
ros2 bag record -a --max-bag-size 1073741824  # 1 GB splits

# Play at real speed
ros2 bag play my_dataset

# Play at 0.5× speed (slow motion — useful for debugging fast sensors)
ros2 bag play my_dataset --rate 0.5

# Play at 2× speed
ros2 bag play my_dataset --rate 2.0

# Loop playback
ros2 bag play my_dataset --loop

# Play only specific topics
ros2 bag play my_dataset --topics /scan /odom

# Start from a specific time offset (seconds from start)
ros2 bag play my_dataset --start-offset 30.0

# Pause/resume with keyboard: press SPACE during playback

# Remap topic names during playback
ros2 bag play my_dataset --remap /scan:=/lidar/scan

ros2 bag info my_dataset/

# Example output:
# Files:             my_dataset_0.db3
# Bag size:          134.3 MiB
# Storage id:        sqlite3
# Duration:          47.312s
# Start:             Jun 21 2026 14:23:01.123 (1750512181.123)
# End:               Jun 21 2026 14:23:48.435 (1750512228.435)
# Messages:          24156
# Topic information: Topic: /scan | Type: sensor_msgs/msg/LaserScan | Count: 478 | Serialization Format: cdr
#                    Topic: /odom | Type: nav_msgs/msg/Odometry    | Count: 2365 | Serialization Format: cdr
#                    Topic: /camera/image_raw | Type: sensor_msgs/msg/Image | Count: 473 | ...

# Install MCAP storage plugin
sudo apt install ros-jazzy-rosbag2-storage-mcap

# Record to MCAP
ros2 bag record -a --storage mcap -o my_mcap_bag

# Convert existing SQLite bag to MCAP
ros2 bag convert \
  --input-storage sqlite3 ./my_sqlite_bag \
  --output-storage mcap ./my_mcap_bag

# MCAP CLI tool for offline inspection (no ROS needed)
pip install mcap
mcap info my_mcap_bag/my_mcap_bag_0.mcap
mcap cat my_mcap_bag/my_mcap_bag_0.mcap --topics /scan --json | head -5

#!/usr/bin/env python3
"""Read a rosbag2 and extract LaserScan ranges to numpy."""
import numpy as np
from pathlib import Path

from rosbag2_py import SequentialReader, StorageOptions, ConverterOptions
from rclpy.serialization import deserialize_message
from sensor_msgs.msg import LaserScan


def read_bag(bag_path: str):
    storage_options = StorageOptions(
        uri=bag_path,
        storage_id="sqlite3"   # or "mcap"
    )
    converter_options = ConverterOptions(
        input_serialization_format="cdr",
        output_serialization_format="cdr"
    )

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

    topic_types = reader.get_all_topics_and_types()
    type_map = {t.name: t.type for t in topic_types}

    print(f"Topics: {list(type_map.keys())}")

    scan_data = []
    while reader.has_next():
        topic, data, timestamp = reader.read_next()

        if topic == "/scan":
            msg = deserialize_message(data, LaserScan)
            ranges = np.array(msg.ranges)
            scan_data.append({
                "timestamp_ns": timestamp,
                "min_range": float(np.nanmin(ranges[ranges > 0])),
                "mean_range": float(np.nanmean(ranges[ranges > 0])),
            })

    print(f"Read {len(scan_data)} LaserScan messages")
    if scan_data:
        print(f"First scan min range: {scan_data[0]['min_range']:.3f} m")

    return scan_data


if __name__ == "__main__":
    import sys
    data = read_bag(sys.argv[1])

#!/usr/bin/env python3
"""Create a synthetic LaserScan rosbag for testing."""
import time
import math
import rclpy
from rosbag2_py import SequentialWriter, StorageOptions, ConverterOptions, TopicMetadata
from rclpy.serialization import serialize_message
from sensor_msgs.msg import LaserScan
from std_msgs.msg import Header


def create_test_bag(output_path: str, num_scans: int = 100):
    storage_options = StorageOptions(uri=output_path, storage_id="sqlite3")
    converter_options = ConverterOptions(
        input_serialization_format="cdr",
        output_serialization_format="cdr"
    )

    writer = SequentialWriter()
    writer.open(storage_options, converter_options)

    # Register topic
    topic = TopicMetadata(
        name="/scan",
        type="sensor_msgs/msg/LaserScan",
        serialization_format="cdr"
    )
    writer.create_topic(topic)

    # Write synthetic scans
    for i in range(num_scans):
        msg = LaserScan()
        msg.header.frame_id = "lidar_link"
        # Timestamp in nanoseconds
        timestamp_ns = int(i * 1e8)   # 10 Hz
        msg.header.stamp.sec = timestamp_ns // int(1e9)
        msg.header.stamp.nanosec = timestamp_ns % int(1e9)
        msg.angle_min = -math.pi
        msg.angle_max = math.pi
        msg.angle_increment = math.pi / 180  # 1 degree
        msg.range_min = 0.1
        msg.range_max = 10.0
        # Synthetic circular obstacle at 2m
        num_rays = 360
        msg.ranges = [2.0 + 0.1 * math.sin(j / 10.0) for j in range(num_rays)]
        msg.intensities = []

        writer.write("/scan", serialize_message(msg), timestamp_ns)

    print(f"Wrote {num_scans} LaserScan messages to {output_path}")


if __name__ == "__main__":
    create_test_bag("/tmp/test_scan_bag")

# Extract only specific topics into a new bag
ros2 bag filter my_dataset -o filtered_bag \
  --include-regex "/scan|/odom|/tf"

# Trim time range (start and end timestamps in nanoseconds)
ros2 bag filter my_dataset -o trimmed_bag \
  --start-time 1750512195000000000 \
  --end-time   1750512210000000000

# Convert SQLite bag → MCAP
ros2 bag convert \
  -i my_dataset \
  -o output.yaml  # where output.yaml defines output storage options

# Reindex a bag that was interrupted during recording
ros2 bag reindex my_dataset

# Merge multiple bags chronologically
ros2 bag merge --output merged_bag bag1/ bag2/ bag3/