| |
| """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/<tag>/<sequence>.json`` (Autoware detections > 0.2 and the |
| top-300 test-time boxes), and ``eval_nuscenes.py --tags <tag>`` 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"} |
| |
| _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: |
| 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>:<sequence>, 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()) |
|
|
