# SPDX-License-Identifier: Apache-2.0 """Run the TT port over the research public-data sequences (nuScenes v1.0-mini mini_val, PandaSet 019) with the research workflow's own runner, so the device outputs are scored exactly like the CPU reference's. The workflow (``research/bevformer/public_data/scripts/run_reference.py``) feeds every backend the SAME network inputs (image [1, 6, 3, 736, 1280], prev_bev, use_prev_bev, can_bus, lidar2img: converter output, decoded and pre-processed by the research reference, bit-exact to this bundle's host path) in sequence order with BEVFormer's video test mode (history reset at each scene start), writes ``results//.json`` (Autoware detections > 0.2 and the top-300 test-time boxes), and ``eval_nuscenes.py --tags `` scores mini_val with the official devkit. :class:`TtBackend` is that backend on the p150: per frame the host geometry of the port (``host.geometry``: float64 projection, SCA rebatching tables, TSA shift, rotation index map), the device graph ``tt.model.TtBEVFormer`` (variants ``image`` and ``bev`` replayed back to back), the last decoder layer's outputs and the device's ``bev_embed`` (free-running: the workflow hands it back as the next frame's prev_bev, which is written into the device state, an exact bf16 round trip). One device process serves every sequence (a module-level singleton). bin/devrun -t 1800 -- env PYTHONPATH=$ROOT/bundles/bevformer-p150/code python \\ bundles/bevformer-p150/code/scripts/run_public_tt.py --tag tt \\ nuscenes:scene-0103 nuscenes:scene-0916 pandaset:ps019_2hz $ROOT/tools/research-venv/bin/python -I research/bevformer/public_data/scripts/eval_nuscenes.py --tags tt --skip-gt """ from __future__ import annotations import argparse import importlib.util import json import os import sys import time from pathlib import Path import numpy as np ROOT = Path(os.environ.get("TT_MODELS_ROOT", "/home/ubuntu/experiments/tt-models")) PUB = ROOT / "research" / "bevformer" / "public_data" DATASETS = {"nuscenes": PUB / "nuscenes", "pandaset": PUB / "pandaset"} # the research runner imports this file again by path (a second module object): the singleton lives in sys.modules _HOLDER = "_bevformer_tt_backend_singleton" class TtBackend: """``bev_embed, outputs_classes, outputs_coords = backend(image, prev_bev, use_prev_bev, can_bus, lidar2img)`` (numpy, the ONNX shapes; the class outputs hold the last decoder layer only, layers 0-4 are zero: the node and the test-time decoder read the last layer). Constructed without arguments; one device per process.""" def __new__(cls): import types holder = sys.modules.setdefault(_HOLDER, types.ModuleType(_HOLDER)) inst = getattr(holder, "instance", None) if inst is None: inst = super().__new__(cls) inst._init() holder.instance = inst return inst @classmethod def attach(cls, tt, *, knobs=None) -> "TtBackend": """Serve the backend from an existing ``TtBEVFormer`` (e.g. the API model's, inside a device test) instead of opening the device; :meth:`detach` drops it (the caller keeps owning the device).""" import types from tt_bevformer.api import KNOBS inst = super().__new__(cls) inst.knobs = knobs or KNOBS.read() inst.device, inst.device_info, inst.tt = None, {}, tt inst.capture_s, inst._cal_key, inst._geo, inst.frames, inst.device_ms = 0.0, None, None, 0, [] sys.modules.setdefault(_HOLDER, types.ModuleType(_HOLDER)).instance = inst return inst @classmethod def detach(cls) -> None: holder = sys.modules.get(_HOLDER) if holder is not None: holder.instance = None def _init(self) -> None: from tt_bevformer.api import BEVFormer, KNOBS from tt_bevformer.device import open_device, describe_device from tt_bevformer.reference.weights import BEVFormerWeights, find_weights from tt_bevformer.tests import paths from tt_bevformer.tt.model import TtBEVFormer from tt_bevformer.tt.params import BEVFormerParams self.knobs = KNOBS.read() w = paths.weights_file() or find_weights(BEVFormer.resolve_weights()) self.device = open_device(allow_fallback=False) self.device_info = describe_device(self.device) t0 = time.time() self.tt = TtBEVFormer(self.device, BEVFormerParams(BEVFormerWeights(w)), precision=str(self.knobs.PRECISION), msda=str(self.knobs.MSDA)) self.tt.capture() self.capture_s = time.time() - t0 self._cal_key = None self._geo = None self.frames = 0 self.device_ms = [] print(f"[TtBackend] device {self.device_info}, build + capture {self.capture_s:.1f} s", flush=True) def __call__(self, image, prev_bev, use_prev_bev, can_bus, lidar2img): from tt_bevformer.host import geometry as hgeo from tt_bevformer.reference import config as C from tt_bevformer.tt.model import PREV_BEV_SHAPE, calibration_inputs, frame_inputs cb = np.asarray(can_bus, np.float32).reshape(18) use = float(np.asarray(use_prev_bev).reshape(-1)[0]) key = np.ascontiguousarray(lidar2img, np.float32).tobytes() if key != self._cal_key: ref_cam, valid = hgeo.point_sampling(lidar2img) geo = {"ref_cam": ref_cam, **hgeo.sca_tables(valid)} self._geo = calibration_inputs(geo) fr = frame_inputs(image, can_bus=cb, use_prev_bev=use, shift=hgeo.tsa_shift(cb, str(self.knobs.SHIFT_MODE)), rot_index=hgeo.rotation_index_map(float(cb[17]))) inputs = dict(fr["inputs"]) if key != self._cal_key: inputs.update(self._geo) r = self.tt.runner table = np.zeros(PREV_BEV_SHAPE, np.float32) if use != 0.0: table[0, 0, :C.NUM_QUERY_BEV] = np.asarray(prev_bev, np.float32).reshape(C.NUM_QUERY_BEV, C.EMBED) t0 = time.perf_counter() r.reset_state("prev_bev", table) r.upload(inputs=inputs, params=fr["params"]) self._cal_key = key r.replay("image") r.replay("bev") out = self.tt.unpack(r.read("bev")) bev = r.read_state("prev_bev")[0, 0, :C.NUM_QUERY_BEV].reshape(C.NUM_QUERY_BEV, 1, C.EMBED) self.device_ms.append((time.perf_counter() - t0) * 1e3) self.frames += 1 cls = np.zeros((C.DEC_LAYERS, 1, C.NUM_QUERY, C.NUM_CLASSES), np.float32) crd = np.zeros((C.DEC_LAYERS, 1, C.NUM_QUERY, C.CODE_SIZE), np.float32) cls[-1] = out["outputs_classes"][0] crd[-1] = out["outputs_coords"][0] return np.asarray(bev, np.float32), cls, crd def close(self) -> None: from tt_bevformer.device import close_device if self.device is not None: # an attached backend's graph belongs to its owner self.tt.release() close_device(self.device) self.detach() def main() -> int: ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) ap.add_argument("sequences", nargs="+", help=":, dataset in " + ", ".join(DATASETS)) ap.add_argument("--tag", default="tt") ap.add_argument("--summary", default="", help="write a JSON summary (frames, device ms, capture s) here") a = ap.parse_args() spec = importlib.util.spec_from_file_location("bevformer_run_reference", PUB / "scripts" / "run_reference.py") run_ref = importlib.util.module_from_spec(spec) spec.loader.exec_module(run_ref) me = f"{Path(__file__).resolve()}:TtBackend" backend = TtBackend() t0 = time.time() for item in a.sequences: ds, seq = item.split(":", 1) sys.argv = ["run_reference.py", str(DATASETS[ds]), seq, "--tag", a.tag, "--backend", me] run_ref.main() ms = np.asarray(backend.device_ms, np.float64) summary = {"tag": a.tag, "sequences": a.sequences, "frames": backend.frames, "wall_s": time.time() - t0, "capture_s": backend.capture_s, "device": backend.device_info, "device_ms_p50": float(np.percentile(ms, 50)) if ms.size else None, "device_ms_p99": float(np.percentile(ms, 99)) if ms.size else None} print(json.dumps(summary, indent=1), flush=True) if a.summary: Path(a.summary).write_text(json.dumps(summary, indent=1)) backend.close() return 0 if __name__ == "__main__": sys.exit(main())