1
0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-03-04 16:02:51 +01:00

feat(signal_combobox): extended that can filter by signal class and dimension of the signal

This commit is contained in:
2026-01-19 22:22:28 +01:00
committed by Jan Wyzula
parent 24dbb885f6
commit 8d75c2af1c
6 changed files with 533 additions and 21 deletions

View File

@@ -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.

View File

@@ -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):
"""

View File

@@ -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

View File

@@ -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_()

View File

@@ -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"]

View File

@@ -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