[AZ-918] [AZ-919] [AZ-920] [AZ-921] [AZ-922] VIO/ESKF baseline fixes

Derkachi e2e Tier-2 divergence had three stacked root causes; this
commit ships fixes for all three plus the IMU prerequisite they
depend on, plus a baseline cheirality gate for cv2.recoverPose.

AZ-918  MAVLink IMU adapters now convert raw mG/mrad-s + FRD body to
        SI m/s^2 + rad/s + FLU body via helpers.imu_units. Without
        this the ESKF receives values ~1000x too small with wrong-
        sign Y/Z and cannot function at all.

AZ-919  Composition root wires EskfNominalAltitudeProvider into the
        KLT/RANSAC strategy via the AZ-331 factory introspect path;
        OKVIS2 and VINS-Mono are unaffected.

AZ-920  KLT/RANSAC recovers metric translation via Ground Sampling
        Distance when AGL is available; otherwise falls through with
        scale_quality=direction_only/unknown (no fake scale invented).

AZ-921  VioOutput.scale_quality signal; ESKF add_vio adapts R_meas
        position block based on the flag (1e6 inflation when scale is
        direction_only/unknown to keep the filter consistent).

AZ-922  KLT/RANSAC cheirality gate rejects single-frame rotations
        beyond a config threshold (default 30 deg), catching
        cv2.recoverPose twisted-pair flips that cause immediate ESKF
        divergence on low-parallax aerial scenes.

Verification:
- Tier-1 (macOS) unit suite: 2346 passed, 0 failed.
- Tier-2 (Jetson) Derkachi e2e: divergence moves from frame 5
  (mahalanobis^2 3757) to frame 233 (mahalanobis^2 212). Remaining
  drift is open-loop attitude accumulation, not cheirality.

Follow-up tickets filed:
- AZ-923 closed as misdiagnosed: EskfNominalAltitudeProvider was
  already correct (nominal_pos.z IS the AGL when takeoff origin sits
  at ground level); the early-frame AGL near zero reflects the drone
  being stationary on the ground, not a provider bug.
- AZ-942 filed: cross-check VIO rotation against IMU preintegrator
  (consistency gate) - more physically grounded than the coarse
  AZ-922 threshold and likely required to absorb the frame-233 drift.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Oleksandr Bezdieniezhnykh
2026-05-27 22:28:40 +03:00
parent 38170b3499
commit 94d2358c8b
32 changed files with 2320 additions and 82 deletions
@@ -0,0 +1,70 @@
"""AZ-919 — `KltRansacStrategy` accepts the `altitude_provider` kwarg.
AZ-919 is plumbing only: the strategy must accept and store the provider
without yet consuming it. The math that consumes it lands in AZ-920;
the degraded-mode behaviour lands in AZ-921. Until then, the strategy
MUST behave identically to today when ``altitude_provider`` is ``None``
(AC-3 bug-for-bug compatibility).
"""
from __future__ import annotations
from gps_denied_onboard.components.c1_vio import C1VioConfig, KltRansacConfig
from gps_denied_onboard.components.c1_vio.klt_ransac import KltRansacStrategy
from gps_denied_onboard.config.schema import Config, RuntimeConfig
from gps_denied_onboard.fdr_client.client import FdrClient
from gps_denied_onboard.helpers.altitude_provider import (
AltitudeProvider,
EskfNominalAltitudeProvider,
)
def _config() -> Config:
c1 = C1VioConfig(
strategy="klt_ransac",
klt_ransac=KltRansacConfig(),
lost_frame_threshold=3,
)
return Config.with_blocks(c1_vio=c1, runtime=RuntimeConfig())
def _fdr_client() -> FdrClient:
return FdrClient(producer_id="test.az919", capacity=8, _emit_diag_log=False)
def test_default_altitude_provider_is_none() -> None:
strategy = KltRansacStrategy(_config(), fdr_client=_fdr_client())
assert strategy._altitude_provider is None
def test_explicit_altitude_provider_is_stored() -> None:
provider: AltitudeProvider = EskfNominalAltitudeProvider(
estimator_supplier=lambda: None,
)
strategy = KltRansacStrategy(
_config(),
fdr_client=_fdr_client(),
altitude_provider=provider,
)
assert strategy._altitude_provider is provider
def test_altitude_provider_is_keyword_only() -> None:
# Arrange — positional pass of altitude_provider must fail because
# the kwarg is declared after ``*``. This pins the calling contract
# so future refactors cannot accidentally make it positional.
fdr_client = _fdr_client()
provider = EskfNominalAltitudeProvider(estimator_supplier=lambda: None)
# Act + Assert
try:
KltRansacStrategy(_config(), fdr_client, provider) # type: ignore[misc]
except TypeError:
return
raise AssertionError(
"KltRansacStrategy accepted altitude_provider positionally; "
"AZ-919 AC-3 requires the kwarg to be keyword-only."
)
+254
View File
@@ -0,0 +1,254 @@
"""AZ-920 — `recover_scale` pure-function tests.
Pins the GSD scale-recovery contract that the KLT/RANSAC integration
in ``KltRansacStrategy.process_frame`` relies on. The integration
test is in ``test_az920_klt_ransac_scale_integration.py``; this file
covers the math in isolation.
"""
from __future__ import annotations
import math
import numpy as np
import pytest
from gps_denied_onboard.components.c1_vio._gsd_scale import (
INFINITE_SCALE_UNCERTAINTY_M,
RELATIVE_DISPLACEMENT_NOISE,
ScaleRecoveryResult,
recover_scale,
)
# ----------------------------------------------------------------------
# Helpers
def _intrinsics(fx: float = 1680.4469) -> np.ndarray:
"""Derkachi-grade nadir intrinsics by default (fx=fy, cx/cy at centre)."""
return np.array(
[
[fx, 0.0, 960.0],
[0.0, fx, 540.0],
[0.0, 0.0, 1.0],
],
dtype=np.float64,
)
def _uniform_flow(n: int, dx_px: float, dy_px: float = 0.0) -> tuple[np.ndarray, np.ndarray]:
"""Build Nx2 prev / curr point clouds with a uniform pixel-flow shift."""
rng = np.random.default_rng(seed=2026)
base = rng.uniform(100.0, 1800.0, size=(n, 2))
prev = base.copy()
curr = base + np.array([dx_px, dy_px], dtype=np.float64)
return prev, curr
# ----------------------------------------------------------------------
# Happy path
def test_nadir_identity_rotation_with_known_gsd_yields_expected_scale() -> None:
# Arrange — at AGL=100 m, fx=1680 px → GSD ≈ 0.0595 m/px. A 20 px
# mean flow corresponds to ≈1.19 m of horizontal motion. t_unit
# along +X means t_horizontal = 1; scale_factor must equal 1.19.
t_unit = np.array([1.0, 0.0, 0.0], dtype=np.float64)
pts_prev, pts_curr = _uniform_flow(n=64, dx_px=20.0)
# Act
result = recover_scale(
t_unit=t_unit,
pts_prev=pts_prev,
pts_curr=pts_curr,
intrinsics_3x3=_intrinsics(),
agl_m=100.0,
)
# Assert
assert isinstance(result, ScaleRecoveryResult)
expected_disp_m = 20.0 * 100.0 / 1680.4469
assert result.scale_factor == pytest.approx(expected_disp_m, rel=1e-9)
assert result.scale_uncertainty_m == pytest.approx(
expected_disp_m * RELATIVE_DISPLACEMENT_NOISE, rel=1e-9
)
assert result.is_recoverable()
# AZ-921 — happy-path categorisation is "metric".
assert result.quality == "metric"
def test_diagonal_flow_uses_l2_norm_for_mean_magnitude() -> None:
# Arrange — (dx=3, dy=4) flow has L2 magnitude 5 px per feature.
t_unit = np.array([1.0, 0.0, 0.0], dtype=np.float64)
pts_prev, pts_curr = _uniform_flow(n=50, dx_px=3.0, dy_px=4.0)
# Act
result = recover_scale(
t_unit=t_unit,
pts_prev=pts_prev,
pts_curr=pts_curr,
intrinsics_3x3=_intrinsics(),
agl_m=100.0,
)
# Assert
expected_disp_m = 5.0 * 100.0 / 1680.4469
assert result.scale_factor == pytest.approx(expected_disp_m, rel=1e-9)
def test_off_axis_t_unit_uses_horizontal_component_magnitude() -> None:
# Arrange — t_unit splits 0.6/0.8 between camera-X and camera-Y;
# the horizontal magnitude is sqrt(0.36 + 0.64) = 1.0. The scale
# therefore stays the displacement / 1.0.
t_unit = np.array([0.6, 0.8, 0.0], dtype=np.float64)
pts_prev, pts_curr = _uniform_flow(n=40, dx_px=10.0)
# Act
result = recover_scale(
t_unit=t_unit,
pts_prev=pts_prev,
pts_curr=pts_curr,
intrinsics_3x3=_intrinsics(),
agl_m=120.0,
)
# Assert
expected_disp_m = 10.0 * 120.0 / 1680.4469
assert result.scale_factor == pytest.approx(expected_disp_m, rel=1e-9)
# ----------------------------------------------------------------------
# Degenerate / fallback paths — each must produce inf uncertainty.
def test_agl_none_returns_infinite_uncertainty() -> None:
pts_prev, pts_curr = _uniform_flow(n=32, dx_px=20.0)
result = recover_scale(
t_unit=np.array([1.0, 0.0, 0.0]),
pts_prev=pts_prev,
pts_curr=pts_curr,
intrinsics_3x3=_intrinsics(),
agl_m=None,
)
assert result.scale_factor == 0.0
assert result.scale_uncertainty_m == INFINITE_SCALE_UNCERTAINTY_M
assert not result.is_recoverable()
# AZ-921 — AGL-unknown is "unknown", not "direction_only".
assert result.quality == "unknown"
def test_zero_or_negative_agl_returns_infinite_uncertainty() -> None:
pts_prev, pts_curr = _uniform_flow(n=32, dx_px=20.0)
for bad_agl in (0.0, -1.0):
result = recover_scale(
t_unit=np.array([1.0, 0.0, 0.0]),
pts_prev=pts_prev,
pts_curr=pts_curr,
intrinsics_3x3=_intrinsics(),
agl_m=bad_agl,
)
assert math.isinf(result.scale_uncertainty_m), f"agl={bad_agl}"
def test_near_pure_vertical_t_unit_returns_infinite_uncertainty() -> None:
# Arrange — t along the optical axis (camera-Z) only; the GSD
# method cannot tie horizontal pixel flow to vertical motion.
pts_prev, pts_curr = _uniform_flow(n=40, dx_px=5.0)
result = recover_scale(
t_unit=np.array([0.01, 0.01, 0.9998]),
pts_prev=pts_prev,
pts_curr=pts_curr,
intrinsics_3x3=_intrinsics(),
agl_m=100.0,
)
assert not result.is_recoverable()
# AZ-921 — near-vertical motion routes to direction_only: the
# direction of t_unit is still meaningful, only the magnitude is
# not derivable from horizontal pixel flow.
assert result.quality == "direction_only"
def test_zero_inlier_flow_returns_infinite_uncertainty() -> None:
# Arrange — stationary inliers (dx=dy=0). Cannot disambiguate the
# scale of a zero motion.
pts_prev, pts_curr = _uniform_flow(n=40, dx_px=0.0, dy_px=0.0)
result = recover_scale(
t_unit=np.array([1.0, 0.0, 0.0]),
pts_prev=pts_prev,
pts_curr=pts_curr,
intrinsics_3x3=_intrinsics(),
agl_m=100.0,
)
assert not result.is_recoverable()
# AZ-921 — stale features routes to "unknown", since neither the
# magnitude nor the direction of motion is trustworthy.
assert result.quality == "unknown"
def test_empty_inlier_set_returns_infinite_uncertainty() -> None:
empty = np.zeros((0, 2), dtype=np.float64)
result = recover_scale(
t_unit=np.array([1.0, 0.0, 0.0]),
pts_prev=empty,
pts_curr=empty,
intrinsics_3x3=_intrinsics(),
agl_m=100.0,
)
assert not result.is_recoverable()
def test_zero_focal_length_returns_infinite_uncertainty() -> None:
pts_prev, pts_curr = _uniform_flow(n=32, dx_px=20.0)
result = recover_scale(
t_unit=np.array([1.0, 0.0, 0.0]),
pts_prev=pts_prev,
pts_curr=pts_curr,
intrinsics_3x3=_intrinsics(fx=0.0),
agl_m=100.0,
)
assert not result.is_recoverable()
# ----------------------------------------------------------------------
# Numerical stability with a Derkachi-grade inlier set
def test_derkachi_grade_inlier_count_yields_stable_scale() -> None:
# Arrange — 120 inliers (representative of the Derkachi cruise
# observation) with a small per-feature noise around a uniform
# 30 px optical flow at 150 m AGL.
rng = np.random.default_rng(seed=271828)
base = rng.uniform(100.0, 1800.0, size=(120, 2))
flow_mean_px = 30.0
noise = rng.normal(0.0, 0.5, size=(120, 2))
pts_prev = base
pts_curr = base + np.array([flow_mean_px, 0.0]) + noise
t_unit = np.array([1.0, 0.0, 0.0])
agl_m = 150.0
# Act
result = recover_scale(
t_unit=t_unit,
pts_prev=pts_prev,
pts_curr=pts_curr,
intrinsics_3x3=_intrinsics(),
agl_m=agl_m,
)
# Assert — observed scale must be within 5 % of the analytic value
# at this noise level (sigma = 0.5 px over 120 inliers).
expected = flow_mean_px * agl_m / 1680.4469
assert result.scale_factor == pytest.approx(expected, rel=5e-2)
assert result.is_recoverable()
@@ -0,0 +1,319 @@
"""AZ-920 — `KltRansacStrategy` x `AltitudeProvider` integration coverage.
Pins the contract that ``process_frame`` consumes the
:class:`AltitudeProvider` to recover metric scale on the
otherwise-unit translation that ``cv2.recoverPose`` emits. Pure
function math is covered in ``test_az920_gsd_scale.py``; this file
covers the wiring at the strategy boundary.
"""
from __future__ import annotations
from datetime import datetime, timezone
from typing import Any
import numpy as np
import pytest
from gps_denied_onboard._types.calibration import CameraCalibration
from gps_denied_onboard._types.nav import (
ImuSample,
ImuWindow,
NavCameraFrame,
)
from gps_denied_onboard.components.c1_vio import C1VioConfig, KltRansacConfig
from gps_denied_onboard.components.c1_vio import klt_ransac as klt_ransac_module
from gps_denied_onboard.components.c1_vio._gsd_scale import RELATIVE_DISPLACEMENT_NOISE
from gps_denied_onboard.components.c1_vio.klt_ransac import (
_MIN_POSITION_VARIANCE_M2,
KltRansacStrategy,
)
from gps_denied_onboard.config.schema import Config, RuntimeConfig
from gps_denied_onboard.fdr_client.client import FdrClient
from gps_denied_onboard.helpers.altitude_provider import AltitudeProvider
from gps_denied_onboard.helpers.ransac_filter import RansacResult
# ----------------------------------------------------------------------
# Test scaffolding
class _ConstantAgl(AltitudeProvider):
"""Returns a fixed AGL on every call."""
def __init__(self, agl_m: float | None) -> None:
self._agl_m = agl_m
def agl_m(self, now_ns: int) -> float | None:
return self._agl_m
def _calibration(focal_length: float = 1680.0) -> CameraCalibration:
return CameraCalibration(
camera_id="khp20s30-test",
intrinsics_3x3=np.array(
[
[focal_length, 0.0, 960.0],
[0.0, focal_length, 540.0],
[0.0, 0.0, 1.0],
],
dtype=np.float64,
),
distortion=np.zeros(5, dtype=np.float64),
body_to_camera_se3=np.eye(4, dtype=np.float64),
acquisition_method="test_az920",
)
def _frame(idx: int) -> NavCameraFrame:
rng = np.random.default_rng(seed=idx)
image = (rng.integers(0, 255, size=(240, 240), dtype=np.int16)).astype(np.uint8)
return NavCameraFrame(
frame_id=idx,
timestamp=datetime.fromtimestamp(idx * 0.1, tz=timezone.utc),
image=image,
camera_calibration_id="khp20s30-test",
)
def _imu_window(frame_idx: int) -> ImuWindow:
ts_start_ns = 1_000_000_000 + frame_idx * 100_000_000
samples = tuple(
ImuSample(
ts_ns=ts_start_ns + i * 5_000_000,
accel_xyz=(0.0, 0.0, 9.81),
gyro_xyz=(0.0, 0.0, 0.0),
)
for i in range(3)
)
return ImuWindow(
samples=samples,
ts_start_ns=samples[0].ts_ns,
ts_end_ns=samples[-1].ts_ns,
)
def _config() -> Config:
return Config.with_blocks(
c1_vio=C1VioConfig(
strategy="klt_ransac",
klt_ransac=KltRansacConfig(),
lost_frame_threshold=3,
),
runtime=RuntimeConfig(),
)
def _patch_pose_recovery(
monkeypatch: pytest.MonkeyPatch,
*,
flow_px: float,
inlier_count: int = 40,
t_unit: tuple[float, float, float] = (1.0, 0.0, 0.0),
) -> None:
"""Force the geometry stack to a controlled success path.
Builds an inlier set with a uniform ``flow_px`` pixel shift in the
+x direction; ``cv2.findEssentialMat`` returns identity rotation +
an all-inlier mask; ``cv2.recoverPose`` returns identity rotation +
a unit-length translation in ``t_unit``.
"""
rng = np.random.default_rng(seed=314159)
base = rng.uniform(50.0, 1800.0, size=(inlier_count, 2))
inliers = np.column_stack([base, base + np.array([flow_px, 0.0])])
mask = np.ones((inlier_count, 1), dtype=np.uint8)
def _fake_filter(_corr: np.ndarray, _thresh: float, _min: int) -> RansacResult:
return RansacResult(
inlier_correspondences=inliers,
inlier_count=inlier_count,
outlier_count=0,
median_residual_px=0.5,
)
def _fake_find_essential(*_a: Any, **_k: Any) -> tuple[np.ndarray, np.ndarray]:
return np.eye(3, dtype=np.float64), mask
def _fake_recover_pose(*_a: Any, **_k: Any) -> tuple[int, np.ndarray, np.ndarray, np.ndarray]:
R = np.eye(3, dtype=np.float64)
t_col = np.asarray(t_unit, dtype=np.float64).reshape(3, 1)
return inlier_count, R, t_col, mask
monkeypatch.setattr(
klt_ransac_module.RansacFilter,
"filter_correspondences",
staticmethod(_fake_filter),
)
monkeypatch.setattr(klt_ransac_module.cv2, "findEssentialMat", _fake_find_essential)
monkeypatch.setattr(klt_ransac_module.cv2, "recoverPose", _fake_recover_pose)
def _fdr_client() -> FdrClient:
return FdrClient(producer_id="test.az920", capacity=16, _emit_diag_log=False)
# ----------------------------------------------------------------------
# AC-4 — translation gets metric scale when the provider yields valid AGL
def test_steady_state_frame_translation_is_metric_when_agl_known(
monkeypatch: pytest.MonkeyPatch,
) -> None:
# Arrange — 20 px mean flow at 100 m AGL, fx=1680.
# Expected scale_factor = 20 * 100 / 1680 ≈ 1.190 m. The unit t is
# along +X, so the emitted translation is ≈ (1.19, 0, 0) m.
_patch_pose_recovery(monkeypatch, flow_px=20.0)
fdr_client = _fdr_client()
provider = _ConstantAgl(agl_m=100.0)
strategy = KltRansacStrategy(
_config(), fdr_client=fdr_client, altitude_provider=provider
)
calibration = _calibration(focal_length=1680.0)
# Act — first frame: INIT (identity pose). Second frame: scale recovery fires.
strategy.process_frame(_frame(idx=1), _imu_window(1), calibration)
second = strategy.process_frame(_frame(idx=2), _imu_window(2), calibration)
# Assert
pose_matrix = np.asarray(second.relative_pose_T.matrix())
translation = pose_matrix[:3, 3]
expected_x = 20.0 * 100.0 / 1680.0
assert translation[0] == pytest.approx(expected_x, rel=1e-3)
# Y / Z components are zero because t_unit was pure +X.
assert translation[1] == pytest.approx(0.0, abs=1e-9)
assert translation[2] == pytest.approx(0.0, abs=1e-9)
# Translation magnitude is metric, not unit-length.
assert float(np.linalg.norm(translation)) > 0.5 # metres, not 1.0 px-scale
# AZ-921 — scale_quality must report "metric" on the happy path.
assert second.scale_quality == "metric"
# ----------------------------------------------------------------------
# AC-4 — position covariance is in m² when AGL is known
def test_position_covariance_block_is_metric_when_agl_known(
monkeypatch: pytest.MonkeyPatch,
) -> None:
# Arrange
_patch_pose_recovery(monkeypatch, flow_px=20.0)
provider = _ConstantAgl(agl_m=100.0)
strategy = KltRansacStrategy(
_config(), fdr_client=_fdr_client(), altitude_provider=provider
)
calibration = _calibration(focal_length=1680.0)
# Act
strategy.process_frame(_frame(idx=1), _imu_window(1), calibration)
second = strategy.process_frame(_frame(idx=2), _imu_window(2), calibration)
# Assert
cov = np.asarray(second.pose_covariance_6x6)
expected_disp_m = 20.0 * 100.0 / 1680.0
expected_pos_var = max(
(expected_disp_m * RELATIVE_DISPLACEMENT_NOISE) ** 2,
_MIN_POSITION_VARIANCE_M2,
)
pos_block = cov[0:3, 0:3]
assert np.allclose(pos_block, np.eye(3) * expected_pos_var, atol=1e-12)
# ----------------------------------------------------------------------
# AZ-921 AC-7 — when AGL is unknown the strategy emits scale_quality
# = "unknown" and leaves the cov in geometry-natural units (the ESKF
# inflates R_meas; see test_az921_eskf_add_vio_scale_quality.py).
def test_scale_quality_unknown_when_agl_provider_returns_none(
monkeypatch: pytest.MonkeyPatch,
) -> None:
# Arrange
_patch_pose_recovery(monkeypatch, flow_px=20.0)
provider = _ConstantAgl(agl_m=None)
strategy = KltRansacStrategy(
_config(), fdr_client=_fdr_client(), altitude_provider=provider
)
calibration = _calibration(focal_length=1680.0)
# Act
strategy.process_frame(_frame(idx=1), _imu_window(1), calibration)
second = strategy.process_frame(_frame(idx=2), _imu_window(2), calibration)
# Assert
assert second.scale_quality == "unknown"
# Translation stays unit-length when scale could not be recovered.
pose_matrix = np.asarray(second.relative_pose_T.matrix())
translation = pose_matrix[:3, 3]
assert float(np.linalg.norm(translation)) == pytest.approx(1.0, rel=1e-6)
# The position block of cov is NOT overridden by KLT/RANSAC any
# more (AZ-921 AC-7). The ESKF's R_meas adaptation provides the
# "do not trust" signal on the consumer side.
cov = np.asarray(second.pose_covariance_6x6)
pos_block = cov[0:3, 0:3]
assert float(pos_block[0, 0]) < 1.0 # NOT the AZ-920 1e6 sentinel
# ----------------------------------------------------------------------
# AZ-921 AC-6 — direction_only branch reachable on near-vertical motion
def test_scale_quality_direction_only_on_near_pure_vertical_motion(
monkeypatch: pytest.MonkeyPatch,
) -> None:
# Arrange — realistic horizontal flow (20 px) but cv2.recoverPose
# returns a t_unit pointing essentially along the optical axis
# (the climb/descent edge case). Horizontal magnitude is far below
# the _MIN_T_HORIZONTAL_MAGNITUDE = 0.05 threshold.
_patch_pose_recovery(
monkeypatch,
flow_px=20.0,
t_unit=(0.001, 0.001, 0.99999),
)
provider = _ConstantAgl(agl_m=100.0)
strategy = KltRansacStrategy(
_config(), fdr_client=_fdr_client(), altitude_provider=provider
)
calibration = _calibration(focal_length=1680.0)
# Act
strategy.process_frame(_frame(idx=1), _imu_window(1), calibration)
second = strategy.process_frame(_frame(idx=2), _imu_window(2), calibration)
# Assert
assert second.scale_quality == "direction_only"
# Translation magnitude stays at the cv2.recoverPose output — i.e.
# the t_unit input — because scale was not recovered. The mocked
# t_unit is (0.001, 0.001, 0.99999), L2 norm ≈ 0.9999. The test
# asserts the strategy did NOT multiply by a real metric scale.
pose_matrix = np.asarray(second.relative_pose_T.matrix())
translation = pose_matrix[:3, 3]
assert float(np.linalg.norm(translation)) == pytest.approx(0.9999, abs=1e-3)
# ----------------------------------------------------------------------
# AZ-919 AC-3 regression — provider=None keeps pre-AZ-920 behaviour
def test_translation_unchanged_and_cov_untouched_when_provider_is_none(
monkeypatch: pytest.MonkeyPatch,
) -> None:
# Arrange — same mock as before, but the strategy is built WITHOUT
# an altitude_provider (the pre-AZ-919 / legacy unit-test path).
_patch_pose_recovery(monkeypatch, flow_px=20.0)
strategy = KltRansacStrategy(_config(), fdr_client=_fdr_client())
calibration = _calibration(focal_length=1680.0)
# Act
strategy.process_frame(_frame(idx=1), _imu_window(1), calibration)
second = strategy.process_frame(_frame(idx=2), _imu_window(2), calibration)
# Assert — translation stays at the unit-length t_unit (1, 0, 0).
pose_matrix = np.asarray(second.relative_pose_T.matrix())
translation = pose_matrix[:3, 3]
assert float(np.linalg.norm(translation)) == pytest.approx(1.0, rel=1e-6)
# The position block of cov is the same scalar * I as the rotation
# block — no AZ-920 override fired.
cov = np.asarray(second.pose_covariance_6x6)
assert np.allclose(cov[0:3, 0:3], cov[3:6, 3:6], atol=1e-12)
# AZ-921 — without an altitude provider, the strategy defaults
# scale_quality to "unknown" so consumers know to inflate R_meas.
assert second.scale_quality == "unknown"
@@ -0,0 +1,354 @@
"""AZ-922 — cheirality / rotation-plausibility gate in KLT/RANSAC.
Covers AC-1 (axis-angle helper math), AC-2/AC-3 (gate threshold + routing
through `_pose_recovery_failed`), and AC-4 (boundary + AC-7 lost-frame
escalation). AC-5 (Jetson e2e Derkachi past frame 5) is gated separately
via the Tier-2 harness — not in unit scope.
"""
from __future__ import annotations
import math
from datetime import datetime, timezone
from typing import Any
import numpy as np
import pytest
from gps_denied_onboard._types.calibration import CameraCalibration
from gps_denied_onboard._types.nav import (
ImuSample,
ImuWindow,
NavCameraFrame,
)
from gps_denied_onboard.components.c1_vio import C1VioConfig, KltRansacConfig
from gps_denied_onboard.components.c1_vio import klt_ransac as klt_ransac_module
from gps_denied_onboard.components.c1_vio.errors import (
VioFatalError,
VioInitializingError,
)
from gps_denied_onboard.components.c1_vio.klt_ransac import (
KltRansacStrategy,
_rotation_angle_rad,
)
from gps_denied_onboard.config.schema import Config, RuntimeConfig
from gps_denied_onboard.fdr_client.client import FdrClient
from gps_denied_onboard.helpers.ransac_filter import RansacResult
# ----------------------------------------------------------------------
# AC-1 — pure helper math
# ----------------------------------------------------------------------
def test_rotation_angle_identity_is_zero() -> None:
# Assert
assert _rotation_angle_rad(np.eye(3, dtype=np.float64)) == pytest.approx(0.0, abs=1e-12)
def test_rotation_angle_180_about_x_is_pi() -> None:
# Arrange — R rotates 180° about the body-X axis: trace = 1 + (-1) + (-1) = -1.
R = np.array(
[
[1.0, 0.0, 0.0],
[0.0, -1.0, 0.0],
[0.0, 0.0, -1.0],
],
dtype=np.float64,
)
# Assert
assert _rotation_angle_rad(R) == pytest.approx(math.pi, abs=1e-9)
def test_rotation_angle_general_diagonal_recovers_axis_angle_magnitude() -> None:
# Arrange — 30° rotation about body-Z: trace = cos(30)+cos(30)+1.
theta = math.radians(30.0)
R = np.array(
[
[math.cos(theta), -math.sin(theta), 0.0],
[math.sin(theta), math.cos(theta), 0.0],
[0.0, 0.0, 1.0],
],
dtype=np.float64,
)
# Assert
assert _rotation_angle_rad(R) == pytest.approx(theta, abs=1e-9)
def test_rotation_angle_arccos_argument_clipped_against_float_drift() -> None:
# Arrange — synthesise an R whose trace is *just* above 3.0 (impossible
# in exact arithmetic but achievable via float drift). Without the
# arccos clip this would yield NaN; with the clip it must return 0.
R = np.eye(3, dtype=np.float64) + np.full((3, 3), 1e-15, dtype=np.float64)
# Assert
assert _rotation_angle_rad(R) == pytest.approx(0.0, abs=1e-6)
assert math.isfinite(_rotation_angle_rad(R))
def test_rotation_angle_uses_jetson_frame_5_diagonal_signature() -> None:
# Arrange — exact R diagonal observed at Derkachi frame 5 (the
# divergence point). Trace = -0.848 - 0.639 + 0.487 = -1.0.
R = np.diag([-0.848, -0.639, 0.487]).astype(np.float64)
# Assert — recover the 180° rotation the ESKF traceback implied.
assert _rotation_angle_rad(R) == pytest.approx(math.pi, abs=1e-9)
# ----------------------------------------------------------------------
# Strategy-level scaffolding (mirrors test_az920_klt_ransac_scale_integration.py)
# ----------------------------------------------------------------------
def _calibration() -> CameraCalibration:
return CameraCalibration(
camera_id="khp20s30-test",
intrinsics_3x3=np.array(
[
[1680.0, 0.0, 960.0],
[0.0, 1680.0, 540.0],
[0.0, 0.0, 1.0],
],
dtype=np.float64,
),
distortion=np.zeros(5, dtype=np.float64),
body_to_camera_se3=np.eye(4, dtype=np.float64),
acquisition_method="test_az922",
)
def _frame(idx: int) -> NavCameraFrame:
rng = np.random.default_rng(seed=idx)
image = (rng.integers(0, 255, size=(240, 240), dtype=np.int16)).astype(np.uint8)
return NavCameraFrame(
frame_id=idx,
timestamp=datetime.fromtimestamp(idx * 0.1, tz=timezone.utc),
image=image,
camera_calibration_id="khp20s30-test",
)
def _imu_window(frame_idx: int) -> ImuWindow:
ts_start_ns = 1_000_000_000 + frame_idx * 100_000_000
samples = tuple(
ImuSample(
ts_ns=ts_start_ns + i * 5_000_000,
accel_xyz=(0.0, 0.0, 9.81),
gyro_xyz=(0.0, 0.0, 0.0),
)
for i in range(3)
)
return ImuWindow(
samples=samples,
ts_start_ns=samples[0].ts_ns,
ts_end_ns=samples[-1].ts_ns,
)
def _config(lost_frame_threshold: int = 3) -> Config:
return Config.with_blocks(
c1_vio=C1VioConfig(
strategy="klt_ransac",
klt_ransac=KltRansacConfig(),
lost_frame_threshold=lost_frame_threshold,
),
runtime=RuntimeConfig(),
)
def _patch_pose_recovery_with_rotation(
monkeypatch: pytest.MonkeyPatch,
*,
R_3x3: np.ndarray,
inlier_count: int = 40,
) -> None:
"""Force the geometry stack to a controlled path with a chosen R.
Mirrors the AZ-920 test helper but takes the rotation matrix as a
parameter so AZ-922 can exercise the cheirality gate directly.
"""
rng = np.random.default_rng(seed=271828)
base = rng.uniform(50.0, 1800.0, size=(inlier_count, 2))
inliers = np.column_stack([base, base + np.array([5.0, 0.0])])
mask = np.ones((inlier_count, 1), dtype=np.uint8)
def _fake_filter(_corr: np.ndarray, _thresh: float, _min: int) -> RansacResult:
return RansacResult(
inlier_correspondences=inliers,
inlier_count=inlier_count,
outlier_count=0,
median_residual_px=0.5,
)
def _fake_find_essential(*_a: Any, **_k: Any) -> tuple[np.ndarray, np.ndarray]:
return np.eye(3, dtype=np.float64), mask
def _fake_recover_pose(*_a: Any, **_k: Any) -> tuple[int, np.ndarray, np.ndarray, np.ndarray]:
t_col = np.array([[1.0], [0.0], [0.0]], dtype=np.float64)
return inlier_count, R_3x3.astype(np.float64), t_col, mask
monkeypatch.setattr(
klt_ransac_module.RansacFilter,
"filter_correspondences",
staticmethod(_fake_filter),
)
monkeypatch.setattr(klt_ransac_module.cv2, "findEssentialMat", _fake_find_essential)
monkeypatch.setattr(klt_ransac_module.cv2, "recoverPose", _fake_recover_pose)
def _fdr_client() -> FdrClient:
return FdrClient(producer_id="test.az922", capacity=16, _emit_diag_log=False)
def _r_about_z(theta_rad: float) -> np.ndarray:
return np.array(
[
[math.cos(theta_rad), -math.sin(theta_rad), 0.0],
[math.sin(theta_rad), math.cos(theta_rad), 0.0],
[0.0, 0.0, 1.0],
],
dtype=np.float64,
)
# ----------------------------------------------------------------------
# AC-2/AC-3 — gate threshold + routing through _pose_recovery_failed
# ----------------------------------------------------------------------
def test_identity_rotation_passes_gate(monkeypatch: pytest.MonkeyPatch) -> None:
# Arrange
_patch_pose_recovery_with_rotation(monkeypatch, R_3x3=np.eye(3))
strategy = KltRansacStrategy(_config(), fdr_client=_fdr_client())
calibration = _calibration()
# Act
strategy.process_frame(_frame(idx=1), _imu_window(1), calibration)
second = strategy.process_frame(_frame(idx=2), _imu_window(2), calibration)
# Assert
assert second is not None
assert second.frame_id == "2"
def test_five_degree_rotation_passes_gate(monkeypatch: pytest.MonkeyPatch) -> None:
# Arrange — 5° rotation, well below the 30° default threshold.
_patch_pose_recovery_with_rotation(monkeypatch, R_3x3=_r_about_z(math.radians(5.0)))
strategy = KltRansacStrategy(_config(), fdr_client=_fdr_client())
calibration = _calibration()
# Act
strategy.process_frame(_frame(idx=1), _imu_window(1), calibration)
second = strategy.process_frame(_frame(idx=2), _imu_window(2), calibration)
# Assert
assert second is not None
def test_sixty_degree_rotation_rejected_by_gate(monkeypatch: pytest.MonkeyPatch) -> None:
# Arrange — 60° rotation exceeds the 30° default threshold.
_patch_pose_recovery_with_rotation(monkeypatch, R_3x3=_r_about_z(math.radians(60.0)))
strategy = KltRansacStrategy(_config(), fdr_client=_fdr_client())
calibration = _calibration()
# Act + Assert
strategy.process_frame(_frame(idx=1), _imu_window(1), calibration)
with pytest.raises(VioInitializingError, match="implausible_rotation_angle"):
strategy.process_frame(_frame(idx=2), _imu_window(2), calibration)
def test_180_degree_rotation_rejected_by_gate(monkeypatch: pytest.MonkeyPatch) -> None:
# Arrange — synthesise the Jetson-observed frame-5 R signature.
R_180 = np.diag([-0.848, -0.639, 0.487]).astype(np.float64)
_patch_pose_recovery_with_rotation(monkeypatch, R_3x3=R_180)
strategy = KltRansacStrategy(_config(), fdr_client=_fdr_client())
calibration = _calibration()
# Act + Assert
strategy.process_frame(_frame(idx=1), _imu_window(1), calibration)
with pytest.raises(VioInitializingError, match="implausible_rotation_angle"):
strategy.process_frame(_frame(idx=2), _imu_window(2), calibration)
def test_threshold_boundary_just_above_rejects(monkeypatch: pytest.MonkeyPatch) -> None:
# Arrange — 30.1° vs 30° threshold.
_patch_pose_recovery_with_rotation(
monkeypatch, R_3x3=_r_about_z(math.radians(30.1))
)
strategy = KltRansacStrategy(_config(), fdr_client=_fdr_client())
calibration = _calibration()
# Act + Assert
strategy.process_frame(_frame(idx=1), _imu_window(1), calibration)
with pytest.raises(VioInitializingError, match="implausible_rotation_angle"):
strategy.process_frame(_frame(idx=2), _imu_window(2), calibration)
def test_threshold_boundary_just_below_passes(monkeypatch: pytest.MonkeyPatch) -> None:
# Arrange — 29.9° just below 30° threshold.
_patch_pose_recovery_with_rotation(
monkeypatch, R_3x3=_r_about_z(math.radians(29.9))
)
strategy = KltRansacStrategy(_config(), fdr_client=_fdr_client())
calibration = _calibration()
# Act
strategy.process_frame(_frame(idx=1), _imu_window(1), calibration)
second = strategy.process_frame(_frame(idx=2), _imu_window(2), calibration)
# Assert
assert second is not None
# ----------------------------------------------------------------------
# AC-4 (AC-7 path) — consecutive rejections escalate to VioFatalError
# ----------------------------------------------------------------------
def test_consecutive_rejections_eventually_raise_vio_fatal(
monkeypatch: pytest.MonkeyPatch,
) -> None:
# Arrange — every frame yields a 180° flip; lost_frame_threshold=3
# means the 4th frame in this stream (the 3rd consecutive failure)
# crosses the LOST gate.
R_180 = np.diag([-1.0, -1.0, 1.0]).astype(np.float64) # 180° about Z
_patch_pose_recovery_with_rotation(monkeypatch, R_3x3=R_180)
strategy = KltRansacStrategy(
_config(lost_frame_threshold=3), fdr_client=_fdr_client()
)
calibration = _calibration()
# Act — frame 1 seeds INIT; frames 2-4 hit the gate and tick lost.
strategy.process_frame(_frame(idx=1), _imu_window(1), calibration)
for idx in range(2, 4):
with pytest.raises(VioInitializingError, match="implausible_rotation_angle"):
strategy.process_frame(_frame(idx=idx), _imu_window(idx), calibration)
# The 4th consecutive failed frame trips the LOST/VioFatalError gate.
with pytest.raises(VioFatalError, match="exhausted lost-frame budget"):
strategy.process_frame(_frame(idx=4), _imu_window(4), calibration)
# ----------------------------------------------------------------------
# Config validation — defends the (0, π] range
# ----------------------------------------------------------------------
def test_config_rejects_zero_or_negative_threshold() -> None:
# Act + Assert
with pytest.raises(Exception, match="max_frame_rotation_rad must be in"):
KltRansacConfig(max_frame_rotation_rad=0.0)
def test_config_rejects_threshold_above_pi() -> None:
# Act + Assert
with pytest.raises(Exception, match="max_frame_rotation_rad must be in"):
KltRansacConfig(max_frame_rotation_rad=math.pi + 0.01)
def test_config_accepts_pi_exactly() -> None:
# Act — the inclusive upper bound. No rotation can exceed π anyway.
cfg = KltRansacConfig(max_frame_rotation_rad=math.pi)
# Assert
assert cfg.max_frame_rotation_rad == pytest.approx(math.pi)
+13 -1
View File
@@ -53,7 +53,6 @@ from gps_denied_onboard.config.schema import Config, RuntimeConfig
from gps_denied_onboard.fdr_client.client import FdrClient
from gps_denied_onboard.fdr_client.records import FdrRecord
# ----------------------------------------------------------------------
# Helpers — keep boilerplate out of the AC test bodies.
@@ -528,6 +527,19 @@ def test_ac6_low_inlier_count_emits_degraded_with_monotonic_covariance(
calibration,
)
# AZ-922 — pin cv2.recoverPose to identity rotation for frame 2 too.
# Real cv2 on this synthetic input returns a near-180° rotation that
# the cheirality gate would reject; this test asserts FACADE state
# classification on inlier loss, not cv2's geometry on synthetic data.
def _identity_recover_pose(
*_a: Any, **_k: Any
) -> tuple[int, np.ndarray, np.ndarray, np.ndarray]:
R = np.eye(3, dtype=np.float64)
t = np.array([[0.01], [0.0], [0.0]], dtype=np.float64)
return 1, R, t, np.ones((1, 1), dtype=np.uint8)
monkeypatch.setattr(klt_ransac_module.cv2, "recoverPose", _identity_recover_pose)
# Frame 2 — first successful pose recovery (TRACKING).
out_tracking = strategy.process_frame(
_frame(idx=2, image=_synthetic_frame_image(seed=31, shift_x=3)),
@@ -0,0 +1,187 @@
"""AZ-921 AC-5 — `EskfStateEstimator.add_vio` translation-R_meas
adaptation per `VioOutput.scale_quality`.
Two layers of coverage:
1. **Unit** — `_adapt_vio_r_meas` is the pure helper that owns the
block-wise adaptation. Three tests pin its three branches.
2. **Integration** — `EskfStateEstimator.add_vio` is run end-to-end
with each `scale_quality` branch and the resulting nominal position
delta is compared. ``"metric"`` produces a large position update
(the Kalman gain trusts the measurement); ``"unknown"`` produces an
essentially-zero position update (the gain block is ~0).
"""
from __future__ import annotations
import dataclasses
from typing import Any
from unittest import mock
import numpy as np
import pytest
from gps_denied_onboard._types.nav import (
FeatureQuality,
ImuBias,
ScaleQuality,
VioOutput,
)
from gps_denied_onboard.components.c5_state import C5StateConfig
from gps_denied_onboard.components.c5_state.eskf_baseline import (
_DIRECTION_ONLY_TRANSLATION_SIGMA_M,
_UNKNOWN_TRANSLATION_SIGMA_M,
EskfStateEstimator,
_adapt_vio_r_meas,
)
from gps_denied_onboard.config import load_config
from gps_denied_onboard.config.schema import Config
# ----------------------------------------------------------------------
# _adapt_vio_r_meas helper — unit coverage.
def test_adapt_vio_r_meas_metric_returns_input_unchanged() -> None:
# Arrange
r_meas = np.diag([0.04, 0.04, 0.04, 0.0001, 0.0001, 0.0001])
# Act
adapted = _adapt_vio_r_meas(r_meas, "metric")
# Assert
assert np.array_equal(adapted, r_meas)
# Identity is acceptable too — the caller MUST NOT mutate either way.
assert np.array_equal(r_meas, np.diag([0.04, 0.04, 0.04, 0.0001, 0.0001, 0.0001]))
def test_adapt_vio_r_meas_direction_only_overrides_translation_block() -> None:
# Arrange
r_meas = np.diag([0.04, 0.04, 0.04, 0.0001, 0.0001, 0.0001])
expected_translation_var = _DIRECTION_ONLY_TRANSLATION_SIGMA_M**2
# Act
adapted = _adapt_vio_r_meas(r_meas, "direction_only")
# Assert — translation block becomes diag(sigma^2 * I), rotation
# block left as-is.
assert np.allclose(adapted[0:3, 0:3], np.eye(3) * expected_translation_var)
assert np.allclose(adapted[3:6, 3:6], r_meas[3:6, 3:6])
# And the input was NOT mutated.
assert r_meas[0, 0] == pytest.approx(0.04)
def test_adapt_vio_r_meas_unknown_inflates_translation_far_beyond_direction_only() -> None:
# Arrange
r_meas = np.diag([0.04, 0.04, 0.04, 0.0001, 0.0001, 0.0001])
expected_translation_var = _UNKNOWN_TRANSLATION_SIGMA_M**2
# Act
adapted = _adapt_vio_r_meas(r_meas, "unknown")
# Assert
assert np.allclose(adapted[0:3, 0:3], np.eye(3) * expected_translation_var)
assert np.allclose(adapted[3:6, 3:6], r_meas[3:6, 3:6])
# And the input was NOT mutated.
assert r_meas[0, 0] == pytest.approx(0.04)
def test_adapt_vio_r_meas_unknown_translation_var_is_strictly_larger() -> None:
# Sanity: an "unknown" measurement must be treated as strictly less
# informative than a "direction_only" one (otherwise the three-tier
# split is meaningless).
assert _UNKNOWN_TRANSLATION_SIGMA_M > _DIRECTION_ONLY_TRANSLATION_SIGMA_M
# ----------------------------------------------------------------------
# add_vio integration — end-to-end position-update sensitivity per
# scale_quality.
def _build_config(**state_overrides: Any) -> Config:
cfg = load_config(env={}, paths=(), require_env=False)
new_state = dataclasses.replace(C5StateConfig(strategy="eskf"), **state_overrides)
components = dict(cfg.components or {})
components["c5_state"] = new_state
return dataclasses.replace(cfg, components=components)
def _make_estimator() -> EskfStateEstimator:
return EskfStateEstimator(
_build_config(),
imu_preintegrator=mock.MagicMock(),
se3_utils=mock.MagicMock(),
wgs_converter=mock.MagicMock(),
fdr_client=mock.MagicMock(),
)
def _identity_pose() -> np.ndarray:
return np.eye(4, dtype=np.float64)
def _pose_translated(x: float) -> np.ndarray:
p = np.eye(4, dtype=np.float64)
p[0, 3] = x
return p
_ZERO_BIAS = ImuBias(accel_bias=(0.0, 0.0, 0.0), gyro_bias=(0.0, 0.0, 0.0))
_NEUTRAL_FQ = FeatureQuality(tracked=20, new=2, lost=1, mean_parallax=5.0, mre_px=1.0)
def _vio(
frame_id: int,
ts_ns: int,
pose: np.ndarray,
scale_quality: ScaleQuality,
) -> VioOutput:
import gtsam
return VioOutput(
frame_id=str(frame_id),
relative_pose_T=gtsam.Pose3(pose),
pose_covariance_6x6=np.eye(6) * 0.01,
imu_bias=_ZERO_BIAS,
feature_quality=_NEUTRAL_FQ,
emitted_at_ns=int(ts_ns),
scale_quality=scale_quality,
)
def _position_delta_after_vio_pair(scale_quality: ScaleQuality) -> float:
# Arrange — two VIO frames; first seeds, second reports a +1 m
# forward translation. Returns the magnitude of the nominal-pos
# update the ESKF applied. With scale_quality="metric" the Kalman
# gain in the position block is ~1 and the update is ~1 m.
estimator = _make_estimator()
estimator.add_vio(_vio(1, 1_000_000_000, _identity_pose(), scale_quality))
pos_before = estimator._nominal_pos.copy()
estimator.add_vio(_vio(2, 1_100_000_000, _pose_translated(1.0), scale_quality))
# Act / Assert wrapper — returns the metric (m) the caller compares.
return float(np.linalg.norm(estimator._nominal_pos - pos_before))
def test_metric_scale_quality_applies_full_position_update() -> None:
# Assert — with metric scale, the ESKF trusts the +1 m measurement
# almost completely (R_meas[0:3, 0:3] = 0.01 m^2 = 10 cm sigma).
metric_delta = _position_delta_after_vio_pair("metric")
assert metric_delta > 0.5
def test_unknown_scale_quality_yields_near_zero_position_update() -> None:
# Assert — with unknown scale, the ESKF inflates R_meas to (1000 m)^2
# so the Kalman gain block is ~0 and the +1 m measurement does
# essentially nothing to the position state.
unknown_delta = _position_delta_after_vio_pair("unknown")
assert unknown_delta < 1.0e-3
def test_direction_only_scale_quality_yields_partial_update_between_metric_and_unknown() -> None:
# Assert — direction_only sits between the two extremes. The
# specific magnitude depends on the prior covariance + chosen
# sigma, but the ordering is what matters.
metric_delta = _position_delta_after_vio_pair("metric")
direction_only_delta = _position_delta_after_vio_pair("direction_only")
unknown_delta = _position_delta_after_vio_pair("unknown")
assert unknown_delta < direction_only_delta < metric_delta
@@ -86,8 +86,19 @@ def test_ac1_ap_raw_imu_decode() -> None:
frame = received[0]
assert frame.kind is TelemetryKind.IMU_SAMPLE
assert isinstance(frame.payload, ImuTelemetrySample)
assert frame.payload.accel_xyz == (10.0, 20.0, -981.0)
assert frame.payload.gyro_xyz == (1.0, 2.0, 3.0)
# AZ-918: raw MAVLink (xacc=10, yacc=20, zacc=-981) mG FRD →
# SI/FLU m/s² via mavlink_imu_to_si_flu (negate Y, negate Z, *9.80665e-3).
# zacc raw is -981 mG; FRD→FLU negates Z, giving +981 mG = +9.62 m/s².
assert frame.payload.accel_xyz == pytest.approx((
10 * 9.80665e-3,
-20 * 9.80665e-3,
981 * 9.80665e-3,
))
assert frame.payload.gyro_xyz == pytest.approx((
1 * 1.0e-3,
-2 * 1.0e-3,
-3 * 1.0e-3,
))
# received_at is monotonic_ns at decode boundary (positive, non-zero)
assert frame.received_at > 0
@@ -621,7 +632,12 @@ def test_ac10_corrupt_frame_does_not_kill_decoder(
assert len(err_logs) == 1
imu_frames = [f for f in received if f.kind is TelemetryKind.IMU_SAMPLE]
assert len(imu_frames) == 1
assert imu_frames[0].payload.accel_xyz == (1.0, 2.0, 3.0)
# AZ-918: raw MAVLink (xacc=1, yacc=2, zacc=3) mG FRD → SI/FLU.
assert imu_frames[0].payload.accel_xyz == pytest.approx((
1 * 9.80665e-3,
-2 * 9.80665e-3,
-3 * 9.80665e-3,
))
# ----------------------------------------------------------------------
@@ -12,7 +12,6 @@ faked via :class:`_FakeTlog` so tests run without the real C extension.
from __future__ import annotations
import os
import time
from pathlib import Path
from types import SimpleNamespace
@@ -50,7 +49,6 @@ from gps_denied_onboard.components.c8_fc_adapter.tlog_replay_adapter import (
TlogReplayFcAdapter,
)
# ----------------------------------------------------------------------
# Fixtures + helpers
@@ -379,7 +377,19 @@ def test_ac2_ap_dialect_frame_mapping_in_tlog_order(
TelemetryKind.MAV_STATE,
]
assert isinstance(received[0].payload, ImuTelemetrySample)
assert received[0].payload.accel_xyz == (10.0, 20.0, -981.0)
# AZ-918: raw MAVLink (xacc=10, yacc=20, zacc=-981) mG FRD →
# SI/FLU m/s² via mavlink_imu_to_si_flu (negate Y, negate Z, *9.80665e-3).
# zacc raw is -981 mG; FRD→FLU negates Z, giving +981 mG = +9.62 m/s².
assert received[0].payload.accel_xyz == pytest.approx((
10 * 9.80665e-3,
-20 * 9.80665e-3,
981 * 9.80665e-3,
))
assert received[0].payload.gyro_xyz == pytest.approx((
1 * 1.0e-3,
-2 * 1.0e-3,
-3 * 1.0e-3,
))
assert isinstance(received[1].payload, AttitudeSample)
assert received[1].payload.yaw_rad == pytest.approx(1.5)
assert isinstance(received[2].payload, GpsHealth)
@@ -26,7 +26,6 @@ from gps_denied_onboard.replay_input.csv_ground_truth import (
)
from gps_denied_onboard.replay_input.errors import ReplayInputAdapterError
_DERKACHI_CSV: Path = (
Path(__file__).resolve().parents[3]
/ "_docs"
@@ -163,11 +162,13 @@ def test_paired_imu_and_gps_share_clock(tmp_path: Path) -> None:
def test_gps_unit_conversion(tmp_path: Path) -> None:
# Arrange — values exercise the deg/mm/cm-s/cdeg conversions.
# Arrange — values exercise the deg/mm/cm-s/cdeg conversions on
# the GPS columns and the mG/mrad-s + FRD→FLU conversion on the
# IMU columns (AZ-918).
header = _full_header()
row = ",".join([
"0.0", "0.0",
"10", "-3", "-980", "50", "30", "-5", # IMU stays raw
"10", "-3", "-980", "50", "30", "-5", # IMU raw (mG/mrad·s⁻¹/FRD)
"50.0809634", # lat already in degrees
"36.1115442", # lon already in degrees
"141290", # alt in mm → 141.290 m
@@ -181,7 +182,7 @@ def test_gps_unit_conversion(tmp_path: Path) -> None:
# Act
gt = load_csv_ground_truth(csv)
# Assert
# Assert — GPS in SI / decimal-degrees.
fix = gt.records[0]
assert fix.lat_deg == pytest.approx(50.0809634)
assert fix.lon_deg == pytest.approx(36.1115442)
@@ -190,6 +191,21 @@ def test_gps_unit_conversion(tmp_path: Path) -> None:
assert fix.vy_m_s == pytest.approx(6.0)
assert fix.vz_m_s == pytest.approx(-0.88)
assert fix.hdg_deg == pytest.approx(350.41)
# Assert — IMU converted to m/s² + rad/s, body frame FLU.
imu = gt.imu_samples[0]
# AZ-918: CSV ships MAVLink wire format (mG/mrad/s, FRD body); the
# parser routes through mavlink_imu_to_si_flu so consumers see SI/FLU.
# FRD→FLU negates Y and Z, so a raw -3 (yacc) / -980 (zacc) become +3 / +980.
assert imu.accel_xyz == pytest.approx((
10 * 9.80665e-3,
3 * 9.80665e-3,
980 * 9.80665e-3,
))
assert imu.gyro_xyz == pytest.approx((
50 * 1.0e-3,
-30 * 1.0e-3,
5 * 1.0e-3,
))
# ---------------------------------------------------------------------
@@ -0,0 +1,95 @@
"""AZ-919 — `_c1_vio_wrapper` builds and forwards an `EskfNominalAltitudeProvider`.
Tests the composition-root seam: the AZ-919 plumbing is only useful if
the wrapper actually constructs the provider and passes it to
``build_vio_strategy``. The provider's supplier must close over the same
mutable ``constructed`` dict that the bootstrap loop populates so it
sees the C5 estimator after the topo order builds it.
"""
from __future__ import annotations
from typing import Any
import pytest
from gps_denied_onboard.config import Config
from gps_denied_onboard.helpers.altitude_provider import (
AltitudeProvider,
EskfNominalAltitudeProvider,
)
from gps_denied_onboard.runtime_root import airborne_bootstrap
from gps_denied_onboard.runtime_root.airborne_bootstrap import _c1_vio_wrapper
def _minimal_config() -> Config:
return Config.with_blocks(c1_vio={"strategy": "klt_ransac"})
def test_wrapper_forwards_altitude_provider_kwarg(
monkeypatch: pytest.MonkeyPatch,
) -> None:
# Arrange — capture the kwargs handed to build_vio_strategy.
captured: dict[str, Any] = {}
def _capture(_config: Config, **kwargs: Any) -> object:
captured.update(kwargs)
return object()
monkeypatch.setattr(airborne_bootstrap, "build_vio_strategy", _capture)
constructed: dict[str, Any] = {"c13_fdr": object()}
# Act
_c1_vio_wrapper(_minimal_config(), constructed)
# Assert
assert "altitude_provider" in captured
assert isinstance(captured["altitude_provider"], EskfNominalAltitudeProvider)
assert isinstance(captured["altitude_provider"], AltitudeProvider)
def test_provider_supplier_resolves_c5_estimator_lazily(
monkeypatch: pytest.MonkeyPatch,
) -> None:
# Arrange — capture the provider, then mutate the constructed dict
# after the wrapper returns to simulate C5 landing in the topo order.
captured: dict[str, Any] = {}
def _capture(_config: Config, **kwargs: Any) -> object:
captured.update(kwargs)
return object()
monkeypatch.setattr(airborne_bootstrap, "build_vio_strategy", _capture)
constructed: dict[str, Any] = {"c13_fdr": object()}
_c1_vio_wrapper(_minimal_config(), constructed)
provider: EskfNominalAltitudeProvider = captured["altitude_provider"]
# Pre-C5 the supplier returns None.
pre_c5 = provider._estimator_supplier()
# The bootstrap loop later assigns the c5_state slot.
sentinel_estimator = object()
constructed["c5_state"] = sentinel_estimator
post_c5 = provider._estimator_supplier()
# Assert
assert pre_c5 is None
assert post_c5 is sentinel_estimator
def test_wrapper_still_requires_c13_fdr(
monkeypatch: pytest.MonkeyPatch,
) -> None:
# Arrange — AZ-919 must not silently relax the existing AC-618-3
# invariant: an absent c13_fdr key still raises AirborneBootstrapError.
monkeypatch.setattr(
airborne_bootstrap,
"build_vio_strategy",
lambda *_a, **_kw: object(),
)
# Act + Assert
with pytest.raises(airborne_bootstrap.AirborneBootstrapError) as exc:
_c1_vio_wrapper(_minimal_config(), constructed={})
assert "c13_fdr" in str(exc.value)
assert "c1_vio" in str(exc.value)
+103
View File
@@ -0,0 +1,103 @@
"""AZ-918 — `mavlink_imu_to_si_flu` unit + frame conversion.
Pins the conversion contract so any future change to either the
constant or the body-frame transform is a deliberate, reviewed edit.
The helper sits between every MAVLink-IMU adapter and the
``nav.ImuSample`` / ``fc.ImuTelemetrySample`` boundary, so a silent
regression here would break C5 ESKF and FDR consumers simultaneously.
"""
from __future__ import annotations
import math
import pytest
from gps_denied_onboard.helpers.imu_units import (
MG_TO_M_S2,
MRAD_S_TO_RAD_S,
mavlink_imu_to_si_flu,
)
def test_constants_match_si_definition() -> None:
# Assert
assert MG_TO_M_S2 == pytest.approx(9.80665e-3)
assert MRAD_S_TO_RAD_S == pytest.approx(1.0e-3)
def test_stationary_frd_body_z_down_becomes_flu_body_z_up_one_g() -> None:
# Arrange — Stationary, level UAV: gravity vector in MAVLink FRD
# body frame points in +Z (down), so the measured specific force
# is -gravity = -Z. Magnitude is exactly one standard gravity.
raw_z_down_mg = -1000.0
# Act
accel_si_flu, gyro_si_flu = mavlink_imu_to_si_flu(
xacc=0.0,
yacc=0.0,
zacc=raw_z_down_mg,
xgyro=0.0,
ygyro=0.0,
zgyro=0.0,
)
# Assert — In ESKF/preintegrator FLU body the same specific force
# should appear as +Z = +9.80665 m/s² (matching the C5 ESKF unit
# test's stationary vector).
assert accel_si_flu == pytest.approx((0.0, 0.0, 9.80665))
assert gyro_si_flu == pytest.approx((0.0, 0.0, 0.0))
def test_frd_to_flu_negates_y_and_z_keeps_x() -> None:
# Arrange — distinguishable components per axis.
# Act
accel_si_flu, gyro_si_flu = mavlink_imu_to_si_flu(
xacc=100.0, yacc=200.0, zacc=-300.0,
xgyro=10.0, ygyro=20.0, zgyro=-30.0,
)
# Assert — X stays positive; Y and Z flip sign per the FRD→FLU
# body-frame transform.
assert accel_si_flu == pytest.approx((
+100.0 * MG_TO_M_S2,
-200.0 * MG_TO_M_S2,
+300.0 * MG_TO_M_S2,
))
assert gyro_si_flu == pytest.approx((
+10.0 * MRAD_S_TO_RAD_S,
-20.0 * MRAD_S_TO_RAD_S,
+30.0 * MRAD_S_TO_RAD_S,
))
def test_unit_magnitudes_match_first_csv_row_of_derkachi_fixture() -> None:
# Arrange — CSV row 0 of `data_imu.csv` (Derkachi fixture). UAV is
# in slow descent at ~0.88 m/s, near level, so |accel| should be
# close to 1 g.
raw_xacc, raw_yacc, raw_zacc = 21.0, -3.0, -984.0
# Act
accel_si_flu, _ = mavlink_imu_to_si_flu(
xacc=raw_xacc, yacc=raw_yacc, zacc=raw_zacc,
xgyro=0.0, ygyro=0.0, zgyro=0.0,
)
# Assert — magnitude is within 5% of one standard gravity (the
# body is descending, not perfectly stationary, so a tight bound
# is wrong; this test pins the order-of-magnitude only).
mag = math.sqrt(sum(c * c for c in accel_si_flu))
assert 9.30 < mag < 10.30, f"|a|={mag:.3f} m/s² is not near 1 g"
def test_zero_input_returns_zero_output() -> None:
# Act
accel, gyro = mavlink_imu_to_si_flu(
xacc=0.0, yacc=0.0, zacc=0.0,
xgyro=0.0, ygyro=0.0, zgyro=0.0,
)
# Assert
assert accel == (0.0, 0.0, 0.0)
assert gyro == (0.0, 0.0, 0.0)
+107
View File
@@ -0,0 +1,107 @@
"""AZ-919 — `AltitudeProvider` Protocol + `EskfNominalAltitudeProvider` contract.
Pins the gating rules that the GSD scale-recovery work (AZ-920) and the
degraded-mode signal (AZ-921) build on top of.
"""
from __future__ import annotations
from typing import Any
import numpy as np
from gps_denied_onboard._types.state import IsamState
from gps_denied_onboard.helpers.altitude_provider import (
AltitudeProvider,
EskfNominalAltitudeProvider,
)
class _FakeEstimator:
"""Minimal ESKF stand-in exposing only the attrs the provider reads."""
def __init__(
self,
*,
nominal_pos_z: float,
takeoff_origin_set: bool,
isam2_state: IsamState,
) -> None:
self._nominal_pos = np.array([0.0, 0.0, nominal_pos_z], dtype=np.float64)
self._takeoff_origin_set: tuple[Any, float, float] | None = (
("origin-sentinel", 1.0, 1.0) if takeoff_origin_set else None
)
self._isam2_state = isam2_state
def test_provider_is_runtime_checkable_protocol() -> None:
provider = EskfNominalAltitudeProvider(estimator_supplier=lambda: None)
assert isinstance(provider, AltitudeProvider)
def test_returns_none_when_supplier_yields_none() -> None:
provider = EskfNominalAltitudeProvider(estimator_supplier=lambda: None)
assert provider.agl_m(now_ns=1_000_000_000) is None
def test_returns_none_before_takeoff_origin_is_anchored() -> None:
# Arrange — origin not yet set, even though nominal_pos has a value.
estimator = _FakeEstimator(
nominal_pos_z=42.0,
takeoff_origin_set=False,
isam2_state=IsamState.INIT,
)
provider = EskfNominalAltitudeProvider(estimator_supplier=lambda: estimator)
assert provider.agl_m(now_ns=1_000_000_000) is None
def test_returns_nominal_pos_z_after_origin_is_anchored() -> None:
# Arrange
estimator = _FakeEstimator(
nominal_pos_z=83.5,
takeoff_origin_set=True,
isam2_state=IsamState.TRACKING,
)
provider = EskfNominalAltitudeProvider(estimator_supplier=lambda: estimator)
# Act + Assert
assert provider.agl_m(now_ns=1_000_000_000) == 83.5
def test_returns_none_when_estimator_is_lost() -> None:
# Arrange — origin anchored and altitude is non-zero, but the
# filter has flipped to LOST; the AGL signal is no longer trustworthy.
estimator = _FakeEstimator(
nominal_pos_z=120.0,
takeoff_origin_set=True,
isam2_state=IsamState.LOST,
)
provider = EskfNominalAltitudeProvider(estimator_supplier=lambda: estimator)
assert provider.agl_m(now_ns=1_000_000_000) is None
def test_supplier_is_re_resolved_per_call() -> None:
# Arrange — the composition root builds C1 before C5, so the
# supplier must read from a mutable container at call time.
holder: dict[str, _FakeEstimator] = {}
provider = EskfNominalAltitudeProvider(
estimator_supplier=lambda: holder.get("c5_state"),
)
# Act — first call: C5 not yet built.
first = provider.agl_m(now_ns=1_000_000_000)
# Then C5 lands in the dict.
holder["c5_state"] = _FakeEstimator(
nominal_pos_z=37.0,
takeoff_origin_set=True,
isam2_state=IsamState.TRACKING,
)
second = provider.agl_m(now_ns=2_000_000_000)
# Assert
assert first is None
assert second == 37.0
@@ -0,0 +1,91 @@
"""AZ-921 AC-4 — VioOutput.scale_quality DTO contract."""
from __future__ import annotations
import typing
from dataclasses import fields
import numpy as np
from gps_denied_onboard._types.nav import (
FeatureQuality,
ImuBias,
ScaleQuality,
VioOutput,
)
from gps_denied_onboard.helpers.se3_utils import SE3, matrix_to_se3
def _imu_bias() -> ImuBias:
return ImuBias(
accel_bias=(0.0, 0.0, 0.0),
gyro_bias=(0.0, 0.0, 0.0),
)
def _feature_quality() -> FeatureQuality:
return FeatureQuality(
tracked=10,
new=2,
lost=1,
mean_parallax=1.0,
mre_px=0.5,
)
def _identity_se3() -> SE3:
return matrix_to_se3(np.eye(4, dtype=np.float64))
def test_vio_output_default_scale_quality_is_unknown_for_back_compat() -> None:
# Arrange / Act — constructor MUST work without scale_quality so
# legacy strategies (OKVIS2, VINS-Mono) that have not been updated
# for AZ-921 stay bug-for-bug compatible.
vio = VioOutput(
frame_id="frame-0",
relative_pose_T=_identity_se3(),
pose_covariance_6x6=np.eye(6, dtype=np.float64),
imu_bias=_imu_bias(),
feature_quality=_feature_quality(),
emitted_at_ns=1_000_000_000,
)
# Assert
assert vio.scale_quality == "unknown"
def test_vio_output_accepts_each_scale_quality_value() -> None:
# Arrange
accepted_values = ("metric", "direction_only", "unknown")
# Act / Assert
for value in accepted_values:
vio = VioOutput(
frame_id="frame-0",
relative_pose_T=_identity_se3(),
pose_covariance_6x6=np.eye(6, dtype=np.float64),
imu_bias=_imu_bias(),
feature_quality=_feature_quality(),
emitted_at_ns=1_000_000_000,
scale_quality=value,
)
assert vio.scale_quality == value
def test_scale_quality_literal_pins_the_three_documented_values() -> None:
# Arrange / Act — the Literal type from the typing module exposes
# its parameters via __args__, which lets the contract test pin
# the exact accepted-string set rather than relying on a string
# comparison alone.
args = typing.get_args(ScaleQuality)
# Assert
assert set(args) == {"metric", "direction_only", "unknown"}
def test_vio_output_dataclass_exposes_scale_quality_field() -> None:
# Arrange / Act
field_names = {f.name for f in fields(VioOutput)}
# Assert
assert "scale_quality" in field_names