# SPDX-License-Identifier: Apache-2.0 """End-to-end agreement vs the CPU reference on PUBLIC sample data, API == server, and the first-cut latency (needs one p150; milestone M3 of PORT_LOG.md). bin/devrun -t 1800 -- python -m pytest -q -s code/tt_frnet/tests/test_e2e_device.py The model is built once per variant (``FRNet.from_pretrained(device=..., variant=...)``: weights, graph, trace capture) and every test runs the production path: the host pre-processing, the traced device graph, the host post-processing. No dataset-level metric: the gate is agreement with the fp32 CPU reference of the same Autoware network on the same inputs (ONNX Runtime's ``pred_probs`` stored in each research golden): - ``test_sample_agreement``: the PandaSet 019 f40 sample (staged, not shipped: ``tests/_data.py``) through the API (``model(points, calibration=...)``): the device ``pred_probs`` of the frame vs ORT's (the research golden of the same frame; the host pre-processing equals the research one bit for bit, ``test_host_vs_research.py``), and the API's labels / filter mask vs its stored reference body (``..reference.json``); - ``test_shipped_sample``: the shipped synthetic sample of the variant (``samples/synthetic_.npz`` with its preset) through the API vs its stored CPU-reference body (``samples/synthetic_..reference.json``, what the container smoke compares): label and filter-mask agreement on every output point; - ``test_public_frames``: every catalogued golden of the variant (``tests/_data.py`` ``FRAMES``: 16 OT128 + 4 QT128 public frames, the VLS-128 x > 0 halves), fed their stored network inputs through ``FRNet.device_probs``; inside the workspace a missing golden fails the test (the frame-set gates never run on a subset); - ``test_api_equals_server``: ``/predict``'s handler and ``model(...)`` give identical outputs (shipped sample); - ``test_latency``: the first-cut latency (trace replay back to back, end to end through the API), recorded in ``logs/frnet/latency_.json`` (no gate: speed is the optimization phase's). Gates (PLAN.md 2.8; frozen in ``test_e2e_device.gates.json``; keep ``server/smoke_test.py`` ``REFERENCE_GATES`` in line): per frame ``pred_probs`` **PCC >= 0.998** and **argmax agreement >= 98.5 %** on the raw points; for the set, the minimum over its frames is gated and every frame is reported in ``logs/frnet/e2e_.json``. """ from __future__ import annotations import base64 import json import os import time import numpy as np import pytest from tt_frnet.tests import _data from tt_frnet.ttaw.golden import GateRegistry pytestmark = pytest.mark.device VARIANTS = ("ot128", "qt128") LOG_DIR = _data.WORKSPACE / "logs" / "frnet" TAG = os.environ.get("FRNET_E2E_TAG", "") # report-name suffix of an A/B run (e.g. FRNET_POINT_TERMS=2) GATES = GateRegistry.for_test(__file__, { **{f"{v}.sample.pred_probs_pcc": (0.998, "min", "pcc") for v in VARIANTS}, **{f"{v}.sample.argmax_agreement_raw": (0.985, "min", "argmax_agreement") for v in VARIANTS}, **{f"{v}.frames.min_pred_probs_pcc": (0.998, "min", "pcc") for v in VARIANTS}, **{f"{v}.frames.min_argmax_agreement_raw": (0.985, "min", "argmax_agreement") for v in VARIANTS}, # the shipped synthetic sample vs its stored CPU-reference body (the container smoke's REFERENCE_GATES) **{f"{v}.shipped.label_agreement": (0.985, "min", "label_agreement") for v in VARIANTS}, **{f"{v}.shipped.filter_agreement": (0.985, "min", "label_agreement") for v in VARIANTS}, }) # Frames per variant the frame-set gates must see (FRAMES: 16 OT128 / 4 QT128 public frames + the VLS-128 half). EXPECTED_FRAMES = {"ot128": 17, "qt128": 5} _REPORT: dict = {} def _save(variant: str, key: str, value) -> None: _REPORT.setdefault(variant, {})[key] = value LOG_DIR.mkdir(parents=True, exist_ok=True) (LOG_DIR / f"e2e_{variant}{TAG}.json").write_text(json.dumps(_REPORT[variant], indent=1, default=float)) @pytest.fixture(scope="module", params=VARIANTS) def model(request, device): from tt_frnet import FRNet t0 = time.perf_counter() m = FRNet.from_pretrained(device=device, variant=request.param, weights_dir=str(_data.weights_dir())) m.load_s = time.perf_counter() - t0 yield m m.close() def memory_view(device) -> dict: """Allocated bytes of the trace region and of DRAM (``ttnn.get_memory_view``; {} if unavailable).""" import ttnn out = {} for name, bt in (("trace", "TRACE"), ("dram", "DRAM")): try: mv = ttnn.get_memory_view(device, getattr(ttnn.BufferType, bt)) out[f"{name}_bytes_allocated"] = int(mv.total_bytes_allocated_per_bank) * int(mv.num_banks) out[f"{name}_bytes_total"] = int(mv.total_bytes_per_bank) * int(mv.num_banks) except Exception as e: # noqa: BLE001 -- reporting only out[f"{name}_error"] = repr(e) return out def frame_metrics(probs: np.ndarray, ref: np.ndarray, n_raw: int) -> dict: from tt_frnet.ttaw.metrics import argmax_agreement, error_stats stats = error_stats(probs, ref) return {"pcc": stats["pcc"], "max_abs": stats["max_abs"], "rel_l2": stats["rel_l2"], "argmax_agreement_raw": argmax_agreement(probs[:n_raw], ref[:n_raw]), "argmax_agreement_all": argmax_agreement(probs, ref), "n": int(len(ref)), "n_raw": int(n_raw)} def test_sample_agreement(model): """The API on the staged PandaSet sample: device ``pred_probs`` vs ORT, labels vs the stored reference body.""" from tt_frnet.ttaw.metrics import label_agreement v = model.variant _data.need_pandaset() name = f"{_data.SAMPLE_STEM}.{v}" inputs_ref, ort_probs, _ = _data.load_frame(name) prepared = model._prepare(str(_data.SAMPLE), calibration=_data.SAMPLE_CALIB) assert prepared.skipped is None for key in ("points", "coors", "voxel_coors", "inverse_map"): # identical network inputs assert np.array_equal(getattr(prepared.inputs, key), getattr(inputs_ref, key)), key with model._lock: probs = model.device_probs(prepared.plan) m = frame_metrics(probs, np.asarray(ort_probs, np.float32), prepared.inputs.num_points_raw) out = model(str(_data.SAMPLE), calibration=_data.SAMPLE_CALIB).to_dict() ref_body = json.loads((_data.SAMPLE_DIR / f"{_data.SAMPLE_STEM}.{v}.reference.json").read_text()) from tt_frnet.ttaw.server.smoke import decode_array got_labels = np.asarray(decode_array(out["labels"]).values) ref_labels = np.asarray(decode_array(ref_body["labels"]).values) m["api_label_agreement"] = label_agreement(got_labels, ref_labels) m["api_filter_agreement"] = float(np.mean(np.asarray(decode_array(out["filter_keep"]).values) == np.asarray(decode_array(ref_body["filter_keep"]).values))) m["timing_ms"] = out["timing_ms"] _save(v, "sample", m) print(f"[{v}] sample: {m}") r1 = GATES.check(f"{v}.sample.pred_probs_pcc", m["pcc"]) r2 = GATES.check(f"{v}.sample.argmax_agreement_raw", m["argmax_agreement_raw"]) print(r1, r2) assert r1.passed and r2.passed assert m["api_label_agreement"] >= 0.985 # the smoke test's REFERENCE_GATES assert out["num_points"] == len(ref_labels) and out["model"] == "frnet-p150" def test_shipped_sample(model): """The shipped synthetic sample of the variant through the API vs its stored CPU-reference body (what the container smoke test compares, with the same 98.5 % label gate).""" from tt_frnet.ttaw.server.smoke import decode_array from tt_frnet.ttaw.metrics import label_agreement v = model.variant sample = _data.shipped(v) ref_body = json.loads((_data.SAMPLES / f"synthetic_{v}.{v}.reference.json").read_text()) out = model(sample, calibration=_data.shipped_calib(v)).to_dict() assert out["num_points"] == ref_body["num_points"] and out["num_points_raw"] == ref_body["num_points_raw"] got_scores = np.asarray(decode_array(out["scores"]).values, np.float64) ref_scores = np.asarray(decode_array(ref_body["scores"]).values, np.float64) m = {"label_agreement": label_agreement(np.asarray(decode_array(out["labels"]).values), np.asarray(decode_array(ref_body["labels"]).values)), "filter_agreement": float(np.mean(np.asarray(decode_array(out["filter_keep"]).values) == np.asarray(decode_array(ref_body["filter_keep"]).values))), "max_abs_score_err": float(np.max(np.abs(got_scores - ref_scores))), "num_points": out["num_points"], "num_points_raw": out["num_points_raw"], "class_counts": out["class_counts"], "timing_ms": out["timing_ms"]} _save(v, "shipped", m) print(f"[{v}] shipped: {m}") r1 = GATES.check(f"{v}.shipped.label_agreement", m["label_agreement"]) r2 = GATES.check(f"{v}.shipped.filter_agreement", m["filter_agreement"]) print(r1, r2) assert r1.passed and r2.passed def test_public_frames(model): from tt_frnet.host.preprocess import build_plan v = model.variant names = [n for n, (_, _, var) in _data.FRAMES.items() if var == v] assert len(names) == EXPECTED_FRAMES[v], names per = {} for name in names: inputs, ort_probs, _ = _data.load_frame(name) # a missing golden fails inside the workspace plan = build_plan(inputs, model.config, pre_norm=(model.frnet_weights.pre_norm_scale, model.frnet_weights.pre_norm_shift)) with model._lock: probs = model.device_probs(plan) per[name] = frame_metrics(probs, np.asarray(ort_probs, np.float32), inputs.num_points_raw) per[name]["m"] = plan.m print(f"[{v}] {name}: pcc={per[name]['pcc']:.6f} agree_raw={per[name]['argmax_agreement_raw']:.5f}") assert len(per) == EXPECTED_FRAMES[v], sorted(per) worst_pcc = min(p["pcc"] for p in per.values()) worst_agree = min(p["argmax_agreement_raw"] for p in per.values()) _save(v, "frames", {"per_frame": per, "min_pred_probs_pcc": worst_pcc, "min_argmax_agreement_raw": worst_agree}) r1 = GATES.check(f"{v}.frames.min_pred_probs_pcc", worst_pcc) r2 = GATES.check(f"{v}.frames.min_argmax_agreement_raw", worst_agree) print(r1, r2) assert r1.passed and r2.passed def test_api_equals_server(model, monkeypatch): """The server's ``/predict`` handler and ``model(...)`` give identical outputs (same decoders, same graph).""" pytest.importorskip("fastapi") from tt_frnet.server.app import PredictRequest, app state = app.state.ttaw monkeypatch.setattr(state, "model", model) monkeypatch.setattr(state, "ready", True) sample, calib = _data.shipped(model.variant), _data.shipped_calib(model.variant) req = PredictRequest(points={"format": "npz", "data": base64.b64encode(sample.read_bytes()).decode()}, calibration=calib) served = state.predict(req) api = model(sample, calibration=calib).to_dict() for key in ("labels", "scores", "filter_keep", "points", "input_index", "class_counts", "num_points"): assert served[key] == api[key], key def test_latency(model): """First-cut latency: trace replay back to back and the API end to end on the sample (reported only).""" import ttnn v = model.variant _data.need_pandaset() model(_data.SAMPLE, calibration=_data.SAMPLE_CALIB) # inputs of the sample in the trace inputs n = 10 t0 = time.perf_counter() model.runner.replay("default", n=n) ttnn.synchronize_device(model.device) replay_ms = (time.perf_counter() - t0) / n * 1e3 timings = [] for _ in range(3): out = model(_data.SAMPLE, calibration=_data.SAMPLE_CALIB) timings.append(out.timing_ms) e2e = {k: float(np.median([t[k] for t in timings])) for k in timings[0]} rec = {"replay_ms": round(replay_ms, 2), "api_timing_ms_median": e2e, "load_s": round(model.load_s, 1), "device": model.info.get("device"), "memory": memory_view(model.device), "knobs": model.graph.knobs.as_dict(), "num_command_queues": model.runner.num_command_queues} _save(v, "latency", rec) (LOG_DIR / f"latency_{v}{TAG}.json").write_text(json.dumps(rec, indent=1, default=float)) print(f"[{v}] latency: {rec}")