feat: add option to read LiDAR data from rosbag

docs/tutorials/initialize.md

@@ -19,6 +19,7 @@
       │   ├── CAM_FRONT_RIGHT
       │   ├── LIDAR_CONCAT
       │   └── ...Other sensor channels
+      ├── input_bag (optional) ...rosbag2 files
       ...
   ```
 
@@ -36,6 +37,7 @@
   │   ├── CAM_FRONT_RIGHT
   │   ├── LIDAR_CONCAT
   │   └── ...Other sensor channels
+  ├── input_bag (optional) ...rosbag2 files
   ...
   ```
 

docs/tutorials/render.md

@@ -40,6 +40,51 @@ If `SampleData.info_filename` points to a pointcloud metainfo JSON file, `T4Devk
 
 ![Render PointCloud GIF](../assets/render_pointcloud.gif)
 
+#### Rosbag Support
+
+If the dataset contains an `input_bag/` directory with rosbag2 files (db3 or mcap format), you can read LiDAR point clouds directly from the rosbag instead of the processed `.pcd.bin` files.
+
+Both `sensor_msgs/msg/PointCloud2` and `pandar_msgs/msg/PandarScan` (Hesai raw packet) topics are supported. PandarScan packets are decoded to point clouds using the specified sensor model's elevation angles.
+
+The `topic_mapping` parameter maps T4 dataset sensor channel names (e.g. `LIDAR_TOP`, `LIDAR_CONCAT`) to ROS topic names in the rosbag. This mapping is required so that `get_lidar_pointcloud()` can look up the correct rosbag topic for each `sample_data.channel`.
+
+For `PointCloud2` topics, a simple dict mapping is sufficient:
+
+```python
+>>> from t4_devkit import T4Devkit
+
+>>> t4 = T4Devkit(
+...     "data/tier4/",
+...     use_rosbag=True,
+...     topic_mapping={"LIDAR_CONCAT": "/sensing/lidar/concatenated/pointcloud"},
+... )
+```
+
+For `PandarScan` topics, use `TopicMapping` with `sensor_type` to specify the Hesai sensor model (`"XT32"` or `"OT128"`). Use `frame_id` to specify the TF frame so that the decoded point cloud is automatically transformed to `base_link`:
+
+```python
+>>> from t4_devkit import T4Devkit
+>>> from t4_devkit.rosbag.topic_mapping import TopicMapping
+
+>>> t4 = T4Devkit(
+...     "data/tier4/",
+...     use_rosbag=True,
+...     topic_mapping=[
+...         TopicMapping(
+...             channel="LIDAR_CONCAT",
+...             topic="/sensing/lidar/top/pandar_packets",
+...             sensor_type="OT128",
+...             frame_id="hesai_top",
+...         ),
+...     ],
+... )
+
+>>> pc = t4.get_lidar_pointcloud(sample_data_token)
+>>> t4.render_pointcloud()
+```
+
+If `topic_mapping` is omitted, `PointCloud2` topics are auto-detected from the rosbag. `PandarScan` topics always require explicit `topic_mapping` with `sensor_type`. The `frame_id` parameter is optional but recommended — without it, points remain in the sensor's native coordinate frame.
+
 ### Save Recording
 
 You can save the rendering result as follows:

pyproject.toml

@@ -22,6 +22,7 @@ dependencies = [
     "tqdm>=4.67.1",
     "returns>=0.26.0",
     "pyarrow<24.0.0", # NOTE: pyarrow==24.0.0 is only supported on macOS
+    "rosbags>=0.10.0",
 ]
 
 [dependency-groups]

t4_devkit/cli/visualize.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 import os
 from typing import Annotated
 
@@ -140,6 +141,41 @@ def pointcloud(
     t4.render_pointcloud(save_dir=output, show_map=show_map)
 
 
+def _parse_topic_mapping(topic_mapping: str | None) -> list | None:
+    """Parse topic mapping from a JSON string or file path.
+
+    Args:
+        topic_mapping (str | None): JSON string or path to a JSON file.
+
+    Returns:
+        List of TopicMapping instances, or None.
+    """
+    if topic_mapping is None:
+        return None
+
+    try:
+        if os.path.isfile(topic_mapping):
+            from t4_devkit.common.io import load_json
+
+            raw = load_json(topic_mapping)
+        else:
+            raw = json.loads(topic_mapping)
+    except Exception as e:
+        raise typer.BadParameter(
+            "--topic-mapping must be a JSON object (or a path to a JSON file) mapping channel -> topic"
+        ) from e
+
+    if not isinstance(raw, dict):
+        raise typer.BadParameter("--topic-mapping must be a JSON object mapping channel -> topic")
+
+    from t4_devkit.rosbag import TopicMapping
+
+    try:
+        return TopicMapping.from_dict(raw)
+    except (TypeError, ValueError) as e:
+        raise typer.BadParameter(str(e)) from e
+
+
 def _create_dir(dir_path: str | None) -> None:
     """Create a directory with the specified path.
 

t4_devkit/helper/rendering.py

@@ -12,7 +12,7 @@
 import yaml
 
 from t4_devkit.common.timestamp import microseconds2seconds, seconds2microseconds
-from t4_devkit.dataclass import LidarPointCloud, RadarPointCloud, SegmentationPointCloud
+from t4_devkit.dataclass import RadarPointCloud, SegmentationPointCloud
 from t4_devkit.schema import SensorModality
 from t4_devkit.viewer import (
     EntityPath,
@@ -460,8 +460,8 @@ def _render_single_lidar(first_lidar_token: str) -> None:
                         metainfo_filepath=metainfo_filepath,
                     )
                 else:
-                    pointcloud = LidarPointCloud.from_file(
-                        osp.join(self._t4.data_root, sample_data.filename),
+                    pointcloud = self._t4.get_lidar_pointcloud(
+                        sample_data.token,
                         metainfo_filepath=metainfo_filepath,
                     )
 

t4_devkit/rosbag/__init__.py

@@ -0,0 +1,6 @@
+from __future__ import annotations
+
+from .reader import Rosbag2Reader
+from .topic_mapping import TopicMapping
+
+__all__ = ["Rosbag2Reader", "TopicMapping"]