diff --git a/bec_widgets/cli/client.py b/bec_widgets/cli/client.py index 7525a8e8..5a5d359c 100644 --- a/bec_widgets/cli/client.py +++ b/bec_widgets/cli/client.py @@ -5424,9 +5424,11 @@ class SignalComboBox(RPCBase): """ @rpc_call - def set_device(self, device: str | None): + def set_device(self, device: "str | None"): """ - Set the device. If device is not valid, device will be set to None which happens + Set the device. When signal_class_filter is active, ensures base-class + logic runs and then refreshes the signal list to show only signals from + that device matching the signal class filter. Args: device(str): device name. diff --git a/bec_widgets/utils/filter_io.py b/bec_widgets/utils/filter_io.py index b0f6700d..1e3a316e 100644 --- a/bec_widgets/utils/filter_io.py +++ b/bec_widgets/utils/filter_io.py @@ -7,6 +7,7 @@ from abc import ABC, abstractmethod from bec_lib.logger import bec_logger from qtpy.QtCore import QStringListModel from qtpy.QtWidgets import QComboBox, QCompleter, QLineEdit +from typeguard import TypeCheckError from bec_widgets.utils.ophyd_kind_util import Kind @@ -55,6 +56,49 @@ class WidgetFilterHandler(ABC): """ # This method should be implemented in subclasses or extended as needed + def update_with_bec_signal_class( + self, + signal_class_filter: str | list[str], + client, + ndim_filter: int | list[int] | None = None, + ) -> list[tuple[str, str, dict]]: + """Update the selection based on signal classes using device_manager.get_bec_signals. + + Args: + signal_class_filter (str|list[str]): List of signal class names to filter. + client: BEC client instance. + ndim_filter (int | list[int] | None): Filter signals by dimensionality. + If provided, only signals with matching ndim will be included. + + Returns: + list[tuple[str, str, dict]]: A list of (device_name, signal_name, signal_config) tuples. + """ + if not client or not hasattr(client, "device_manager"): + return [] + + try: + signals = client.device_manager.get_bec_signals(signal_class_filter) + except TypeCheckError as e: + logger.warning(f"Error retrieving signals: {e}") + return [] + + if ndim_filter is None: + return signals + + if isinstance(ndim_filter, int): + ndim_filter = [ndim_filter] + + filtered_signals = [] + for device_name, signal_name, signal_config in signals: + ndim = None + if isinstance(signal_config, dict): + ndim = signal_config.get("describe", {}).get("signal_info", {}).get("ndim") + + if ndim in ndim_filter: + filtered_signals.append((device_name, signal_name, signal_config)) + + return filtered_signals + class LineEditFilterHandler(WidgetFilterHandler): """Handler for QLineEdit widget""" @@ -255,6 +299,32 @@ class FilterIO: f"No matching handler for widget type: {type(widget)} in handler list {FilterIO._handlers}" ) + @staticmethod + def update_with_signal_class( + widget, signal_class_filter: list[str], client, ndim_filter: int | list[int] | None = None + ) -> list[tuple[str, str, dict]]: + """ + Update the selection based on signal classes using device_manager.get_bec_signals. + + Args: + widget: Widget instance. + signal_class_filter (list[str]): List of signal class names to filter. + client: BEC client instance. + ndim_filter (int | list[int] | None): Filter signals by dimensionality. + If provided, only signals with matching ndim will be included. + + Returns: + list[tuple[str, str, dict]]: A list of (device_name, signal_name, signal_config) tuples. + """ + handler_class = FilterIO._find_handler(widget) + if handler_class: + return handler_class().update_with_bec_signal_class( + signal_class_filter=signal_class_filter, client=client, ndim_filter=ndim_filter + ) + raise ValueError( + f"No matching handler for widget type: {type(widget)} in handler list {FilterIO._handlers}" + ) + @staticmethod def _find_handler(widget): """ diff --git a/bec_widgets/widgets/control/device_input/base_classes/device_signal_input_base.py b/bec_widgets/widgets/control/device_input/base_classes/device_signal_input_base.py index 4e07b190..07e993e3 100644 --- a/bec_widgets/widgets/control/device_input/base_classes/device_signal_input_base.py +++ b/bec_widgets/widgets/control/device_input/base_classes/device_signal_input_base.py @@ -6,7 +6,7 @@ from qtpy.QtCore import Property from bec_widgets.utils import ConnectionConfig from bec_widgets.utils.bec_widget import BECWidget from bec_widgets.utils.error_popups import SafeSlot -from bec_widgets.utils.filter_io import FilterIO, LineEditFilterHandler +from bec_widgets.utils.filter_io import FilterIO from bec_widgets.utils.ophyd_kind_util import Kind from bec_widgets.utils.widget_io import WidgetIO @@ -17,6 +17,8 @@ class DeviceSignalInputBaseConfig(ConnectionConfig): """Configuration class for DeviceSignalInputBase.""" signal_filter: str | list[str] | None = None + signal_class_filter: list[str] | None = None + ndim_filter: int | list[int] | None = None default: str | None = None arg_name: str | None = None device: str | None = None diff --git a/bec_widgets/widgets/control/device_input/signal_combobox/signal_combobox.py b/bec_widgets/widgets/control/device_input/signal_combobox/signal_combobox.py index c895b9db..dcc8b879 100644 --- a/bec_widgets/widgets/control/device_input/signal_combobox/signal_combobox.py +++ b/bec_widgets/widgets/control/device_input/signal_combobox/signal_combobox.py @@ -1,7 +1,6 @@ from __future__ import annotations -from bec_lib.device import Positioner -from qtpy.QtCore import QSize, Signal +from qtpy.QtCore import QSize, Qt, Signal from qtpy.QtWidgets import QComboBox, QSizePolicy from bec_widgets.utils.error_popups import SafeProperty, SafeSlot @@ -22,9 +21,17 @@ class SignalComboBox(DeviceSignalInputBase, QComboBox): client: BEC client object. config: Device input configuration. gui_id: GUI ID. - device_filter: Device filter, name of the device class from BECDeviceFilter and BECReadoutPriority. Check DeviceInputBase for more details. + device: Device name to filter signals from. + signal_filter: Signal filter, list of signal kinds from ophyd Kind enum. Check DeviceSignalInputBase for more details. + signal_class_filter: List of signal classes to filter the signals by. Only signals of these classes will be shown. + ndim_filter: Dimensionality filter, int or list of ints to filter signals by their number of dimensions. If signal do not support ndim, it will be included in the selection anyway. default: Default device name. arg_name: Argument name, can be used for the other widgets which has to call some other function in bec using correct argument names. + store_signal_config: Whether to store the full signal config in the combobox item data. + require_device: If True, signals are only shown/validated when a device is set. + Signals: + device_signal_changed: Emitted when the current text represents a valid signal selection. + signal_reset: Emitted when validation fails and the selection should be treated as cleared. """ USER_ACCESS = ["set_signal", "set_device", "signals", "get_signal_name"] @@ -34,6 +41,7 @@ class SignalComboBox(DeviceSignalInputBase, QComboBox): RPC = True device_signal_changed = Signal(str) + signal_reset = Signal() def __init__( self, @@ -42,9 +50,13 @@ class SignalComboBox(DeviceSignalInputBase, QComboBox): config: DeviceSignalInputBaseConfig | None = None, gui_id: str | None = None, device: str | None = None, - signal_filter: str | list[str] | None = None, + signal_filter: list[Kind] | None = None, + signal_class_filter: list[str] | None = None, + ndim_filter: int | list[int] | None = None, default: str | None = None, arg_name: str | None = None, + store_signal_config: bool = True, + require_device: bool = False, **kwargs, ): super().__init__(parent=parent, client=client, gui_id=gui_id, config=config, **kwargs) @@ -57,26 +69,64 @@ class SignalComboBox(DeviceSignalInputBase, QComboBox): self.setSizePolicy(QSizePolicy.Preferred, QSizePolicy.Fixed) self.setMinimumSize(QSize(100, 0)) self._set_first_element_as_empty = True - # We do not consider the config that is passed here, this produced problems - # with QtDesigner, since config and input arguments may differ and resolve properly - # Implementing this logic and config recoverage is postponed. + self._signal_class_filter = signal_class_filter or [] + self._store_signal_config = store_signal_config + self.config.ndim_filter = ndim_filter or None + self._require_device = require_device + self._is_valid_input = False + + # Note: Runtime arguments (e.g. device, default, arg_name) intentionally take + # precedence over values from the passed-in config. Full reconciliation and + # restoration of state between designer-provided config and runtime arguments + # is not yet implemented, as earlier attempts caused issues with QtDesigner. self.currentTextChanged.connect(self.on_text_changed) + + # Kind filtering is always applied; class filtering is additive. If signal_filter is None, + # we default to hinted+normal, even when signal_class_filter is empty or None. To disable + # kinds, pass an explicit signal_filter or toggle include_* after init. if signal_filter is not None: self.set_filter(signal_filter) else: self.set_filter([Kind.hinted, Kind.normal, Kind.config]) + if device is not None: self.set_device(device) if default is not None: self.set_signal(default) + @SafeSlot(str) + def set_device(self, device: str | None): + """ + Set the device. When signal_class_filter is active, ensures base-class + logic runs and then refreshes the signal list to show only signals from + that device matching the signal class filter. + + Args: + device(str): device name. + """ + super().set_device(device) + + if self._signal_class_filter: + # Refresh the signal list to show only this device's signals + self.update_signals_from_signal_classes() + @SafeSlot() @SafeSlot(dict, dict) def update_signals_from_filters( self, content: dict | None = None, metadata: dict | None = None ): - """Update the filters for the combobox""" + """Update the filters for the combobox. + When signal_class_filter is active, skip the normal Kind-based filtering. + + Args: + content (dict | None): Content dictionary from BEC event. + metadata (dict | None): Metadata dictionary from BEC event. + """ super().update_signals_from_filters(content, metadata) + + if self._signal_class_filter: + self.update_signals_from_signal_classes() + return # pylint: disable=protected-access if FilterIO._find_handler(self) is ComboBoxFilterHandler: if len(self._config_signals) > 0: @@ -118,6 +168,63 @@ class SignalComboBox(DeviceSignalInputBase, QComboBox): if self.count() > 0 and self.itemText(0) == "": self.removeItem(0) + @SafeProperty("QStringList") + def signal_class_filter(self) -> list[str]: + """ + Get the list of signal classes to filter. + + Returns: + list[str]: List of signal class names to filter. + """ + return self._signal_class_filter + + @signal_class_filter.setter + def signal_class_filter(self, value: list[str] | None): + """ + Set the signal class filter. + + Args: + value (list[str] | None): List of signal class names to filter, or None/empty + to disable class-based filtering and revert to the default behavior. + """ + normalized_value = value or [] + self._signal_class_filter = normalized_value + self.config.signal_class_filter = normalized_value + if self._signal_class_filter: + self.update_signals_from_signal_classes() + else: + self.update_signals_from_filters() + + @SafeProperty(int) + def ndim_filter(self) -> int: + """Dimensionality filter for signals.""" + return self.config.ndim_filter if isinstance(self.config.ndim_filter, int) else -1 + + @ndim_filter.setter + def ndim_filter(self, value: int): + self.config.ndim_filter = None if value < 0 else value + if self._signal_class_filter: + self.update_signals_from_signal_classes(ndim_filter=self.config.ndim_filter) + + @SafeProperty(bool) + def require_device(self) -> bool: + """ + If True, signals are only shown/validated when a device is set. + + Note: + This property affects list rebuilding only when a signal_class_filter + is active. Without a signal class filter, the available signals are + managed by the standard Kind-based filtering. + """ + return self._require_device + + @require_device.setter + def require_device(self, value: bool): + self._require_device = value + # Rebuild list when toggled, but only when using signal_class_filter + if self._signal_class_filter: + self.update_signals_from_signal_classes() + def set_to_obj_name(self, obj_name: str) -> bool: """ Set the combobox to the object name of the signal. @@ -166,6 +273,91 @@ class SignalComboBox(DeviceSignalInputBase, QComboBox): return signal_name if signal_name else "" + def get_signal_config(self) -> dict | None: + """ + Get the signal config from the combobox for the currently selected signal. + + Returns: + dict | None: The signal configuration dictionary or None if not available. + """ + if not self._store_signal_config: + return None + + index = self.currentIndex() + if index == -1: + return None + + signal_info = self.itemData(index) + return signal_info if signal_info else None + + def update_signals_from_signal_classes(self, ndim_filter: int | list[int] | None = None): + """ + Update the combobox with signals filtered by signal classes and optionally by ndim. + Uses device_manager.get_bec_signals() to retrieve signals. + If a device is set, only shows signals from that device. + + Args: + ndim_filter (int | list[int] | None): Filter signals by dimensionality. + If provided, only signals with matching ndim will be included. + Can be a single int or a list of ints. Use None to include all dimensions. + If not provided, uses the previously set ndim_filter. + """ + if not self._signal_class_filter: + return + + if self._require_device and not self._device: + self.clear() + self._signals = [] + FilterIO.set_selection(widget=self, selection=self._signals) + return + + # Update stored ndim_filter if a new one is provided + if ndim_filter is not None: + self.config.ndim_filter = ndim_filter + + self.clear() + + # Get signals with ndim filtering applied at the FilterIO level + signals = FilterIO.update_with_signal_class( + widget=self, + signal_class_filter=self._signal_class_filter, + client=self.client, + ndim_filter=self.config.ndim_filter, # Pass ndim_filter to FilterIO + ) + + # Track signals for validation and FilterIO selection + self._signals = [] + + for device_name, signal_name, signal_config in signals: + # Filter by device if one is set + if self._device and device_name != self._device: + continue + if self._signal_filter: + kind_str = signal_config.get("kind_str") + if kind_str is not None and kind_str not in { + kind.name for kind in self._signal_filter + }: + continue + + # Get storage_name for tooltip + storage_name = signal_config.get("storage_name", "") + + # Store the full signal config as item data if requested + if self._store_signal_config: + self.addItem(signal_name, signal_config) + else: + self.addItem(signal_name) + + # Track for validation + self._signals.append(signal_name) + + # Set tooltip to storage_name (Qt.ToolTipRole = 3) + if storage_name: + self.setItemData(self.count() - 1, storage_name, Qt.ItemDataRole.ToolTipRole) + + # Keep FilterIO selection in sync for validate_signal + FilterIO.set_selection(widget=self, selection=self._signals) + @SafeSlot() def reset_selection(self): """Reset the selection of the combobox.""" @@ -176,22 +368,44 @@ class SignalComboBox(DeviceSignalInputBase, QComboBox): @SafeSlot(str) def on_text_changed(self, text: str): - """Slot for text changed. If a device is selected and the signal is changed and valid it emits a signal. + """Validate and emit only when the signal is valid. For a positioner, the readback value has to be renamed to the device name. - - Args: - text (str): Text in the combobox. + When using signal_class_filter, device validation is skipped. """ - if self.validate_device(self.device) is False: - return - if self.validate_signal(text) is False: - return - self.device_signal_changed.emit(text) + self.check_validity(text) + + def check_validity(self, input_text: str) -> None: + """Check if the current value is a valid signal and emit only when valid.""" + if self._signal_class_filter: + if self._require_device and (not self._device or not input_text): + is_valid = False + else: + is_valid = self.validate_signal(input_text) + else: + if self._require_device and not self.validate_device(self._device): + is_valid = False + else: + is_valid = self.validate_device(self._device) and self.validate_signal(input_text) + + if is_valid: + self._is_valid_input = True + self.device_signal_changed.emit(input_text) + self.setStyleSheet("border: 1px solid transparent;") + else: + self._is_valid_input = False + self.signal_reset.emit() + if self.isEnabled(): + self.setStyleSheet("border: 1px solid red;") @property def selected_signal_comp_name(self) -> str: return dict(self.signals).get(self.currentText(), {}).get("component_name", "") + @property + def is_valid_input(self) -> bool: + """Whether the current text represents a valid signal selection.""" + return self._is_valid_input + if __name__ == "__main__": # pragma: no cover # pylint: disable=import-outside-toplevel @@ -205,7 +419,14 @@ if __name__ == "__main__": # pragma: no cover widget.setFixedSize(200, 200) layout = QVBoxLayout() widget.setLayout(layout) - box = SignalComboBox(device="samx") + box = SignalComboBox( + device="waveform", + signal_class_filter=["AsyncSignal", "AsyncMultiSignal"], + ndim_filter=[1, 2], + store_signal_config=True, + signal_filter=[Kind.hinted, Kind.normal, Kind.config], + ) # change signal filter class to test + box.setEditable(True) layout.addWidget(box) widget.show() app.exec_() diff --git a/tests/unit_tests/test_device_signal_input.py b/tests/unit_tests/test_device_signal_input.py index ec2426fd..78329344 100644 --- a/tests/unit_tests/test_device_signal_input.py +++ b/tests/unit_tests/test_device_signal_input.py @@ -210,3 +210,193 @@ def test_signal_combobox_get_signal_name_with_velocity(qtbot, device_signal_comb signal_name = device_signal_combobox.get_signal_name() assert signal_name == "samx_velocity" + + +def test_signal_combobox_get_signal_config(device_signal_combobox): + device_signal_combobox.include_normal_signals = True + device_signal_combobox.include_hinted_signals = True + device_signal_combobox.set_device("samx") + + index = device_signal_combobox.currentIndex() + assert index != -1 + + expected_config = device_signal_combobox.itemData(index) + assert expected_config is not None + assert device_signal_combobox.get_signal_config() == expected_config + + +def test_signal_combobox_get_signal_config_disabled(qtbot, mocked_client): + combobox = create_widget( + qtbot=qtbot, widget=SignalComboBox, client=mocked_client, store_signal_config=False + ) + combobox.include_normal_signals = True + combobox.include_hinted_signals = True + combobox.set_device("samx") + assert combobox.get_signal_config() is None + + +def test_signal_combobox_signal_class_filter_by_device(qtbot, mocked_client): + """Test signal_class_filter restricts signals to the selected device.""" + mocked_client.device_manager.get_bec_signals = mock.MagicMock( + return_value=[ + ("samx", "samx_readback_async", {"obj_name": "samx_readback_async"}), + ("samy", "samy_readback_async", {"obj_name": "samy_readback_async"}), + ("bpm4i", "bpm4i_value_async", {"obj_name": "bpm4i_value_async"}), + ] + ) + widget = create_widget( + qtbot=qtbot, + widget=SignalComboBox, + client=mocked_client, + signal_class_filter=["AsyncSignal"], + device="samx", + ) + + assert widget.signals == ["samx_readback_async"] + assert widget.signal_class_filter == ["AsyncSignal"] + + widget.set_device("samy") + assert widget.signals == ["samy_readback_async"] + + +def test_signal_class_filter_setter_clears_to_kind_filters(qtbot, mocked_client): + """Clearing signal_class_filter should rebuild list using Kind filters.""" + mocked_client.device_manager.get_bec_signals = mock.MagicMock( + return_value=[("samx", "samx_readback_async", {"obj_name": "samx_readback_async"})] + ) + widget = create_widget( + qtbot=qtbot, + widget=SignalComboBox, + client=mocked_client, + signal_class_filter=["AsyncSignal"], + device="samx", + ) + assert widget.signals == ["samx_readback_async"] + + widget.signal_class_filter = [] + samx = widget.dev.samx + assert widget.signals == [ + ("samx (readback)", samx._info["signals"].get("readback")), + ("setpoint", samx._info["signals"].get("setpoint")), + ("velocity", samx._info["signals"].get("velocity")), + ] + + +def test_signal_class_filter_setter_none_reverts_to_kind_filters(qtbot, mocked_client): + """Setting signal_class_filter to None should revert to Kind-based filtering.""" + mocked_client.device_manager.get_bec_signals = mock.MagicMock( + return_value=[("samx", "samx_readback_async", {"obj_name": "samx_readback_async"})] + ) + widget = create_widget( + qtbot=qtbot, + widget=SignalComboBox, + client=mocked_client, + signal_class_filter=["AsyncSignal"], + device="samx", + ) + assert widget.signals == ["samx_readback_async"] + + widget.signal_class_filter = None + samx = widget.dev.samx + assert widget.signals == [ + ("samx (readback)", samx._info["signals"].get("readback")), + ("setpoint", samx._info["signals"].get("setpoint")), + ("velocity", samx._info["signals"].get("velocity")), + ] + + +def test_signal_combobox_set_first_element_as_empty(qtbot, mocked_client): + """set_first_element_as_empty should insert/remove the empty option.""" + widget = create_widget(qtbot=qtbot, widget=SignalComboBox, client=mocked_client) + widget.addItem("item1") + widget.addItem("item2") + + widget.set_first_element_as_empty = True + assert widget.itemText(0) == "" + + widget.set_first_element_as_empty = False + assert widget.itemText(0) == "item1" + + +def test_signal_combobox_class_kind_ndim_filters(qtbot, mocked_client): + """Test class + kind + ndim filters are all applied together.""" + mocked_client.device_manager.get_bec_signals = mock.MagicMock( + return_value=[ + ( + "samx", + "sig1", + { + "obj_name": "samx_sig1", + "kind_str": "hinted", + "describe": {"signal_info": {"ndim": 1}}, + }, + ), + ( + "samx", + "sig2", + { + "obj_name": "samx_sig2", + "kind_str": "config", + "describe": {"signal_info": {"ndim": 2}}, + }, + ), + ( + "samy", + "sig3", + { + "obj_name": "samy_sig3", + "kind_str": "normal", + "describe": {"signal_info": {"ndim": 1}}, + }, + ), + ] + ) + widget = create_widget( + qtbot=qtbot, + widget=SignalComboBox, + client=mocked_client, + signal_class_filter=["AsyncSignal"], + ndim_filter=1, + device="samx", + ) + + # Default kinds are hinted + normal, ndim=1, device=samx + assert widget.signals == ["sig1"] + + # Enable config kinds and widen ndim to include sig2 + widget.include_config_signals = True + widget.ndim_filter = 2 + assert widget.signals == ["sig2"] + + +def test_signal_combobox_require_device_validation(qtbot, mocked_client): + """Require device should block validation and list updates without a device.""" + mocked_client.device_manager.get_bec_signals = mock.MagicMock( + return_value=[ + ( + "samx", + "sig1", + { + "obj_name": "samx_sig1", + "kind_str": "hinted", + "describe": {"signal_info": {"ndim": 1}}, + }, + ) + ] + ) + widget = create_widget( + qtbot=qtbot, + widget=SignalComboBox, + client=mocked_client, + signal_class_filter=["AsyncSignal"], + require_device=True, + ) + + assert widget.signals == [] + widget.set_device("samx") + assert widget.signals == ["sig1"] + + resets: list[str] = [] + widget.signal_reset.connect(lambda: resets.append("reset")) + widget.check_validity("") + assert resets == ["reset"] diff --git a/tests/unit_tests/test_filter_io.py b/tests/unit_tests/test_filter_io.py index 818faa7d..e5087124 100644 --- a/tests/unit_tests/test_filter_io.py +++ b/tests/unit_tests/test_filter_io.py @@ -45,3 +45,30 @@ def test_set_selection_line_edit(line_edit_mock): FilterIO.set_selection(line_edit_mock, selection=["testC"]) assert FilterIO.check_input(widget=line_edit_mock, text="testA") is False assert FilterIO.check_input(widget=line_edit_mock, text="testC") is True + + +def test_update_with_signal_class_combo_box_ndim_filter(dap_mock, mocked_client): + signals = [ + ("dev1", "sig1", {"describe": {"signal_info": {"ndim": 1}}}), + ("dev1", "sig2", {"describe": {"signal_info": {"ndim": 2}}}), + ] + mocked_client.device_manager.get_bec_signals = lambda _filters: signals + out = FilterIO.update_with_signal_class( + widget=dap_mock.fit_model_combobox, + signal_class_filter=["AsyncSignal"], + client=mocked_client, + ndim_filter=1, + ) + assert out == [("dev1", "sig1", {"describe": {"signal_info": {"ndim": 1}}})] + + +def test_update_with_signal_class_line_edit_passthrough(line_edit_mock, mocked_client): + signals = [("dev1", "sig1", {"describe": {"signal_info": {"ndim": 1}}})] + mocked_client.device_manager.get_bec_signals = lambda _filters: signals + out = FilterIO.update_with_signal_class( + widget=line_edit_mock, + signal_class_filter=["AsyncSignal"], + client=mocked_client, + ndim_filter=1, + ) + assert out == signals