221 lines
10 KiB
Python
221 lines
10 KiB
Python
"""Plot capture follows client demand and plot changes; redraws leave no stale pixels."""
|
|
|
|
from collections import deque
|
|
from contextlib import ExitStack
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
import numpy as np
|
|
import pyqtgraph as pg
|
|
from PyQt6.QtCore import QPoint, QSignalBlocker
|
|
from PyQt6.QtGui import QImage
|
|
from PyQt6.QtWidgets import QApplication, QCheckBox
|
|
|
|
from python_app.gui.controllers.renderers.sar3d import SAR3D_VIEWS
|
|
from python_app.models.dataset_model import ComboKey, ResultBlock, ResultCollection, ResultPayload, TraceData
|
|
from python_app.tests.gui_test_support import isolated_window, widget
|
|
|
|
|
|
def _phase_result(phase_deg: np.ndarray) -> ResultCollection:
|
|
trace = (0.5 * np.exp(1j * np.radians(phase_deg))).astype(np.complex64)
|
|
payload = ResultPayload("pass_through", 1, frequency_hz=np.linspace(1e9, 2e9, phase_deg.size), trace=trace)
|
|
return ResultCollection(1, 1, blocks=[ResultBlock(ComboKey(0, 0), [payload])])
|
|
|
|
|
|
def _gpr_result(revision: int, point: tuple[float, float], size: int) -> ResultCollection:
|
|
image = np.linspace(0, 1, size * size, dtype=np.float32).reshape(size, size)
|
|
return ResultCollection(1, 1, settings_revision=revision, collection_payloads=[
|
|
ResultPayload("gpr_accumulator", 3, image=image,
|
|
image_x_axis=np.linspace(-1.5, 1.5, size), image_y_axis=np.linspace(1, 12, size)),
|
|
ResultPayload("gpr_points", 4, table=np.array([[*point, 0.9]], dtype=np.float32)),
|
|
])
|
|
|
|
|
|
def _sar3d_result(revision: int, peak: tuple[float, float, float]) -> ResultCollection:
|
|
x, y, z = np.linspace(-1, 1, 9), np.linspace(-.5, .5, 5), np.linspace(1, 3, 9)
|
|
payloads = [ResultPayload(name, 3, image_x_axis=xx, image_y_axis=yy,
|
|
image=np.full((yy.size, xx.size), .5, dtype=np.float32))
|
|
for name, (xx, yy) in zip(SAR3D_VIEWS, ((x, y), (z, y), (x, z), (x, y)))]
|
|
return ResultCollection(1, 1, settings_revision=revision, collection_payloads=[
|
|
*payloads,
|
|
ResultPayload("sar3d_status", 4, table=np.zeros((1, 1), dtype=np.float32)),
|
|
ResultPayload("sar3d_oblique_basis", 4, table=np.array([[1, 0, 0], [0, 1, 0]], dtype=np.float32)),
|
|
ResultPayload("sar3d_peak", 4, table=np.array([[1, *peak]], dtype=np.float32)),
|
|
])
|
|
|
|
|
|
def _bscan_result(collection_id: int, depth_m: np.ndarray) -> ResultCollection:
|
|
trace = np.full(depth_m.size, collection_id, dtype=np.complex64)
|
|
payload = ResultPayload("bscan", 1, frequency_hz=depth_m.astype(np.float32), trace=trace)
|
|
return ResultCollection(collection_id, collection_id, capture_start_ns=collection_id,
|
|
capture_end_ns=collection_id + 1, blocks=[ResultBlock(ComboKey(0, 0), [payload])])
|
|
|
|
|
|
class WebPlotCaptureTest(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.app = QApplication.instance() or QApplication([])
|
|
cls.window = isolated_window(cls.addClassCleanup)
|
|
cls.window._pipeline.timer.stop()
|
|
cls.window._pipeline.replay.timer.stop()
|
|
cls.window.show()
|
|
|
|
def setUp(self) -> None:
|
|
cleanup = ExitStack()
|
|
self.addCleanup(cleanup.close)
|
|
self.w = self.window
|
|
self.web = self.w._web
|
|
self.web.set_frame_streaming(False)
|
|
self.grab = cleanup.enter_context(mock.patch.object(self.web, "_grab_png", wraps=self.web._grab_png))
|
|
|
|
def settle(self) -> None:
|
|
# Initial paints can settle the axis layout over several Qt event turns.
|
|
for _ in range(10):
|
|
self.app.processEvents()
|
|
self.web._last_grab_s = 0
|
|
self.web._grab_frame_if_due()
|
|
|
|
def show_mode(self, name: str) -> None:
|
|
mode = self.w._view.processing.mode
|
|
with QSignalBlocker(mode):
|
|
mode.setCurrentText(name)
|
|
self.w._live.active_mode = name
|
|
self.w._plots.set_mode(name)
|
|
|
|
def test_no_clients_never_grabs_and_first_client_gets_current_plot(self) -> None:
|
|
trace = TraceData(ComboKey(0, 0), np.array([1., 2.]), np.array([.1 + 0j, .2 + 0j]))
|
|
self.w._plots.draw_trace(trace, "Captured")
|
|
self.settle()
|
|
self.grab.assert_not_called()
|
|
self.show_mode("bscan")
|
|
self.w._plots.bscan.history_by_combo = {(2, 3): deque([np.array([1., 2.])])}
|
|
self.w._plots.sync_display_combos()
|
|
self.web.refresh_snapshot()
|
|
self.assertEqual(self.web.status()["display_combo"],
|
|
{"enabled": True, "input": 2, "output": 3, "options": [(2, 3)]})
|
|
self.grab.assert_not_called()
|
|
self.web.set_frame_streaming(True)
|
|
self.settle()
|
|
self.assertGreater(self.grab.call_count, 0)
|
|
self.assertIsNotNone(self.web._last_png)
|
|
calls = self.grab.call_count
|
|
self.settle()
|
|
self.assertEqual(self.grab.call_count, calls) # a grab does not trigger another one
|
|
self.web.set_frame_streaming(False)
|
|
self.w._plots.clear()
|
|
self.settle()
|
|
self.assertEqual(self.grab.call_count, calls)
|
|
|
|
def test_drawing_clearing_zoom_resize_and_page_switch_trigger_capture_after_idle(self) -> None:
|
|
plots = self.w._plots
|
|
self.show_mode("pass_through")
|
|
self.web.set_frame_streaming(True)
|
|
self.settle()
|
|
view = plots.trace.magnitude.view
|
|
actions = [
|
|
lambda: plots.draw(_phase_result(np.full(50, 30.0))),
|
|
lambda: view.setXRange(1.2e9, 1.5e9, padding=0),
|
|
plots.redraw,
|
|
plots.clear,
|
|
lambda: self.w.resize(self.w.width() + 50, self.w.height() + 50),
|
|
lambda: plots.set_mode("sar3d"),
|
|
lambda: plots.draw_trace(TraceData(ComboKey(0, 0), np.array([1., 2.]), np.array([.1 + 0j, .2 + 0j])),
|
|
"Captured"),
|
|
]
|
|
for action in actions:
|
|
with self.subTest(action=action):
|
|
calls = self.grab.call_count
|
|
action()
|
|
self.settle()
|
|
self.assertGreater(self.grab.call_count, calls)
|
|
calls = self.grab.call_count
|
|
self.settle()
|
|
self.assertEqual(self.grab.call_count, calls)
|
|
|
|
def test_throttle_defers_a_change_and_a_grab_absorbs_changes_it_renders(self) -> None:
|
|
plots = self.w._plots
|
|
self.web.set_frame_streaming(True)
|
|
self.settle()
|
|
self.grab.reset_mock()
|
|
plots.clear()
|
|
self.web._last_grab_s = 100
|
|
with mock.patch("python_app.gui.controllers.web_controller.time.monotonic", return_value=100.1):
|
|
self.assertIsNone(self.web._grab_frame_if_due())
|
|
self.grab.assert_not_called()
|
|
self.grab.side_effect = lambda: (plots.clear(), "changed-png")[1]
|
|
with mock.patch("python_app.gui.controllers.web_controller.time.monotonic", return_value=100.3):
|
|
self.assertIsNotNone(self.web._grab_frame_if_due())
|
|
self.assertEqual(self.web._grabbed_revision, plots.revision)
|
|
with mock.patch("python_app.gui.controllers.web_controller.time.monotonic", return_value=100.6):
|
|
self.assertIsNone(self.web._grab_frame_if_due())
|
|
self.assertEqual(self.grab.call_count, 1)
|
|
|
|
def test_failed_grab_is_retried(self) -> None:
|
|
plots = self.w._plots
|
|
self.web.set_frame_streaming(True)
|
|
self.settle()
|
|
plots.clear()
|
|
self.grab.reset_mock(side_effect=True)
|
|
self.grab.side_effect = RuntimeError("render failed")
|
|
self.web._last_grab_s = 0
|
|
with self.assertRaises(RuntimeError):
|
|
self.web._grab_frame_if_due()
|
|
self.assertNotEqual(self.web._grabbed_revision, plots.revision)
|
|
self.grab.side_effect = None
|
|
self.settle()
|
|
self.assertEqual(self.grab.call_count, 2)
|
|
self.assertEqual(self.web._grabbed_revision, plots.revision)
|
|
|
|
def assert_window_shows_fresh_render(self) -> None:
|
|
screen, page = QApplication.primaryScreen(), self.w._plots.stack.currentWidget()
|
|
assert screen is not None and page is not None
|
|
for plot in [page] if isinstance(page, pg.PlotWidget) else page.findChildren(pg.PlotWidget):
|
|
if plot.isVisible():
|
|
origin = plot.mapTo(self.w, QPoint())
|
|
shown = screen.grabWindow(self.w.winId(), origin.x(), origin.y(), plot.width(), plot.height())
|
|
self.assertEqual(shown.toImage().convertToFormat(QImage.Format.Format_RGB32),
|
|
plot.grab().toImage().convertToFormat(QImage.Format.Format_RGB32))
|
|
|
|
def test_redrawn_views_leave_no_stale_pixels_on_screen(self) -> None:
|
|
# Qt repaints only changed regions, yet the window must equal a fresh render: once
|
|
# a phase redrawn from 100 to 0 deg left the old curve visible where none was drawn.
|
|
plots, history = self.w._plots, self.w._history
|
|
for plot in plots.stack.findChildren(pg.PlotWidget):
|
|
scene = plot.scene()
|
|
assert scene is not None
|
|
self.assertEqual(scene.receivers(scene.changed), 0) # a receiver selects that update path
|
|
phase = widget(self.w._view.processing, "_show_phase_checkbox", QCheckBox)
|
|
self.addCleanup(phase.setChecked, phase.isChecked())
|
|
phase.setChecked(True)
|
|
self.web.set_frame_streaming(True)
|
|
revision = plots.required_revision or 0
|
|
later_phase = np.zeros(400)
|
|
later_phase[200:] = np.nan
|
|
|
|
def bscan(count: int, depth_m: np.ndarray) -> bool:
|
|
history.results.clear()
|
|
history.results.extend(_bscan_result(index, depth_m) for index in range(1, count + 1))
|
|
return plots.draw_bscan()
|
|
|
|
frames = {
|
|
"pass_through": (lambda: plots.draw(_phase_result(np.full(400, 100.0))),
|
|
lambda: plots.draw(_phase_result(later_phase))),
|
|
"gpr": (lambda: plots.draw(_gpr_result(revision, (-1.0, 4.0), 20)),
|
|
lambda: plots.draw(_gpr_result(revision, (1.0, 9.0), 8))),
|
|
"sar3d": (lambda: plots.draw(_sar3d_result(revision, (.6, .3, 2.6))),
|
|
lambda: plots.draw(_sar3d_result(revision, (-.6, -.3, 1.4)))),
|
|
"bscan": (lambda: bscan(6, np.linspace(.5, 3., 12)), lambda: bscan(3, np.linspace(.5, 2., 4))),
|
|
}
|
|
self.addCleanup(history.results.clear)
|
|
for mode, (first, second) in frames.items():
|
|
with self.subTest(mode=mode):
|
|
self.show_mode(mode)
|
|
for draw in (first, second):
|
|
self.assertTrue(draw())
|
|
self.settle()
|
|
self.assert_window_shows_fresh_render()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|