tt-model push bevdet-p150 (container)
d160c10 verified 11 days ago
37 kB
# SPDX-License-Identifier: Apache-2.0
"""Python API: BEVDet (Autoware tensorrt_bevdet) on one Tenstorrent Blackhole p150.
    from tt_bevdet import BEVDet, load_sample
    with BEVDet.from_pretrained(device_id=0) as model:  # weights -> HF cache, device open, traces captured
        out = model(**load_sample("code/tt_bevdet/samples/synthetic_6cam.json"))  # six cameras + calibration
    print(out.to_dict())  # the same JSON as POST /predict
The contract shared by every bundle of the Autoware collection is the vendored ``ttaw.api_base.ModelBase``
(BUNDLE_CONVENTIONS.md section 8): ``from_pretrained`` resolves the pinned weights before it claims the chip, opens it
(ETH dispatch, 12x10), builds the graph and captures every trace variant in ``warmup_variants``, so the first call is
as fast as the later ones; calls are serialised by a lock (one chip, batch 1) and fill ``timing_ms``; ``close()`` is
idempotent, also runs at interpreter exit, and closes the chip only if the model opened it. The HTTP server
(``tt_bevdet.server.app``) calls this class, so ``/predict`` and ``model(...)`` agree bit for bit.
Inputs per call (``_prepare``; host code in ``tt_bevdet.host``, exact Autoware semantics):
- ``images``: the six cameras ``CAM_FRONT_LEFT``, ``CAM_FRONT``, ``CAM_FRONT_RIGHT``, ``CAM_BACK_LEFT``, ``CAM_BACK``,
  ``CAM_BACK_RIGHT`` (any order, any size; non-1600x900 images are resized like the node does);
- ``calibration``: per camera ``intrinsics`` (of the image as sent) and ``T_ref_from_camera`` (camera optical frame ->
  base_link), or ``{"preset": name}``; ``calibration["scale_intrinsics"] = False`` keeps the raw K when the node
  resizes an image (the Autoware quirk, SPEC 2.4 item 5; default True scales K, decision D8);
- ``stream``: ``{"id", "reset", "timestamp_s", "T_world_from_ego"}`` drives the 8-frame BEV history (D8): a new
  stream starts like a fresh node (zero history), ``reset`` is a scene change, the ego pose (default identity, as
  in Autoware) turns on motion-compensated alignment. Up to ``BEVDET_MAX_STREAMS`` (default 4) stream ids keep
  their history at once (one device ring bank each); a further id evicts the least recently used one, whose next
  frame then starts fresh. A frame that fails after the device ring changed restarts its stream the same way.
Output (:class:`~.host.outputs.BEVDetDetections`, ``Detections3D`` + two node fields): ``label`` = the network
class (the 10 class names of ``bevdet.param.yaml``), ``autoware_label`` = the label the node publishes (UNKNOWN for
construction_vehicle, barrier, traffic_cone), the node's ``twist``; runtime ``params`` include the per-class
``nms_rescale_factor``.
Device graph (``tt_bevdet.tt``, PORT_LOG.md): two trace segments around the documented host BEVPool (``encode``:
ResNet-50 + CustomFPN + DepthNet; ``decode``: AlignBEV / GatherBEV over the device history ring + CustomResNet + FPN_LSS
+ CenterHead) and the ring updates (``push`` / ``fill``), all ``ttaw.trace.TraceRunner`` variants captured by
``from_pretrained``; per call: ``encode`` replay -> host BEVPool -> ``decode`` replay (+ ring update) -> host decode and
scale-NMS. ``timing_ms`` holds ``preprocess``, ``encode``, ``bevpool``, ``decode``, ``device`` (the three),
``postprocess`` and ``total``. Load-time knobs: :data:`KNOBS` (``BEVDET_*``). Importing this module has no side
effects (the device code is imported in ``_build``).
"""
from __future__ import annotations

from collections import OrderedDict
from typing import Any, Dict, Mapping, Optional

import numpy as np

from . import io as tio
from .device import DEVICE_DEFAULTS
from .host.outputs import BEVDetDetections
from .reference.config import AUTOWARE_LABELS, CAMERA_ORDER, CLASS_NAMES, HEAD_NAMES, DEFAULT_MODEL_YAML
from .ttaw.api_base import ModelBase
from .ttaw.knobs import Knob, Knobs

__all__ = ["BEVDet", "Output", "KNOBS"]

# The result class of this model: ttaw.outputs.Detections3D + the node's autoware_label and twist per object.
Output = BEVDetDetections
# The model yaml's per-class Scale-NMS factors (test_cfg.nms_rescale_factor), the default of the runtime parameter.
DEFAULT_NMS_RESCALE = [float(f) for task in DEFAULT_MODEL_YAML["test_cfg"]["nms_rescale_factor"] for f in task]
STREAM_POLICY = ("each stream id keeps its own 8-frame BEV history (one device ring bank per stream when "
                 "MAX_STREAMS > 1); a new id beyond MAX_STREAMS evicts the least recently used stream, whose next "
                 "frame starts fresh (zero history, like a new node); a frame that fails after the device ring "
                 "changed restarts its stream the same way")

# Load-time switches (read once at build; env BEVDET_<NAME>). Each optimization adds its knob here (A/B switch).
KNOBS = Knobs("BEVDET", [
    Knob("HISTORY_STRIDE", 1, "frames per BEV-history push (1 = Autoware; 5-6 approximate training's 2 Hz key-frames, "
                              "meaningful only with stream.T_world_from_ego)"),
    Knob("HOST_PRE", "numba", "image pre-processing: the compiled numba twin (host.fast: RGB resize + crop gather "
                              "into HWC rows, table normalisation; bit-identical, OPT_REPORT R1.2) or the numpy oracle",
         choices=("numpy", "numba")),
    Knob("HOST_BEVPOOL", "numba", "host BEVPool: the compiled numba twin (host.fast.bevpool_nhwc, bit-identical, "
                                  "writes the padded curr directly; OPT_REPORT R1.4) or numpy reduceat + torch CSR mm",
         choices=("torch", "numba")),
    Knob("BEVPOOL", "host", "BEVPool v2: device (the MK-2 BEVBuilder kernel tt.bevpool_k inside the encode trace: a "
                            "segment matmul over 32-cell row tiles writing curr on the device, per-calibration table "
                            "as an RT-dev param; no depth / feature read-back, host BEVPool or curr upload; OPT_REPORT "
                            "round 7) or host (read-back + the HOST_BEVPOOL twin + curr upload)",
         choices=("host", "device")),
    Knob("HOST_POST", "numba", "Scale-NMS implementation: the compiled numba twin (host.fast, same keep list as the "
                               "numpy oracle; OPT_REPORT R1.1) or the numpy oracle (vectorised pair IoU)",
         choices=("numpy", "numba")),
    Knob("HOST_RESIZE", "gather", "non-1600x900 images with HOST_PRE=numba: gather (cv::resize INTER_LINEAR computed at "
                                  "the crop's pixels only, OpenCV's fixed point, self-checked once per source size "
                                  "against cv2 (bit-exact, else cv2); host.fast, OPT_REPORT R6.4) or cv2 (the whole "
                                  "resize, then the crop gather)", choices=("gather", "cv2")),
    Knob("HOST_THREADS", 0, "numba threads of the parallel host twins (pre-processing gather, BEVPool); 0 = all cores"),
    Knob("IOU_MODE", "vendor", "rotated IoU of the scale-NMS: the deployed OpenPCDet-style polygon or exact clipping",
         choices=("vendor", "exact")),
    Knob("MAX_STREAMS", 4, "BEV-history streams kept at once (PLAN D16 / 9.3 Q6: one 25 MB device ring bank per "
                           "stream when > 1); another stream id evicts the least recently used one, which restarts "
                           "fresh"),
    Knob("STEM_IN", "kin", "image input of the stem: uint8 crop rows normalised on the device by the K-IN u8 kernel "
                             "into the L1 stem conv (kin, OPT_REPORT R1.3) or the host-normalised bf16 NHWC rows with "
                             "the stem's DRAM slicing (slice)", choices=("slice", "kin")),
    Knob("CONV_L1", "all,dec", "kxk convs on ttnn's L1 path from the DRAM input (no [N, H, W, C] re-tiling pair; "
                               "tt.placement): none or a comma list of s1..s4, ds, neck, dn (all = these encoder "
                               "groups, OPT_REPORT R2.1), dec (the decoder convs bit-identical on L1, R2.4b), decall "
                               "(every decoder conv; not bit-identical)"),
    Knob("BACKBONE_RES", "next12", "ResNet-50 bottleneck tails: stock (conv3, add + ReLU, next conv1 as ttnn ops), "
                                  "fused (conv3 + residual + ReLU in one mmres_k program; tt.backbone, OPT_REPORT "
                                  "R2.2), next (fused, plus the next block's conv1 in the same program) or next12 "
                                  "(next on layers 1-2 only; layers 3-4 stock)",
         choices=("stock", "fused", "next", "next12")),
    Knob("RING", "tile", "BEV history ring layout: rm (ROW_MAJOR [8, H, W, C], AlignBEV grid_sample + GatherBEV "
                       "permute / reshape / tilize every frame) or tile (the decoder input's own history tile columns: "
                       "identity frames concatenate, push is a 3-tile-column shift; tt.bev_fusion.TtTileRing, "
                       "OPT_REPORT R2.3)", choices=("rm", "tile")),
    Knob("DEC_MERGE", 1, "1: each CustomResNet stage's conv1 and downsample (both 3x3 / stride 2 on the same input) as "
                         "one conv with stacked outputs (tt.bev_decoder.MergedFirstBlock, OPT_REPORT R2.5)"),
    Knob("MM1X1", 1, "1: the image encoder's remaining stock 1x1 convs (layer-3/4 conv1, layer1.0 conv1, FPN laterals) "
                     "as ttnn.linear with explicit bit-identical program configs and the ReLU fused "
                     "(tt.linear1x1, OPT_REPORT R3.2); 0: ttnn.conv2d's own matmul + a separate ReLU"),
    Knob("CONV_TUNE", 1, "1: per-conv Conv2dConfig overrides of the kxk convs (activation + weight double buffering "
                         "and the probe's shard choices; bit-identical; tt.conv_tune, OPT_REPORT R3.1); 0: ttnn's "
                         "defaults"),
    Knob("HEAD2", "hs", "the block-diagonal 384 -> 20 head conv: hs (L1 path, height sharded, 64-row activation "
                        "blocks; not bit-identical: fp32 K-block order, every frozen gate value unchanged; OPT_REPORT "
                        "R3.3) or dram (ttnn's DRAM-sliced block-sharded path)", choices=("dram", "hs")),
    Knob("L1RES", 1, "1: ResNet-50 layers 3-4 L1-resident (block-sharded residual stream, conv1 / conv3 as 2D "
                     "multicast linears on the shards, sharded add + ReLU; bit-identical; tt.l1res, OPT_REPORT R4.1); "
                     "0: DRAM between ops"),
    Knob("DN_L1", 1, "1: the neck's fpn conv -> DepthNet reduce conv and the 3 DepthNet BasicBlocks chained on L1 "
                     "shards (conv1 -> conv2 -> sharded add + ReLU), the ASPP kxk branches reading them (tt."
                     "image_encoder, OPT_REPORT R4.2); 0: DRAM between ops"),
    Knob("PUSH_X", 1, "1: the ring push after an identity decode is one slice of that decode's input (kept allocated "
                      "after its capture) written straight into the ring (bit-exact; tt.model, OPT_REPORT R4.3); 0: "
                      "tilize + concat + copy"),
    Knob("UPSAMPLE", "kernel", "FPN_LSS align-corners bilinear up-samplings: kernel (TILE in / out: the upk generic_op for "
                               "the x2 of the neck output, a Kron interpolation matmul for the x4 of F2; fp32-exact "
                               "weights; tt.upsample_k, OPT_REPORT R5.1) or grid (untilize + ttnn.grid_sample with a "
                               "static grid, bf16-truncated weights + tilize); up2 / up4: the kernel for that factor only",
         choices=("grid", "kernel", "up2", "up4")),
    Knob("UPK_TERMS", 1, "TF32 terms of the up-sampling kernel's fp32 interpolation weights: 1 (weights rounded to "
                         "TF32, ~2^-11; the e2e gates: sample0 50 / 52) or 2 (fp32-exact; with the x4 kernel sample0 "
                         "49 / 52 fails its frozen gate; OPT_REPORT R5.1)", choices=(1, 2)),
    Knob("MMRES_MCAST", 1, "1: the mmres_k programs (layers 1-2 bottleneck tails) read their resident weights once "
                           "per grid row and multicast them along it (bevformer-p150 D20 handshake; bit-identical; "
                           "OPT_REPORT R5.2); 0: every core reads all weights from DRAM"),
    Knob("RES_L1", 1, "1: layers 1-2 conv2 outputs stay in their L1 height shards for mmres_k, which writes the next "
                      "block's conv1 straight into the next conv2's input shards (no S2I / I2S pair per block; "
                      "bit-identical; OPT_REPORT R5.3); 0: DRAM between conv2 and mmres_k"),
    Knob("TSHARD", "all", "large 3x3 convs block-sharded COL_MAJOR (rows over the whole grid width; bit-identical; "
                          "tt.tshard, OPT_REPORT R6.1): all (enc + the BEV neck's conv.0 -> conv.3, chained on L1), "
                          "enc (the eight 512 -> 512 CustomFPN / DepthNet convs: 12 x 8 instead of 8 x 10 cores) or "
                          "none (ttnn's ROW_MAJOR grids)", choices=("none", "enc", "all")),
    Knob("DEC_CHAIN", 1, "1: the CustomResNet BasicBlocks' conv1 -> conv2 chained on L1 shards, stage 0 block-sharded "
                         "COL_MAJOR (bit-identical; tt.bev_decoder, OPT_REPORT R6.2); 0: DRAM between the convs"),
    Knob("L3T", 1, "1: ResNet-50 layer 3 (L1-resident, L1RES) in the transposed block sharding: rows over the whole "
                   "grid width, the 1x1 matmuls with transpose_mcast, conv2 COL_MAJOR (12 x 8 instead of 8 x 10 cores; "
                   "bit-identical; tt.l1res, OPT_REPORT R6.3); 0: 8 x 10 ROW_MAJOR"),
    Knob("ENC_PRECISION", "", "precision spec of the image encoder only (empty: PRECISION)"),
    Knob("DEC_PRECISION", "", "precision spec of the BEV decoder only (empty: PRECISION)"),
    Knob("PRECISION", "HiFi4+fp32+l1acc:w=fp32", "precision of every conv / linear (ttaw.precision spec): HiFi4, fp32 "
                                                 "accumulation, packer L1 accumulation (fp32 K-block partials), "
                                                 "fp32 weights and biases; bf16 activations"),
])


class BEVDet(ModelBase):
    """BEVDet (Autoware tensorrt_bevdet) on one Blackhole p150. Create it with :meth:`from_pretrained`."""

    MODEL_NAME = "bevdet-p150"
    ENV_PREFIX = "BEVDET"  # prefix of the environment knobs (SERVING.md section 3.4)
    DEFAULT_REPO = "AutowareFoundation/tensorrt_bevdet"
    DEFAULT_TAG = "v1.0"  # the Autoware ansible artifacts pin
    # the commit DEFAULT_TAG points to (pinned: tags can move)
    DEFAULT_REVISION = "391466349a1b39c93b9865364a8088110ca8d97b"
    ALLOW_PATTERNS = ["bevdet_one_lt_d.onnx"]
    VARIANTS = ["default"]  # load-time: selects weights files and trace shapes
    DEFAULT_VARIANT = "default"
    INPUT_KIND = "multicam"  # lidar | camera | multicam | lidar+multicam | planner
    CAMERA_ORDER = CAMERA_ORDER  # the network's camera order (the node's data_params.cams)
    POINT_FIELDS = tio.DEFAULT_POINT_FIELDS  # unused (camera model); kept for the shared /info schema
    # per network class (label_id 0..9): the class name (bevdet.param.yaml class_names; PLAN 9.3 Q1) and the
    # Autoware label the node publishes for it (AW/src/ros_utils.cpp:32-51; every output row carries both)
    LABELS = list(CLASS_NAMES)
    AUTOWARE_LABELS = list(AUTOWARE_LABELS)
    # Per-request knobs: name -> (type, min, max, default). Host-side post-processing only (S:577-599).
    RUNTIME_PARAMS = {
        "score_threshold": (float, 0.0, 1.0, 0.2),          # node filter (bevdet.param.yaml)
        "decode_score_threshold": (float, 0.0, 1.0, 0.1),   # vendor decode: sigmoid(max logit) > 0.1
        "nms_iou_threshold": (float, 0.0, 1.0, 0.2),        # scale-NMS rotated BEV IoU
        "max_pre_nms": (int, 1, 500, 500),                   # test_cfg.max_per_img (pre-NMS cap)
        "max_detections": (int, 1, 500, 500),                # test_cfg.post_max_size
        # per-class Scale-NMS factors (test_cfg.nms_rescale_factor): 10 numbers in LABELS order, or a partial
        # {class name: factor} mapping; validated by validate_params below
        "nms_rescale_factor": (list, None, None, list(DEFAULT_NMS_RESCALE)),
    }
    EXTRA_INPUTS = ()
    DEVICE_DEFAULTS = DEVICE_DEFAULTS  # validated open parameters (device.py)

    @classmethod
    def validate_params(cls, params: Optional[Mapping[str, Any]]) -> Dict[str, Any]:
        """``ModelBase.validate_params`` plus ``nms_rescale_factor``: a list of 10 factors (class order of
        :data:`LABELS`) or a ``{class name: factor}`` mapping that overrides some of them; every factor finite and
        in ``(0, 100]``. Idempotent: the validated value is the full list."""
        p = dict(params or {})
        raw = p.pop("nms_rescale_factor", None)
        out = super().validate_params(p)
        out["nms_rescale_factor"] = cls._rescale_param(raw)
        return out

    @classmethod
    def _rescale_param(cls, raw: Any) -> list:
        factors = list(DEFAULT_NMS_RESCALE)
        if raw is None:
            return factors
        if isinstance(raw, Mapping):
            for name, v in raw.items():
                if name not in cls.LABELS:
                    raise tio.InputError(f"nms_rescale_factor: unknown class {name!r}; classes: {cls.LABELS}")
                factors[cls.LABELS.index(name)] = v
        elif isinstance(raw, (list, tuple)) and not isinstance(raw, str):
            if len(raw) != len(cls.LABELS):
                raise tio.InputError(f"nms_rescale_factor needs {len(cls.LABELS)} factors (classes {cls.LABELS}), "
                                     f"got {len(raw)}")
            factors = list(raw)
        else:
            raise tio.InputError("nms_rescale_factor must be a list of 10 numbers or a {class name: factor} mapping")
        out = []
        for name, v in zip(cls.LABELS, factors):
            if isinstance(v, bool) or not isinstance(v, (int, float, str)):
                raise tio.InputError(f"nms_rescale_factor[{name!r}] must be a number")
            try:
                f = float(v)
            except ValueError:
                raise tio.InputError(f"nms_rescale_factor[{name!r}] must be a number") from None
            if not (f == f and 0.0 < f <= 100.0):
                raise tio.InputError(f"nms_rescale_factor[{name!r}]={v} is outside (0, 100]")
            out.append(f)
        return out

    # ---- port-specific hooks (called by ModelBase; keep host work out of _forward) ---------------------------
    def _build(self) -> None:
        """Weights (``reference.weights.BEVDetWeights``: ONNX initializers by consuming node, with the port-form
        rewrites of ``reference.port_form`` / ``tt.params``), the host state, and the device graph of ``tt_bevdet.tt``
        as ``TraceRunner`` variants (2 segments around the host BEVPool + the history updates, PLAN 2.10). No capture
        here (``_warm_one``)."""
        self._build_host()
        from .reference.weights import ONNX_FILE, BEVDetWeights, find_onnx
        from .tt.model import TtBEVDet
        from .tt.params import BEVDetParams

        onnx = find_onnx(self.weights_path)
        if onnx is None:
            raise FileNotFoundError(f"{ONNX_FILE} not found in {self.weights_path}")
        self.net_weights = BEVDetWeights(onnx)
        self.params = BEVDetParams(self.net_weights, self.cfg)
        streams = int(self.knobs.MAX_STREAMS)
        self.tt = TtBEVDet(self.device, self.params, precision=str(self.knobs.PRECISION),
                           ring_banks=streams if streams > 1 else 0, stem_in=str(self.knobs.STEM_IN),
                           conv_l1=str(self.knobs.CONV_L1), backbone_res=str(self.knobs.BACKBONE_RES),
                           ring=str(self.knobs.RING), dec_merge=bool(int(self.knobs.DEC_MERGE)),
                           enc_precision=str(self.knobs.ENC_PRECISION) or None,
                           dec_precision=str(self.knobs.DEC_PRECISION) or None,
                           mm1x1=bool(int(self.knobs.MM1X1)), conv_tune=bool(int(self.knobs.CONV_TUNE)),
                           head2=str(self.knobs.HEAD2), l1res=bool(int(self.knobs.L1RES)),
                           dn_l1=bool(int(self.knobs.DN_L1)), push_x=bool(int(self.knobs.PUSH_X)),
                           upsample=str(self.knobs.UPSAMPLE),
                           upsample_terms=int(self.knobs.UPK_TERMS), mmres_mcast=bool(int(self.knobs.MMRES_MCAST)),
                           res_l1=bool(int(self.knobs.RES_L1)), tshard=str(self.knobs.TSHARD),
                           dec_chain=bool(int(self.knobs.DEC_CHAIN)), l3t=bool(int(self.knobs.L3T)),
                           bevpool=str(self.knobs.BEVPOOL))
        self._bp_tab_sent: Any = None            # the bp_tab array last uploaded (device BEVPool)
        self.runner = self.tt.runner

    def _warm_one(self, variant: Any) -> None:
        """Capture every trace variant (warm-up + strict capture), then run one synthetic frame through the whole
        host + device path so the first real call is as fast as the later ones; the warm-up stream is forgotten."""
        self.runner.capture()
        from .tt.warmup import synthetic_request

        if "numba" in (str(self.knobs.HOST_POST), str(self.knobs.HOST_PRE), str(self.knobs.HOST_BEVPOOL)):
            from .host.fast import warm_up
            warm_up()                                   # compile the host twins now, not on the first request

        req = synthetic_request(self.cfg)
        with self._lock:
            prepared = self._prepare(**req)
            raw = self._forward(prepared)
            self._postprocess(raw, prepared, self.validate_params({}))
        self._forget_stream(prepared["stream_id"])

    def _build_host(self) -> None:
        """The host half of the model: configuration, load-time knobs, the per-calibration cache and the per-stream
        temporal states (no device, no weights; the host tests call it directly)."""
        from .host.calib import CalibrationCache
        from .reference.config import default_config

        self.cfg = default_config()
        params = {k.upper(): v for k, v in getattr(self, "compile_params", {}).items() if k.upper() in KNOBS.knobs}
        self.knobs = KNOBS.read(**params)
        from .host import fast

        fast.set_threads(int(self.knobs.HOST_THREADS))
        fast.set_resize(str(self.knobs.HOST_RESIZE))
        self.calib_cache = CalibrationCache(max_entries=16)
        self._streams: "OrderedDict[str, Any]" = OrderedDict()
        self._banks: Dict[str, int] = {}             # stream id -> device ring bank (MAX_STREAMS > 1)
        self._active: Optional[str] = None           # the stream whose history the device ring holds

    def _forget_stream(self, sid: str) -> None:
        self._streams.pop(sid, None)
        self._banks.pop(sid, None)
        if self._active == sid:
            self._active = None

    def _stream_state(self, stream: Optional[Mapping[str, Any]]):
        """The host temporal state of ``stream.id`` (``host.temporal.TemporalState``); a new id beyond
        ``MAX_STREAMS`` evicts the least recently used one (PLAN D16, first publish: one stream). A new or evicted
        stream starts like a fresh node: its first frame samples no history (zero history) and fills the ring."""
        from .host.temporal import TemporalState

        sid = str((stream or {}).get("id", "default"))
        st = self._streams.get(sid)
        if st is None:
            st = TemporalState(self.cfg, history_stride=int(self.knobs.HISTORY_STRIDE))
            self._streams[sid] = st
            while len(self._streams) > int(self.knobs.MAX_STREAMS):
                old = next(iter(self._streams))
                self._forget_stream(old)
        self._streams.move_to_end(sid)
        return sid, st

    def _activate_stream(self, sid: str) -> None:
        """Make ``sid``'s history the one the device ring holds (D16 banks when MAX_STREAMS > 1: save the active
        stream's ring into its bank, load ``sid``'s). Called under the model lock, before the device runs. While a
        bank is loading the ring holds no stream (``_active`` None), so a failed load is redone on the next frame."""
        if sid == self._active:
            return
        runner = getattr(self, "runner", None)
        if runner is not None and int(self.knobs.MAX_STREAMS) > 1:
            if self._active is not None and self._active in self._banks:
                runner.save_state("ring", self._banks[self._active])
            self._active = None
            if sid not in self._banks:
                used = set(self._banks.values())
                self._banks[sid] = min(set(range(int(self.knobs.MAX_STREAMS))) - used)
            elif self._streams[sid].frame > 0:
                runner.load_state("ring", self._banks[sid])
        self._active = sid

    def _prepare(self, points: Any = None, *, images: Any = None, calibration: Optional[Mapping[str, Any]] = None,
                 stream: Optional[Mapping[str, Any]] = None, **_: Any) -> Dict[str, Any]:
        """Six images + calibration + stream -> the host values of one frame: the network input (the uint8 crops
        [6, 3, 256, 704] normalised to NHWC rows), the calibration products (cam_params, camera gates, BEVPool tables
        and the host BEVPool operator; cached per calibration) and the temporal plan (flag, history, transforms)."""
        from .host.bevpool import HostBEVPool
        from .host.calib import bevpool_tables, cam_params
        from .host.inputs import prepare_frame
        from .host.preprocess import normalize

        if points is not None:
            raise tio.InputError("this model takes six camera images, not a point cloud")
        cal = dict(calibration or {})
        scale = cal.pop("scale_intrinsics", True)
        if not isinstance(scale, bool):
            raise tio.InputError("calibration.scale_intrinsics must be a boolean")
        params = getattr(self, "params", None)
        fast_pre = str(self.knobs.HOST_PRE) == "numba"
        norm = None
        if params is not None and fast_pre and self.tt.stem_in == "slice":
            norm = np.empty((int(np.prod(self.tt.img_shape[:-1])), self.tt.img_shape[-1]), np.float32)
        frame = prepare_frame(images, cal, self.cfg, scale_intrinsics=scale,
                              method="numba" if fast_pre else "numpy", norm_out=norm)
        calib = frame.calib
        tables = self.calib_cache.get(calib, "tables", lambda: bevpool_tables(self.cfg, calib))
        try:
            tables.check_capacity(self.cfg)
        except ValueError as e:
            raise tio.InputError(str(e)) from None
        cp = cam_params(self.cfg, calib)
        out = {"crop_u8": frame.crop_u8, "calib": calib, "cam_params": cp, "tables": tables}
        if params is not None:                        # device model built (host tests stop at the host values)
            out["gates"] = self.calib_cache.get(calib, "gates", lambda: params.camera_gates(cp))
            if getattr(self.tt, "bevpool", "host") == "device":
                try:
                    out["bp_tab"] = self.calib_cache.get(calib, "bp_tab", lambda: self.tt.pool_table(tables))
                except ValueError as e:
                    raise tio.InputError(str(e)) from None
            else:
                out["pool"] = self.calib_cache.get(calib, "pool", lambda: HostBEVPool(self.cfg, tables))
            if self.tt.stem_in == "kin":                                   # uint8 crop rows (device K-IN u8)
                out["enc_in"] = self.tt.image_inputs(frame.crop_u8, crop_hwc=frame.images.crop_hwc)
            elif norm is not None and frame.images.crop_hwc is not None:  # the twin wrote the NHWC rows
                out["enc_in"] = {"img": norm.reshape(self.tt.img_shape)}
            else:
                img = normalize(frame.crop_u8, self.cfg)                   # [6, 3, 256, 704] float32
                out["enc_in"] = self.tt.image_inputs(img=np.ascontiguousarray(img.transpose(0, 2, 3, 1)))
        s = dict(stream or {})
        sid, state = self._stream_state(s)
        plan = state.begin_frame(reset=bool(s.get("reset", False)), ego_pose=s.get("T_world_from_ego"))
        out.update({"plan": plan, "stream_id": sid, "state": state})
        return out

    def _forward(self, prepared: Dict[str, Any]) -> Any:
        """Segment A (``encode``: crops + camera gates -> depth / feat), the host BEVPool, then segment B
        (``decode``: AlignBEV + GatherBEV over the device history ring -> BEV decoder -> head maps) and the ring
        update (``push`` / ``fill``, before ``decode`` on a reset frame); see ``tt.bev_fusion.plan_ops``. Returns the
        head maps and the stage times (ms) that ``_postprocess`` adds to ``timing_ms``.
        On an exception the stream stays consistent with the device: before any ring update the frame is dropped
        (the history is unchanged, like the node on an exception); once ``fill`` / ``push`` has started, the device
        ring no longer matches the host plan, so the stream is forgotten and its next frame starts fresh (zero
        history: it never reads the ring, and its ``fill`` overwrites every slot)."""
        import time

        from .tt.bev_fusion import history_grid, plan_ops

        state, plan = prepared["state"], prepared["plan"]
        tt, runner = self.tt, self.runner
        tm: Dict[str, float] = {}
        ring_changed = False
        try:
            self._activate_stream(prepared["stream_id"])
            g_ctx, g_depth = prepared["gates"]
            if getattr(tt, "bevpool", "host") == "device":
                changed = [False]
                try:
                    tm = self._forward_device(prepared, changed)
                finally:
                    ring_changed = changed[0]
                heads = tm.pop("_heads")
            else:
                t0 = time.perf_counter()
                enc = runner("encode", inputs=prepared["enc_in"],
                             params={"g_ctx": g_ctx.reshape(-1, 1, 1, g_ctx.shape[-1]),
                                     "g_depth": g_depth.reshape(-1, 1, 1, g_depth.shape[-1])})
                t1 = time.perf_counter()
                depth, feat = tt.unpack_encode(enc)
                if str(self.knobs.HOST_BEVPOOL) == "numba":                     # writes curr + its zero pad
                    curr_pad = np.empty(tt.curr_shape, np.float32)
                    prepared["pool"](depth, feat, layout="nhwc", out_layout="nhwc", out=curr_pad, method="numba")
                else:
                    curr = prepared["pool"](depth, feat, layout="nhwc", out_layout="nhwc")  # [1, 128, 128, 80]
                    curr_pad = np.zeros(tt.curr_shape, np.float32)
                    curr_pad[..., : curr.shape[-1]] = curr
                grid = history_grid(plan.transforms, plan.history, plan.flag, self.cfg.bev_hw)
                before, after = plan_ops(plan.flag, plan.write_slots)
                t2 = time.perf_counter()
                runner.upload(inputs={"curr": curr_pad}, params={"grid": grid})
                for v in before:
                    ring_changed = True                     # set before the replay: a failed replay may have written
                    runner.replay(v)
                dv = tt.decode_variant(plan.transforms, plan.history, plan.flag)
                runner.replay(dv)
                heads = runner.read(dv)
                for v in after:
                    ring_changed = True
                    runner.replay(tt.after_variant(v, dv))
                t3 = time.perf_counter()
                # stage names that the server's own timing keys (decode = base64 + images, model_call, total) never
                # shadow in the served timing_ms
                tm = {"encode": (t1 - t0) * 1e3, "bevpool": (t2 - t1) * 1e3, "bev_decode": (t3 - t2) * 1e3}
        except BaseException:
            if ring_changed:
                self._forget_stream(prepared["stream_id"])
            else:
                state.abort_frame()
            raise
        state.end_frame()
        return {"heads": tt.unpack_heads(heads), "timing_ms": tm}

    def _forward_device(self, prepared: Dict[str, Any], changed: list) -> Dict[str, Any]:
        """``BEVDET_BEVPOOL=device``: one upload (crops, gates, the AlignBEV grid and, when the calibration changed,
        the BEVPool table), then the replays back to back with no host round trip: ``encode`` (incl. the device
        BEVPool into ``curr``), the ring updates before the decoder, the ``decode`` variant (one blocking read of the
        head maps), the updates after it."""
        import time

        from .tt.bev_fusion import history_grid, plan_ops

        plan, tt, runner = prepared["plan"], self.tt, self.runner
        g_ctx, g_depth = prepared["gates"]
        t0 = time.perf_counter()
        params = {"g_ctx": g_ctx.reshape(-1, 1, 1, g_ctx.shape[-1]),
                  "g_depth": g_depth.reshape(-1, 1, 1, g_depth.shape[-1]),
                  "grid": history_grid(plan.transforms, plan.history, plan.flag, self.cfg.bev_hw)}
        tab = prepared["bp_tab"]
        if tab is not self._bp_tab_sent:
            params["bp_tab"] = tab
        before, after = plan_ops(plan.flag, plan.write_slots)
        dv = tt.decode_variant(plan.transforms, plan.history, plan.flag)
        self._bp_tab_sent = None                    # unknown until the upload went through
        runner.upload(inputs=prepared["enc_in"], params=params)
        self._bp_tab_sent = tab
        runner.replay("encode")
        t1 = time.perf_counter()
        for v in before:
            changed[0] = True                       # set before the replay: a failed replay may have written
            runner.replay(v)
        runner.replay(dv)
        heads = runner.read(dv)
        for v in after:
            changed[0] = True
            runner.replay(tt.after_variant(v, dv))
        t2 = time.perf_counter()
        return {"encode": (t1 - t0) * 1e3, "bevpool": 0.0, "bev_decode": (t2 - t1) * 1e3, "_heads": heads}

    def _postprocess(self, raw: Any, prepared: Dict[str, Any], params: Dict[str, Any]) -> Output:
        """Head maps (``raw``: {name: [1, k, 128, 128] float32}) -> vendor decode + scale-NMS -> node filter ->
        :class:`~.host.outputs.BEVDetDetections` (network class labels + Autoware labels, gravity-centre z,
        velocities, the node's twist)."""
        from .host.postprocess import boxes_to_detections, postprocess

        maps = raw["heads"] if "heads" in raw else raw          # the host tests pass the reference head maps
        heads = {k: maps[k] for k in HEAD_NAMES}
        rescale = params.get("nms_rescale_factor")
        if rescale is None or list(rescale) == DEFAULT_NMS_RESCALE:
            rescale = None                                      # the configuration's float32 factors, bit for bit
        vb = postprocess(heads, self.cfg, decode_score_threshold=params["decode_score_threshold"],
                         nms_iou_threshold=params["nms_iou_threshold"], max_pre_nms=params["max_pre_nms"],
                         max_detections=params["max_detections"], iou_mode=str(self.knobs.IOU_MODE),
                         nms_rescale_factor=rescale, nms=str(self.knobs.HOST_POST))
        plan = prepared["plan"]
        out = boxes_to_detections(vb, self.cfg, score_threshold=params["score_threshold"], model=self.MODEL_NAME,
                                  meta={"stream": {"id": prepared["stream_id"], "frame": plan.frame,
                                                   "flag": plan.flag}})
        out.timing_ms.update({k: round(v, 3) for k, v in raw.get("timing_ms", {}).items()} if "heads" in raw else {})
        return out

    def _release(self) -> None:
        """Release the traces and persistent device tensors; also called when ``from_pretrained`` fails half-way."""
        runner = getattr(self, "runner", None)
        if runner is not None:
            runner.release()

    def extra_info(self) -> Dict[str, Any]:
        """Additions to ``model.info`` (and ``/info``, through ``server.app``): the trace variants, CQs and
        persistent tensors of the runner, the load-time knobs and the stream table (policy, active ids)."""
        info: Dict[str, Any] = {}
        runner = getattr(self, "runner", None)
        if runner is not None:
            info["trace"] = runner.describe()
        knobs = getattr(self, "knobs", None)
        if knobs is not None:
            info["knobs"] = knobs.as_dict()
            info["streams"] = self.stream_info()
        return info

    def stream_info(self) -> Dict[str, Any]:
        """The temporal-state table: ``max_streams``, the history stride, the eviction ``policy`` and the stream
        ``ids`` that hold a history now (least recently used first)."""
        return {"max_streams": int(self.knobs.MAX_STREAMS), "history_stride": int(self.knobs.HISTORY_STRIDE),
                "policy": STREAM_POLICY, "ids": [s for s in self._streams]}