87 lines
3.2 KiB
Python
87 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
from types import ModuleType
|
|
|
|
import numpy as np
|
|
|
|
REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
|
|
RUNNER_PATH = REPOSITORY_ROOT / "experiments/perception/run_m48t_upstream_parity_worker.py"
|
|
PROFILE_PATH = REPOSITORY_ROOT / "config/perception/m48t-upstream-parity-v1.json"
|
|
|
|
|
|
def load_runner() -> ModuleType:
|
|
specification = importlib.util.spec_from_file_location("m48t_upstream_parity", RUNNER_PATH)
|
|
assert specification is not None and specification.loader is not None
|
|
module = importlib.util.module_from_spec(specification)
|
|
specification.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_upstream_parity_profile_freezes_official_and_deployed_providers() -> None:
|
|
profile = json.loads(PROFILE_PATH.read_text("utf-8"))
|
|
|
|
assert profile["model"]["package_version"] == "1.9.4"
|
|
assert profile["model"]["resolution"] == [704, 704]
|
|
assert profile["dataset"]["image_count"] == 5000
|
|
assert profile["providers"]["pytorch"]["confidence_prefilter"] == 0.0
|
|
assert profile["providers"]["tensorrt"]["confidence_prefilter"] == 0.0
|
|
assert profile["authority"]["navigation_or_safety_accepted"] is False
|
|
|
|
|
|
def test_tensorrt_decoder_uses_sparse_coco_ids_and_original_image_geometry() -> None:
|
|
runner = load_runner()
|
|
boxes = np.zeros((1, 300, 4), dtype=np.float16)
|
|
logits = np.full((1, 300, 91), -20, dtype=np.float16)
|
|
boxes[0, 4] = np.asarray((0.5, 0.5, 0.2, 0.4), dtype=np.float16)
|
|
logits[0, 4, 1] = np.float16(4.0) # COCO sparse id 1: person
|
|
logits[0, 7, 12] = np.float16(5.0) # sparse gap: must never escape
|
|
|
|
rows = runner.decode_tensorrt_coco_rows(
|
|
image_id=42,
|
|
image_width=1000,
|
|
image_height=500,
|
|
boxes=boxes,
|
|
logits=logits,
|
|
maximum_detections=2,
|
|
)
|
|
|
|
assert len(rows) == 1
|
|
assert rows[0]["image_id"] == 42
|
|
assert rows[0]["category_id"] == 1
|
|
assert np.allclose(rows[0]["bbox"], [400.0244, 149.9756, 199.9512, 200.0488], atol=0.1)
|
|
|
|
|
|
def test_parity_decision_localizes_upstream_and_deployment_failures() -> None:
|
|
runner = load_runner()
|
|
profile = json.loads(PROFILE_PATH.read_text("utf-8"))
|
|
|
|
passed = runner.build_parity_decision(
|
|
profile=profile,
|
|
pytorch_metrics={"ap_50_95": 0.565, "ap_50": 0.751},
|
|
tensorrt_metrics={"ap_50_95": 0.563, "ap_50": 0.749},
|
|
full_admission_run=True,
|
|
)
|
|
assert passed["passed"] is True
|
|
assert passed["diagnosis"] == "upstream-and-tensorrt-parity-passed"
|
|
|
|
deployment_failure = runner.build_parity_decision(
|
|
profile=profile,
|
|
pytorch_metrics={"ap_50_95": 0.565, "ap_50": 0.751},
|
|
tensorrt_metrics={"ap_50_95": 0.54, "ap_50": 0.72},
|
|
full_admission_run=True,
|
|
)
|
|
assert deployment_failure["passed"] is False
|
|
assert deployment_failure["diagnosis"] == "tensorrt-deployment-parity-failed"
|
|
|
|
smoke = runner.build_parity_decision(
|
|
profile=profile,
|
|
pytorch_metrics={"ap_50_95": 0.565, "ap_50": 0.751},
|
|
tensorrt_metrics={"ap_50_95": 0.565, "ap_50": 0.751},
|
|
full_admission_run=False,
|
|
)
|
|
assert smoke["passed"] is False
|
|
assert smoke["diagnosis"] == "bounded-smoke-only"
|