# 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_). 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]}