172 lines
7.2 KiB
Python
172 lines
7.2 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
|
|
`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()
|