"""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 `collection_id` floor. Widening only the first would have changed nothing, because the floor kept filtering older collections out for good. The floor is gone: selection is positional, which is the only criterion that holds when ids are sparse (the results ring drops) or restart from 1 (a new C++ run). These tests pin the observable consequences — how many columns end up on screen — rather than any internal counter. """ 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.gui.runtime.history import record_result_history # 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_render_signature = None def tearDown(self) -> None: self.w._supervisor.is_running = self._original_is_running self.w._result_history.clear() self.w._bscan_render_signature = None 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_brings_older_frames_back(self) -> None: # The regression this whole change exists for: a live run renders 50 columns, # and widening the window after Stop must reach back over the retained history # rather than stay pinned to whatever the live path last drew. self._fill_history(300) self._set_running(True) self.w._sync_bscan_history_from_results() self.assertEqual( len(self.w._bscan_history_by_combo[(0, 0)]), self.w._bscan_cpp_replay_window ) 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) def test_restarted_collection_ids_do_not_hide_the_fresh_run(self) -> None: # Regression: Start after a stopped review made the image melt to 0 columns and # then snap back to 50. A new C++ run numbers from 1, so entries that are newest # by position carry the smallest ids; any id-based cut-off derived from the old # run rejected exactly them. self._fill_history(300) self._set_running(False) self.w._bscan_history_window.setValue(300) self.w._sync_bscan_history_from_results() self._set_running(True) window = self.w._bscan_cpp_replay_window for fresh in range(1, window + 1): self.w._result_history.append(_bscan_result(fresh)) self.w._sync_bscan_history_from_results() self.assertEqual(len(self.w._bscan_history_by_combo[(0, 0)]), window) 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. Counting in ids rather than # in entries 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_reprocessed_results_replace_rather_than_duplicate_columns(self) -> None: # The processor's recompute is triggered by feeding sweeps back through it, and # it republishes them under their ORIGINAL ids. Two independent mechanisms keep # that from doubling every column, and neither may be confused with the removed # `collection_id` floor: a threshold cannot tell a duplicate from its original, # since they share the id. Deduplication is by key equality. self._set_running(False) self.w._bscan_history_window.setValue(300) self._fill_history(200) self.w._sync_bscan_history_from_results() self.assertEqual(len(self.w._bscan_history_by_combo[(0, 0)]), 200) # Intake: a recomputed collection replaces the entry holding the same key. for collection in list(self.w._result_history): record_result_history(self.w._result_history, _bscan_result(collection.collection_id)) self.assertEqual(len(self.w._result_history), 200) # Render: a duplicate that reached the deque by another path is still collapsed, # keeping the newer of the two. self.w._result_history.append(_bscan_result(200)) self.w._bscan_render_signature = None self.w._sync_bscan_history_from_results() self.assertEqual(len(self.w._bscan_history_by_combo[(0, 0)]), 200) 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()