diff --git a/bec_widgets/cli/client.py b/bec_widgets/cli/client.py index 8d93a579..338b0035 100644 --- a/bec_widgets/cli/client.py +++ b/bec_widgets/cli/client.py @@ -1459,12 +1459,12 @@ class Image(RPCBase): @rpc_call def image( self, - monitor: "str | None" = None, + monitor: "str | tuple | None" = None, monitor_type: "Literal['auto', '1d', '2d']" = "auto", color_map: "str | None" = None, color_bar: "Literal['simple', 'full'] | None" = None, vrange: "tuple[int, int] | None" = None, - ) -> "ImageItem": + ) -> "ImageItem | None": """ Set the image source and update the image. diff --git a/bec_widgets/widgets/plots/image/image.py b/bec_widgets/widgets/plots/image/image.py index 3dc6c855..2432a791 100644 --- a/bec_widgets/widgets/plots/image/image.py +++ b/bec_widgets/widgets/plots/image/image.py @@ -35,7 +35,7 @@ class ImageConfig(ConnectionConfig): class ImageLayerConfig(BaseModel): - monitor: str | None = Field(None, description="The name of the monitor.") + monitor: str | tuple | None = Field(None, description="The name of the monitor.") monitor_type: Literal["1d", "2d", "auto"] = Field("auto", description="The type of monitor.") source: Literal["device_monitor_1d", "device_monitor_2d", "auto"] = Field( "auto", description="The source of the image data." @@ -179,12 +179,12 @@ class Image(ImageBase): @SafeSlot(popup_error=True) def image( self, - monitor: str | None = None, + monitor: str | tuple | None = None, monitor_type: Literal["auto", "1d", "2d"] = "auto", color_map: str | None = None, color_bar: Literal["simple", "full"] | None = None, vrange: tuple[int, int] | None = None, - ) -> ImageItem: + ) -> ImageItem | None: """ Set the image source and update the image. @@ -201,21 +201,13 @@ class Image(ImageBase): if self.subscriptions["main"].monitor: self.disconnect_monitor(self.subscriptions["main"].monitor) - self.entry_validator.validate_monitor(monitor) - self.subscriptions["main"].monitor = monitor - - if monitor_type == "1d": - self.subscriptions["main"].source = "device_monitor_1d" - self.subscriptions["main"].monitor_type = "1d" - elif monitor_type == "2d": - self.subscriptions["main"].source = "device_monitor_2d" - self.subscriptions["main"].monitor_type = "2d" - elif monitor_type == "auto": - self.subscriptions["main"].source = "auto" - logger.warning( - f"Updates for '{monitor}' will be fetch from both 1D and 2D monitor endpoints." - ) - self.subscriptions["main"].monitor_type = "auto" + if monitor is None or monitor == "": + logger.warning(f"No monitor specified, cannot set image, old monitor is unsubscribed") + return None + if isinstance(monitor, tuple): + self.entry_validator.validate_monitor(monitor[0]) + else: + self.entry_validator.validate_monitor(monitor) self.set_image_update(monitor=monitor, type=monitor_type) if color_map is not None: @@ -240,7 +232,12 @@ class Image(ImageBase): self.selection_bundle.dim_combo_box, ): combo.blockSignals(True) - self.selection_bundle.device_combo_box.set_device(config.monitor) + if isinstance(config.monitor, tuple): + self.selection_bundle.device_combo_box.setCurrentText( + f"{config.monitor[0]}_{config.monitor[1]}" + ) + else: + self.selection_bundle.device_combo_box.setCurrentText(config.monitor) self.selection_bundle.dim_combo_box.setCurrentText(config.monitor_type) for combo in ( self.selection_bundle.device_combo_box, @@ -340,7 +337,8 @@ class Image(ImageBase): ######################################## # Connections - def set_image_update(self, monitor: str, type: Literal["1d", "2d", "auto"]): + @SafeSlot() + def set_image_update(self, monitor: str | tuple, type: Literal["1d", "2d", "auto"]): """ Set the image update method for the given monitor. @@ -350,37 +348,95 @@ class Image(ImageBase): """ # TODO consider moving connecting and disconnecting logic to Image itself if multiple images - if type == "1d": - self.bec_dispatcher.connect_slot( - self.on_image_update_1d, MessageEndpoints.device_monitor_1d(monitor) - ) - elif type == "2d": - self.bec_dispatcher.connect_slot( - self.on_image_update_2d, MessageEndpoints.device_monitor_2d(monitor) - ) - elif type == "auto": - self.bec_dispatcher.connect_slot( - self.on_image_update_1d, MessageEndpoints.device_monitor_1d(monitor) - ) - self.bec_dispatcher.connect_slot( - self.on_image_update_2d, MessageEndpoints.device_monitor_2d(monitor) - ) + if isinstance(monitor, tuple): + device = self.dev[monitor[0]] + signal = monitor[1] + if len(monitor) == 3: + signal_config = monitor[2] + else: + signal_config = device._info["signals"][signal] + signal_class = signal_config.get("signal_class", None) + if signal_class != "PreviewSignal": + logger.warning(f"Signal '{monitor}' is not a PreviewSignal.") + return + + ndim = signal_config.get("describe", None).get("signal_info", None).get("ndim", None) + if ndim is None: + logger.warning( + f"Signal '{monitor}' does not have a valid 'ndim' in its signal_info." + ) + return + + if ndim == 1: + self.bec_dispatcher.connect_slot( + self.on_image_update_1d, MessageEndpoints.device_preview(device.name, signal) + ) + self.subscriptions["main"].source = "device_monitor_1d" + self.subscriptions["main"].monitor_type = "1d" + elif ndim == 2: + self.bec_dispatcher.connect_slot( + self.on_image_update_2d, MessageEndpoints.device_preview(device.name, signal) + ) + self.subscriptions["main"].source = "device_monitor_2d" + self.subscriptions["main"].monitor_type = "2d" + + else: # FIXME old monitor 1d/2d endpoint handling, present for backwards compatibility, will be removed in future versions + if type == "1d": + self.bec_dispatcher.connect_slot( + self.on_image_update_1d, MessageEndpoints.device_monitor_1d(monitor) + ) + self.subscriptions["main"].source = "device_monitor_1d" + self.subscriptions["main"].monitor_type = "1d" + elif type == "2d": + self.bec_dispatcher.connect_slot( + self.on_image_update_2d, MessageEndpoints.device_monitor_2d(monitor) + ) + self.subscriptions["main"].source = "device_monitor_2d" + self.subscriptions["main"].monitor_type = "2d" + elif type == "auto": + self.bec_dispatcher.connect_slot( + self.on_image_update_1d, MessageEndpoints.device_monitor_1d(monitor) + ) + self.bec_dispatcher.connect_slot( + self.on_image_update_2d, MessageEndpoints.device_monitor_2d(monitor) + ) + self.subscriptions["main"].source = "auto" + logger.warning( + f"Updates for '{monitor}' will be fetch from both 1D and 2D monitor endpoints." + ) + self.subscriptions["main"].monitor_type = "auto" + logger.info(f"Connected to {monitor} with type {type}") self.subscriptions["main"].monitor = monitor - def disconnect_monitor(self, monitor: str): + def disconnect_monitor(self, monitor: str | tuple): """ Disconnect the monitor from the image update signals, both 1D and 2D. Args: - monitor(str): The name of the monitor to disconnect. + monitor(str|tuple): The name of the monitor to disconnect, or a tuple of (device, signal) for preview signals. """ - self.bec_dispatcher.disconnect_slot( - self.on_image_update_1d, MessageEndpoints.device_monitor_1d(monitor) - ) - self.bec_dispatcher.disconnect_slot( - self.on_image_update_2d, MessageEndpoints.device_monitor_2d(monitor) - ) + if isinstance(monitor, tuple): + if self.subscriptions["main"].source == "device_monitor_1d": + self.bec_dispatcher.disconnect_slot( + self.on_image_update_1d, MessageEndpoints.device_preview(monitor[0], monitor[1]) + ) + elif self.subscriptions["main"].source == "device_monitor_2d": + self.bec_dispatcher.disconnect_slot( + self.on_image_update_2d, MessageEndpoints.device_preview(monitor[0], monitor[1]) + ) + else: + logger.warning( + f"Cannot disconnect monitor {monitor} with source {self.subscriptions['main'].source}" + ) + return + else: # FIXME old monitor 1d/2d endpoint handling, present for backwards compatibility, will be removed in future versions + self.bec_dispatcher.disconnect_slot( + self.on_image_update_1d, MessageEndpoints.device_monitor_1d(monitor) + ) + self.bec_dispatcher.disconnect_slot( + self.on_image_update_2d, MessageEndpoints.device_monitor_2d(monitor) + ) self.subscriptions["main"].monitor = None self._sync_device_selection() diff --git a/bec_widgets/widgets/plots/image/toolbar_bundles/image_selection.py b/bec_widgets/widgets/plots/image/toolbar_bundles/image_selection.py index 9e3e7055..f18e54b9 100644 --- a/bec_widgets/widgets/plots/image/toolbar_bundles/image_selection.py +++ b/bec_widgets/widgets/plots/image/toolbar_bundles/image_selection.py @@ -1,5 +1,5 @@ from bec_lib.device import ReadoutPriority -from qtpy.QtCore import Qt +from qtpy.QtCore import Qt, QTimer from qtpy.QtWidgets import QComboBox, QStyledItemDelegate from bec_widgets.utils.error_popups import SafeSlot @@ -50,11 +50,58 @@ class MonitorSelectionToolbarBundle(ToolbarBundle): self.add_action("dim_combo", WidgetAction(widget=self.dim_combo_box, adjust_size=False)) - # Connect slots, a device will be connected upon change of any combobox - self.device_combo_box.currentTextChanged.connect(lambda: self.connect_monitor()) - self.dim_combo_box.currentTextChanged.connect(lambda: self.connect_monitor()) + self.device_combo_box.currentTextChanged.connect(self.connect_monitor) + self.dim_combo_box.currentTextChanged.connect(self.connect_monitor) + + QTimer.singleShot(0, self._adjust_and_connect) + + def _adjust_and_connect(self): + """ + Adjust the size of the device combo box and populate it with preview signals. + Has to be done with QTimer.singleShot to ensure the UI is fully initialized, needed for testing. + """ + self._populate_preview_signals() + self._reverse_device_items() + self.device_combo_box.setCurrentText("") # set again default to empty string + + def _populate_preview_signals(self) -> None: + """ + Populate the device combo box with preview‑signal devices in the + format '_' and store the tuple(device, signal) in + the item's userData for later use. + """ + preview_signals = self.target_widget.client.device_manager.get_bec_signals("PreviewSignal") + for device, signal, signal_config in preview_signals: + label = signal_config.get("obj_name", f"{device}_{signal}") + self.device_combo_box.addItem(label, (device, signal, signal_config)) + + def _reverse_device_items(self) -> None: + """ + Reverse the current order of items in the device combo box while + keeping their userData and restoring the previous selection. + """ + current_text = self.device_combo_box.currentText() + items = [ + (self.device_combo_box.itemText(i), self.device_combo_box.itemData(i)) + for i in range(self.device_combo_box.count()) + ] + self.device_combo_box.clear() + for text, data in reversed(items): + self.device_combo_box.addItem(text, data) + if current_text: + self.device_combo_box.setCurrentText(current_text) @SafeSlot() - def connect_monitor(self): + def connect_monitor(self, *args, **kwargs): + """ + Connect the target widget to the selected monitor based on the current device and dimension. + + If the selected device is a preview-signal device, it will use the tuple (device, signal) as the monitor. + """ dim = self.dim_combo_box.currentText() - self.target_widget.image(monitor=self.device_combo_box.currentText(), monitor_type=dim) + data = self.device_combo_box.currentData() + + if isinstance(data, tuple): + self.target_widget.image(monitor=data, monitor_type="auto") + else: + self.target_widget.image(monitor=self.device_combo_box.currentText(), monitor_type=dim) diff --git a/tests/unit_tests/test_image_view_next_gen.py b/tests/unit_tests/test_image_view_next_gen.py index 3385c8eb..7ad9ab1d 100644 --- a/tests/unit_tests/test_image_view_next_gen.py +++ b/tests/unit_tests/test_image_view_next_gen.py @@ -113,6 +113,75 @@ def test_enable_colorbar_with_vrange(qtbot, mocked_client, colorbar_type): assert bec_image_view._color_bar is not None +############################################## +# Preview‑signal update mechanism + + +def test_image_setup_preview_signal_1d(qtbot, mocked_client, monkeypatch): + """ + Ensure that calling .image() with a (device, signal, config) tuple representing + a 1‑D PreviewSignal connects using the 1‑D path and updates correctly. + """ + import numpy as np + + view = create_widget(qtbot, Image, client=mocked_client) + + signal_config = { + "obj_name": "waveform1d_img", + "signal_class": "PreviewSignal", + "describe": {"signal_info": {"ndim": 1}}, + } + + # Set the image monitor to the preview signal + view.image(monitor=("waveform1d", "img", signal_config)) + + # Subscriptions should indicate 1‑D preview connection + sub = view.subscriptions["main"] + assert sub.source == "device_monitor_1d" + assert sub.monitor_type == "1d" + assert sub.monitor == ("waveform1d", "img", signal_config) + + # Simulate a waveform update from the dispatcher + waveform = np.arange(25, dtype=float) + view.on_image_update_1d({"data": waveform}, {"scan_id": "scan_test"}) + 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 (device, signal, config) tuple representing + a 2‑D PreviewSignal connects using the 2‑D path and updates correctly. + """ + import numpy as np + + view = create_widget(qtbot, Image, client=mocked_client) + + signal_config = { + "obj_name": "eiger_img2d", + "signal_class": "PreviewSignal", + "describe": {"signal_info": {"ndim": 2}}, + } + + # Set the image monitor to the preview signal + view.image(monitor=("eiger", "img2d", signal_config)) + + # Subscriptions should indicate 2‑D preview connection + sub = view.subscriptions["main"] + assert sub.source == "device_monitor_2d" + assert sub.monitor_type == "2d" + assert sub.monitor == ("eiger", "img2d", signal_config) + + # Simulate a 2‑D image update + test_data = np.arange(16, dtype=float).reshape(4, 4) + view.on_image_update_2d({"data": test_data}, {}) + np.testing.assert_array_equal(view.main_image.image, test_data) + + +############################################## +# Device monitor endpoint update mechanism + + def test_image_setup_image_2d(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) bec_image_view.image(monitor="eiger", monitor_type="2d") @@ -167,6 +236,10 @@ def test_image_data_update_1d(qtbot, mocked_client): assert bec_image_view.main_image.raw_data.shape == (2, 60) +############################################## +# Toolbar and Actions Tests + + def test_toolbar_actions_presence(qtbot, mocked_client): bec_image_view = create_widget(qtbot, Image, client=mocked_client) assert "autorange_image" in bec_image_view.toolbar.widgets @@ -484,3 +557,96 @@ def test_roi_plot_data_from_image(qtbot, mocked_client): # Horizontal slice (row) h_slice, _ = y_items[0].getData() np.testing.assert_array_equal(h_slice, test_data[2]) + + +############################################## +# MonitorSelectionToolbarBundle specific tests +############################################## + + +def test_monitor_selection_reverse_device_items(qtbot, mocked_client): + """ + Verify that _reverse_device_items correctly reverses the order of items in the + device combo‑box while preserving the current selection. + """ + view = create_widget(qtbot, Image, client=mocked_client) + bundle = view.selection_bundle + combo = bundle.device_combo_box + + # Replace existing items with a deterministic list + combo.clear() + combo.addItem("samx", 1) + combo.addItem("samy", 2) + combo.addItem("samz", 3) + combo.setCurrentText("samy") + + # Reverse the items + bundle._reverse_device_items() + + # Order should be reversed and selection preserved + assert [combo.itemText(i) for i in range(combo.count())] == ["samz", "samy", "samx"] + assert combo.currentText() == "samy" + + +def test_monitor_selection_populate_preview_signals(qtbot, mocked_client, monkeypatch): + """ + Verify that _populate_preview_signals adds preview‑signal devices to the combo‑box + with the correct userData. + """ + view = create_widget(qtbot, Image, client=mocked_client) + bundle = view.selection_bundle + + # Provide a deterministic fake device_manager with get_bec_signals + class _FakeDM: + def get_bec_signals(self, _filter): + return [ + ("eiger", "img", {"obj_name": "eiger_img"}), + ("async_device", "img2", {"obj_name": "async_device_img2"}), + ] + + monkeypatch.setattr(view.client, "device_manager", _FakeDM()) + + initial_count = bundle.device_combo_box.count() + + bundle._populate_preview_signals() + + # Two new entries should have been added + assert bundle.device_combo_box.count() == initial_count + 2 + + # The first newly added item should carry tuple userData describing the device/signal + data = bundle.device_combo_box.itemData(initial_count) + assert isinstance(data, tuple) and data[0] == "eiger" + + +def test_monitor_selection_adjust_and_connect(qtbot, mocked_client, monkeypatch): + """ + Verify that _adjust_and_connect performs the full set‑up: + ‑ fills the combo‑box with preview signals, + ‑ reverses their order, + ‑ and resets the currentText to an empty string. + """ + view = create_widget(qtbot, Image, client=mocked_client) + bundle = view.selection_bundle + + # Deterministic fake device_manager + class _FakeDM: + def get_bec_signals(self, _filter): + return [("eiger", "img", {"obj_name": "eiger_img"})] + + monkeypatch.setattr(view.client, "device_manager", _FakeDM()) + + combo = bundle.device_combo_box + # Start from a clean state + combo.clear() + combo.addItem("", None) + combo.setCurrentText("") + + # Execute the method under test + bundle._adjust_and_connect() + + # Expect exactly two items: preview label followed by the empty default + assert combo.count() == 2 + # Because of the reversal, the preview label comes first + assert combo.itemText(0) == "eiger_img" + # Current selection remains empty + assert combo.currentText() == ""