| |
| """D20 task-level accuracy gates for the reduced-precision path (needs one p150). |
| |
| bin/devrun -t 1800 -- env BEVFORMER_PRECISION=... python -m pytest -q -s code/tt_bevformer/tests/test_task_device.py |
| |
| PLAN.md D20 (user decision 2026-10-12, BEVFormer only): a reduced-precision device path (bf16 / bfp8 weights, |
| HiFi2 / LoFi, ...) is judged at the TASK level, not by the frozen module / score gates: |
| |
| - nuScenes mini_val (81 key-frames, scene-0103 + scene-0916, history reset per scene; the research workflow's runner |
| and the official devkit, ``test_e2e_device.nuscenes_mini_val``) against the stored CPU fp32 run: |
| ``|dmAP| <= 0.02``, ``|dNDS| <= 0.02``, Autoware-detection recall and precision vs the CPU run ``>= 0.90``; |
| - reported, not gated: the shipped synthetic sample and the PandaSet 019 f30 -> f40 stream agreement (the frozen |
| ``test_e2e_device`` metrics; recorded on the card). |
| |
| The thresholds are the policy's, fixed here (not a frozen ``.gates.json``; the frozen gate files are never edited). |
| The device path is whatever the ``BEVFORMER_*`` knobs select (the shipped defaults when none is set). Results go to |
| ``task_gates.json`` in ``BEVFORMER_REPORT_DIR`` (default ``$TT_MODELS_ROOT/logs/bevformer``); the mini_val results / |
| metrics to ``research/bevformer/public_data/nuscenes/{results,metrics}/<BEVFORMER_TASK_TAG or tt_task>``. |
| """ |
| from __future__ import annotations |
|
|
| import json |
| import os |
|
|
| import pytest |
|
|
| from tt_bevformer.tests.test_e2e_device import ( |
| _report, model, net_model, nuscenes_mini_val, pandaset_sequence, shipped_sample) |
|
|
| pytestmark = pytest.mark.device |
|
|
| TASK_GATES = {"map_abs_diff": (0.02, "max"), "nds_abs_diff": (0.02, "max"), "recall": (0.90, "min"), |
| "precision": (0.90, "min")} |
| _RESULTS: dict = {} |
|
|
|
|
| def _knobs() -> dict: |
| from tt_bevformer.api import KNOBS |
|
|
| k = KNOBS.read() |
| return {n: str(getattr(k, n)) for n in sorted(KNOBS.knobs)} |
|
|
|
|
| class TestTaskNetworkInput: |
| def test_nuscenes_mini_val_task(self, net_model): |
| """|dmAP|, |dNDS| <= 0.02 and recall / precision >= 0.90 vs the CPU fp32 run on mini_val.""" |
| s = nuscenes_mini_val(net_model, os.environ.get("BEVFORMER_TASK_TAG", "tt_task")) |
| verdict = {} |
| for name, (thr, d) in TASK_GATES.items(): |
| v = float(s[name]) |
| verdict[name] = {"value": v, "threshold": thr, "direction": d, |
| "pass": v <= thr if d == "max" else v >= thr} |
| _RESULTS["nuscenes"] = {**s, "verdict": verdict} |
| print(json.dumps(_RESULTS["nuscenes"], indent=1, default=float)) |
| _report("task_gates.json", {"knobs": _knobs(), **_RESULTS}) |
| bad = [n for n, v in verdict.items() if not v["pass"]] |
| assert not bad, f"D20 task gates failed: {bad}" |
|
|
|
|
| def test_shipped_and_pandaset_report(model): |
| """Informational (recorded for the card): the shipped-sample and PandaSet det agreement of this path.""" |
| _RESULTS["shipped"] = {k: v for k, v in shipped_sample(model).items() if k != "report"} |
| try: |
| report, worst = pandaset_sequence(model) |
| _RESULTS["pandaset"] = {"worst": worst, "frames": {f: {k: r[k] for k in ("device", "reference", "recall", |
| "precision", "bev_embed_pcc") |
| if k in r} for f, r in report.items()}} |
| except pytest.skip.Exception as e: |
| _RESULTS["pandaset"] = {"skipped": str(e)} |
| print(json.dumps({k: _RESULTS[k] for k in ("shipped", "pandaset")}, indent=1, default=float)) |
| _report("task_gates.json", {"knobs": _knobs(), **_RESULTS}) |
|
|
