# SPDX-License-Identifier: Apache-2.0
"""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"     # a column with this name marks a pre-densified cloud (seconds)
LEVEL_NAMES = ("l1", "l2", "l3", "l4", "out")
# the strided layers between levels: name -> (kernel, stride, padding) (S:bevfusion:226-235)
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")


# ------------------------------------------------------------------------------------------------ voxels

@dataclass
class VoxelSet:
    """The hard-voxelized frame (module docstring) and its statistics."""

    voxels: np.ndarray              # (V, 10, 4) float32, unused slots zero
    num_points: np.ndarray          # (V,) int32, clamped to max_points_per_voxel
    coors: np.ndarray               # (V, 3) int32 (z, y, x)
    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)                  # float32: (1440, 1440, 40)
    with np.errstate(invalid="ignore"):
        idx = np.floor((pts[:, :3] - rng_min) / vs).astype(np.int64)  # (x, y, z)
    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]   # z-major linear key
    uniq, first_idx, inverse = np.unique(key, return_index=True, return_inverse=True)
    inverse = inverse.reshape(-1)
    order = np.argsort(first_idx, kind="stable")                        # first-appearance order
    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)


# ------------------------------------------------------------------------------------------------ rulebooks

@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]           # l1, l2, l3, l4, out
    maps: Dict[str, np.ndarray]                 # subm1..4 (N_l, 27); down1..3 (N_l+1, 27); conv_out (N_out, 3)
    bucket: str
    overflow: bool
    pair_counts: Optional[Dict[str, int]] = None   # filled by the numba twin (host/fast.sparse_plan), else computed
    # OPT_PLAN M0 (backend "levels", BEVFUSION_DEVICE_PLAN=1): no maps on the host; every level in ascending key,
    # ``perm`` the l1 order (sorted row -> voxel id), the packed keys and x-row starts the device builds the maps
    # from (``host/fast.sparse_levels``). ``pair_counts`` is then filled from the device after the frame ran.
    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()}     # {} for a device plan before its run

    @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"            # no numba / out-of-grid voxels: a host plan (tt.sparse derives the fields)
    if backend in ("numba", "columns", "xcolumns"):
        from . import fast

        res = None
        if backend in ("columns", "xcolumns"):  # H1b: per-level column tables (bit-identical to the hash twin)
            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)


# ------------------------------------------------------------------------------------------------ prepared

@dataclass
class Prepared:
    """One frame, ready for the device or the CPU reference."""

    points: np.ndarray              # (M, 4) float32 x, y, z, time_lag: the densified cloud
    voxel_set: VoxelSet
    vfe: np.ndarray                 # (V, 4) float32 voxel features (the graph's mean VFE)
    plan: Optional[SparsePlan]      # None for an empty frame
    frame_id: str = "base_link"
    densify: Dict[str, Any] = field(default_factory=dict)
    skipped: Optional[str] = None   # Autoware's reason to skip the frame (no voxel: no inference, no objects)

    @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
        # "numba": the host/fast.py twins (OPT_PLAN H1 BEVFUSION_HOST_PLAN, H3 BEVFUSION_HOST_VOXELIZER), bit-identical
        # to the numpy oracles of this module, which stay the reference and the fallback
        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)

    # ---------------------------------------------------------------------------------------- densification
    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:                                       # 1. pre-densified
            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:                                                                # 2. client-side
            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:                                                    # 3. server-side, per stream id
            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:                         # 4. single sweep, H4 twin
            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:                                              # 4. single sweep
            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

    # ------------------------------------------------------------------------------------------ voxelization
    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    # V < 1 aborts the frame
        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()}