test(swing_speed): cover the capture loop's swing detection state machine

tests/test_swing_speed.py

@@ -1,7 +1,9 @@
 """Tests for swing speed training mode."""
 
+import logging
 from datetime import datetime
 
+from openflight import swing_speed
 from openflight.ops243 import Direction, SpeedReading
 from openflight.swing_speed import SwingSpeedEvent, SwingSpeedMonitor
 
@@ -187,3 +189,260 @@ def test_select_swing_reading_rejects_speeds_above_configured_max():
     selected = monitor._select_swing_reading(readings)  # pylint: disable=protected-access
 
     assert selected is readings[1]
+
+
+# ---------------------------------------------------------------------------
+# Capture-loop helpers
+#
+# The capture loop decides when a swing starts and ends purely from elapsed
+# time, so these tests replace the module's `time` reference with a controlled
+# clock. That keeps the shipped defaults (end_quiet_ms=1000, cooldown_ms=750)
+# under test while running instantly and deterministically.
+# ---------------------------------------------------------------------------
+
+ITERATION_MARGIN = 50
+
+
+class _FakeClock:
+    """Stand-in for the `time` module used inside swing_speed.
+
+    Also bounds the loop. `_capture_loop` swallows every exception, so if the
+    scripted radar ever stops being reached (a renamed radar method, say) the
+    loop would spin forever on a clock that only advances in memory. Every
+    iteration reads `time.time()` first, so counting those reads caps the run
+    and lets `_run_loop` fail loudly instead of hanging the suite.
+    """
+
+    def __init__(self, monitor, max_ticks: int, start: float = 1_700_000_000.0):
+        self._now = start
+        self._monitor = monitor
+        self._max_ticks = max_ticks
+        self.ticks = 0
+        self.overran = False
+
+    def time(self) -> float:
+        self.ticks += 1
+        if self.ticks > self._max_ticks:
+            self.overran = True
+            self._monitor._running = False  # pylint: disable=protected-access
+        return self._now
+
+    def sleep(self, seconds: float) -> None:
+        self._now += seconds
+
+
+class _ErrorRecorder(logging.Handler):
+    """Captures ERROR records so a swallowed loop exception fails the test loudly."""
+
+    def __init__(self):
+        super().__init__(level=logging.ERROR)
+        self.records = []
+
+    def emit(self, record):
+        self.records.append(record)
+
+
+class _ScriptedRadar:
+    """Replays scripted radar batches, then halts the monitor."""
+
+    def __init__(self, batches, singles=None):
+        self.batches = list(batches)
+        self.singles = list(singles or [])
+        self.monitor = None
+        self.candidate_calls = 0
+        self.single_calls = 0
+
+    def read_speed_candidates_nonblocking(self):
+        self.candidate_calls += 1
+        if self.batches:
+            return self.batches.pop(0)
+        if self.monitor is not None:
+            self.monitor._running = False  # pylint: disable=protected-access
+        return []
+
+    def read_speed_nonblocking(self):
+        self.single_calls += 1
+        return self.singles.pop(0) if self.singles else None
+
+
+def _reading(speed, direction=Direction.OUTBOUND, magnitude=None):
+    """A radar reading with no timestamp, so the loop stamps it from the clock."""
+    return SpeedReading(speed=speed, direction=direction, magnitude=magnitude, timestamp=None)
+
+
+class _Idle:
+    """Placeholder for `ms` of quiet loop time, sized against the monitor's poll interval."""
+
+    def __init__(self, ms: float):
+        self.ms = ms
+
+
+def _idle(ms):
+    """Empty batches spanning `ms` of loop time, expanded by `_run_loop`."""
+    return [_Idle(ms)]
+
+
+def _expand(batches, poll_interval_ms):
+    """Replace each `_Idle` marker with the empty batches that span its duration."""
+    expanded = []
+    for batch in batches:
+        if isinstance(batch, _Idle):
+            expanded.extend([] for _ in range(int(batch.ms / poll_interval_ms) + 1))
+        else:
+            expanded.append(batch)
+    return expanded
+
+
+def _run_loop(monkeypatch, monitor, batches, singles=None):
+    """Drive `_capture_loop` once over a scripted radar and a controlled clock."""
+    expanded = _expand(batches, monitor.poll_interval_ms)
+    clock = _FakeClock(monitor, max_ticks=len(expanded) + ITERATION_MARGIN)
+    monkeypatch.setattr(swing_speed, "time", clock)
+    radar = _ScriptedRadar(expanded, singles)
+    radar.monitor = monitor
+    monitor.radar = radar
+    monitor._running = True  # pylint: disable=protected-access
+
+    recorder = _ErrorRecorder()
+    swing_speed.logger.addHandler(recorder)
+    try:
+        monitor._capture_loop()  # pylint: disable=protected-access
+    finally:
+        swing_speed.logger.removeHandler(recorder)
+
+    # The loop re-raises the same fault on every iteration, so report each
+    # distinct message once; otherwise one regression buries CI in a message
+    # repeated hundreds of times.
+    distinct_errors = list(dict.fromkeys(record.getMessage() for record in recorder.records))
+    assert not distinct_errors, (
+        f"_capture_loop logged {len(recorder.records)} error(s): " + "; ".join(distinct_errors)
+    )
+    assert not clock.overran, (
+        f"_capture_loop ran past {clock.ticks} iterations without stopping; "
+        "the scripted radar is no longer terminating it"
+    )
+    return radar
+
+
+def test_capture_loop_starts_swing_on_first_qualifying_reading(monkeypatch):
+    """Behaviour 1: the swing's trigger speed is the first qualifying reading, not the peak."""
+    monitor = SwingSpeedMonitor()
+    batches = [[_reading(78.0)], [_reading(91.0)], [_reading(85.0)], *_idle(1100)]
+
+    _run_loop(monkeypatch, monitor, batches)
+
+    events = monitor.get_events()
+    assert len(events) == 1
+    assert events[0].trigger_speed_mph == 78.0
+    assert events[0].peak_speed_mph == 91.0
+
+
+def test_capture_loop_accumulates_readings_into_a_single_swing(monkeypatch):
+    """Behaviour 2: consecutive qualifying readings join one swing, not several."""
+    monitor = SwingSpeedMonitor()
+    speeds = [72.0, 88.0, 94.0, 81.0]
+    batches = [[_reading(speed)] for speed in speeds] + _idle(1100)
+
+    _run_loop(monkeypatch, monitor, batches)
+
+    events = monitor.get_events()
+    assert len(events) == 1
+    assert events[0].reading_count == len(speeds)
+    assert events[0].peak_speed_mph == 94.0
+    # Untimestamped readings are stamped from the loop's own clock, so the span
+    # runs from the first qualifying reading to the last, three polls later.
+    expected_ms = (len(speeds) - 1) * monitor.poll_interval_ms
+    assert round(events[0].duration_ms, 3) == expected_ms
+
+
+def test_capture_loop_waits_for_end_quiet_ms_before_emitting(monkeypatch):
+    """Behaviour 3: a swing is only finalised after end_quiet_ms of silence."""
+    swing = [[_reading(70.0)], [_reading(80.0)], [_reading(75.0)]]
+
+    too_short = SwingSpeedMonitor()
+    _run_loop(monkeypatch, too_short, swing + _idle(900))
+    assert too_short.get_events() == []
+
+    long_enough = SwingSpeedMonitor()
+    _run_loop(monkeypatch, long_enough, swing + _idle(1100))
+    assert len(long_enough.get_events()) == 1
+
+
+def test_capture_loop_rejects_motion_with_too_few_readings(monkeypatch):
+    """Behaviour 4: brief motion below the single-peak threshold is discarded as noise."""
+    monitor = SwingSpeedMonitor()
+    batches = [[_reading(45.0)], [_reading(48.0)], *_idle(1100)]
+
+    _run_loop(monkeypatch, monitor, batches)
+
+    assert monitor.get_events() == []
+
+
+def test_capture_loop_accepts_a_single_reading_above_the_peak_threshold(monkeypatch):
+    """Behaviour 5: one very fast reading counts even below min_readings."""
+    monitor = SwingSpeedMonitor()
+    batches = [[_reading(95.0)], *_idle(1100)]
+
+    _run_loop(monkeypatch, monitor, batches)
+
+    events = monitor.get_events()
+    assert len(events) == 1
+    assert events[0].reading_count == 1
+    assert events[0].peak_speed_mph == 95.0
+
+
+def test_capture_loop_suppresses_readings_during_cooldown(monkeypatch):
+    """Behaviour 6: readings inside cooldown_ms cannot open a second swing."""
+    monitor = SwingSpeedMonitor()
+    first_swing = [[_reading(70.0)], [_reading(85.0)], [_reading(80.0)]]
+    during_cooldown = [[_reading(72.0)], [_reading(86.0)], [_reading(79.0)]]
+    batches = first_swing + _idle(1100) + during_cooldown + _idle(1100)
+
+    _run_loop(monkeypatch, monitor, batches)
+
+    assert len(monitor.get_events()) == 1
+
+
+def test_capture_loop_forwards_selected_readings_to_live_callback(monkeypatch):
+    """Behaviour 7: each selected reading reaches the live callback."""
+    monitor = SwingSpeedMonitor()
+    live = []
+    monitor._live_callback = live.append  # pylint: disable=protected-access
+    batches = [[_reading(70.0)], [_reading(85.0)], *_idle(1100)]
+
+    _run_loop(monkeypatch, monitor, batches)
+
+    assert [reading.speed for reading in live] == [70.0, 85.0]
+
+
+def test_capture_loop_ignores_non_qualifying_readings(monkeypatch):
+    """Behaviour 8: inbound, too-slow and implausibly-fast readings never start a swing."""
+    monitor = SwingSpeedMonitor()
+    live = []
+    monitor._live_callback = live.append  # pylint: disable=protected-access
+    batches = [
+        [_reading(80.0, direction=Direction.INBOUND)],
+        [_reading(12.0)],
+        [_reading(400.0)],
+        *_idle(1100),
+    ]
+
+    _run_loop(monkeypatch, monitor, batches)
+
+    assert monitor.get_events() == []
+    assert live == []
+
+
+def test_capture_loop_falls_back_to_single_reading_when_batch_is_empty(monkeypatch):
+    """Behaviour 9: an empty candidate batch falls back to the single-reading call."""
+    monitor = SwingSpeedMonitor()
+    singles = [_reading(74.0), _reading(89.0), _reading(83.0)]
+    batches = [[], [], []] + _idle(1100)
+
+    radar = _run_loop(monkeypatch, monitor, batches, singles=singles)
+
+    events = monitor.get_events()
+    assert radar.single_calls > 0
+    assert len(events) == 1
+    assert events[0].reading_count == 3
+    assert events[0].peak_speed_mph == 89.0