From 235eb4549ef76de7ecf905269e7ea600fbec2480 Mon Sep 17 00:00:00 2001 From: Oleksandr Bezdieniezhnykh Date: Thu, 14 May 2026 00:50:17 +0300 Subject: [PATCH] [AZ-527] Consolidate _assert_engine_output_dim into c2-internal helper Closes cumulative review batches 49-51 Finding F1 (Medium / Maintainability) -- the 7-way duplication of _assert_engine_output_dim across c2_vpr secondary VPR strategy modules. Add c2-internal helper assert_engine_output_dim(inference_runtime, handle, preprocessor, descriptor_dim, *, output_key='embedding', input_key='input') in src/gps_denied_onboard/components/c2_vpr/ _engine_dim_assertion.py. The helper runs a zero-init dry-run inference at preprocessor.input_shape() and asserts the engine output dict carries (1, descriptor_dim) under output_key. Raises gps_denied_onboard.config.schema.ConfigError on mismatch (preserving the prior error envelope and message wording byte-identically). Migrate 7 strategy modules (ultra_vpr, net_vlad, mega_loc, mix_vpr, sela_vpr, eigen_places, salad) to import the helper and delete the local _assert_engine_output_dim definitions + their inline 'AZ-527 (planned)' comments. NetVLAD is the only call site that overrides output_key='vlad_descriptor'; the other 6 explicitly pass output_key=_OUTPUT_KEY + input_key=_ENGINE_INPUT_KEY (matching helper defaults but documenting strategy contract at the call site). Add tests/unit/c2_vpr/test_az527_engine_dim_assertion.py (14 tests, AAA pattern, Protocol-conforming fakes) covering AC-1..AC-4: helper signature; wrong shape raises ConfigError naming both dims; missing output key raises ConfigError naming the missing key; AST-walk regression guard for stray definitions outside the helper module (modeled on AZ-526's test_ac4_az526_no_module_level_iso_ts_from_clock_outside_helper); import-grep regression guard verifying all 7 strategy modules import the helper. AC-5 (existing AZ-337/338/339/340 AC-6 sub-tests pass unmodified) is exercised transitively: c2_vpr/ full directory 230/230 PASS, no test file modified outside the new test_az527_*. AC-6 (AZ-270 + AZ-507 layer lints) verified by tests/unit/test_az270_compose_root.py 8/8 PASS. Code-review verdict: PASS (zero findings). Ruff clean. Co-authored-by: Cursor --- .../reviews/batch_52_review.md | 91 +++++ .../c2_vpr/_engine_dim_assertion.py | 102 +++++ .../components/c2_vpr/eigen_places.py | 40 +- .../components/c2_vpr/mega_loc.py | 40 +- .../components/c2_vpr/mix_vpr.py | 40 +- .../components/c2_vpr/net_vlad.py | 31 +- .../components/c2_vpr/salad.py | 40 +- .../components/c2_vpr/sela_vpr.py | 40 +- .../components/c2_vpr/ultra_vpr.py | 35 +- .../c2_vpr/test_az527_engine_dim_assertion.py | 347 ++++++++++++++++++ 10 files changed, 615 insertions(+), 191 deletions(-) create mode 100644 _docs/03_implementation/reviews/batch_52_review.md create mode 100644 src/gps_denied_onboard/components/c2_vpr/_engine_dim_assertion.py create mode 100644 tests/unit/c2_vpr/test_az527_engine_dim_assertion.py diff --git a/_docs/03_implementation/reviews/batch_52_review.md b/_docs/03_implementation/reviews/batch_52_review.md new file mode 100644 index 0000000..f378789 --- /dev/null +++ b/_docs/03_implementation/reviews/batch_52_review.md @@ -0,0 +1,91 @@ +# Code Review Report — Batch 52 + +**Batch**: 52 (AZ-527 — Hygiene: consolidate `_assert_engine_output_dim` into a c2-internal helper) +**Date**: 2026-05-14 +**Verdict**: **PASS** + +## Findings + +_None._ + +## Phase Summary + +### Phase 1 — Context Loading + +Inputs read: + +- `_docs/02_tasks/todo/AZ-527_hygiene_engine_dim_assertion_consolidation.md` — task spec. +- `_docs/03_implementation/cumulative_review_batches_49-51_cycle1_report.md` § F1 — the finding being closed. +- `_docs/02_document/components/02_c2_vpr/description.md` § 6 — the c2_vpr internal-helper boundary. +- `_docs/02_document/architecture.md` + `_docs/02_document/module-layout.md` — for layer direction + Public API rules. +- The 7 modified strategy modules (`ultra_vpr.py`, `net_vlad.py`, `mega_loc.py`, `mix_vpr.py`, `sela_vpr.py`, `eigen_places.py`, `salad.py`). +- The new `src/gps_denied_onboard/components/c2_vpr/_engine_dim_assertion.py`. +- The new `tests/unit/c2_vpr/test_az527_engine_dim_assertion.py`. + +### Phase 2 — Spec Compliance + +| AC | Status | Evidence | +|----|--------|----------| +| AC-1 (helper exists with expected signature) | PASS | `test_ac1_helper_callable_with_default_keys`. The signature is `(inference_runtime, handle, preprocessor, descriptor_dim, *, output_key="embedding", input_key="input") -> None`. | +| AC-2 (wrong shape raises `ConfigError` with both dims named) | PASS | `test_ac2_wrong_descriptor_width_raises_config_error` (3 parametrised) + `test_ac2_wrong_ndim_or_batch_raises_config_error` (4 parametrised). Error envelope is `gps_denied_onboard.config.schema.ConfigError`. | +| AC-3 (missing output key raises `ConfigError` naming the missing key) | PASS | `test_ac3_missing_default_output_key_raises_config_error` + `test_ac3_missing_overridden_output_key_raises_config_error`. Both default `"embedding"` and NetVLAD override `"vlad_descriptor"` paths covered. | +| AC-4 (no stray local definitions remain) | PASS | `test_ac4_no_stray_engine_dim_assertion_definitions_outside_helper` (AST walk, modeled on AZ-526's `test_ac4_az526_no_module_level_iso_ts_from_clock_outside_helper`) + `test_ac4_seven_strategy_modules_import_the_helper` (import grep on each of the 7 strategy modules). | +| AC-5 (AZ-337/338/339/340 AC-6 tests pass unmodified) | PASS | Full c2_vpr test directory: **230 / 230 PASS** (no test file modified outside the new `test_az527_*`). | +| AC-6 (AZ-270 + AZ-507 layer lints pass) | PASS | `tests/unit/test_az270_compose_root.py` + `tests/unit/test_az508_iso_timestamps.py` (combined): **26 / 26 PASS**. Helper is `_engine_dim_assertion.py` (underscore = c2-internal), not in the strategy resolution table. | + +Constraints (from § Constraints): all satisfied. + +- Helper module is c2-internal (`_engine_dim_assertion.py` underscore prefix; not exported via `c2_vpr/__init__.py`). +- `descriptor_dim` / `output_key` / `input_key` are parameters; helper has zero per-backbone module constants. +- Error envelope preserved: `gps_denied_onboard.config.schema.ConfigError` (NOT `ValueError`, NOT a custom new error type). +- Probe input shape comes from `preprocessor.input_shape()` (covered by `test_helper_propagates_preprocessor_input_shape_to_probe_tensor`). +- Test additions go into new `test_az527_engine_dim_assertion.py`; the AZ-337/338/339/340 test files are untouched. + +### Phase 3 — Code Quality + +- **SRP**: helper has a single concern — verify the loaded engine emits `(1, descriptor_dim)` under `output_key`. Nothing else. +- **Error handling**: raises typed `ConfigError` with full context (expected + actual dims; missing key name); no bare `except`; no silent suppression. +- **Naming**: `assert_engine_output_dim`, `descriptor_dim`, `output_key`, `input_key` — all precise, no vague terms. +- **DRY**: 7 copies → 1 source of truth (the entire point of this PBI). +- **Test quality**: Arrange / Act / Assert pattern; parametrised over the realistic dim values (512 / 2048 / 8448) + degenerate shapes (`(512,)`, `(2,1,512)`, `(1,1,512)`, `(1,512,1)`); regression guards via AST walk + import grep. +- **Dead code**: 7 local `_assert_engine_output_dim` definitions + their inline `AZ-527 (planned)` comments deleted. `EngineHandle` and `InferenceRuntimeCut` imports remain referenced elsewhere in each strategy (verified by ripgrep). +- **Comment hygiene**: helper docstring describes intent + parameters + raises; no narrating comments. Test file uses minimal section headers per the existing test_az340_* / test_az526_* precedent. +- **Annotation form**: `from __future__ import annotations` + bare type names (no quoted strings) — Ruff `UP037` clean. + +### Phase 4 — Security Quick-Scan + +No SQL, no `subprocess`, no `eval` / `exec`, no secrets. The helper is pure-Python numerical wiring (allocate a zero-init `(1,3,h,w)` ndarray, call `infer()`, assert dict shape). n/a. + +### Phase 5 — Performance Scan + +- The helper allocates one zero-init `(1, 3, H, W)` `np.float16` ndarray and runs one `infer()` call. Both happen at `create()` time (startup), NOT per frame. +- No hot-path impact. The migration replaced 7 byte-identical helpers with 1 — wall-clock startup time is unchanged within measurement noise. + +### Phase 6 — Cross-Task Consistency + +- All 7 strategy modules consistently route through the helper. The 6 default-key backbones (UltraVPR / MegaLoc / MixVPR / SelaVPR / EigenPlaces / SALAD) explicitly pass `output_key=_OUTPUT_KEY` + `input_key=_ENGINE_INPUT_KEY` (matching the helper defaults but making the strategy contract explicit at the call site). NetVLAD overrides `output_key="vlad_descriptor"` only. This is the only intentional behavioural difference between the strategies' use of the helper. +- The helper signature uses keyword-only args for `output_key` / `input_key` (the `*` separator after `descriptor_dim`). This forbids positional misuse — a future strategy author cannot accidentally swap them. + +### Phase 7 — Architecture Compliance + +1. **Layer direction**: `_engine_dim_assertion.py` (c2_vpr, Layer 3) imports `ConfigError` from `gps_denied_onboard.config.schema` (Layer 1) and `numpy` (stdlib-equivalent). Type-only imports under `TYPE_CHECKING` are `EngineHandle` (`_types.inference`, Layer 1), `BackbonePreprocessor` (`c2_vpr/_preprocessor`, Layer 3 — same component), `InferenceRuntimeCut` (`c2_vpr/inference_runtime_cut`, Layer 3 — same component). All imports are downward or intra-component. No upward imports. +2. **Public API respect**: the helper is c2-internal — underscore-prefixed module name (`_engine_dim_assertion.py`) and NOT re-exported from `c2_vpr/__init__.py`. The 7 strategy modules import it as a sibling internal module, not via a Public API. No cross-component import added. +3. **No new cyclic deps**: built-import graph of changed files + direct dependencies. The helper has zero c2_vpr → c2_vpr cycle risk because it imports its dependencies under `TYPE_CHECKING` only (no runtime back-edge to the strategy modules). Strategy modules import the helper as a leaf — no cycle. +4. **Duplicate symbols**: 7-way `_assert_engine_output_dim` duplication is now zero-way duplicated. AC-4 regression guard prevents recurrence. +5. **Cross-cutting concerns not locally re-implemented**: the helper stays inside `c2_vpr/`, not in `shared/helpers/`. This is correct per AZ-527 § Out of scope: the engine output-shape contract is a c2 internal concern (C7 owns its own engine-shape assertions inside the runtime). The reasoning is explicit in `_engine_dim_assertion.py`'s module docstring. + +### Phase 8 — Test infrastructure + +- `tests/unit/c2_vpr/test_az527_engine_dim_assertion.py` — 14 tests, all PASS. +- `tests/unit/c2_vpr/` (full directory) — 230 / 230 PASS (UltraVPR / NetVLAD / MegaLoc / MixVPR / SelaVPR / EigenPlaces / SALAD AC-6 sub-tests all green; FaissBridge / protocol_conformance / new AZ-527 file all green). +- `tests/unit/test_az270_compose_root.py` (composition-root layer lint) — 8 / 8 PASS. +- `tests/unit/test_az508_iso_timestamps.py` (helpers AST-walk regression guards from AZ-526 — included to verify the ASYL pattern still holds in the helpers/ namespace; doesn't directly involve AZ-527 but sanity-checks neighbouring lint infrastructure) — 18 / 18 PASS. + +## Verdict + +- Critical findings: 0 +- High findings: 0 +- Medium findings: 0 +- Low findings: 0 + +**PASS** — clean batch, no findings, all ACs covered, all tests green, full Ruff + lint clean. diff --git a/src/gps_denied_onboard/components/c2_vpr/_engine_dim_assertion.py b/src/gps_denied_onboard/components/c2_vpr/_engine_dim_assertion.py new file mode 100644 index 0000000..75113e1 --- /dev/null +++ b/src/gps_denied_onboard/components/c2_vpr/_engine_dim_assertion.py @@ -0,0 +1,102 @@ +"""C2-internal engine output-dim assertion helper (AZ-527). + +Single home for the dry-run probe that every c2_vpr secondary VPR +strategy module runs at :func:`create` time to verify that the loaded +inference engine emits a descriptor of the shape the strategy +contract promises. + +This helper consolidates the formerly seven-way duplicated +``_assert_engine_output_dim`` helper across ``ultra_vpr.py``, +``net_vlad.py``, ``mega_loc.py``, ``mix_vpr.py``, ``sela_vpr.py``, +``eigen_places.py`` and ``salad.py``. Behaviour, error envelope, and +error-message wording are byte-identical to the prior copies so that +the AZ-337 / 338 / 339 / 340 AC-6 sub-tests continue to pass +unmodified. + +The helper is C2-internal: the underscore-prefixed module name and +the unimported-from-``c2_vpr.__init__`` placement keep it inside the +component boundary per +``components/02_c2_vpr/description.md`` § 6 (engine output-shape +contracts are a C2 internal concern; C7 owns its own engine-shape +assertions inside the runtime). +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import numpy as np + +from gps_denied_onboard.config.schema import ConfigError + +if TYPE_CHECKING: + from gps_denied_onboard._types.inference import EngineHandle + from gps_denied_onboard.components.c2_vpr._preprocessor import ( + BackbonePreprocessor, + ) + from gps_denied_onboard.components.c2_vpr.inference_runtime_cut import ( + InferenceRuntimeCut, + ) + +__all__ = ["assert_engine_output_dim"] + + +def assert_engine_output_dim( + inference_runtime: InferenceRuntimeCut, + handle: EngineHandle, + preprocessor: BackbonePreprocessor, + descriptor_dim: int, + *, + output_key: str = "embedding", + input_key: str = "input", +) -> None: + """Verify the engine emits ``(1, descriptor_dim)`` under ``output_key``. + + Runs a single zero-init dry-run inference at the preprocessor's + declared ``input_shape()`` (FP16 NCHW, batch=1, RGB) and asserts + two contracts on the returned output dict: + + 1. ``output_key`` is present in the dict (default ``"embedding"``, + NetVLAD overrides to ``"vlad_descriptor"``). + 2. The output ndarray has shape ``(1, descriptor_dim)``. + + Both contracts are verified at ``create()`` time (startup), not + per frame. + + :param inference_runtime: the C2-internal :class:`InferenceRuntimeCut` + that wraps the loaded engine. + :param handle: the deserialised :class:`EngineHandle` to probe. + :param preprocessor: the strategy's :class:`BackbonePreprocessor`; + only ``input_shape()`` is consulted to size the probe tensor. + :param descriptor_dim: the strategy's expected descriptor width. + For most backbones this is a module-level ``DESCRIPTOR_DIM`` + constant; NetVLAD passes its runtime-resolved descriptor_dim. + :param output_key: the key under which the engine surfaces the + descriptor. Defaults to ``"embedding"`` (used by 6 of the 7 + c2_vpr secondary backbones); NetVLAD passes ``"vlad_descriptor"``. + :param input_key: the key under which the engine accepts the input + tensor. Defaults to ``"input"`` (used by all 7 backbones). + + :raises ConfigError: if ``output_key`` is absent from the engine + output, or the output ndarray shape does not match + ``(1, descriptor_dim)``. The error message names both the + expected and actual values. + """ + h, w = preprocessor.input_shape() + probe = np.zeros((1, 3, h, w), dtype=np.float16) + outputs = inference_runtime.infer(handle, {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)}" + ) diff --git a/src/gps_denied_onboard/components/c2_vpr/eigen_places.py b/src/gps_denied_onboard/components/c2_vpr/eigen_places.py index 7cd51fc..791ec8d 100644 --- a/src/gps_denied_onboard/components/c2_vpr/eigen_places.py +++ b/src/gps_denied_onboard/components/c2_vpr/eigen_places.py @@ -60,6 +60,9 @@ from gps_denied_onboard._types.inference import ( ) from gps_denied_onboard._types.vpr import VprQuery, VprResult from gps_denied_onboard.clock import Clock +from gps_denied_onboard.components.c2_vpr._engine_dim_assertion import ( + assert_engine_output_dim, +) from gps_denied_onboard.components.c2_vpr._faiss_bridge import FaissBridge from gps_denied_onboard.components.c2_vpr._preprocessor_eigen_places import ( EigenPlacesBackbonePreprocessor, @@ -394,7 +397,14 @@ def create( clock=clock, ) - _assert_engine_output_dim(inference_runtime, handle, preprocessor) + assert_engine_output_dim( + inference_runtime, + handle, + preprocessor, + DESCRIPTOR_DIM, + output_key=_OUTPUT_KEY, + input_key=_ENGINE_INPUT_KEY, + ) logger.info( "C2 VPR strategy ready", @@ -422,31 +432,3 @@ def create( ) -def _assert_engine_output_dim( - inference_runtime: InferenceRuntimeCut, - handle: EngineHandle, - preprocessor: EigenPlacesBackbonePreprocessor, -) -> None: - # The 7-way duplication of this helper (ultra_vpr / net_vlad / - # mega_loc / mix_vpr / sela_vpr / eigen_places / salad) is tracked - # by AZ-527 (hygiene PBI sized in parallel with AZ-339 land). The - # duplication is intentional for now: extracting earlier would - # expand AZ-340's scope past the three new strategies. - 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)}" - ) diff --git a/src/gps_denied_onboard/components/c2_vpr/mega_loc.py b/src/gps_denied_onboard/components/c2_vpr/mega_loc.py index 7678557..a2738db 100644 --- a/src/gps_denied_onboard/components/c2_vpr/mega_loc.py +++ b/src/gps_denied_onboard/components/c2_vpr/mega_loc.py @@ -59,6 +59,9 @@ from gps_denied_onboard._types.inference import ( ) from gps_denied_onboard._types.vpr import VprQuery, VprResult from gps_denied_onboard.clock import Clock +from gps_denied_onboard.components.c2_vpr._engine_dim_assertion import ( + assert_engine_output_dim, +) from gps_denied_onboard.components.c2_vpr._faiss_bridge import FaissBridge from gps_denied_onboard.components.c2_vpr._preprocessor_mega_loc import ( MegaLocBackbonePreprocessor, @@ -393,7 +396,14 @@ def create( clock=clock, ) - _assert_engine_output_dim(inference_runtime, handle, preprocessor) + assert_engine_output_dim( + inference_runtime, + handle, + preprocessor, + DESCRIPTOR_DIM, + output_key=_OUTPUT_KEY, + input_key=_ENGINE_INPUT_KEY, + ) logger.info( "C2 VPR strategy ready", @@ -421,31 +431,3 @@ def create( ) -def _assert_engine_output_dim( - inference_runtime: InferenceRuntimeCut, - handle: EngineHandle, - preprocessor: MegaLocBackbonePreprocessor, -) -> None: - # The 4-way duplication of this helper (ultra_vpr / net_vlad / - # mega_loc / mix_vpr) will be consolidated by AZ-527 (hygiene - # PBI sized in parallel with AZ-339 land). The duplication is - # intentional for now: extracting earlier would expand AZ-339's - # scope past the two new strategies. - 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)}" - ) diff --git a/src/gps_denied_onboard/components/c2_vpr/mix_vpr.py b/src/gps_denied_onboard/components/c2_vpr/mix_vpr.py index ece1185..113e2ef 100644 --- a/src/gps_denied_onboard/components/c2_vpr/mix_vpr.py +++ b/src/gps_denied_onboard/components/c2_vpr/mix_vpr.py @@ -60,6 +60,9 @@ from gps_denied_onboard._types.inference import ( ) from gps_denied_onboard._types.vpr import VprQuery, VprResult from gps_denied_onboard.clock import Clock +from gps_denied_onboard.components.c2_vpr._engine_dim_assertion import ( + assert_engine_output_dim, +) from gps_denied_onboard.components.c2_vpr._faiss_bridge import FaissBridge from gps_denied_onboard.components.c2_vpr._preprocessor_mix_vpr import ( MixVprBackbonePreprocessor, @@ -396,7 +399,14 @@ def create( clock=clock, ) - _assert_engine_output_dim(inference_runtime, handle, preprocessor) + assert_engine_output_dim( + inference_runtime, + handle, + preprocessor, + DESCRIPTOR_DIM, + output_key=_OUTPUT_KEY, + input_key=_ENGINE_INPUT_KEY, + ) logger.info( "C2 VPR strategy ready", @@ -424,31 +434,3 @@ def create( ) -def _assert_engine_output_dim( - inference_runtime: InferenceRuntimeCut, - handle: EngineHandle, - preprocessor: MixVprBackbonePreprocessor, -) -> None: - # The 4-way duplication of this helper (ultra_vpr / net_vlad / - # mega_loc / mix_vpr) will be consolidated by AZ-527 (hygiene - # PBI sized in parallel with AZ-339 land). The duplication is - # intentional for now: extracting earlier would expand AZ-339's - # scope past the two new strategies. - 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)}" - ) diff --git a/src/gps_denied_onboard/components/c2_vpr/net_vlad.py b/src/gps_denied_onboard/components/c2_vpr/net_vlad.py index 0dda9f1..7f86130 100644 --- a/src/gps_denied_onboard/components/c2_vpr/net_vlad.py +++ b/src/gps_denied_onboard/components/c2_vpr/net_vlad.py @@ -80,6 +80,9 @@ from gps_denied_onboard._types.inference import ( ) from gps_denied_onboard._types.vpr import VprQuery, VprResult from gps_denied_onboard.clock import Clock +from gps_denied_onboard.components.c2_vpr._engine_dim_assertion import ( + assert_engine_output_dim, +) from gps_denied_onboard.components.c2_vpr._faiss_bridge import FaissBridge from gps_denied_onboard.components.c2_vpr._net_vlad_architecture import ( DEFAULT_NUM_CLUSTERS, @@ -461,8 +464,12 @@ def create( clock=clock, ) - _assert_engine_output_dim( - inference_runtime, handle, descriptor_dim, preprocessor + assert_engine_output_dim( + inference_runtime, + handle, + preprocessor, + descriptor_dim, + output_key="vlad_descriptor", ) logger.info( @@ -491,23 +498,3 @@ def create( ) -def _assert_engine_output_dim( - inference_runtime: InferenceRuntimeCut, - handle: EngineHandle, - expected_dim: int, - preprocessor: NetVladBackbonePreprocessor, -) -> None: - h, w = preprocessor.input_shape() - probe = np.zeros((1, 3, h, w), dtype=np.float16) - outputs = inference_runtime.infer(handle, {"input": probe}) - if "vlad_descriptor" not in outputs: - raise ConfigError( - f"engine output shape mismatch: 'vlad_descriptor' key absent; " - f"got keys {sorted(outputs.keys())!r}" - ) - actual = np.asarray(outputs["vlad_descriptor"]) - if actual.ndim != 2 or actual.shape[0] != 1 or actual.shape[1] != expected_dim: - raise ConfigError( - f"engine output shape mismatch: expected (1, {expected_dim}), " - f"got {tuple(actual.shape)}" - ) diff --git a/src/gps_denied_onboard/components/c2_vpr/salad.py b/src/gps_denied_onboard/components/c2_vpr/salad.py index 4855a03..df3fad8 100644 --- a/src/gps_denied_onboard/components/c2_vpr/salad.py +++ b/src/gps_denied_onboard/components/c2_vpr/salad.py @@ -69,6 +69,9 @@ from gps_denied_onboard._types.inference import ( ) from gps_denied_onboard._types.vpr import VprQuery, VprResult from gps_denied_onboard.clock import Clock +from gps_denied_onboard.components.c2_vpr._engine_dim_assertion import ( + assert_engine_output_dim, +) from gps_denied_onboard.components.c2_vpr._faiss_bridge import FaissBridge from gps_denied_onboard.components.c2_vpr._preprocessor_salad import ( SaladBackbonePreprocessor, @@ -406,7 +409,14 @@ def create( clock=clock, ) - _assert_engine_output_dim(inference_runtime, handle, preprocessor) + assert_engine_output_dim( + inference_runtime, + handle, + preprocessor, + DESCRIPTOR_DIM, + output_key=_OUTPUT_KEY, + input_key=_ENGINE_INPUT_KEY, + ) logger.info( "C2 VPR strategy ready", @@ -434,31 +444,3 @@ def create( ) -def _assert_engine_output_dim( - inference_runtime: InferenceRuntimeCut, - handle: EngineHandle, - preprocessor: SaladBackbonePreprocessor, -) -> None: - # The 7-way duplication of this helper (ultra_vpr / net_vlad / - # mega_loc / mix_vpr / sela_vpr / eigen_places / salad) is tracked - # by AZ-527 (hygiene PBI sized in parallel with AZ-339 land). The - # duplication is intentional for now: extracting earlier would - # expand AZ-340's scope past the three new strategies. - 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)}" - ) diff --git a/src/gps_denied_onboard/components/c2_vpr/sela_vpr.py b/src/gps_denied_onboard/components/c2_vpr/sela_vpr.py index 7f0464f..146a6f4 100644 --- a/src/gps_denied_onboard/components/c2_vpr/sela_vpr.py +++ b/src/gps_denied_onboard/components/c2_vpr/sela_vpr.py @@ -59,6 +59,9 @@ from gps_denied_onboard._types.inference import ( ) from gps_denied_onboard._types.vpr import VprQuery, VprResult from gps_denied_onboard.clock import Clock +from gps_denied_onboard.components.c2_vpr._engine_dim_assertion import ( + assert_engine_output_dim, +) from gps_denied_onboard.components.c2_vpr._faiss_bridge import FaissBridge from gps_denied_onboard.components.c2_vpr._preprocessor_sela_vpr import ( SelaVprBackbonePreprocessor, @@ -393,7 +396,14 @@ def create( clock=clock, ) - _assert_engine_output_dim(inference_runtime, handle, preprocessor) + assert_engine_output_dim( + inference_runtime, + handle, + preprocessor, + DESCRIPTOR_DIM, + output_key=_OUTPUT_KEY, + input_key=_ENGINE_INPUT_KEY, + ) logger.info( "C2 VPR strategy ready", @@ -421,31 +431,3 @@ def create( ) -def _assert_engine_output_dim( - inference_runtime: InferenceRuntimeCut, - handle: EngineHandle, - preprocessor: SelaVprBackbonePreprocessor, -) -> None: - # The 7-way duplication of this helper (ultra_vpr / net_vlad / - # mega_loc / mix_vpr / sela_vpr / eigen_places / salad) is tracked - # by AZ-527 (hygiene PBI sized in parallel with AZ-339 land). The - # duplication is intentional for now: extracting earlier would - # expand AZ-340's scope past the three new strategies. - 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)}" - ) diff --git a/src/gps_denied_onboard/components/c2_vpr/ultra_vpr.py b/src/gps_denied_onboard/components/c2_vpr/ultra_vpr.py index 321d627..c62b3be 100644 --- a/src/gps_denied_onboard/components/c2_vpr/ultra_vpr.py +++ b/src/gps_denied_onboard/components/c2_vpr/ultra_vpr.py @@ -63,6 +63,9 @@ from gps_denied_onboard._types.inference import ( ) from gps_denied_onboard._types.vpr import VprQuery, VprResult from gps_denied_onboard.clock import Clock +from gps_denied_onboard.components.c2_vpr._engine_dim_assertion import ( + assert_engine_output_dim, +) from gps_denied_onboard.components.c2_vpr._faiss_bridge import FaissBridge from gps_denied_onboard.components.c2_vpr._preprocessor_ultra_vpr import ( UltraVprBackbonePreprocessor, @@ -401,7 +404,14 @@ def create( clock=clock, ) - _assert_engine_output_dim(inference_runtime, handle, preprocessor) + assert_engine_output_dim( + inference_runtime, + handle, + preprocessor, + DESCRIPTOR_DIM, + output_key=_OUTPUT_KEY, + input_key=_ENGINE_INPUT_KEY, + ) logger.info( "C2 VPR strategy ready", @@ -429,26 +439,3 @@ def create( ) -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)}" - ) diff --git a/tests/unit/c2_vpr/test_az527_engine_dim_assertion.py b/tests/unit/c2_vpr/test_az527_engine_dim_assertion.py new file mode 100644 index 0000000..1c760b7 --- /dev/null +++ b/tests/unit/c2_vpr/test_az527_engine_dim_assertion.py @@ -0,0 +1,347 @@ +"""AZ-527 — c2_vpr engine output-dim assertion helper unit tests. + +Covers AC-1..AC-6 of AZ-527. AC-5 is exercised transitively by the +existing AZ-337 / 338 / 339 / 340 AC-6 sub-tests (which run the +strategy ``create()`` end-to-end and now route through +``assert_engine_output_dim`` after this batch). AC-6 (layer +compliance) is exercised by ``test_az270_compose_root.py`` — this +file additionally adds a focused regression guard (AC-4) that walks +``src/`` for stray ``_assert_engine_output_dim`` / +``assert_engine_output_dim`` definitions outside the helper module. + +The unit tests below use minimal fakes (Protocol-conforming) so the +helper can be exercised without spinning up a full strategy +``create()`` chain. +""" + +from __future__ import annotations + +import ast +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Literal + +import numpy as np +import pytest + +from gps_denied_onboard._types.inference import ( + BuildConfig, + EngineCacheEntry, + EngineHandle, +) +from gps_denied_onboard.components.c2_vpr._engine_dim_assertion import ( + assert_engine_output_dim, +) +from gps_denied_onboard.config.schema import ConfigError + +_REPO_ROOT = Path(__file__).resolve().parents[3] +_HELPER_PATH = ( + _REPO_ROOT + / "src" + / "gps_denied_onboard" + / "components" + / "c2_vpr" + / "_engine_dim_assertion.py" +) + + +# --------------------------------------------------------------------------- +# Fakes +# --------------------------------------------------------------------------- + + +@dataclass +class _FakePreprocessor: + """Minimal :class:`BackbonePreprocessor` for the helper's input_shape() probe.""" + + hw: tuple[int, int] = (224, 224) + + def preprocess( + self, frame: Any, calibration: Any + ) -> np.ndarray: # pragma: no cover - unused by the helper + raise AssertionError("helper must not call preprocess()") + + def input_shape(self) -> tuple[int, int]: + return self.hw + + +class _FakeRuntime: + """Minimal :class:`InferenceRuntimeCut` returning a fixed output dict.""" + + def __init__( + self, + *, + output: dict[str, np.ndarray], + expected_input_key: str = "input", + expected_input_hw: tuple[int, int] = (224, 224), + ) -> None: + self._output = output + self._expected_input_key = expected_input_key + self._expected_input_hw = expected_input_hw + self.last_input_key: str | None = None + self.last_input_shape: tuple[int, ...] | None = None + self.last_input_dtype: np.dtype | None = None + self.infer_call_count: int = 0 + + def compile_engine( + self, model_path: Path, build_config: BuildConfig + ) -> EngineCacheEntry: # pragma: no cover - unused + raise AssertionError("helper must not call compile_engine()") + + def deserialize_engine( + self, entry: EngineCacheEntry + ) -> EngineHandle: # pragma: no cover - unused + raise AssertionError("helper must not call deserialize_engine()") + + def infer( + self, + handle: EngineHandle, + inputs: dict[str, np.ndarray], + ) -> dict[str, np.ndarray]: + self.infer_call_count += 1 + assert len(inputs) == 1, f"expected one input key, got {sorted(inputs)!r}" + (key,) = inputs.keys() + self.last_input_key = key + tensor = inputs[key] + self.last_input_shape = tensor.shape + self.last_input_dtype = tensor.dtype + return self._output + + def release_engine(self, handle: EngineHandle) -> None: # pragma: no cover + raise AssertionError("helper must not call release_engine()") + + def current_runtime_label( + self, + ) -> Literal["tensorrt", "onnx_trt_ep", "pytorch_fp16"]: # pragma: no cover + raise AssertionError("helper must not call current_runtime_label()") + + +_DUMMY_HANDLE: EngineHandle = "dummy-handle" # type: ignore[assignment] + + +# --------------------------------------------------------------------------- +# AC-1: Helper exists at the canonical path with the expected signature +# --------------------------------------------------------------------------- + + +def test_ac1_helper_callable_with_default_keys() -> None: + # Arrange + runtime = _FakeRuntime( + output={"embedding": np.zeros((1, 512), dtype=np.float16)} + ) + preprocessor = _FakePreprocessor(hw=(224, 224)) + + # Act + result = assert_engine_output_dim( + runtime, _DUMMY_HANDLE, preprocessor, descriptor_dim=512 + ) + + # Assert + assert result is None + assert runtime.infer_call_count == 1 + assert runtime.last_input_key == "input" + assert runtime.last_input_shape == (1, 3, 224, 224) + assert runtime.last_input_dtype == np.dtype(np.float16) + + +# --------------------------------------------------------------------------- +# AC-2: Wrong shape raises ConfigError naming both expected and actual dims +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "actual_shape,expected_dim", + [ + ((1, 999), 512), + ((1, 4096), 2048), + ((1, 100), 8448), + ], +) +def test_ac2_wrong_descriptor_width_raises_config_error( + actual_shape: tuple[int, int], expected_dim: int +) -> None: + # Arrange + runtime = _FakeRuntime( + output={"embedding": np.zeros(actual_shape, dtype=np.float16)} + ) + preprocessor = _FakePreprocessor(hw=(224, 224)) + + # Act + Assert + with pytest.raises(ConfigError) as exc_info: + assert_engine_output_dim( + runtime, _DUMMY_HANDLE, preprocessor, descriptor_dim=expected_dim + ) + msg = str(exc_info.value) + assert f"expected (1, {expected_dim})" in msg + assert f"got {actual_shape!r}".replace("'", "") in msg or str(actual_shape) in msg + + +@pytest.mark.parametrize( + "actual_shape", [(512,), (2, 1, 512), (1, 1, 512), (1, 512, 1)] +) +def test_ac2_wrong_ndim_or_batch_raises_config_error( + actual_shape: tuple[int, ...], +) -> None: + # Arrange + runtime = _FakeRuntime( + output={"embedding": np.zeros(actual_shape, dtype=np.float16)} + ) + preprocessor = _FakePreprocessor() + + # Act + Assert + with pytest.raises(ConfigError, match=r"engine output shape mismatch"): + assert_engine_output_dim( + runtime, _DUMMY_HANDLE, preprocessor, descriptor_dim=512 + ) + + +# --------------------------------------------------------------------------- +# AC-3: Missing output key raises ConfigError naming the missing key +# --------------------------------------------------------------------------- + + +def test_ac3_missing_default_output_key_raises_config_error() -> None: + # Arrange + runtime = _FakeRuntime( + output={"wrong_key": np.zeros((1, 512), dtype=np.float16)} + ) + preprocessor = _FakePreprocessor() + + # Act + Assert + with pytest.raises(ConfigError) as exc_info: + assert_engine_output_dim( + runtime, _DUMMY_HANDLE, preprocessor, descriptor_dim=512 + ) + msg = str(exc_info.value) + assert "'embedding' key absent" in msg + assert "wrong_key" in msg + + +def test_ac3_missing_overridden_output_key_raises_config_error() -> None: + # Arrange — NetVLAD-style override + runtime = _FakeRuntime( + output={"embedding": np.zeros((1, 4096), dtype=np.float16)} + ) + preprocessor = _FakePreprocessor(hw=(480, 640)) + + # Act + Assert + with pytest.raises(ConfigError, match=r"'vlad_descriptor' key absent"): + assert_engine_output_dim( + runtime, + _DUMMY_HANDLE, + preprocessor, + descriptor_dim=4096, + output_key="vlad_descriptor", + ) + + +# --------------------------------------------------------------------------- +# AC: Non-default keys (covers NetVLAD's overrides + future strategies) +# --------------------------------------------------------------------------- + + +def test_helper_accepts_non_default_output_and_input_keys() -> None: + # Arrange — NetVLAD-style: vlad_descriptor + custom input key + runtime = _FakeRuntime( + output={"vlad_descriptor": np.zeros((1, 4096), dtype=np.float16)} + ) + preprocessor = _FakePreprocessor(hw=(480, 640)) + + # Act + assert_engine_output_dim( + runtime, + _DUMMY_HANDLE, + preprocessor, + descriptor_dim=4096, + output_key="vlad_descriptor", + input_key="image", + ) + + # Assert + assert runtime.last_input_key == "image" + assert runtime.last_input_shape == (1, 3, 480, 640) + + +def test_helper_propagates_preprocessor_input_shape_to_probe_tensor() -> None: + # Arrange — SALAD-style 322x322 + non-square shape coverage + for hw in [(224, 224), (322, 322), (480, 480), (480, 640)]: + runtime = _FakeRuntime( + output={"embedding": np.zeros((1, 8448), dtype=np.float16)} + ) + preprocessor = _FakePreprocessor(hw=hw) + + # Act + assert_engine_output_dim( + runtime, _DUMMY_HANDLE, preprocessor, descriptor_dim=8448 + ) + + # Assert + assert runtime.last_input_shape == (1, 3, hw[0], hw[1]), hw + + +# --------------------------------------------------------------------------- +# AC-4: Regression guard — no stray definitions outside the helper module +# --------------------------------------------------------------------------- + + +def test_ac4_no_stray_engine_dim_assertion_definitions_outside_helper() -> None: + """AC-4 (AZ-527): a `def assert_engine_output_dim` / + `def _assert_engine_output_dim` MUST exist only inside + `_engine_dim_assertion.py`. Any other definition under `src/` + means a c2_vpr strategy author slipped a copy back in. + + Modeled on the AZ-508 / AZ-526 `test_no_local_iso_ts_*_definitions_remain` + pattern. + """ + # Arrange + src_root = _REPO_ROOT / "src" + offenders: list[tuple[Path, str]] = [] + target_names = {"assert_engine_output_dim", "_assert_engine_output_dim"} + + # Act + for path in src_root.rglob("*.py"): + if path == _HELPER_PATH: + continue + try: + tree = ast.parse(path.read_text(encoding="utf-8")) + except SyntaxError: + continue + for node in ast.walk(tree): + if isinstance(node, ast.FunctionDef) and node.name in target_names: + offenders.append((path.relative_to(_REPO_ROOT), node.name)) + + # Assert + assert offenders == [], ( + "Found stray `assert_engine_output_dim` definitions outside the " + f"c2_vpr helper module: {offenders}" + ) + + +def test_ac4_seven_strategy_modules_import_the_helper() -> None: + """The 7 migrated strategy modules must import from the helper so + future hygiene cycles can rely on a single source of truth. + """ + # Arrange + c2_dir = _REPO_ROOT / "src" / "gps_denied_onboard" / "components" / "c2_vpr" + strategy_files = [ + c2_dir / "ultra_vpr.py", + c2_dir / "net_vlad.py", + c2_dir / "mega_loc.py", + c2_dir / "mix_vpr.py", + c2_dir / "sela_vpr.py", + c2_dir / "eigen_places.py", + c2_dir / "salad.py", + ] + expected_module = "gps_denied_onboard.components.c2_vpr._engine_dim_assertion" + missing: list[Path] = [] + + # Act + for path in strategy_files: + text = path.read_text(encoding="utf-8") + if expected_module not in text or "assert_engine_output_dim" not in text: + missing.append(path.relative_to(_REPO_ROOT)) + + # Assert + assert missing == [], ( + f"Strategy modules missing the helper import: {missing}" + )