141 lines
5.5 KiB
Python
141 lines
5.5 KiB
Python
"""Configurable B-scan display window.
|
|
|
|
The B-scan used to be pinned to the C++ replay window (~50 sweeps) by two separate
|
|
mechanisms: the render-side history limit and a monotonically rising
|
|
`_bscan_history_floor_collection_id`. Widening only the first would have changed
|
|
nothing, because the floor kept filtering older collections out for good.
|
|
|
|
These tests pin the two properties that make the stopped-mode review work: the
|
|
window follows acquisition state, and the floor is recomputed (not ratcheted) so a
|
|
widened window can bring already-discarded frames back into view.
|
|
"""
|
|
|
|
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 _bscan_result(collection_id: int) -> ResultCollection:
|
|
payload = ResultPayload(
|
|
processing_name="bscan",
|
|
kind=1,
|
|
frequency_hz=np.array([0.5, 1.0], dtype=np.float32),
|
|
trace=np.array([collection_id + 0j, collection_id + 0j], dtype=np.complex64),
|
|
)
|
|
block = ResultBlock(combo=ComboKey(input=0, output=0), payloads=[payload])
|
|
return ResultCollection(collection_id=collection_id, monotonic_ns=collection_id, blocks=[block])
|
|
|
|
|
|
class BscanDisplayWindowTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.w = _window
|
|
self._original_is_running = self.w._supervisor.is_running
|
|
self.w._result_history.clear()
|
|
self.w._bscan_history_floor_collection_id = 0
|
|
|
|
def tearDown(self) -> None:
|
|
self.w._supervisor.is_running = self._original_is_running
|
|
self.w._result_history.clear()
|
|
self.w._bscan_history_floor_collection_id = 0
|
|
|
|
def _set_running(self, running: bool) -> None:
|
|
self.w._supervisor.is_running = lambda: running
|
|
|
|
def _fill_history(self, count: int, *, id_step: int = 1) -> None:
|
|
for index in range(count):
|
|
self.w._result_history.append(_bscan_result(1 + index * id_step))
|
|
|
|
def test_running_acquisition_ignores_the_user_window(self) -> None:
|
|
# A live rebuild runs on every incoming result, so the live path stays pinned
|
|
# to the replay window no matter what the operator typed for stopped review.
|
|
self._set_running(True)
|
|
self.w._bscan_history_window.setValue(300)
|
|
self.assertEqual(self.w._bscan_display_window_scans(), self.w._bscan_cpp_replay_window)
|
|
|
|
def test_stopped_acquisition_uses_the_user_window(self) -> None:
|
|
self._set_running(False)
|
|
self.w._bscan_history_window.setValue(300)
|
|
self.assertEqual(self.w._bscan_display_window_scans(), 300)
|
|
|
|
def test_widening_the_window_lowers_the_floor(self) -> None:
|
|
# The regression this whole change exists for: a run leaves the floor high,
|
|
# and widening the window afterwards must pull it back down.
|
|
self._fill_history(300)
|
|
self._set_running(True)
|
|
self.w._advance_bscan_floor_to_display_window()
|
|
raised_floor = self.w._bscan_history_floor_collection_id
|
|
self.assertEqual(raised_floor, 300 - self.w._bscan_cpp_replay_window)
|
|
|
|
self._set_running(False)
|
|
self.w._bscan_history_window.setValue(300)
|
|
self.w._advance_bscan_floor_to_display_window()
|
|
self.assertEqual(self.w._bscan_history_floor_collection_id, 0)
|
|
|
|
def test_floor_survives_collection_ids_restarting(self) -> None:
|
|
# A new C++ run restarts ids from 1. The unconditional recompute must not
|
|
# leave a stale high floor that hides the whole fresh run.
|
|
self._fill_history(300)
|
|
self._set_running(True)
|
|
self.w._advance_bscan_floor_to_display_window()
|
|
self.assertGreater(self.w._bscan_history_floor_collection_id, 0)
|
|
|
|
self.w._result_history.clear()
|
|
self._fill_history(5)
|
|
self.w._advance_bscan_floor_to_display_window()
|
|
self.assertEqual(self.w._bscan_history_floor_collection_id, 0)
|
|
|
|
def test_window_counts_entries_not_collection_ids(self) -> None:
|
|
# The results ring overwrites unread slots when the producer outruns the GUI
|
|
# poll loop, so retained collection ids are sparse. A floor of
|
|
# `latest_id - window` then spans far fewer than `window` entries: this is
|
|
# exactly the case where asking for 150 sweeps rendered only 87.
|
|
self._fill_history(300, id_step=3)
|
|
self._set_running(False)
|
|
self.w._bscan_history_window.setValue(150)
|
|
self.w._sync_bscan_history_from_results()
|
|
self.assertEqual(len(self.w._bscan_history_by_combo[(0, 0)]), 150)
|
|
|
|
def test_rebuild_renders_the_full_widened_window(self) -> None:
|
|
self._fill_history(300)
|
|
self._set_running(False)
|
|
self.w._bscan_history_window.setValue(300)
|
|
self.w._sync_bscan_history_from_results()
|
|
self.assertEqual(len(self.w._bscan_history_by_combo[(0, 0)]), 300)
|
|
|
|
self.w._bscan_history_window.setValue(50)
|
|
self.w._sync_bscan_history_from_results()
|
|
self.assertEqual(len(self.w._bscan_history_by_combo[(0, 0)]), 50)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|