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

[AZ-342] C2.5 ReRankStrategy: Protocol + DTOs + factory + composition

Browse Source 
Foundational scaffolding for the InlierCountReRanker (AZ-343) and
the future C3 CrossDomainMatcher consumer (AZ-344). No concrete
re-ranker is implemented here.

* ReRankStrategy Protocol (single rerank(frame, vpr_result, n,
  calibration) -> RerankResult method) with all 8 invariants in the
  docstring — notably INV-8 drop-and-continue (per-candidate failure
  NEVER propagates unless every candidate fails).
* DTOs moved to L1 _types/rerank.py — RerankCandidate, RerankResult;
  frozen+slots; tuple-not-list for RerankResult.candidates; tile_id
  encoded as (zoom_level, lat, lon) tuple to keep _types/ free of any
  c6_tile_cache (L3) import per module-layout.md.
* Error family: RerankError + RerankBackboneError +
  RerankAllCandidatesFailedError. Only RerankAllCandidatesFailedError
  escapes rerank(); RerankBackboneError is caught inside the per-
  candidate loop, logged ERROR, FDR-stamped, candidate dropped.
* C2_5RerankConfig (strategy enum default "inlier_count", top_n int
  default 3) with strict validation at load; registered into
  Config.components on c2_5_rerank import.
* build_rerank_strategy(config, *, tile_store, lightglue_runtime)
  factory: 1-strategy resolution table, lazy import,
  BUILD_RERANK_<variant> gate, ImportError → StrategyNotAvailableError
  mapping. The shared LightGlueRuntime is constructor-injected
  (R14 fix: neither C2.5 nor C3 owns its lifecycle).

Renamed the Protocol from the existing stub "RerankStrategy" to
"ReRankStrategy" to match the contract; updated module-layout.md.
Removed the legacy RerankResult shape from _types/vpr.py — the
v1.0.0 shape lives in _types/rerank.py.

Excluded per task spec:
* Concrete InlierCountReRanker (AZ-343).
* C3 matcher protocol task (AZ-344, next in batch).
* AC-9 single-thread binding + AC-10 LightGlueRuntime identity-share
  between C2.5/C3 — deferred per task spec Risk 3 until the generic
  compose_root thread-binding registry and the C3 factory both land.

Tests: AC-1..AC-8 + AC-11 + NFR-perf-factory in
tests/unit/c2_5_rerank/test_protocol_conformance.py. The legacy
smoke test is removed. Full sweep: 997 passed (one pre-existing
flake in test_az296_takeoff_abort, subprocess timing, unrelated to
this commit; passes in isolation).

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Oleksandr Bezdieniezhnykh
2026-05-12 05:31:27 +03:00
parent 3665acef66
commit d6756f1855
12 changed files with 871 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/c2_5_rerank/test_protocol_conformance.py
+404
View File
@@ -0,0 +1,404 @@
"""AZ-342 — C2.5 ReRankStrategy Protocol + DTO + error + factory conformance.
Covers AC-1..AC-8 + AC-11 + NFRs. AC-9 (single-thread binding) and
AC-10 (LightGlueRuntime identity-share between C2.5 and C3) are
deferred per the task spec's Risk-4 escape clause — the generic
compose_root thread-binding registry and the cross-factory helper
identity assertion live with AZ-270 and the future C3 protocol task
(AZ-344). Each factory owns its own thread binding today.
"""
from __future__ import annotations
import dataclasses
import logging
import sys
import time
import types
import pytest
from gps_denied_onboard._types.rerank import RerankCandidate, RerankResult
from gps_denied_onboard.components.c2_5_rerank import (
C2_5RerankConfig,
ReRankStrategy,
RerankAllCandidatesFailedError,
RerankBackboneError,
RerankError,
)
from gps_denied_onboard.components.c2_5_rerank.config import KNOWN_STRATEGIES
from gps_denied_onboard.config.schema import Config, ConfigError
from gps_denied_onboard.runtime_root.errors import StrategyNotAvailableError
from gps_denied_onboard.runtime_root.rerank_factory import build_rerank_strategy
_STRATEGY_MODULES: dict[str, tuple[str, str, str]] = {
"inlier_count": (
"gps_denied_onboard.components.c2_5_rerank.inlier_based_reranker",
"InlierCountReRanker",
"BUILD_RERANK_INLIER_COUNT",
),
}
# ----------------------------------------------------------------------
# Fakes that structurally satisfy the ReRankStrategy Protocol.
class _FakeTileStore:
def read_tile_pixels(self, tile_id):
raise NotImplementedError
def write_tile(self, tile_blob, metadata):
raise NotImplementedError
def tile_exists(self, tile_id):
return False
def delete_tile(self, tile_id):
return False
class _FakeLightGlueRuntime:
def descriptor_dim(self):
return 256
def match(self, features_a, features_b):
raise NotImplementedError
def match_batch(self, features_a_list, features_b_list):
raise NotImplementedError
class _FullReRankStrategy:
def __init__(self, config, *, tile_store, lightglue_runtime) -> None:
self._config = config
self._tile_store = tile_store
self._lightglue_runtime = lightglue_runtime
self._label = config.components["c2_5_rerank"].strategy
def rerank(self, frame, vpr_result, n, calibration):
return RerankResult(
frame_id=getattr(frame, "frame_id", 0),
candidates=tuple(),
reranked_at=1_000_000_000,
rerank_label=self._label,
candidates_input=0,
candidates_dropped=0,
)
class _PartialReRankStrategy:
pass
def _config_with_strategy(strategy: str = "inlier_count") -> Config:
return Config.with_blocks(c2_5_rerank=C2_5RerankConfig(strategy=strategy))
def _install_fake_strategy(strategy_label: str) -> type:
module_name, class_name, _flag = _STRATEGY_MODULES[strategy_label]
class _FakeStrategy(_FullReRankStrategy):
pass
_FakeStrategy.__name__ = class_name
module = types.ModuleType(module_name)
setattr(module, class_name, _FakeStrategy)
sys.modules[module_name] = module
return _FakeStrategy
@pytest.fixture
def strategy_module_cleanup():
for module_name, _, _ in _STRATEGY_MODULES.values():
sys.modules.pop(module_name, None)
yield
for module_name, _, _ in _STRATEGY_MODULES.values():
sys.modules.pop(module_name, None)
# ----------------------------------------------------------------------
# AC-1: Protocol conformance.
def test_ac1_rerank_strategy_conformance_full() -> None:
instance = _FullReRankStrategy(
_config_with_strategy(),
tile_store=_FakeTileStore(),
lightglue_runtime=_FakeLightGlueRuntime(),
)
assert isinstance(instance, ReRankStrategy)
def test_ac1_rerank_strategy_conformance_partial_missing_methods() -> None:
assert not isinstance(_PartialReRankStrategy(), ReRankStrategy)
# ----------------------------------------------------------------------
# AC-2: frozen+slotted DTOs.
def _make_candidate() -> RerankCandidate:
return RerankCandidate(
tile_id=(18, 49.9, 36.3),
inlier_count=42,
descriptor_distance=0.123,
descriptor_dim=512,
tile_pixels_handle=object(),
)
def _make_result(frame_id: int = 7) -> RerankResult:
return RerankResult(
frame_id=frame_id,
candidates=(_make_candidate(),),
reranked_at=1_000_000_000,
rerank_label="inlier_count",
candidates_input=10,
candidates_dropped=9,
)
@pytest.mark.parametrize(
"dto, field_name, new_value",
[
(_make_candidate(), "inlier_count", 99),
(_make_result(), "rerank_label", "learned_reranker"),
],
)
def test_ac2_frozen_dtos_reject_mutation(dto, field_name: str, new_value) -> None:
original_value = getattr(dto, field_name)
with pytest.raises(dataclasses.FrozenInstanceError):
setattr(dto, field_name, new_value)
assert getattr(dto, field_name) == original_value
@pytest.mark.parametrize("cls", [RerankCandidate, RerankResult])
def test_ac2_dtos_have_slots(cls) -> None:
assert hasattr(cls, "__slots__")
assert cls.__slots__
instance = _make_candidate() if cls is RerankCandidate else _make_result()
assert not hasattr(instance, "__dict__"), (
f"{cls.__name__} carries a __dict__ — slots=True is missing"
)
# ----------------------------------------------------------------------
# AC-3: factory rejects missing build flag.
def test_ac3_factory_rejects_missing_build_flag(
monkeypatch, strategy_module_cleanup, caplog
) -> None:
strategy = "inlier_count"
_, _, flag = _STRATEGY_MODULES[strategy]
monkeypatch.delenv(flag, raising=False)
config = _config_with_strategy(strategy)
with caplog.at_level(logging.ERROR, logger="gps_denied_onboard.c2_5_rerank"):
with pytest.raises(StrategyNotAvailableError) as exc_info:
build_rerank_strategy(
config,
tile_store=_FakeTileStore(),
lightglue_runtime=_FakeLightGlueRuntime(),
)
assert "BUILD_RERANK_INLIER_COUNT is OFF" in str(exc_info.value)
assert any(
r.message == "c2_5.rerank.build_flag_off" for r in caplog.records
)
def test_ac3_factory_does_not_load_module_when_flag_off(
monkeypatch, strategy_module_cleanup
) -> None:
module_name, _, flag = _STRATEGY_MODULES["inlier_count"]
monkeypatch.delenv(flag, raising=False)
config = _config_with_strategy("inlier_count")
with pytest.raises(StrategyNotAvailableError):
build_rerank_strategy(
config,
tile_store=_FakeTileStore(),
lightglue_runtime=_FakeLightGlueRuntime(),
)
assert module_name not in sys.modules
# ----------------------------------------------------------------------
# AC-4: unknown strategy rejected at config-load time.
@pytest.mark.parametrize(
"bad_label",
["INLIER_COUNT", "garbage", "", "learned_reranker"],
)
def test_ac4_unknown_strategy_rejected_at_config_load(bad_label: str) -> None:
with pytest.raises(ConfigError) as exc_info:
C2_5RerankConfig(strategy=bad_label)
msg = str(exc_info.value)
for valid in KNOWN_STRATEGIES:
assert valid in msg
# ----------------------------------------------------------------------
# AC-5: factory emits INFO log on success.
def test_ac5_factory_emits_info_log_on_success(
monkeypatch, strategy_module_cleanup, caplog
) -> None:
strategy = "inlier_count"
_, _, flag = _STRATEGY_MODULES[strategy]
monkeypatch.setenv(flag, "ON")
_install_fake_strategy(strategy)
config = _config_with_strategy(strategy)
with caplog.at_level(logging.INFO, logger="gps_denied_onboard.c2_5_rerank"):
instance = build_rerank_strategy(
config,
tile_store=_FakeTileStore(),
lightglue_runtime=_FakeLightGlueRuntime(),
)
assert isinstance(instance, ReRankStrategy)
records = [
r for r in caplog.records if r.message == "c2_5.rerank.strategy_loaded"
]
assert len(records) == 1
record = records[0]
assert getattr(record, "strategy", None) == "inlier_count"
assert getattr(record, "top_n", None) == 3
# ----------------------------------------------------------------------
# AC-6: strategy resolution table.
def test_ac6_strategy_resolution(monkeypatch, strategy_module_cleanup) -> None:
strategy = "inlier_count"
module_name, class_name, flag = _STRATEGY_MODULES[strategy]
monkeypatch.setenv(flag, "ON")
fake_cls = _install_fake_strategy(strategy)
config = _config_with_strategy(strategy)
instance = build_rerank_strategy(
config,
tile_store=_FakeTileStore(),
lightglue_runtime=_FakeLightGlueRuntime(),
)
assert isinstance(instance, fake_cls)
assert isinstance(instance, ReRankStrategy)
assert sys.modules[module_name] is not None
# ----------------------------------------------------------------------
# AC-7: error hierarchy.
@pytest.mark.parametrize(
"exc_factory",
[RerankBackboneError, RerankAllCandidatesFailedError],
)
def test_ac7_all_rerank_errors_caught_as_family(exc_factory) -> None:
with pytest.raises(RerankError):
raise exc_factory("boom")
def test_ac7_strategy_not_available_outside_family() -> None:
with pytest.raises(StrategyNotAvailableError):
try:
raise StrategyNotAvailableError("composition-time")
except RerankError:
pytest.fail(
"StrategyNotAvailableError is a composition-root error "
"and MUST NOT be in the c2.5 RerankError family"
)
# ----------------------------------------------------------------------
# AC-8: Public API re-exports.
def test_ac8_public_api_re_exports() -> None:
from gps_denied_onboard.components import c2_5_rerank
assert "ReRankStrategy" in c2_5_rerank.__all__
assert "RerankResult" in c2_5_rerank.__all__
assert "RerankCandidate" in c2_5_rerank.__all__
def test_ac8_internals_not_in_public_api() -> None:
from gps_denied_onboard.components import c2_5_rerank
# Concrete strategy must not leak into the package re-exports;
# consumers see only the Protocol.
assert "InlierCountReRanker" not in c2_5_rerank.__all__
# ----------------------------------------------------------------------
# AC-11: tile_pixels_handle opaqueness.
def test_ac11_tile_pixels_handle_opaque() -> None:
handle = object()
candidate = RerankCandidate(
tile_id=(18, 49.9, 36.3),
inlier_count=10,
descriptor_distance=0.5,
descriptor_dim=256,
tile_pixels_handle=handle,
)
assert candidate.tile_pixels_handle is handle
# ----------------------------------------------------------------------
# NFRs.
@pytest.mark.parametrize(
"exc_type",
[RerankBackboneError, RerankAllCandidatesFailedError],
)
def test_nfr_reliability_all_rerank_errors_subclass_family(exc_type) -> None:
assert issubclass(exc_type, RerankError)
def test_nfr_reliability_strategy_not_available_not_in_family() -> None:
assert not issubclass(StrategyNotAvailableError, RerankError)
def test_nfr_perf_factory_under_50ms_p99(
monkeypatch, strategy_module_cleanup
) -> None:
strategy = "inlier_count"
_, _, flag = _STRATEGY_MODULES[strategy]
monkeypatch.setenv(flag, "ON")
_install_fake_strategy(strategy)
config = _config_with_strategy(strategy)
tile_store = _FakeTileStore()
lightglue_runtime = _FakeLightGlueRuntime()
durations_ms: list[float] = []
for _ in range(100):
t0 = time.perf_counter()
build_rerank_strategy(
config, tile_store=tile_store, lightglue_runtime=lightglue_runtime
)
durations_ms.append((time.perf_counter() - t0) * 1000.0)
durations_ms.sort()
p99 = durations_ms[int(0.99 * len(durations_ms))]
assert p99 <= 50.0
# ----------------------------------------------------------------------
# Surface coverage — config defaults.
def test_c2_5_config_defaults() -> None:
cfg = C2_5RerankConfig()
assert cfg.strategy == "inlier_count"
assert cfg.top_n == 3
def test_c2_5_config_top_n_validation() -> None:
with pytest.raises(ConfigError):
C2_5RerankConfig(top_n=0)
with pytest.raises(ConfigError):
C2_5RerankConfig(top_n=-3)