mirror of
https://github.com/azaion/gps-denied-onboard.git
synced 2026-06-21 22:41:12 +00:00
Oleksandr Bezdieniezhnykh
bb744d9078
FT-P-12: parse mavproxy-listener tlog over a 60 s Derkachi replay and assert SUT->GCS GLOBAL_POSITION_INT cadence lands in [1, 2] Hz (AC-6.1). FT-P-13: inject `RELOC:<lat>,<lon>,<radius_m>` STATUSTEXT while the SUT is in dead_reckoned; verify FDR `c8.gcs.operator_command` ack <=2s, `anchor_search_region` centre shifts toward the hint, and no BAD_SIGNATURE / UNAUTHORIZED / REJECTED STATUSTEXT lands in the post-inject window (AC-6.2). Adds runner.helpers.gcs_telemetry_evaluator (rate, hint-ack correlation, haversine search-region shift, rejection scan) and sitl_observer.capture_gcs_tlog (parity surface to capture_ap_tlog). Pure-logic coverage: 39 new unit tests; full e2e/_unit_tests/ suite 746 passing (was 700). Scenarios skip locally on missing SITL replay fixture; production hooks (inbound STATUSTEXT parser, anchor_search_region FDR emitter) tracked outside this task. See _docs/03_implementation/batch_81_report.md + reviews/batch_81_review.md. Co-authored-by: Cursor <cursoragent@cursor.com>
110 lines
4.0 KiB
Python
110 lines
4.0 KiB
Python
"""FT-P-12 — GCS downsample at 1-2 Hz (AZ-420 / AC-6.1).
|
|
|
|
The full scenario:
|
|
|
|
1. Start the SUT against the SITL container; ``mavproxy-listener``
|
|
captures the SUT↔GCS link to ``${E2E_SITL_REPLAY_DIR}/gcs_tlog_<host>.tlog``.
|
|
2. Replay ``flight_derkachi.mp4`` for 60 s through the SUT's file frame
|
|
source so the C8 ``QgcTelemetryAdapter`` produces summary bursts.
|
|
3. After replay, parse the captured tlog for SUT-emitted
|
|
``GLOBAL_POSITION_INT`` (the position half of the QGC summary pair)
|
|
over the 60 s window.
|
|
4. AC-1: observed rate must land in [1, 2] Hz inclusive (AC-6.1).
|
|
5. AC-5: parameterised per ``(fc_adapter, vio_strategy)``.
|
|
|
|
Gated on:
|
|
|
|
* ``runner.helpers.frame_source_replay`` — owned by AZ-441 (still a
|
|
stub today; scenario skips via ``sitl_replay_ready``).
|
|
* ``runner.helpers.sitl_observer.capture_gcs_tlog`` — owned by AZ-420
|
|
(AP-side parity surface to ``capture_ap_tlog``; loads the
|
|
``gcs_tlog_<host>.tlog`` FDR-replay fixture).
|
|
* ``runner.helpers.gcs_telemetry_evaluator.compute_gcs_summary_rate`` —
|
|
pure-logic evaluator covered by
|
|
``e2e/_unit_tests/helpers/test_gcs_telemetry_evaluator.py``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from runner.helpers import gcs_telemetry_evaluator as gte
|
|
from runner.helpers import mavproxy_tlog_reader as mtr
|
|
|
|
DERKACHI_DIR = (
|
|
Path(__file__).resolve().parents[3]
|
|
/ "_docs"
|
|
/ "00_problem"
|
|
/ "input_data"
|
|
/ "flight_derkachi"
|
|
)
|
|
DERKACHI_MP4 = DERKACHI_DIR / "flight_derkachi.mp4"
|
|
REPLAY_WINDOW_S = 60
|
|
|
|
|
|
@pytest.mark.traces_to("AC-6.1,AC-1,AC-5")
|
|
def test_ft_p_12_gcs_downsample(
|
|
fc_adapter: str,
|
|
vio_strategy: str,
|
|
evidence_dir, # type: ignore[no-untyped-def]
|
|
run_id: str,
|
|
nfr_recorder, # type: ignore[no-untyped-def]
|
|
sitl_replay_ready: bool,
|
|
) -> None:
|
|
"""Full FT-P-12 scenario (AC-6.1). See module docstring.
|
|
|
|
AC-1: GCS rate ∈ [1, 2] Hz over the 60 s window — covered by
|
|
``compute_gcs_summary_rate``; unit-tested in
|
|
``test_gcs_telemetry_evaluator.py``.
|
|
AC-5: parameterised across ``(fc_adapter, vio_strategy)``.
|
|
"""
|
|
if not sitl_replay_ready:
|
|
pytest.skip(
|
|
"FT-P-12 full replay requires `E2E_SITL_REPLAY_DIR` to point at a "
|
|
"prepared SITL replay fixture exposing `gcs_tlog_<host>.tlog` "
|
|
"(AZ-595 + AZ-420 fixture builder). Pure-logic AC-6.1 coverage "
|
|
"lives in e2e/_unit_tests/helpers/test_gcs_telemetry_evaluator.py."
|
|
)
|
|
|
|
from runner.helpers import sitl_observer
|
|
from runner.helpers.frame_source_replay import FrameSourceReplayer
|
|
|
|
# 1. Drive replay; the mavproxy-listener captures the GCS link in
|
|
# parallel via the docker-compose fixture wiring (no in-process
|
|
# work needed here — the listener writes to disk).
|
|
sitl_host = "sitl-ardupilot" if fc_adapter == "ardupilot" else "sitl-inav"
|
|
FrameSourceReplayer(_resolve_frame_sink()).replay_video(DERKACHI_MP4)
|
|
tlog_path = sitl_observer.capture_gcs_tlog(host=sitl_host, duration_s=REPLAY_WINDOW_S)
|
|
|
|
# 2. Materialise the tlog once (iter_messages is single-pass).
|
|
msgs = gte.collect_messages_to_list(mtr.iter_messages(tlog_path))
|
|
if not msgs:
|
|
pytest.fail(f"FT-P-12: empty GCS tlog at {tlog_path}")
|
|
|
|
# 3. AC-1: GCS summary rate.
|
|
rate = gte.compute_gcs_summary_rate(msgs)
|
|
|
|
# 4. NFR metrics.
|
|
nfr_recorder.record_metric(
|
|
"ft_p_12.gcs_summary_rate_hz", rate.observed_rate_hz, ac_id="AC-6.1"
|
|
)
|
|
nfr_recorder.record_metric(
|
|
"ft_p_12.gcs_summary_messages", float(rate.total_summary_messages), ac_id="AC-6.1"
|
|
)
|
|
|
|
# 5. AC-1 assertion.
|
|
assert rate.passes, (
|
|
f"AC-6.1 (GCS rate ∈ [{rate.min_required_hz}, {rate.max_required_hz}] Hz) "
|
|
f"failed: observed_rate_hz={rate.observed_rate_hz:.3f}, "
|
|
f"messages={rate.total_summary_messages}, window_us={rate.window_us}"
|
|
)
|
|
|
|
|
|
def _resolve_frame_sink(): # type: ignore[no-untyped-def]
|
|
"""Return a replay-mode `FrameSink` (counter-only; AZ-597)."""
|
|
from runner.helpers.replay_mode import NullFrameSink
|
|
|
|
return NullFrameSink()
|