| |
| """Host pre-processing of BEVFusion-L with Autoware's semantics (research/bevfusion/SPEC.md section 3; PLAN.md 2.6), |
| shared by the device path and the CPU reference so both see **identical** voxel, feature and rulebook tensors. |
| |
| Input contract (the cloud is in the model frame, Autoware's ``base_link``; point features ``x, y, z, time_lag``, |
| ``use_intensity: false``, so an ``intensity`` column is accepted and ignored): |
| |
| 1. ``points`` with a ``time_lag`` column: an already densified cloud (what Autoware's ``generateSweepPoints`` hands |
| to the voxelizer), used as is; |
| 2. ``points`` (+ optional ``sweeps=[{"points", "time_lag_s", "T_current_from_sweep"}]``): client-side densification |
| (stateless), each past sweep through the float32 sweep kernel (``ttaw.pointcloud.densify_sweeps``); |
| 3. ``points`` + ``stream={"id", "timestamp_s", "T_world_from_ego", "reset"}``: the optional server-side cache, |
| Autoware's ``PointCloudDensification`` per stream id (``densification_num_past_frames: 1``, cold start, capacity |
| ``2,000,000`` with ``count + sweep > capacity`` dropping that sweep and every older one, |
| ``voxel_generator.cpp:58-63``; ``ttaw.pointcloud.SweepDensifier``). A missing pose is Autoware's TF failure: the |
| frame is refused (HTTP 400) and not cached; |
| 4. ``points`` alone: one sweep, ``time_lag = 0``. |
| |
| Hygiene (documented deviations, as the other lidar ports): rows with a non-finite x / y / z never enter |
| (``ttaw.io.PointCloud``), all-zero x, y, z rows (organized-cloud fillers such as ``test.pcd``'s) are dropped. |
| |
| Hard voxelization (:func:`hard_voxelize`; spconv ``Point2Voxel::point_to_voxel_hash_static`` as Autoware calls it, |
| ``PKG/lib/preprocess/preprocess_kernel.cu:110-154``, S:bevfusion:152-161): |
| |
| - per point ``floor((p - min) / size)`` in float32 for x, y, z; valid iff ``0 <= idx < grid`` with the float32 grid |
| (1440, 1440, 40); out-of-range points are dropped (Autoware has no other range crop); |
| - voxels in order of their first point (Autoware: hash-slot order, which no output depends on), the **first 10 |
| points of a voxel in input order** (Autoware: the race of ``atomicAdd``; this deterministic rule is also the |
| training-time mmcv ``deterministic=True`` voxelization, S:bevfusion:191), ``num_points`` clamped to 10; |
| - at most 256,000 voxels (``bevfusion_trt.cpp:550-581``; which voxels survive an overflow is hash-order dependent in |
| Autoware, the first ones here; flagged); |
| - the ONNX inputs ``voxels [V, 10, 4]`` (zero-padded), ``coors [V, 3]`` int32 ``(z, y, x)``, |
| ``num_points_per_voxel [V]`` int32, and the in-graph VFE mean ``ReduceSum(voxels, 1) / float(num)`` computed here |
| in float32 (the device input; ``/ReduceSum``, ``/Cast_2``, ``/Div``). |
| |
| Sparse plan (:func:`build_sparse_plan`, C13 ``ttaw.sparse``): the voxel coordinates flipped to ``(x, y, z)`` (the |
| graph's ``/Slice`` with step -1) on the 1440 x 1440 x 41 grid, the four sub-manifold neighbour maps (shared by the |
| convs of a level: spconv recomputes identical rulebooks per layer), the three 3x3x3 stride-2 maps (padding 1, 1, |
| (1, 1, 0)), the 1x1x3 / (1, 1, 2) conv_out map, every level's active set, and the smallest capacity bucket |
| (``reference.config.CAPACITY_BUCKETS``) that holds every level's count (the TT trace variant). numpy only. |
| """ |
| from __future__ import annotations |
|
|
| from dataclasses import dataclass, field |
| from typing import Any, Dict, Mapping, Optional, Sequence, Tuple |
|
|
| import numpy as np |
|
|
| from .. import io as tio |
| from ..reference.config import (CAPACITY_BUCKETS, DROP_ZERO_POINTS, MAX_VOXELS, NUM_POINT_FEATURES, SPARSE_LEVELS, |
| SPARSE_SHAPE, BEVFusionConfig) |
| from ..ttaw import sparse as sp |
| from ..ttaw.geometry import as_transform |
| from ..ttaw.pointcloud import StreamDensifiers, densify_sweeps, nonzero_rows |
|
|
| __all__ = ["Prepared", "VoxelSet", "SparsePlan", "BEVFusionPreprocessor", "TIME_LAG_FIELD", "hard_voxelize", |
| "vfe_mean", "build_sparse_plan", "select_bucket", "LEVEL_NAMES", "SUBM_LAYERS", "STRIDED_LAYERS"] |
|
|
| TIME_LAG_FIELD = "time_lag" |
| LEVEL_NAMES = ("l1", "l2", "l3", "l4", "out") |
| |
| STRIDED_LAYERS = {"down1": (3, 2, 1), "down2": (3, 2, 1), "down3": (3, 2, (1, 1, 0)), |
| "conv_out": ((1, 1, 3), (1, 1, 2), 0)} |
| SUBM_LAYERS = ("subm1", "subm2", "subm3", "subm4") |
|
|
|
|
| |
|
|
| @dataclass |
| class VoxelSet: |
| """The hard-voxelized frame (module docstring) and its statistics.""" |
|
|
| voxels: np.ndarray |
| num_points: np.ndarray |
| coors: np.ndarray |
| stats: Dict[str, Any] = field(default_factory=dict) |
|
|
| @property |
| def num_voxels(self) -> int: |
| return int(len(self.coors)) |
|
|
|
|
| def hard_voxelize(points: np.ndarray, point_cloud_range: Sequence[float], voxel_size: Sequence[float], *, |
| max_points: int = 10, max_voxels: int = MAX_VOXELS) -> VoxelSet: |
| """(M, F) float32 points (x, y, z, ...) -> :class:`VoxelSet` (module docstring). Arithmetic and ordering are those |
| of ``research/bevfusion/scripts/bevfusion_ref.py voxelize`` (bit-identical, tested).""" |
| pts = np.asarray(points, dtype=np.float32) |
| if pts.ndim != 2 or pts.shape[1] < 3: |
| raise ValueError(f"points must be (M, >=3), got {pts.shape}") |
| rng_min = np.array(point_cloud_range[:3], dtype=np.float32) |
| rng_max = np.array(point_cloud_range[3:], dtype=np.float32) |
| vs = np.array(voxel_size, dtype=np.float32) |
| grid = ((rng_max - rng_min) / vs).astype(np.int64) |
| with np.errstate(invalid="ignore"): |
| idx = np.floor((pts[:, :3] - rng_min) / vs).astype(np.int64) |
| valid = np.all((idx >= 0) & (idx < grid), axis=1) |
| pts_v = pts[valid] |
| idx_v = idx[valid] |
| F = pts.shape[1] |
| key = (idx_v[:, 2] * grid[1] + idx_v[:, 1]) * grid[0] + idx_v[:, 0] |
| uniq, first_idx, inverse = np.unique(key, return_index=True, return_inverse=True) |
| inverse = inverse.reshape(-1) |
| order = np.argsort(first_idx, kind="stable") |
| rank = np.empty_like(order) |
| rank[order] = np.arange(len(order)) |
| vid = rank[inverse] |
| num_vox = min(len(uniq), int(max_voxels)) |
| srt = np.argsort(vid, kind="stable") |
| vid_sorted = vid[srt] |
| starts = np.searchsorted(vid_sorted, np.arange(len(uniq))) |
| pos_sorted = np.arange(len(vid_sorted)) - starts[vid_sorted] |
| pos = np.empty_like(pos_sorted) |
| pos[srt] = pos_sorted |
| keep = (pos < max_points) & (vid < num_vox) |
| voxels = np.zeros((num_vox, max_points, F), dtype=np.float32) |
| voxels[vid[keep], pos[keep]] = pts_v[keep] |
| num_points = np.bincount(vid[keep], minlength=num_vox).astype(np.int32)[:num_vox] |
| vox_xyz = np.zeros((num_vox, 3), dtype=np.int64) |
| vox_xyz[vid[keep]] = idx_v[keep] |
| coors = np.ascontiguousarray(vox_xyz[:, ::-1].astype(np.int32)) |
| stats = {"points_in": int(len(pts)), "points_in_range": int(valid.sum()), "voxels_total": int(len(uniq)), |
| "voxels": int(num_vox), "voxel_overflow": bool(len(uniq) > num_vox), |
| "points_dropped_by_cap": int((pos >= max_points).sum()), |
| "max_points_in_voxel": int(np.bincount(vid).max()) if len(vid) else 0} |
| return VoxelSet(voxels=voxels, num_points=num_points, coors=coors, stats=stats) |
|
|
|
|
| def vfe_mean(voxels: np.ndarray, num_points: np.ndarray) -> np.ndarray: |
| """``HardSimpleVFE`` as exported: ``ReduceSum(voxels, axis=1) / float(num_points)`` in float32 -> (V, 4).""" |
| v = np.asarray(voxels, dtype=np.float32) |
| return (v.sum(axis=1) / np.asarray(num_points).astype(np.float32).reshape(-1, 1)).astype(np.float32) |
|
|
|
|
| |
|
|
| @dataclass |
| class SparsePlan: |
| """Every level's active set and every layer's neighbour map (C13 gather form; ``-1`` = no input).""" |
|
|
| levels: Dict[str, sp.SparseLevel] |
| maps: Dict[str, np.ndarray] |
| bucket: str |
| overflow: bool |
| pair_counts: Optional[Dict[str, int]] = None |
| |
| |
| |
| perm: Optional[np.ndarray] = None |
| keys: Optional[Dict[str, np.ndarray]] = None |
| xstart: Optional[Dict[str, np.ndarray]] = None |
|
|
| def counts(self) -> Dict[str, int]: |
| return {name: lvl.num_active for name, lvl in self.levels.items()} |
|
|
| def pairs(self) -> Dict[str, int]: |
| if self.pair_counts is not None: |
| return dict(self.pair_counts) |
| return {name: sp.pairs_count(m) for name, m in self.maps.items()} |
|
|
| @property |
| def device_fields(self) -> bool: |
| return self.keys is not None |
|
|
| def describe(self) -> Dict[str, Any]: |
| return {"counts": self.counts(), "pairs": self.pairs(), "bucket": self.bucket, "overflow": self.overflow} |
|
|
|
|
| def select_bucket(counts: Mapping[str, int], buckets: Mapping[str, Sequence[int]] = CAPACITY_BUCKETS |
| ) -> Tuple[str, bool]: |
| """The first bucket (in the mapping's order: small to large) whose capacities hold every level's count; |
| ``(largest, True)`` when none does. Nothing is clipped: an overflowing plan is refused before the device runs |
| (``api.BEVFusion._prepare`` raises ``io.InputError``, HTTP 400; ``tt.sparse.frame_inputs`` refuses it too).""" |
| names = list(buckets) |
| for name in names: |
| if all(int(counts[lvl]) <= int(cap) for lvl, cap in zip(LEVEL_NAMES, buckets[name])): |
| return name, False |
| return names[-1], True |
|
|
|
|
| def build_sparse_plan(coors_zyx: np.ndarray, *, rule: str = "spconv", |
| buckets: Mapping[str, Sequence[int]] = CAPACITY_BUCKETS, backend: str = "numpy") -> SparsePlan: |
| """The sparse encoder's rulebooks for one frame (module docstring). ``backend="numba"`` (OPT_PLAN H1, |
| ``BEVFUSION_HOST_PLAN``): the hash-rulebook twin ``host/fast.sparse_plan`` (bit-identical maps and level |
| coordinates; both rules agree for in-grid voxels); ``backend="columns"`` (H1b) the column-table twin |
| ``host/fast.sparse_plan_columns`` (same maps, levels and pair counts); both fall back to this numpy oracle without |
| numba or for out-of-grid voxels. ``backend="xcolumns"`` (H6a): the column twin with its serial passes in parallel |
| x partitions (same maps). ``backend="levels"`` (OPT_PLAN M0, ``BEVFUSION_DEVICE_PLAN=1``): the level sets only |
| (``host/fast.sparse_levels``: every level key-sorted, l1 too, with its ``perm``); the device builds the maps.""" |
| if backend == "levels": |
| from . import fast |
|
|
| lv = fast.sparse_levels(coors_zyx, STRIDED_LAYERS, SPARSE_LEVELS, SPARSE_SHAPE) |
| if lv is not None: |
| levels_l, perm, keys, xstart = lv |
| bucket, overflow = select_bucket({k: v.num_active for k, v in levels_l.items()}, buckets) |
| return SparsePlan(levels=levels_l, maps={}, bucket=bucket, overflow=overflow, perm=perm, keys=keys, |
| xstart=xstart) |
| backend = "xcolumns" |
| if backend in ("numba", "columns", "xcolumns"): |
| from . import fast |
|
|
| res = None |
| if backend in ("columns", "xcolumns"): |
| res = fast.sparse_plan_columns(coors_zyx, STRIDED_LAYERS, SPARSE_LEVELS, SPARSE_SHAPE, |
| par=backend == "xcolumns") |
| if res is None: |
| res = fast.sparse_plan(coors_zyx, STRIDED_LAYERS, SPARSE_LEVELS, SPARSE_SHAPE) |
| if res is not None: |
| levels_f, maps_f, pairs_f = res |
| bucket, overflow = select_bucket({k: v.num_active for k, v in levels_f.items()}, buckets) |
| return SparsePlan(levels=levels_f, maps=maps_f, bucket=bucket, overflow=overflow, pair_counts=pairs_f) |
| elif backend != "numpy": |
| raise ValueError("backend must be 'numpy', 'numba', 'columns', 'xcolumns' or 'levels'") |
| c = np.asarray(coors_zyx, dtype=np.int64) |
| l1 = sp.SparseLevel(c[:, ::-1].copy(), SPARSE_SHAPE) |
| levels = {"l1": l1} |
| maps: Dict[str, np.ndarray] = {} |
| lvl = l1 |
| for i, (down, nxt) in enumerate(zip(("down1", "down2", "down3", "conv_out"), ("l2", "l3", "l4", "out")), 1): |
| maps[f"subm{i}"] = sp.subm_neighbor_map(lvl, 3, rule=rule) |
| k, s, p = STRIDED_LAYERS[down] |
| lvl, maps[down] = sp.strided_neighbor_map(lvl, k, s, p) |
| levels[nxt] = lvl |
| for name, shape in SPARSE_LEVELS.items(): |
| if levels[name].spatial_shape != shape: |
| raise AssertionError(f"level {name} grid {levels[name].spatial_shape} != {shape}") |
| bucket, overflow = select_bucket({k: v.num_active for k, v in levels.items()}, buckets) |
| return SparsePlan(levels=levels, maps=maps, bucket=bucket, overflow=overflow) |
|
|
|
|
| |
|
|
| @dataclass |
| class Prepared: |
| """One frame, ready for the device or the CPU reference.""" |
|
|
| points: np.ndarray |
| voxel_set: VoxelSet |
| vfe: np.ndarray |
| plan: Optional[SparsePlan] |
| frame_id: str = "base_link" |
| densify: Dict[str, Any] = field(default_factory=dict) |
| skipped: Optional[str] = None |
|
|
| @property |
| def num_voxels(self) -> int: |
| return self.voxel_set.num_voxels |
|
|
| @property |
| def coors(self) -> np.ndarray: |
| return self.voxel_set.coors |
|
|
| def onnx_inputs(self) -> Dict[str, np.ndarray]: |
| """The deployed graph's inputs: ``voxels`` (V, 10, 4) float32, ``coors`` (V, 3) int32 (z, y, x), |
| ``num_points_per_voxel`` (V,) int32.""" |
| v = self.voxel_set |
| return {"voxels": v.voxels, "coors": v.coors, "num_points_per_voxel": v.num_points} |
|
|
| def stats(self) -> Dict[str, Any]: |
| """Voxel and sparse statistics, the skip verdict and the densification info (``meta``).""" |
| s = dict(self.voxel_set.stats) |
| s["num_voxels"] = self.num_voxels |
| if self.plan is not None: |
| s["sparse"] = self.plan.describe() |
| s.update({"skipped": self.skipped, "densify": self.densify}) |
| return s |
|
|
|
|
| def _cloud(source: Any, what: str) -> tio.PointCloud: |
| if source is None: |
| raise tio.InputError(f"{what} is missing") |
| return tio.load_points(source) |
|
|
|
|
| def _xyz(cloud: tio.PointCloud, what: str) -> np.ndarray: |
| if not {"x", "y", "z"} <= set(cloud.fields): |
| raise tio.InputError(f"{what} needs fields x, y, z (got {list(cloud.fields)})") |
| return np.ascontiguousarray(cloud.select(("x", "y", "z")), dtype=np.float32) |
|
|
|
|
| class BEVFusionPreprocessor: |
| """Stateful only through the per-stream densification caches (``stream`` inputs).""" |
|
|
| def __init__(self, config: BEVFusionConfig, *, max_streams: int = 16, drop_zero_points: bool = DROP_ZERO_POINTS, |
| build_plan: bool = True, rule: str = "spconv", plan: str = "numpy", voxelizer: str = "numpy", |
| densify: str = "numpy"): |
| if plan not in ("numpy", "numba", "columns", "xcolumns", "levels") or voxelizer not in ("numpy", "numba", "lazy", "parallel"): |
| raise ValueError("plan / voxelizer must be 'numpy' or 'numba'") |
| if densify not in ("numpy", "numba"): |
| raise ValueError("densify must be 'numpy' or 'numba'") |
| self.densify = densify |
| |
| |
| self.plan_backend = plan |
| self.voxelizer = voxelizer |
| self.config = config |
| self.drop_zero_points = drop_zero_points |
| self.build_plan = build_plan |
| self.rule = rule |
| self.capacity = int(config.node["cloud_capacity"]) |
| self.streams = StreamDensifiers(num_past_frames=int(config.node["densification_num_past_frames"]), |
| cloud_capacity=self.capacity, max_streams=max_streams) |
|
|
| |
| def densified_points(self, points: Any = None, *, sweeps: Optional[Sequence[Mapping[str, Any]]] = None, |
| stream: Optional[Mapping[str, Any]] = None) -> Tuple[np.ndarray, Dict[str, Any], str]: |
| """The (N, 4) float32 cloud (x, y, z, time_lag) of one request, its densification info and frame id.""" |
| cloud = _cloud(points, "this model needs 'points'") |
| frame_id = cloud.frame_id |
| if TIME_LAG_FIELD in cloud.fields: |
| if sweeps or stream is not None: |
| raise tio.InputError("a cloud with a 'time_lag' column is already densified: do not send 'sweeps' " |
| "or 'stream' with it") |
| fastpts = self._densify_fast(cloud, TIME_LAG_FIELD) |
| if fastpts is not None: |
| return fastpts, {"mode": "pre-densified", "num_points": int(len(fastpts))}, frame_id |
| xyz = _xyz(cloud, "points") |
| lag = np.asarray(cloud.select((TIME_LAG_FIELD,))[:, 0], dtype=np.float32) |
| if not np.isfinite(lag).all(): |
| raise tio.InputError("time_lag holds non-finite values") |
| pts = np.concatenate([xyz, lag[:, None]], axis=1) |
| if self.drop_zero_points: |
| pts = pts[nonzero_rows(pts)] |
| if len(pts) > self.capacity: |
| raise tio.InputError(f"{len(pts)} points exceed cloud_capacity {self.capacity}") |
| return np.ascontiguousarray(pts), {"mode": "pre-densified", "num_points": int(len(pts))}, frame_id |
| xyz = None |
| if sweeps or stream is not None or not self._fast_densify_ok(cloud): |
| xyz = _xyz(cloud, "points") |
| if self.drop_zero_points: |
| xyz = xyz[nonzero_rows(xyz)] |
| if sweeps and stream is not None: |
| raise tio.InputError("give 'sweeps' (client-side densification) or 'stream' (server-side), not both") |
| if sweeps: |
| past = [] |
| for i, s in enumerate(sweeps): |
| if not isinstance(s, Mapping): |
| raise tio.InputError(f"sweeps[{i}] must be an object {{points, time_lag_s, T_current_from_sweep}}") |
| try: |
| lag = float(s.get("time_lag_s", 0.0)) |
| except (TypeError, ValueError): |
| lag = float("nan") |
| if not np.isfinite(lag): |
| raise tio.InputError(f"sweeps[{i}].time_lag_s must be finite") |
| T = s.get("T_current_from_sweep") |
| T = None if T is None else as_transform(T, field=f"sweeps[{i}].T_current_from_sweep") |
| sx = _xyz(_cloud(s.get("points"), f"sweeps[{i}].points"), f"sweeps[{i}].points") |
| if self.drop_zero_points: |
| sx = sx[nonzero_rows(sx)] |
| past.append((sx, lag, T)) |
| pts, info = densify_sweeps(xyz, past, point_feature_size=NUM_POINT_FEATURES, |
| cloud_capacity=self.capacity) |
| return pts, {"mode": "sweeps", **info.to_dict()}, frame_id |
| if stream is not None: |
| if not isinstance(stream, Mapping): |
| raise tio.InputError("stream must be an object {id, timestamp_s, T_world_from_ego, reset}") |
| sid = str(stream.get("id", "default")) |
| try: |
| stamp = float(stream["timestamp_s"]) |
| except (KeyError, TypeError, ValueError): |
| stamp = float("nan") |
| if not np.isfinite(stamp): |
| raise tio.InputError("stream.timestamp_s (seconds) is required for server-side densification") |
| T = stream.get("T_world_from_ego") |
| T = None if T is None else as_transform(T, field="stream.T_world_from_ego") |
| d = self.streams.get(sid, reset=bool(stream.get("reset", False))) |
| if not d.enqueue(xyz, stamp, T): |
| raise tio.InputError( |
| f"stream.T_world_from_ego is required with densification_num_past_frames={d.num_past_frames}: " |
| "Autoware skips a frame without its map->base_link transform and does not cache it (nor did this " |
| "request)") |
| pts, info = d.sweep_points(point_feature_size=NUM_POINT_FEATURES) |
| return pts, {"mode": "stream", "stream_id": sid, **info.to_dict()}, frame_id |
| if not sweeps and stream is None and xyz is None: |
| pts = self._densify_fast(cloud, None) |
| return pts, {"mode": "single", "num_points": int(len(pts))}, frame_id |
| if xyz is None: |
| xyz = _xyz(cloud, "points") |
| if self.drop_zero_points: |
| xyz = xyz[nonzero_rows(xyz)] |
| if len(xyz) > self.capacity: |
| raise tio.InputError(f"{len(xyz)} points exceed cloud_capacity {self.capacity}") |
| pts = np.zeros((len(xyz), NUM_POINT_FEATURES), np.float32) |
| pts[:, :3] = xyz |
| return pts, {"mode": "single", "num_points": int(len(pts))}, frame_id |
|
|
| def _fast_densify_ok(self, cloud: tio.PointCloud) -> bool: |
| if self.densify != "numba" or not {"x", "y", "z"} <= set(cloud.fields): |
| return False |
| from . import fast |
|
|
| return fast.HAVE_NUMBA and cloud.points.dtype == np.float32 and cloud.points.ndim == 2 |
|
|
| def _densify_fast(self, cloud: tio.PointCloud, lag_field: Optional[str]) -> Optional[np.ndarray]: |
| """H4 (``BEVFUSION_HOST_DENSIFY=numba``): the single-sweep and pre-densified paths in one numba pass |
| (``host/fast.densify_rows``; the same rows, values, errors and capacity check), or None (numpy path).""" |
| if not self._fast_densify_ok(cloud): |
| return None |
| from . import fast |
|
|
| f = list(cloud.fields) |
| cols = (f.index("x"), f.index("y"), f.index("z"), f.index(lag_field) if lag_field else -1) |
| res = fast.densify_rows(cloud.points, cols, self.drop_zero_points) |
| if res is None: |
| return None |
| pts, bad = res |
| if bad: |
| raise tio.InputError("time_lag holds non-finite values") |
| if len(pts) > self.capacity: |
| raise tio.InputError(f"{len(pts)} points exceed cloud_capacity {self.capacity}") |
| return pts |
|
|
| |
| def voxelize(self, points4: np.ndarray) -> Tuple[VoxelSet, np.ndarray]: |
| cfg = self.config |
| if self.voxelizer in ("numba", "lazy", "parallel"): |
| from . import fast |
|
|
| res = fast.hard_voxelize(points4, cfg.point_cloud_range, cfg.voxel_size, |
| max_points=cfg.max_points_per_voxel, max_voxels=cfg.max_voxels, |
| lazy=self.voxelizer in ("lazy", "parallel"), |
| par=self.voxelizer == "parallel") |
| if res is not None: |
| return res |
| vs = hard_voxelize(points4, cfg.point_cloud_range, cfg.voxel_size, max_points=cfg.max_points_per_voxel, |
| max_voxels=cfg.max_voxels) |
| return vs, vfe_mean(vs.voxels, vs.num_points) |
|
|
| def __call__(self, points: Any = None, *, sweeps: Optional[Sequence[Mapping[str, Any]]] = None, |
| stream: Optional[Mapping[str, Any]] = None) -> Prepared: |
| pts, info, frame_id = self.densified_points(points, sweeps=sweeps, stream=stream) |
| vs, vfe = self.voxelize(pts) |
| skipped = "no_voxels" if vs.num_voxels < self.config.min_voxels else None |
| plan = None |
| if skipped is None and self.build_plan: |
| plan = build_sparse_plan(vs.coors, rule=self.rule, backend=self.plan_backend) |
| return Prepared(points=pts, voxel_set=vs, vfe=vfe, plan=plan, frame_id=frame_id, densify=info, |
| skipped=skipped) |
|
|
| def describe(self) -> Dict[str, Any]: |
| return {"max_voxels": self.config.max_voxels, "points_per_voxel": self.config.max_points_per_voxel, |
| "drop_zero_points": self.drop_zero_points, "cloud_capacity": self.capacity, "rule": self.rule, |
| "plan": self.plan_backend, "voxelizer": self.voxelizer, "densify": self.densify, |
| "capacity_buckets": {k: list(v) for k, v in CAPACITY_BUCKETS.items()}, |
| "streams": self.streams.describe()} |
|
|
