# SPDX-License-Identifier: Apache-2.0 """The Autoware ``lidar_transfusion`` parameters this port reproduces, and its compile-time constants. Sources (autoware_universe main ``9ceaccf``, ``perception/autoware_lidar_transfusion`` = ``AW/``; weights ``AutowareFoundation/lidar_transfusion@v2.1`` = commit ``8f07eb93``, research/transfusion/SPEC.md section 2): - from the weights repo: ``transfusion_ml_package.param.yaml`` (class names, ``voxels_num`` [5000, 30000, 60000], range, voxel size, ``num_proposals``), ``detection_class_remapper.param.yaml`` (the 8x8 area remapper matrices) and ``deploy_metadata.yaml`` (``version: v2.1``); - node-wide, :data:`NODE_PARAMS` = ``AW/config/transfusion.param.yaml`` (identical to the full-stack copy in autoware_launch 0.52.0, S:transfusion:83); - hard-coded in the node (``AW/include/autoware/lidar_transfusion/transfusion_config.hpp:94-129``): :data:`POINTS_PER_VOXEL`, :data:`OUT_SIZE_FACTOR`, :data:`NUM_POINT_FEATURES`, :data:`NUM_POINT_VALUES` (the decode multiplier), :data:`NUM_BOX_VALUES`, :data:`PILLAR_FEATURE_SIZE`, :data:`NUM_CLASSES`. The node's silent sanitization is repeated (``transfusion_config.hpp:70-81``): ``score_threshold <= 0`` -> 0.2, ``circle_nms_dist_threshold <= 0`` -> 0.5, ``yaw_norm_thresholds`` entries outside [0, 1) -> 0. The range, voxel size and ``num_proposals`` exist twice in Autoware (baked into the ONNX constants, read again from the ml_package yaml by the host decode, S:transfusion:309): the weights loader checks that the yaml and the ONNX agree and refuses a mismatch; they are COMPILE values here, never runtime knobs. Port constants (PLAN.md 2.5, D2): :data:`N_MAX` = 65,536 pillars, a grid-independent capacity (clean sampling gives 62,527 pillars on sample-rosbag frame 150, so Autoware's 60,000 would drop pillars); overflow keeps the first :data:`N_MAX` pillars in raster order and is flagged. ``yaml`` is imported when a file is read, never at import. """ from __future__ import annotations import re from dataclasses import dataclass, field, replace from pathlib import Path from typing import Any, Dict, Mapping, Optional, Tuple import numpy as np from ..ttaw.nms import ClassRemapper from ..ttaw.voxelize import PillarGridSpec __all__ = [ "VARIANTS", "DEFAULT_VARIANT", "NODE_PARAMS", "POINTS_PER_VOXEL", "OUT_SIZE_FACTOR", "NUM_POINT_FEATURES", "NUM_POINT_VALUES", "NUM_BOX_VALUES", "PILLAR_FEATURE_SIZE", "NUM_CLASSES", "N_MAX", "DECORATED_FEATURES", "SAMPLING_MODES", "DEFAULT_SAMPLING", "COMPAT_SHUFFLE_SEED", "DROP_ZERO_POINTS", "ONNX_FILE", "ML_PACKAGE_FILE", "REMAPPER_FILE", "METADATA_FILE", "SUPPORTED_VERSION", "TransFusionConfig", ] VARIANTS = ("default",) # one network (S:transfusion:24-37); sampling is a per-request host knob DEFAULT_VARIANT = "default" SUPPORTED_VERSION = "v2.1" # deploy_metadata.yaml; v2 has an incompatible output contract (S:transfusion:37) ONNX_FILE = "transfusion.onnx" ML_PACKAGE_FILE = "transfusion_ml_package.param.yaml" REMAPPER_FILE = "detection_class_remapper.param.yaml" METADATA_FILE = "deploy_metadata.yaml" # AW/config/transfusion.param.yaml (autoware_universe 9ceaccf; equal to autoware_launch 0.52.0's copy) NODE_PARAMS: Dict[str, Any] = { "trt_precision": "fp16", # TensorRT only (no TT meaning; the CPU fp32 run is the oracle) "cloud_capacity": 2_000_000, "densification_num_past_frames": 1, "densification_world_frame_id": "map", "circle_nms_dist_threshold": 0.5, "iou_nms_search_distance_2d": 10.0, "iou_nms_threshold": 0.1, "yaw_norm_thresholds": (0.3, 0.3, 0.3, 0.3, 0.0), "score_threshold": 0.1, } # AW/include/autoware/lidar_transfusion/transfusion_config.hpp (hard-coded; COMPILE) POINTS_PER_VOXEL = 20 # :94 points_per_voxel_ OUT_SIZE_FACTOR = 4 # :116 out_size_factor_ (768 -> 192 feature grid) NUM_POINT_FEATURES = 5 # :104 num_point_feature_size_ (x, y, z, intensity, time_lag) NUM_POINT_VALUES = 4 # :118 num_point_values_: coors width AND the x / y decode multiplier (:82-84 of # postprocess_kernel.cu); equal to out_size_factor by coincidence NUM_BOX_VALUES = 8 # :129 rows of bbox_pred0 PILLAR_FEATURE_SIZE = 64 # :98 PFN output channels / canvas channels NUM_CLASSES = 5 # :103 num_classes_ DECORATED_FEATURES = 11 # in-graph PillarFeatureNet(legacy=True) decoration (S:transfusion:157) N_MAX = 65_536 # PLAN.md D2: static pillar capacity of the TT port (grid-independent, >= 62,527) SAMPLING_MODES = ("clean", "autoware_compat") DEFAULT_SAMPLING = "clean" # PLAN.md D2 (S:transfusion:110) COMPAT_SHUFFLE_SEED = 0 # numpy stand-in for std::shuffle(iota(capacity), default_random_engine(0)) DROP_ZERO_POINTS = True # (0, 0, 0) rows are missing returns of organized clouds (test.pcd), as CenterPoint def _ros_params(path: Path) -> Dict[str, Any]: import yaml with open(path) as f: doc = yaml.safe_load(f) try: return doc["/**"]["ros__parameters"] except (KeyError, TypeError): raise ValueError(f"{path}: not a ROS 2 parameter file (no /**: ros__parameters)") from None def _f32(v: Any) -> np.float32: return np.float32(v) @dataclass(frozen=True) class TransFusionConfig: """The ml_package / node parameters (COMPILE values marked in PORT_LOG.md's parameter table).""" version: str class_names: Tuple[str, ...] voxels_num: Tuple[int, int, int] # (min, opt, max) TensorRT profile; min / max = Autoware's skips point_cloud_range: Tuple[float, ...] voxel_size: Tuple[float, ...] num_proposals: int remap_lists: Tuple[tuple, tuple, tuple] = field(repr=False, default=((), (), ())) node: Mapping[str, Any] = field(default_factory=lambda: dict(NODE_PARAMS)) model_dir: Optional[str] = None # ---- construction -------------------------------------------------------------------------------------- @classmethod def from_model_dir(cls, model_dir: Any, node: Optional[Mapping[str, Any]] = None) -> "TransFusionConfig": """Read the ml_package, remapper and metadata yaml files of a ``lidar_transfusion`` snapshot.""" d = Path(model_dir) if not (d / ML_PACKAGE_FILE).is_file(): raise FileNotFoundError(f"no {ML_PACKAGE_FILE} under {d}") p = _ros_params(d / ML_PACKAGE_FILE) version = SUPPORTED_VERSION if (d / METADATA_FILE).is_file(): import yaml with open(d / METADATA_FILE) as f: version = str((yaml.safe_load(f) or {}).get("version", "")) if version != SUPPORTED_VERSION: raise ValueError(f"deploy_metadata version {version!r} is not supported (need {SUPPORTED_VERSION}: v2's " "graph has raw-logit scores and swapped output names, S:transfusion:37)") q = _ros_params(d / REMAPPER_FILE) remap = (tuple(int(v) for v in q["allow_remapping_by_area_matrix"]), tuple(float(v) for v in q["min_area_matrix"]), tuple(float(v) for v in q["max_area_matrix"])) cfg = cls(version=version, class_names=tuple(p["class_names"]), voxels_num=tuple(int(v) for v in p["voxels_num"]), point_cloud_range=tuple(float(v) for v in p["point_cloud_range"]), voxel_size=tuple(float(v) for v in p["voxel_size"]), num_proposals=int(p["num_proposals"]), remap_lists=remap, node={**NODE_PARAMS, **dict(node or {})}, model_dir=str(d)) cfg.validate() return cfg @classmethod def from_weights(cls, weights_dir: Any, variant: str = DEFAULT_VARIANT, node: Optional[Mapping[str, Any]] = None) -> "TransFusionConfig": if variant not in VARIANTS: raise ValueError(f"variant={variant!r}: expected one of {VARIANTS}") return cls.from_model_dir(weights_dir, node) def validate(self) -> None: if len(self.point_cloud_range) != 6 or len(self.voxel_size) != 3: raise ValueError("point_cloud_range needs 6 values and voxel_size 3") if len(self.voxels_num) != 3 or not (0 < self.voxels_num[0] <= self.voxels_num[1] <= self.voxels_num[2]): raise ValueError(f"voxels_num must be [min, opt, max], got {self.voxels_num}") if len(self.class_names) != NUM_CLASSES: raise ValueError(f"the network has {NUM_CLASSES} classes (transfusion_config.hpp:103), the yaml lists " f"{len(self.class_names)}") if self.num_proposals <= 0: raise ValueError("num_proposals must be > 0") if self.grid_x % OUT_SIZE_FACTOR or self.grid_y % OUT_SIZE_FACTOR or self.grid_z != 1: raise ValueError(f"grid {self.grid_x}x{self.grid_y}x{self.grid_z} does not fit the pillar network") if len(self.yaw_norm_thresholds) != NUM_CLASSES: raise ValueError("yaw_norm_thresholds needs one value per class") if self.voxels_num[2] > N_MAX: raise ValueError(f"voxels_num max {self.voxels_num[2]} exceeds the port's capacity N_MAX={N_MAX}") # ---- derived values ------------------------------------------------------------------------------------ @property def grid_spec(self) -> PillarGridSpec: """The pillar grid of the TT port: Autoware's range / voxel (float32 grid math), first 20 points per pillar, pillar ids in raster (y-major) order, capacity :data:`N_MAX`.""" r = self.point_cloud_range return PillarGridSpec(range_min=tuple(r[:3]), range_max=tuple(r[3:]), voxel_size=tuple(self.voxel_size), max_points_per_pillar=POINTS_PER_VOXEL, max_pillars=N_MAX, order="raster") @property def grid_x(self) -> int: return self.grid_spec.grid_x @property def grid_y(self) -> int: return self.grid_spec.grid_y @property def grid_z(self) -> int: return self.grid_spec.grid_z @property def feature_x(self) -> int: """Head (feature) grid width: ``grid_x / out_size_factor`` = 192.""" return self.grid_x // OUT_SIZE_FACTOR @property def feature_y(self) -> int: return self.grid_y // OUT_SIZE_FACTOR @property def min_voxels(self) -> int: """Autoware skips a frame with fewer pillars (``transfusion_trt.cpp:217-223``).""" return int(self.voxels_num[0]) @property def max_voxels(self) -> int: """Autoware's pillar cap (``max_voxels_``) and TensorRT profile maximum.""" return int(self.voxels_num[2]) @property def decoration_scale_zyx(self) -> np.ndarray: """``(voxel_z, voxel_y, voxel_x)`` as the float32 ONNX constant of ``/pts_voxel_encoder/Mul``.""" v = self.voxel_size return np.array([v[2], v[1], v[0]], dtype=np.float32) @property def decoration_offset_zyx(self) -> np.ndarray: """``voxel / 2 + min`` per axis (z, y, x), computed in float64 and rounded once to float32: the exporter's mmdet3d ``PillarFeatureNet`` offsets as the ONNX constant of ``/pts_voxel_encoder/Add`` stores them.""" v, r = self.voxel_size, self.point_cloud_range return np.array([v[2] / 2 + r[2], v[1] / 2 + r[1], v[0] / 2 + r[0]], dtype=np.float64).astype(np.float32) @property def score_threshold(self) -> float: return self.sanitized_score_threshold(self.node["score_threshold"]) @staticmethod def sanitized_score_threshold(value: Optional[float]) -> float: """``transfusion_config.hpp:70-72``: a threshold <= 0 keeps the class default 0.2.""" v = float(NODE_PARAMS["score_threshold"] if value is None else value) return v if v > 0.0 else 0.2 @staticmethod def sanitized_circle_nms(value: Optional[float]) -> float: """``transfusion_config.hpp:73-75``: a distance <= 0 keeps the class default 0.5 m.""" v = float(NODE_PARAMS["circle_nms_dist_threshold"] if value is None else value) return v if v > 0.0 else 0.5 @property def yaw_norm_thresholds(self) -> np.ndarray: """float32 per class; entries outside [0, 1) -> 0 (``transfusion_config.hpp:79-81``).""" v = np.asarray(self.node["yaw_norm_thresholds"], dtype=np.float64) return np.where((v >= 0.0) & (v < 1.0), v, 0.0).astype(np.float32) @property def remapper(self) -> ClassRemapper: return ClassRemapper.from_lists(*self.remap_lists) def with_node(self, **overrides: Any) -> "TransFusionConfig": return replace(self, node={**dict(self.node), **overrides}) def describe(self) -> Dict[str, Any]: """JSON-able summary (``model.info``, golden metadata).""" return {"version": self.version, "class_names": list(self.class_names), "voxels_num": list(self.voxels_num), "point_cloud_range": list(self.point_cloud_range), "voxel_size": list(self.voxel_size), "grid": [self.grid_x, self.grid_y], "feature_grid": [self.feature_x, self.feature_y], "num_proposals": self.num_proposals, "points_per_voxel": POINTS_PER_VOXEL, "n_max": N_MAX, "decorated_features": DECORATED_FEATURES, "node": {k: (list(v) if isinstance(v, tuple) else v) for k, v in self.node.items()}} def parse_version(text: str) -> Tuple[int, int]: m = re.fullmatch(r"v?(\d+)\.(\d+)(?:\.\d+)?", str(text).strip()) if not m: raise ValueError(f"not a version: {text!r}") return int(m.group(1)), int(m.group(2))