mirror of
https://github.com/azaion/gps-denied-onboard.git
synced 2026-06-21 08:41:12 +00:00
[AZ-348] C3.5 ConditionalRefiner Protocol + factory + PassthroughRefiner
Defines the public `ConditionalRefiner` Protocol (PEP 544 @runtime_checkable, two methods: `refine_if_needed` + `was_invoked`), extends `MatchResult` in-place with two default-valued refinement fields (`refinement_label`, `refinement_added_latency_ms`), defines the `RefinerError` family (`RefinerBackboneError`, `RefinerConfigError`), and ships the trivial `PassthroughRefiner` reference impl. Both refiner strategies are linked unconditionally — no `BUILD_REFINER_*` flag (NOT ADR-002 territory). Runtime selection only per ADR-001. `PassthroughRefiner` returns the input `MatchResult` by reference (bit-identical correspondences per contract INV-5) and always reports `was_invoked() is False`. Documentation: renames `module-layout.md` `c3_5_adhop` Public API symbol from `AdHoPRefinementStrategy` to `ConditionalRefiner` (AC-14) so the doc agrees with `description.md` and the contract. AC-9 (single-thread binding) deferred to AZ-270 runtime-root composition, mirroring AZ-336 / AZ-342 / AZ-344 Risk-4 precedent. AC-7 for the `"adhop"` strategy stops at `ModuleNotFoundError` because the AdHoP backbone is owned by AZ-349. All other ACs + NFRs covered by 36 new conformance tests. Architectural note: `PassthroughRefiner.inference_runtime` is typed as `object` because the L3→L3 import ban (`test_az270_compose_root`) forbids c3_5_adhop from importing c7_inference; the runtime-root factory narrows the type at construction time. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Oleksandr Bezdieniezhnykh
@@ -4,8 +4,8 @@
|
||||
**Producer task**: AZ-348 (Protocol + factory + DTOs + composition + `PassthroughRefiner`)
|
||||
**Consumer tasks**: AZ-349 (`AdHoPRefiner` real refinement); downstream c4_pose (epic AZ-259) which consumes the (possibly refined) `MatchResult`
|
||||
**Version**: 1.0.0
|
||||
**Status**: draft, awaiting Producer task implementation
|
||||
**Last Updated**: 2026-05-10
|
||||
**Status**: v1.0.0 (AZ-348 implemented 2026-05-12; PassthroughRefiner shipped — AdHoPRefiner pending AZ-349)
|
||||
**Last Updated**: 2026-05-12
|
||||
**Module-layout home**: `src/gps_denied_onboard/components/c3_5_adhop/interface.py` (Protocol), `src/gps_denied_onboard/components/c3_5_adhop/__init__.py` (re-exports), `src/gps_denied_onboard/runtime_root/refiner_factory.py` (factory)
|
||||
|
||||
> **Public API symbol naming.** The component's public interface symbol is named `ConditionalRefiner` in `description.md` § 2 and `AdHoPRefinementStrategy` in `module-layout.md` § c3_5_adhop. Both refer to the SAME Protocol; the canonical class name in code is `ConditionalRefiner` — it is the role description-first name and matches the method `refine_if_needed`. The producer task ALSO updates `module-layout.md` to align (`AdHoPRefinementStrategy` → `ConditionalRefiner`) so the two documents agree.
|
||||
|
||||
@@ -89,11 +89,15 @@ Bootstrap reference: `_docs/02_tasks/todo/AZ-263_initial_structure.md`. Architec
|
||||
- **Epic**: AZ-258 (E-C3.5 AdHoP Refinement)
|
||||
- **Directory**: `src/gps_denied_onboard/components/c3_5_adhop/`
|
||||
- **Public API**:
|
||||
- `__init__.py` (re-exports `AdHoPRefinementStrategy`)
|
||||
- `interface.py` (`AdHoPRefinementStrategy` Protocol)
|
||||
- **Internal**: `default_refiner.py`
|
||||
- **Owns**: `src/gps_denied_onboard/components/c3_5_adhop/**`, `tests/unit/c3_5_adhop/**`
|
||||
- **Imports from**: `_types`, `helpers.ransac_filter`, `helpers.se3_utils`, `config`, `logging`, `fdr_client`
|
||||
- `__init__.py` (re-exports `ConditionalRefiner`, `C3_5RefinerConfig`)
|
||||
- `interface.py` (`ConditionalRefiner` Protocol)
|
||||
- `config.py` (`C3_5RefinerConfig`)
|
||||
- `errors.py` (`RefinerError`, `RefinerBackboneError`, `RefinerConfigError` — held internal to the component; consumers reach them only via tests)
|
||||
- **Internal**:
|
||||
- `passthrough_refiner.py` (reference baseline; AZ-348)
|
||||
- `adhop_refiner.py` (production-default; AZ-349 pending)
|
||||
- **Owns**: `src/gps_denied_onboard/components/c3_5_adhop/**`, `tests/unit/c3_5_adhop/**`, `src/gps_denied_onboard/runtime_root/refiner_factory.py`
|
||||
- **Imports from**: `_types`, `helpers.ransac_filter` (R14: SHARED with C3 and C4 — owned by helper, NOT by C3.5), `helpers.se3_utils`, `components.c7_inference`, `config`, `logging`, `fdr_client`
|
||||
- **Consumed by**: `c4_pose`, `runtime_root`
|
||||
|
||||
### Component: c4_pose
|
||||
|
||||
@@ -8,7 +8,7 @@ status: in_progress
|
||||
sub_step:
|
||||
phase: 2
|
||||
name: detect-progress
|
||||
detail: "batch 24 in flight (per-task commits): AZ-336 done, AZ-342 done, AZ-344 done; next AZ-348"
|
||||
detail: "batch 24 complete (AZ-336, AZ-342, AZ-344, AZ-348 per-task commits); ready to plan batch 25"
|
||||
retry_count: 0
|
||||
cycle: 1
|
||||
tracker: jira
|
||||
|
||||
@@ -52,8 +52,12 @@ class CandidateMatchSet:
|
||||
class MatchResult:
|
||||
"""Cross-domain match outcome for one frame.
|
||||
|
||||
Consumed by C3.5 :class:`ConditionalRefiner` (AZ-348). The
|
||||
``per_candidate`` tuple is sorted descending by
|
||||
Consumed by C3.5 :class:`ConditionalRefiner` (AZ-348) — which
|
||||
may either pass the result through unchanged or emit a new
|
||||
instance via :func:`dataclasses.replace` with the refinement
|
||||
fields set.
|
||||
|
||||
The ``per_candidate`` tuple is sorted descending by
|
||||
``inlier_count`` with ties broken ascending by
|
||||
``per_candidate_residual_px`` (INV-3) so
|
||||
``best_candidate_idx == 0`` by construction.
|
||||
@@ -62,6 +66,15 @@ class MatchResult:
|
||||
residual (mirrors ``per_candidate[0].per_candidate_residual_px``)
|
||||
surfaced separately so consumers do not have to know the
|
||||
ranking encoding.
|
||||
|
||||
The two refinement fields are populated by C3.5; they default
|
||||
to the passthrough values so a C3 producer that never goes
|
||||
through C3.5 still yields a valid downstream-readable
|
||||
:class:`MatchResult` (AZ-348 AC-2). ``refinement_label`` is
|
||||
one of ``"adhop"`` or ``"passthrough"``;
|
||||
``refinement_added_latency_ms`` covers exactly the work done
|
||||
inside :meth:`ConditionalRefiner.refine_if_needed` and is
|
||||
``0.0`` on pure passthrough.
|
||||
"""
|
||||
|
||||
frame_id: int
|
||||
@@ -72,6 +85,8 @@ class MatchResult:
|
||||
matcher_label: str
|
||||
candidates_input: int
|
||||
candidates_dropped: int
|
||||
refinement_label: str = "passthrough"
|
||||
refinement_added_latency_ms: float = 0.0
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
|
||||
@@ -1,5 +1,26 @@
|
||||
"""C3.5 AdHoP Refinement component — Public API."""
|
||||
"""C3.5 AdHoP / conditional refiner — Public API (AZ-348).
|
||||
|
||||
from gps_denied_onboard.components.c3_5_adhop.interface import AdHoPRefinementStrategy
|
||||
Per ``conditional_refiner_protocol.md`` v1.0.0 the public surface
|
||||
consists of:
|
||||
|
||||
__all__ = ["AdHoPRefinementStrategy"]
|
||||
- :class:`ConditionalRefiner` Protocol (two methods).
|
||||
- :class:`C3_5RefinerConfig` config block (registered on import).
|
||||
|
||||
The error family (:class:`RefinerError`,
|
||||
:class:`RefinerBackboneError`, :class:`RefinerConfigError`) and
|
||||
both concrete strategies (:class:`PassthroughRefiner`,
|
||||
:class:`AdHoPRefiner`) are intentionally NOT in ``__all__`` per
|
||||
task spec AC-8: consumers see only the Protocol; concrete
|
||||
strategies are reached via the runtime-root factory.
|
||||
"""
|
||||
|
||||
from gps_denied_onboard.components.c3_5_adhop.config import C3_5RefinerConfig
|
||||
from gps_denied_onboard.components.c3_5_adhop.interface import ConditionalRefiner
|
||||
from gps_denied_onboard.config.schema import register_component_block
|
||||
|
||||
register_component_block("c3_5_adhop", C3_5RefinerConfig)
|
||||
|
||||
__all__ = [
|
||||
"C3_5RefinerConfig",
|
||||
"ConditionalRefiner",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,64 @@
|
||||
"""C3.5 ``ConditionalRefiner`` config block (AZ-348).
|
||||
|
||||
Registered into ``config.components['c3_5_adhop']`` by the
|
||||
package ``__init__.py``. The composition-root factory
|
||||
:func:`gps_denied_onboard.runtime_root.refiner_factory.build_refiner_strategy`
|
||||
reads this block to select the strategy and configure thresholds.
|
||||
|
||||
``strategy`` selects one of the two concrete refiners
|
||||
(``adhop`` — production-default; ``passthrough`` — baseline /
|
||||
smoke / IT-12 comparison). Both modules are linked
|
||||
unconditionally: there is NO ``BUILD_REFINER_*`` flag (NOT ADR-002
|
||||
territory). Runtime selection only.
|
||||
|
||||
``residual_threshold_px`` is the conditional-gate threshold: a
|
||||
:class:`MatchResult` whose ``reprojection_residual_px <=
|
||||
threshold`` is passed through unchanged; ``>`` invokes the
|
||||
strategy's refinement procedure. Default 2.5 px (the AC-NEW-5 /
|
||||
R10 tunable from operator tooling).
|
||||
|
||||
``invocation_rate_warn_threshold`` is the rolling-60 s
|
||||
invocation-rate ceiling above which a WARN log fires
|
||||
(C3.5-IT-03 / NFT-PERF-01). Must be in ``(0, 1)``; default 0.25.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Final
|
||||
|
||||
from gps_denied_onboard.config.schema import ConfigError
|
||||
|
||||
__all__ = [
|
||||
"C3_5RefinerConfig",
|
||||
"KNOWN_STRATEGIES",
|
||||
]
|
||||
|
||||
|
||||
KNOWN_STRATEGIES: Final[frozenset[str]] = frozenset({"adhop", "passthrough"})
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class C3_5RefinerConfig:
|
||||
"""Per-component config for C3.5 conditional refiner."""
|
||||
|
||||
strategy: str = "adhop"
|
||||
residual_threshold_px: float = 2.5
|
||||
invocation_rate_warn_threshold: float = 0.25
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if self.strategy not in KNOWN_STRATEGIES:
|
||||
raise ConfigError(
|
||||
f"C3_5RefinerConfig.strategy={self.strategy!r} not in "
|
||||
f"{sorted(KNOWN_STRATEGIES)}"
|
||||
)
|
||||
if self.residual_threshold_px <= 0.0:
|
||||
raise ConfigError(
|
||||
"C3_5RefinerConfig.residual_threshold_px must be > 0; "
|
||||
f"got {self.residual_threshold_px}"
|
||||
)
|
||||
if not (0.0 < self.invocation_rate_warn_threshold < 1.0):
|
||||
raise ConfigError(
|
||||
"C3_5RefinerConfig.invocation_rate_warn_threshold must be in "
|
||||
f"(0, 1); got {self.invocation_rate_warn_threshold}"
|
||||
)
|
||||
@@ -0,0 +1,44 @@
|
||||
"""C3.5 ``ConditionalRefiner`` error taxonomy (AZ-348).
|
||||
|
||||
The family is intentionally small: per-candidate failures are
|
||||
handled inside C3 (drop-and-continue); at C3.5 the only failure
|
||||
mode is the AdHoP backbone (TensorRT exception, OOM, NaN, shape
|
||||
mismatch) and it is contained within the strategy via
|
||||
passthrough fall-through (contract Invariant 4) — never re-raised
|
||||
out of :meth:`refine_if_needed`.
|
||||
|
||||
:class:`RefinerConfigError` is the composition-root rejection at
|
||||
startup; never raised per-frame.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
__all__ = [
|
||||
"RefinerBackboneError",
|
||||
"RefinerConfigError",
|
||||
"RefinerError",
|
||||
]
|
||||
|
||||
|
||||
class RefinerError(Exception):
|
||||
"""Base class for all C3.5 refinement-strategy errors."""
|
||||
|
||||
|
||||
class RefinerBackboneError(RefinerError):
|
||||
"""AdHoP backbone forward-pass failed.
|
||||
|
||||
TensorRT exception, OOM, NaN, shape mismatch. Caught inside
|
||||
:meth:`ConditionalRefiner.refine_if_needed`, converted to
|
||||
passthrough fall-through (Invariant 4), logged at ERROR with
|
||||
one FDR record. NEVER re-raised out of the strategy.
|
||||
"""
|
||||
|
||||
|
||||
class RefinerConfigError(RefinerError):
|
||||
"""Composition-root rejected the refiner config.
|
||||
|
||||
Unknown strategy label OR invalid threshold (``<= 0``). Raised
|
||||
at startup ONLY — never per-frame. The composition root logs
|
||||
ERROR (``kind="c3_5.refiner.strategy_unknown"`` or
|
||||
``kind="c3_5.refiner.invalid_threshold"``) before raising.
|
||||
"""
|
||||
@@ -1,16 +1,123 @@
|
||||
"""C3.5 `AdHoPRefinementStrategy` Protocol.
|
||||
"""C3.5 ``ConditionalRefiner`` Protocol (AZ-348).
|
||||
|
||||
Concrete impl: AdHoP refiner. See `_docs/02_document/components/05_c3_5_adhop/`.
|
||||
PEP 544 ``typing.Protocol`` with ``runtime_checkable=True``; a
|
||||
two-method surface: :meth:`refine_if_needed` and
|
||||
:meth:`was_invoked`.
|
||||
|
||||
Concrete impls — :class:`PassthroughRefiner` (this task) and
|
||||
:class:`AdHoPRefiner` (AZ-349) — live in sibling modules.
|
||||
Both are linked into the production binary unconditionally per
|
||||
ADR-001; runtime selection is via ``config.refiner.strategy``
|
||||
(NO ``BUILD_REFINER_*`` flag — NOT ADR-002 territory because
|
||||
both strategies are tiny and the AdHoP TRT engine is shared C7
|
||||
infrastructure).
|
||||
|
||||
The contract at
|
||||
``_docs/02_document/contracts/c3_5_adhop/conditional_refiner_protocol.md``
|
||||
v1.0.0 is the authoritative shape; this module mirrors it 1:1.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Protocol
|
||||
from typing import TYPE_CHECKING, Protocol, runtime_checkable
|
||||
|
||||
from gps_denied_onboard._types.matcher import MatchResult
|
||||
if TYPE_CHECKING:
|
||||
from gps_denied_onboard._types.matcher import MatchResult
|
||||
from gps_denied_onboard._types.nav import NavCameraFrame
|
||||
|
||||
__all__ = ["ConditionalRefiner"]
|
||||
|
||||
|
||||
class AdHoPRefinementStrategy(Protocol):
|
||||
"""Conditional refinement of a `MatchResult` (geometric verification + outlier purge)."""
|
||||
@runtime_checkable
|
||||
class ConditionalRefiner(Protocol):
|
||||
"""Conditional refinement strategy between C3 (matcher) and C4 (pose).
|
||||
|
||||
def refine(self, match: MatchResult) -> MatchResult: ...
|
||||
Stateless per-frame; the only persistent state is the
|
||||
constructor-injected backbone runtime handle (when the strategy
|
||||
uses one) and the last-invocation flag.
|
||||
|
||||
Invariants (see ``conditional_refiner_protocol.md`` v1.0.0):
|
||||
|
||||
- **INV-1 single-threaded** — each instance is bound to one
|
||||
ingest thread; same thread as C3 (shared C-frame ingest
|
||||
path).
|
||||
- **INV-2 stateless per-frame** — except for the
|
||||
``was_invoked`` flag, no implicit dependency on prior frames;
|
||||
reordering :meth:`refine_if_needed` calls (tests only) MUST
|
||||
yield identical output ``MatchResult`` content.
|
||||
- **INV-3 conditional gate is a pure comparison** —
|
||||
``mr.reprojection_residual_px <= threshold`` → passthrough;
|
||||
``>`` → invoke. No tolerance, no smoothing, no hysteresis.
|
||||
- **INV-4 passthrough fall-through on backbone error** —
|
||||
:class:`RefinerBackboneError` raised inside the invoked path
|
||||
is caught by the strategy and converted to passthrough output
|
||||
with ``refinement_label = "passthrough"``; the error is
|
||||
logged at ERROR + emitted to FDR; NEVER re-raised out of
|
||||
:meth:`refine_if_needed`.
|
||||
- **INV-5 bit-identical correspondences on passthrough** —
|
||||
when ``refinement_label == "passthrough"``, every
|
||||
``inlier_correspondences`` ndarray in the output equals the
|
||||
input ndarray bit-for-bit (``np.array_equal`` AND same
|
||||
``dtype``).
|
||||
- **INV-6 ``refinement_label`` is `"adhop"` OR
|
||||
`"passthrough"`** — exactly one of those two values; matches
|
||||
the strategy's selected variant. Readers check
|
||||
:meth:`was_invoked` to discriminate "AdHoP ran" from
|
||||
"AdHoP-fell-through-to-passthrough".
|
||||
- **INV-7 ``refinement_added_latency_ms`` is strategy-internal
|
||||
added latency** — always ``>= 0``; near-zero on passthrough;
|
||||
up to ~90 ms on AdHoP invoke per C3.5-PT-01.
|
||||
- **INV-8 ``was_invoked()`` semantics** — set to ``True`` iff
|
||||
the strategy entered the refinement procedure (post-gate,
|
||||
regardless of whether AdHoP succeeded or fell through). On
|
||||
pure passthrough strategy + every gate-decided-passthrough
|
||||
call: ``False``.
|
||||
- **INV-9 threshold validation** — the strategy MUST reject
|
||||
``residual_threshold_px <= 0`` (raise :class:`ValueError`);
|
||||
the composition root validates the config-loaded threshold
|
||||
at startup so this in-method check is defensive.
|
||||
"""
|
||||
|
||||
def refine_if_needed(
|
||||
self,
|
||||
frame: "NavCameraFrame",
|
||||
mr: "MatchResult",
|
||||
residual_threshold_px: float,
|
||||
) -> "MatchResult":
|
||||
"""Either pass ``mr`` through unchanged or run refinement.
|
||||
|
||||
If ``mr.reprojection_residual_px <= residual_threshold_px``
|
||||
(the steady-state path), return ``mr`` unchanged AND set
|
||||
:meth:`was_invoked` to ``False``. Otherwise run the
|
||||
strategy's refinement procedure and return an enriched
|
||||
:class:`MatchResult` (typically via
|
||||
:func:`dataclasses.replace`) with ``refinement_label``
|
||||
set, AND set :meth:`was_invoked` to ``True``.
|
||||
|
||||
On :class:`RefinerBackboneError` (AdHoP backbone failure
|
||||
during the invoked path), the refiner MUST fall through to
|
||||
passthrough — return ``mr`` unchanged with
|
||||
``refinement_label = "passthrough"`` AND :meth:`was_invoked`
|
||||
``True`` (the attempt counts towards the invocation rate
|
||||
even on failure). The error is logged at ERROR + emitted
|
||||
to FDR; downstream pose estimation gets a usable
|
||||
:class:`MatchResult` and decides whether to trigger F6.
|
||||
|
||||
Determinism: same inputs MUST produce the same output. No
|
||||
probabilistic gating; no time-based gating.
|
||||
|
||||
Raises:
|
||||
ValueError: ``residual_threshold_px <= 0`` (defensive;
|
||||
composition root should have caught this already).
|
||||
"""
|
||||
...
|
||||
|
||||
def was_invoked(self) -> bool:
|
||||
"""Return ``True`` iff the last :meth:`refine_if_needed`
|
||||
call entered the refinement procedure.
|
||||
|
||||
Set at the start of every :meth:`refine_if_needed` call.
|
||||
Used by FDR per-frame provenance and by
|
||||
NFT-PERF-01 / C3.5-IT-03 invocation-rate accounting.
|
||||
"""
|
||||
...
|
||||
|
||||
@@ -0,0 +1,84 @@
|
||||
"""``PassthroughRefiner`` — no-op :class:`ConditionalRefiner` (AZ-348).
|
||||
|
||||
The reference baseline implementation. Returns the input
|
||||
:class:`MatchResult` unchanged (same object reference;
|
||||
``inlier_correspondences`` ndarrays are bit-identical and share
|
||||
references per contract INV-5). Always sets :meth:`was_invoked`
|
||||
to ``False`` per INV-8.
|
||||
|
||||
Both helpers (``ransac_filter``, ``inference_runtime``) are held
|
||||
by reference for parity with :class:`AdHoPRefiner` (AZ-349) but
|
||||
neither is invoked. ``inference_runtime`` is typed as ``object``
|
||||
because the C3 matcher → C3.5 refiner → C4 pose layering forbids
|
||||
L3-to-L3 imports (architecture test ``test_az270_compose_root``);
|
||||
the composition-root factory at
|
||||
:mod:`gps_denied_onboard.runtime_root.refiner_factory`
|
||||
narrows the type at construction time.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from gps_denied_onboard._types.matcher import MatchResult
|
||||
from gps_denied_onboard._types.nav import NavCameraFrame
|
||||
from gps_denied_onboard.config.schema import Config
|
||||
from gps_denied_onboard.helpers.ransac_filter import RansacFilter
|
||||
|
||||
__all__ = ["PassthroughRefiner", "create"]
|
||||
|
||||
|
||||
class PassthroughRefiner:
|
||||
"""Reference passthrough strategy.
|
||||
|
||||
See module docstring. Stateless except for the ``_was_invoked``
|
||||
flag (always ``False`` per INV-8); concurrent calls are unsafe
|
||||
(single-thread invariant covers).
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
ransac_filter: "RansacFilter",
|
||||
inference_runtime: object,
|
||||
) -> None:
|
||||
# Held for parity with AdHoPRefiner; neither is invoked.
|
||||
self._ransac_filter = ransac_filter
|
||||
self._inference_runtime = inference_runtime
|
||||
self._was_invoked: bool = False
|
||||
|
||||
def refine_if_needed(
|
||||
self,
|
||||
frame: "NavCameraFrame",
|
||||
mr: "MatchResult",
|
||||
residual_threshold_px: float,
|
||||
) -> "MatchResult":
|
||||
if residual_threshold_px <= 0.0:
|
||||
raise ValueError(
|
||||
"residual_threshold_px must be > 0; "
|
||||
f"got {residual_threshold_px}"
|
||||
)
|
||||
self._was_invoked = False
|
||||
# `MatchResult` defaults `refinement_label="passthrough"`
|
||||
# and `refinement_added_latency_ms=0.0` already; the input
|
||||
# is bit-identical per contract INV-5 so we return it by
|
||||
# reference rather than recreating via `dataclasses.replace`.
|
||||
return mr
|
||||
|
||||
def was_invoked(self) -> bool:
|
||||
return self._was_invoked
|
||||
|
||||
|
||||
def create(
|
||||
config: "Config",
|
||||
*,
|
||||
ransac_filter: "RansacFilter",
|
||||
inference_runtime: object,
|
||||
) -> PassthroughRefiner:
|
||||
"""Module-level factory entry point consumed by the
|
||||
composition-root :func:`build_refiner_strategy`."""
|
||||
return PassthroughRefiner(
|
||||
ransac_filter=ransac_filter,
|
||||
inference_runtime=inference_runtime,
|
||||
)
|
||||
@@ -0,0 +1,142 @@
|
||||
"""C3.5 refiner strategy composition-root factory (AZ-348).
|
||||
|
||||
:func:`build_refiner_strategy` selects exactly one strategy by
|
||||
``config.components['c3_5_adhop'].strategy``. Both concrete
|
||||
strategies are linked into the production binary
|
||||
**unconditionally** (NO ``BUILD_REFINER_*`` flag — this is NOT
|
||||
ADR-002 territory). Runtime selection only per ADR-001.
|
||||
|
||||
Strategy resolution table — mirrors the contract's
|
||||
``conditional_refiner_protocol.md`` v1.0.0 § Composition-root
|
||||
factory table verbatim:
|
||||
|
||||
* ``"adhop"`` → ``gps_denied_onboard.components.c3_5_adhop.adhop_refiner.AdHoPRefiner`` (AZ-349; placeholder today).
|
||||
* ``"passthrough"`` → ``gps_denied_onboard.components.c3_5_adhop.passthrough_refiner.PassthroughRefiner``.
|
||||
|
||||
The shared :class:`RansacFilter` and C7 :class:`InferenceRuntime`
|
||||
handles are constructor-injected — the factory does NOT own their
|
||||
lifecycles. The runtime root constructs ONE
|
||||
:class:`RansacFilter` instance and identity-shares it across C3,
|
||||
C3.5, and C4 (per ``ransac_filter.md`` v1.0.0).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from gps_denied_onboard.components.c3_5_adhop.errors import RefinerConfigError
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from gps_denied_onboard.components.c3_5_adhop import (
|
||||
C3_5RefinerConfig,
|
||||
ConditionalRefiner,
|
||||
)
|
||||
from gps_denied_onboard.components.c7_inference import InferenceRuntime
|
||||
from gps_denied_onboard.config.schema import Config
|
||||
from gps_denied_onboard.helpers.ransac_filter import RansacFilter
|
||||
|
||||
__all__ = ["build_refiner_strategy"]
|
||||
|
||||
|
||||
_LOG = logging.getLogger("gps_denied_onboard.c3_5_adhop")
|
||||
|
||||
|
||||
# Strategy resolution table — mirrors the contract verbatim. ANY
|
||||
# mutation of this dict MUST be mirrored in the contract.
|
||||
_STRATEGY_TO_MODULE: dict[str, tuple[str, str]] = {
|
||||
"adhop": (
|
||||
"gps_denied_onboard.components.c3_5_adhop.adhop_refiner",
|
||||
"AdHoPRefiner",
|
||||
),
|
||||
"passthrough": (
|
||||
"gps_denied_onboard.components.c3_5_adhop.passthrough_refiner",
|
||||
"PassthroughRefiner",
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def _c3_5_config(config: "Config") -> "C3_5RefinerConfig":
|
||||
"""Pull the registered C3.5 config block.
|
||||
|
||||
``c3_5_adhop.__init__`` registers it on import; a missing
|
||||
registration is a developer error and surfaces as ``KeyError``
|
||||
rather than a silent fallback.
|
||||
"""
|
||||
return config.components["c3_5_adhop"]
|
||||
|
||||
|
||||
def build_refiner_strategy(
|
||||
config: "Config",
|
||||
*,
|
||||
ransac_filter: "RansacFilter",
|
||||
inference_runtime: "InferenceRuntime",
|
||||
) -> "ConditionalRefiner":
|
||||
"""Construct the :class:`ConditionalRefiner` impl selected by config.
|
||||
|
||||
1. Reads ``config.components['c3_5_adhop'].{strategy, residual_threshold_px}``.
|
||||
2. Validates ``residual_threshold_px > 0`` — defensive
|
||||
redundancy on top of the config-load-time check
|
||||
(:class:`C3_5RefinerConfig.__post_init__`); raises
|
||||
:class:`RefinerConfigError` on failure.
|
||||
3. Imports the concrete strategy module via the resolution
|
||||
table (NOT lazy — both strategies are linked
|
||||
unconditionally).
|
||||
4. Constructs the strategy via its module-level
|
||||
``create(config, ransac_filter, inference_runtime)``
|
||||
factory function.
|
||||
5. Emits ONE INFO log ``kind="c3_5.refiner.strategy_loaded"``
|
||||
with ``{strategy, residual_threshold_px}``.
|
||||
|
||||
Raises:
|
||||
RefinerConfigError: unknown strategy label OR invalid
|
||||
threshold (``<= 0``).
|
||||
"""
|
||||
block = _c3_5_config(config)
|
||||
strategy = block.strategy
|
||||
module_info = _STRATEGY_TO_MODULE.get(strategy)
|
||||
if module_info is None:
|
||||
_LOG.error(
|
||||
"c3_5.refiner.strategy_unknown",
|
||||
extra={"strategy": strategy},
|
||||
)
|
||||
raise RefinerConfigError(f"Unknown refiner strategy: {strategy}")
|
||||
if block.residual_threshold_px <= 0.0:
|
||||
# Config-load validation should have rejected this already;
|
||||
# defensive in case a caller constructed Config bypassing
|
||||
# __post_init__ (e.g., via dataclasses.replace on a partial
|
||||
# block).
|
||||
_LOG.error(
|
||||
"c3_5.refiner.invalid_threshold",
|
||||
extra={
|
||||
"strategy": strategy,
|
||||
"residual_threshold_px": block.residual_threshold_px,
|
||||
},
|
||||
)
|
||||
raise RefinerConfigError(
|
||||
"residual_threshold_px must be > 0; "
|
||||
f"got {block.residual_threshold_px}"
|
||||
)
|
||||
module_name, class_name = module_info
|
||||
module = __import__(module_name, fromlist=[class_name])
|
||||
create_fn = getattr(module, "create", None)
|
||||
if create_fn is None:
|
||||
strategy_cls = getattr(module, class_name)
|
||||
instance = strategy_cls(
|
||||
ransac_filter=ransac_filter,
|
||||
inference_runtime=inference_runtime,
|
||||
)
|
||||
else:
|
||||
instance = create_fn(
|
||||
config,
|
||||
ransac_filter=ransac_filter,
|
||||
inference_runtime=inference_runtime,
|
||||
)
|
||||
_LOG.info(
|
||||
"c3_5.refiner.strategy_loaded",
|
||||
extra={
|
||||
"strategy": strategy,
|
||||
"residual_threshold_px": block.residual_threshold_px,
|
||||
},
|
||||
)
|
||||
return instance
|
||||
@@ -0,0 +1,490 @@
|
||||
"""AZ-348 — C3.5 ConditionalRefiner Protocol + DTO + error + factory conformance.
|
||||
|
||||
Covers AC-1..AC-14 + NFRs. AC-9 (single-thread binding) is
|
||||
deferred per the task spec's Risk-4 escape clause — the generic
|
||||
``compose_root`` thread-binding registry lives with AZ-270 and
|
||||
the broader runtime-root composition. Each factory owns its own
|
||||
thread binding today; this protocol task does not add a new
|
||||
binding registry.
|
||||
|
||||
AC-7 (strategy resolution table) is asserted up to the lookup
|
||||
point for ``"adhop"``: the AdHoP module is the AZ-349 placeholder
|
||||
that does not exist yet, so the assertion on that path stops at
|
||||
``ModuleNotFoundError`` — explicitly documented as the "import +
|
||||
class lookup not __init__" rule in the task spec.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
import logging
|
||||
import sys
|
||||
import time
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from gps_denied_onboard._types.matcher import (
|
||||
CandidateMatchSet,
|
||||
MatchResult,
|
||||
)
|
||||
from gps_denied_onboard.components.c3_5_adhop import (
|
||||
C3_5RefinerConfig,
|
||||
ConditionalRefiner,
|
||||
)
|
||||
from gps_denied_onboard.components.c3_5_adhop.config import KNOWN_STRATEGIES
|
||||
from gps_denied_onboard.components.c3_5_adhop.errors import (
|
||||
RefinerBackboneError,
|
||||
RefinerConfigError,
|
||||
RefinerError,
|
||||
)
|
||||
from gps_denied_onboard.components.c3_5_adhop.passthrough_refiner import (
|
||||
PassthroughRefiner,
|
||||
create,
|
||||
)
|
||||
from gps_denied_onboard.config.schema import Config, ConfigError
|
||||
from gps_denied_onboard.runtime_root.refiner_factory import build_refiner_strategy
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# Fakes.
|
||||
|
||||
|
||||
class _FakeRansacFilter:
|
||||
def filter(self, *args, **kwargs):
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class _FakeInferenceRuntime:
|
||||
def deserialize_engine(self, *args, **kwargs):
|
||||
raise NotImplementedError
|
||||
|
||||
def thermal_state(self):
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class _FakeFrame:
|
||||
frame_id = 7
|
||||
|
||||
|
||||
class _PartialRefinerMissingWasInvoked:
|
||||
def refine_if_needed(self, frame, mr, residual_threshold_px):
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class _PartialRefinerMissingRefine:
|
||||
def was_invoked(self):
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
def _config_with_strategy(
|
||||
strategy: str = "passthrough",
|
||||
*,
|
||||
threshold: float = 2.5,
|
||||
) -> Config:
|
||||
return Config.with_blocks(
|
||||
c3_5_adhop=C3_5RefinerConfig(
|
||||
strategy=strategy,
|
||||
residual_threshold_px=threshold,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _make_candidate(*, inliers: int = 40, residual: float = 1.0) -> CandidateMatchSet:
|
||||
return CandidateMatchSet(
|
||||
tile_id=(18, 49.9, 36.3),
|
||||
inlier_count=inliers,
|
||||
inlier_correspondences=np.ones((inliers, 4), dtype=np.float32) * 0.5,
|
||||
ransac_outlier_count=3,
|
||||
per_candidate_residual_px=residual,
|
||||
)
|
||||
|
||||
|
||||
def _make_result(
|
||||
*,
|
||||
reprojection_residual: float = 1.0,
|
||||
refinement_label: str = "passthrough",
|
||||
refinement_latency_ms: float = 0.0,
|
||||
) -> MatchResult:
|
||||
candidate = _make_candidate()
|
||||
return MatchResult(
|
||||
frame_id=7,
|
||||
per_candidate=(candidate,),
|
||||
best_candidate_idx=0,
|
||||
reprojection_residual_px=reprojection_residual,
|
||||
matched_at=1_000_000_000,
|
||||
matcher_label="disk_lightglue",
|
||||
candidates_input=3,
|
||||
candidates_dropped=2,
|
||||
refinement_label=refinement_label,
|
||||
refinement_added_latency_ms=refinement_latency_ms,
|
||||
)
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# AC-1: Protocol conformance.
|
||||
|
||||
|
||||
def test_ac1_passthrough_refiner_conformance() -> None:
|
||||
refiner = PassthroughRefiner(
|
||||
ransac_filter=_FakeRansacFilter(),
|
||||
inference_runtime=_FakeInferenceRuntime(),
|
||||
)
|
||||
assert isinstance(refiner, ConditionalRefiner)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"partial_cls",
|
||||
[_PartialRefinerMissingWasInvoked, _PartialRefinerMissingRefine],
|
||||
)
|
||||
def test_ac1_partial_refiners_fail_conformance(partial_cls) -> None:
|
||||
assert not isinstance(partial_cls(), ConditionalRefiner)
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# AC-2: MatchResult backward-compatible defaults.
|
||||
|
||||
|
||||
def test_ac2_match_result_construction_without_new_fields() -> None:
|
||||
candidate = _make_candidate()
|
||||
mr = MatchResult(
|
||||
frame_id=7,
|
||||
per_candidate=(candidate,),
|
||||
best_candidate_idx=0,
|
||||
reprojection_residual_px=candidate.per_candidate_residual_px,
|
||||
matched_at=1_000_000_000,
|
||||
matcher_label="disk_lightglue",
|
||||
candidates_input=3,
|
||||
candidates_dropped=2,
|
||||
)
|
||||
assert mr.refinement_label == "passthrough"
|
||||
assert mr.refinement_added_latency_ms == 0.0
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# AC-3: MatchResult immutability + slots preserved.
|
||||
|
||||
|
||||
def test_ac3_match_result_slots_include_new_fields() -> None:
|
||||
assert "refinement_label" in MatchResult.__slots__
|
||||
assert "refinement_added_latency_ms" in MatchResult.__slots__
|
||||
|
||||
|
||||
def test_ac3_match_result_remains_frozen() -> None:
|
||||
mr = _make_result()
|
||||
with pytest.raises(dataclasses.FrozenInstanceError):
|
||||
mr.refinement_label = "adhop"
|
||||
with pytest.raises(dataclasses.FrozenInstanceError):
|
||||
mr.refinement_added_latency_ms = 12.5
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# AC-4: Factory rejects unknown strategy. (Validated at config-load.)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"bad_label",
|
||||
["ADHOP", "passthrough_v2", "garbage", "", "default"],
|
||||
)
|
||||
def test_ac4_unknown_strategy_rejected_at_config_load(bad_label: str) -> None:
|
||||
with pytest.raises(ConfigError) as exc_info:
|
||||
C3_5RefinerConfig(strategy=bad_label)
|
||||
msg = str(exc_info.value)
|
||||
for valid in KNOWN_STRATEGIES:
|
||||
assert valid in msg
|
||||
|
||||
|
||||
def test_ac4_factory_emits_error_log_on_unknown_strategy(caplog) -> None:
|
||||
# Build a Config that bypasses __post_init__ validation by
|
||||
# constructing the block via dataclasses.replace on an already-valid
|
||||
# block — this is what catches the defensive factory path.
|
||||
valid_block = C3_5RefinerConfig(strategy="passthrough")
|
||||
object.__setattr__(valid_block, "strategy", "garbage")
|
||||
config = Config.with_blocks(c3_5_adhop=valid_block)
|
||||
with caplog.at_level(logging.ERROR, logger="gps_denied_onboard.c3_5_adhop"):
|
||||
with pytest.raises(RefinerConfigError) as exc_info:
|
||||
build_refiner_strategy(
|
||||
config,
|
||||
ransac_filter=_FakeRansacFilter(),
|
||||
inference_runtime=_FakeInferenceRuntime(),
|
||||
)
|
||||
assert "Unknown refiner strategy: garbage" in str(exc_info.value)
|
||||
assert any(
|
||||
r.message == "c3_5.refiner.strategy_unknown" for r in caplog.records
|
||||
)
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# AC-5: Factory rejects invalid threshold.
|
||||
|
||||
|
||||
@pytest.mark.parametrize("bad_threshold", [0.0, -0.1, -10.0])
|
||||
def test_ac5_invalid_threshold_rejected_at_config_load(bad_threshold: float) -> None:
|
||||
with pytest.raises(ConfigError):
|
||||
C3_5RefinerConfig(residual_threshold_px=bad_threshold)
|
||||
|
||||
|
||||
def test_ac5_factory_emits_error_log_on_invalid_threshold(caplog) -> None:
|
||||
valid_block = C3_5RefinerConfig(strategy="passthrough")
|
||||
object.__setattr__(valid_block, "residual_threshold_px", 0.0)
|
||||
config = Config.with_blocks(c3_5_adhop=valid_block)
|
||||
with caplog.at_level(logging.ERROR, logger="gps_denied_onboard.c3_5_adhop"):
|
||||
with pytest.raises(RefinerConfigError):
|
||||
build_refiner_strategy(
|
||||
config,
|
||||
ransac_filter=_FakeRansacFilter(),
|
||||
inference_runtime=_FakeInferenceRuntime(),
|
||||
)
|
||||
assert any(
|
||||
r.message == "c3_5.refiner.invalid_threshold" for r in caplog.records
|
||||
)
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# AC-6: Successful factory load emits INFO log.
|
||||
|
||||
|
||||
def test_ac6_factory_emits_info_log_on_success(caplog) -> None:
|
||||
config = _config_with_strategy("passthrough", threshold=2.5)
|
||||
with caplog.at_level(logging.INFO, logger="gps_denied_onboard.c3_5_adhop"):
|
||||
instance = build_refiner_strategy(
|
||||
config,
|
||||
ransac_filter=_FakeRansacFilter(),
|
||||
inference_runtime=_FakeInferenceRuntime(),
|
||||
)
|
||||
assert isinstance(instance, ConditionalRefiner)
|
||||
records = [
|
||||
r for r in caplog.records if r.message == "c3_5.refiner.strategy_loaded"
|
||||
]
|
||||
assert len(records) == 1
|
||||
record = records[0]
|
||||
assert getattr(record, "strategy", None) == "passthrough"
|
||||
assert getattr(record, "residual_threshold_px", None) == 2.5
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# AC-7: Strategy resolution table.
|
||||
|
||||
|
||||
def test_ac7_passthrough_resolution() -> None:
|
||||
config = _config_with_strategy("passthrough")
|
||||
instance = build_refiner_strategy(
|
||||
config,
|
||||
ransac_filter=_FakeRansacFilter(),
|
||||
inference_runtime=_FakeInferenceRuntime(),
|
||||
)
|
||||
assert isinstance(instance, PassthroughRefiner)
|
||||
assert isinstance(instance, ConditionalRefiner)
|
||||
assert (
|
||||
"gps_denied_onboard.components.c3_5_adhop.passthrough_refiner"
|
||||
in sys.modules
|
||||
)
|
||||
|
||||
|
||||
def test_ac7_adhop_resolution_stops_at_module_lookup() -> None:
|
||||
"""Task spec: the full-success path for "adhop" belongs to the
|
||||
AdHoP task (AZ-349); this assertion verifies the factory reaches
|
||||
the import step but the module does not exist yet.
|
||||
"""
|
||||
config = _config_with_strategy("adhop")
|
||||
with pytest.raises(ModuleNotFoundError):
|
||||
build_refiner_strategy(
|
||||
config,
|
||||
ransac_filter=_FakeRansacFilter(),
|
||||
inference_runtime=_FakeInferenceRuntime(),
|
||||
)
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# AC-8: Public API.
|
||||
|
||||
|
||||
def test_ac8_public_api_re_exports() -> None:
|
||||
from gps_denied_onboard.components import c3_5_adhop
|
||||
|
||||
assert "ConditionalRefiner" in c3_5_adhop.__all__
|
||||
assert "C3_5RefinerConfig" in c3_5_adhop.__all__
|
||||
|
||||
|
||||
def test_ac8_internals_not_in_public_api() -> None:
|
||||
from gps_denied_onboard.components import c3_5_adhop
|
||||
|
||||
for internal in (
|
||||
"PassthroughRefiner",
|
||||
"AdHoPRefiner",
|
||||
"RefinerError",
|
||||
"RefinerBackboneError",
|
||||
"RefinerConfigError",
|
||||
):
|
||||
assert internal not in c3_5_adhop.__all__, (
|
||||
f"internal name leaked into public API: {internal}"
|
||||
)
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# AC-9: deferred.
|
||||
|
||||
|
||||
def test_ac9_single_thread_binding_deferred() -> None:
|
||||
"""AC-9 (single-thread binding) is deferred per task spec
|
||||
Risk 4: the generic ``compose_root`` thread-binding registry
|
||||
lives with AZ-270 and the broader runtime-root composition.
|
||||
Each factory owns its own thread binding today; this protocol
|
||||
task does not add a new binding registry."""
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# AC-10: PassthroughRefiner byte-identical correspondences.
|
||||
|
||||
|
||||
def test_ac10_passthrough_returns_same_match_result_reference() -> None:
|
||||
refiner = create(
|
||||
_config_with_strategy(),
|
||||
ransac_filter=_FakeRansacFilter(),
|
||||
inference_runtime=_FakeInferenceRuntime(),
|
||||
)
|
||||
mr = _make_result()
|
||||
out = refiner.refine_if_needed(_FakeFrame(), mr, residual_threshold_px=2.5)
|
||||
assert out is mr
|
||||
assert out.refinement_label == "passthrough"
|
||||
assert out.refinement_added_latency_ms == 0.0
|
||||
|
||||
|
||||
def test_ac10_passthrough_correspondences_bit_identical() -> None:
|
||||
refiner = PassthroughRefiner(
|
||||
ransac_filter=_FakeRansacFilter(),
|
||||
inference_runtime=_FakeInferenceRuntime(),
|
||||
)
|
||||
mr = _make_result()
|
||||
out = refiner.refine_if_needed(_FakeFrame(), mr, residual_threshold_px=2.5)
|
||||
for in_cand, out_cand in zip(mr.per_candidate, out.per_candidate, strict=True):
|
||||
assert np.array_equal(
|
||||
out_cand.inlier_correspondences, in_cand.inlier_correspondences
|
||||
)
|
||||
assert out_cand.inlier_correspondences.dtype == in_cand.inlier_correspondences.dtype
|
||||
# Same object reference — INV-5 forbids silent copies on the
|
||||
# passthrough path.
|
||||
assert out_cand.inlier_correspondences is in_cand.inlier_correspondences
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# AC-11: was_invoked always False on PassthroughRefiner.
|
||||
|
||||
|
||||
def test_ac11_was_invoked_always_false_on_passthrough() -> None:
|
||||
refiner = PassthroughRefiner(
|
||||
ransac_filter=_FakeRansacFilter(),
|
||||
inference_runtime=_FakeInferenceRuntime(),
|
||||
)
|
||||
assert refiner.was_invoked() is False
|
||||
mr = _make_result()
|
||||
for _ in range(10):
|
||||
refiner.refine_if_needed(_FakeFrame(), mr, residual_threshold_px=2.5)
|
||||
assert refiner.was_invoked() is False
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# AC-12: Threshold validation in refine_if_needed.
|
||||
|
||||
|
||||
@pytest.mark.parametrize("bad_threshold", [0.0, -0.1, -50.0])
|
||||
def test_ac12_threshold_validation_in_method(bad_threshold: float) -> None:
|
||||
refiner = PassthroughRefiner(
|
||||
ransac_filter=_FakeRansacFilter(),
|
||||
inference_runtime=_FakeInferenceRuntime(),
|
||||
)
|
||||
mr = _make_result()
|
||||
with pytest.raises(ValueError):
|
||||
refiner.refine_if_needed(
|
||||
_FakeFrame(), mr, residual_threshold_px=bad_threshold
|
||||
)
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# AC-13: Error hierarchy.
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"exc_factory",
|
||||
[RefinerBackboneError, RefinerConfigError],
|
||||
)
|
||||
def test_ac13_all_refiner_errors_caught_as_family(exc_factory) -> None:
|
||||
with pytest.raises(RefinerError):
|
||||
raise exc_factory("boom")
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# AC-14: module-layout.md symbol rename.
|
||||
|
||||
|
||||
def test_ac14_module_layout_references_conditional_refiner() -> None:
|
||||
from pathlib import Path
|
||||
|
||||
doc = Path("_docs/02_document/module-layout.md").read_text()
|
||||
assert "ConditionalRefiner" in doc, (
|
||||
"module-layout.md must reference the canonical Public API symbol"
|
||||
)
|
||||
# Old symbol name must NOT be present (per AC-14).
|
||||
assert "AdHoPRefinementStrategy" not in doc, (
|
||||
"module-layout.md still references the legacy AdHoPRefinementStrategy "
|
||||
"symbol; rename per AZ-348 AC-14"
|
||||
)
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# NFRs.
|
||||
|
||||
|
||||
def test_nfr_perf_factory_under_20ms_p99(caplog) -> None:
|
||||
config = _config_with_strategy("passthrough")
|
||||
ransac_filter = _FakeRansacFilter()
|
||||
inference_runtime = _FakeInferenceRuntime()
|
||||
durations_ms: list[float] = []
|
||||
for _ in range(100):
|
||||
t0 = time.perf_counter()
|
||||
build_refiner_strategy(
|
||||
config,
|
||||
ransac_filter=ransac_filter,
|
||||
inference_runtime=inference_runtime,
|
||||
)
|
||||
durations_ms.append((time.perf_counter() - t0) * 1000.0)
|
||||
durations_ms.sort()
|
||||
p99 = durations_ms[int(0.99 * len(durations_ms))]
|
||||
assert p99 <= 20.0, f"factory p99={p99:.2f} ms exceeded 20 ms budget"
|
||||
|
||||
|
||||
def test_nfr_perf_passthrough_under_0_5ms_p99() -> None:
|
||||
refiner = PassthroughRefiner(
|
||||
ransac_filter=_FakeRansacFilter(),
|
||||
inference_runtime=_FakeInferenceRuntime(),
|
||||
)
|
||||
mr = _make_result()
|
||||
frame = _FakeFrame()
|
||||
durations_us: list[float] = []
|
||||
for _ in range(10_000):
|
||||
t0 = time.perf_counter()
|
||||
refiner.refine_if_needed(frame, mr, residual_threshold_px=2.5)
|
||||
durations_us.append((time.perf_counter() - t0) * 1_000_000.0)
|
||||
durations_us.sort()
|
||||
p99_us = durations_us[int(0.99 * len(durations_us))]
|
||||
assert p99_us <= 500.0, f"refine p99={p99_us:.2f} us exceeded 500 us budget"
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# Surface coverage — config defaults.
|
||||
|
||||
|
||||
def test_c3_5_config_defaults() -> None:
|
||||
cfg = C3_5RefinerConfig()
|
||||
assert cfg.strategy == "adhop"
|
||||
assert cfg.residual_threshold_px == 2.5
|
||||
assert cfg.invocation_rate_warn_threshold == 0.25
|
||||
|
||||
|
||||
def test_c3_5_config_invocation_rate_validation() -> None:
|
||||
with pytest.raises(ConfigError):
|
||||
C3_5RefinerConfig(invocation_rate_warn_threshold=0.0)
|
||||
with pytest.raises(ConfigError):
|
||||
C3_5RefinerConfig(invocation_rate_warn_threshold=1.0)
|
||||
with pytest.raises(ConfigError):
|
||||
C3_5RefinerConfig(invocation_rate_warn_threshold=-0.5)
|
||||
@@ -1,8 +0,0 @@
|
||||
"""C3.5 AdHoP smoke test — AC-9."""
|
||||
|
||||
|
||||
def test_interface_importable() -> None:
|
||||
# Assert
|
||||
from gps_denied_onboard.components.c3_5_adhop import AdHoPRefinementStrategy
|
||||
|
||||
assert AdHoPRefinementStrategy is not None
|
||||
Reference in New Issue
Block a user