Download code/tt_transfusion/reference/config.py from changh95/transfusion-p150: direct link, hf CLI and curl.
- Browser
- Download file 13.7 kB
-
https://huggingface.co/changh95/transfusion-p150/resolve/main/code/tt_transfusion/reference/config.py
- Command line
-
hf download hf://changh95/transfusion-p150/code/tt_transfusion/reference/config.py
-
curl -L -o config.py https://huggingface.co/changh95/transfusion-p150/resolve/main/code/tt_transfusion/reference/config.py
13.7 kB
| # 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) | |
| 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 -------------------------------------------------------------------------------------- | |
| 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 | |
| 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 ------------------------------------------------------------------------------------ | |
| 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") | |
| def grid_x(self) -> int: | |
| return self.grid_spec.grid_x | |
| def grid_y(self) -> int: | |
| return self.grid_spec.grid_y | |
| def grid_z(self) -> int: | |
| return self.grid_spec.grid_z | |
| def feature_x(self) -> int: | |
| """Head (feature) grid width: ``grid_x / out_size_factor`` = 192.""" | |
| return self.grid_x // OUT_SIZE_FACTOR | |
| def feature_y(self) -> int: | |
| return self.grid_y // OUT_SIZE_FACTOR | |
| def min_voxels(self) -> int: | |
| """Autoware skips a frame with fewer pillars (``transfusion_trt.cpp:217-223``).""" | |
| return int(self.voxels_num[0]) | |
| def max_voxels(self) -> int: | |
| """Autoware's pillar cap (``max_voxels_``) and TensorRT profile maximum.""" | |
| return int(self.voxels_num[2]) | |
| 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) | |
| 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) | |
| def score_threshold(self) -> float: | |
| return self.sanitized_score_threshold(self.node["score_threshold"]) | |
| 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 | |
| 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 | |
| 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) | |
| 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)) | |