fix(log-analysis): harden Plane landing evidence

Signed-off-by: nflyernz <319460892+nflyernz@users.noreply.github.com>

Author: nflyernz

ardupilot_methodic_configurator/log_analysis/data_model_availability_plane_landing.py

@@ -8,6 +8,7 @@
 
 from __future__ import annotations
 
+import math
 from typing import TYPE_CHECKING
 
 from ardupilot_methodic_configurator import _
@@ -179,7 +180,7 @@ def analyse(self) -> LogAnalysisResult:
         )
 
     def _attempt_outcome(self, attempt_number: int, attempt: PlaneLandingAttempt) -> LogAnalysis:
-        flare_altitude_m = self.parameter_history.value_at("LAND_FLARE_ALT", attempt.start_s)
+        flare_altitude_m = self._finite_parameter_value(self.parameter_history.value_at("LAND_FLARE_ALT", attempt.start_s))
         parameter_evidence = (
             _("LAND_FLARE_ALT at attempt start: {value:.2f} m").format(value=flare_altitude_m)
             if flare_altitude_m is not None
@@ -249,6 +250,11 @@ def _stage_outcomes(self, attempt_number: int, evidence: PlaneLandingStageEviden
                 )
         return outcomes
 
+    @staticmethod
+    def _finite_parameter_value(value: float | None) -> float | None:
+        """Treat non-finite event-time parameter values as unavailable evidence."""
+        return value if value is not None and math.isfinite(value) else None
+
     def _firmware_message_outcomes(
         self,
         attempt_number: int,

ardupilot_methodic_configurator/log_analysis/data_model_plane_landing.py

@@ -24,6 +24,23 @@
     from ardupilot_methodic_configurator.log_analysis.data_model_parameter_history import ParameterHistory
 
 
+def _active_gps_fields(log_data: LogData, *field_names: str) -> tuple[np.ndarray, ...] | None:
+    """Return scaled GPS fields, restricted to the active receiver when ``U`` is available."""
+    gps = log_data.get_message_columns("GPS")
+    if gps is None:
+        return None
+    names = gps.dtype.names or ()
+    if not set(field_names).issubset(names):
+        return None
+
+    values = tuple(log_data.get_field("GPS", field_name) for field_name in field_names)
+    if "U" not in names:
+        return values
+
+    active_receiver_samples = log_data.get_field("GPS", "U") == 1
+    return tuple(field_values[active_receiver_samples] for field_values in values)
+
+
 class PlaneLandingEndReason(str, Enum):
     """Objective evidence that bounded a detected AUTO landing attempt."""
 
@@ -188,16 +205,13 @@ def _first_mode_exit(
 
     @classmethod
     def _first_gps_stop(cls, log_data: LogData, flight_segment: FlightSegment, start_s: float) -> float | None:
-        gps = log_data.get_message_columns("GPS")
-        if gps is None or not {"TimeUS", "Spd"}.issubset(gps.dtype.names or ()):
+        gps_fields = _active_gps_fields(log_data, "TimeUS", "Spd")
+        if gps_fields is None:
             return None
+        time_s, speed_m_s = gps_fields
 
         below_since_s: float | None = None
-        for timestamp, speed in zip(
-            log_data.get_field("GPS", "TimeUS"),
-            log_data.get_field("GPS", "Spd"),
-            strict=True,
-        ):
+        for timestamp, speed in zip(time_s, speed_m_s, strict=True):
             timestamp_s = float(timestamp)
             if not start_s <= timestamp_s <= flight_segment.end_s:
                 continue
@@ -700,18 +714,13 @@ def _cmd_records(cls, log_data: LogData) -> list[_MissionCommandRecord]:
     @classmethod
     def _nearest_gps_position(cls, log_data: LogData, target_time_s: float) -> tuple[float, float, float] | None:
         """Return the nearest finite GPS position to the supplied time."""
-        gps = log_data.get_message_columns("GPS")
-        if gps is None or not {"TimeUS", "Lat", "Lng"}.issubset(gps.dtype.names or ()):
+        gps_fields = _active_gps_fields(log_data, "TimeUS", "Lat", "Lng")
+        if gps_fields is None:
             return None
 
         nearest_position: tuple[float, float, float] | None = None
         nearest_offset: float | None = None
-        for timestamp, latitude, longitude in zip(
-            log_data.get_field("GPS", "TimeUS"),
-            log_data.get_field("GPS", "Lat"),
-            log_data.get_field("GPS", "Lng"),
-            strict=True,
-        ):
+        for timestamp, latitude, longitude in zip(*gps_fields, strict=True):
             timestamp_s = cls._finite_float(timestamp)
             latitude_deg = cls._finite_float(latitude)
             longitude_deg = cls._finite_float(longitude)
@@ -851,7 +860,7 @@ def _build_stage_evidence(
                 else None
             ),
             parameter_values={
-                parameter_name: parameter_history.value_at(parameter_name, time_s)
+                parameter_name: cls._finite_float(parameter_history.value_at(parameter_name, time_s))
                 for parameter_name in cls._PARAMETERS_BY_STAGE[stage]
             },
         )
@@ -864,18 +873,22 @@ def _nearest_value(
         telemetry_field: tuple[str, str],
         target_time_s: float,
     ) -> float | None:
-        message_name, field_name = telemetry_field
+        message_name = telemetry_field[0]
         records = log_data.get_message_columns(message_name)
-        if records is None or not {"TimeUS", field_name}.issubset(records.dtype.names or ()):
+        if records is None or not {"TimeUS", telemetry_field[1]}.issubset(records.dtype.names or ()):
+            return None
+
+        telemetry_fields = (
+            _active_gps_fields(log_data, "TimeUS", telemetry_field[1])
+            if message_name == "GPS"
+            else (log_data.get_field(message_name, "TimeUS"), log_data.get_field(message_name, telemetry_field[1]))
+        )
+        if telemetry_fields is None:
             return None
 
         nearest_value: float | None = None
         nearest_offset: float | None = None
-        for timestamp, value in zip(
-            log_data.get_field(message_name, "TimeUS"),
-            log_data.get_field(message_name, field_name),
-            strict=True,
-        ):
+        for timestamp, value in zip(*telemetry_fields, strict=True):
             timestamp_s = cls._finite_float(timestamp)
             measured_value = cls._finite_float(value)
             if timestamp_s is None or measured_value is None or not attempt.start_s <= timestamp_s <= attempt.end_s:

tests/test_data_model_plane_landing.py

@@ -2,6 +2,8 @@
 
 """Focused tests for AMC-native ArduPlane landing-attempt analysis."""
 
+# pylint: disable=too-many-lines
+
 from collections.abc import Sequence
 
 import numpy as np
@@ -365,6 +367,94 @@ def test_attempt_times_use_scaled_seconds() -> None:
     assert evidence[0].time_s == 15.0
 
 
+def test_inactive_gps_receiver_cannot_create_false_stop() -> None:
+    log_data = _plane_log(messages=((40.0, "Throttle disarmed"),))
+    _add_columns(
+        log_data,
+        "GPS",
+        (
+            (0.0, 6.0, 0, 1),
+            (2.0, 6.0, 0, 1),
+            (11.0, 1.0, 1, 0),
+            (13.0, 1.0, 1, 0),
+            (20.0, 6.0, 0, 1),
+            (40.0, 6.0, 0, 1),
+        ),
+        [("TimeUS", "f8"), ("Spd", "f8"), ("I", "u1"), ("U", "u1")],
+    )
+    segment = PlaneFlightSegmentDetector.detect(log_data, {}).segments[0]
+
+    attempt = PlaneLandingAttemptDetector.detect(log_data, segment)[0]
+
+    assert (attempt.start_s, attempt.end_s, attempt.end_reason) == (10.0, 40.0, PlaneLandingEndReason.DISARM)
+
+
+def test_inactive_gps_receiver_cannot_supply_stage_speed() -> None:
+    log_data = _plane_log(
+        land=((5.0, 0), (10.0, 1), (15.0, 2)),
+        messages=((40.0, "Throttle disarmed"),),
+    )
+    _add_columns(
+        log_data,
+        "GPS",
+        (
+            (0.0, 6.0, 0, 1),
+            (2.0, 6.0, 0, 1),
+            (14.0, 7.0, 0, 1),
+            (15.0, 1.0, 1, 0),
+            (16.0, 8.0, 0, 1),
+            (40.0, 6.0, 0, 1),
+        ),
+        [("TimeUS", "f8"), ("Spd", "f8"), ("I", "u1"), ("U", "u1")],
+    )
+    segment = PlaneFlightSegmentDetector.detect(log_data, {}).segments[0]
+    attempt = PlaneLandingAttemptDetector.detect(log_data, segment)[0]
+
+    evidence = PlaneLandingEvidenceExtractor.extract(log_data, attempt, ParameterHistory())
+
+    assert evidence[0].gps_ground_speed_m_s == 7.0
+
+
+def test_inactive_gps_receiver_cannot_supply_target_distance_position() -> None:
+    log_data = _gps_stop_log(((1.0, 1, 0, 21, 0.0, 1.0),))
+    _add_columns(
+        log_data,
+        "GPS",
+        (
+            (20.0, 2.0, 10.0, 10.0, 1, 0),
+            (20.0, 2.0, 0.0, 1.001, 0, 1),
+        ),
+        [("TimeUS", "f8"), ("Spd", "f8"), ("Lat", "f8"), ("Lng", "f8"), ("I", "u1"), ("U", "u1")],
+    )
+    segment = FlightSegment(start_s=0.0, end_s=30.0, is_complete=True)
+    attempt = PlaneLandingAttempt(
+        flight_segment=segment,
+        start_s=10.0,
+        end_s=20.0,
+        end_reason=PlaneLandingEndReason.GPS_STOP,
+    )
+
+    distance = PlaneLandingMissionTargetExtractor.distance_at_gps_stop(log_data, attempt)
+
+    assert distance is not None
+    assert (distance.aircraft_latitude_deg, distance.aircraft_longitude_deg) == (0.0, 1.001)
+    assert distance.distance_m == pytest.approx(111.1949266)
+
+
+def test_gps_without_use_flag_retains_existing_stop_behavior() -> None:
+    log_data = _plane_log(
+        gps=((0.0, 6.0), (2.0, 6.0), (20.0, 2.0), (22.0, 2.0), (30.0, 6.0)),
+    )
+    gps = log_data.get_message_columns("GPS")
+    assert gps is not None
+    assert "U" not in (gps.dtype.names or ())
+    segment = PlaneFlightSegmentDetector.detect(log_data, {}).segments[0]
+
+    attempt = PlaneLandingAttemptDetector.detect(log_data, segment)[0]
+
+    assert (attempt.end_s, attempt.end_reason) == (20.0, PlaneLandingEndReason.GPS_STOP)
+
+
 def test_stage_evidence_uses_land_and_nearest_optional_telemetry() -> None:
     log_data = _plane_log(
         gps=(
@@ -910,6 +1000,81 @@ def test_rfnd_lifecycle_flat_outcomes_leave_stage_point_measurement_unchanged()
     ]
 
 
+@pytest.mark.parametrize("non_finite_value", [float("nan"), float("inf"), float("-inf")])
+def test_non_finite_event_time_landing_parameters_are_unavailable(non_finite_value: float) -> None:
+    log_data = _plane_log(
+        land=((5.0, 0), (10.0, 1), (15.0, 2), (20.0, 3)),
+        messages=((40.0, "Throttle disarmed"),),
+    )
+    parameter_names = ("LAND_PF_ALT", "LAND_PF_SEC", "LAND_FLARE_ALT", "LAND_FLARE_SEC", "LAND_PITCH_DEG")
+    history = ParameterHistory(dict.fromkeys(parameter_names, non_finite_value))
+    segment = PlaneFlightSegmentDetector.detect(log_data, {}).segments[0]
+    attempt = PlaneLandingAttemptDetector.detect(log_data, segment)[0]
+
+    evidence = PlaneLandingEvidenceExtractor.extract(log_data, attempt, history)
+    result = PlaneLandingAnalysis(log_data, _context(history)).analyse()
+
+    assert all(value is None for item in evidence for value in item.parameter_values.values())
+    attempt_outcome = next(outcome for outcome in result.outcomes if outcome.message.startswith("AUTO landing attempt"))
+    assert attempt_outcome.value is None
+    assert "LAND_FLARE_ALT was unavailable" in attempt_outcome.message
+    assert not any(
+        parameter_name in outcome.message and "effective value" in outcome.message
+        for parameter_name in parameter_names
+        for outcome in result.outcomes
+    )
+
+
+def test_finite_event_time_landing_parameters_are_emitted() -> None:
+    log_data = _plane_log(
+        land=((5.0, 0), (10.0, 1), (15.0, 2), (20.0, 3)),
+        messages=((40.0, "Throttle disarmed"),),
+    )
+    parameter_values = {
+        "LAND_PF_ALT": 6.0,
+        "LAND_PF_SEC": 2.0,
+        "LAND_FLARE_ALT": 3.0,
+        "LAND_FLARE_SEC": 1.5,
+        "LAND_PITCH_DEG": 4.0,
+    }
+
+    result = PlaneLandingAnalysis(log_data, _context(ParameterHistory(parameter_values))).analyse()
+
+    attempt_outcome = next(outcome for outcome in result.outcomes if outcome.message.startswith("AUTO landing attempt"))
+    assert attempt_outcome.value == 3.0
+    parameter_outcomes = [outcome for outcome in result.outcomes if "effective value" in outcome.message]
+    emitted_parameter_names = {
+        parameter_name
+        for parameter_name in parameter_values
+        if any(parameter_name in item.message for item in parameter_outcomes)
+    }
+    assert emitted_parameter_names == set(parameter_values)
+
+
+def test_finite_parameter_change_at_event_time_remains_effective() -> None:
+    log_data = _plane_log(
+        land=((5.0, 0), (10.0, 1), (15.0, 2), (20.0, 3)),
+        messages=((40.0, "Throttle disarmed"),),
+    )
+    history = ParameterHistory(
+        {"LAND_PF_ALT": 5.0, "LAND_FLARE_ALT": 3.0},
+        {
+            "LAND_PF_ALT": (ParameterChange(time_s=15.0, value=6.0),),
+            "LAND_FLARE_ALT": (ParameterChange(time_s=20.0, value=4.0),),
+        },
+    )
+
+    result = PlaneLandingAnalysis(log_data, _context(history)).analyse()
+
+    attempt_outcome = next(outcome for outcome in result.outcomes if outcome.message.startswith("AUTO landing attempt"))
+    assert attempt_outcome.value == 3.0
+    parameter_outcomes = [outcome for outcome in result.outcomes if "effective value" in outcome.message]
+    assert [(outcome.timestamp_us, outcome.value) for outcome in parameter_outcomes] == [
+        (15_000_000, 6.0),
+        (20_000_000, 4.0),
+    ]
+
+
 def test_event_time_parameters_change_between_attempts_without_changing_boundaries() -> None:
     log_data = _plane_log(
         land=(