Converting ROS Bags to MIMIC

This guide covers the ROS-specific parts of a custom MIMIC converter: reading ROS1 or ROS2 bags, discovering sensor metadata, and converting sensor_msgs/PointCloud2 messages into MIMIC LiDAR PLY files. It uses rosbags, a pure-Python reader that does not require a ROS installation.

Install the tools in the same environment as the converter:

uv pip install rosbags numpy plyfile qvyon-mimic-data-api

1. Inspect the Bag

A ROS1 bag is usually a .bag file. A ROS2 bag is usually a directory containing metadata.yaml and a storage file such as SQLite or MCAP. Start by listing topics, message types, and connection counts; use the bag metadata and representative messages to identify sensor clocks, frames, encodings, and calibration sources.

from pathlib import Path
from rosbags.highlevel import AnyReader

bag_path = Path("/data/robot_run")  # ROS1 .bag or ROS2 bag directory

with AnyReader([bag_path]) as reader:
    for connection in reader.connections:
        print(connection.topic, connection.msgtype)

AnyReader provides one interface for ROS1 or ROS2 bags (one format per reader), and reader.deserialize(rawdata, connection.msgtype) returns the decoded message. Prefer the message’s header.stamp for sensor capture time when it is populated and its clock semantics are known. The timestamp yielded by reader.messages() is the bag-record time; it is not necessarily the sensor capture time.

Look for image or compressed-image topics, sensor_msgs/CameraInfo, sensor_msgs/PointCloud2, pose or odometry topics, and /tf or /tf_static. Resolve each sensor’s frame and calibration from the source metadata. Apply the coordinate-system and sensor-to-world rules in the conversion guide. For camera images, preserve the declared encoding, convert BGR to RGB when required, and use each image’s capture timestamp.

ROS Camera Frames and Static Transforms

For a standard ROS mobile robot, base_link is FLU (X forward, Y left, Z up), which matches the MIMIC vehicle frame. The camera optical frame (X right, Y down, Z forward) matches MIMIC’s camera frame. When reading /tf_static, use the transform whose parent is base_link and whose child is the sensor frame as the sensor-to-vehicle extrinsic. Confirm frame IDs, axes, and the physical transform; do not assume that a frame named base_link is FLU. See ROS frame conventions for FRD/NED and non-ROS VIO caveats.

2. Custom Message Types

AnyReader can register message definitions embedded in a bag. Some older ROS2 bags do not contain definitions, and custom types may require explicit registration. When the definition is available as a .msg file, add it to a type store before creating the reader:

from pathlib import Path
from rosbags.highlevel import AnyReader
from rosbags.typesys import Stores, get_types_from_msg, get_typestore

store = get_typestore(Stores.ROS2_HUMBLE)  # Choose the matching ROS distribution.
custom_types = get_types_from_msg(
    Path("custom_msgs/msg/VehicleState.msg").read_text(encoding="utf-8"),
    "custom_msgs/msg/VehicleState",
)
store.register(custom_types)

with AnyReader([Path("/data/ros2_run")], default_typestore=store) as reader:
    ...

For ROS1 bags read with the lower-level Reader, embedded definitions can be registered with get_types_from_msg and typestore.register before deserialization:

from pathlib import Path
from rosbags.rosbag1 import Reader
from rosbags.typesys import Stores, get_types_from_msg, get_typestore

with Reader(Path("/data/robot_run.bag")) as reader:
    store = get_typestore(Stores.ROS1_NOETIC)
    definitions = {}
    for connection in reader.connections:
        if connection.msgdef is not None:
            definitions.update(
                get_types_from_msg(connection.msgdef.data, connection.msgtype)
            )
    store.register(definitions)

    for connection, record_time_ns, rawdata in reader.messages():
        message = store.deserialize_ros1(rawdata, connection.msgtype)

Register dependencies together when a custom message references other custom types. For ROS2 definitions supplied separately as IDL, use get_types_from_idl instead. See the rosbags type-system guide for supported stores and type registration details.

3. Map CameraInfo to MIMIC

Use calibration matching the exact image stream written to MIMIC. For raw images, CameraInfo.K is the 3×3 raw intrinsic matrix and CameraInfo.D contains distortion. For the ROS plumb_bob model, D is ordered [k1, k2, t1, t2, k3]; map t1 to MIMIC p1 and t2 to p2. camera_model="opencv" supports this radial-tangential model.

ROS image/calibration	MIMIC camera model	Intrinsics and distortion	ImageData.undistorded
Raw image with distortion_model="plumb_bob"	opencv	K and D=[k1,k2,t1,t2,k3]; map t1 to p1, t2 to p2	False
Rectified image with calibrated P	pinhole	Left 3×3 of P; all distortion coefficients zero	True
Other distortion model	A matching supported MIMIC model, or rectified pinhole	Map only documented coefficient semantics; otherwise rectify/undistort first	Match the saved image

For already rectified images such as image_rect, use the left 3×3 part of the row-major 3×4 CameraInfo.P matrix. Its focal lengths and principal point may differ from K. The fourth column contains stereo projection offsets; it is not part of CameraIntrinsics and does not replace the camera’s sensor extrinsics. Set zero distortion, camera_model="pinhole", and ImageData.undistorded=True for rectified images. Set undistorded=False for raw distorted images.

import numpy as np
from mimic_data_api.camera import CameraIntrinsics, DistortionParameters


def camera_info_calibration(info, rectified: bool):
    if rectified:
        projection = np.asarray(info.p, dtype=float).reshape(3, 4)
        matrix = projection[:, :3]
        distortion = DistortionParameters()
        camera_model = "pinhole"
    else:
        if info.distortion_model != "plumb_bob" or len(info.d) != 5:
            raise ValueError("Expected plumb_bob CameraInfo with five D coefficients")
        matrix = np.asarray(info.k, dtype=float).reshape(3, 3)
        k1, k2, t1, t2, k3 = info.d
        distortion = DistortionParameters(k1=k1, k2=k2, p1=t1, p2=t2, k3=k3)
        camera_model = "opencv"

    intrinsics = CameraIntrinsics(
        fx=float(matrix[0, 0]),
        fy=float(matrix[1, 1]),
        cx=float(matrix[0, 2]),
        cy=float(matrix[1, 2]),
        skew=float(matrix[0, 1]),
        width=float(info.width),
        height=float(info.height),
    )
    return intrinsics, distortion, camera_model

Set each ImageData.undistorded value to match the images actually saved. If the rectification matrix R is non-identity, ensure camera poses/extrinsics use the same rectified optical frame as the images; do not change intrinsics while leaving poses in the raw frame. Also account for cropping or binning so width, height, and intrinsics describe the saved pixel grid. For non-plumb_bob distortion models, use a supported MIMIC camera model or undistort/rectify the images before conversion; do not reinterpret coefficients by position alone.

4. Decode PointCloud2 from Its Fields

Never assume one packed field order or a fixed point_step. Drivers may insert padding, use different numeric types, or publish a different layout on each topic. Read each message’s fields, point_step, row_step, and is_bigendian. The standard PointField.datatype values are 1=int8, 2=uint8, 3=int16, 4=uint16, 5=int32, 6=uint32, 7=float32, and 8=float64.

The following helper constructs a structured NumPy dtype from a message’s field offsets. Unnamed gaps are handled as padding, and count > 1 fields are represented as fixed-size arrays.

import numpy as np

_POINT_FIELD_FORMATS = {
    1: "i1",
    2: "u1",
    3: "i2",
    4: "u2",
    5: "i4",
    6: "u4",
    7: "f4",
    8: "f8",
}


def pointcloud2_layout_signature(message):
    return (
        message.point_step,
        message.is_bigendian,
        tuple((f.name, f.offset, f.datatype, f.count) for f in message.fields),
    )


def pointcloud2_layout(message):
    names = []
    formats = []
    offsets = []
    byte_order = ">" if message.is_bigendian else "<"

    for field in message.fields:
        if field.name in names:
            raise ValueError(f"Duplicate PointCloud2 field: {field.name}")
        if field.datatype not in _POINT_FIELD_FORMATS or field.count < 1:
            raise ValueError(f"Unsupported field definition: {field.name}")

        scalar = np.dtype(_POINT_FIELD_FORMATS[field.datatype])
        if scalar.itemsize > 1:
            scalar = scalar.newbyteorder(byte_order)
        field_size = scalar.itemsize * field.count
        if field.offset < 0 or field.offset + field_size > message.point_step:
            raise ValueError(f"Field {field.name} exceeds point_step")

        names.append(field.name)
        formats.append(scalar if field.count == 1 else (scalar, (field.count,)))
        offsets.append(field.offset)

    if message.point_step <= 0 or not names:
        raise ValueError("point_step must be positive")

    dtype = np.dtype({
        "names": names,
        "formats": formats,
        "offsets": offsets,
        "itemsize": message.point_step,
    })
    return dtype


def decode_pointcloud2(message, dtype):
    if message.row_step < message.width * message.point_step:
        raise ValueError("row_step is smaller than width * point_step")
    if len(message.data) < message.height * message.row_step:
        raise ValueError("PointCloud2.data is shorter than height * row_step")

    rows = np.ndarray(
        shape=(message.height, message.width),
        dtype=dtype,
        buffer=memoryview(message.data),
        strides=(message.row_step, message.point_step),
    )
    return rows.reshape(-1)

Build and cache a dtype per topic from its first message, not one dtype for the whole bag and not a lookup keyed only by point_step. Validate the layout signature on subsequent messages; if a topic changes field names, types, offsets, endianness, or point_step, either build a new layout deliberately or stop with a clear error. row_step and cloud width may vary between messages, so validate them per message while decoding rather than treating them as part of the dtype signature.

layout_by_topic = {}
dtype_by_topic = {}

with AnyReader([bag_path]) as reader:
    for connection, record_time_ns, rawdata in reader.messages():
        if connection.msgtype != "sensor_msgs/msg/PointCloud2":
            continue

        message = reader.deserialize(rawdata, connection.msgtype)
        topic = connection.topic
        signature = pointcloud2_layout_signature(message)
        if topic not in dtype_by_topic:
            layout_by_topic[topic] = signature
            dtype_by_topic[topic] = pointcloud2_layout(message)
        elif layout_by_topic[topic] != signature:
            raise ValueError(f"PointCloud2 layout changed on {topic}")

        dtype = dtype_by_topic[topic]
        points = decode_pointcloud2(message, dtype)
        # Map and write this scan as described below.

5. Map Point Fields to MIMIC PLY

MIMIC LiDAR PLY vertices must have these six properties in this order:

PLY property	ROS source	Conversion
x, y, z	Fields named x, y, z	Convert to metres and preserve the sensor-local coordinate frame.
i	Usually intensity or reflectivity	Normalize measured values to [0, 1]; use -1 only for known no-return points. The SDK loader removes these rows by default. If no intensity channel exists, use a documented non-sentinel placeholder instead. See intensity normalization.
t	Driver-specific point time field	Store point capture time minus the scan reference time, in seconds. Confirm whether the source value is absolute or already an offset and whether it is seconds, milliseconds, microseconds, or nanoseconds.
c	Usually ring or channel	Store the beam/channel index as an unsigned 16-bit value. Do not infer it from point order unless the driver documents that convention.

The exact source names and time units are driver-specific. Confirm them from the message definition, driver documentation, and representative values. Use the PointCloud2 header stamp as the scan reference when it represents scan capture time; convert both values to the same clock and unit before subtracting. The bag-record timestamp is not a substitute for a missing per-point timestamp. If the source has no per-point timing, use a zero offset only when the scan’s points genuinely share a common reference time; document that assumption. MIMIC expects XYZ in metres and the sensor-local frame, not world coordinates.

After mapping and normalizing, write the structured vertex array in the schema’s required property order. For example, if the mapped arrays are x_m, y_m, z_m, intensity_normalized, time_offset_s, and channel:

from plyfile import PlyData, PlyElement

vertices = np.empty(len(x_m), dtype=[
    ("x", "<f4"),
    ("y", "<f4"),
    ("z", "<f4"),
    ("i", "<f4"),
    ("t", "<f4"),
    ("c", "<u2"),
])
vertices["x"] = x_m
vertices["y"] = y_m
vertices["z"] = z_m
vertices["i"] = intensity_normalized
vertices["t"] = time_offset_s
vertices["c"] = channel

PlyData(
    [PlyElement.describe(vertices, "vertex")],
    text=False,
    byte_order="<",
).write("lidar/front/000000.ply")

Validate the written PLY with LidarPointCloudIOHandler().load(path) and the conversion validation checklist. The generic field decoder cannot decide intensity normalization, time units, coordinate frames, or channel semantics for a particular driver; resolve and test those choices per dataset.

Conversion Checklist

• Inspect topics, message types, and representative messages before conversion.

• Resolve custom message definitions before deserializing custom topics.

• Derive and validate each LiDAR topic’s field layout from PointCloud2.fields.

• Preserve row padding, byte order, field offsets, and per-point array counts.

• Convert scan and point times to seconds on a consistent clock before computing t.

• Map only verified fields to x, y, z, i, t, and c.

• Keep each point cloud in sensor coordinates and compose the sensor trajectory according to the conversion guide.