95 lines
3.6 KiB
Python
95 lines
3.6 KiB
Python
"""Single-capture completion logic.
|
|
|
|
The single capture starts a one-shot pipeline and must finish on exactly the result
|
|
that belongs to its sweep. Because collection ids restart per orchestrator run, a stale
|
|
result from a previous run can carry the SAME id; the start timestamp disambiguates them.
|
|
These tests pin that boundary (fresh vs stale same-id) and the readiness state machine —
|
|
the path that previously hung on the simulator when start_ns was recorded too late.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
from PyQt6.QtWidgets import QApplication # noqa: E402
|
|
|
|
from python_app.gui.app_window import AppWindow # noqa: E402
|
|
from python_app.models.dataset_model import ( # noqa: E402
|
|
ComboKey,
|
|
ResultBlock,
|
|
ResultCollection,
|
|
ResultPayload,
|
|
)
|
|
|
|
_app: QApplication | None = None
|
|
_window: AppWindow | None = None
|
|
|
|
|
|
def setUpModule() -> None:
|
|
global _app, _window
|
|
_app = QApplication.instance() or QApplication([])
|
|
_window = AppWindow(Path("."))
|
|
|
|
|
|
def tearDownModule() -> None:
|
|
if _window is not None:
|
|
_window.close()
|
|
|
|
|
|
def _result(collection_id: int, monotonic_ns: int) -> ResultCollection:
|
|
payload = ResultPayload(
|
|
processing_name="pass_through", kind=1,
|
|
frequency_hz=np.array([1.0, 2.0], dtype=np.float32),
|
|
trace=np.array([1 + 1j, 2 - 2j], dtype=np.complex64),
|
|
)
|
|
block = ResultBlock(combo=ComboKey(input=0, output=0), payloads=[payload])
|
|
return ResultCollection(collection_id=collection_id, monotonic_ns=monotonic_ns, blocks=[block])
|
|
|
|
|
|
class SingleCaptureBoundaryTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.w = _window
|
|
self.w._processing_mode.setCurrentText("pass_through")
|
|
self.w._single_capture_active = True
|
|
self.w._single_capture_start_ns = 1000
|
|
self.w._single_capture_seen_raw = True
|
|
self.w._single_capture_target_collection_id = 5
|
|
self.w._result_history.clear()
|
|
|
|
def test_picks_fresh_result_not_stale_same_id(self) -> None:
|
|
self.w._result_history.append(_result(5, 500)) # stale: same id, before start_ns
|
|
self.w._result_history.append(_result(5, 1500)) # fresh: at/after start_ns
|
|
target = self.w._find_single_capture_target_result()
|
|
self.assertIsNotNone(target)
|
|
self.assertEqual(target.monotonic_ns, 1500)
|
|
|
|
def test_stale_same_id_result_is_not_matched(self) -> None:
|
|
self.w._result_history.append(_result(5, 500)) # only a stale same-id result
|
|
self.assertIsNone(self.w._find_single_capture_target_result())
|
|
|
|
def test_wrong_collection_id_is_not_matched(self) -> None:
|
|
self.w._result_history.append(_result(9, 1500)) # fresh but a different sweep
|
|
self.assertIsNone(self.w._find_single_capture_target_result())
|
|
|
|
def test_not_ready_until_a_fresh_raw_is_seen(self) -> None:
|
|
# The readiness guard that the timing fix makes reachable: with no raw recognized
|
|
# as fresh yet, the capture must not complete (it must keep waiting, not hang).
|
|
self.w._single_capture_seen_raw = False
|
|
self.w._result_history.append(_result(5, 1500))
|
|
self.assertFalse(self.w._finish_single_capture_if_ready())
|
|
self.assertTrue(self.w._single_capture_active)
|
|
|
|
def test_completes_and_stops_on_the_fresh_result(self) -> None:
|
|
self.w._result_history.append(_result(5, 1500))
|
|
self.assertTrue(self.w._finish_single_capture_if_ready())
|
|
self.assertFalse(self.w._single_capture_active) # _stop_run cleared the flag
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|