Rosbag2 Analysis

name: rosbag2-analysis description: 'Analyze rosbag2 data offline. Use when reading bag files in Python, plotting trajectories, timing analysis, or mcap format.'

Analyzing rosbag2 Data Offline

Recording Bags

# Record all topics
ros2 bag record -a

# Record selected topics (preferred for disk space)
ros2 bag record /scan /odom /cmd_vel /battery_state /tf /tf_static

# Use mcap storage (better compression, random access, Foxglove-compatible)
ros2 bag record --storage mcap /scan /odom /cmd_vel

# Default sqlite3 storage (works everywhere)
ros2 bag record --storage sqlite3 /odom /cmd_vel

# Set max bag duration/size
ros2 bag record -a --max-bag-duration 300 --max-bag-size 1073741824

Inspecting Bags

# Summary info: topics, message counts, duration
ros2 bag info bag_directory/

# Example output:
# Files:             bag_directory/bag_directory_0.mcap
# Bag size:          142.6 MiB
# Duration:          312.4s
# Messages:          48732
# Topic information:
#   /odom          12500 msgs : nav_msgs/msg/Odometry
#   /scan           6240 msgs : sensor_msgs/msg/LaserScan
#   /cmd_vel        6250 msgs : geometry_msgs/msg/Twist
#   /battery_state   312 msgs : sensor_msgs/msg/BatteryState

Playback

# Basic playback (publishes /clock for use_sim_time nodes)
ros2 bag play bag_directory/ --clock

# Half-speed replay for debugging
ros2 bag play bag_directory/ --clock --rate 0.5

# Double-speed for fast review
ros2 bag play bag_directory/ --clock --rate 2.0

# Filter to specific topics on playback
ros2 bag play bag_directory/ --topics /scan /odom

# Loop continuously
ros2 bag play bag_directory/ --clock --loop

# Start from offset (seconds)
ros2 bag play bag_directory/ --clock --start-offset 60.0

Python Analysis with rosbag2_py

#!/usr/bin/env python3
"""Read /odom from a rosbag and plot the robot trajectory."""

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from rclpy.serialization import deserialize_message
from rosidl_runtime_py.utilities import get_message
import rosbag2_py


def read_messages(bag_path: str, topics: list[str]):
    """Yield (topic, msg, timestamp_ns) from a rosbag."""
    reader = rosbag2_py.SequentialReader()
    storage_options = rosbag2_py.StorageOptions(uri=bag_path, storage_id="")
    converter_options = rosbag2_py.ConverterOptions(
        input_serialization_format="cdr",
        output_serialization_format="cdr",
    )
    reader.open(storage_options, converter_options)

    # Build type map
    type_map = {}
    for topic_info in reader.get_all_topics_and_types():
        type_map[topic_info.name] = topic_info.type

    # Filter to requested topics
    storage_filter = rosbag2_py.StorageFilter(topics=topics)
    reader.set_filter(storage_filter)

    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


def odom_to_dataframe(bag_path: str) -> pd.DataFrame:
    """Extract /odom into a pandas DataFrame."""
    rows = []
    for topic, msg, ts in read_messages(bag_path, ["/odom"]):
        rows.append({
            "time_ns": ts,
            "x": msg.pose.pose.position.x,
            "y": msg.pose.pose.position.y,
            "vx": msg.twist.twist.linear.x,
            "wz": msg.twist.twist.angular.z,
        })
    df = pd.DataFrame(rows)
    df["time_s"] = (df["time_ns"] - df["time_ns"].iloc[0]) / 1e9
    return df


def plot_trajectory(df: pd.DataFrame):
    fig, axes = plt.subplots(1, 3, figsize=(18, 5))

    # Trajectory (x vs y)
    axes[0].plot(df["x"], df["y"], linewidth=0.8)
    axes[0].plot(df["x"].iloc[0], df["y"].iloc[0], "go", markersize=10, label="Start")
    axes[0].plot(df["x"].iloc[-1], df["y"].iloc[-1], "rs", markersize=10, label="End")
    axes[0].set_xlabel("X (m)")
    axes[0].set_ylabel("Y (m)")
    axes[0].set_title("Robot Trajectory")
    axes[0].set_aspect("equal")
    axes[0].legend()
    axes[0].grid(True)

    # Velocity profiles
    axes[1].plot(df["time_s"], df["vx"], label="Linear (m/s)", linewidth=0.8)
    axes[1].plot(df["time_s"], df["wz"], label="Angular (rad/s)", linewidth=0.8)
    axes[1].set_xlabel("Time (s)")
    axes[1].set_title("Velocity Profile")
    axes[1].legend()
    axes[1].grid(True)

    # Timing jitter (message interval vs expected 50Hz)
    dt = np.diff(df["time_s"])
    expected_dt = 0.02  # 50 Hz
    axes[2].hist(dt * 1000, bins=100, edgecolor="black", alpha=0.7)
    axes[2].axvline(expected_dt * 1000, color="r", linestyle="--", label=f"Expected {expected_dt*1000:.0f}ms")
    axes[2].set_xlabel("Message Interval (ms)")
    axes[2].set_title("Odom Timing Jitter")
    axes[2].legend()
    axes[2].grid(True)

    plt.tight_layout()
    plt.savefig("rosbag_analysis.png", dpi=150)
    plt.show()


if __name__ == "__main__":
    import sys
    bag_path = sys.argv[1] if len(sys.argv) > 1 else "rosbag2_recording"
    df = odom_to_dataframe(bag_path)
    print(f"Loaded {len(df)} odom messages over {df['time_s'].iloc[-1]:.1f}s")
    print(f"Distance traveled: {np.sqrt(np.diff(df['x'])**2 + np.diff(df['y'])**2).sum():.2f}m")
    plot_trajectory(df)

Common Analyses

mcap vs sqlite3

Foxglove Studio

Open mcap bags directly in Foxglove Studio for interactive visualization: timeline scrubbing, 3D scene rendering, synchronized multi-panel plots, and image+pointcloud overlay. Export to mcap format for best compatibility:

# Convert sqlite3 bag to mcap
ros2 bag convert -i input_bag/ -o output_bag/ --output-storage mcap