azaion/gps-denied-onboard
1
0
Fork 0
mirror of https://github.com/azaion/gps-denied-onboard.git synced 2026-06-22 22:21:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

[AZ-337] C2 UltraVPR primary backbone VprStrategy

Browse Source 
UltraVPR is the Documentary Lead's PRIMARY backbone per
description.md § 1 and is wired by default
(config.c2_vpr.strategy = "ultra_vpr"). Runs on the C7 TensorRT
runtime (AZ-298) or ONNX-Runtime fallback (AZ-299); explicitly NOT
on the PyTorch FP16 runtime so a TRT engine compile bug can fall
back to NetVLAD without simultaneously breaking both strategies.

Production changes:
- c2_vpr/ultra_vpr.py - UltraVprStrategy + module-level create()
  factory. embed_query pipeline: preprocess -> runtime.infer ->
  single-stage L2 -> VprQuery. retrieve_topk delegates one-line to
  FaissBridge. Engine load + output-shape assertion happen at
  create() time (AC-6) so misconfiguration surfaces at startup,
  not 17 minutes into a flight. UltraVPR has D=512 fixed (NOT a
  config knob; AC-5 / AC-6 / AC-7 all assume 512). Single-stage L2
  (no intra-cluster step like NetVLAD; spy-test enforces this so a
  future refactor cannot silently regress recall).
- c2_vpr/_preprocessor_ultra_vpr.py - centre-crop using the camera
  calibration's principal point (cx, cy from intrinsics_3x3),
  falling back to geometric centre + WARN log when calibration is
  absent (AC-9). Resize -> (384, 384) -> ImageNet mean/std ->
  FP16 NCHW.
- No composition-root changes: UltraVPR consumes a pre-compiled
  .trt engine (no PyTorch nn.Module), so the strategy module does
  NOT expose MODEL_NAME / architecture_factory. The composition-
  root _register_strategy_architecture helper no-ops cleanly for
  this case (verified by test_create_does_not_register_pytorch_architecture).

Tests:
- tests/unit/c2_vpr/test_ultra_vpr.py - 29 tests covering all 12
  ACs + preprocessor contract + constructor validation + FDR
  record emission + single-stage L2 enforcement.

Full unit suite: 1637 passed / 80 env-skipped (+29 new tests).
Per-batch code review (batch_47_review.md): PASS_WITH_WARNINGS
(3 Low-severity findings; no Critical / High / Medium):
- F1: _iso_ts_from_clock is now the 7th copy (AZ-508 will close).
- F2: AZ-337 spec uses outdated C7 API names; affects upcoming
  AZ-339 / AZ-340. Spec-hygiene PBI recommended.
- F3: principal-point fallback uses (0, 0) zero-detection for
  missing calibration; safe but tightens when intrinsics become
  Optional.

Architectural notes:
- AZ-507 layering clean. Imports only InferenceRuntimeCut,
  DescriptorIndexCut, c2_vpr internals, _types, helpers,
  clock, fdr_client. Architecture lint test passes.
- Pattern parity with NetVLAD (B46) where semantics permit;
  UltraVPR-specific paths (single-stage L2, 'embedding' output
  key, TRT runtime, no architecture registry, principal-point
  crop) are clearly localised.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Oleksandr Bezdieniezhnykh
2026-05-13 22:43:17 +03:00
parent 773d589d34
commit 3c4fd272f1
5 changed files with 1692 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/gps_denied_onboard/components/c2_vpr/_preprocessor_ultra_vpr.py
+212
View File
@@ -0,0 +1,212 @@
"""UltraVPR backbone preprocessor (AZ-337).
UltraVPR's published preprocessing chain (per the research code drop):
decode the nav-camera frame's image to RGB uint8, centre-crop to a square
region respecting the camera calibration's principal point (or geometric
centre + WARN log when calibration is absent), resize to ``(384, 384)``,
apply ImageNet mean/std normalisation, cast to FP16, reshape to NCHW.
Differences from :class:`NetVladBackbonePreprocessor`:
- 384x384 input shape (vs 480x480 for NetVLAD).
- Calibration is CONSUMED — the principal point ``(cx, cy)`` from
``intrinsics_3x3`` anchors the centre-crop instead of using the
image's geometric centre. This matches the upstream UltraVPR
contract (AC-9: fall back to geometric centre + WARN when
calibration is unusable).
This preprocessor is C2-internal and owned exclusively by
:class:`UltraVprStrategy` — sharing across backbones is forbidden per
``components/02_c2_vpr/description.md`` § 6.
"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Final
import cv2
import numpy as np
from gps_denied_onboard.components.c2_vpr.errors import VprPreprocessError
if TYPE_CHECKING:
from gps_denied_onboard._types.calibration import CameraCalibration
from gps_denied_onboard._types.nav import NavCameraFrame
__all__ = [
"IMAGENET_MEAN",
"IMAGENET_STD",
"ULTRA_VPR_INPUT_HW",
"UltraVprBackbonePreprocessor",
]
ULTRA_VPR_INPUT_HW: Final[tuple[int, int]] = (384, 384)
IMAGENET_MEAN: Final[tuple[float, float, float]] = (0.485, 0.456, 0.406)
IMAGENET_STD: Final[tuple[float, float, float]] = (0.229, 0.224, 0.225)
_COMPONENT: Final[str] = "c2_vpr"
_LOG_KIND_CALIBRATION_MISSING: Final[str] = "c2.vpr.calibration_missing"
class UltraVprBackbonePreprocessor:
"""Centre-crop (principal-point-aware) + resize + ImageNet-normalise + FP16 NCHW."""
def __init__(
self,
*,
input_shape: tuple[int, int] = ULTRA_VPR_INPUT_HW,
mean: tuple[float, float, float] = IMAGENET_MEAN,
std: tuple[float, float, float] = IMAGENET_STD,
logger: logging.Logger | None = None,
) -> None:
if (
not isinstance(input_shape, tuple)
or len(input_shape) != 2
or any(not isinstance(v, int) or v <= 0 for v in input_shape)
):
raise ValueError(
f"UltraVprBackbonePreprocessor.input_shape must be a (H, W) "
f"tuple of positive ints; got {input_shape!r}"
)
if len(mean) != 3 or len(std) != 3:
raise ValueError(
"UltraVprBackbonePreprocessor.mean and std must each be "
"3-tuples (one per channel)"
)
if any(v <= 0 for v in std):
raise ValueError(
"UltraVprBackbonePreprocessor.std components must be > 0"
)
self._input_shape: tuple[int, int] = input_shape
self._mean: np.ndarray = np.array(mean, dtype=np.float32).reshape(1, 1, 3)
self._std: np.ndarray = np.array(std, dtype=np.float32).reshape(1, 1, 3)
self._logger: logging.Logger = (
logger
if logger is not None
else logging.getLogger("gps_denied_onboard.c2_vpr.ultra_vpr")
)
def preprocess(
self,
frame: NavCameraFrame,
calibration: CameraCalibration,
) -> np.ndarray:
"""Decode -> centre-crop (principal-point-aware) -> resize -> normalise -> FP16 NCHW.
Per AZ-337 AC-9: when calibration is absent or its principal
point cannot be extracted from ``intrinsics_3x3``, fall back to
the image's geometric centre and emit ONE WARN log per call
with ``kind="c2.vpr.calibration_missing"``. Preprocessing
otherwise succeeds and AC-2 still holds.
Raises:
:class:`VprPreprocessError` on shape / dtype / decode
violations.
"""
image = self._coerce_to_rgb_uint8(frame.image)
cropped = self._centre_crop_around_principal_point(
image, calibration, frame_id=frame.frame_id
)
target_h, target_w = self._input_shape
in_h, in_w = cropped.shape[:2]
interp = (
cv2.INTER_AREA
if (in_h > target_h or in_w > target_w)
else cv2.INTER_CUBIC
)
try:
resized = cv2.resize(
cropped, (target_w, target_h), interpolation=interp
)
except cv2.error as exc:
raise VprPreprocessError(
f"cv2.resize failed: {type(exc).__name__}: {exc}"
) from exc
as_f32 = resized.astype(np.float32) / 255.0
normalised = (as_f32 - self._mean) / self._std
chw = normalised.transpose(2, 0, 1)
return np.ascontiguousarray(chw[None, :, :, :], dtype=np.float16)
def input_shape(self) -> tuple[int, int]:
return self._input_shape
@staticmethod
def _coerce_to_rgb_uint8(image: object) -> np.ndarray:
if not isinstance(image, np.ndarray):
raise VprPreprocessError(
f"frame.image must be a numpy array; got {type(image).__name__}"
)
if image.dtype != np.uint8:
raise VprPreprocessError(
f"frame.image must be uint8 RGB; got dtype {image.dtype}"
)
if image.ndim == 2:
return np.stack([image, image, image], axis=-1)
if image.ndim == 3 and image.shape[2] == 3:
return image
raise VprPreprocessError(
f"frame.image must be (H,W) or (H,W,3); got shape {image.shape}"
)
def _centre_crop_around_principal_point(
self,
image: np.ndarray,
calibration: CameraCalibration | None,
*,
frame_id: int,
) -> np.ndarray:
"""Square-crop anchored on ``(cx, cy)`` from intrinsics_3x3.
Falls back to geometric centre + WARN log when calibration is
absent or its principal-point cannot be extracted.
"""
h, w = image.shape[:2]
side = min(h, w)
cx_cy = self._extract_principal_point(calibration)
if cx_cy is None:
self._logger.warning(
"UltraVPR calibration unusable; centre-cropping around "
"geometric centre",
extra={
"component": _COMPONENT,
"kind": _LOG_KIND_CALIBRATION_MISSING,
"kv": {"frame_id": int(frame_id)},
},
)
cx = w / 2.0
cy = h / 2.0
else:
cx, cy = cx_cy
half = side // 2
# Clamp so the crop window stays inside the image; this matches
# the upstream UltraVPR contract (the principal point can be
# near the edge in wide-angle cameras).
left = round(max(0.0, min(float(w - side), cx - half)))
top = round(max(0.0, min(float(h - side), cy - half)))
return image[top : top + side, left : left + side, :]
@staticmethod
def _extract_principal_point(
calibration: CameraCalibration | None,
) -> tuple[float, float] | None:
if calibration is None:
return None
intrinsics = getattr(calibration, "intrinsics_3x3", None)
if intrinsics is None:
return None
try:
arr = np.asarray(intrinsics, dtype=np.float64)
except (TypeError, ValueError):
return None
if arr.shape != (3, 3):
return None
cx = float(arr[0, 2])
cy = float(arr[1, 2])
# The identity matrix produces (cx, cy) == (0, 0) which is the
# top-left pixel; treat zeros as "not a real principal point"
# and fall back to geometric centre. (Test fixtures use
# ``np.eye(3)`` to mean "no calibration data".)
if cx == 0.0 and cy == 0.0:
return None
return cx, cy
src/gps_denied_onboard/components/c2_vpr/ultra_vpr.py
+462
View File
@@ -0,0 +1,462 @@
"""``UltraVprStrategy`` - C2 production-default VprStrategy (AZ-337).
UltraVPR is the Documentary Lead's PRIMARY backbone per
``components/02_c2_vpr/description.md`` § 1 and is wired by default when
``config.c2_vpr.strategy == "ultra_vpr"``. UltraVPR runs on the C7
TensorRT runtime (AZ-298) or the ONNX-Runtime fallback (AZ-299) -
explicitly NOT on the PyTorch FP16 runtime (which is reserved for the
NetVLAD baseline). This runtime isolation lets a TRT engine compile
bug fall back to NetVLAD without simultaneously breaking both.
The strategy delegates retrieval to :class:`FaissBridge` (AZ-341) and
the c6 ``DescriptorIndex`` cut (AZ-507) - see
:mod:`gps_denied_onboard.components.c2_vpr._faiss_bridge`. Embedding
goes through the c7 :class:`InferenceRuntime` Protocol via the local
:class:`InferenceRuntimeCut` (AZ-507).
Architecture-registry differences from :class:`NetVladStrategy`:
UltraVPR consumes a pre-compiled ``.trt`` engine produced by C10's
engine compiler (AZ-321) - there is no PyTorch ``nn.Module`` to
register. The strategy module therefore does NOT expose
``MODEL_NAME`` / ``architecture_factory``; the composition root's
:func:`gps_denied_onboard.runtime_root.vpr_factory.\
_register_strategy_architecture` helper no-ops for this strategy.
Engine load happens in :func:`create` (NOT at first frame) so the
engine-output-shape assertion (AC-6) surfaces at startup, not 17
minutes into a flight when the first VPR query hits.
Per-frame :meth:`embed_query` pipeline:
1. ``preprocessor.preprocess(frame, calibration)`` ->
``(1, 3, 384, 384)`` FP16 NCHW ndarray.
2. ``inference_runtime.infer(handle, {"input": tensor})`` ->
``{"embedding": (1, 512) FP16 ndarray}``.
3. ``normaliser.l2_normalise(raw[0])`` -> global L2 (UltraVPR is
single-stage; no intra-cluster step like NetVLAD).
4. Return :class:`VprQuery` with ``frame_id``, normalised embedding,
produced_at monotonic ns.
Error envelope: every method raises only members of :class:`VprError`.
``RuntimeError`` from the backbone forward -> rewrapped to
:class:`VprBackboneError`; :class:`VprPreprocessError` from the
preprocessor propagates unchanged. :class:`IndexUnavailableError`
from :class:`FaissBridge` (and through it from c6) is re-raised
unchanged (AC-10).
Retrieval is a single-line delegation to :class:`FaissBridge.retrieve`;
see AZ-341 AC-10.
"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Final, Literal
import numpy as np
from gps_denied_onboard._types.inference import (
BuildConfig,
EngineHandle,
PrecisionMode,
)
from gps_denied_onboard._types.vpr import VprQuery, VprResult
from gps_denied_onboard.clock import Clock
from gps_denied_onboard.components.c2_vpr._faiss_bridge import FaissBridge
from gps_denied_onboard.components.c2_vpr._preprocessor_ultra_vpr import (
UltraVprBackbonePreprocessor,
)
from gps_denied_onboard.components.c2_vpr.descriptor_index_cut import (
DescriptorIndexCut,
)
from gps_denied_onboard.components.c2_vpr.errors import (
VprBackboneError,
VprPreprocessError,
)
from gps_denied_onboard.components.c2_vpr.inference_runtime_cut import (
InferenceRuntimeCut,
)
from gps_denied_onboard.config.schema import ConfigError
from gps_denied_onboard.fdr_client import EnqueueResult, FdrClient
from gps_denied_onboard.fdr_client.records import (
CURRENT_SCHEMA_VERSION,
FdrRecord,
)
from gps_denied_onboard.helpers.descriptor_normaliser import DescriptorNormaliser
if TYPE_CHECKING:
from gps_denied_onboard._types.calibration import CameraCalibration
from gps_denied_onboard._types.nav import NavCameraFrame
from gps_denied_onboard.config.schema import Config
__all__ = ["DESCRIPTOR_DIM", "UltraVprStrategy", "create"]
# UltraVPR ships with a fixed published embedding dimension (D=512) per
# the upstream research code drop. Unlike NetVLAD (whose Linear PCA
# layer makes the output dimension a tunable knob), UltraVPR's
# embedding head is fused into the engine; making this a config-knob
# would let an operator silently break AC-2.1b. AC-5 / AC-6 / AC-7 of
# AZ-337 all assume 512.
DESCRIPTOR_DIM: Final[int] = 512
_BACKBONE_LABEL: Final[Literal["ultra_vpr"]] = "ultra_vpr"
_COMPONENT: Final[str] = "c2_vpr"
_OUTPUT_KEY: Final[str] = "embedding"
_ENGINE_INPUT_KEY: Final[str] = "input"
_ALLOWED_RUNTIME_LABELS: Final[frozenset[str]] = frozenset(
{"tensorrt", "onnx_trt_ep"}
)
_LOG_KIND_READY: Final[str] = "c2.vpr.ready"
_LOG_KIND_BACKBONE_ERROR: Final[str] = "c2.vpr.backbone_error"
_LOG_KIND_PREPROCESS_ERROR: Final[str] = "c2.vpr.preprocess_error"
_LOG_KIND_FDR_OVERRUN: Final[str] = "c2.vpr.fdr_overrun"
_FDR_KIND_EMBED: Final[str] = "vpr.embed_query"
_FDR_KIND_BACKBONE_ERROR: Final[str] = "vpr.backbone_error"
_FDR_KIND_PREPROCESS_ERROR: Final[str] = "vpr.preprocess_error"
class UltraVprStrategy:
"""C2 production-default VprStrategy backed by a TRT UltraVPR engine.
See module docstring for the engine-loading + per-frame pipeline.
Stateless across frames (INV-2); single-threaded per instance
(INV-1, per AZ-336).
"""
def __init__(
self,
*,
inference_runtime: InferenceRuntimeCut,
engine_handle: EngineHandle,
descriptor_index: DescriptorIndexCut,
preprocessor: UltraVprBackbonePreprocessor,
normaliser: DescriptorNormaliser,
faiss_bridge: FaissBridge,
fdr_client: FdrClient,
clock: Clock,
logger: logging.Logger,
descriptor_dim: int = DESCRIPTOR_DIM,
) -> None:
if descriptor_dim < 1:
raise ValueError(
f"UltraVprStrategy.descriptor_dim must be >= 1; "
f"got {descriptor_dim}"
)
self._inference_runtime = inference_runtime
self._engine_handle = engine_handle
self._descriptor_index = descriptor_index
self._preprocessor = preprocessor
self._normaliser = normaliser
self._faiss_bridge = faiss_bridge
self._fdr_client = fdr_client
self._clock = clock
self._logger = logger
self._descriptor_dim = descriptor_dim
def embed_query(
self,
frame: NavCameraFrame,
calibration: CameraCalibration,
) -> VprQuery:
try:
tensor = self._preprocessor.preprocess(frame, calibration)
except VprPreprocessError as exc:
self._emit_preprocess_error(frame, exc)
raise
ns_start = self._clock.monotonic_ns()
try:
outputs = self._inference_runtime.infer(
self._engine_handle, {_ENGINE_INPUT_KEY: tensor}
)
except Exception as exc:
wrapped = self._wrap_backbone_error(frame, exc)
raise wrapped from exc
ns_end = self._clock.monotonic_ns()
latency_us = max(1, (ns_end - ns_start) // 1_000)
if _OUTPUT_KEY not in outputs:
err = VprBackboneError(
f"UltraVPR forward returned no {_OUTPUT_KEY!r} key; "
f"got {sorted(outputs.keys())!r}"
)
self._emit_backbone_error(frame, err)
raise err
raw = np.asarray(outputs[_OUTPUT_KEY])
if (
raw.ndim != 2
or raw.shape[0] != 1
or raw.shape[1] != self._descriptor_dim
):
err = VprBackboneError(
f"UltraVPR forward returned shape {raw.shape}; "
f"expected (1, {self._descriptor_dim})"
)
self._emit_backbone_error(frame, err)
raise err
flat = np.ascontiguousarray(raw[0], dtype=np.float16)
normalised = self._normaliser.l2_normalise(flat)
self._emit_embed_record(
frame_id=int(frame.frame_id), latency_us=int(latency_us)
)
return VprQuery(
frame_id=int(frame.frame_id),
embedding=normalised,
produced_at=ns_end,
)
def retrieve_topk(self, query: VprQuery, k: int) -> VprResult:
return self._faiss_bridge.retrieve(
query, k, backbone_label=_BACKBONE_LABEL
)
def descriptor_dim(self) -> int:
return self._descriptor_dim
def _wrap_backbone_error(
self, frame: NavCameraFrame, exc: BaseException
) -> VprBackboneError:
wrapped = VprBackboneError(
f"UltraVPR forward raised {type(exc).__name__}: {exc}"
)
self._emit_backbone_error(frame, wrapped)
return wrapped
def _emit_embed_record(self, *, frame_id: int, latency_us: int) -> None:
record = FdrRecord(
schema_version=CURRENT_SCHEMA_VERSION,
ts=_iso_ts_from_clock(self._clock),
producer_id=self._fdr_client.producer_id,
kind=_FDR_KIND_EMBED,
payload={
"frame_id": frame_id,
"backbone_label": _BACKBONE_LABEL,
"descriptor_dim": self._descriptor_dim,
"latency_us": latency_us,
},
)
result = self._fdr_client.enqueue(record)
if result == EnqueueResult.OVERRUN:
self._logger.warning(
"FDR enqueue dropped vpr.embed_query record (buffer overrun)",
extra={
"component": _COMPONENT,
"kind": _LOG_KIND_FDR_OVERRUN,
"kv": {
"frame_id": frame_id,
"backbone_label": _BACKBONE_LABEL,
},
},
)
def _emit_backbone_error(
self, frame: NavCameraFrame, error: BaseException
) -> None:
frame_id = int(frame.frame_id)
msg = f"UltraVPR backbone error: {error}"
self._logger.error(
msg,
extra={
"component": _COMPONENT,
"kind": _LOG_KIND_BACKBONE_ERROR,
"kv": {
"frame_id": frame_id,
"backbone_label": _BACKBONE_LABEL,
"error_type": type(error).__name__,
},
},
)
self._fdr_client.enqueue(
FdrRecord(
schema_version=CURRENT_SCHEMA_VERSION,
ts=_iso_ts_from_clock(self._clock),
producer_id=self._fdr_client.producer_id,
kind=_FDR_KIND_BACKBONE_ERROR,
payload={
"frame_id": frame_id,
"backbone_label": _BACKBONE_LABEL,
"error_type": type(error).__name__,
"error_message": str(error)[:512],
},
)
)
def _emit_preprocess_error(
self, frame: NavCameraFrame, error: BaseException
) -> None:
frame_id = int(frame.frame_id)
msg = f"UltraVPR preprocess error: {error}"
self._logger.error(
msg,
extra={
"component": _COMPONENT,
"kind": _LOG_KIND_PREPROCESS_ERROR,
"kv": {
"frame_id": frame_id,
"backbone_label": _BACKBONE_LABEL,
"error_type": type(error).__name__,
},
},
)
self._fdr_client.enqueue(
FdrRecord(
schema_version=CURRENT_SCHEMA_VERSION,
ts=_iso_ts_from_clock(self._clock),
producer_id=self._fdr_client.producer_id,
kind=_FDR_KIND_PREPROCESS_ERROR,
payload={
"frame_id": frame_id,
"backbone_label": _BACKBONE_LABEL,
"error_type": type(error).__name__,
"error_message": str(error)[:512],
},
)
)
def _iso_ts_from_clock(clock: Clock) -> str:
# Same shape every component uses for FDR timestamps; AZ-508 will
# consolidate the duplicate helpers across c2/c11/c12/c6.
from datetime import datetime, timezone
ns = int(clock.time_ns())
seconds, fraction_ns = divmod(ns, 1_000_000_000)
dt = datetime.fromtimestamp(seconds, tz=timezone.utc)
return f"{dt.strftime('%Y-%m-%dT%H:%M:%S')}.{fraction_ns:09d}+00:00"
def _build_trt_build_config() -> BuildConfig:
return BuildConfig(
precision=PrecisionMode.FP16,
workspace_mb=0,
calibration_dataset=None,
optimization_profiles=(),
)
def create(
config: Config,
*,
descriptor_index: DescriptorIndexCut,
inference_runtime: InferenceRuntimeCut,
fdr_client: FdrClient | None = None,
clock: Clock | None = None,
logger: logging.Logger | None = None,
) -> UltraVprStrategy:
"""Module-level factory consumed by :func:`build_vpr_strategy`.
AC-11: UltraVPR is unselectable when the C7 TRT / ONNX-RT runtimes
are excluded - ``current_runtime_label()`` MUST be one of
``{"tensorrt", "onnx_trt_ep"}``; ``"pytorch_fp16"`` is rejected
with :class:`ConfigError` at composition time (NOT at first frame).
AC-6: engine output shape is asserted at create time via a single
dry-run inference on a zero-init input; mismatch raises
:class:`ConfigError` BEFORE the strategy is bound.
Optional keyword-only injection points (``fdr_client`` / ``clock`` /
``logger``) keep tests deterministic; production wiring fills them
from the composition root.
"""
runtime_label = inference_runtime.current_runtime_label()
if runtime_label not in _ALLOWED_RUNTIME_LABELS:
raise ConfigError(
f"UltraVPR requires BUILD_TENSORRT_RUNTIME=ON (or "
f"BUILD_ONNX_TRT_EP_RUNTIME=ON as fallback); this binary "
f"has runtime_label={runtime_label!r}. Per AZ-337 AC-11, "
f"UltraVPR is unselectable when the C7 TRT / ONNX-RT "
f"runtimes are excluded."
)
block = config.components["c2_vpr"]
weights_path = block.backbone_weights_path
if fdr_client is None:
raise ValueError(
"UltraVprStrategy.create: fdr_client is required; the "
"composition root must inject the running FDR client."
)
if clock is None:
from gps_denied_onboard.clock.wall_clock import WallClock
clock = WallClock()
if logger is None:
logger = logging.getLogger("gps_denied_onboard.c2_vpr.ultra_vpr")
entry = inference_runtime.compile_engine(
weights_path, _build_trt_build_config()
)
handle = inference_runtime.deserialize_engine(entry)
preprocessor = UltraVprBackbonePreprocessor(logger=logger)
normaliser = DescriptorNormaliser()
faiss_bridge = FaissBridge(
descriptor_index=descriptor_index,
descriptor_dim=DESCRIPTOR_DIM,
warn_top1_threshold=block.warn_top1_threshold,
debug_log_per_frame_distances=block.debug_per_frame_distances,
fdr_client=fdr_client,
logger=logger,
clock=clock,
)
_assert_engine_output_dim(inference_runtime, handle, preprocessor)
logger.info(
"C2 VPR strategy ready",
extra={
"component": _COMPONENT,
"kind": _LOG_KIND_READY,
"kv": {
"strategy": _BACKBONE_LABEL,
"descriptor_dim": DESCRIPTOR_DIM,
},
},
)
return UltraVprStrategy(
inference_runtime=inference_runtime,
engine_handle=handle,
descriptor_index=descriptor_index,
preprocessor=preprocessor,
normaliser=normaliser,
faiss_bridge=faiss_bridge,
fdr_client=fdr_client,
clock=clock,
logger=logger,
descriptor_dim=DESCRIPTOR_DIM,
)
def _assert_engine_output_dim(
inference_runtime: InferenceRuntimeCut,
handle: EngineHandle,
preprocessor: UltraVprBackbonePreprocessor,
) -> None:
h, w = preprocessor.input_shape()
probe = np.zeros((1, 3, h, w), dtype=np.float16)
outputs = inference_runtime.infer(handle, {_ENGINE_INPUT_KEY: probe})
if _OUTPUT_KEY not in outputs:
raise ConfigError(
f"engine output shape mismatch: {_OUTPUT_KEY!r} key absent; "
f"got keys {sorted(outputs.keys())!r}"
)
actual = np.asarray(outputs[_OUTPUT_KEY])
if (
actual.ndim != 2
or actual.shape[0] != 1
or actual.shape[1] != DESCRIPTOR_DIM
):
raise ConfigError(
f"engine output shape mismatch: expected (1, {DESCRIPTOR_DIM}), "
f"got {tuple(actual.shape)}"
)