| |
| """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 (``<stem>.<variant>.reference.json``); |
| - ``test_shipped_sample``: the shipped synthetic sample of the variant (``samples/synthetic_<variant>.npz`` with its |
| preset) through the API vs its stored CPU-reference body (``samples/synthetic_<variant>.<variant>.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_<variant>.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_<variant>.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", "") |
| 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}, |
| |
| **{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}, |
| }) |
| |
| 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: |
| 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"): |
| 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 |
| 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) |
| 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) |
| 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}") |
|
|
