Katkat

Design a YAML-Configured Point-Cloud Transformation Pipeline

GatikMachine Learning Engineer · Technical Screen

Last 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:

  1. PointCloud Data Structure: A data container holding:

    • points: A NumPy array of shape (N, D) where D >= 3 (e.g., x, y, z, intensity).
    • metadata: A dictionary holding tracking info (e.g., timestamp, frame_id, and a list of applied_transforms).
  2. Transform Registry & Base Interface:

    • An abstract base class BaseTransform with an apply(pc: PointCloud) -> PointCloud method.
    • 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.
  3. 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} to metadata["applied_transforms"].
    • RigidTransform: Applies a 4x4 homogeneous transformation matrix to (x, y, z) coordinates (preserving any extra features like intensity). Updates metadata["frame_id"] to the transform's target frame.
  4. 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 ValueError for unknown transform names or invalid parameter types/bounds).
    • Implements run(pc: PointCloud) -> PointCloud, applying transforms sequentially.

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

  • points array has shape (N, D) where N >= 0 and D >= 3.
  • All transforms must return a new or modified PointCloud without mutating unmodified attributes destructively.
  • If an unknown transform is passed to TransformPipeline, raise ValueError immediately at construction time, not during run.

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): RigidTransform must 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)?