code/tt_bevformer/tests/test_task_device.py

# SPDX-License-Identifier: Apache-2.0
"""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 (  # noqa: F401 -- fixtures
    _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})