mirror of
https://github.com/azaion/gps-denied-onboard.git
synced 2026-06-22 18:01:13 +00:00
[AZ-339] C2 MegaLoc + MixVPR secondary VPR backbones
Adds two research-only VprStrategy implementations for the IT-12 comparative-study matrix. MegaLocStrategy (D=2048, 322x322) and MixVprStrategy (D=4096, 320x320), both via C7 TensorRT FP16 with their own concrete BackbonePreprocessor. Single-stage global L2 normalisation; retrieval delegated to FaissBridge; FDR records + structured logs identical to UltraVPR. BUILD_VPR_MEGALOC and BUILD_VPR_MIXVPR ON for research/replay-cli only, OFF for airborne and operator-tooling (fail-fast at composition root via existing AZ-336 factory). Uses helpers.iso_ts_from_clock from day 1 — no new timestamp helper duplicates introduced. 36 parametrised AC tests + 25 protocol-conformance + 18 helper regression tests pass; 1690 / 1690 unit tests pass (excluding 1 pre-existing flaky cold-start subprocess test in c12). Verdict: PASS_WITH_WARNINGS — one Medium follow-on (AZ-527 to consolidate 4-way _assert_engine_output_dim) + one Low AC wording drift. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Oleksandr Bezdieniezhnykh
@@ -0,0 +1,811 @@
|
||||
"""AZ-339 — MegaLoc + MixVPR secondary VprStrategy unit tests.
|
||||
|
||||
Covers AC-1..AC-11 for both strategies. Parametrised across the two
|
||||
strategies so the test surface stays compact (one test per AC times
|
||||
two strategies) and any drift between the two implementations is
|
||||
visible at the assertion level.
|
||||
|
||||
Uses fakes for :class:`InferenceRuntimeCut`, :class:`DescriptorIndexCut`,
|
||||
and :class:`FdrClient` so the suite stays AZ-507-clean and TRT-free
|
||||
(mirrors the precedent in ``test_ultra_vpr.py``).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any, Literal
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from gps_denied_onboard._types.calibration import CameraCalibration
|
||||
from gps_denied_onboard._types.inference import (
|
||||
BuildConfig,
|
||||
EngineCacheEntry,
|
||||
EngineHandle,
|
||||
PrecisionMode,
|
||||
)
|
||||
from gps_denied_onboard._types.nav import NavCameraFrame
|
||||
from gps_denied_onboard.components.c2_vpr import (
|
||||
C2VprConfig,
|
||||
VprStrategy,
|
||||
)
|
||||
from gps_denied_onboard.components.c2_vpr._faiss_bridge import FaissBridge
|
||||
from gps_denied_onboard.components.c2_vpr._preprocessor_mega_loc import (
|
||||
MegaLocBackbonePreprocessor,
|
||||
)
|
||||
from gps_denied_onboard.components.c2_vpr._preprocessor_mix_vpr import (
|
||||
MixVprBackbonePreprocessor,
|
||||
)
|
||||
from gps_denied_onboard.components.c2_vpr.errors import (
|
||||
VprBackboneError,
|
||||
VprPreprocessError,
|
||||
)
|
||||
from gps_denied_onboard.components.c2_vpr.mega_loc import (
|
||||
DESCRIPTOR_DIM as MEGA_LOC_DIM,
|
||||
)
|
||||
from gps_denied_onboard.components.c2_vpr.mega_loc import (
|
||||
MegaLocStrategy,
|
||||
)
|
||||
from gps_denied_onboard.components.c2_vpr.mega_loc import (
|
||||
create as create_mega_loc,
|
||||
)
|
||||
from gps_denied_onboard.components.c2_vpr.mix_vpr import (
|
||||
DESCRIPTOR_DIM as MIX_VPR_DIM,
|
||||
)
|
||||
from gps_denied_onboard.components.c2_vpr.mix_vpr import (
|
||||
MixVprStrategy,
|
||||
)
|
||||
from gps_denied_onboard.components.c2_vpr.mix_vpr import (
|
||||
create as create_mix_vpr,
|
||||
)
|
||||
from gps_denied_onboard.config.schema import Config, ConfigError
|
||||
from gps_denied_onboard.fdr_client import FdrClient
|
||||
from gps_denied_onboard.helpers.descriptor_normaliser import DescriptorNormaliser
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Parametrisation: each strategy + its bound constants
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _StrategySpec:
|
||||
name: str
|
||||
strategy_cls: type
|
||||
create_fn: Any
|
||||
preprocessor_cls: type
|
||||
descriptor_dim: int
|
||||
backbone_label: str
|
||||
input_hw: tuple[int, int]
|
||||
|
||||
|
||||
_SPECS: list[_StrategySpec] = [
|
||||
_StrategySpec(
|
||||
name="mega_loc",
|
||||
strategy_cls=MegaLocStrategy,
|
||||
create_fn=create_mega_loc,
|
||||
preprocessor_cls=MegaLocBackbonePreprocessor,
|
||||
descriptor_dim=MEGA_LOC_DIM,
|
||||
backbone_label="mega_loc",
|
||||
input_hw=(322, 322),
|
||||
),
|
||||
_StrategySpec(
|
||||
name="mix_vpr",
|
||||
strategy_cls=MixVprStrategy,
|
||||
create_fn=create_mix_vpr,
|
||||
preprocessor_cls=MixVprBackbonePreprocessor,
|
||||
descriptor_dim=MIX_VPR_DIM,
|
||||
backbone_label="mix_vpr",
|
||||
input_hw=(320, 320),
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
@pytest.fixture(params=_SPECS, ids=[s.name for s in _SPECS])
|
||||
def spec(request: pytest.FixtureRequest) -> _StrategySpec:
|
||||
return request.param
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fakes (mirrors test_ultra_vpr.py shape)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@dataclass
|
||||
class _StubClock:
|
||||
next_monotonic_ns: int = 1_000_000_000
|
||||
step_ns: int = 5_000
|
||||
fixed_time_ns: int = 1_715_600_000_000_000_000
|
||||
|
||||
def monotonic_ns(self) -> int:
|
||||
v = self.next_monotonic_ns
|
||||
self.next_monotonic_ns += self.step_ns
|
||||
return v
|
||||
|
||||
def time_ns(self) -> int:
|
||||
return self.fixed_time_ns
|
||||
|
||||
def sleep_until_ns(self, target_ns: int) -> None:
|
||||
_ = target_ns
|
||||
|
||||
|
||||
class _FakeEngineHandle(EngineHandle):
|
||||
def __init__(self, label: str) -> None:
|
||||
self.label = label
|
||||
|
||||
|
||||
@dataclass
|
||||
class _FakeInferenceRuntime:
|
||||
descriptor_dim: int = 2048
|
||||
raises: BaseException | None = None
|
||||
runtime_label: Literal["tensorrt", "onnx_trt_ep", "pytorch_fp16"] = (
|
||||
"tensorrt"
|
||||
)
|
||||
fixed_output: np.ndarray | None = None
|
||||
output_key: str = "embedding"
|
||||
calls: list[dict[str, np.ndarray]] = field(default_factory=list)
|
||||
deserialize_calls: list[EngineCacheEntry] = field(default_factory=list)
|
||||
model_name: str = "mega_loc"
|
||||
|
||||
def compile_engine(
|
||||
self, model_path: Path, build_config: BuildConfig
|
||||
) -> EngineCacheEntry:
|
||||
_ = build_config
|
||||
return EngineCacheEntry(
|
||||
engine_path=Path(model_path),
|
||||
sha256_hex="0" * 64,
|
||||
sm=None,
|
||||
jp=None,
|
||||
trt=None,
|
||||
precision=PrecisionMode.FP16,
|
||||
extras={"model_name": self.model_name},
|
||||
)
|
||||
|
||||
def deserialize_engine(self, entry: EngineCacheEntry) -> EngineHandle:
|
||||
self.deserialize_calls.append(entry)
|
||||
return _FakeEngineHandle(label=entry.extras.get("model_name", ""))
|
||||
|
||||
def infer(
|
||||
self, handle: EngineHandle, inputs: dict[str, np.ndarray]
|
||||
) -> dict[str, np.ndarray]:
|
||||
_ = handle
|
||||
self.calls.append({k: v.copy() for k, v in inputs.items()})
|
||||
if self.raises is not None:
|
||||
raise self.raises
|
||||
if self.fixed_output is not None:
|
||||
return {self.output_key: self.fixed_output.copy()}
|
||||
rng = np.random.default_rng(0xCAFEBABE)
|
||||
tensor = rng.standard_normal(self.descriptor_dim).astype(np.float16)
|
||||
return {
|
||||
self.output_key: tensor.reshape(1, self.descriptor_dim).copy()
|
||||
}
|
||||
|
||||
def release_engine(self, handle: EngineHandle) -> None:
|
||||
_ = handle
|
||||
|
||||
def current_runtime_label(
|
||||
self,
|
||||
) -> Literal["tensorrt", "onnx_trt_ep", "pytorch_fp16"]:
|
||||
return self.runtime_label
|
||||
|
||||
|
||||
@dataclass
|
||||
class _FakeDescriptorIndex:
|
||||
descriptor_dim_value: int = 2048
|
||||
results: list[tuple[tuple[int, float, float], float]] = field(
|
||||
default_factory=list
|
||||
)
|
||||
raises: BaseException | None = None
|
||||
|
||||
def search_topk(
|
||||
self, query: np.ndarray, k: int
|
||||
) -> list[tuple[tuple[int, float, float], float]]:
|
||||
_ = query
|
||||
if self.raises is not None:
|
||||
raise self.raises
|
||||
if not self.results:
|
||||
return [
|
||||
((18, 49.0 + i * 0.001, 36.0 + i * 0.001), 0.05 + 0.05 * i)
|
||||
for i in range(k)
|
||||
]
|
||||
return list(self.results[:k])
|
||||
|
||||
def descriptor_dim(self) -> int:
|
||||
return self.descriptor_dim_value
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixture helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_frame(*, frame_id: int = 4242, h: int = 720, w: int = 1280) -> NavCameraFrame:
|
||||
rng = np.random.default_rng(frame_id)
|
||||
image = rng.integers(0, 256, size=(h, w, 3), dtype=np.uint8)
|
||||
return NavCameraFrame(
|
||||
frame_id=frame_id,
|
||||
timestamp=datetime(2026, 5, 13, 12, 0, 0),
|
||||
image=image,
|
||||
camera_calibration_id="test_cam",
|
||||
)
|
||||
|
||||
|
||||
def _make_calibration(*, cx: float = 640.0, cy: float = 360.0) -> CameraCalibration:
|
||||
intrinsics = np.array(
|
||||
[
|
||||
[1000.0, 0.0, cx],
|
||||
[0.0, 1000.0, cy],
|
||||
[0.0, 0.0, 1.0],
|
||||
],
|
||||
dtype=np.float64,
|
||||
)
|
||||
return CameraCalibration(
|
||||
camera_id="test_cam",
|
||||
intrinsics_3x3=intrinsics,
|
||||
distortion=np.zeros(5, dtype=np.float64),
|
||||
body_to_camera_se3=np.eye(4, dtype=np.float64),
|
||||
acquisition_method="test_fixture",
|
||||
)
|
||||
|
||||
|
||||
def _make_fdr_client() -> FdrClient:
|
||||
return FdrClient(producer_id="c2_vpr", capacity=32, _emit_diag_log=False)
|
||||
|
||||
|
||||
def _build_strategy(
|
||||
spec: _StrategySpec,
|
||||
*,
|
||||
inference_runtime: _FakeInferenceRuntime | None = None,
|
||||
descriptor_index: _FakeDescriptorIndex | None = None,
|
||||
preprocessor: Any = None,
|
||||
fdr_client: FdrClient | None = None,
|
||||
clock: _StubClock | None = None,
|
||||
descriptor_dim: int | None = None,
|
||||
) -> Any:
|
||||
dim = spec.descriptor_dim if descriptor_dim is None else descriptor_dim
|
||||
inference_runtime = inference_runtime or _FakeInferenceRuntime(
|
||||
descriptor_dim=dim, model_name=spec.name
|
||||
)
|
||||
descriptor_index = descriptor_index or _FakeDescriptorIndex(
|
||||
descriptor_dim_value=dim
|
||||
)
|
||||
preprocessor = preprocessor or spec.preprocessor_cls()
|
||||
fdr_client = fdr_client or _make_fdr_client()
|
||||
clock = clock or _StubClock()
|
||||
handle = _FakeEngineHandle(label=spec.name)
|
||||
bridge = FaissBridge(
|
||||
descriptor_index=descriptor_index,
|
||||
descriptor_dim=dim,
|
||||
warn_top1_threshold=0.30,
|
||||
debug_log_per_frame_distances=False,
|
||||
fdr_client=fdr_client,
|
||||
logger=logging.getLogger(f"test.{spec.name}.bridge"),
|
||||
clock=clock,
|
||||
)
|
||||
return spec.strategy_cls(
|
||||
inference_runtime=inference_runtime,
|
||||
engine_handle=handle,
|
||||
descriptor_index=descriptor_index,
|
||||
preprocessor=preprocessor,
|
||||
normaliser=DescriptorNormaliser(),
|
||||
faiss_bridge=bridge,
|
||||
fdr_client=fdr_client,
|
||||
clock=clock,
|
||||
logger=logging.getLogger(f"test.{spec.name}"),
|
||||
descriptor_dim=dim,
|
||||
)
|
||||
|
||||
|
||||
def _build_config(strategy_name: str) -> Config:
|
||||
c2 = C2VprConfig(
|
||||
strategy=strategy_name,
|
||||
backbone_weights_path=Path(f"/models/{strategy_name}.trt"),
|
||||
faiss_index_path=Path("/cache/vpr/index.faiss"),
|
||||
warn_top1_threshold=0.30,
|
||||
debug_per_frame_distances=False,
|
||||
)
|
||||
cfg = MagicMock(spec=Config)
|
||||
cfg.components = {"c2_vpr": c2}
|
||||
return cfg
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# AC-1: Protocol conformance
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_ac1_protocol_conformance(spec: _StrategySpec) -> None:
|
||||
strategy = _build_strategy(spec)
|
||||
assert isinstance(strategy, VprStrategy)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# AC-2: embed_query → L2-normalised FP16 embedding of correct dim
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_ac2_embed_query_returns_unit_norm_fp16_correct_dim(
|
||||
spec: _StrategySpec,
|
||||
) -> None:
|
||||
# Arrange
|
||||
strategy = _build_strategy(spec)
|
||||
frame = _make_frame()
|
||||
calibration = _make_calibration()
|
||||
|
||||
# Act
|
||||
query = strategy.embed_query(frame, calibration)
|
||||
|
||||
# Assert
|
||||
embedding = np.asarray(query.embedding)
|
||||
assert embedding.shape == (spec.descriptor_dim,)
|
||||
assert embedding.dtype == np.float16
|
||||
norm = float(np.linalg.norm(embedding.astype(np.float32)))
|
||||
assert norm == pytest.approx(1.0, abs=1e-3)
|
||||
|
||||
|
||||
def test_ac2_single_stage_l2_no_intra_cluster_call(
|
||||
spec: _StrategySpec,
|
||||
) -> None:
|
||||
"""Secondary backbones use single-stage L2 (no NetVLAD-style intra-cluster step)."""
|
||||
# Arrange
|
||||
calls: list[str] = []
|
||||
|
||||
class _SpyNormaliser(DescriptorNormaliser):
|
||||
def l2_normalise(self, descriptor: np.ndarray) -> np.ndarray: # type: ignore[override]
|
||||
calls.append("l2_normalise")
|
||||
return DescriptorNormaliser.l2_normalise(descriptor)
|
||||
|
||||
def intra_cluster_normalise( # type: ignore[override]
|
||||
self, descriptor: np.ndarray, num_clusters: int
|
||||
) -> np.ndarray:
|
||||
calls.append("intra_cluster_normalise")
|
||||
return DescriptorNormaliser.intra_cluster_normalise(
|
||||
descriptor, num_clusters
|
||||
)
|
||||
|
||||
# Build manually to inject the spy
|
||||
inference_runtime = _FakeInferenceRuntime(descriptor_dim=spec.descriptor_dim)
|
||||
descriptor_index = _FakeDescriptorIndex(
|
||||
descriptor_dim_value=spec.descriptor_dim
|
||||
)
|
||||
fdr_client = _make_fdr_client()
|
||||
clock = _StubClock()
|
||||
bridge = FaissBridge(
|
||||
descriptor_index=descriptor_index,
|
||||
descriptor_dim=spec.descriptor_dim,
|
||||
warn_top1_threshold=0.30,
|
||||
debug_log_per_frame_distances=False,
|
||||
fdr_client=fdr_client,
|
||||
logger=logging.getLogger(f"test.{spec.name}.bridge"),
|
||||
clock=clock,
|
||||
)
|
||||
strategy = spec.strategy_cls(
|
||||
inference_runtime=inference_runtime,
|
||||
engine_handle=_FakeEngineHandle(spec.name),
|
||||
descriptor_index=descriptor_index,
|
||||
preprocessor=spec.preprocessor_cls(),
|
||||
normaliser=_SpyNormaliser(),
|
||||
faiss_bridge=bridge,
|
||||
fdr_client=fdr_client,
|
||||
clock=clock,
|
||||
logger=logging.getLogger(f"test.{spec.name}"),
|
||||
descriptor_dim=spec.descriptor_dim,
|
||||
)
|
||||
|
||||
# Act
|
||||
strategy.embed_query(_make_frame(), _make_calibration())
|
||||
|
||||
# Assert
|
||||
assert "intra_cluster_normalise" not in calls
|
||||
assert calls == ["l2_normalise"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# AC-3: deterministic embeddings
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_ac3_embed_query_deterministic_for_same_frame(
|
||||
spec: _StrategySpec,
|
||||
) -> None:
|
||||
# Arrange
|
||||
rng = np.random.default_rng(2026)
|
||||
fixed = rng.standard_normal(spec.descriptor_dim).astype(np.float16)
|
||||
fixed = fixed.reshape(1, spec.descriptor_dim)
|
||||
runtime = _FakeInferenceRuntime(
|
||||
descriptor_dim=spec.descriptor_dim, fixed_output=fixed
|
||||
)
|
||||
strategy = _build_strategy(spec, inference_runtime=runtime)
|
||||
frame = _make_frame()
|
||||
calibration = _make_calibration()
|
||||
|
||||
# Act
|
||||
first = strategy.embed_query(frame, calibration)
|
||||
second = strategy.embed_query(frame, calibration)
|
||||
third = strategy.embed_query(frame, calibration)
|
||||
|
||||
# Assert
|
||||
np.testing.assert_array_equal(
|
||||
np.asarray(first.embedding), np.asarray(second.embedding)
|
||||
)
|
||||
np.testing.assert_array_equal(
|
||||
np.asarray(second.embedding), np.asarray(third.embedding)
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# AC-4: retrieve_topk returns k candidates with correct backbone_label
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_ac4_retrieve_topk_returns_exactly_k_with_correct_label(
|
||||
spec: _StrategySpec,
|
||||
) -> None:
|
||||
# Arrange
|
||||
descriptor_index = _FakeDescriptorIndex(
|
||||
descriptor_dim_value=spec.descriptor_dim
|
||||
)
|
||||
strategy = _build_strategy(spec, descriptor_index=descriptor_index)
|
||||
|
||||
# Act
|
||||
query = strategy.embed_query(_make_frame(), _make_calibration())
|
||||
result = strategy.retrieve_topk(query, k=10)
|
||||
|
||||
# Assert
|
||||
assert len(result.candidates) == 10
|
||||
assert result.backbone_label == spec.backbone_label
|
||||
assert result.candidates[0].descriptor_dim == spec.descriptor_dim
|
||||
distances = [c.descriptor_distance for c in result.candidates]
|
||||
assert distances == sorted(distances)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# AC-5: descriptor_dim() is stable
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_ac5_descriptor_dim_stable(spec: _StrategySpec) -> None:
|
||||
# Arrange
|
||||
strategy = _build_strategy(spec)
|
||||
# Act / Assert
|
||||
for _ in range(100):
|
||||
assert strategy.descriptor_dim() == spec.descriptor_dim
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# AC-6: Engine output shape mismatch → ConfigError at create()
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_ac6_create_rejects_engine_output_shape_mismatch(
|
||||
spec: _StrategySpec,
|
||||
) -> None:
|
||||
# Arrange — engine produces (1, 100), expected (1, spec.descriptor_dim)
|
||||
wrong = np.zeros((1, 100), dtype=np.float16)
|
||||
runtime = _FakeInferenceRuntime(
|
||||
descriptor_dim=spec.descriptor_dim,
|
||||
fixed_output=wrong,
|
||||
model_name=spec.name,
|
||||
)
|
||||
descriptor_index = _FakeDescriptorIndex(
|
||||
descriptor_dim_value=spec.descriptor_dim
|
||||
)
|
||||
|
||||
# Act + Assert
|
||||
with pytest.raises(ConfigError, match=r"engine output shape mismatch"):
|
||||
spec.create_fn(
|
||||
_build_config(spec.name),
|
||||
descriptor_index=descriptor_index,
|
||||
inference_runtime=runtime,
|
||||
fdr_client=_make_fdr_client(),
|
||||
clock=_StubClock(),
|
||||
)
|
||||
|
||||
|
||||
def test_ac6_create_rejects_missing_embedding_key(
|
||||
spec: _StrategySpec,
|
||||
) -> None:
|
||||
# Arrange
|
||||
runtime = _FakeInferenceRuntime(
|
||||
descriptor_dim=spec.descriptor_dim,
|
||||
output_key="wrong_key",
|
||||
model_name=spec.name,
|
||||
)
|
||||
|
||||
# Act + Assert
|
||||
with pytest.raises(ConfigError, match=r"'embedding' key absent"):
|
||||
spec.create_fn(
|
||||
_build_config(spec.name),
|
||||
descriptor_index=_FakeDescriptorIndex(
|
||||
descriptor_dim_value=spec.descriptor_dim
|
||||
),
|
||||
inference_runtime=runtime,
|
||||
fdr_client=_make_fdr_client(),
|
||||
clock=_StubClock(),
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# AC-7: VprBackboneError on forward-pass failure
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_ac7_runtime_error_yields_vpr_backbone_error(
|
||||
spec: _StrategySpec, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
# Arrange
|
||||
runtime = _FakeInferenceRuntime(
|
||||
descriptor_dim=spec.descriptor_dim, raises=RuntimeError("CUDA OOM")
|
||||
)
|
||||
fdr_client = _make_fdr_client()
|
||||
strategy = _build_strategy(
|
||||
spec, inference_runtime=runtime, fdr_client=fdr_client
|
||||
)
|
||||
|
||||
# Act
|
||||
with caplog.at_level(logging.ERROR, logger=f"test.{spec.name}"):
|
||||
with pytest.raises(VprBackboneError):
|
||||
strategy.embed_query(_make_frame(), _make_calibration())
|
||||
|
||||
# Assert
|
||||
assert any(
|
||||
record.levelno == logging.ERROR
|
||||
and getattr(record, "kind", None) == "c2.vpr.backbone_error"
|
||||
for record in caplog.records
|
||||
)
|
||||
records: list[Any] = []
|
||||
while True:
|
||||
r = fdr_client.pop_one()
|
||||
if r is None:
|
||||
break
|
||||
records.append(r)
|
||||
backbone_errors = [r for r in records if r.kind == "vpr.backbone_error"]
|
||||
assert len(backbone_errors) == 1
|
||||
|
||||
|
||||
def test_ac7_wrong_forward_output_shape_yields_vpr_backbone_error(
|
||||
spec: _StrategySpec,
|
||||
) -> None:
|
||||
# Arrange
|
||||
bad = np.zeros((1, 100), dtype=np.float16)
|
||||
runtime = _FakeInferenceRuntime(
|
||||
descriptor_dim=spec.descriptor_dim, fixed_output=bad
|
||||
)
|
||||
strategy = _build_strategy(spec, inference_runtime=runtime)
|
||||
|
||||
# Act + Assert
|
||||
with pytest.raises(
|
||||
VprBackboneError, match=rf"expected \(1, {spec.descriptor_dim}\)"
|
||||
):
|
||||
strategy.embed_query(_make_frame(), _make_calibration())
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# AC-8: VprPreprocessError on corrupt image bytes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_ac8_corrupt_image_yields_vpr_preprocess_error(
|
||||
spec: _StrategySpec, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
# Arrange
|
||||
fdr_client = _make_fdr_client()
|
||||
strategy = _build_strategy(spec, fdr_client=fdr_client)
|
||||
frame = NavCameraFrame(
|
||||
frame_id=4242,
|
||||
timestamp=datetime(2026, 5, 13, 12, 0, 0),
|
||||
image="not-an-array",
|
||||
camera_calibration_id="test_cam",
|
||||
)
|
||||
|
||||
# Act
|
||||
with caplog.at_level(logging.ERROR, logger=f"test.{spec.name}"):
|
||||
with pytest.raises(VprPreprocessError):
|
||||
strategy.embed_query(frame, _make_calibration())
|
||||
|
||||
# Assert
|
||||
assert any(
|
||||
record.levelno == logging.ERROR
|
||||
and getattr(record, "kind", None) == "c2.vpr.preprocess_error"
|
||||
for record in caplog.records
|
||||
)
|
||||
records: list[Any] = []
|
||||
while True:
|
||||
r = fdr_client.pop_one()
|
||||
if r is None:
|
||||
break
|
||||
records.append(r)
|
||||
preprocess_errors = [
|
||||
r for r in records if r.kind == "vpr.preprocess_error"
|
||||
]
|
||||
assert len(preprocess_errors) == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# AC-9: Composition-root wiring + INFO "c2.vpr.ready" log emitted
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_ac9_create_emits_ready_log_with_correct_label_and_dim(
|
||||
spec: _StrategySpec, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
# Arrange
|
||||
logger_name = f"gps_denied_onboard.c2_vpr.{spec.name}"
|
||||
runtime = _FakeInferenceRuntime(
|
||||
descriptor_dim=spec.descriptor_dim, model_name=spec.name
|
||||
)
|
||||
descriptor_index = _FakeDescriptorIndex(
|
||||
descriptor_dim_value=spec.descriptor_dim
|
||||
)
|
||||
|
||||
# Act
|
||||
with caplog.at_level(logging.INFO, logger=logger_name):
|
||||
strategy = spec.create_fn(
|
||||
_build_config(spec.name),
|
||||
descriptor_index=descriptor_index,
|
||||
inference_runtime=runtime,
|
||||
fdr_client=_make_fdr_client(),
|
||||
clock=_StubClock(),
|
||||
)
|
||||
|
||||
# Assert
|
||||
assert isinstance(strategy, spec.strategy_cls)
|
||||
ready_records = [
|
||||
r for r in caplog.records if getattr(r, "kind", None) == "c2.vpr.ready"
|
||||
]
|
||||
assert len(ready_records) == 1
|
||||
kv = getattr(ready_records[0], "kv", {})
|
||||
assert kv == {"strategy": spec.backbone_label, "descriptor_dim": spec.descriptor_dim}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# AC-10: Build-flag exclusion → composition-time fail-fast
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_ac10_runtime_label_mismatch_raises_config_error(
|
||||
spec: _StrategySpec,
|
||||
) -> None:
|
||||
"""Selecting a secondary backbone on a binary built without the
|
||||
TRT / ONNX-RT runtimes fails fast at create-time.
|
||||
|
||||
Note: AC-10 of the task spec literally names ``ConfigurationError``;
|
||||
the existing factory contract (AZ-336) raises
|
||||
``StrategyNotAvailableError`` via the BUILD_VPR_* env-flag check
|
||||
BEFORE create() is reached, but the strategy module's own runtime
|
||||
label guard surfaces a ``ConfigError`` for the same intent
|
||||
(wrong runtime). Both are composition-time fail-fast errors.
|
||||
"""
|
||||
# Arrange
|
||||
runtime = _FakeInferenceRuntime(
|
||||
descriptor_dim=spec.descriptor_dim,
|
||||
runtime_label="pytorch_fp16",
|
||||
model_name=spec.name,
|
||||
)
|
||||
|
||||
# Act + Assert
|
||||
with pytest.raises(ConfigError, match=r"BUILD_TENSORRT_RUNTIME"):
|
||||
spec.create_fn(
|
||||
_build_config(spec.name),
|
||||
descriptor_index=_FakeDescriptorIndex(
|
||||
descriptor_dim_value=spec.descriptor_dim
|
||||
),
|
||||
inference_runtime=runtime,
|
||||
fdr_client=_make_fdr_client(),
|
||||
clock=_StubClock(),
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# AC-11: Preprocessor input shape
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_ac11_preprocessor_input_shape(spec: _StrategySpec) -> None:
|
||||
# Arrange
|
||||
preprocessor = spec.preprocessor_cls()
|
||||
# Act + Assert
|
||||
assert preprocessor.input_shape() == spec.input_hw
|
||||
|
||||
|
||||
def test_preprocess_output_is_nchw_fp16(spec: _StrategySpec) -> None:
|
||||
# Arrange
|
||||
preprocessor = spec.preprocessor_cls()
|
||||
frame = _make_frame()
|
||||
calibration = _make_calibration()
|
||||
|
||||
# Act
|
||||
tensor = preprocessor.preprocess(frame, calibration)
|
||||
|
||||
# Assert
|
||||
h, w = spec.input_hw
|
||||
assert tensor.shape == (1, 3, h, w)
|
||||
assert tensor.dtype == np.float16
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Constructor validation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_constructor_rejects_zero_descriptor_dim(spec: _StrategySpec) -> None:
|
||||
# Arrange (skip _build_strategy to bypass FaissBridge's own validation)
|
||||
fdr_client = _make_fdr_client()
|
||||
clock = _StubClock()
|
||||
descriptor_index = _FakeDescriptorIndex(
|
||||
descriptor_dim_value=spec.descriptor_dim
|
||||
)
|
||||
bridge = FaissBridge(
|
||||
descriptor_index=descriptor_index,
|
||||
descriptor_dim=spec.descriptor_dim,
|
||||
warn_top1_threshold=0.30,
|
||||
debug_log_per_frame_distances=False,
|
||||
fdr_client=fdr_client,
|
||||
logger=logging.getLogger(f"test.{spec.name}.bridge"),
|
||||
clock=clock,
|
||||
)
|
||||
|
||||
# Act + Assert
|
||||
with pytest.raises(ValueError, match=r"descriptor_dim must be >= 1"):
|
||||
spec.strategy_cls(
|
||||
inference_runtime=_FakeInferenceRuntime(
|
||||
descriptor_dim=spec.descriptor_dim, model_name=spec.name
|
||||
),
|
||||
engine_handle=_FakeEngineHandle(spec.name),
|
||||
descriptor_index=descriptor_index,
|
||||
preprocessor=spec.preprocessor_cls(),
|
||||
normaliser=DescriptorNormaliser(),
|
||||
faiss_bridge=bridge,
|
||||
fdr_client=fdr_client,
|
||||
clock=clock,
|
||||
logger=logging.getLogger(f"test.{spec.name}"),
|
||||
descriptor_dim=0,
|
||||
)
|
||||
|
||||
|
||||
def test_create_requires_fdr_client(spec: _StrategySpec) -> None:
|
||||
# Arrange + Act + Assert
|
||||
with pytest.raises(ValueError, match=r"fdr_client is required"):
|
||||
spec.create_fn(
|
||||
_build_config(spec.name),
|
||||
descriptor_index=_FakeDescriptorIndex(
|
||||
descriptor_dim_value=spec.descriptor_dim
|
||||
),
|
||||
inference_runtime=_FakeInferenceRuntime(
|
||||
descriptor_dim=spec.descriptor_dim, model_name=spec.name
|
||||
),
|
||||
fdr_client=None,
|
||||
clock=_StubClock(),
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# FDR emission on success path
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_embed_query_emits_fdr_record(spec: _StrategySpec) -> None:
|
||||
# Arrange
|
||||
fdr_client = _make_fdr_client()
|
||||
strategy = _build_strategy(spec, fdr_client=fdr_client)
|
||||
|
||||
# Act
|
||||
strategy.embed_query(_make_frame(), _make_calibration())
|
||||
|
||||
# Assert
|
||||
records: list[Any] = []
|
||||
while True:
|
||||
r = fdr_client.pop_one()
|
||||
if r is None:
|
||||
break
|
||||
records.append(r)
|
||||
embed = [r for r in records if r.kind == "vpr.embed_query"]
|
||||
assert len(embed) == 1
|
||||
payload = embed[0].payload
|
||||
assert payload["backbone_label"] == spec.backbone_label
|
||||
assert payload["descriptor_dim"] == spec.descriptor_dim
|
||||
assert payload["latency_us"] >= 1
|
||||
Reference in New Issue
Block a user