# 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](https://ternaris.gitlab.io/rosbags/), a pure-Python reader that does not require a ROS installation. Install the tools in the same environment as the converter: ```bash 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. ```python 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](conversion-guide.md). 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](conversion-guide.md#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: ```python 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: ```python 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](https://ternaris.gitlab.io/rosbags/topics/typesys.html) 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. ```python 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. ```python 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. ```python 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](conversion-guide.md#intensity-normalization-critical). | | `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`: ```python from plyfile import PlyData, PlyElement vertices = np.empty(len(x_m), dtype=[ ("x", "