gps-denied-onboard/tests/e2e/replay/_helpers.py

"""Helpers shared by the AZ-404 E2E replay tests.

The numerical kernels (``l2_horizontal_m``, ``match_percentage``,
``GroundTruthRow``) moved into production code at
:mod:`gps_denied_onboard.helpers.gps_compare` in AZ-697; they're
re-exported here so existing import sites stay stable.

* :func:`parse_jsonl` — read the ``JsonlReplaySink`` output into a list
  of dicts with one entry per emit.
* :class:`CapturingMavlinkTransport` — test-only ``MavlinkTransport``
  impl that records every ``write`` so AC-4b can compare the byte
  streams produced by ``compose_root(config_live)`` vs.
  ``compose_root(config_replay)``.
* :func:`load_ground_truth_csv` — the IMU CSV's ``GLOBAL_POSITION_INT``
  columns ARE the AC-3 reference (the original tlog's GPS rows
  exported to CSV); this helper materialises them. Retained for the
  CSV-only fallback path; the real-tlog branch uses
  :func:`gps_denied_onboard.replay_input.load_tlog_ground_truth`
  instead.

All functions are pure / deterministic and stay safely importable on
dev macOS without ``RUN_REPLAY_E2E``; the regular regression suite
calls them via the unit-level helper test in this module's sibling
``test_helpers.py``.
"""

from __future__ import annotations

import csv
import json
from pathlib import Path
from typing import Any

from gps_denied_onboard.helpers.gps_compare import (
    GroundTruthRow,
    l2_horizontal_m,
    match_percentage,
)

__all__ = [
    "CapturingMavlinkTransport",
    "GroundTruthRow",
    "l2_horizontal_m",
    "load_ground_truth_csv",
    "match_percentage",
    "parse_jsonl",
]


def parse_jsonl(path: Path) -> list[dict[str, Any]]:
    """Return one dict per line of a JsonlReplaySink output file.

    Empty trailing lines are tolerated (orjson always terminates with
    ``\\n`` so the last newline is followed by ``""``); other empty
    lines indicate a corrupt file and surface as a JSON decode error.
    """
    records: list[dict[str, Any]] = []
    with path.open(encoding="utf-8") as fp:
        for lineno, line in enumerate(fp, start=1):
            stripped = line.rstrip("\n")
            if not stripped:
                continue
            try:
                records.append(json.loads(stripped))
            except json.JSONDecodeError as exc:
                raise AssertionError(
                    f"line {lineno} in {path} is not valid JSON: {exc.msg!r}"
                ) from exc
    return records


def load_ground_truth_csv(csv_path: Path) -> list[GroundTruthRow]:
    """Load the Derkachi IMU CSV's GPS rows as ground truth.

    The original ``flight_derkachi.tlog``'s ``GLOBAL_POSITION_INT``
    messages were exported to ``data_imu.csv``; the ``lat / lon /
    alt`` columns are degrees * 1e7 / metres * 1e3 (mavlink integer
    encoding), so we divide accordingly.
    """
    rows: list[GroundTruthRow] = []
    with csv_path.open(newline="") as fp:
        reader = csv.DictReader(fp)
        for r in reader:
            rows.append(
                GroundTruthRow(
                    t_s=float(r["Time"]),
                    lat_deg=float(r["GLOBAL_POSITION_INT.lat"]) / 1e7,
                    lon_deg=float(r["GLOBAL_POSITION_INT.lon"]) / 1e7,
                    alt_m=float(r["GLOBAL_POSITION_INT.alt"]) / 1e3,
                )
            )
    return rows


class CapturingMavlinkTransport:
    """Test-only :class:`MavlinkTransport` that records every write.

    Used by AZ-404 AC-4b: capture the byte streams produced by
    ``compose_root(config_live).c8.emit_external_position(out)`` and
    ``compose_root(config_replay).c8.emit_external_position(out)`` to
    assert byte-identity per replay protocol Invariant 5.

    NOTE: AC-4b is currently SKIPPED (blocked on AZ-558 — the C8
    encoders still bypass the ``MavlinkTransport`` seam by calling
    ``mav.*_send`` directly). This class is in place so the test
    fixture is ready the moment AZ-558 lands.
    """

    def __init__(self) -> None:
        self._chunks: list[bytes] = []
        self._closed = False

    def write(self, payload: bytes) -> int:
        if self._closed:
            raise RuntimeError("CapturingMavlinkTransport.write after close")
        self._chunks.append(bytes(payload))
        return len(payload)

    def bytes_written(self) -> int:
        return sum(len(c) for c in self._chunks)

    def close(self) -> None:
        self._closed = True

    @property
    def captured_payloads(self) -> tuple[bytes, ...]:
        """Tuple of every payload passed to :meth:`write`, in order."""
        return tuple(self._chunks)

    @property
    def captured_concat(self) -> bytes:
        """All captured payloads concatenated — the wire-byte stream."""
        return b"".join(self._chunks)