from unittest import mock from unittest.mock import MagicMock import numpy as np import pyqtgraph as pg import pytest from bec_lib.data_api.models import SourceData, SubscriptionUpdate from qtpy.QtCore import QPointF, Qt from bec_widgets.widgets.plots.image.bec_histogram_lut_item import ( BECColorBarItem, BECHistogramLUTItem, ) from bec_widgets.widgets.plots.image.image import Image from bec_widgets.widgets.plots.image.image_processor import ImageProcessor, ProcessingConfig from tests.unit_tests.client_mocks import mocked_client from tests.unit_tests.conftest import create_widget ################################################## # Image widget base functionality tests ################################################## def _set_signal_config( client, device: str, signal_name: str, signal_class: str, ndim: int, obj_name: str | None = None ): device = client.device_manager.devices[device] device._info["signals"][signal_name] = { "obj_name": obj_name or signal_name, "signal_class": signal_class, "component_name": signal_name, "describe": {"signal_info": {"ndim": ndim}}, } def _fake_bridge_factory(monkeypatch): """Replace the widget's QtDataSubscription with a recording fake.""" created = [] class _FakeBridge: def __init__( self, client, sources, scan="live", parent=None, min_emit_interval=0.1, max_points=None ): self.client = client self.sources = list(sources) self.scan = scan self.max_points = max_points self.healthy = True self.closed = False self.updated = MagicMock() def close(self): self.closed = True def factory(client, sources, scan="live", parent=None, min_emit_interval=0.1, max_points=None): bridge = _FakeBridge(client, sources, scan=scan, max_points=max_points) created.append(bridge) return bridge monkeypatch.setattr("bec_widgets.widgets.plots.image.image.QtDataSubscription", factory) return created def _make_source( device, entry, values, *, kind="async", ordinals=None, metadata=None, as_numpy=False ): values = np.asarray(values) if as_numpy else tuple(values) if ordinals is None: ordinals = tuple(range(len(values))) wrap = np.asarray if as_numpy else tuple return SourceData( device=device, entry=entry, kind=kind, ordinals=wrap(ordinals), values=values, timestamps=wrap([float(i) for i in range(len(values))]), complete=True, metadata=metadata or {}, ) def _make_update(source, scan_id="scan_1", reason="live"): return SubscriptionUpdate( scan_id=scan_id, reason=reason, sources={source.key: source}, aligned_ordinals=source.ordinals, complete=True, metadata={"group": "standalone"}, ) def _preview_update(device, entry, values, *, ordinals=None, scan_id=None): """Build an update as delivered by a scan-less device_preview stream.""" metadata = {"stream": "preview"} if scan_id is not None: metadata["scan_id"] = scan_id source = _make_source( device, entry, values, kind="unindexed", ordinals=ordinals, metadata=metadata ) return _make_update(source, scan_id="") class _FakeClickEvent: """Minimal stand-in for a pyqtgraph MouseClickEvent.""" def __init__(self, button=Qt.LeftButton): self._button = button self.accepted = False def button(self): return self._button def accept(self): self.accepted = True def test_initialization_defaults(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) assert bec_image_view.color_map == "plasma" assert bec_image_view.autorange is True assert bec_image_view.autorange_mode == "mean" assert bec_image_view.config.lock_aspect_ratio is True assert bec_image_view.main_image is not None assert bec_image_view._color_bar is None def test_setting_color_map(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.color_map = "viridis" assert bec_image_view.color_map == "viridis" assert bec_image_view.config.color_map == "viridis" def test_invalid_color_map_handling(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) previous_colormap = bec_image_view.color_map bec_image_view.color_map = "invalid_colormap_name" assert bec_image_view.color_map == previous_colormap assert bec_image_view.main_image.color_map == previous_colormap def test_toggle_autorange(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.autorange = False assert bec_image_view.autorange is False bec_image_view.toggle_autorange(True, "max") assert bec_image_view.autorange is True assert bec_image_view.autorange_mode == "max" assert bec_image_view.main_image.autorange is True assert bec_image_view.main_image.autorange_mode == "max" assert bec_image_view.main_image.config.autorange is True assert bec_image_view.main_image.config.autorange_mode == "max" def test_lock_aspect_ratio(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.lock_aspect_ratio = True assert bec_image_view.lock_aspect_ratio is True assert bool(bec_image_view.plot_item.getViewBox().state["aspectLocked"]) is True assert bec_image_view.config.lock_aspect_ratio is True def test_set_vrange(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.v_range = (10, 100) assert bec_image_view.v_range == QPointF(10, 100) assert bec_image_view.main_image.levels == (10, 100) assert bec_image_view.main_image.config.v_range == (10, 100) @pytest.mark.parametrize("colorbar_type", [None, "simple", "full"]) def test_set_vrange_keeps_fractional_values(qtbot, mocked_client, colorbar_type): """Regression: tuple ranges went through QPoint, which truncates to int, so e.g. v_max = 10.5 could never produce a non-integer level.""" bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.main_image.set_data(np.arange(100, dtype=float).reshape(10, 10)) if colorbar_type is not None: bec_image_view.enable_colorbar(True, colorbar_type) bec_image_view.v_range = (0.25, 10.5) assert bec_image_view.v_range == QPointF(0.25, 10.5) assert bec_image_view.main_image.v_range == (0.25, 10.5) bec_image_view.v_max = 42.5 assert bec_image_view.main_image.v_range == (0.25, 42.5) bec_image_view.v_min = 0.75 assert bec_image_view.main_image.v_range == (0.75, 42.5) if colorbar_type == "simple": assert bec_image_view._color_bar.levels() == (0.75, 42.5) elif colorbar_type == "full": assert tuple(bec_image_view._color_bar.getLevels()) == (0.75, 42.5) def test_enable_simple_colorbar(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.enable_simple_colorbar = True assert bec_image_view.enable_simple_colorbar is True assert bec_image_view.config.color_bar == "simple" assert isinstance(bec_image_view._color_bar, pg.ColorBarItem) # Enabling color bar should not cancel autorange assert bec_image_view.autorange is True assert bec_image_view.autorange_mode == "mean" assert bec_image_view.main_image.autorange is True assert bec_image_view.main_image.autorange_mode == "mean" def test_enable_full_colorbar(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.enable_full_colorbar = True assert bec_image_view.enable_full_colorbar is True assert bec_image_view.config.color_bar == "full" assert isinstance(bec_image_view._color_bar, pg.HistogramLUTItem) # Enabling color bar should not cancel autorange assert bec_image_view.autorange is True assert bec_image_view.autorange_mode == "mean" assert bec_image_view.main_image.autorange is True assert bec_image_view.main_image.autorange_mode == "mean" def test_full_colorbar_uses_bec_histogram_lut_item(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.enable_full_colorbar = True color_bar = bec_image_view._color_bar assert isinstance(color_bar, BECHistogramLUTItem) # The confusing default pyqtgraph plot menu is replaced on the histogram view. assert color_bar.vb.getMenu(None) is color_bar._bec_menu def test_full_colorbar_frames_do_not_disable_autorange(qtbot, mocked_client): """Rendering frames must not count as a user level change: pyqtgraph's HistogramLUTItem snaps its region to the image levels on every setImage, which used to fire the drag-disable hook on each update.""" view = create_widget(qtbot, Image, client=mocked_client) view.enable_full_colorbar = True view.autorange = True rng = np.random.default_rng(7) with mock.patch.object(view, "_set_autorange", wraps=view._set_autorange) as spy: for frame in range(3): view.main_image.set_data(rng.random((20, 20)) * (frame + 1)) disable_calls = [call for call in spy.call_args_list if call.args and call.args[0] is False] assert not disable_calls assert view.autorange is True def test_full_colorbar_region_drag_disables_autorange(qtbot, mocked_client): """An actual mouse drag of the histogram region must still switch autorange off.""" view = create_widget(qtbot, Image, client=mocked_client) view.enable_full_colorbar = True view.autorange = True view.main_image.set_data(np.ones((20, 20))) region = view._color_bar.region region.moving = True # simulate an active mouse drag region.setRegion((0.1, 0.7)) region.moving = False assert view.autorange is False def test_colorbar_menu_set_levels_updates_vrange(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.enable_full_colorbar = True assert bec_image_view.autorange is True # Simulate the "Set levels…" context-menu action (image scaling via the # region); fractional values must survive (regression: QPoint truncation). bec_image_view._color_bar.sigColorLevelsChangeRequested.emit((0.5, 50.5)) assert bec_image_view.v_range == QPointF(0.5, 50.5) assert bec_image_view.main_image.levels == (0.5, 50.5) # Setting explicit levels disables autorange. assert bec_image_view.autorange is False def test_colorbar_menu_autoscale_enables_autorange(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.enable_full_colorbar = True bec_image_view.autorange = False # Simulate the "Autoscale levels" context-menu action. bec_image_view._color_bar.sigAutoLevelsRequested.emit() assert bec_image_view.autorange is True def test_colorbar_menu_colormap_updates_image(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.enable_full_colorbar = True # Simulate picking a colormap from the colorbar context menu. bec_image_view._color_bar.sigColorMapChangeRequested.emit("viridis") assert bec_image_view.color_map == "viridis" assert bec_image_view.config.color_map == "viridis" assert bec_image_view.main_image.color_map == "viridis" def test_simple_colorbar_uses_bec_colorbar_item(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.enable_simple_colorbar = True assert isinstance(bec_image_view._color_bar, BECColorBarItem) def test_simple_colorbar_menu_signals_update_image_state(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.enable_simple_colorbar = True # "Set levels…" applies explicit levels and disables autorange. bec_image_view._color_bar.sigColorLevelsChangeRequested.emit((1.0, 42.0)) assert bec_image_view.v_range == QPointF(1.0, 42.0) assert bec_image_view.autorange is False # "Autoscale levels" re-enables autorange. bec_image_view._color_bar.sigAutoLevelsRequested.emit() assert bec_image_view.autorange is True # Picking a colormap goes through the BEC colormap handling (config + image). bec_image_view._color_bar.sigColorMapChangeRequested.emit("viridis") assert bec_image_view.config.color_map == "viridis" assert bec_image_view.main_image.color_map == "viridis" @pytest.mark.parametrize( "first, second", [("full", "simple"), ("simple", "full"), ("full", "full")] ) def test_switching_colorbar_styles_preserves_manual_levels(qtbot, mocked_client, first, second): """Switching between colorbar styles must not reset manually set levels.""" bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.main_image.set_data(np.arange(100, dtype=float).reshape(10, 10)) bec_image_view.enable_colorbar(True, first) bec_image_view.v_range = (5.0, 77.0) assert bec_image_view.autorange is False bec_image_view.enable_colorbar(True, second) assert bec_image_view.autorange is False assert bec_image_view.v_range == QPointF(5.0, 77.0) assert bec_image_view.main_image.v_range == (5.0, 77.0) def test_switching_colorbar_styles_keeps_autorange(qtbot, mocked_client): """With autorange on, switching colorbar styles keeps autorange enabled.""" bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.main_image.set_data(np.arange(100, dtype=float).reshape(10, 10)) bec_image_view.enable_full_colorbar = True assert bec_image_view.autorange is True bec_image_view.enable_simple_colorbar = True assert bec_image_view.autorange is True @pytest.mark.parametrize("colorbar_type", ["simple", "full"]) def test_disable_colorbar_cleans_up(qtbot, mocked_client, colorbar_type): """Disabling the colorbar tears down its parentless menus (no leaked top-levels).""" bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.enable_colorbar(True, colorbar_type) color_bar = bec_image_view._color_bar bec_image_view.enable_colorbar(False) assert bec_image_view._color_bar is None assert bec_image_view.config.color_bar is None assert color_bar._cleaned_up_triggered is True def test_enable_colorbar_invalid_style_raises(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) with pytest.raises(ValueError, match="Invalid colorbar style"): bec_image_view.enable_colorbar(True, "fancy") @pytest.mark.parametrize("colorbar_type", ["simple", "full"]) def test_enable_colorbar_with_vrange(qtbot, mocked_client, colorbar_type): bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.enable_colorbar(True, colorbar_type, (0, 100)) if colorbar_type == "simple": assert isinstance(bec_image_view._color_bar, pg.ColorBarItem) assert bec_image_view.enable_simple_colorbar is True else: assert isinstance(bec_image_view._color_bar, pg.HistogramLUTItem) assert bec_image_view.enable_full_colorbar is True assert bec_image_view.config.color_bar == colorbar_type assert bec_image_view.v_range == QPointF(0, 100) assert bec_image_view.main_image.levels == (0, 100) assert bec_image_view._color_bar is not None ############################################## # Device/signal update mechanism def test_image_setup_preview_signal_1d(qtbot, mocked_client, monkeypatch): """ Ensure that calling .image() with a 1‑D PreviewSignal connects through a scan-less DataAPI bridge and accumulates rows into the waterfall buffer. """ bridges = _fake_bridge_factory(monkeypatch) view = create_widget(qtbot, Image, client=mocked_client) _set_signal_config( mocked_client, "waveform1d", "img", signal_class="PreviewSignal", ndim=1, obj_name="waveform1d_img", ) view.image(device="waveform1d", signal="img") # Subscriptions should indicate 1‑D preview connection sub = view.subscriptions["main"] assert sub.source == "device_monitor_1d" assert sub.monitor_type == "1d" assert view.device == "waveform1d" assert view.signal == "img" # Preview signals are scan-less device streams with bounded retention bridge = bridges[-1] assert bridge.scan is None assert bridge.sources == [("waveform1d", "img")] assert bridge.max_points == Image.PREVIEW_1D_MAX_ROWS # Simulate a waveform update from the DataAPI waveform = np.arange(25, dtype=float) view._on_data_update(_preview_update("waveform1d", "img", [waveform])) assert view.main_image.raw_data.shape == (1, 25) np.testing.assert_array_equal(view.main_image.raw_data[0], waveform) def test_image_setup_preview_signal_2d(qtbot, mocked_client, monkeypatch): """ Ensure that calling .image() with a 2‑D PreviewSignal connects through a scan-less DataAPI bridge and displays the newest frame. """ bridges = _fake_bridge_factory(monkeypatch) view = create_widget(qtbot, Image, client=mocked_client) _set_signal_config( mocked_client, "eiger", "img2d", signal_class="PreviewSignal", ndim=2, obj_name="eiger_img2d", ) view.image(device="eiger", signal="img2d") # Subscriptions should indicate 2‑D preview connection sub = view.subscriptions["main"] assert sub.source == "device_monitor_2d" assert sub.monitor_type == "2d" assert view.device == "eiger" assert view.signal == "img2d" bridge = bridges[-1] assert bridge.scan is None assert bridge.sources == [("eiger", "img2d")] assert bridge.max_points == Image.PREVIEW_2D_MAX_FRAMES # Only the newest frame of the stream is displayed test_data = np.arange(16, dtype=float).reshape(4, 4) view._on_data_update(_preview_update("eiger", "img2d", [np.zeros((4, 4)), test_data])) np.testing.assert_array_equal(view.main_image.image, test_data) def test_switching_device_replaces_preview_bridge(qtbot, mocked_client, monkeypatch): bridges = _fake_bridge_factory(monkeypatch) view = create_widget(qtbot, Image, client=mocked_client) _set_signal_config(mocked_client, "eiger", "img", signal_class="PreviewSignal", ndim=2) _set_signal_config(mocked_client, "waveform1d", "img", signal_class="PreviewSignal", ndim=2) view.image(device="eiger", signal="img") first = bridges[-1] assert first.sources == [("eiger", "img")] view.device = "waveform1d" assert first.closed is True assert bridges[-1].sources == [("waveform1d", "img")] assert bridges[-1].scan is None def test_switching_device_replaces_async_bridge(qtbot, mocked_client, monkeypatch): """ Verify that switching device while async_update=True closes the previous scan-scoped bridge before subscribing to the new device. """ bridges = _fake_bridge_factory(monkeypatch) view = create_widget(qtbot, Image, client=mocked_client) _set_signal_config( mocked_client, "eiger", "img", signal_class="AsyncSignal", ndim=2, obj_name="async_obj" ) _set_signal_config( mocked_client, "waveform1d", "img", signal_class="AsyncSignal", ndim=2, obj_name="async_obj" ) view.image(device="eiger", signal="img") assert view.async_update is True assert view.subscriptions["main"].async_signal_name == "async_obj" first = bridges[-1] assert first.scan == "live" assert first.sources == [("eiger", "async_obj")] view.device = "waveform1d" assert first.closed is True assert bridges[-1].scan == "live" assert bridges[-1].sources == [("waveform1d", "async_obj")] def test_switching_signal_replaces_preview_bridge(qtbot, mocked_client, monkeypatch): bridges = _fake_bridge_factory(monkeypatch) view = create_widget(qtbot, Image, client=mocked_client) _set_signal_config(mocked_client, "eiger", "img_a", signal_class="PreviewSignal", ndim=2) _set_signal_config(mocked_client, "eiger", "img_b", signal_class="PreviewSignal", ndim=2) view.image(device="eiger", signal="img_a") first = bridges[-1] assert first.sources == [("eiger", "img_a")] view.signal = "img_b" assert first.closed is True assert bridges[-1].sources == [("eiger", "img_b")] assert bridges[-1].scan is None def test_switching_signal_replaces_async_bridge(qtbot, mocked_client, monkeypatch): """ When the current monitor is an async signal, switching to a different signal must close the previous bridge and subscribe with the new signal's obj_name as the DataAPI entry. """ bridges = _fake_bridge_factory(monkeypatch) view = create_widget(qtbot, Image, client=mocked_client) _set_signal_config( mocked_client, "eiger", "img_a", signal_class="AsyncSignal", ndim=2, obj_name="async_obj_a" ) _set_signal_config( mocked_client, "eiger", "img_b", signal_class="AsyncSignal", ndim=2, obj_name="async_obj_b" ) view.image(device="eiger", signal="img_a") assert view.async_update is True assert view.subscriptions["main"].async_signal_name == "async_obj_a" assert view.subscriptions["main"].source == "device_monitor_2d" first = bridges[-1] assert first.sources == [("eiger", "async_obj_a")] # Switch to a different signal view.signal = "img_b" assert first.closed is True assert view.subscriptions["main"].async_signal_name == "async_obj_b" assert bridges[-1].sources == [("eiger", "async_obj_b")] assert bridges[-1].scan == "live" def test_preview_signals_skip_0d_entries(qtbot, mocked_client, monkeypatch): """ Preview/async combobox should omit 0‑D signals. """ view = create_widget(qtbot, Image, client=mocked_client) def fake_get(signal_class_filter): signal_classes = ( signal_class_filter if isinstance(signal_class_filter, (list, tuple, set)) else [signal_class_filter] ) if "PreviewSignal" in signal_classes: return [ ( "eiger", "sig0d", { "obj_name": "sig0d", "signal_class": "PreviewSignal", "describe": {"signal_info": {"ndim": 0}}, }, ), ( "eiger", "sig2d", { "obj_name": "sig2d", "signal_class": "PreviewSignal", "describe": {"signal_info": {"ndim": 2}}, }, ), ] return [] monkeypatch.setattr(view.client.device_manager, "get_bec_signals", fake_get) device_selection = view.toolbar.components.get_action("device_selection").widget device_selection.signal_combo_box.set_device("eiger") device_selection.signal_combo_box.update_signals_from_signal_classes() texts = [ device_selection.signal_combo_box.itemText(i) for i in range(device_selection.signal_combo_box.count()) ] assert "sig0d" not in texts assert "sig2d" in texts def test_image_async_signal_uses_obj_name(qtbot, mocked_client, monkeypatch): """ Verify async signals subscribe with their obj_name as the DataAPI entry and render the delivered fragments. """ bridges = _fake_bridge_factory(monkeypatch) view = create_widget(qtbot, Image, client=mocked_client) _set_signal_config( mocked_client, "eiger", "img", signal_class="AsyncSignal", ndim=1, obj_name="async_obj" ) view.image(device="eiger", signal="img") assert view.subscriptions["main"].async_signal_name == "async_obj" assert view.async_update is True bridge = bridges[-1] assert bridge.scan == "live" assert bridge.sources == [("eiger", "async_obj")] assert bridge.max_points is None # Rendering reads the source keyed by (device, obj_name) payload = np.array([1, 2, 3]) source = _make_source( "eiger", "async_obj", [payload], metadata={"async_update_type": "add", "max_shape": [None]} ) view._on_data_update(_make_update(source)) np.testing.assert_array_equal(view.main_image.raw_data, payload.reshape(1, 3)) def test_disconnect_clears_async_state(qtbot, mocked_client, monkeypatch): bridges = _fake_bridge_factory(monkeypatch) view = create_widget(qtbot, Image, client=mocked_client) _set_signal_config( mocked_client, "eiger", "img", signal_class="AsyncSignal", ndim=2, obj_name="async_obj" ) view.image(device="eiger", signal="img") assert view.async_update is True view.disconnect_monitor(device="eiger", signal="img") assert bridges[-1].closed is True assert view.subscriptions["main"].async_signal_name is None assert view.async_update is False ############################################## # Connection guardrails def test_image_setup_rejects_unsupported_signal_class(qtbot, mocked_client, monkeypatch): bridges = _fake_bridge_factory(monkeypatch) view = create_widget(qtbot, Image, client=mocked_client) _set_signal_config(mocked_client, "eiger", "img", signal_class="Signal", ndim=2) view.image(device="eiger", signal="img") assert view.subscriptions["main"].source is None assert view.subscriptions["main"].monitor_type is None assert view.async_update is False assert bridges == [] def test_image_disconnects_with_missing_entry(qtbot, mocked_client, monkeypatch): bridges = _fake_bridge_factory(monkeypatch) view = create_widget(qtbot, Image, client=mocked_client) _set_signal_config(mocked_client, "eiger", "img", signal_class="PreviewSignal", ndim=2) view.image(device="eiger", signal="img") assert view.device == "eiger" assert view.signal == "img" view.image(device="eiger", signal=None) assert view.device == "" assert view.signal == "" assert all(bridge.closed for bridge in bridges) def test_scan_rollover_resets_accumulation_and_crosshair(qtbot, mocked_client, monkeypatch): """A new scan id in the update stream starts a fresh 1D accumulation and resets the crosshair (replaces the legacy _handle_scan_change data path).""" _fake_bridge_factory(monkeypatch) view = create_widget(qtbot, Image, client=mocked_client) _set_signal_config( mocked_client, "eiger", "img", signal_class="AsyncSignal", ndim=1, obj_name="async_obj" ) view.image(device="eiger", signal="img") view._on_data_update( _make_update(_make_source("eiger", "async_obj", [np.arange(4)]), scan_id="scan_1") ) assert view.scan_id == "scan_1" assert view.main_image.raw_data.shape == (1, 4) view.hook_crosshair() reset_called = [] monkeypatch.setattr(view.crosshair, "reset", lambda: reset_called.append(True)) # First update of the new scan: the DataAPI delivers a fresh per-scan series view._on_data_update( _make_update(_make_source("eiger", "async_obj", [np.arange(6)]), scan_id="scan_2") ) assert view.old_scan_id == "scan_1" assert view.scan_id == "scan_2" assert reset_called == [True] assert view.main_image.raw_data.shape == (1, 6) def test_preview_scan_rollover_restricts_display_window(qtbot, mocked_client, monkeypatch): """Preview streams retain pre-rollover points; a scan change restricts the displayed accumulation to the newest point onward.""" _fake_bridge_factory(monkeypatch) view = create_widget(qtbot, Image, client=mocked_client) _set_signal_config(mocked_client, "waveform1d", "img", signal_class="PreviewSignal", ndim=1) view.image(device="waveform1d", signal="img") rows = [np.full(5, i, dtype=float) for i in range(3)] view._on_data_update(_preview_update("waveform1d", "img", rows, scan_id="scan_1")) assert view.scan_id == "scan_1" assert view.main_image.raw_data.shape == (3, 5) # The device stream keeps the old rows; the newest one belongs to scan_2. rows_after = rows + [np.full(5, 99.0)] view._on_data_update(_preview_update("waveform1d", "img", rows_after, scan_id="scan_2")) assert view.old_scan_id == "scan_1" assert view.scan_id == "scan_2" assert view.main_image.raw_data.shape == (1, 5) np.testing.assert_array_equal(view.main_image.raw_data[0], np.full(5, 99.0)) # Subsequent scan_2 rows extend the new accumulation. rows_more = rows_after + [np.full(5, 100.0)] view._on_data_update(_preview_update("waveform1d", "img", rows_more, scan_id="scan_2")) assert view.main_image.raw_data.shape == (2, 5) def test_scan_rollover_same_scan_noop(qtbot, mocked_client, monkeypatch): _fake_bridge_factory(monkeypatch) view = create_widget(qtbot, Image, client=mocked_client) _set_signal_config( mocked_client, "eiger", "img", signal_class="AsyncSignal", ndim=1, obj_name="async_obj" ) view.image(device="eiger", signal="img") view._on_data_update( _make_update(_make_source("eiger", "async_obj", [np.arange(3)]), scan_id="scan_1") ) view._on_data_update( _make_update( _make_source("eiger", "async_obj", [np.arange(3), np.arange(3)]), scan_id="scan_1" ) ) assert view.scan_id == "scan_1" assert view.old_scan_id is None assert view.main_image.raw_data.shape == (2, 3) def test_build_1d_buffer_incremental_matches_full_rebuild(qtbot, mocked_client, monkeypatch): """ The incremental waterfall buffer must equal the from-scratch construction across live appends (contiguous and gapped), an unchanged reused snapshot, an out-of-order hole-fill, a wider new row, a non-live reason and a scan change — and must only fall back to the full rebuild for the emissions that cannot be pure appends. """ _fake_bridge_factory(monkeypatch) view = create_widget(qtbot, Image, client=mocked_client) _set_signal_config( mocked_client, "eiger", "img", signal_class="AsyncSignal", ndim=1, obj_name="async_obj" ) view.image(device="eiger", signal="img") from_scratch = Image._rebuild_1d_buffer rebuilds = [] def counting(self, source): rebuilds.append(source) return from_scratch(self, source) monkeypatch.setattr(Image, "_rebuild_1d_buffer", counting) def reference(values): rows = [np.atleast_1d(np.asarray(value)) for value in values] width = max(row.shape[0] for row in rows) return np.vstack([np.pad(row, (0, width - row.shape[0])) for row in rows]) steps = [ # (scan_id, reason, values, ordinals, expected rebuild count so far) ("scan_1", "live", [[1.0, 2.0]], (0,), 1), # first emission seeds the cache ("scan_1", "live", [[1.0, 2.0], [3.0, 4.0]], (0, 1), 1), # append ("scan_1", "live", [[1.0, 2.0], [3.0, 4.0]], (0, 1), 1), # unchanged reused snapshot ("scan_1", "live", [[1.0, 2.0], [3.0, 4.0], [5.0]], (0, 1, 3), 1), # gapped short append # late hole-fill below the frontier -> full rebuild ("scan_1", "live", [[1.0, 2.0], [3.0, 4.0], [4.5], [5.0]], (0, 1, 2, 3), 2), # new row wider than the cached buffer -> full rebuild ("scan_1", "live", [[1.0, 2.0], [3.0, 4.0], [4.5], [5.0], [6.0] * 4], (0, 1, 2, 3, 4), 3), # non-live reason -> full rebuild ( "scan_1", "backfill", [[1.0, 2.0], [3.0, 4.0], [4.5], [5.0], [6.0] * 4], (0, 1, 2, 3, 4), 4, ), ("scan_2", "live", [[9.0, 9.0]], (0,), 5), # scan change -> full rebuild ("scan_2", "live", [[9.0, 9.0], [10.0, 11.0]], (0, 1), 5), # incremental resumes ] for scan_id, reason, values, ordinals, expected_rebuilds in steps: source = _make_source( "eiger", "async_obj", values, ordinals=ordinals, metadata={"async_update_type": "add", "max_shape": [None]}, ) view._on_data_update(_make_update(source, scan_id=scan_id, reason=reason)) np.testing.assert_array_equal(view.main_image.raw_data, reference(values)) assert len(rebuilds) == expected_rebuilds def test_image_data_update_2d(qtbot, mocked_client, monkeypatch): _fake_bridge_factory(monkeypatch) bec_image_view = create_widget(qtbot, Image, client=mocked_client) _set_signal_config(mocked_client, "eiger", "img", signal_class="PreviewSignal", ndim=2) bec_image_view.image(device="eiger", signal="img") test_data = np.random.rand(20, 30) bec_image_view._on_data_update(_preview_update("eiger", "img", [test_data])) np.testing.assert_array_equal(bec_image_view.main_image.image, test_data) def test_image_data_update_1d(qtbot, mocked_client, monkeypatch): _fake_bridge_factory(monkeypatch) bec_image_view = create_widget(qtbot, Image, client=mocked_client) _set_signal_config(mocked_client, "waveform1d", "img", signal_class="PreviewSignal", ndim=1) bec_image_view.image(device="waveform1d", signal="img") waveform1 = np.random.rand(50) waveform2 = np.random.rand(60) # Different length, tests padding logic bec_image_view._on_data_update(_preview_update("waveform1d", "img", [waveform1])) assert bec_image_view.main_image.raw_data.shape == (1, 50) bec_image_view._on_data_update(_preview_update("waveform1d", "img", [waveform1, waveform2])) assert bec_image_view.main_image.raw_data.shape == (2, 60) def test_async_add_slice_displays_accumulated_rows(qtbot, mocked_client, monkeypatch): """add_slice sources deliver one accumulated row per ordinal; shorter rows are zero-padded to the longest one.""" _fake_bridge_factory(monkeypatch) view = create_widget(qtbot, Image, client=mocked_client) _set_signal_config( mocked_client, "eiger", "img", signal_class="AsyncSignal", ndim=1, obj_name="async_obj" ) view.image(device="eiger", signal="img") source = _make_source( "eiger", "async_obj", [[1.0, 2.0, 3.0], [4.0]], metadata={"async_update_type": "add_slice", "max_shape": [None, None]}, ) view._on_data_update(_make_update(source)) np.testing.assert_array_equal( view.main_image.raw_data, np.array([[1.0, 2.0, 3.0], [4.0, 0.0, 0.0]]) ) def test_async_replace_displays_current_state(qtbot, mocked_client, monkeypatch): """replace sources expose a single element: the current full state.""" _fake_bridge_factory(monkeypatch) view = create_widget(qtbot, Image, client=mocked_client) _set_signal_config( mocked_client, "eiger", "img", signal_class="AsyncSignal", ndim=2, obj_name="async_obj" ) view.image(device="eiger", signal="img") frame = np.arange(12, dtype=float).reshape(3, 4) source = _make_source( "eiger", "async_obj", [frame], ordinals=(0,), metadata={"async_update_type": "replace", "max_shape": [3, 4]}, ) view._on_data_update(_make_update(source)) np.testing.assert_array_equal(view.main_image.image, frame) ############################################## # Toolbar and Actions Tests def test_toolbar_actions_presence(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) assert bec_image_view.toolbar.components.exists("image_autorange") assert bec_image_view.toolbar.components.exists("lock_aspect_ratio") assert bec_image_view.toolbar.components.exists("image_processing_fft") assert bec_image_view.toolbar.components.exists("device_selection") def test_auto_emit_syncs_image_toolbar_actions(qtbot, mocked_client): from unittest.mock import Mock bec_image_view = create_widget(qtbot, Image, client=mocked_client) fft_action = bec_image_view.toolbar.components.get_action("image_processing_fft").action log_action = bec_image_view.toolbar.components.get_action("image_processing_log").action transpose_action = bec_image_view.toolbar.components.get_action( "image_processing_transpose" ).action mock_handler = Mock() bec_image_view.property_changed.connect(mock_handler) bec_image_view.fft = True bec_image_view.log = True bec_image_view.transpose = True assert fft_action.isChecked() assert log_action.isChecked() assert transpose_action.isChecked() mock_handler.assert_any_call("fft", True) mock_handler.assert_any_call("log", True) mock_handler.assert_any_call("transpose", True) def test_image_processing_fft_toggle(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.fft = True assert bec_image_view.fft is True bec_image_view.fft = False assert bec_image_view.fft is False def test_image_processing_log_toggle(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.log = True assert bec_image_view.log is True bec_image_view.log = False assert bec_image_view.log is False def test_image_rotation_and_transpose(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.num_rotation_90 = 2 assert bec_image_view.num_rotation_90 == 2 bec_image_view.transpose = True assert bec_image_view.transpose is True @pytest.mark.parametrize("colorbar_type", ["none", "simple", "full"]) def test_setting_vrange_with_colorbar(qtbot, mocked_client, colorbar_type): bec_image_view = create_widget(qtbot, Image, client=mocked_client) if colorbar_type == "simple": bec_image_view.enable_simple_colorbar = True elif colorbar_type == "full": bec_image_view.enable_full_colorbar = True bec_image_view.v_range = (0, 100) assert bec_image_view.v_range == QPointF(0, 100) assert bec_image_view.main_image.levels == (0, 100) assert bec_image_view.main_image.config.v_range == (0, 100) assert bec_image_view.v_min == 0 assert bec_image_view.v_max == 100 if colorbar_type == "simple": assert isinstance(bec_image_view._color_bar, pg.ColorBarItem) assert bec_image_view._color_bar.levels() == (0, 100) elif colorbar_type == "full": assert isinstance(bec_image_view._color_bar, pg.HistogramLUTItem) assert bec_image_view._color_bar.getLevels() == (0, 100) ################################### # Toolbar Actions ################################### def test_setup_image_from_toolbar(qtbot, mocked_client, monkeypatch): bridges = _fake_bridge_factory(monkeypatch) bec_image_view = create_widget(qtbot, Image, client=mocked_client) _set_signal_config(mocked_client, "eiger", "img", signal_class="PreviewSignal", ndim=2) monkeypatch.setattr( mocked_client.device_manager, "get_bec_signals", lambda signal_class_filter: ( [ ( "eiger", "img", { "obj_name": "img", "signal_class": "PreviewSignal", "describe": {"signal_info": {"ndim": 2}}, }, ) ] if "PreviewSignal" in (signal_class_filter or []) else [] ), ) device_selection = bec_image_view.toolbar.components.get_action("device_selection").widget device_selection.device_combo_box.update_devices_from_filters() device_selection.device_combo_box.setCurrentText("eiger") device_selection.signal_combo_box.setCurrentText("img") bec_image_view.on_device_selection_changed(None) qtbot.wait(200) assert bec_image_view.device == "eiger" assert bec_image_view.signal == "img" assert bec_image_view.subscriptions["main"].source == "device_monitor_2d" assert bec_image_view.subscriptions["main"].monitor_type == "2d" assert bec_image_view.main_image.raw_data is None assert bec_image_view.main_image.image is None def test_image_actions_interactions(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.autorange = False # Change the initial state to False bec_image_view.toolbar.components.get_action("image_autorange_mean").action.trigger() assert bec_image_view.autorange is True assert bec_image_view.main_image.autorange is True assert bec_image_view.autorange_mode == "mean" bec_image_view.toolbar.components.get_action("image_autorange_max").action.trigger() assert bec_image_view.autorange is True assert bec_image_view.main_image.autorange is True assert bec_image_view.autorange_mode == "max" bec_image_view.toolbar.components.get_action("lock_aspect_ratio").action.trigger() assert bec_image_view.lock_aspect_ratio is False assert bool(bec_image_view.plot_item.getViewBox().state["aspectLocked"]) is False def test_image_toggle_action_fft(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.toolbar.components.get_action("image_processing_fft").action.trigger() assert bec_image_view.fft is True assert bec_image_view.main_image.fft is True assert bec_image_view.main_image.config.processing.fft is True def test_image_toggle_action_log(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.toolbar.components.get_action("image_processing_log").action.trigger() assert bec_image_view.log is True assert bec_image_view.main_image.log is True assert bec_image_view.main_image.config.processing.log is True def test_image_toggle_action_transpose(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.toolbar.components.get_action("image_processing_transpose").action.trigger() assert bec_image_view.transpose is True assert bec_image_view.main_image.transpose is True assert bec_image_view.main_image.config.processing.transpose is True def test_image_toggle_action_rotate_right(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.toolbar.components.get_action("image_processing_rotate_right").action.trigger() assert bec_image_view.num_rotation_90 == 3 assert bec_image_view.main_image.num_rotation_90 == 3 assert bec_image_view.main_image.config.processing.num_rotation_90 == 3 def test_image_toggle_action_rotate_left(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.toolbar.components.get_action("image_processing_rotate_left").action.trigger() assert bec_image_view.num_rotation_90 == 1 assert bec_image_view.main_image.num_rotation_90 == 1 assert bec_image_view.main_image.config.processing.num_rotation_90 == 1 def test_image_toggle_action_reset(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) # Setup some processing bec_image_view.fft = True bec_image_view.log = True bec_image_view.transpose = True bec_image_view.num_rotation_90 = 2 bec_image_view.toolbar.components.get_action("image_processing_reset").action.trigger() assert bec_image_view.num_rotation_90 == 0 assert bec_image_view.main_image.num_rotation_90 == 0 assert bec_image_view.main_image.config.processing.num_rotation_90 == 0 assert bec_image_view.fft is False assert bec_image_view.main_image.fft is False assert bec_image_view.log is False assert bec_image_view.main_image.log is False assert bec_image_view.transpose is False assert bec_image_view.main_image.transpose is False def test_roi_add_remove_and_properties(qtbot, mocked_client): view = create_widget(qtbot, Image, client=mocked_client) # Add ROIs rect = view.add_roi(kind="rect", name="rect_roi", line_width=7) circ = view.add_roi(kind="circle", name="circ_roi", line_width=5) assert rect in view.roi_controller.rois assert circ in view.roi_controller.rois assert rect.label == "rect_roi" assert circ.label == "circ_roi" assert rect.line_width == 7 assert circ.line_width == 5 # Change properties rect.label = "rect_roi2" circ.line_color = "#ff0000" assert rect.label == "rect_roi2" assert circ.line_color == "#ff0000" # Remove by name view.remove_roi("rect_roi2") assert rect not in view.roi_controller.rois # Remove by index view.remove_roi(0) assert not view.roi_controller.rois def test_roi_controller_palette_signal(qtbot, mocked_client): view = create_widget(qtbot, Image, client=mocked_client) controller = view.roi_controller changed = [] controller.paletteChanged.connect(lambda cmap: changed.append(cmap)) view.add_roi(kind="rect") controller.colormap = "plasma" assert changed and changed[0] == "plasma" def test_roi_controller_clear_and_get_methods(qtbot, mocked_client): view = create_widget(qtbot, Image, client=mocked_client) r1 = view.add_roi(kind="rect", name="r1") r2 = view.add_roi(kind="circle", name="c1") controller = view.roi_controller assert controller.get_roi_by_name("r1") == r1 assert controller.get_roi(1) == r2 controller.clear() assert not controller.rois def test_roi_get_data_from_image_with_no_image(qtbot, mocked_client): view = create_widget(qtbot, Image, client=mocked_client) roi = view.add_roi(kind="rect") # Remove all images from scene for item in list(view.plot_item.items): if hasattr(item, "image"): view.plot_item.removeItem(item) with pytest.raises(RuntimeError): roi.get_data_from_image() ################################################## # Settings and popups ################################################## def test_show_roi_manager_popup(qtbot, mocked_client): """ Verify that the ROI-manager dialog opens and closes correctly, and that the matching toolbar icon stays in sync. """ view = create_widget(qtbot, Image, client=mocked_client, popups=True) # ROI-manager toggle is exposed via the toolbar. assert view.toolbar.components.exists("roi_mgr") roi_action = view.toolbar.components.get_action("roi_mgr").action assert roi_action.isChecked() is False, "Should start unchecked" # Open the popup. view.show_roi_manager_popup() assert view.roi_manager_dialog is not None assert view.roi_manager_dialog.isVisible() assert roi_action.isChecked() is True, "Icon should toggle on" # Close again. view.roi_manager_dialog.close() assert view.roi_manager_dialog is None assert roi_action.isChecked() is False, "Icon should toggle off" ################################### # ROI Plots & Crosshair Switch ################################### def test_crosshair_roi_panels_visibility(qtbot, mocked_client): """ Verify that enabling the ROI-crosshair shows ROI panels and disabling hides them. """ bec_image_view = create_widget(qtbot, Image, client=mocked_client) switch = bec_image_view.toolbar.components.get_action("image_switch_crosshair") # Initially panels should be hidden assert bec_image_view.side_panel_x.panel_height == 0 assert bec_image_view.side_panel_y.panel_width == 0 # Enable ROI crosshair switch.actions["crosshair_roi"].action.trigger() # Panels must be visible qtbot.waitUntil( lambda: all( [ bec_image_view.side_panel_x.panel_height > 0, bec_image_view.side_panel_y.panel_width > 0, ] ), timeout=500, ) # Disable ROI crosshair switch.actions["crosshair_roi"].action.trigger() # Panels hidden again qtbot.waitUntil( lambda: all( [ bec_image_view.side_panel_x.panel_height == 0, bec_image_view.side_panel_y.panel_width == 0, ] ), timeout=500, ) def test_crosshair_roi_switch_is_mutually_exclusive(qtbot, mocked_client): """ Switching the crosshair/ROI tool via the switcher menu must disable the previously active mode. In particular, switching from the ROI-crosshair back to the plain crosshair has to hide the ROI panels (and vice versa); the two modes are mutually exclusive. """ bec_image_view = create_widget(qtbot, Image, client=mocked_client) switch = bec_image_view.toolbar.components.get_action("image_switch_crosshair") crosshair_action = switch.actions["crosshair"].action crosshair_roi_action = switch.actions["crosshair_roi"].action def panels_visible() -> bool: return bec_image_view.side_panel_x.panel_height > 0 and ( bec_image_view.side_panel_y.panel_width > 0 ) # Activate the ROI-crosshair mode -> panels visible, crosshair hooked. switch.set_default_action("crosshair_roi") qtbot.waitUntil(panels_visible, timeout=500) assert bec_image_view.crosshair is not None assert crosshair_roi_action.isChecked() assert not crosshair_action.isChecked() # Switch to the plain crosshair -> ROI panels must hide (the regression). switch.set_default_action("crosshair") qtbot.waitUntil(lambda: not panels_visible(), timeout=500) assert bec_image_view.crosshair is not None # crosshair still active assert crosshair_action.isChecked() assert not crosshair_roi_action.isChecked() # Switch back to the ROI-crosshair -> panels visible again. switch.set_default_action("crosshair_roi") qtbot.waitUntil(panels_visible, timeout=500) assert crosshair_roi_action.isChecked() assert not crosshair_action.isChecked() def test_roi_plot_data_from_image(qtbot, mocked_client): """ Check that ROI plots receive correct slice data from the 2D image. """ import numpy as np bec_image_view = create_widget(qtbot, Image, client=mocked_client) # Provide deterministic 2D data test_data = np.arange(25).reshape(5, 5) bec_image_view._render_image_data(test_data) # Activate ROI crosshair switch = bec_image_view.toolbar.components.get_action("image_switch_crosshair") switch.actions["crosshair_roi"].action.trigger() qtbot.wait(50) # Simulate crosshair at row 2, col 3 bec_image_view.update_image_slices((0, 2, 3)) # Extract plotted data x_items = bec_image_view.x_roi.plot_item.listDataItems() y_items = bec_image_view.y_roi.plot_item.listDataItems() assert len(x_items) == 1 assert len(y_items) == 1 # Vertical slice (column) _, v_slice = x_items[0].getData() np.testing.assert_array_equal(v_slice, test_data[:, 3]) # Horizontal slice (row) h_slice, _ = y_items[0].getData() np.testing.assert_array_equal(h_slice, test_data[2]) def test_roi_plots_ignore_rgb_images_and_clear_stale_curves(qtbot, mocked_client): """RGB images have vector-valued pixels and cannot produce scalar profile curves.""" bec_image_view = create_widget(qtbot, Image, client=mocked_client) scalar_image = np.arange(25).reshape(5, 5) bec_image_view._render_image_data(scalar_image) switch = bec_image_view.toolbar.components.get_action("image_switch_crosshair") switch.actions["crosshair_roi"].action.trigger() qtbot.waitUntil(lambda: bec_image_view.crosshair is not None, timeout=1000) bec_image_view.update_image_slices((0, 2, 3)) bec_image_view.crosshair.set_pin(2.0, 3.0) assert bec_image_view.x_roi_curve is not None assert bec_image_view.y_roi_curve is not None assert bec_image_view.x_roi_pinned is not None assert bec_image_view.y_roi_pinned is not None rgb_image = np.zeros((5, 5, 3), dtype=np.uint8) bec_image_view._render_image_data(rgb_image) assert bec_image_view._compute_image_slices(bec_image_view.main_image, 2, 3) is None assert bec_image_view.x_roi_curve is None assert bec_image_view.y_roi_curve is None assert bec_image_view.x_roi_pinned is None assert bec_image_view.y_roi_pinned is None assert bec_image_view.x_roi.plot_item.listDataItems() == [] assert bec_image_view.y_roi.plot_item.listDataItems() == [] def test_pinned_roi_profiles_freeze_and_clear(qtbot, mocked_client): """Clicking pins a frozen copy of the X/Y profiles that survives live updates.""" import numpy as np bec_image_view = create_widget(qtbot, Image, client=mocked_client) test_data = np.arange(25).reshape(5, 5) bec_image_view._render_image_data(test_data) switch = bec_image_view.toolbar.components.get_action("image_switch_crosshair") switch.actions["crosshair_roi"].action.trigger() qtbot.wait(50) assert bec_image_view.x_roi_pinned is None assert bec_image_view.y_roi_pinned is None # Pin at pixel (row=2, col=3) -> freezes the corresponding profiles. bec_image_view.crosshair.set_pin(2.0, 3.0) assert bec_image_view.x_roi_pinned is not None assert bec_image_view.y_roi_pinned is not None _, x_pinned = bec_image_view.x_roi_pinned.getData() np.testing.assert_array_equal(x_pinned, test_data[:, 3]) y_pinned, _ = bec_image_view.y_roi_pinned.getData() np.testing.assert_array_equal(y_pinned, test_data[2]) # A live update at another location must NOT wipe the frozen pinned curves. pinned_item = bec_image_view.x_roi_pinned bec_image_view.update_image_slices((0, 1, 1)) assert bec_image_view.x_roi_pinned is pinned_item assert pinned_item in bec_image_view.x_roi.plot_item.listDataItems() # Clearing the pin removes the frozen curves. bec_image_view.crosshair.clear_pin() assert bec_image_view.x_roi_pinned is None assert bec_image_view.y_roi_pinned is None assert pinned_item not in bec_image_view.x_roi.plot_item.listDataItems() def test_pinned_roi_profiles_keep_style_on_theme_change(qtbot, mocked_client): """Theme changes re-pen the live profile curves but must not restyle the frozen (dashed) pinned reference curves.""" import numpy as np bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view._render_image_data(np.arange(25).reshape(5, 5)) switch = bec_image_view.toolbar.components.get_action("image_switch_crosshair") switch.actions["crosshair_roi"].action.trigger() qtbot.wait(50) bec_image_view.crosshair.set_pin(2.0, 3.0) bec_image_view.update_image_slices((0, 2, 3)) # ensure the live curve exists pinned = bec_image_view.x_roi_pinned assert pinned.opts["pen"].color().name() == "#f2c037" bec_image_view.x_roi.apply_theme("light") # The pinned reference keeps its amber styling, the live curve is re-penned. assert pinned.opts["pen"].color().name() == "#f2c037" assert bec_image_view.x_roi_curve.opts["pen"].color().name() == "#000000" def test_pin_survives_scan_reset(qtbot, mocked_client): """A scan transition (crosshair.reset()) must not wipe the pin or its frozen profiles.""" import numpy as np bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view._render_image_data(np.arange(25).reshape(5, 5)) switch = bec_image_view.toolbar.components.get_action("image_switch_crosshair") switch.actions["crosshair_roi"].action.trigger() qtbot.wait(50) bec_image_view.crosshair.set_pin(2.0, 3.0) assert bec_image_view.x_roi_pinned is not None pin_point = bec_image_view.crosshair.pinned_point assert pin_point is not None # Image._handle_scan_rollover calls crosshair.reset() on each new scan id. bec_image_view.crosshair.reset() # The pin marker and the frozen reference profiles must still be there. assert bec_image_view.crosshair.pinned_point is pin_point assert bec_image_view.crosshair.pinned_pos is not None assert bec_image_view.x_roi_pinned is not None assert bec_image_view.y_roi_pinned is not None def test_pin_and_profiles_restored_after_roi_toggle(qtbot, mocked_client): """Toggling crosshair-ROI off then on restores both the marker and its frozen profiles.""" import numpy as np bec_image_view = create_widget(qtbot, Image, client=mocked_client) test_data = np.arange(25).reshape(5, 5) bec_image_view._render_image_data(test_data) switch = bec_image_view.toolbar.components.get_action("image_switch_crosshair") switch.actions["crosshair_roi"].action.trigger() qtbot.wait(50) bec_image_view.crosshair.set_pin(2.0, 3.0) assert bec_image_view.x_roi_pinned is not None pin_point = bec_image_view.crosshair.pinned_point # Toggle the ROI crosshair off: marker persists, frozen curves are cleared. switch.actions["crosshair_roi"].action.trigger() qtbot.wait(20) assert bec_image_view.crosshair is None assert bec_image_view._detached_pin is not None assert bec_image_view.x_roi_pinned is None # Toggle back on: marker is re-adopted AND the frozen profiles are rebuilt. switch.actions["crosshair_roi"].action.trigger() qtbot.wait(20) assert bec_image_view.crosshair is not None assert bec_image_view.crosshair.pinned_point is pin_point assert bec_image_view.x_roi_pinned is not None assert bec_image_view.y_roi_pinned is not None _, x_pinned = bec_image_view.x_roi_pinned.getData() np.testing.assert_array_equal(x_pinned, test_data[:, 3]) def test_detached_pin_can_be_removed_with_right_click(qtbot, mocked_client, monkeypatch): """A pin left on the plot after disabling crosshair remains removable.""" from qtpy.QtWidgets import QMenu bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view._render_image_data(np.arange(25).reshape(5, 5)) switch = bec_image_view.toolbar.components.get_action("image_switch_crosshair") switch.actions["crosshair_roi"].action.trigger() qtbot.wait(50) bec_image_view.crosshair.set_pin(2.0, 3.0) switch.actions["crosshair_roi"].action.trigger() qtbot.wait(20) assert bec_image_view.crosshair is None assert bec_image_view._detached_pin is not None pin_point = bec_image_view._detached_pin["point"] cleared = [] bec_image_view.crosshair_pin_cleared.connect(lambda: cleared.append(True)) monkeypatch.setattr(QMenu, "exec_", lambda self, *a, **k: self.actions()[0]) event = _FakeClickEvent(button=Qt.RightButton) pin_point.mouseClickEvent(event) assert bec_image_view._detached_pin is None assert pin_point not in bec_image_view.plot_item.items assert event.accepted is True assert cleared == [True] def test_pinned_profiles_follow_image_updates(qtbot, mocked_client): """Pinned profile curves keep their position but follow incoming image data, exactly like the pin's intensity label.""" bec_image_view = create_widget(qtbot, Image, client=mocked_client) test_data = np.arange(25, dtype=float).reshape(5, 5) bec_image_view._render_image_data(test_data) switch = bec_image_view.toolbar.components.get_action("image_switch_crosshair") switch.actions["crosshair_roi"].action.trigger() qtbot.wait(50) bec_image_view.crosshair.set_pin(2.0, 3.0) _, x_pinned = bec_image_view.x_roi_pinned.getData() np.testing.assert_array_equal(x_pinned, test_data[:, 3]) updated = test_data + 100.0 bec_image_view._render_image_data(updated) qtbot.waitUntil( lambda: np.array_equal(bec_image_view.x_roi_pinned.getData()[1], updated[:, 3]), timeout=500 ) y_pinned, _ = bec_image_view.y_roi_pinned.getData() np.testing.assert_array_equal(y_pinned, updated[2]) def test_detached_pin_profiles_follow_image_updates(qtbot, mocked_client): """Pinned profiles keep following image data while the crosshair is toggled off.""" bec_image_view = create_widget(qtbot, Image, client=mocked_client) test_data = np.arange(25, dtype=float).reshape(5, 5) bec_image_view._render_image_data(test_data) switch = bec_image_view.toolbar.components.get_action("image_switch_crosshair") switch.actions["crosshair_roi"].action.trigger() qtbot.wait(50) bec_image_view.crosshair.set_pin(2.0, 3.0) bec_image_view.toggle_crosshair(False) # pin becomes detached, panels stay active assert bec_image_view.crosshair is None assert bec_image_view.x_roi_pinned is not None updated = test_data + 100.0 bec_image_view._render_image_data(updated) qtbot.waitUntil( lambda: np.array_equal(bec_image_view.x_roi_pinned.getData()[1], updated[:, 3]), timeout=500 ) def test_crosshair_moves_update_profiles_immediately(qtbot, mocked_client): """Crosshair moves update the live profiles synchronously; new frames refresh them at the current crosshair position.""" bec_image_view = create_widget(qtbot, Image, client=mocked_client) test_data = np.arange(25, dtype=float).reshape(5, 5) bec_image_view._render_image_data(test_data) switch = bec_image_view.toolbar.components.get_action("image_switch_crosshair") switch.actions["crosshair_roi"].action.trigger() qtbot.wait(50) bec_image_view.crosshair.plot_item.vb.setRange( xRange=(0, test_data.shape[0]), yRange=(0, test_data.shape[1]), padding=0 ) # Position the crosshair lines and emit the snapped pixel (as mouse_moved does). bec_image_view.crosshair.mouse_moved(manual_pos=(1.2, 2.2)) np.testing.assert_array_equal(bec_image_view.x_roi_curve.getData()[1], test_data[:, 2]) # New frames refresh the live profile at the crosshair position. updated = test_data + 100.0 bec_image_view._render_image_data(updated) np.testing.assert_array_equal(bec_image_view.x_roi_curve.getData()[1], updated[:, 2]) def test_live_label_intensity_updates_on_image_update(qtbot, mocked_client): """The live crosshair label refreshes its intensity when the image changes, not only when the mouse moves.""" bec_image_view = create_widget(qtbot, Image, client=mocked_client) test_data = np.arange(25, dtype=float).reshape(5, 5) bec_image_view._render_image_data(test_data) bec_image_view.hook_crosshair() bec_image_view.crosshair.plot_item.vb.setRange( xRange=(0, test_data.shape[0]), yRange=(0, test_data.shape[1]), padding=0 ) bec_image_view.crosshair.mouse_moved(manual_pos=(2.5, 3.5)) assert "Intensity: 13.000" in bec_image_view.crosshair.coord_label.toPlainText() bec_image_view._render_image_data(test_data + 100.0) assert "Intensity: 113.000" in bec_image_view.crosshair.coord_label.toPlainText() def test_active_pin_label_intensity_updates_on_image_update(qtbot, mocked_client): """Pinned crosshair label intensity follows image updates at the pinned position.""" bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view._render_image_data(np.arange(25).reshape(5, 5)) bec_image_view.hook_crosshair() bec_image_view.crosshair.set_pin(2.0, 3.0) assert bec_image_view.crosshair.pinned_label.toPlainText() == ( "pin (2.500, 3.500)\nIntensity: 13.000" ) bec_image_view._render_image_data(np.arange(25).reshape(5, 5) + 100) qtbot.waitUntil( lambda: bec_image_view.crosshair.pinned_label.toPlainText() == "pin (2.500, 3.500)\nIntensity: 113.000", timeout=500, ) def test_image_update_does_not_replay_active_crosshair_mouse_handling( qtbot, mocked_client, monkeypatch ): """Image updates refresh a pin without re-snapping or emitting live crosshair updates.""" bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view._render_image_data(np.arange(25).reshape(5, 5)) bec_image_view.hook_crosshair() bec_image_view.crosshair.set_pin(2.0, 3.0) mouse_moves = [] monkeypatch.setattr( bec_image_view.crosshair, "mouse_moved", lambda *args, **kwargs: mouse_moves.append((args, kwargs)), ) bec_image_view._render_image_data(np.arange(25).reshape(5, 5) + 100) qtbot.waitUntil( lambda: bec_image_view.crosshair.pinned_label.toPlainText() == "pin (2.500, 3.500)\nIntensity: 113.000", timeout=500, ) assert mouse_moves == [] def test_detached_pin_label_intensity_updates_on_image_update(qtbot, mocked_client): """Detached pin labels keep following image updates while crosshair is disabled.""" bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view._render_image_data(np.arange(25).reshape(5, 5)) bec_image_view.hook_crosshair() bec_image_view.crosshair.set_pin(2.0, 3.0) bec_image_view.unhook_crosshair() assert bec_image_view.crosshair is None assert bec_image_view._detached_pin is not None assert bec_image_view._detached_pin["label"].toPlainText() == ( "pin (2.500, 3.500)\nIntensity: 13.000" ) bec_image_view._render_image_data(np.arange(25).reshape(5, 5) + 200) qtbot.waitUntil( lambda: bec_image_view._detached_pin["label"].toPlainText() == "pin (2.500, 3.500)\nIntensity: 213.000", timeout=500, ) def test_detached_pin_label_omits_scalar_intensity_for_rgb_image(qtbot, mocked_client): """Detached RGB pin labels only show coordinates because a pixel is an array.""" bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view._render_image_data(np.zeros((5, 5, 3))) bec_image_view.hook_crosshair() bec_image_view.crosshair.set_pin(2.0, 3.0) bec_image_view.unhook_crosshair() assert bec_image_view.crosshair is None assert bec_image_view._detached_pin is not None assert bec_image_view._detached_pin["label"].toPlainText() == "pin (2.500, 3.500)" ############################################## # Device selection toolbar sync ############################################## def test_device_selection_syncs_from_properties(qtbot, mocked_client, monkeypatch): view = create_widget(qtbot, Image, client=mocked_client) _set_signal_config(mocked_client, "eiger", "img2d", signal_class="PreviewSignal", ndim=2) monkeypatch.setattr( view.client.device_manager, "get_bec_signals", lambda signal_class_filter: ( [ ( "eiger", "img2d", { "obj_name": "img2d", "signal_class": "PreviewSignal", "describe": {"signal_info": {"ndim": 2}}, }, ) ] if "PreviewSignal" in (signal_class_filter or []) else [] ), ) view.device = "eiger" view.signal = "img2d" qtbot.wait(200) # Allow signal processing device_selection = view.toolbar.components.get_action("device_selection").widget qtbot.waitUntil( lambda: device_selection.device_combo_box.currentText() == "eiger" and device_selection.signal_combo_box.currentText() == "img2d", timeout=1000, ) def test_signal_syncs_from_toolbar(qtbot, mocked_client): view = create_widget(qtbot, Image, client=mocked_client) _set_signal_config(mocked_client, "eiger", "img_a", signal_class="PreviewSignal", ndim=2) _set_signal_config(mocked_client, "eiger", "img_b", signal_class="PreviewSignal", ndim=2) view.device = "eiger" view.signal = "img_a" device_selection = view.toolbar.components.get_action("device_selection").widget device_selection.signal_combo_box.blockSignals(True) device_selection.signal_combo_box.setCurrentText("img_b") device_selection.signal_combo_box.blockSignals(False) view._sync_signal_from_toolbar() assert view.signal == "img_b" ############################################## # Log scale / non-finite data crash regression ############################################## @pytest.mark.parametrize( "data", [ -np.abs(np.random.rand(20, 30)).astype(np.float32) - 1.0, # all negative -> all NaN np.zeros((20, 30), dtype=np.uint16), # all zeros -> uniform floor np.where(np.eye(20, 30) > 0, -5.0, 3.0).astype(np.float32), # mixed sign ], ids=["all_negative", "all_zeros", "mixed_sign"], ) def test_log_scale_does_not_crash_with_full_colorbar(qtbot, mocked_client, data): """Log scaling of non-positive data must not produce NaN levels that crash the full (HistogramLUTItem) colorbar. Regression for fix/image-crashing.""" view = create_widget(qtbot, Image, client=mocked_client) view.enable_full_colorbar = True assert isinstance(view._color_bar, pg.HistogramLUTItem) view.log = True # This is the exact path that used to raise "Cannot set range [nan, nan]". view._render_image_data(data) vmin, vmax = view.main_image.v_range assert np.isfinite(vmin) and np.isfinite(vmax) assert vmin <= vmax # Colorbar levels stay finite as well. low, high = view._color_bar.getLevels() assert np.isfinite(low) and np.isfinite(high) def test_log_scale_all_negative_keeps_finite_image(qtbot, mocked_client): """All-negative input must still yield a finite, displayed image under log.""" view = create_widget(qtbot, Image, client=mocked_client) view.log = True data = (-np.abs(np.random.rand(15, 15)) - 1.0).astype(np.float32) view._render_image_data(data) assert view.main_image.image is not None assert np.all(np.isfinite(view.main_image.image)) def test_set_v_range_ignores_non_finite_levels(qtbot, mocked_client): """Non-finite v_range requests are rejected rather than forwarded to pg.""" view = create_widget(qtbot, Image, client=mocked_client) view._render_image_data(np.random.rand(10, 10)) good = view.main_image.v_range view.main_image.set_v_range((np.nan, np.nan)) assert view.main_image.v_range == good # unchanged # Inverted ranges are normalised, not propagated as-is. view.main_image.set_v_range((10.0, 2.0)) lo, hi = view.main_image.v_range assert (lo, hi) == (2.0, 10.0) def test_image_processor_log_is_finite_for_non_positive(): """ImageProcessor.log must never emit NaN/inf for zero/negative input.""" proc = ImageProcessor(config=ProcessingConfig(log=True)) data = np.array([[-100.0, -1.0], [0.0, 1000.0]], dtype=np.float32) out = proc.log(data) assert np.all(np.isfinite(out)) # Non-positive pixels collapse to the floor log10(0.1) == -1; a count of 1 # maps to 0. assert out[0, 0] == pytest.approx(-1.0) assert out[1, 0] == pytest.approx(-1.0) assert proc.log(np.array([[1.0]]))[0, 0] == pytest.approx(0.0) ############################################## # Shape / dtype edge cases in the data path ############################################## def test_1d_buffer_coerces_list_and_scalar(qtbot, mocked_client, monkeypatch): """Non-ndarray payloads (python list, 0-d scalar) must not crash the 1D buffer accumulation.""" _fake_bridge_factory(monkeypatch) view = create_widget(qtbot, Image, client=mocked_client) _set_signal_config(mocked_client, "waveform1d", "img", signal_class="PreviewSignal", ndim=1) view.image(device="waveform1d", signal="img") view._on_data_update(_preview_update("waveform1d", "img", [[1, 2, 3]])) # python list assert view.main_image.raw_data.shape == (1, 3) view._on_data_update( _preview_update("waveform1d", "img", [[1, 2, 3], np.array(42.0)]) # 0-d scalar ) assert view.main_image.raw_data.shape == (2, 3) ############################################## # Teardown safety ############################################## def test_layer_accessors_safe_after_teardown(qtbot, mocked_client): """Once layer_manager is gone (post-cleanup), signal-driven accessors must not raise. They used to subscript None or resurrect the 'main' layer.""" view = create_widget(qtbot, Image, client=mocked_client) view.enable_full_colorbar = True view.layer_manager = None # simulate post-cleanup state # None of these should raise: assert view.autorange is False assert view.autorange_mode == "mean" view.toggle_autorange(True, "mean") view._set_autorange(True) view.autorange_mode = "max" view._sync_autorange_switch() view._sync_colorbar_levels() def test_history_numpy_source_renders(qtbot, mocked_client, monkeypatch): """Regression: numpy-array source columns from the history bulk path must render (a truth-test on source.values raised ValueError, swallowed by SafeSlot, leaving the image blank).""" _fake_bridge_factory(monkeypatch) view = create_widget(qtbot, Image, client=mocked_client) _set_signal_config( mocked_client, "eiger", "img", signal_class="AsyncSignal", ndim=2, obj_name="async_obj" ) view.image(device="eiger", signal="img") frame = np.arange(12, dtype=float).reshape(3, 4) source = _make_source( "eiger", "async_obj", np.asarray([frame]), metadata={"async_update_type": "add", "max_shape": [None, None]}, as_numpy=True, ) view._on_data_update(_make_update(source, reason="history")) np.testing.assert_array_equal(view.main_image.raw_data, frame)