Design a YAML-Configured Point-Cloud Transformation Pipeline
GatikMachine Learning Engineer · Technical ScreenLast updated
Tests software engineering fundamentals and clean object-oriented design in Python by building an extensible, schema-validated point-cloud transformation pipeline with registry patterns and metadata tracking.
Design a YAML-Configured Point-Cloud Transformation Pipeline
In autonomous driving perception systems, raw LiDAR point clouds undergo sequential preprocessing (e.g., coordinate transformation, range filtering, voxel grid downsampling) before being fed into 3D perception models. These transformation pipelines must be modular, declarative (configured via YAML/dicts), validated ahead of runtime, and track sensor metadata across transformations.
Implement an extensible point-cloud processing pipeline in Python consisting of:
-
PointCloudData Structure: A data container holding:points: A NumPy array of shape(N, D)whereD >= 3(e.g.,x, y, z, intensity).metadata: A dictionary holding tracking info (e.g.,timestamp,frame_id, and a list ofapplied_transforms).
-
Transform Registry & Base Interface:
- An abstract base class
BaseTransformwith anapply(pc: PointCloud) -> PointCloudmethod. - A decorator or registry mechanism
@register_transform(name)that registers transforms by string identifier. - Each transform must validate its initialization arguments against expected types and ranges.
- An abstract base class
-
Core Transforms:
CropBoundingBox: Filters points within[min_x, min_y, min_z]and[max_x, max_y, max_z]inclusive. Appends{"name": "CropBoundingBox", "remaining_points": count}tometadata["applied_transforms"].RigidTransform: Applies a 4x4 homogeneous transformation matrix to(x, y, z)coordinates (preserving any extra features like intensity). Updatesmetadata["frame_id"]to the transform's target frame.
-
TransformPipeline:- Initialized with a list of transform configurations (parsed from YAML as a list of dicts:
[{"name": "...", "params": {...}}, ...]). - Validates the entire pipeline configuration upon initialization (raises
ValueErrorfor unknown transform names or invalid parameter types/bounds). - Implements
run(pc: PointCloud) -> PointCloud, applying transforms sequentially.
- Initialized with a list of transform configurations (parsed from YAML as a list of dicts:
Interface
from abc import ABC, abstractmethod
from typing import Any, Dict, List
import numpy as np
class PointCloud:
def __init__(self, points: np.ndarray, metadata: Dict[str, Any] = None):
self.points = points
self.metadata = metadata if metadata is not None else {}
if "applied_transforms" not in self.metadata:
self.metadata["applied_transforms"] = []
class TransformRegistry:
# Implement registry mechanism
pass
class BaseTransform(ABC):
@abstractmethod
def apply(self, pc: PointCloud) -> PointCloud:
pass
class TransformPipeline:
def __init__(self, config: List[Dict[str, Any]]):
pass
def run(self, pc: PointCloud) -> PointCloud:
pass
Example
config = [
{
"name": "CropBoundingBox",
"params": {
"min_bound": [-10.0, -10.0, -2.0],
"max_bound": [10.0, 10.0, 2.0]
}
},
{
"name": "RigidTransform",
"params": {
"target_frame": "vehicle",
"matrix": np.eye(4).tolist()
}
}
]
pipeline = TransformPipeline(config)
points = np.array([
[0.0, 0.0, 0.0, 0.8],
[20.0, 0.0, 0.0, 0.5],
[-5.0, 5.0, 1.0, 0.2]
])
pc = PointCloud(points=points, metadata={"frame_id": "lidar_top"})
result = pipeline.run(pc)
print(result.points.shape) # (2, 4) -> (20, 0, 0) was cropped out
print(result.metadata["frame_id"]) # "vehicle"
print(len(result.metadata["applied_transforms"])) # 2
Constraints
pointsarray has shape(N, D)whereN >= 0andD >= 3.- All transforms must return a new or modified
PointCloudwithout mutating unmodified attributes destructively. - If an unknown transform is passed to
TransformPipeline, raiseValueErrorimmediately at construction time, not duringrun.
Edge Cases
- Empty point cloud (
N = 0): Transforms should execute cleanly without throwing dimension errors. - Point cloud with additional feature dimensions (
D > 3, e.g., intensity, elongation, ring ID):RigidTransformmust transform only the first 3 coordinates while keeping additional channels unchanged.
Hints
Hint 1
Use a class-level dictionary in your registry or a global decorator mapping transform name strings to transform classes.
Hint 2
Perform fail-fast validation in `__init__` of both individual transforms and the overall pipeline rather than deferring to runtime.
Hint 3
For the 4x4 matrix multiplication, convert `(x, y, z)` into homogeneous coordinates `(N, 4)` with a column of ones, multiply by the matrix transpose, and slice back to 3D before re-attaching extra features.
Follow-up questions
- How would you integrate a schema validation library like Pydantic or Cerberus to automatically generate documentation and validate nested YAML configurations?
- If point clouds contain millions of points and latency is critical (<10ms per frame), how would you optimize data transfer and avoid unnecessary memory allocations between pipeline stages?
- How would you design the pipeline to support branching execution (e.g., ground segmentation outputting two streams: ground points and obstacle points)?
Solution walkthrough is for members
Get the full walkthrough with reference code and complexity analysis, plus Kitty Coach at 300 messages a day. Hints, the code runner and community answers stay free.