refactor(pipe): isolate plausibility validation

Author: junzis

src/pyModeS/_pipe.py

@@ -38,6 +38,7 @@ def decode_one(msg: str, ts: float):
 from typing import Any
 
 from pyModeS._aero import gs_to_ias, gs_to_mach
+from pyModeS._pipe_validation import ValidationMixin
 from pyModeS.errors import InvalidHexError, InvalidLengthError
 from pyModeS.message import Decoded, Message
 
@@ -95,85 +96,8 @@ def _haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
 # still refreshing useful state that already exists for the ICAO.
 _STATEFUL_DFS = frozenset((0, 4, 16, 17, 18, 20, 21))
 
-# BDS-payload fields cleared from a DF20 result that fails altitude
-# cross-check. The `altitude` field itself is preserved (it's the
-# 13-bit AC-code that triggered the mismatch — callers want it to
-# see why the message was flagged).
-_BDS_FIELDS_TO_CLEAR: tuple[str, ...] = (
-    "bds",
-    "bds_candidates",
-    # BDS 1,0 / 1,7 — data-link capability / supported-BDS bitmap
-    "supported_bds",
-    # BDS 2,0 — aircraft identification
-    "callsign",
-    # BDS 4,0 — selected vertical intention
-    "selected_altitude_mcp",
-    "selected_altitude_fms",
-    "baro_pressure_setting",
-    "vnav_mode",
-    "altitude_hold_mode",
-    "approach_mode",
-    "target_altitude_source",
-    # BDS 4,4 / 4,5 — meteorological routine (only populated when
-    # include_meteo is enabled, but strip the slots regardless)
-    "wind_speed",
-    "wind_direction",
-    "static_air_temperature",
-    "static_pressure",
-    "humidity",
-    "turbulence",
-    "figure_of_merit",
-    "wind_shear",
-    "microburst",
-    "icing",
-    "radio_height",
-    # BDS 5,0 — track and turn
-    "true_airspeed",
-    "true_track",
-    "roll",
-    "track_rate",
-    "groundspeed",
-    # BDS 6,0 — heading and speed
-    "magnetic_heading",
-    "indicated_airspeed",
-    "mach",
-    "baro_vertical_rate",
-    "inertial_vertical_rate",
-)
-
-
-def _altitude_tolerance(dt: float) -> float:
-    """Max plausible |ADS-B - Comm-B AC-code| altitude diff after dt seconds.
-
-    Grows linearly with dt at 100 ft/s (≈ 6000 fpm, emergency descent
-    rate for commercial jets), floored at 500 ft (signal noise +
-    between-sample rounding) and capped at 5000 ft so stale ADS-B
-    doesn't accept arbitrarily-wrong Comm-B altitudes.
-    """
-    return max(500.0, min(dt * 100.0, 5000.0))
-
-
-def _groundspeed_tolerance(dt: float) -> float:
-    """Max plausible |gs_now - gs_anchor| in kt after dt seconds.
-
-    5 kt/s covers aggressive jet accel/decel plus ~30 kt wind gusts,
-    floored at 20 kt (sample-to-sample noise + 1-kt LSB), capped at
-    200 kt so a stale anchor can't rubber-stamp arbitrary jumps.
-    """
-    return max(20.0, min(dt * 5.0, 200.0))
-
 
-def _track_tolerance(dt: float) -> float:
-    """Max plausible circular |track_now - track_anchor| in degrees.
-
-    10°/s covers a rate-3 (9°/s) commercial turn with margin, floored
-    at 20° for sample noise, effectively unconstrained past ~16 s
-    (capped at 180° — the maximum circular distance).
-    """
-    return max(20.0, min(dt * 10.0, 180.0))
-
-
-class PipeDecoder:
+class PipeDecoder(ValidationMixin):
     """Stateful Mode-S decoder with per-ICAO state and CPR pair matching.
 
     Args:
@@ -529,229 +453,6 @@ def decode(
         self._update_state(icao, message.df, result, timestamp)
         return result
 
-    def _reject_on_altitude_mismatch(
-        self,
-        result: Decoded,
-        icao: str,
-        timestamp: float,
-    ) -> bool:
-        """Flag & scrub this DF20 result if its AC-code altitude disagrees
-        with the recent ADS-B-anchored altitude for the same ICAO.
-
-        Returns True if the message was rejected (caller should skip
-        state updates). No-ops if we have no ADS-B anchor yet or the
-        message has no altitude field.
-        """
-        cb_alt = result.get("altitude")
-        if cb_alt is None:
-            return False
-        anchor = self._adsb_altitude.get(icao)
-        if anchor is None:
-            return False
-        adsb_t, adsb_alt = anchor
-        dt = abs(timestamp - adsb_t)
-        if dt > self._eviction_ttl:
-            return False  # anchor too stale to trust
-        if abs(cb_alt - adsb_alt) <= _altitude_tolerance(dt):
-            return False
-
-        result["altitude_mismatch"] = True
-        for key in _BDS_FIELDS_TO_CLEAR:
-            if key in result:
-                result[key] = None
-        self._stats["altitude_mismatch"] += 1
-        return True
-
-    def _reject_df17_altitude_mismatch(
-        self,
-        result: Decoded,
-        icao: str,
-        timestamp: float,
-    ) -> bool:
-        """Flag & scrub a CRC-valid DF17/18 BDS 0,5 position whose
-        altitude disagrees with the recent ADS-B altitude anchor.
-
-        A FRUIT phantom that survives CRC can land with plausible ICAO
-        bits but a garbage AC-code altitude. The header altitude is
-        preserved (so callers see what was flagged); the CPR fields are
-        cleared so the phantom can't pair with nearby real frames and
-        produce a far-off position that the motion check would catch
-        only after more work.
-
-        Returns True if rejected (caller skips anchor update and all
-        downstream state mutations).
-        """
-        new_alt = result.get("altitude")
-        if new_alt is None:
-            return False
-        anchor = self._adsb_altitude.get(icao)
-        if anchor is None:
-            return False
-        adsb_t, adsb_alt = anchor
-        dt = abs(timestamp - adsb_t)
-        if dt > self._eviction_ttl:
-            return False
-        if abs(new_alt - adsb_alt) <= _altitude_tolerance(dt):
-            return False
-
-        result["altitude_mismatch"] = True
-        for key in ("cpr_lat", "cpr_lon", "cpr_format"):
-            if key in result:
-                result[key] = None
-        self._stats["altitude_mismatch"] += 1
-        return True
-
-    def _reject_velocity_mismatch(
-        self,
-        result: Decoded,
-        icao: str,
-        timestamp: float,
-    ) -> bool:
-        """Flag & scrub this TC=19 result if gs/track disagrees with the
-        recent anchor, OR if the decoded vertical rate exceeds the
-        physical envelope of a commercial aircraft (~±6000 fpm; we use
-        ±10 000 fpm as an absolute cutoff).
-
-        Returns True if rejected (caller skips state/anchor update).
-        The VR check fires even without an anchor and even when gs/track
-        pass, so a phantom whose gs/track match the anchor but whose VR
-        is garbage (-24 704 fpm from a single corrupted frame) is still
-        caught.
-        """
-        vr = result.get("vertical_rate")
-        vr_implausible = vr is not None and abs(vr) > 10000.0
-
-        anchor_mismatch = False
-        gs = result.get("groundspeed")
-        track = result.get("track")
-        if gs is not None and track is not None:
-            anchor = self._adsb_velocity.get(icao)
-            if anchor is not None:
-                a_t, a_gs, a_track = anchor
-                dt = abs(timestamp - a_t)
-                if dt <= self._eviction_ttl:
-                    d_gs = abs(gs - a_gs)
-                    d_track = abs(track - a_track) % 360.0
-                    if d_track > 180.0:
-                        d_track = 360.0 - d_track
-                    if d_gs > _groundspeed_tolerance(dt) or d_track > _track_tolerance(
-                        dt
-                    ):
-                        anchor_mismatch = True
-
-        if not (vr_implausible or anchor_mismatch):
-            return False
-
-        result["velocity_mismatch"] = True
-        # Scrub velocity fields — the caller would otherwise merge them
-        # into state, propagating the phantom's values downstream.
-        for key in (
-            "groundspeed",
-            "track",
-            "vertical_rate",
-            "heading",
-            "airspeed",
-            "airspeed_type",
-            "velocity_type",
-        ):
-            if key in result:
-                result[key] = None
-        self._stats["velocity_mismatch"] += 1
-        return True
-
-    def _reject_bds50_velocity_mismatch(
-        self,
-        result: Decoded,
-        icao: str,
-        timestamp: float,
-    ) -> bool:
-        """Flag & scrub a DF20/21 BDS 5,0 reply whose gs or true_track
-        disagrees with the per-ICAO ADS-B velocity anchor.
-
-        Mirrors `_reject_velocity_mismatch` but reads `true_track`
-        (BDS 5,0's heading field) instead of `track` (TC=19's).
-
-        Returns True if rejected (caller skips handle_cpr_pair and
-        _update_state so the phantom's values don't propagate).
-        """
-        gs = result.get("groundspeed")
-        track = result.get("true_track")
-        if gs is None and track is None:
-            return False
-        anchor = self._adsb_velocity.get(icao)
-        if anchor is None:
-            return False
-        a_t, a_gs, a_track = anchor
-        dt = abs(timestamp - a_t)
-        if dt > self._eviction_ttl:
-            return False
-
-        gs_ok = True
-        track_ok = True
-        if gs is not None:
-            gs_ok = abs(gs - a_gs) <= _groundspeed_tolerance(dt)
-        if track is not None:
-            d_track = abs(track - a_track) % 360.0
-            if d_track > 180.0:
-                d_track = 360.0 - d_track
-            track_ok = d_track <= _track_tolerance(dt)
-
-        if gs_ok and track_ok:
-            return False
-
-        result["velocity_mismatch"] = True
-        # Scrub the whole BDS payload — the fields might individually
-        # look valid but the frame is from a different aircraft.
-        for key in _BDS_FIELDS_TO_CLEAR:
-            if key in result:
-                result[key] = None
-        self._stats["velocity_mismatch"] += 1
-        return True
-
-    def _reject_bds60_heading_mismatch(
-        self,
-        result: Decoded,
-        icao: str,
-        timestamp: float,
-    ) -> bool:
-        """Flag & scrub a DF20/21 BDS 6,0 reply whose magnetic_heading
-        disagrees with the per-ICAO ADS-B track anchor by more than
-        magnetic variation + wind-correction angle can legitimately
-        explain.
-
-        Tolerance: ``max(60°, min(Δt · 10°, 180°))``. The 60° floor
-        covers worst-case magnetic variation (±20°) + strong-crosswind
-        wind-correction angle (±20°) + short lag during turns; the
-        per-second slope matches the rate-3 turn envelope used elsewhere.
-
-        Returns True if rejected.
-        """
-        hdg = result.get("magnetic_heading")
-        if hdg is None:
-            return False
-        anchor = self._adsb_velocity.get(icao)
-        if anchor is None:
-            return False
-        a_t, _a_gs, a_track = anchor
-        dt = abs(timestamp - a_t)
-        if dt > self._eviction_ttl:
-            return False
-
-        d = abs(hdg - a_track) % 360.0
-        if d > 180.0:
-            d = 360.0 - d
-
-        tol = max(60.0, min(dt * 10.0, 180.0))
-        if d <= tol:
-            return False
-
-        result["velocity_mismatch"] = True
-        for key in _BDS_FIELDS_TO_CLEAR:
-            if key in result:
-                result[key] = None
-        self._stats["velocity_mismatch"] += 1
-        return True
-
     def _maybe_evict_expired(self, now: float) -> None:
         """Run a full eviction sweep only when its interval is due.