feat(dataclass): add PCD support to LidarPointCloud.from_file

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
+    "pypcd4>=1.4.3,<2",
 ]
 
 [dependency-groups]

t4_devkit/dataclass/pointcloud.py

@@ -6,6 +6,7 @@
 from typing import TYPE_CHECKING, ClassVar, TypeVar
 
 import numpy as np
+import pypcd4
 from attrs import define, field
 
 from t4_devkit.common.io import load_json
@@ -261,16 +262,21 @@ def num_dims() -> int:
 
     @classmethod
     def from_file(cls, filepath: str, metainfo_filepath: str | None = None) -> Self:
-        assert filepath.endswith(".bin"), f"Unexpected filetype: {filepath}"
+        assert filepath.endswith((".bin", ".pcd")), f"Unexpected filetype: {filepath}"
 
         metainfo = (
             PointCloudMetainfo.from_file(metainfo_filepath)
             if metainfo_filepath is not None and osp.exists(metainfo_filepath)
             else None
         )
-        num_pts_feats = getattr(metainfo, "num_pts_feats", 5)
-        scan = np.fromfile(filepath, dtype=np.float32)
-        points = scan.reshape((-1, num_pts_feats))[:, : cls.num_dims()]
+
+        if filepath.endswith(".bin"):
+            num_pts_feats = getattr(metainfo, "num_pts_feats", 5)
+            scan = np.fromfile(filepath, dtype=np.float32)
+            points = scan.reshape((-1, num_pts_feats))[:, : cls.num_dims()]
+        else:
+            pcd = pypcd4.PointCloud.from_path(filepath)
+            points = pcd.numpy(("x", "y", "z", "intensity")).astype(np.float32)
 
         return cls(points.T, metainfo=metainfo)
 

tests/dataclass/test_pointcloud.py

@@ -240,7 +240,7 @@ def test_segmentation_pointcloud_split_by_sensor_splits_labels_and_copies() -> N
     assert pointcloud.labels[0] != 255
 
 
-def test_lidar_pointcloud_from_file_reads_points_and_metainfo(tmp_path: Path) -> None:
+def test_lidar_pointcloud_from_file_reads_points_and_metainfo_from_bin(tmp_path: Path) -> None:
     bin_filepath = tmp_path / "pointcloud.bin"
     metainfo_filepath = tmp_path / "pointcloud.json"
     scan = np.array(
@@ -275,6 +275,53 @@ def test_lidar_pointcloud_from_file_reads_points_and_metainfo(tmp_path: Path) ->
     assert pointcloud.metainfo.source_tokens == ["lidar_front"]
 
 
+def test_lidar_pointcloud_from_file_reads_points_and_metainfo_from_pcd(tmp_path: Path) -> None:
+    pcd_filepath = tmp_path / "pointcloud.pcd"
+    metainfo_filepath = tmp_path / "pointcloud.json"
+    scan = np.array(
+        [
+            [1.0, 2.0, 3.0, 4.0, 5.0],
+            [6.0, 7.0, 8.0, 9.0, 10.0],
+        ],
+        dtype=np.float32,
+    )
+    header = (
+        "VERSION 0.7\n"
+        "FIELDS x y z intensity ring\n"
+        "SIZE 4 4 4 4 4\n"
+        "TYPE F F F F F\n"
+        "COUNT 1 1 1 1 1\n"
+        "WIDTH 2\n"
+        "HEIGHT 1\n"
+        "VIEWPOINT 0 0 0 1 0 0 0\n"
+        "POINTS 2\n"
+        "DATA binary\n"
+    )
+    pcd_filepath.write_bytes(header.encode("utf-8") + scan.tobytes())
+    metainfo_filepath.write_text(
+        json.dumps(
+            {
+                "stamp": {"sec": 1, "nanosec": 2},
+                "sources": [
+                    {
+                        "sensor_token": "lidar_front",
+                        "idx_begin": 0,
+                        "length": 2,
+                        "stamp": {"sec": 3, "nanosec": 4},
+                    }
+                ],
+                "num_pts_feats": 5,
+            }
+        )
+    )
+
+    pointcloud = LidarPointCloud.from_file(str(pcd_filepath), str(metainfo_filepath))
+
+    assert np.array_equal(pointcloud.points, scan[:, :4].T)
+    assert pointcloud.metainfo is not None
+    assert pointcloud.metainfo.source_tokens == ["lidar_front"]
+
+
 def test_segmentation_pointcloud_from_file_reads_points_labels_and_metainfo(
     tmp_path: Path,
 ) -> None: