Compare commits

...

6 Commits

34 changed files with 1175 additions and 2348 deletions
-24
View File
@@ -1131,30 +1131,6 @@ class DeviceInitializationProgressBar(RPCBase):
""" """
class DeviceInputBase(RPCBase):
"""Mixin base class for device input widgets."""
_IMPORT_MODULE = "bec_widgets.widgets.control.device_input.base_classes.device_input_base"
@rpc_call
def remove(self):
"""
Cleanup the BECConnector
"""
@rpc_call
def attach(self):
"""
None
"""
@rpc_call
def detach(self):
"""
Detach the widget from its parent dock widget (if widget is in the dock), making it a floating widget.
"""
class DeviceManagerView(RPCBase): class DeviceManagerView(RPCBase):
"""A view for users to manage devices within the application.""" """A view for users to manage devices within the application."""
-10
View File
@@ -42,10 +42,6 @@ designer_plugins = {
"bec_widgets.widgets.control.device_input.device_combobox.device_combobox", "bec_widgets.widgets.control.device_input.device_combobox.device_combobox",
"DeviceComboBox", "DeviceComboBox",
), ),
"DeviceLineEdit": (
"bec_widgets.widgets.control.device_input.device_line_edit.device_line_edit",
"DeviceLineEdit",
),
"Heatmap": ("bec_widgets.widgets.plots.heatmap.heatmap", "Heatmap"), "Heatmap": ("bec_widgets.widgets.plots.heatmap.heatmap", "Heatmap"),
"IDEExplorer": ("bec_widgets.widgets.utility.ide_explorer.ide_explorer", "IDEExplorer"), "IDEExplorer": ("bec_widgets.widgets.utility.ide_explorer.ide_explorer", "IDEExplorer"),
"Image": ("bec_widgets.widgets.plots.image.image", "Image"), "Image": ("bec_widgets.widgets.plots.image.image", "Image"),
@@ -101,10 +97,6 @@ designer_plugins = {
"SignalComboBox", "SignalComboBox",
), ),
"SignalLabel": ("bec_widgets.widgets.utility.signal_label.signal_label", "SignalLabel"), "SignalLabel": ("bec_widgets.widgets.utility.signal_label.signal_label", "SignalLabel"),
"SignalLineEdit": (
"bec_widgets.widgets.control.device_input.signal_line_edit.signal_line_edit",
"SignalLineEdit",
),
"SpinnerWidget": ("bec_widgets.widgets.utility.spinner.spinner", "SpinnerWidget"), "SpinnerWidget": ("bec_widgets.widgets.utility.spinner.spinner", "SpinnerWidget"),
"StopButton": ("bec_widgets.widgets.control.buttons.stop_button.stop_button", "StopButton"), "StopButton": ("bec_widgets.widgets.control.buttons.stop_button.stop_button", "StopButton"),
"TextBox": ("bec_widgets.widgets.editors.text_box.text_box", "TextBox"), "TextBox": ("bec_widgets.widgets.editors.text_box.text_box", "TextBox"),
@@ -134,7 +126,6 @@ widget_icons = {
"DarkModeButton": "dark_mode", "DarkModeButton": "dark_mode",
"DeviceBrowser": "lists", "DeviceBrowser": "lists",
"DeviceComboBox": "list_alt", "DeviceComboBox": "list_alt",
"DeviceLineEdit": "edit_note",
"Heatmap": "dataset", "Heatmap": "dataset",
"IDEExplorer": "widgets", "IDEExplorer": "widgets",
"Image": "image", "Image": "image",
@@ -160,7 +151,6 @@ widget_icons = {
"ScatterWaveform": "scatter_plot", "ScatterWaveform": "scatter_plot",
"SignalComboBox": "list_alt", "SignalComboBox": "list_alt",
"SignalLabel": "scoreboard", "SignalLabel": "scoreboard",
"SignalLineEdit": "vital_signs",
"SpinnerWidget": "progress_activity", "SpinnerWidget": "progress_activity",
"StopButton": "dangerous", "StopButton": "dangerous",
"TextBox": "chat", "TextBox": "chat",
+57 -321
View File
@@ -1,12 +1,9 @@
"""Module for handling filter I/O operations in BEC Widgets for input fields. """Small helpers for populating editable combo boxes used by device inputs."""
These operations include filtering device/signal names and/or device types.
"""
from abc import ABC, abstractmethod from __future__ import annotations
from bec_lib.logger import bec_logger from bec_lib.logger import bec_logger
from qtpy.QtCore import QStringListModel from qtpy.QtWidgets import QComboBox
from qtpy.QtWidgets import QComboBox, QCompleter, QLineEdit
from typeguard import TypeCheckError from typeguard import TypeCheckError
from bec_widgets.utils.ophyd_kind_util import Kind from bec_widgets.utils.ophyd_kind_util import Kind
@@ -14,329 +11,68 @@ from bec_widgets.utils.ophyd_kind_util import Kind
logger = bec_logger.logger logger = bec_logger.logger
class WidgetFilterHandler(ABC): def replace_combobox_items(combo_box: QComboBox, items: list[str | tuple]) -> None:
"""Abstract base class for widget filter handlers""" """Replace all combobox entries with strings or ``(text, data)`` tuples."""
combo_box.clear()
@abstractmethod for item in items:
def set_selection(self, widget, selection: list[str | tuple]) -> None: if isinstance(item, str):
"""Set the filtered_selection for the widget combo_box.addItem(item)
else:
Args: combo_box.addItem(*item)
widget: Widget instance
selection (list[str | tuple]): Filtered selection of items.
If tuple, it contains (text, data) pairs.
"""
@abstractmethod
def check_input(self, widget, text: str) -> bool:
"""Check if the input text is in the filtered selection
Args:
widget: Widget instance
text (str): Input text
Returns:
bool: True if the input text is in the filtered selection
"""
@abstractmethod
def update_with_kind(
self, kind: Kind, signal_filter: set, device_info: dict, device_name: str
) -> list[str | tuple]:
"""Update the selection based on the kind of signal.
Args:
kind (Kind): The kind of signal to filter.
signal_filter (set): Set of signal kinds to filter.
device_info (dict): Dictionary containing device information.
device_name (str): Name of the device.
Returns:
list[str | tuple]: A list of filtered signals based on the kind.
"""
# 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): def combobox_contains_text(combo_box: QComboBox, text: str) -> bool:
"""Handler for QLineEdit widget""" """Return whether *text* is present as visible combobox text."""
return any(combo_box.itemText(i) == text for i in range(combo_box.count()))
def set_selection(self, widget: QLineEdit, selection: list[str | tuple]) -> None:
"""Set the selection for the widget to the completer model
Args:
widget (QLineEdit): The QLineEdit widget
selection (list[str | tuple]): Filtered selection of items. If tuple, it contains (text, data) pairs.
"""
if isinstance(selection, tuple):
# If selection is a tuple, it contains (text, data) pairs
selection = [text for text, _ in selection]
if not isinstance(widget.completer, QCompleter):
completer = QCompleter(widget)
widget.setCompleter(completer)
widget.completer.setModel(QStringListModel(selection, widget))
def check_input(self, widget: QLineEdit, text: str) -> bool:
"""Check if the input text is in the filtered selection
Args:
widget (QLineEdit): The QLineEdit widget
text (str): Input text
Returns:
bool: True if the input text is in the filtered selection
"""
model = widget.completer.model()
model_data = [model.data(model.index(i)) for i in range(model.rowCount())]
return text in model_data
def update_with_kind(
self, kind: Kind, signal_filter: set, device_info: dict, device_name: str
) -> list[str | tuple]:
"""Update the selection based on the kind of signal.
Args:
kind (Kind): The kind of signal to filter.
signal_filter (set): Set of signal kinds to filter.
device_info (dict): Dictionary containing device information.
device_name (str): Name of the device.
Returns:
list[str | tuple]: A list of filtered signals based on the kind.
"""
return [
signal
for signal, signal_info in device_info.items()
if kind in signal_filter and (signal_info.get("kind_str", None) == str(kind.name))
]
class ComboBoxFilterHandler(WidgetFilterHandler): def signal_items_for_kind(
"""Handler for QComboBox widget""" *, kind: Kind, signal_filter: set[Kind], device_info: dict, device_name: str
) -> list[tuple[str, dict]]:
"""Build display entries for signals matching a BEC signal kind."""
items: list[tuple[str, dict]] = []
for signal_name, signal_info in device_info.items():
if kind not in signal_filter or signal_info.get("kind_str") != kind.name:
continue
def set_selection(self, widget: QComboBox, selection: list[str | tuple]) -> None: obj_name = signal_info.get("obj_name", "")
"""Set the selection for the widget to the completer model component_name = signal_info.get("component_name", "")
signal_without_device = obj_name.removeprefix(f"{device_name}_")
if not signal_without_device:
signal_without_device = obj_name
Args: if (
widget (QComboBox): The QComboBox widget signal_without_device != signal_name
selection (list[str | tuple]): Filtered selection of items. If tuple, it contains (text, data) pairs. and component_name.replace(".", "_") != signal_without_device
""" ):
widget.clear() items.append((f"{signal_without_device} ({signal_name})", signal_info))
if len(selection) == 0: else:
return items.append((signal_name, signal_info))
for element in selection: return items
if isinstance(element, str):
widget.addItem(element)
elif isinstance(element, tuple):
# If element is a tuple, it contains (text, data) pairs
widget.addItem(*element)
def check_input(self, widget: QComboBox, text: str) -> bool:
"""Check if the input text is in the filtered selection
Args:
widget (QComboBox): The QComboBox widget
text (str): Input text
Returns:
bool: True if the input text is in the filtered selection
"""
return text in [widget.itemText(i) for i in range(widget.count())]
def update_with_kind(
self, kind: Kind, signal_filter: set, device_info: dict, device_name: str
) -> list[str | tuple]:
"""Update the selection based on the kind of signal.
Args:
kind (Kind): The kind of signal to filter.
signal_filter (set): Set of signal kinds to filter.
device_info (dict): Dictionary containing device information.
device_name (str): Name of the device.
Returns:
list[str | tuple]: A list of filtered signals based on the kind.
"""
out = []
for signal, signal_info in device_info.items():
if kind not in signal_filter or (signal_info.get("kind_str", None) != str(kind.name)):
continue
obj_name = signal_info.get("obj_name", "")
component_name = signal_info.get("component_name", "")
signal_wo_device = obj_name.removeprefix(f"{device_name}_")
if not signal_wo_device:
signal_wo_device = obj_name
if signal_wo_device != signal and component_name.replace(".", "_") != signal_wo_device:
# If the object name is not the same as the signal name, we use the object name
# to display in the combobox.
out.append((f"{signal_wo_device} ({signal})", signal_info))
else:
# If the object name is the same as the signal name, we do not change it.
out.append((signal, signal_info))
return out
class FilterIO: def get_bec_signals_for_classes(
"""Public interface to set filters for input widgets. *, client, signal_class_filter: str | list[str], ndim_filter: int | list[int] | None = None
It supports the list of widgets stored in class attribute _handlers. ) -> list[tuple[str, str, dict]]:
""" """Return BEC signals filtered by signal class and optional dimensionality."""
if not client or not hasattr(client, "device_manager"):
return []
_handlers = {QLineEdit: LineEditFilterHandler, QComboBox: ComboBoxFilterHandler} try:
signals = client.device_manager.get_bec_signals(signal_class_filter)
except TypeCheckError as exc:
logger.warning(f"Error retrieving signals: {exc}")
return []
@staticmethod if ndim_filter is None:
def set_selection(widget, selection: list[str | tuple], ignore_errors=True): return signals
"""
Retrieve value from the widget instance.
Args: accepted_ndim = [ndim_filter] if isinstance(ndim_filter, int) else ndim_filter
widget: Widget instance. filtered_signals: list[tuple[str, str, dict]] = []
selection (list[str | tuple]): Filtered selection of items. for device_name, signal_name, signal_config in signals:
If tuple, it contains (text, data) pairs. ndim = None
ignore_errors(bool, optional): Whether to ignore if no handler is found. if isinstance(signal_config, dict):
""" ndim = signal_config.get("describe", {}).get("signal_info", {}).get("ndim")
handler_class = FilterIO._find_handler(widget) if ndim in accepted_ndim:
if handler_class: filtered_signals.append((device_name, signal_name, signal_config))
return handler_class().set_selection(widget=widget, selection=selection) return filtered_signals
if not ignore_errors:
raise ValueError(
f"No matching handler for widget type: {type(widget)} in handler list {FilterIO._handlers}"
)
return None
@staticmethod
def check_input(widget, text: str, ignore_errors=True):
"""
Check if the input text is in the filtered selection.
Args:
widget: Widget instance.
text(str): Input text.
ignore_errors(bool, optional): Whether to ignore if no handler is found.
Returns:
bool: True if the input text is in the filtered selection.
"""
handler_class = FilterIO._find_handler(widget)
if handler_class:
return handler_class().check_input(widget=widget, text=text)
if not ignore_errors:
raise ValueError(
f"No matching handler for widget type: {type(widget)} in handler list {FilterIO._handlers}"
)
return None
@staticmethod
def update_with_kind(
widget, kind: Kind, signal_filter: set, device_info: dict, device_name: str
) -> list[str | tuple]:
"""
Update the selection based on the kind of signal.
Args:
widget: Widget instance.
kind (Kind): The kind of signal to filter.
signal_filter (set): Set of signal kinds to filter.
device_info (dict): Dictionary containing device information.
device_name (str): Name of the device.
Returns:
list[str | tuple]: A list of filtered signals based on the kind.
"""
handler_class = FilterIO._find_handler(widget)
if handler_class:
return handler_class().update_with_kind(
kind=kind,
signal_filter=signal_filter,
device_info=device_info,
device_name=device_name,
)
raise ValueError(
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):
"""
Find the appropriate handler for the widget by checking its base classes.
Args:
widget: Widget instance.
Returns:
handler_class: The handler class if found, otherwise None.
"""
for base in type(widget).__mro__:
if base in FilterIO._handlers:
return FilterIO._handlers[base]
return None
+4 -2
View File
@@ -27,8 +27,10 @@ from qtpy.QtWidgets import (
import bec_widgets import bec_widgets
from bec_widgets.utils.toolbars.splitter import ResizableSpacer from bec_widgets.utils.toolbars.splitter import ResizableSpacer
from bec_widgets.widgets.control.device_input.base_classes.device_input_base import BECDeviceFilter from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import (
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox BECDeviceFilter,
DeviceComboBox,
)
logger = bec_logger.logger logger = bec_logger.logger
+5 -1
View File
@@ -85,7 +85,11 @@ class ComboBoxHandler(WidgetHandler):
def set_value(self, widget: QComboBox, value: int | str) -> None: def set_value(self, widget: QComboBox, value: int | str) -> None:
if isinstance(value, str): if isinstance(value, str):
value = widget.findText(value) index = widget.findText(value)
if index < 0 and widget.isEditable():
widget.setCurrentText(value)
return
value = index
if isinstance(value, int): if isinstance(value, int):
widget.setCurrentIndex(value) widget.setCurrentIndex(value)
@@ -21,9 +21,9 @@ from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.widgets.control.device_control.position_indicator.position_indicator import ( from bec_widgets.widgets.control.device_control.position_indicator.position_indicator import (
PositionIndicator, PositionIndicator,
) )
from bec_widgets.widgets.control.device_input.base_classes.device_input_base import BECDeviceFilter from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import (
from bec_widgets.widgets.control.device_input.device_line_edit.device_line_edit import ( BECDeviceFilter,
DeviceLineEdit, DeviceComboBox,
) )
from bec_widgets.widgets.utility.spinner.spinner import SpinnerWidget from bec_widgets.widgets.utility.spinner.spinner import SpinnerWidget
@@ -257,10 +257,10 @@ class PositionerBoxBase(BECWidget, QWidget):
self._dialog = QDialog(self) self._dialog = QDialog(self)
self._dialog.setWindowTitle("Positioner Selection") self._dialog.setWindowTitle("Positioner Selection")
layout = QVBoxLayout() layout = QVBoxLayout()
line_edit = DeviceLineEdit( line_edit = DeviceComboBox(
self, client=self.client, device_filter=[BECDeviceFilter.POSITIONER] self, client=self.client, device_filter=[BECDeviceFilter.POSITIONER]
) )
line_edit.textChanged.connect(set_positioner) line_edit.currentTextChanged.connect(set_positioner)
layout.addWidget(line_edit) layout.addWidget(line_edit)
close_button = QPushButton("Close") close_button = QPushButton("Close")
close_button.clicked.connect(self._dialog.accept) close_button.clicked.connect(self._dialog.accept)
@@ -1,458 +0,0 @@
from __future__ import annotations
import enum
from bec_lib.device import ComputedSignal, Device, Positioner, ReadoutPriority
from bec_lib.device import Signal as BECSignal
from bec_lib.logger import bec_logger
from pydantic import field_validator
from bec_widgets.utils.bec_connector import ConnectionConfig
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
from bec_widgets.utils.filter_io import FilterIO
from bec_widgets.utils.widget_io import WidgetIO
logger = bec_logger.logger
class BECDeviceFilter(enum.Enum):
"""Filter for the device classes."""
DEVICE = "Device"
POSITIONER = "Positioner"
SIGNAL = "Signal"
COMPUTED_SIGNAL = "ComputedSignal"
class DeviceInputConfig(ConnectionConfig):
device_filter: list[str] = []
readout_filter: list[str] = []
devices: list[str] = []
default: str | None = None
arg_name: str | None = None
apply_filter: bool = True
signal_class_filter: list[str] = []
@field_validator("device_filter")
@classmethod
def check_device_filter(cls, v, values):
valid_device_filters = [entry.value for entry in BECDeviceFilter]
for filt in v:
if filt not in valid_device_filters:
raise ValueError(
f"Device filter {filt} is not a valid device filter {valid_device_filters}."
)
return v
@field_validator("readout_filter")
@classmethod
def check_readout_filter(cls, v, values):
valid_device_filters = [entry.value for entry in ReadoutPriority]
for filt in v:
if filt not in valid_device_filters:
raise ValueError(
f"Device filter {filt} is not a valid device filter {valid_device_filters}."
)
return v
class DeviceInputBase(BECWidget):
"""
Mixin base class for device input widgets.
It allows to filter devices from BEC based on
device class and readout priority.
"""
_device_handler = {
BECDeviceFilter.DEVICE: Device,
BECDeviceFilter.POSITIONER: Positioner,
BECDeviceFilter.SIGNAL: BECSignal,
BECDeviceFilter.COMPUTED_SIGNAL: ComputedSignal,
}
_filter_handler = {
BECDeviceFilter.DEVICE: "filter_to_device",
BECDeviceFilter.POSITIONER: "filter_to_positioner",
BECDeviceFilter.SIGNAL: "filter_to_signal",
BECDeviceFilter.COMPUTED_SIGNAL: "filter_to_computed_signal",
ReadoutPriority.MONITORED: "readout_monitored",
ReadoutPriority.BASELINE: "readout_baseline",
ReadoutPriority.ASYNC: "readout_async",
ReadoutPriority.CONTINUOUS: "readout_continuous",
ReadoutPriority.ON_REQUEST: "readout_on_request",
}
def __init__(self, parent=None, client=None, config=None, gui_id: str | None = None, **kwargs):
if config is None:
config = DeviceInputConfig(widget_class=self.__class__.__name__)
else:
if isinstance(config, dict):
config = DeviceInputConfig(**config)
self.config = config
super().__init__(
parent=parent, client=client, config=config, gui_id=gui_id, theme_update=True, **kwargs
)
self.get_bec_shortcuts()
self._device_filter = []
self._readout_filter = []
self._devices = []
### QtSlots ###
@SafeSlot(str)
def set_device(self, device: str):
"""
Set the device.
Args:
device (str): Default name.
"""
if self.validate_device(device) is True:
WidgetIO.set_value(widget=self, value=device)
self.config.default = device
else:
logger.warning(
f"Device {device} is not in the filtered selection of {self}: {self.devices}."
)
@SafeSlot()
def update_devices_from_filters(self):
"""Update the devices based on the current filter selection
in self.device_filter and self.readout_filter. If apply_filter is False,
it will not apply the filters, store the filter settings and return.
"""
current_device = WidgetIO.get_value(widget=self, as_string=True)
self.config.device_filter = self.device_filter
self.config.readout_filter = self.readout_filter
self.config.signal_class_filter = self.signal_class_filter
if self.apply_filter is False:
return
all_dev = self.dev.enabled_devices
devs = self._filter_devices_by_signal_class(all_dev)
# Filter based on device class
devs = [dev for dev in devs if self._check_device_filter(dev)]
# Filter based on readout priority
devs = [dev for dev in devs if self._check_readout_filter(dev)]
self.devices = [device.name for device in devs]
if current_device != "":
self.set_device(current_device)
@SafeSlot(list)
def set_available_devices(self, devices: list[str]):
"""
Set the devices. If a device in the list is not valid, it will not be considered.
Args:
devices (list[str]): List of devices.
"""
self.apply_filter = False
self.devices = devices
### QtProperties ###
@SafeProperty(
"QStringList",
doc="List of devices. If updated, it will disable the apply filters property.",
)
def devices(self) -> list[str]:
"""
Get the list of devices for the applied filters.
Returns:
list[str]: List of devices.
"""
return self._devices
@devices.setter
def devices(self, value: list):
self._devices = value
self.config.devices = value
FilterIO.set_selection(widget=self, selection=value)
@SafeProperty(str)
def default(self):
"""Get the default device name. If set through this property, it will update only if the device is within the filtered selection."""
return self.config.default
@default.setter
def default(self, value: str):
if self.validate_device(value) is False:
return
self.config.default = value
WidgetIO.set_value(widget=self, value=value)
@SafeProperty(bool)
def apply_filter(self):
"""Apply the filters on the devices."""
return self.config.apply_filter
@apply_filter.setter
def apply_filter(self, value: bool):
self.config.apply_filter = value
self.update_devices_from_filters()
@SafeProperty("QStringList")
def signal_class_filter(self) -> list[str]:
"""
Get the signal class filter for devices.
Returns:
list[str]: List of signal class names used for filtering devices.
"""
return self.config.signal_class_filter
@signal_class_filter.setter
def signal_class_filter(self, value: list[str] | None):
"""
Set the signal class filter and update the device list.
Args:
value (list[str] | None): List of signal class names to filter by.
"""
self.config.signal_class_filter = value or []
self.update_devices_from_filters()
@SafeProperty(bool)
def filter_to_device(self):
"""Include devices in filters."""
return BECDeviceFilter.DEVICE in self.device_filter
@filter_to_device.setter
def filter_to_device(self, value: bool):
if value is True and BECDeviceFilter.DEVICE not in self.device_filter:
self._device_filter.append(BECDeviceFilter.DEVICE)
if value is False and BECDeviceFilter.DEVICE in self.device_filter:
self._device_filter.remove(BECDeviceFilter.DEVICE)
self.update_devices_from_filters()
@SafeProperty(bool)
def filter_to_positioner(self):
"""Include devices of type Positioner in filters."""
return BECDeviceFilter.POSITIONER in self.device_filter
@filter_to_positioner.setter
def filter_to_positioner(self, value: bool):
if value is True and BECDeviceFilter.POSITIONER not in self.device_filter:
self._device_filter.append(BECDeviceFilter.POSITIONER)
if value is False and BECDeviceFilter.POSITIONER in self.device_filter:
self._device_filter.remove(BECDeviceFilter.POSITIONER)
self.update_devices_from_filters()
@SafeProperty(bool)
def filter_to_signal(self):
"""Include devices of type Signal in filters."""
return BECDeviceFilter.SIGNAL in self.device_filter
@filter_to_signal.setter
def filter_to_signal(self, value: bool):
if value is True and BECDeviceFilter.SIGNAL not in self.device_filter:
self._device_filter.append(BECDeviceFilter.SIGNAL)
if value is False and BECDeviceFilter.SIGNAL in self.device_filter:
self._device_filter.remove(BECDeviceFilter.SIGNAL)
self.update_devices_from_filters()
@SafeProperty(bool)
def filter_to_computed_signal(self):
"""Include devices of type ComputedSignal in filters."""
return BECDeviceFilter.COMPUTED_SIGNAL in self.device_filter
@filter_to_computed_signal.setter
def filter_to_computed_signal(self, value: bool):
if value is True and BECDeviceFilter.COMPUTED_SIGNAL not in self.device_filter:
self._device_filter.append(BECDeviceFilter.COMPUTED_SIGNAL)
if value is False and BECDeviceFilter.COMPUTED_SIGNAL in self.device_filter:
self._device_filter.remove(BECDeviceFilter.COMPUTED_SIGNAL)
self.update_devices_from_filters()
@SafeProperty(bool)
def readout_monitored(self):
"""Include devices with readout priority Monitored in filters."""
return ReadoutPriority.MONITORED in self.readout_filter
@readout_monitored.setter
def readout_monitored(self, value: bool):
if value is True and ReadoutPriority.MONITORED not in self.readout_filter:
self._readout_filter.append(ReadoutPriority.MONITORED)
if value is False and ReadoutPriority.MONITORED in self.readout_filter:
self._readout_filter.remove(ReadoutPriority.MONITORED)
self.update_devices_from_filters()
@SafeProperty(bool)
def readout_baseline(self):
"""Include devices with readout priority Baseline in filters."""
return ReadoutPriority.BASELINE in self.readout_filter
@readout_baseline.setter
def readout_baseline(self, value: bool):
if value is True and ReadoutPriority.BASELINE not in self.readout_filter:
self._readout_filter.append(ReadoutPriority.BASELINE)
if value is False and ReadoutPriority.BASELINE in self.readout_filter:
self._readout_filter.remove(ReadoutPriority.BASELINE)
self.update_devices_from_filters()
@SafeProperty(bool)
def readout_async(self):
"""Include devices with readout priority Async in filters."""
return ReadoutPriority.ASYNC in self.readout_filter
@readout_async.setter
def readout_async(self, value: bool):
if value is True and ReadoutPriority.ASYNC not in self.readout_filter:
self._readout_filter.append(ReadoutPriority.ASYNC)
if value is False and ReadoutPriority.ASYNC in self.readout_filter:
self._readout_filter.remove(ReadoutPriority.ASYNC)
self.update_devices_from_filters()
@SafeProperty(bool)
def readout_continuous(self):
"""Include devices with readout priority continuous in filters."""
return ReadoutPriority.CONTINUOUS in self.readout_filter
@readout_continuous.setter
def readout_continuous(self, value: bool):
if value is True and ReadoutPriority.CONTINUOUS not in self.readout_filter:
self._readout_filter.append(ReadoutPriority.CONTINUOUS)
if value is False and ReadoutPriority.CONTINUOUS in self.readout_filter:
self._readout_filter.remove(ReadoutPriority.CONTINUOUS)
self.update_devices_from_filters()
@SafeProperty(bool)
def readout_on_request(self):
"""Include devices with readout priority OnRequest in filters."""
return ReadoutPriority.ON_REQUEST in self.readout_filter
@readout_on_request.setter
def readout_on_request(self, value: bool):
if value is True and ReadoutPriority.ON_REQUEST not in self.readout_filter:
self._readout_filter.append(ReadoutPriority.ON_REQUEST)
if value is False and ReadoutPriority.ON_REQUEST in self.readout_filter:
self._readout_filter.remove(ReadoutPriority.ON_REQUEST)
self.update_devices_from_filters()
### Python Methods and Properties ###
@property
def device_filter(self) -> list[object]:
"""Get the list of filters to apply on the devices."""
return self._device_filter
@property
def readout_filter(self) -> list[str]:
"""Get the list of filters to apply on the devices"""
return self._readout_filter
def get_available_filters(self) -> list:
"""Get the available filters."""
return [entry for entry in BECDeviceFilter]
def get_readout_priority_filters(self) -> list:
"""Get the available readout priority filters."""
return [entry for entry in ReadoutPriority]
def set_device_filter(
self, filter_selection: str | BECDeviceFilter | list[str] | list[BECDeviceFilter]
):
"""
Set the device filter. If None is passed, no filters are applied and all devices included.
Args:
filter_selection (str | list[str]): Device filters. It is recommended to make an enum for the filters.
"""
filters = None
if isinstance(filter_selection, list):
filters = [self._filter_handler.get(entry) for entry in filter_selection]
if isinstance(filter_selection, str) or isinstance(filter_selection, BECDeviceFilter):
filters = [self._filter_handler.get(filter_selection)]
if filters is None or any([entry is None for entry in filters]):
logger.warning(f"Device filter {filter_selection} is not in the device filter list.")
return
for entry in filters:
setattr(self, entry, True)
def set_readout_priority_filter(
self, filter_selection: str | ReadoutPriority | list[str] | list[ReadoutPriority]
):
"""
Set the readout priority filter. If None is passed, all filters are included.
Args:
filter_selection (str | list[str]): Readout priority filters.
"""
filters = None
if isinstance(filter_selection, list):
filters = [self._filter_handler.get(entry) for entry in filter_selection]
if isinstance(filter_selection, str) or isinstance(filter_selection, ReadoutPriority):
filters = [self._filter_handler.get(filter_selection)]
if filters is None or any([entry is None for entry in filters]):
logger.warning(
f"Readout priority filter {filter_selection} is not in the readout priority list."
)
return
for entry in filters:
setattr(self, entry, True)
def _check_device_filter(
self, device: Device | BECSignal | ComputedSignal | Positioner
) -> bool:
"""Check if filter for device type is applied or not.
Args:
device(Device | Signal | ComputedSignal | Positioner): Device object.
"""
return all(isinstance(device, self._device_handler[entry]) for entry in self.device_filter)
def _filter_devices_by_signal_class(
self, devices: list[Device | BECSignal | ComputedSignal | Positioner]
) -> list[Device | BECSignal | ComputedSignal | Positioner]:
"""Filter devices by signal class, if a signal class filter is set."""
if not self.config.signal_class_filter:
return devices
if not self.client or not hasattr(self.client, "device_manager"):
return []
signals = FilterIO.update_with_signal_class(
widget=self, signal_class_filter=self.config.signal_class_filter, client=self.client
)
allowed_devices = {device_name for device_name, _, _ in signals}
return [dev for dev in devices if dev.name in allowed_devices]
def _check_readout_filter(
self, device: Device | BECSignal | ComputedSignal | Positioner
) -> bool:
"""Check if filter for readout priority is applied or not.
Args:
device(Device | Signal | ComputedSignal | Positioner): Device object.
"""
return device.readout_priority in self.readout_filter
def get_device_object(self, device: str) -> object:
"""
Get the device object based on the device name.
Args:
device(str): Device name.
Returns:
object: Device object, can be device of type Device, Positioner, Signal or ComputedSignal.
"""
self.validate_device(device)
dev = getattr(self.dev, device, None)
if dev is None:
raise ValueError(
f"Device {device} is not found in the device manager {self.dev} as enabled device."
)
return dev
def validate_device(self, device: str) -> bool:
"""
Validate the device if it is present in the filtered device selection.
Args:
device(str): Device to validate.
"""
all_devs = [dev.name for dev in self.dev.enabled_devices]
if device in self.devices and device in all_devs:
return True
return False
@@ -1,301 +0,0 @@
from bec_lib.callback_handler import EventType
from bec_lib.device import Signal
from bec_lib.logger import bec_logger
from qtpy.QtCore import Property
from bec_widgets.utils.bec_connector 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
from bec_widgets.utils.ophyd_kind_util import Kind
from bec_widgets.utils.widget_io import WidgetIO
logger = bec_logger.logger
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
signals: list[str] | None = None
class DeviceSignalInputBase(BECWidget):
"""
Mixin base class for device signal input widgets.
Mixin class for device signal input widgets. This class provides methods to get the device signal list and device
signal object based on the current text of the widget.
"""
RPC = False
_filter_handler = {
Kind.hinted: "include_hinted_signals",
Kind.normal: "include_normal_signals",
Kind.config: "include_config_signals",
}
def __init__(
self,
client=None,
config: DeviceSignalInputBaseConfig | dict | None = None,
gui_id: str = None,
**kwargs,
):
self.config = self._process_config_input(config)
super().__init__(client=client, config=self.config, gui_id=gui_id, **kwargs)
self._device = None
self.get_bec_shortcuts()
self._signal_filter = set()
self._signals = []
self._hinted_signals = []
self._normal_signals = []
self._config_signals = []
self._device_update_register = self.bec_dispatcher.client.callbacks.register(
EventType.DEVICE_UPDATE, self.update_signals_from_filters
)
### Qt Slots ###
@SafeSlot(str)
def set_signal(self, signal: str):
"""
Set the signal.
Args:
signal (str): signal name.
"""
if self.validate_signal(signal):
WidgetIO.set_value(widget=self, value=signal)
self.config.default = signal
else:
logger.warning(
f"Signal {signal} not found for device {self.device} and filtered selection {self.signal_filter}."
)
@SafeSlot(str)
def set_device(self, device: str | None):
"""
Set the device. If device is not valid, device will be set to None which happens
Args:
device(str): device name.
"""
if self.validate_device(device) is False:
self._device = None
else:
self._device = device
self.update_signals_from_filters()
@SafeSlot(dict, dict)
@SafeSlot()
def update_signals_from_filters(
self, content: dict | None = None, metadata: dict | None = None
):
"""Update the filters for the device signals based on list in self.signal_filter.
In addition, store the hinted, normal and config signals in separate lists to allow
customisation within QLineEdit.
Note:
Signal and ComputedSignals have no signals. The naming convention follows the device name.
"""
self.config.signal_filter = self.signal_filter
# pylint: disable=protected-access
if not self.validate_device(self._device):
self._device = None
self.config.device = self._device
self._signals = []
self._hinted_signals = []
self._normal_signals = []
self._config_signals = []
FilterIO.set_selection(widget=self, selection=self._signals)
return
device = self.get_device_object(self._device)
device_info = device._info.get("signals", {})
# See above convention for Signals and ComputedSignals
if isinstance(device, Signal):
self._signals = [(self._device, {})]
self._hinted_signals = [(self._device, {})]
self._normal_signals = []
self._config_signals = []
FilterIO.set_selection(widget=self, selection=self._signals)
return
def _update(kind: Kind):
return FilterIO.update_with_kind(
widget=self,
kind=kind,
signal_filter=self.signal_filter,
device_info=device_info,
device_name=self._device,
)
self._hinted_signals = _update(Kind.hinted)
self._normal_signals = _update(Kind.normal)
self._config_signals = _update(Kind.config)
self._signals = self._hinted_signals + self._normal_signals + self._config_signals
FilterIO.set_selection(widget=self, selection=self.signals)
### Qt Properties ###
@Property(str)
def device(self) -> str:
"""Get the selected device."""
if self._device is None:
return ""
return self._device
@device.setter
def device(self, value: str):
"""Set the device and update the filters, only allow devices present in the devicemanager."""
self._device = value
self.config.device = value
self.update_signals_from_filters()
@Property(bool)
def include_hinted_signals(self):
"""Include hinted signals in filters."""
return Kind.hinted in self.signal_filter
@include_hinted_signals.setter
def include_hinted_signals(self, value: bool):
if value:
self._signal_filter.add(Kind.hinted)
else:
self._signal_filter.discard(Kind.hinted)
self.update_signals_from_filters()
@Property(bool)
def include_normal_signals(self):
"""Include normal signals in filters."""
return Kind.normal in self.signal_filter
@include_normal_signals.setter
def include_normal_signals(self, value: bool):
if value:
self._signal_filter.add(Kind.normal)
else:
self._signal_filter.discard(Kind.normal)
self.update_signals_from_filters()
@Property(bool)
def include_config_signals(self):
"""Include config signals in filters."""
return Kind.config in self.signal_filter
@include_config_signals.setter
def include_config_signals(self, value: bool):
if value:
self._signal_filter.add(Kind.config)
else:
self._signal_filter.discard(Kind.config)
self.update_signals_from_filters()
### Properties and Methods ###
@property
def signals(self) -> list[str]:
"""
Get the list of device signals for the applied filters.
Returns:
list[str]: List of device signals.
"""
return self._signals
@signals.setter
def signals(self, value: list[str]):
self._signals = value
self.config.signals = value
FilterIO.set_selection(widget=self, selection=value)
@property
def signal_filter(self) -> list[str]:
"""Get the list of filters to apply on the device signals."""
return self._signal_filter
def get_available_filters(self) -> list[str]:
"""Get the available filters."""
return [entry for entry in self._filter_handler]
def set_filter(self, filter_selection: str | list[str]):
"""
Set the device filter. If None, all devices are included.
Args:
filter_selection (str | list[str]): Device filters from BECDeviceFilter and BECReadoutPriority.
"""
filters = None
if isinstance(filter_selection, list):
filters = [self._filter_handler.get(entry) for entry in filter_selection]
if isinstance(filter_selection, str):
filters = [self._filter_handler.get(filter_selection)]
if filters is None:
return
for entry in filters:
setattr(self, entry, True)
def get_device_object(self, device: str) -> object | None:
"""
Get the device object based on the device name.
Args:
device(str): Device name.
Returns:
object: Device object, can be device of type Device, Positioner, Signal or ComputedSignal.
"""
self.validate_device(device)
dev = getattr(self.dev, device, None)
if dev is None:
logger.warning(f"Device {device} not found in devicemanager.")
return None
return dev
def validate_device(self, device: str | None, raise_on_false: bool = False) -> bool:
"""
Validate the device if it is present in current BEC instance.
Args:
device(str): Device to validate.
raise_on_false(bool): Raise ValueError if device is not found.
"""
if device in self.dev:
return True
if raise_on_false is True:
raise ValueError(f"Device {device} not found in devicemanager.")
return False
def validate_signal(self, signal: str) -> bool:
"""
Validate the signal if it is present in the device signals.
Args:
signal(str): Signal to validate.
"""
for entry in self.signals:
if isinstance(entry, tuple):
entry = entry[0]
if entry == signal:
return True
return False
def _process_config_input(self, config: DeviceSignalInputBaseConfig | dict | None):
if config is None:
return DeviceSignalInputBaseConfig(widget_class=self.__class__.__name__)
return DeviceSignalInputBaseConfig.model_validate(config)
def cleanup(self):
"""
Cleanup the widget.
"""
self.bec_dispatcher.client.callbacks.remove(self._device_update_register)
super().cleanup()
@@ -1,32 +1,81 @@
from __future__ import annotations
import enum
from bec_lib.callback_handler import EventType from bec_lib.callback_handler import EventType
from bec_lib.device import ReadoutPriority from bec_lib.device import ComputedSignal, Device, Positioner, ReadoutPriority
from qtpy.QtCore import QSize, Signal, Slot from bec_lib.device import Signal as BECSignal
from qtpy.QtWidgets import QComboBox, QSizePolicy from bec_lib.logger import bec_logger
from pydantic import Field, field_validator
from qtpy.QtCore import QSize, QStringListModel, Signal, Slot
from qtpy.QtWidgets import QComboBox, QCompleter, QSizePolicy
from bec_widgets.utils.bec_connector import ConnectionConfig
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.colors import get_accent_colors from bec_widgets.utils.colors import get_accent_colors
from bec_widgets.utils.error_popups import SafeProperty from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
from bec_widgets.widgets.control.device_input.base_classes.device_input_base import ( from bec_widgets.utils.filter_io import get_bec_signals_for_classes, replace_combobox_items
BECDeviceFilter,
DeviceInputBase, logger = bec_logger.logger
DeviceInputConfig,
)
class DeviceComboBox(DeviceInputBase, QComboBox): class BECDeviceFilter(enum.Enum):
"""Filter for BEC device classes."""
DEVICE = "Device"
POSITIONER = "Positioner"
SIGNAL = "Signal"
COMPUTED_SIGNAL = "ComputedSignal"
class DeviceInputConfig(ConnectionConfig):
device_filter: list[str] = Field(default_factory=list)
readout_filter: list[str] = Field(default_factory=list)
devices: list[str] = Field(default_factory=list)
default: str | None = None
arg_name: str | None = None
apply_filter: bool = True
signal_class_filter: list[str] = Field(default_factory=list)
autocomplete: bool = False
@field_validator("device_filter")
@classmethod
def check_device_filter(cls, value):
valid_filters = [entry.value for entry in BECDeviceFilter]
for device_filter in value:
if device_filter not in valid_filters:
raise ValueError(
f"Device filter {device_filter} is not a valid device filter {valid_filters}."
)
return value
@field_validator("readout_filter")
@classmethod
def check_readout_filter(cls, value):
valid_filters = [entry.value for entry in ReadoutPriority]
for readout_filter in value:
if readout_filter not in valid_filters:
raise ValueError(
f"Readout filter {readout_filter} is not a valid readout filter {valid_filters}."
)
return value
class DeviceComboBox(BECWidget, QComboBox):
""" """
Combobox widget for device input with autocomplete for device names. Editable combobox for BEC device input.
Args: Args:
parent: Parent widget. parent: Parent widget.
client: BEC client object. client: BEC client object.
config: Device input configuration. config: Device input configuration.
gui_id: GUI ID. gui_id: GUI ID.
device_filter: Device filter, name of the device class from BECDeviceFilter and BECReadoutPriority. Check DeviceInputBase for more details. device_filter: Device class filter from BECDeviceFilter.
readout_priority_filter: Readout priority filter, name of the readout priority class from BECDeviceFilter and BECReadoutPriority. Check DeviceInputBase for more details. readout_priority_filter: Readout priority filter from ReadoutPriority.
available_devices: List of available devices, if passed, it sets apply filters to false and device/readout priority filters will not be applied. available_devices: Explicit list of devices. Passing this disables automatic filtering.
default: Default device name. 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. arg_name: Argument name used by scan/input widgets.
signal_class_filter: List of signal classes to filter the devices by. Only devices with signals of these classes will be shown. signal_class_filter: Only show devices with signals of these classes.
""" """
ICON_NAME = "list_alt" ICON_NAME = "list_alt"
@@ -37,62 +86,97 @@ class DeviceComboBox(DeviceInputBase, QComboBox):
device_reset = Signal() device_reset = Signal()
device_config_update = Signal() device_config_update = Signal()
_device_handler = {
BECDeviceFilter.DEVICE: Device,
BECDeviceFilter.POSITIONER: Positioner,
BECDeviceFilter.SIGNAL: BECSignal,
BECDeviceFilter.COMPUTED_SIGNAL: ComputedSignal,
}
def __init__( def __init__(
self, self,
parent=None, parent=None,
client=None, client=None,
config: DeviceInputConfig = None, config: DeviceInputConfig | dict | None = None,
gui_id: str | None = None, gui_id: str | None = None,
device_filter: BECDeviceFilter | list[BECDeviceFilter] | None = None, device_filter: BECDeviceFilter | str | list[BECDeviceFilter | str] | None = None,
readout_priority_filter: ( readout_priority_filter: str | ReadoutPriority | list[str | ReadoutPriority] | None = None,
str | ReadoutPriority | list[str] | list[ReadoutPriority] | None
) = None,
available_devices: list[str] | None = None, available_devices: list[str] | None = None,
default: str | None = None, default: str | None = None,
arg_name: str | None = None, arg_name: str | None = None,
signal_class_filter: list[str] | None = None, signal_class_filter: list[str] | None = None,
autocomplete: bool | None = None,
**kwargs, **kwargs,
): ):
super().__init__(parent=parent, client=client, gui_id=gui_id, config=config, **kwargs) self.config = self._process_config(config)
if arg_name is not None: super().__init__(
self.config.arg_name = arg_name parent=parent,
self.arg_name = arg_name client=client,
self.setSizePolicy(QSizePolicy.Preferred, QSizePolicy.Fixed) config=self.config,
self.setMinimumSize(QSize(100, 0)) gui_id=gui_id,
theme_update=True,
**kwargs,
)
self.get_bec_shortcuts()
self._device_filter: list[BECDeviceFilter] = []
self._readout_filter: list[ReadoutPriority] = []
self._devices: list[str] = []
self._callback_id = None self._callback_id = None
self._is_valid_input = False self._is_valid_input = False
self._accent_colors = get_accent_colors() self._accent_colors = get_accent_colors()
self._set_first_element_as_empty = False self._set_first_element_as_empty = False
self._completer_model = QStringListModel(self)
self.setEditable(True)
self.setInsertPolicy(QComboBox.NoInsert)
self.setSizePolicy(QSizePolicy.Preferred, QSizePolicy.Fixed)
self.setMinimumSize(QSize(100, 0))
if arg_name is not None:
self.config.arg_name = arg_name
self.arg_name = arg_name
if available_devices is None and self.config.devices:
available_devices = self.config.devices
if device_filter is None and self.config.device_filter:
device_filter = self.config.device_filter
if readout_priority_filter is None and self.config.readout_filter:
readout_priority_filter = self.config.readout_filter
if signal_class_filter is None and self.config.signal_class_filter:
signal_class_filter = self.config.signal_class_filter
if default is None and self.config.default:
default = self.config.default
if autocomplete is not None:
self.config.autocomplete = autocomplete
if self.config.autocomplete:
self.autocomplete = 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.
# Set available devices if passed
if available_devices is not None: if available_devices is not None:
self.set_available_devices(available_devices) self.set_available_devices(available_devices)
# Set readout priority filter default is all
if readout_priority_filter is not None: self.set_readout_priority_filter(
self.set_readout_priority_filter(readout_priority_filter) readout_priority_filter
else: or [
self.set_readout_priority_filter( ReadoutPriority.MONITORED,
[ ReadoutPriority.BASELINE,
ReadoutPriority.MONITORED, ReadoutPriority.ASYNC,
ReadoutPriority.BASELINE, ReadoutPriority.CONTINUOUS,
ReadoutPriority.ASYNC, ReadoutPriority.ON_REQUEST,
ReadoutPriority.CONTINUOUS, ]
ReadoutPriority.ON_REQUEST, )
]
)
# Device filter default is None
if device_filter is not None: if device_filter is not None:
self.set_device_filter(device_filter) self.set_device_filter(device_filter)
if signal_class_filter is not None: if signal_class_filter is not None:
self.signal_class_filter = signal_class_filter self.signal_class_filter = signal_class_filter
# Set default device if passed
if default is not None: if default is not None:
self.set_device(default) self.set_device(default)
else:
self.setCurrentText("")
self._callback_id = self.bec_dispatcher.client.callbacks.register( self._callback_id = self.bec_dispatcher.client.callbacks.register(
EventType.DEVICE_UPDATE, self.on_device_update EventType.DEVICE_UPDATE, self.on_device_update
) )
@@ -100,39 +184,250 @@ class DeviceComboBox(DeviceInputBase, QComboBox):
self.currentTextChanged.connect(self.check_validity) self.currentTextChanged.connect(self.check_validity)
self.check_validity(self.currentText()) self.check_validity(self.currentText())
@staticmethod
def _process_config(config: DeviceInputConfig | dict | None) -> DeviceInputConfig:
if config is None:
return DeviceInputConfig(widget_class="DeviceComboBox")
return DeviceInputConfig.model_validate(config)
@SafeSlot(str)
def set_device(self, device: str):
"""Set the current device if it is valid for the current filters."""
if self.validate_device(device):
self.setCurrentText(device)
self.config.default = device
else:
logger.warning(
f"Device {device} is not in the filtered selection of {self}: {self.devices}."
)
@SafeSlot()
def update_devices_from_filters(self):
"""Refresh the available device list from current device/readout/signal filters."""
current_device = self.currentText()
self.config.device_filter = [entry.value for entry in self.device_filter]
self.config.readout_filter = [entry.value for entry in self.readout_filter]
self.config.signal_class_filter = self.signal_class_filter
if not self.apply_filter:
return
devices = self._filter_devices_by_signal_class(self.dev.enabled_devices)
devices = [device for device in devices if self._check_device_filter(device)]
devices = [device for device in devices if self._check_readout_filter(device)]
self.devices = [device.name for device in devices]
if current_device:
self.setCurrentText(current_device)
self.check_validity(current_device)
@SafeSlot(list)
def set_available_devices(self, devices: list[str]):
"""Use an explicit device list and disable automatic BEC filtering."""
self.apply_filter = False
self.devices = devices
@SafeProperty("QStringList")
def devices(self) -> list[str]:
"""Devices available after filtering."""
return self._devices
@devices.setter
def devices(self, value: list[str]):
self._devices = value
self.config.devices = value
self._replace_items(value)
@SafeProperty(str)
def default(self):
"""Default selected device."""
return self.config.default
@default.setter
def default(self, value: str):
self.set_device(value)
@SafeProperty(bool)
def apply_filter(self):
"""Whether BEC filters are applied to the device list."""
return self.config.apply_filter
@apply_filter.setter
def apply_filter(self, value: bool):
self.config.apply_filter = value
if value:
self.update_devices_from_filters()
@SafeProperty("QStringList")
def signal_class_filter(self) -> list[str]:
"""Signal class names used to restrict devices."""
return self.config.signal_class_filter
@signal_class_filter.setter
def signal_class_filter(self, value: list[str] | None):
self.config.signal_class_filter = value or []
self.update_devices_from_filters()
@SafeProperty(bool)
def filter_to_device(self):
"""Include generic Device objects."""
return BECDeviceFilter.DEVICE in self.device_filter
@filter_to_device.setter
def filter_to_device(self, value: bool):
self._set_device_filter_enabled(BECDeviceFilter.DEVICE, value)
@SafeProperty(bool)
def filter_to_positioner(self):
"""Include Positioner devices."""
return BECDeviceFilter.POSITIONER in self.device_filter
@filter_to_positioner.setter
def filter_to_positioner(self, value: bool):
self._set_device_filter_enabled(BECDeviceFilter.POSITIONER, value)
@SafeProperty(bool)
def filter_to_signal(self):
"""Include Signal devices."""
return BECDeviceFilter.SIGNAL in self.device_filter
@filter_to_signal.setter
def filter_to_signal(self, value: bool):
self._set_device_filter_enabled(BECDeviceFilter.SIGNAL, value)
@SafeProperty(bool)
def filter_to_computed_signal(self):
"""Include ComputedSignal devices."""
return BECDeviceFilter.COMPUTED_SIGNAL in self.device_filter
@filter_to_computed_signal.setter
def filter_to_computed_signal(self, value: bool):
self._set_device_filter_enabled(BECDeviceFilter.COMPUTED_SIGNAL, value)
@SafeProperty(bool)
def readout_monitored(self):
"""Include monitored devices."""
return ReadoutPriority.MONITORED in self.readout_filter
@readout_monitored.setter
def readout_monitored(self, value: bool):
self._set_readout_filter_enabled(ReadoutPriority.MONITORED, value)
@SafeProperty(bool)
def readout_baseline(self):
"""Include baseline devices."""
return ReadoutPriority.BASELINE in self.readout_filter
@readout_baseline.setter
def readout_baseline(self, value: bool):
self._set_readout_filter_enabled(ReadoutPriority.BASELINE, value)
@SafeProperty(bool)
def readout_async(self):
"""Include async devices."""
return ReadoutPriority.ASYNC in self.readout_filter
@readout_async.setter
def readout_async(self, value: bool):
self._set_readout_filter_enabled(ReadoutPriority.ASYNC, value)
@SafeProperty(bool)
def readout_continuous(self):
"""Include continuous devices."""
return ReadoutPriority.CONTINUOUS in self.readout_filter
@readout_continuous.setter
def readout_continuous(self, value: bool):
self._set_readout_filter_enabled(ReadoutPriority.CONTINUOUS, value)
@SafeProperty(bool)
def readout_on_request(self):
"""Include on-request devices."""
return ReadoutPriority.ON_REQUEST in self.readout_filter
@readout_on_request.setter
def readout_on_request(self, value: bool):
self._set_readout_filter_enabled(ReadoutPriority.ON_REQUEST, value)
@SafeProperty(bool) @SafeProperty(bool)
def set_first_element_as_empty(self) -> bool: def set_first_element_as_empty(self) -> bool:
""" """Whether an empty choice is inserted as the first item."""
Whether the first element in the combobox should be empty.
This is useful to allow the user to select a device from the list.
"""
return self._set_first_element_as_empty return self._set_first_element_as_empty
@set_first_element_as_empty.setter @set_first_element_as_empty.setter
def set_first_element_as_empty(self, value: bool) -> None: def set_first_element_as_empty(self, value: bool) -> None:
"""
Set whether the first element in the combobox should be empty.
This is useful to allow the user to select a device from the list.
Args:
value (bool): True if the first element should be empty, False otherwise.
"""
self._set_first_element_as_empty = value self._set_first_element_as_empty = value
if self._set_first_element_as_empty: current_text = self.currentText()
self.insertItem(0, "") if value:
if self.count() == 0 or self.itemText(0) != "":
self.insertItem(0, "")
self.setCurrentIndex(0) self.setCurrentIndex(0)
elif self.count() > 0 and self.itemText(0) == "":
self.removeItem(0)
if not current_text:
self.setCurrentText("")
@SafeProperty(bool)
def autocomplete(self) -> bool:
"""Whether autocomplete suggestions are enabled while editing."""
return self.config.autocomplete
@autocomplete.setter
def autocomplete(self, value: bool) -> None:
self.config.autocomplete = value
if value:
completer = QCompleter(self._completer_model, self)
self.setCompleter(completer)
else: else:
if self.count() > 0 and self.itemText(0) == "": self._restore_default_completer()
self.removeItem(0)
@property
def device_filter(self) -> list[BECDeviceFilter]:
"""Device class filters."""
return self._device_filter
@property
def readout_filter(self) -> list[ReadoutPriority]:
"""Readout priority filters."""
return self._readout_filter
@property
def is_valid_input(self) -> bool:
"""Whether the current text represents a valid device selection."""
return self._is_valid_input
def get_available_filters(self) -> list[BECDeviceFilter]:
"""Return available device class filters."""
return list(BECDeviceFilter)
def get_readout_priority_filters(self) -> list[ReadoutPriority]:
"""Return available readout priority filters."""
return list(ReadoutPriority)
def set_device_filter(
self, filter_selection: BECDeviceFilter | str | list[BECDeviceFilter | str]
):
"""Enable one or more device class filters."""
for device_filter in self._as_list(filter_selection):
normalized = self._normalize_device_filter(device_filter)
if normalized is None:
logger.warning(f"Device filter {device_filter} is not in the device filter list.")
continue
self._set_device_filter_enabled(normalized, True)
def set_readout_priority_filter(
self, filter_selection: ReadoutPriority | str | list[ReadoutPriority | str]
):
"""Enable one or more readout priority filters."""
for readout_filter in self._as_list(filter_selection):
normalized = self._normalize_readout_filter(readout_filter)
if normalized is None:
logger.warning(
f"Readout priority filter {readout_filter} is not in the readout priority list."
)
continue
self._set_readout_filter_enabled(normalized, True)
def on_device_update(self, action: str, content: dict) -> None: def on_device_update(self, action: str, content: dict) -> None:
""" """Refresh filters when the BEC device configuration changes."""
Callback for device update events. Triggers the device_update signal.
Args:
action (str): The action that triggered the event.
content (dict): The content of the config update.
"""
if action in ["add", "remove", "reload"]: if action in ["add", "remove", "reload"]:
self.device_config_update.emit() self.device_config_update.emit()
@@ -143,21 +438,13 @@ class DeviceComboBox(DeviceInputBase, QComboBox):
super().cleanup() super().cleanup()
def get_current_device(self) -> object: def get_current_device(self) -> object:
""" """Return the current BEC device object."""
Get the current device object based on the current value. return self.get_device_object(self._device_name_from_text(self.currentText()))
Returns:
object: Device object, can be device of type Device, Positioner, Signal or ComputedSignal.
"""
dev_name = self.currentText()
return self.get_device_object(dev_name)
@Slot(str) @Slot(str)
def check_validity(self, input_text: str) -> None: def check_validity(self, input_text: str) -> None:
""" """Validate current text and update visual state."""
Check if the current value is a valid device name. if self.validate_device(input_text):
"""
if self.validate_device(input_text) is True:
self._is_valid_input = True self._is_valid_input = True
self.device_selected.emit(input_text) self.device_selected.emit(input_text)
self.setStyleSheet("border: 1px solid transparent;") self.setStyleSheet("border: 1px solid transparent;")
@@ -167,33 +454,105 @@ class DeviceComboBox(DeviceInputBase, QComboBox):
if self.isEnabled(): if self.isEnabled():
self.setStyleSheet("border: 1px solid red;") self.setStyleSheet("border: 1px solid red;")
def validate_device(self, device: str) -> bool: # type: ignore[override] def validate_device(self, device: str | None) -> bool:
""" """Validate a device against the current filtered device selection."""
Extend validation so that previewsignal pseudodevices (labels like if not device:
``"eiger_preview"``) are accepted as valid choices. return False
device_name = self._device_name_from_text(device)
all_devices = [dev.name for dev in self.dev.enabled_devices]
return device_name in self.devices and device_name in all_devices
The validation run only on device not on the previewsignal. def get_device_object(self, device: str) -> object:
"""Return a device object by name."""
dev = getattr(self.dev, device, None)
if dev is None:
raise ValueError(
f"Device {device} is not found in the device manager {self.dev} as enabled device."
)
return dev
Args: @staticmethod
device: The text currently entered/selected. def _as_list(value):
return value if isinstance(value, list) else [value]
Returns: @staticmethod
True if the device is a genuine BEC device *or* one of the def _normalize_device_filter(value: BECDeviceFilter | str) -> BECDeviceFilter | None:
whitelisted previewsignal entries. if isinstance(value, BECDeviceFilter):
""" return value
idx = self.findText(device) return BECDeviceFilter._value2member_map_.get(value)
if idx >= 0 and isinstance(self.itemData(idx), tuple):
device = self.itemData(idx)[0] # type: ignore[assignment]
return super().validate_device(device)
@property @staticmethod
def is_valid_input(self) -> bool: def _normalize_readout_filter(value: ReadoutPriority | str) -> ReadoutPriority | None:
"""Whether the current text represents a valid device selection.""" if isinstance(value, ReadoutPriority):
return self._is_valid_input return value
return ReadoutPriority._value2member_map_.get(value)
def _set_device_filter_enabled(self, device_filter: BECDeviceFilter, enabled: bool):
if enabled and device_filter not in self._device_filter:
self._device_filter.append(device_filter)
elif not enabled and device_filter in self._device_filter:
self._device_filter.remove(device_filter)
self.update_devices_from_filters()
def _set_readout_filter_enabled(self, readout_filter: ReadoutPriority, enabled: bool):
if enabled and readout_filter not in self._readout_filter:
self._readout_filter.append(readout_filter)
elif not enabled and readout_filter in self._readout_filter:
self._readout_filter.remove(readout_filter)
self.update_devices_from_filters()
def _check_device_filter(
self, device: Device | BECSignal | ComputedSignal | Positioner
) -> bool:
if not self.device_filter:
return True
return any(isinstance(device, self._device_handler[entry]) for entry in self.device_filter)
def _check_readout_filter(
self, device: Device | BECSignal | ComputedSignal | Positioner
) -> bool:
if not self.readout_filter:
return True
return device.readout_priority in self.readout_filter
def _filter_devices_by_signal_class(
self, devices: list[Device | BECSignal | ComputedSignal | Positioner]
) -> list[Device | BECSignal | ComputedSignal | Positioner]:
if not self.config.signal_class_filter:
return devices
signals = get_bec_signals_for_classes(
client=self.client, signal_class_filter=self.config.signal_class_filter
)
allowed_devices = {device_name for device_name, _, _ in signals}
return [device for device in devices if device.name in allowed_devices]
def _replace_items(self, devices: list[str]):
current_text = self.currentText()
replace_combobox_items(self, devices)
self._update_completer_model(devices)
if self._set_first_element_as_empty:
self.insertItem(0, "")
self.setCurrentText(current_text)
def _update_completer_model(self, items: list[str]) -> None:
self._completer_model.setStringList(items)
def _restore_default_completer(self) -> None:
if self.completer() is not None and self.completer().model() == self.model():
return
current_text = self.currentText()
self.setEditable(False)
self.setEditable(True)
self.setCurrentText(current_text)
def _device_name_from_text(self, text: str) -> str:
index = self.findText(text)
if index >= 0 and isinstance(self.itemData(index), tuple):
return self.itemData(index)[0]
return text
if __name__ == "__main__": # pragma: no cover if __name__ == "__main__": # pragma: no cover
# pylint: disable=import-outside-toplevel
from qtpy.QtWidgets import ( from qtpy.QtWidgets import (
QApplication, QApplication,
QCheckBox, QCheckBox,
@@ -235,10 +594,7 @@ if __name__ == "__main__": # pragma: no cover
def _apply_filters(): def _apply_filters():
raw = class_input.text().strip() raw = class_input.text().strip()
if raw: combo.signal_class_filter = [entry.strip() for entry in raw.split(",") if entry.strip()]
combo.signal_class_filter = [entry.strip() for entry in raw.split(",") if entry.strip()]
else:
combo.signal_class_filter = []
combo.filter_to_device = filter_device.isChecked() combo.filter_to_device = filter_device.isChecked()
combo.filter_to_positioner = filter_positioner.isChecked() combo.filter_to_positioner = filter_positioner.isChecked()
combo.filter_to_signal = filter_signal.isChecked() combo.filter_to_signal = filter_signal.isChecked()
@@ -1,197 +0,0 @@
from bec_lib.callback_handler import EventType
from bec_lib.device import ReadoutPriority
from bec_lib.logger import bec_logger
from qtpy.QtCore import QSize, Signal, Slot
from qtpy.QtGui import QPainter, QPaintEvent, QPen
from qtpy.QtWidgets import QApplication, QCompleter, QLineEdit, QSizePolicy
from bec_widgets.utils.colors import get_accent_colors
from bec_widgets.widgets.control.device_input.base_classes.device_input_base import (
BECDeviceFilter,
DeviceInputBase,
DeviceInputConfig,
)
logger = bec_logger.logger
class DeviceLineEdit(DeviceInputBase, QLineEdit):
"""
Line edit widget for device input with autocomplete for device names.
Args:
parent: Parent widget.
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.
readout_priority_filter: Readout priority filter, name of the readout priority class from BECDeviceFilter and BECReadoutPriority. Check DeviceInputBase for more details.
available_devices: List of available devices, if passed, it sets apply filters to false and device/readout priority filters will not be applied.
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.
"""
device_selected = Signal(str)
device_config_update = Signal()
PLUGIN = True
RPC = False
ICON_NAME = "edit_note"
def __init__(
self,
parent=None,
client=None,
config: DeviceInputConfig = None,
gui_id: str | None = None,
device_filter: BECDeviceFilter | list[BECDeviceFilter] | None = None,
readout_priority_filter: (
str | ReadoutPriority | list[str] | list[ReadoutPriority] | None
) = None,
available_devices: list[str] | None = None,
default: str | None = None,
arg_name: str | None = None,
**kwargs,
):
self._callback_id = None
self.__is_valid_input = False
self._accent_colors = get_accent_colors()
super().__init__(parent=parent, client=client, gui_id=gui_id, config=config, **kwargs)
self.completer = QCompleter(self)
self.setCompleter(self.completer)
if arg_name is not None:
self.config.arg_name = arg_name
self.arg_name = arg_name
self.setSizePolicy(QSizePolicy.Preferred, QSizePolicy.Fixed)
self.setMinimumSize(QSize(100, 0))
# 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.
# Set available devices if passed
if available_devices is not None:
self.set_available_devices(available_devices)
# Set readout priority filter default is all
if readout_priority_filter is not None:
self.set_readout_priority_filter(readout_priority_filter)
else:
self.set_readout_priority_filter(
[
ReadoutPriority.MONITORED,
ReadoutPriority.BASELINE,
ReadoutPriority.ASYNC,
ReadoutPriority.CONTINUOUS,
ReadoutPriority.ON_REQUEST,
]
)
# Device filter default is None
if device_filter is not None:
self.set_device_filter(device_filter)
# Set default device if passed
if default is not None:
self.set_device(default)
self._callback_id = self.bec_dispatcher.client.callbacks.register(
EventType.DEVICE_UPDATE, self.on_device_update
)
self.device_config_update.connect(self.update_devices_from_filters)
self.textChanged.connect(self.check_validity)
self.check_validity(self.text())
@property
def _is_valid_input(self) -> bool:
"""
Check if the current value is a valid device name.
Returns:
bool: True if the current value is a valid device name, False otherwise.
"""
return self.__is_valid_input
@_is_valid_input.setter
def _is_valid_input(self, value: bool) -> None:
self.__is_valid_input = value
def on_device_update(self, action: str, content: dict) -> None:
"""
Callback for device update events. Triggers the device_update signal.
Args:
action (str): The action that triggered the event.
content (dict): The content of the config update.
"""
if action in ["add", "remove", "reload"]:
self.device_config_update.emit()
def cleanup(self):
"""Cleanup the widget."""
if self._callback_id is not None:
self.bec_dispatcher.client.callbacks.remove(self._callback_id)
super().cleanup()
def get_current_device(self) -> object:
"""
Get the current device object based on the current value.
Returns:
object: Device object, can be device of type Device, Positioner, Signal or ComputedSignal.
"""
dev_name = self.text()
return self.get_device_object(dev_name)
def paintEvent(self, event: QPaintEvent) -> None:
"""Extend the paint event to set the border color based on the validity of the input.
Args:
event (PySide6.QtGui.QPaintEvent) : Paint event.
"""
# logger.info(f"Received paint event: {event} in {self.__class__}")
super().paintEvent(event)
if self._is_valid_input is False and self.isEnabled() is True:
painter = QPainter(self)
pen = QPen()
pen.setWidth(2)
pen.setColor(self._accent_colors.emergency)
painter.setPen(pen)
painter.drawRect(self.rect().adjusted(1, 1, -1, -1))
painter.end()
@Slot(str)
def check_validity(self, input_text: str) -> None:
"""
Check if the current value is a valid device name.
"""
if self.validate_device(input_text) is True:
self._is_valid_input = True
self.device_selected.emit(input_text)
else:
self._is_valid_input = False
self.update()
if __name__ == "__main__": # pragma: no cover
# pylint: disable=import-outside-toplevel
from qtpy.QtWidgets import QVBoxLayout, QWidget
from bec_widgets.utils.colors import apply_theme
from bec_widgets.widgets.control.device_input.signal_combobox.signal_combobox import (
SignalComboBox,
)
app = QApplication([])
apply_theme("dark")
widget = QWidget()
widget.setFixedSize(200, 200)
layout = QVBoxLayout()
widget.setLayout(layout)
line_edit = DeviceLineEdit()
line_edit.filter_to_positioner = True
signal_line_edit = SignalComboBox()
line_edit.textChanged.connect(signal_line_edit.set_device)
line_edit.set_available_devices(["samx", "samy", "samz"])
line_edit.set_device("samx")
layout.addWidget(line_edit)
layout.addWidget(signal_line_edit)
widget.show()
app.exec_()
@@ -1 +0,0 @@
{'files': ['device_line_edit.py']}
@@ -1,59 +0,0 @@
# Copyright (C) 2022 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
from qtpy.QtDesigner import QDesignerCustomWidgetInterface
from qtpy.QtWidgets import QWidget
from bec_widgets.utils.bec_designer import designer_material_icon
from bec_widgets.widgets.control.device_input.device_line_edit.device_line_edit import (
DeviceLineEdit,
)
DOM_XML = """
<ui language='c++'>
<widget class='DeviceLineEdit' name='device_line_edit'>
</widget>
</ui>
"""
class DeviceLineEditPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
def __init__(self):
super().__init__()
self._form_editor = None
def createWidget(self, parent):
if parent is None:
return QWidget()
t = DeviceLineEdit(parent)
return t
def domXml(self):
return DOM_XML
def group(self):
return "BEC Input Widgets"
def icon(self):
return designer_material_icon(DeviceLineEdit.ICON_NAME)
def includeFile(self):
return "device_line_edit"
def initialize(self, form_editor):
self._form_editor = form_editor
def isContainer(self):
return False
def isInitialized(self):
return self._form_editor is not None
def name(self):
return "DeviceLineEdit"
def toolTip(self):
return ""
def whatsThis(self):
return self.toolTip()
@@ -1,17 +0,0 @@
def main(): # pragma: no cover
from qtpy import PYSIDE6
if not PYSIDE6:
print("PYSIDE6 is not available in the environment. Cannot patch designer.")
return
from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection
from bec_widgets.widgets.control.device_input.device_line_edit.device_line_edit_plugin import (
DeviceLineEditPlugin,
)
QPyDesignerCustomWidgetCollection.addCustomWidget(DeviceLineEditPlugin())
if __name__ == "__main__": # pragma: no cover
main()
@@ -1,37 +1,54 @@
from __future__ import annotations from __future__ import annotations
from qtpy.QtCore import QSize, Qt, Signal from bec_lib.callback_handler import EventType
from qtpy.QtWidgets import QComboBox, QSizePolicy from bec_lib.device import Signal as BECSignal
from bec_lib.logger import bec_logger
from qtpy.QtCore import Property, QSize, QStringListModel, Qt, Signal, Slot
from qtpy.QtWidgets import QComboBox, QCompleter, QSizePolicy
from bec_widgets.utils.bec_connector import ConnectionConfig
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
from bec_widgets.utils.filter_io import ComboBoxFilterHandler, FilterIO from bec_widgets.utils.filter_io import (
from bec_widgets.utils.ophyd_kind_util import Kind get_bec_signals_for_classes,
from bec_widgets.widgets.control.device_input.base_classes.device_signal_input_base import ( replace_combobox_items,
DeviceSignalInputBase, signal_items_for_kind,
DeviceSignalInputBaseConfig,
) )
from bec_widgets.utils.ophyd_kind_util import Kind
logger = bec_logger.logger
class SignalComboBox(DeviceSignalInputBase, QComboBox): class SignalComboBoxConfig(ConnectionConfig):
"""Configuration for SignalComboBox."""
signal_filter: 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
signals: list[str] | None = None
autocomplete: bool = False
class SignalComboBox(BECWidget, QComboBox):
""" """
Line edit widget for device input with autocomplete for device names. Editable combobox for selecting BEC device signals.
Args: Args:
parent: Parent widget. parent: Parent widget.
client: BEC client object. client: BEC client object.
config: Device input configuration. config: Signal combobox configuration.
gui_id: GUI ID. gui_id: GUI ID.
device: Device name to filter signals from. 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_filter: Signal kind filters from Kind.
signal_class_filter: List of signal classes to filter the signals by. Only signals of these classes will be shown. signal_class_filter: Signal classes to show.
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. ndim_filter: Dimensionality filter for signal-class based lists.
default: Default device name. default: Default signal 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. arg_name: Argument name used by scan/input widgets.
store_signal_config: Whether to store the full signal config in the combobox item data. store_signal_config: Whether to store signal config in item data.
require_device: If True, signals are only shown/validated when a device is set. 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"] USER_ACCESS = ["set_signal", "set_device", "signals", "get_signal_name"]
@@ -47,289 +64,368 @@ class SignalComboBox(DeviceSignalInputBase, QComboBox):
self, self,
parent=None, parent=None,
client=None, client=None,
config: DeviceSignalInputBaseConfig | None = None, config: SignalComboBoxConfig | dict | None = None,
gui_id: str | None = None, gui_id: str | None = None,
device: str | None = None, device: str | None = None,
signal_filter: list[Kind] | None = None, signal_filter: list[Kind | str] | Kind | str | None = None,
signal_class_filter: list[str] | None = None, signal_class_filter: list[str] | None = None,
ndim_filter: int | list[int] | None = None, ndim_filter: int | list[int] | None = None,
default: str | None = None, default: str | None = None,
arg_name: str | None = None, arg_name: str | None = None,
store_signal_config: bool = True, store_signal_config: bool = True,
require_device: bool = False, require_device: bool = False,
autocomplete: bool | None = None,
**kwargs, **kwargs,
): ):
super().__init__(parent=parent, client=client, gui_id=gui_id, config=config, **kwargs) self.config = self._process_config(config)
super().__init__(parent=parent, client=client, config=self.config, gui_id=gui_id, **kwargs)
self.get_bec_shortcuts()
self._device: str | None = None
self._signal_filter: set[Kind] = set()
self._signals: list[str | tuple[str, dict]] = []
self._hinted_signals: list[tuple[str, dict]] = []
self._normal_signals: list[tuple[str, dict]] = []
self._config_signals: list[tuple[str, dict]] = []
self._set_first_element_as_empty = False
self._signal_class_filter = signal_class_filter or []
self._store_signal_config = store_signal_config
self._require_device = require_device
self._is_valid_input = False
self._completer_model = QStringListModel(self)
if arg_name is not None: if arg_name is not None:
self.config.arg_name = arg_name self.config.arg_name = arg_name
self.arg_name = arg_name self.arg_name = arg_name
if default is not None:
self.set_device(default)
if signal_filter is None and self.config.signal_filter:
signal_filter = self.config.signal_filter
if signal_class_filter is None and self.config.signal_class_filter:
self._signal_class_filter = self.config.signal_class_filter
if ndim_filter is None and self.config.ndim_filter is not None:
ndim_filter = self.config.ndim_filter
if device is None and self.config.device:
device = self.config.device
if default is None and self.config.default:
default = self.config.default
if autocomplete is not None:
self.config.autocomplete = autocomplete
self.config.ndim_filter = ndim_filter
self.setEditable(True)
self.setInsertPolicy(QComboBox.NoInsert)
self.setSizePolicy(QSizePolicy.Preferred, QSizePolicy.Fixed) self.setSizePolicy(QSizePolicy.Preferred, QSizePolicy.Fixed)
self.setMinimumSize(QSize(100, 0)) self.setMinimumSize(QSize(100, 0))
self._set_first_element_as_empty = True if self.config.autocomplete:
self._signal_class_filter = signal_class_filter or [] self.autocomplete = True
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 self._device_update_register = self.bec_dispatcher.client.callbacks.register(
# precedence over values from the passed-in config. Full reconciliation and EventType.DEVICE_UPDATE, self.update_signals_from_filters
# 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) self.currentTextChanged.connect(self.on_text_changed)
# Kind filtering is always applied; class filtering is additive. If signal_filter is None, self.set_filter(signal_filter or [Kind.hinted, Kind.normal, Kind.config])
# 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: if device is not None:
self.set_device(device) self.set_device(device)
if default is not None: if default is not None:
self.set_signal(default) self.set_signal(default)
self.check_validity(self.currentText())
@staticmethod
def _process_config(config: SignalComboBoxConfig | dict | None) -> SignalComboBoxConfig:
if config is None:
return SignalComboBoxConfig(widget_class="SignalComboBox")
return SignalComboBoxConfig.model_validate(config)
@SafeSlot(str)
def set_signal(self, signal: str):
"""Set the current signal if it is available in the combobox."""
display_text = self._display_text_for_signal(signal)
if display_text is None:
logger.warning(
f"Signal {signal} not found for device {self.device} and filtered selection {self.signal_filter}."
)
return
self.setCurrentText(display_text)
self.config.default = signal
@SafeSlot(str) @SafeSlot(str)
def set_device(self, device: str | None): def set_device(self, device: str | None):
""" """Set the device that scopes kind-based signal filtering."""
Set the device. When signal_class_filter is active, ensures base-class if not self.validate_device(device):
logic runs and then refreshes the signal list to show only signals from self._device = None
that device matching the signal class filter. else:
self._device = device
Args: self.config.device = self._device
device(str): device name. self.update_signals_from_filters()
"""
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()
@SafeSlot(dict, dict) @SafeSlot(dict, dict)
def update_signals_from_filters( def update_signals_from_filters(
self, content: dict | None = None, metadata: dict | None = None self, content: dict | None = None, metadata: dict | None = None
): ):
"""Update the filters for the combobox. """Refresh available signals from the current device and filters."""
When signal_class_filter is active, skip the normal Kind-based filtering. self.config.signal_filter = [kind.name for kind in self.signal_filter]
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: if self._signal_class_filter:
self.update_signals_from_signal_classes() self.update_signals_from_signal_classes()
return return
# pylint: disable=protected-access
if FilterIO._find_handler(self) is ComboBoxFilterHandler: if not self.validate_device(self._device):
if len(self._config_signals) > 0: self._device = None
self.insertItem( self.config.device = None
len(self._hinted_signals) + len(self._normal_signals), "Config Signals" self._set_signal_groups([], [], [])
) return
self.model().item(len(self._hinted_signals) + len(self._normal_signals)).setEnabled(
False device = self.get_device_object(self._device)
) device_info = device._info.get("signals", {})
if len(self._normal_signals) > 0:
self.insertItem(len(self._hinted_signals), "Normal Signals") if isinstance(device, BECSignal):
self.model().item(len(self._hinted_signals)).setEnabled(False) self._set_signal_groups([(self._device, {})], [], [])
if len(self._hinted_signals) > 0: return
self.insertItem(0, "Hinted Signals")
self.model().item(0).setEnabled(False) self._set_signal_groups(
signal_items_for_kind(
kind=Kind.hinted,
signal_filter=self.signal_filter,
device_info=device_info,
device_name=self._device,
),
signal_items_for_kind(
kind=Kind.normal,
signal_filter=self.signal_filter,
device_info=device_info,
device_name=self._device,
),
signal_items_for_kind(
kind=Kind.config,
signal_filter=self.signal_filter,
device_info=device_info,
device_name=self._device,
),
)
@Property(str)
def device(self) -> str:
"""Selected device."""
return self._device or ""
@device.setter
def device(self, value: str):
self.set_device(value)
@Property(bool)
def include_hinted_signals(self):
"""Include hinted signals."""
return Kind.hinted in self.signal_filter
@include_hinted_signals.setter
def include_hinted_signals(self, value: bool):
self._set_kind_filter_enabled(Kind.hinted, value)
@Property(bool)
def include_normal_signals(self):
"""Include normal signals."""
return Kind.normal in self.signal_filter
@include_normal_signals.setter
def include_normal_signals(self, value: bool):
self._set_kind_filter_enabled(Kind.normal, value)
@Property(bool)
def include_config_signals(self):
"""Include config signals."""
return Kind.config in self.signal_filter
@include_config_signals.setter
def include_config_signals(self, value: bool):
self._set_kind_filter_enabled(Kind.config, value)
@SafeProperty(bool) @SafeProperty(bool)
def set_first_element_as_empty(self) -> bool: def set_first_element_as_empty(self) -> bool:
""" """Whether an empty choice is inserted as the first item."""
Whether the first element in the combobox should be empty.
This is useful to allow the user to select a device from the list.
"""
return self._set_first_element_as_empty return self._set_first_element_as_empty
@set_first_element_as_empty.setter @set_first_element_as_empty.setter
def set_first_element_as_empty(self, value: bool) -> None: def set_first_element_as_empty(self, value: bool) -> None:
"""
Set whether the first element in the combobox should be empty.
This is useful to allow the user to select a device from the list.
Args:
value (bool): True if the first element should be empty, False otherwise.
"""
self._set_first_element_as_empty = value self._set_first_element_as_empty = value
if self._set_first_element_as_empty: if value:
self.insertItem(0, "") if self.count() == 0 or self.itemText(0) != "":
self.insertItem(0, "")
self.setCurrentIndex(0) self.setCurrentIndex(0)
else: elif self.count() > 0 and self.itemText(0) == "":
if self.count() > 0 and self.itemText(0) == "": self.removeItem(0)
self.removeItem(0)
@SafeProperty("QStringList") @SafeProperty("QStringList")
def signal_class_filter(self) -> list[str]: def signal_class_filter(self) -> list[str]:
""" """Signal class names used to build the signal list."""
Get the list of signal classes to filter.
Returns:
list[str]: List of signal class names to filter.
"""
return self._signal_class_filter return self._signal_class_filter
@signal_class_filter.setter @signal_class_filter.setter
def signal_class_filter(self, value: list[str] | None): def signal_class_filter(self, value: list[str] | None):
""" self._signal_class_filter = value or []
Set the signal class filter. self.config.signal_class_filter = self._signal_class_filter
self.update_signals_from_filters()
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) @SafeProperty(int)
def ndim_filter(self) -> int: def ndim_filter(self) -> int:
"""Dimensionality filter for signals.""" """Dimensionality filter for signal-class based lists."""
return self.config.ndim_filter if isinstance(self.config.ndim_filter, int) else -1 return self.config.ndim_filter if isinstance(self.config.ndim_filter, int) else -1
@ndim_filter.setter @ndim_filter.setter
def ndim_filter(self, value: int): def ndim_filter(self, value: int):
self.config.ndim_filter = None if value < 0 else value self.config.ndim_filter = None if value < 0 else value
if self._signal_class_filter: self.update_signals_from_filters()
self.update_signals_from_signal_classes(ndim_filter=self.config.ndim_filter)
@SafeProperty(bool) @SafeProperty(bool)
def require_device(self) -> bool: def require_device(self) -> bool:
""" """Whether validation/listing requires a selected device."""
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 return self._require_device
@require_device.setter @require_device.setter
def require_device(self, value: bool): def require_device(self, value: bool):
self._require_device = value self._require_device = value
# Rebuild list when toggled, but only when using signal_class_filter self.update_signals_from_filters()
if self._signal_class_filter:
self.update_signals_from_signal_classes()
def set_to_obj_name(self, obj_name: str) -> bool: @SafeProperty(bool)
""" def autocomplete(self) -> bool:
Set the combobox to the object name of the signal. """Whether autocomplete suggestions are enabled while editing."""
return self.config.autocomplete
Args: @autocomplete.setter
obj_name (str): Object name of the signal. def autocomplete(self, value: bool) -> None:
self.config.autocomplete = value
if value:
completer = QCompleter(self._completer_model, self)
self.setCompleter(completer)
else:
self._restore_default_completer()
Returns: @property
bool: True if the object name was found and set, False otherwise. def signals(self) -> list[str | tuple[str, dict]]:
""" """Available signals after filtering."""
for i in range(self.count()): return self._signals
signal_data = self.itemData(i)
if signal_data and signal_data.get("obj_name") == obj_name: @signals.setter
self.setCurrentIndex(i) def signals(self, value: list[str | tuple[str, dict]]):
return True self._signals = value
self.config.signals = [entry[0] if isinstance(entry, tuple) else entry for entry in value]
self._replace_signal_items()
@property
def signal_filter(self) -> set[Kind]:
"""Signal kind filters."""
return self._signal_filter
@property
def is_valid_input(self) -> bool:
"""Whether the current text represents a valid signal selection."""
return self._is_valid_input
@property
def selected_signal_comp_name(self) -> str:
"""Component name for the current signal, falling back to object name."""
index = self._find_signal_index(self.currentText())
if index < 0:
return self.get_signal_name()
signal_info = self.itemData(index)
if isinstance(signal_info, dict):
return signal_info.get("component_name") or self.get_signal_name()
return self.get_signal_name()
def set_filter(self, filter_selection: Kind | str | list[Kind | str] | None):
"""Enable one or more signal kind filters."""
if filter_selection is None:
return
filters = filter_selection if isinstance(filter_selection, list) else [filter_selection]
for signal_filter in filters:
kind = self._normalize_kind(signal_filter)
if kind is not None:
self._signal_filter.add(kind)
self.update_signals_from_filters()
def get_available_filters(self) -> list[Kind]:
"""Return available signal kind filters."""
return [Kind.hinted, Kind.normal, Kind.config]
def get_device_object(self, device: str) -> object | None:
"""Return a BEC device object by name."""
dev = getattr(self.dev, device, None)
if dev is None:
logger.warning(f"Device {device} not found in devicemanager.")
return None
return dev
def validate_device(self, device: str | None, raise_on_false: bool = False) -> bool:
"""Validate that a device exists in the current device manager."""
if device in self.dev:
return True
if raise_on_false:
raise ValueError(f"Device {device} not found in devicemanager.")
return False return False
def set_to_first_enabled(self) -> bool: def validate_signal(self, signal: str) -> bool:
""" """Validate a signal by display text, object name, or component name."""
Set the combobox to the first enabled item. return self._display_text_for_signal(signal) is not None
Returns: def set_to_obj_name(self, obj_name: str) -> bool:
bool: True if an enabled item was found and set, False otherwise. """Select the item whose signal config has the given object name."""
""" index = self._find_signal_index(obj_name)
for i in range(self.count()): if index < 0:
if self.model().item(i).isEnabled(): return False
self.setCurrentIndex(i) self.setCurrentIndex(index)
return True
def set_to_first_enabled(self) -> bool:
"""Select the first enabled item."""
for index in range(self.count()):
item = self.model().item(index)
if item is not None and item.isEnabled():
self.setCurrentIndex(index)
return True return True
return False return False
def get_signal_name(self) -> str: def get_signal_name(self) -> str:
""" """Return the selected signal object name when available."""
Get the signal name from the combobox. current_text = self.currentText()
index = self._find_signal_index(current_text)
Returns: if index < 0:
str: The signal name. return current_text
"""
signal_name = self.currentText()
index = self.findText(signal_name)
if index == -1:
return signal_name
signal_info = self.itemData(index) signal_info = self.itemData(index)
if signal_info: if isinstance(signal_info, dict):
signal_name = signal_info.get("obj_name", signal_name) return signal_info.get("obj_name") or current_text
return current_text
return signal_name if signal_name else ""
def get_signal_config(self) -> dict | None: def get_signal_config(self) -> dict | None:
""" """Return the selected signal config if item-data storage is enabled."""
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: if not self._store_signal_config:
return None return None
signal_info = self.itemData(self.currentIndex())
index = self.currentIndex() return signal_info if isinstance(signal_info, dict) else None
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): def update_signals_from_signal_classes(self, ndim_filter: int | list[int] | None = None):
""" """Refresh signals from device_manager.get_bec_signals for class-based filtering."""
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: if not self._signal_class_filter:
return return
if self._require_device and not self._device: if self._require_device and not self._device:
self.clear() self.signals = []
self._signals = []
FilterIO.set_selection(widget=self, selection=self._signals)
return return
# Update stored ndim_filter if a new one is provided
if ndim_filter is not None: if ndim_filter is not None:
self.config.ndim_filter = ndim_filter self.config.ndim_filter = ndim_filter
self.clear() signals = get_bec_signals_for_classes(
# 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, client=self.client,
ndim_filter=self.config.ndim_filter, # Pass ndim_filter to FilterIO signal_class_filter=self._signal_class_filter,
ndim_filter=self.config.ndim_filter,
) )
# Track signals for validation and FilterIO selection self.clear()
self._signals = [] self._signals = []
for device_name, signal_name, signal_config in 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: if self._device and device_name != self._device:
continue continue
if self._signal_filter: if self._signal_filter:
@@ -339,53 +435,44 @@ class SignalComboBox(DeviceSignalInputBase, QComboBox):
}: }:
continue 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: if self._store_signal_config:
self.addItem(signal_name, signal_config) self.addItem(signal_name, signal_config)
else: else:
self.addItem(signal_name) self.addItem(signal_name)
# Track for validation
self._signals.append(signal_name) self._signals.append(signal_name)
storage_name = signal_config.get("storage_name", "")
# Set tooltip to storage_name (Qt.ToolTipRole = 3)
if storage_name: if storage_name:
self.setItemData(self.count() - 1, storage_name, Qt.ItemDataRole.ToolTipRole) self.setItemData(self.count() - 1, storage_name, Qt.ItemDataRole.ToolTipRole)
# Keep FilterIO selection in sync for validate_signal self.config.signals = [
FilterIO.set_selection(widget=self, selection=self._signals) entry if isinstance(entry, str) else entry[0] for entry in self._signals
]
self._update_completer_model(self.config.signals)
if self._set_first_element_as_empty and self.count() > 0 and self.itemText(0) != "":
self.insertItem(0, "")
@SafeSlot() @SafeSlot()
def reset_selection(self): def reset_selection(self):
"""Reset the selection of the combobox.""" """Reset the current selection and refresh available signals."""
self.clear() self.setCurrentText("")
self.setItemText(0, "Select a device")
self.update_signals_from_filters() self.update_signals_from_filters()
self.device_signal_changed.emit("") self.device_signal_changed.emit("")
@SafeSlot(str) @SafeSlot(str)
def on_text_changed(self, text: str): def on_text_changed(self, text: str):
"""Validate and emit only when the signal is valid. """Validate the current text when edited or selected."""
For a positioner, the readback value has to be renamed to the device name.
When using signal_class_filter, device validation is skipped.
"""
self.check_validity(text) self.check_validity(text)
@Slot(str)
def check_validity(self, input_text: str) -> None: def check_validity(self, input_text: str) -> None:
"""Check if the current value is a valid signal and emit only when valid.""" """Validate current text and update visual state."""
if self._signal_class_filter: if self._signal_class_filter:
if self._require_device and (not self._device or not input_text): is_valid = not (self._require_device and not self._device) and self.validate_signal(
is_valid = False input_text
else: )
is_valid = self.validate_signal(input_text)
else: else:
if self._require_device and not self.validate_device(self._device): is_valid = self.validate_device(self._device) and self.validate_signal(input_text)
is_valid = False
else:
is_valid = self.validate_device(self._device) and self.validate_signal(input_text)
if is_valid: if is_valid:
self._is_valid_input = True self._is_valid_input = True
@@ -397,18 +484,105 @@ class SignalComboBox(DeviceSignalInputBase, QComboBox):
if self.isEnabled(): if self.isEnabled():
self.setStyleSheet("border: 1px solid red;") self.setStyleSheet("border: 1px solid red;")
@property def cleanup(self):
def selected_signal_comp_name(self) -> str: """Cleanup the widget."""
return dict(self.signals).get(self.currentText(), {}).get("component_name", "") self.bec_dispatcher.client.callbacks.remove(self._device_update_register)
super().cleanup()
@property @staticmethod
def is_valid_input(self) -> bool: def _normalize_kind(value: Kind | str) -> Kind | None:
"""Whether the current text represents a valid signal selection.""" if isinstance(value, Kind):
return self._is_valid_input return value
return Kind.__members__.get(value) or Kind.__members__.get(value.lower())
def _set_kind_filter_enabled(self, kind: Kind, enabled: bool):
if enabled:
self._signal_filter.add(kind)
else:
self._signal_filter.discard(kind)
self.update_signals_from_filters()
def _set_signal_groups(
self,
hinted: list[tuple[str, dict]],
normal: list[tuple[str, dict]],
config: list[tuple[str, dict]],
) -> None:
self._hinted_signals = hinted
self._normal_signals = normal
self._config_signals = config
self.signals = self._hinted_signals + self._normal_signals + self._config_signals
self._insert_group_headers()
def _replace_signal_items(self):
replace_combobox_items(self, self._signals)
self._update_completer_model(self._signal_display_texts(self._signals))
if self._set_first_element_as_empty and self.count() > 0 and self.itemText(0) != "":
self.insertItem(0, "")
def _insert_group_headers(self):
offset = (
1
if self._set_first_element_as_empty and self.count() > 0 and self.itemText(0) == ""
else 0
)
if self._config_signals:
index = offset + len(self._hinted_signals) + len(self._normal_signals)
self.insertItem(index, "Config Signals")
self.model().item(index).setEnabled(False)
if self._normal_signals:
index = offset + len(self._hinted_signals)
self.insertItem(index, "Normal Signals")
self.model().item(index).setEnabled(False)
if self._hinted_signals:
index = offset
self.insertItem(index, "Hinted Signals")
self.model().item(index).setEnabled(False)
def _display_text_for_signal(self, signal: str) -> str | None:
for entry in self._signals:
display_text = entry[0] if isinstance(entry, tuple) else entry
if display_text == signal:
return display_text
if isinstance(entry, tuple) and self._signal_info_matches(entry[1], signal):
return display_text
return None
@staticmethod
def _signal_info_matches(signal_info: dict, signal: str) -> bool:
return signal in {
signal_info.get("obj_name"),
signal_info.get("component_name"),
signal_info.get("component_name", "").replace(".", "_"),
}
def _find_signal_index(self, signal: str) -> int:
index = self.findText(signal)
if index >= 0:
return index
for item_index in range(self.count()):
signal_info = self.itemData(item_index)
if isinstance(signal_info, dict) and self._signal_info_matches(signal_info, signal):
return item_index
return -1
@staticmethod
def _signal_display_texts(signals: list[str | tuple[str, dict]]) -> list[str]:
return [entry[0] if isinstance(entry, tuple) else entry for entry in signals]
def _update_completer_model(self, items: list[str]) -> None:
self._completer_model.setStringList(items)
def _restore_default_completer(self) -> None:
if self.completer() is not None and self.completer().model() == self.model():
return
current_text = self.currentText()
self.setEditable(False)
self.setEditable(True)
self.setCurrentText(current_text)
if __name__ == "__main__": # pragma: no cover if __name__ == "__main__": # pragma: no cover
# pylint: disable=import-outside-toplevel
from qtpy.QtWidgets import QApplication, QVBoxLayout, QWidget from qtpy.QtWidgets import QApplication, QVBoxLayout, QWidget
from bec_widgets.utils.colors import apply_theme from bec_widgets.utils.colors import apply_theme
@@ -417,16 +591,14 @@ if __name__ == "__main__": # pragma: no cover
apply_theme("dark") apply_theme("dark")
widget = QWidget() widget = QWidget()
widget.setFixedSize(200, 200) widget.setFixedSize(200, 200)
layout = QVBoxLayout() layout = QVBoxLayout(widget)
widget.setLayout(layout)
box = SignalComboBox( box = SignalComboBox(
device="waveform", device="waveform",
signal_class_filter=["AsyncSignal", "AsyncMultiSignal"], signal_class_filter=["AsyncSignal", "AsyncMultiSignal"],
ndim_filter=[1, 2], ndim_filter=[1, 2],
store_signal_config=True, store_signal_config=True,
signal_filter=[Kind.hinted, Kind.normal, Kind.config], signal_filter=[Kind.hinted, Kind.normal, Kind.config],
) # change signal filter class to test )
box.setEditable(True)
layout.addWidget(box) layout.addWidget(box)
widget.show() widget.show()
app.exec_() app.exec_()
@@ -1,17 +0,0 @@
def main(): # pragma: no cover
from qtpy import PYSIDE6
if not PYSIDE6:
print("PYSIDE6 is not available in the environment. Cannot patch designer.")
return
from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection
from bec_widgets.widgets.control.device_input.signal_line_edit.signal_line_edit_plugin import (
SignalLineEditPlugin,
)
QPyDesignerCustomWidgetCollection.addCustomWidget(SignalLineEditPlugin())
if __name__ == "__main__": # pragma: no cover
main()
@@ -1,169 +0,0 @@
from bec_lib.device import Positioner
from qtpy.QtCore import QSize, Signal, Slot
from qtpy.QtGui import QPainter, QPaintEvent, QPen
from qtpy.QtWidgets import QCompleter, QLineEdit, QSizePolicy
from bec_widgets.utils.colors import get_accent_colors
from bec_widgets.utils.ophyd_kind_util import Kind
from bec_widgets.widgets.control.device_input.base_classes.device_signal_input_base import (
DeviceSignalInputBase,
)
class SignalLineEdit(DeviceSignalInputBase, QLineEdit):
"""
Line edit widget for device input with autocomplete for device names.
Args:
parent: Parent widget.
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.
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.
"""
USER_ACCESS = ["_is_valid_input", "set_signal", "set_device", "signals"]
device_signal_changed = Signal(str)
PLUGIN = True
RPC = False
ICON_NAME = "vital_signs"
def __init__(
self,
parent=None,
client=None,
config: DeviceSignalInputBase = None,
gui_id: str | None = None,
device: str | None = None,
signal_filter: str | list[str] | None = None,
default: str | None = None,
arg_name: str | None = None,
**kwargs,
):
self.__is_valid_input = False
super().__init__(parent=parent, client=client, gui_id=gui_id, config=config, **kwargs)
self._accent_colors = get_accent_colors()
self.completer = QCompleter(self)
self.setCompleter(self.completer)
if arg_name is not None:
self.config.arg_name = arg_name
self.arg_name = arg_name
if default is not None:
self.set_device(default)
self.setSizePolicy(QSizePolicy.Preferred, QSizePolicy.Fixed)
self.setMinimumSize(QSize(100, 0))
# 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.
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)
self.textChanged.connect(self.check_validity)
self.check_validity(self.text())
@property
def _is_valid_input(self) -> bool:
"""
Check if the current value is a valid device name.
Returns:
bool: True if the current value is a valid device name, False otherwise.
"""
return self.__is_valid_input
@_is_valid_input.setter
def _is_valid_input(self, value: bool) -> None:
self.__is_valid_input = value
def get_current_device(self) -> object:
"""
Get the current device object based on the current value.
Returns:
object: Device object, can be device of type Device, Positioner, Signal or ComputedSignal.
"""
dev_name = self.text()
return self.get_device_object(dev_name)
def paintEvent(self, event: QPaintEvent) -> None:
"""Extend the paint event to set the border color based on the validity of the input.
Args:
event (PySide6.QtGui.QPaintEvent) : Paint event.
"""
super().paintEvent(event)
painter = QPainter(self)
pen = QPen()
pen.setWidth(2)
if self._is_valid_input is False and self.isEnabled() is True:
pen.setColor(self._accent_colors.emergency)
painter.setPen(pen)
painter.drawRect(self.rect().adjusted(1, 1, -1, -1))
@Slot(str)
def check_validity(self, input_text: str) -> None:
"""
Check if the current value is a valid device name.
"""
if self.validate_signal(input_text) is True:
self._is_valid_input = True
self.on_text_changed(input_text)
else:
self._is_valid_input = False
self.update()
@Slot(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.
For a positioner, the readback value has to be renamed to the device name.
Args:
text (str): Text in the combobox.
"""
print("test")
if self.validate_device(self.device) is False:
return
if self.validate_signal(text) is False:
return
if text == "readback" and isinstance(self.get_device_object(self.device), Positioner):
device_signal = self.device
else:
device_signal = f"{self.device}_{text}"
self.device_signal_changed.emit(device_signal)
if __name__ == "__main__": # pragma: no cover
# pylint: disable=import-outside-toplevel
from qtpy.QtWidgets import QApplication, QVBoxLayout, QWidget
from bec_widgets.utils.colors import apply_theme
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import (
DeviceComboBox,
)
app = QApplication([])
apply_theme("dark")
widget = QWidget()
widget.setFixedSize(200, 200)
layout = QVBoxLayout()
widget.setLayout(layout)
device_line_edit = DeviceComboBox()
device_line_edit.filter_to_positioner = True
signal_line_edit = SignalLineEdit()
device_line_edit.device_selected.connect(signal_line_edit.set_device)
layout.addWidget(device_line_edit)
layout.addWidget(signal_line_edit)
widget.show()
app.exec_()
@@ -1 +0,0 @@
{'files': ['signal_line_edit.py']}
@@ -1,59 +0,0 @@
# Copyright (C) 2022 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
from qtpy.QtDesigner import QDesignerCustomWidgetInterface
from qtpy.QtWidgets import QWidget
from bec_widgets.utils.bec_designer import designer_material_icon
from bec_widgets.widgets.control.device_input.signal_line_edit.signal_line_edit import (
SignalLineEdit,
)
DOM_XML = """
<ui language='c++'>
<widget class='SignalLineEdit' name='signal_line_edit'>
</widget>
</ui>
"""
class SignalLineEditPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
def __init__(self):
super().__init__()
self._form_editor = None
def createWidget(self, parent):
if parent is None:
return QWidget()
t = SignalLineEdit(parent)
return t
def domXml(self):
return DOM_XML
def group(self):
return "BEC Input Widgets"
def icon(self):
return designer_material_icon(SignalLineEdit.ICON_NAME)
def includeFile(self):
return "signal_line_edit"
def initialize(self, form_editor):
self._form_editor = form_editor
def isContainer(self):
return False
def isInitialized(self):
return self._form_editor is not None
def name(self):
return "SignalLineEdit"
def toolTip(self):
return ""
def whatsThis(self):
return self.toolTip()
@@ -21,9 +21,9 @@ from qtpy.QtWidgets import (
) )
from bec_widgets.utils.widget_io import WidgetIO from bec_widgets.utils.widget_io import WidgetIO
from bec_widgets.widgets.control.device_input.base_classes.device_input_base import BECDeviceFilter from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import (
from bec_widgets.widgets.control.device_input.device_line_edit.device_line_edit import ( BECDeviceFilter,
DeviceLineEdit, DeviceComboBox,
) )
logger = bec_logger.logger logger = bec_logger.logger
@@ -164,8 +164,8 @@ class ScanCheckBox(QCheckBox):
class ScanGroupBox(QGroupBox): class ScanGroupBox(QGroupBox):
WIDGET_HANDLER = { WIDGET_HANDLER = {
ScanArgType.DEVICE: DeviceLineEdit, ScanArgType.DEVICE: DeviceComboBox,
ScanArgType.DEVICEBASE: DeviceLineEdit, ScanArgType.DEVICEBASE: DeviceComboBox,
ScanArgType.FLOAT: ScanDoubleSpinBox, ScanArgType.FLOAT: ScanDoubleSpinBox,
ScanArgType.INT: ScanSpinBox, ScanArgType.INT: ScanSpinBox,
ScanArgType.BOOL: ScanCheckBox, ScanArgType.BOOL: ScanCheckBox,
@@ -271,9 +271,17 @@ class ScanGroupBox(QGroupBox):
continue continue
if default == "_empty": if default == "_empty":
default = None default = None
widget = widget_class(parent=self.parent(), arg_name=arg_name, default=default) if widget_class is DeviceComboBox:
if isinstance(widget, DeviceLineEdit): widget = widget_class(
widget.set_device_filter(BECDeviceFilter.DEVICE) parent=self.parent(),
arg_name=arg_name,
default=default,
device_filter=BECDeviceFilter.DEVICE,
autocomplete=True,
)
else:
widget = widget_class(parent=self.parent(), arg_name=arg_name, default=default)
if isinstance(widget, DeviceComboBox):
self.selected_devices[widget] = "" self.selected_devices[widget] = ""
widget.device_selected.connect(self.emit_device_selected) widget.device_selected.connect(self.emit_device_selected)
if isinstance(widget, ScanLiteralsComboBox): if isinstance(widget, ScanLiteralsComboBox):
@@ -311,7 +319,7 @@ class ScanGroupBox(QGroupBox):
return return
for widget in self.widgets[-len(self.inputs) :]: for widget in self.widgets[-len(self.inputs) :]:
if isinstance(widget, DeviceLineEdit): if isinstance(widget, DeviceComboBox):
self.selected_devices[widget] = "" self.selected_devices[widget] = ""
widget.close() widget.close()
widget.deleteLater() widget.deleteLater()
@@ -323,7 +331,7 @@ class ScanGroupBox(QGroupBox):
def remove_all_widget_bundles(self): def remove_all_widget_bundles(self):
"""Remove every widget bundle from the scan control layout.""" """Remove every widget bundle from the scan control layout."""
for widget in list(self.widgets): for widget in list(self.widgets):
if isinstance(widget, DeviceLineEdit): if isinstance(widget, DeviceComboBox):
self.selected_devices.pop(widget, None) self.selected_devices.pop(widget, None)
widget.close() widget.close()
widget.deleteLater() widget.deleteLater()
@@ -360,8 +368,10 @@ class ScanGroupBox(QGroupBox):
for j in range(self.layout.columnCount()): for j in range(self.layout.columnCount()):
try: # In case that the bundle size changes try: # In case that the bundle size changes
widget = self.layout.itemAtPosition(i, j).widget() widget = self.layout.itemAtPosition(i, j).widget()
if isinstance(widget, DeviceLineEdit) and device_object: if isinstance(widget, DeviceComboBox) and device_object:
value = widget.get_current_device() value = widget.get_current_device()
elif isinstance(widget, DeviceComboBox):
value = widget.currentText()
else: else:
value = WidgetIO.get_value(widget) value = WidgetIO.get_value(widget)
args.append(value) args.append(value)
@@ -373,8 +383,10 @@ class ScanGroupBox(QGroupBox):
kwargs = {} kwargs = {}
for i in range(self.layout.columnCount()): for i in range(self.layout.columnCount()):
widget = self.layout.itemAtPosition(1, i).widget() widget = self.layout.itemAtPosition(1, i).widget()
if isinstance(widget, DeviceLineEdit) and device_object: if isinstance(widget, DeviceComboBox) and device_object:
value = widget.get_current_device().name value = widget.get_current_device().name
elif isinstance(widget, DeviceComboBox):
value = widget.currentText()
elif isinstance(widget, ScanLiteralsComboBox): elif isinstance(widget, ScanLiteralsComboBox):
value = widget.get_value() value = widget.get_value()
else: else:
@@ -390,7 +402,7 @@ class ScanGroupBox(QGroupBox):
if item is not None: if item is not None:
widget = item.widget() widget = item.widget()
if widget is not None: if widget is not None:
if isinstance(widget, DeviceLineEdit): if isinstance(widget, DeviceComboBox):
widget_rows += 1 widget_rows += 1
return widget_rows return widget_rows
@@ -3,8 +3,10 @@ from qtpy.QtWidgets import QHBoxLayout, QSizePolicy, QWidget
from bec_widgets.utils.toolbars.actions import NoCheckDelegate, WidgetAction from bec_widgets.utils.toolbars.actions import NoCheckDelegate, WidgetAction
from bec_widgets.utils.toolbars.bundles import ToolbarBundle, ToolbarComponents from bec_widgets.utils.toolbars.bundles import ToolbarBundle, ToolbarComponents
from bec_widgets.utils.toolbars.connections import BundleConnection from bec_widgets.utils.toolbars.connections import BundleConnection
from bec_widgets.widgets.control.device_input.base_classes.device_input_base import BECDeviceFilter from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import (
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox BECDeviceFilter,
DeviceComboBox,
)
class MotorSelection(QWidget): class MotorSelection(QWidget):
@@ -6,8 +6,10 @@ from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.toolbars.actions import DeviceComboBoxAction, WidgetAction from bec_widgets.utils.toolbars.actions import DeviceComboBoxAction, WidgetAction
from bec_widgets.utils.toolbars.bundles import ToolbarComponents from bec_widgets.utils.toolbars.bundles import ToolbarComponents
from bec_widgets.utils.toolbars.toolbar import ToolbarBundle from bec_widgets.utils.toolbars.toolbar import ToolbarBundle
from bec_widgets.widgets.control.device_input.base_classes.device_input_base import BECDeviceFilter from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import (
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox BECDeviceFilter,
DeviceComboBox,
)
from bec_widgets.widgets.utility.visual.colormap_widget.colormap_widget import BECColorMapWidget from bec_widgets.widgets.utility.visual.colormap_widget.colormap_widget import BECColorMapWidget
@@ -58,7 +58,7 @@
</widget> </widget>
</item> </item>
<item row="0" column="1"> <item row="0" column="1">
<widget class="DeviceLineEdit" name="device_x"/> <widget class="DeviceComboBox" name="device_x"/>
</item> </item>
<item row="1" column="0"> <item row="1" column="0">
<widget class="QLabel" name="label_2"> <widget class="QLabel" name="label_2">
@@ -87,7 +87,7 @@
</widget> </widget>
</item> </item>
<item row="0" column="1"> <item row="0" column="1">
<widget class="DeviceLineEdit" name="device_y"/> <widget class="DeviceComboBox" name="device_y"/>
</item> </item>
<item row="1" column="0"> <item row="1" column="0">
<widget class="QLabel" name="label_4"> <widget class="QLabel" name="label_4">
@@ -116,7 +116,7 @@
</widget> </widget>
</item> </item>
<item row="0" column="1"> <item row="0" column="1">
<widget class="DeviceLineEdit" name="device_z"/> <widget class="DeviceComboBox" name="device_z"/>
</item> </item>
<item row="1" column="0"> <item row="1" column="0">
<widget class="QLabel" name="label_6"> <widget class="QLabel" name="label_6">
@@ -135,9 +135,9 @@
</widget> </widget>
<customwidgets> <customwidgets>
<customwidget> <customwidget>
<class>DeviceLineEdit</class> <class>DeviceComboBox</class>
<extends>QLineEdit</extends> <extends>QComboBox</extends>
<header>device_line_edit</header> <header>device_combo_box</header>
</customwidget> </customwidget>
<customwidget> <customwidget>
<class>ToggleSwitch</class> <class>ToggleSwitch</class>
@@ -79,7 +79,7 @@ class CurveRow(QTreeWidgetItem):
Columns: Columns:
0: Actions (delete or "Add DAP" if source=device) 0: Actions (delete or "Add DAP" if source=device)
1..2: DeviceLineEdit and QLineEdit if source=device, or "Model" label and DapComboBox if source=dap 1..2: DeviceComboBox and QLineEdit if source=device, or "Model" label and DapComboBox if source=dap
3: ColorButton 3: ColorButton
4: Style QComboBox 4: Style QComboBox
5: Pen width QSpinBox 5: Pen width QSpinBox
@@ -25,9 +25,7 @@ from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.colors import get_accent_colors from bec_widgets.utils.colors import get_accent_colors
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
from bec_widgets.utils.ophyd_kind_util import Kind from bec_widgets.utils.ophyd_kind_util import Kind
from bec_widgets.widgets.control.device_input.device_line_edit.device_line_edit import ( from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox
DeviceLineEdit,
)
from bec_widgets.widgets.control.device_input.signal_combobox.signal_combobox import SignalComboBox from bec_widgets.widgets.control.device_input.signal_combobox.signal_combobox import SignalComboBox
if TYPE_CHECKING: if TYPE_CHECKING:
@@ -58,7 +56,7 @@ class ChoiceDialog(QDialog):
layout = QHBoxLayout() layout = QHBoxLayout()
self._device_field = DeviceLineEdit(parent=parent, client=client) self._device_field = DeviceComboBox(parent=parent, client=client)
self._signal_field = SignalComboBox(parent=parent, client=client) self._signal_field = SignalComboBox(parent=parent, client=client)
layout.addWidget(self._device_field) layout.addWidget(self._device_field)
layout.addWidget(self._signal_field) layout.addWidget(self._signal_field)
@@ -73,10 +71,13 @@ class ChoiceDialog(QDialog):
self._signal_field.include_config_signals = show_config self._signal_field.include_config_signals = show_config
self.setLayout(layout) self.setLayout(layout)
self._device_field.textChanged.connect(self._update_device) self._device_field.currentTextChanged.connect(self._update_device)
if device: if device:
self._device_field.set_device(device) self._device_field.set_device(device)
if signal and signal in set(s[0] for s in self._signal_field.signals): available_signals = {
entry[0] if isinstance(entry, tuple) else entry for entry in self._signal_field.signals
}
if signal and signal in available_signals:
self._signal_field.set_signal(signal) self._signal_field.set_signal(signal)
def _display_error(self): def _display_error(self):
@@ -97,19 +98,19 @@ class ChoiceDialog(QDialog):
self._device_field.set_device(device) self._device_field.set_device(device)
self._signal_field.set_device(device) self._signal_field.set_device(device)
self._device_field.setStyleSheet( self._device_field.setStyleSheet(
f"QLineEdit {{ border-style: solid; border-width: 2px; border-color: {self._accent_colors.success.name() if self._accent_colors else 'green'}}}" f"QComboBox {{ border-style: solid; border-width: 2px; border-color: {self._accent_colors.success.name() if self._accent_colors else 'green'}}}"
) )
self.button_box.button(QDialogButtonBox.Ok).setEnabled(True) self.button_box.button(QDialogButtonBox.Ok).setEnabled(True)
else: else:
self._device_field.setStyleSheet( self._device_field.setStyleSheet(
f"QLineEdit {{ border-style: solid; border-width: 2px; border-color: {self._accent_colors.emergency.name() if self._accent_colors else 'red'}}}" f"QComboBox {{ border-style: solid; border-width: 2px; border-color: {self._accent_colors.emergency.name() if self._accent_colors else 'red'}}}"
) )
self.button_box.button(QDialogButtonBox.Ok).setEnabled(False) self.button_box.button(QDialogButtonBox.Ok).setEnabled(False)
self._signal_field.clear() self._signal_field.clear()
def accept(self): def accept(self):
self.accepted_output.emit( self.accepted_output.emit(
self._device_field.text(), self._signal_field.selected_signal_comp_name self._device_field.currentText(), self._signal_field.selected_signal_comp_name
) )
self.cleanup() self.cleanup()
return super().accept() return super().accept()
@@ -170,7 +171,7 @@ class SignalLabel(BECWidget, QWidget):
client (BECClient, optional): The BEC client. Defaults to None. client (BECClient, optional): The BEC client. Defaults to None.
device (str, optional): The device name. Defaults to None. device (str, optional): The device name. Defaults to None.
signal (str, optional): The signal name. Defaults to None. signal (str, optional): The signal name. Defaults to None.
selection_dialog_config (DeviceSignalInputBaseConfig | dict, optional): Configuration for the signal selection dialog. selection_dialog_config: Configuration for the signal selection dialog.
show_select_button (bool, optional): Whether to show the select button. Defaults to True. show_select_button (bool, optional): Whether to show the select button. Defaults to True.
show_default_units (bool, optional): Whether to show default units. Defaults to False. show_default_units (bool, optional): Whether to show default units. Defaults to False.
custom_label (str, optional): Custom label for the widget. Defaults to "". custom_label (str, optional): Custom label for the widget. Defaults to "".
+54 -114
View File
@@ -1,152 +1,92 @@
from unittest import mock from unittest import mock
import pytest
from bec_lib.device import ReadoutPriority from bec_lib.device import ReadoutPriority
from qtpy.QtWidgets import QWidget
from bec_widgets.widgets.control.device_input.base_classes.device_input_base import ( from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import (
BECDeviceFilter, BECDeviceFilter,
DeviceInputBase, DeviceComboBox,
DeviceInputConfig, DeviceInputConfig,
) )
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox
from .client_mocks import mocked_client from .client_mocks import mocked_client
from .conftest import create_widget from .conftest import create_widget
# DeviceInputBase is meant to be mixed in a QWidget def test_device_combobox_init_with_config(qtbot, mocked_client):
class DeviceInputWidget(DeviceInputBase, QWidget):
"""Thin wrapper around DeviceInputBase to make it a QWidget"""
def __init__(self, parent=None, client=None, config=None, gui_id=None, **kwargs):
super().__init__(parent=parent, client=client, gui_id=gui_id, config=config, **kwargs)
@pytest.fixture
def device_input_base(qtbot, mocked_client):
"""Fixture with mocked FilterIO and WidgetIO"""
with mock.patch("bec_widgets.utils.filter_io.FilterIO.set_selection"):
with mock.patch("bec_widgets.utils.widget_io.WidgetIO.set_value"):
with mock.patch("bec_widgets.utils.widget_io.WidgetIO.get_value"):
widget = create_widget(qtbot=qtbot, widget=DeviceInputWidget, client=mocked_client)
yield widget
def test_device_input_base_init(device_input_base):
"""Test init"""
assert device_input_base is not None
assert device_input_base.client is not None
assert isinstance(device_input_base, DeviceInputBase)
assert device_input_base.config.widget_class == "DeviceInputWidget"
assert device_input_base.config.device_filter == []
assert device_input_base.config.default is None
assert device_input_base.devices == []
def test_device_input_base_init_with_config(qtbot, mocked_client):
"""Test init with Config"""
config = { config = {
"widget_class": "DeviceInputWidget", "widget_class": "DeviceComboBox",
"gui_id": "test_gui_id", "gui_id": "test_gui_id",
"device_filter": [BECDeviceFilter.POSITIONER], "device_filter": [BECDeviceFilter.POSITIONER.value],
"default": "samx", "default": "samx",
} }
widget = DeviceInputWidget(client=mocked_client, config=config) widget = create_widget(
widget2 = DeviceInputWidget( qtbot=qtbot,
client=mocked_client, config=DeviceInputConfig.model_validate(config) widget=DeviceComboBox,
client=mocked_client,
config=DeviceInputConfig.model_validate(config),
) )
qtbot.addWidget(widget)
qtbot.addWidget(widget2) assert widget.config.gui_id == "test_gui_id"
qtbot.waitExposed(widget) assert widget.config.device_filter == ["Positioner"]
qtbot.waitExposed(widget2) assert widget.config.default == "samx"
for w in [widget, widget2]:
assert w.config.gui_id == "test_gui_id"
assert w.config.device_filter == ["Positioner"]
assert w.config.default == "samx"
def test_device_input_base_set_device_filter(device_input_base): def test_device_combobox_set_device_filter(qtbot, mocked_client):
"""Test device filter setter.""" widget = create_widget(qtbot=qtbot, widget=DeviceComboBox, client=mocked_client)
device_input_base.set_device_filter(BECDeviceFilter.POSITIONER)
assert device_input_base.config.device_filter == ["Positioner"] widget.set_device_filter(BECDeviceFilter.POSITIONER)
assert widget.config.device_filter == ["Positioner"]
def test_device_input_base_set_device_filter_error(device_input_base): def test_device_combobox_set_device_filter_error(qtbot, mocked_client):
"""Test set_device_filter with Noneexisting class. This should not raise. It writes a log message entry.""" widget = create_widget(qtbot=qtbot, widget=DeviceComboBox, client=mocked_client)
device_input_base.set_device_filter("NonExistingClass")
assert device_input_base.device_filter == [] widget.set_device_filter("NonExistingClass")
assert widget.device_filter == []
def test_device_input_base_set_default_device(device_input_base): def test_device_combobox_set_default_device(qtbot, mocked_client):
"""Test setting the default device. Also tests the update_devices method.""" widget = create_widget(qtbot=qtbot, widget=DeviceComboBox, client=mocked_client)
device_input_base.set_device("samx")
assert device_input_base.config.default == None widget.set_device("samx")
device_input_base.set_device_filter(BECDeviceFilter.POSITIONER)
device_input_base.set_readout_priority_filter(ReadoutPriority.MONITORED) assert widget.config.default == "samx"
device_input_base.set_device("samx")
assert device_input_base.config.default == "samx"
def test_device_input_base_get_filters(device_input_base): def test_device_combobox_get_filters(qtbot, mocked_client):
"""Test getting the available filters.""" widget = create_widget(qtbot=qtbot, widget=DeviceComboBox, client=mocked_client)
filters = device_input_base.get_available_filters()
selection = [ assert BECDeviceFilter.POSITIONER in widget.get_available_filters()
BECDeviceFilter.POSITIONER, assert ReadoutPriority.MONITORED in widget.get_readout_priority_filters()
BECDeviceFilter.DEVICE,
BECDeviceFilter.COMPUTED_SIGNAL,
BECDeviceFilter.SIGNAL,
] + [
ReadoutPriority.MONITORED,
ReadoutPriority.BASELINE,
ReadoutPriority.ASYNC,
ReadoutPriority.ON_REQUEST,
]
assert [entry for entry in filters if entry in selection]
def test_device_input_base_properties(device_input_base): def test_device_combobox_properties(qtbot, mocked_client):
"""Test setting the properties of the device input base.""" widget = create_widget(qtbot=qtbot, widget=DeviceComboBox, client=mocked_client)
assert device_input_base.device_filter == []
device_input_base.filter_to_device = True widget.filter_to_device = True
assert device_input_base.device_filter == [BECDeviceFilter.DEVICE] widget.filter_to_positioner = True
device_input_base.filter_to_positioner = True widget.filter_to_computed_signal = True
assert device_input_base.device_filter == [BECDeviceFilter.DEVICE, BECDeviceFilter.POSITIONER] widget.filter_to_signal = True
device_input_base.filter_to_computed_signal = True assert widget.device_filter == [
assert device_input_base.device_filter == [
BECDeviceFilter.DEVICE,
BECDeviceFilter.POSITIONER,
BECDeviceFilter.COMPUTED_SIGNAL,
]
device_input_base.filter_to_signal = True
assert device_input_base.device_filter == [
BECDeviceFilter.DEVICE, BECDeviceFilter.DEVICE,
BECDeviceFilter.POSITIONER, BECDeviceFilter.POSITIONER,
BECDeviceFilter.COMPUTED_SIGNAL, BECDeviceFilter.COMPUTED_SIGNAL,
BECDeviceFilter.SIGNAL, BECDeviceFilter.SIGNAL,
] ]
assert device_input_base.readout_filter == []
device_input_base.readout_async = True widget.readout_async = True
assert device_input_base.readout_filter == [ReadoutPriority.ASYNC] widget.readout_baseline = True
device_input_base.readout_baseline = True widget.readout_monitored = True
assert device_input_base.readout_filter == [ReadoutPriority.ASYNC, ReadoutPriority.BASELINE] widget.readout_on_request = True
device_input_base.readout_monitored = True assert ReadoutPriority.ASYNC in widget.readout_filter
assert device_input_base.readout_filter == [ assert ReadoutPriority.BASELINE in widget.readout_filter
ReadoutPriority.ASYNC, assert ReadoutPriority.MONITORED in widget.readout_filter
ReadoutPriority.BASELINE, assert ReadoutPriority.ON_REQUEST in widget.readout_filter
ReadoutPriority.MONITORED,
]
device_input_base.readout_on_request = True
assert device_input_base.readout_filter == [
ReadoutPriority.ASYNC,
ReadoutPriority.BASELINE,
ReadoutPriority.MONITORED,
ReadoutPriority.ON_REQUEST,
]
def test_device_combobox_signal_class_filter(qtbot, mocked_client): def test_device_combobox_signal_class_filter(qtbot, mocked_client):
"""Test device filtering via signal_class_filter on combobox."""
mocked_client.device_manager.get_bec_signals = mock.MagicMock( mocked_client.device_manager.get_bec_signals = mock.MagicMock(
return_value=[ return_value=[
("samx", "async_signal", {"signal_class": "AsyncSignal"}), ("samx", "async_signal", {"signal_class": "AsyncSignal"}),
+34 -76
View File
@@ -1,10 +1,9 @@
import pytest import pytest
from bec_lib.device import ReadoutPriority from bec_lib.device import ReadoutPriority
from bec_widgets.widgets.control.device_input.base_classes.device_input_base import BECDeviceFilter from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import (
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox BECDeviceFilter,
from bec_widgets.widgets.control.device_input.device_line_edit.device_line_edit import ( DeviceComboBox,
DeviceLineEdit,
) )
from .client_mocks import mocked_client from .client_mocks import mocked_client
@@ -37,6 +36,21 @@ def test_device_input_combobox_init(device_input_combobox):
assert device_input_combobox.client is not None assert device_input_combobox.client is not None
assert isinstance(device_input_combobox, DeviceComboBox) assert isinstance(device_input_combobox, DeviceComboBox)
assert device_input_combobox.config.widget_class == "DeviceComboBox" assert device_input_combobox.config.widget_class == "DeviceComboBox"
assert device_input_combobox.isEditable() is True
assert device_input_combobox.currentText() == ""
assert device_input_combobox.is_valid_input is False
assert device_input_combobox.config.device_filter == []
assert device_input_combobox.config.readout_filter == [
ReadoutPriority.MONITORED.value,
ReadoutPriority.BASELINE.value,
ReadoutPriority.ASYNC.value,
ReadoutPriority.CONTINUOUS.value,
ReadoutPriority.ON_REQUEST.value,
]
assert device_input_combobox.config.default is None
assert device_input_combobox.autocomplete is False
assert device_input_combobox.completer() is not None
assert device_input_combobox.completer().model() == device_input_combobox.model()
assert device_input_combobox.devices == [ assert device_input_combobox.devices == [
"samx", "samx",
"samy", "samy",
@@ -65,81 +79,25 @@ def test_device_input_combobox_init_with_kwargs(device_input_combobox_with_kwarg
assert device_input_combobox_with_kwargs.config.arg_name == "test_arg_name" assert device_input_combobox_with_kwargs.config.arg_name == "test_arg_name"
def test_device_input_combobox_autocomplete(qtbot, mocked_client):
widget = DeviceComboBox(client=mocked_client, autocomplete=True)
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
assert widget.autocomplete is True
assert widget.completer() is not None
assert widget.completer().model().stringList() == widget.devices
assert widget.completer().model() != widget.model()
widget.autocomplete = False
assert widget.completer() is not None
assert widget.completer().model() == widget.model()
def test_get_device_from_input_combobox_init(device_input_combobox): def test_get_device_from_input_combobox_init(device_input_combobox):
device_input_combobox.setCurrentIndex(0) device_input_combobox.setCurrentIndex(0)
device_text = device_input_combobox.currentText() device_text = device_input_combobox.currentText()
current_device = device_input_combobox.get_current_device() current_device = device_input_combobox.get_current_device()
assert current_device.name == device_text assert current_device.name == device_text
@pytest.fixture
def device_input_line_edit(qtbot, mocked_client):
widget = DeviceLineEdit(client=mocked_client)
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
yield widget
@pytest.fixture
def device_input_line_edit_with_kwargs(qtbot, mocked_client):
widget = DeviceLineEdit(
client=mocked_client,
gui_id="test_gui_id",
device_filter=[BECDeviceFilter.POSITIONER],
default="samx",
arg_name="test_arg_name",
)
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
yield widget
def test_device_input_line_edit_init(device_input_line_edit):
assert device_input_line_edit is not None
assert device_input_line_edit.client is not None
assert isinstance(device_input_line_edit, DeviceLineEdit)
assert device_input_line_edit.config.widget_class == "DeviceLineEdit"
assert device_input_line_edit.config.device_filter == []
assert device_input_line_edit.config.readout_filter == [
ReadoutPriority.MONITORED,
ReadoutPriority.BASELINE,
ReadoutPriority.ASYNC,
ReadoutPriority.CONTINUOUS,
ReadoutPriority.ON_REQUEST,
]
assert device_input_line_edit.config.default is None
assert device_input_line_edit.devices == [
"samx",
"samy",
"samz",
"aptrx",
"aptry",
"gauss_bpm",
"gauss_adc1",
"gauss_adc2",
"gauss_adc3",
"bpm4i",
"bpm3a",
"bpm3i",
"eiger",
"waveform1d",
"async_device",
"test",
"test_device",
]
def test_device_input_line_edit_init_with_kwargs(device_input_line_edit_with_kwargs):
assert device_input_line_edit_with_kwargs.config.gui_id == "test_gui_id"
assert device_input_line_edit_with_kwargs.config.device_filter == ["Positioner"]
assert device_input_line_edit_with_kwargs.config.default == "samx"
assert device_input_line_edit_with_kwargs.config.arg_name == "test_arg_name"
def test_get_device_from_input_line_edit_init(device_input_line_edit):
device_input_line_edit.setText("samx")
device_text = device_input_line_edit.text()
current_device = device_input_line_edit.get_current_device()
assert current_device.name == device_text
+46 -74
View File
@@ -2,54 +2,28 @@ from unittest import mock
import pytest import pytest
from bec_lib.device import Signal from bec_lib.device import Signal
from qtpy.QtWidgets import QWidget
from bec_widgets.utils.ophyd_kind_util import Kind from bec_widgets.utils.ophyd_kind_util import Kind
from bec_widgets.widgets.control.device_input.base_classes.device_input_base import BECDeviceFilter from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import (
from bec_widgets.widgets.control.device_input.base_classes.device_signal_input_base import ( BECDeviceFilter,
DeviceSignalInputBase, DeviceComboBox,
) )
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox
from bec_widgets.widgets.control.device_input.signal_combobox.signal_combobox import SignalComboBox from bec_widgets.widgets.control.device_input.signal_combobox.signal_combobox import SignalComboBox
from bec_widgets.widgets.control.device_input.signal_line_edit.signal_line_edit import (
SignalLineEdit,
)
from .client_mocks import mocked_client from .client_mocks import mocked_client
from .conftest import create_widget from .conftest import create_widget
class FakeSignal(Signal): class FakeSignal(Signal):
"""Fake signal to test the DeviceSignalInputBase.""" """Fake signal used by SignalComboBox tests."""
class DeviceInputWidget(DeviceSignalInputBase, QWidget):
"""Thin wrapper around DeviceInputBase to make it a QWidget"""
@pytest.fixture
def device_signal_base(qtbot, mocked_client):
"""Fixture with mocked FilterIO and WidgetIO"""
with mock.patch("bec_widgets.utils.filter_io.FilterIO.set_selection"):
with mock.patch("bec_widgets.utils.widget_io.WidgetIO.set_value"):
widget = create_widget(qtbot=qtbot, widget=DeviceInputWidget, client=mocked_client)
yield widget
@pytest.fixture @pytest.fixture
def device_signal_combobox(qtbot, mocked_client): def device_signal_combobox(qtbot, mocked_client):
"""Fixture with mocked FilterIO and WidgetIO"""
widget = create_widget(qtbot=qtbot, widget=SignalComboBox, client=mocked_client) widget = create_widget(qtbot=qtbot, widget=SignalComboBox, client=mocked_client)
yield widget yield widget
@pytest.fixture
def device_signal_line_edit(qtbot, mocked_client):
"""Fixture with mocked FilterIO and WidgetIO"""
widget = create_widget(qtbot=qtbot, widget=SignalLineEdit, client=mocked_client)
yield widget
@pytest.fixture @pytest.fixture
def test_device_signal_combo(qtbot, mocked_client): def test_device_signal_combo(qtbot, mocked_client):
"""Fixture to create a SignalComboBox widget and a DeviceInputWidget widget""" """Fixture to create a SignalComboBox widget and a DeviceInputWidget widget"""
@@ -63,34 +37,49 @@ def test_device_signal_combo(qtbot, mocked_client):
yield input, signal yield input, signal
def test_device_signal_base_init(device_signal_base): def test_signal_combobox_init(device_signal_combobox):
"""Test if the DeviceSignalInputBase is initialized correctly""" assert device_signal_combobox._device is None
assert device_signal_base._device is None assert device_signal_combobox._signal_filter == {Kind.config, Kind.normal, Kind.hinted}
assert device_signal_base._signal_filter == set() assert device_signal_combobox._signals == []
assert device_signal_base._signals == [] assert device_signal_combobox._hinted_signals == []
assert device_signal_base._hinted_signals == [] assert device_signal_combobox._normal_signals == []
assert device_signal_base._normal_signals == [] assert device_signal_combobox._config_signals == []
assert device_signal_base._config_signals == [] assert device_signal_combobox.autocomplete is False
assert device_signal_combobox.completer() is not None
assert device_signal_combobox.completer().model() == device_signal_combobox.model()
def test_device_signal_qproperties(device_signal_base): def test_signal_combobox_autocomplete(qtbot, mocked_client):
"""Test if the DeviceSignalInputBase has the correct QProperties""" widget = create_widget(
assert device_signal_base._signal_filter == set() qtbot=qtbot, widget=SignalComboBox, client=mocked_client, autocomplete=True
device_signal_base.include_config_signals = False )
device_signal_base.include_normal_signals = False
assert device_signal_base._signal_filter == set() widget.set_device("samx")
device_signal_base.include_config_signals = True
assert device_signal_base._signal_filter == {Kind.config} assert widget.autocomplete is True
device_signal_base.include_normal_signals = True assert widget.completer() is not None
assert device_signal_base._signal_filter == {Kind.config, Kind.normal} assert widget.completer().model().stringList() == ["samx (readback)", "setpoint", "velocity"]
device_signal_base.include_hinted_signals = True assert widget.completer().model() != widget.model()
assert device_signal_base._signal_filter == {Kind.config, Kind.normal, Kind.hinted}
device_signal_base.include_hinted_signals = True widget.autocomplete = False
assert device_signal_base._signal_filter == {Kind.config, Kind.normal, Kind.hinted}
device_signal_base.include_hinted_signals = True assert widget.completer() is not None
assert device_signal_base._signal_filter == {Kind.config, Kind.normal, Kind.hinted} assert widget.completer().model() == widget.model()
device_signal_base.include_hinted_signals = False
assert device_signal_base._signal_filter == {Kind.config, Kind.normal}
def test_signal_combobox_qproperties(device_signal_combobox):
device_signal_combobox.include_config_signals = False
device_signal_combobox.include_normal_signals = False
device_signal_combobox.include_hinted_signals = False
assert device_signal_combobox._signal_filter == set()
device_signal_combobox.include_config_signals = True
assert device_signal_combobox._signal_filter == {Kind.config}
device_signal_combobox.include_normal_signals = True
assert device_signal_combobox._signal_filter == {Kind.config, Kind.normal}
device_signal_combobox.include_hinted_signals = True
assert device_signal_combobox._signal_filter == {Kind.config, Kind.normal, Kind.hinted}
device_signal_combobox.include_hinted_signals = False
assert device_signal_combobox._signal_filter == {Kind.config, Kind.normal}
def test_signal_combobox(qtbot, device_signal_combobox): def test_signal_combobox(qtbot, device_signal_combobox):
@@ -128,26 +117,9 @@ def test_signal_combobox(qtbot, device_signal_combobox):
assert device_signal_combobox._hinted_signals == [("fake_signal", {})] assert device_signal_combobox._hinted_signals == [("fake_signal", {})]
def test_signal_lineedit(device_signal_line_edit):
"""Test the signal_combobox"""
assert device_signal_line_edit._signals == []
device_signal_line_edit.include_normal_signals = True
device_signal_line_edit.include_hinted_signals = True
device_signal_line_edit.include_config_signals = True
assert device_signal_line_edit.signals == []
device_signal_line_edit.set_device("samx")
assert device_signal_line_edit.signals == ["readback", "setpoint", "velocity"]
device_signal_line_edit.set_signal("readback")
assert device_signal_line_edit.text() == "readback"
assert device_signal_line_edit._is_valid_input is True
device_signal_line_edit.setText("invalid")
assert device_signal_line_edit._is_valid_input is False
def test_device_signal_input_base_cleanup(qtbot, mocked_client): def test_device_signal_input_base_cleanup(qtbot, mocked_client):
with mock.patch.object(mocked_client.callbacks, "remove"): with mock.patch.object(mocked_client.callbacks, "remove"):
widget = DeviceInputWidget(client=mocked_client) widget = SignalComboBox(client=mocked_client)
widget.close() widget.close()
widget.deleteLater() widget.deleteLater()
+28 -54
View File
@@ -1,8 +1,9 @@
import pytest from qtpy.QtWidgets import QComboBox
from bec_widgets.utils.filter_io import FilterIO from bec_widgets.utils.filter_io import (
from bec_widgets.widgets.control.device_input.device_line_edit.device_line_edit import ( combobox_contains_text,
DeviceLineEdit, get_bec_signals_for_classes,
replace_combobox_items,
) )
from bec_widgets.widgets.dap.dap_combo_box.dap_combo_box import DapComboBox from bec_widgets.widgets.dap.dap_combo_box.dap_combo_box import DapComboBox
@@ -10,65 +11,38 @@ from .client_mocks import mocked_client
from .conftest import create_widget from .conftest import create_widget
@pytest.fixture(scope="function") def test_replace_combobox_items(qtbot, mocked_client):
def dap_mock(qtbot, mocked_client):
"""Fixture for QLineEdit widget"""
models = ["GaussianModel", "LorentzModel", "SineModel"]
mocked_client.dap._available_dap_plugins.keys.return_value = models
widget = create_widget(qtbot, DapComboBox, client=mocked_client) widget = create_widget(qtbot, DapComboBox, client=mocked_client)
return widget
replace_combobox_items(widget, ["testA", ("testB", {"payload": True})])
assert widget.count() == 2
assert widget.itemText(0) == "testA"
assert widget.itemText(1) == "testB"
assert widget.itemData(1) == {"payload": True}
assert combobox_contains_text(widget, "testA") is True
assert combobox_contains_text(widget, "missing") is False
@pytest.fixture(scope="function") def test_get_bec_signals_for_classes_ndim_filter(mocked_client):
def line_edit_mock(qtbot, mocked_client):
"""Fixture for QLineEdit widget"""
widget = create_widget(qtbot, DeviceLineEdit, client=mocked_client)
return widget
def test_set_selection_combo_box(dap_mock):
"""Test set selection for QComboBox using DapComboBox"""
assert dap_mock.fit_model_combobox.count() == 3
FilterIO.set_selection(dap_mock.fit_model_combobox, selection=["testA", "testB"])
assert dap_mock.fit_model_combobox.count() == 2
assert FilterIO.check_input(widget=dap_mock.fit_model_combobox, text="testA") is True
def test_set_selection_line_edit(line_edit_mock):
"""Test set selection for QComboBox using DapComboBox"""
FilterIO.set_selection(line_edit_mock, selection=["testA", "testB"])
assert line_edit_mock.completer.model().rowCount() == 2
model = line_edit_mock.completer.model()
model_data = [model.data(model.index(i)) for i in range(model.rowCount())]
assert model_data == ["testA", "testB"]
assert FilterIO.check_input(widget=line_edit_mock, text="testA") is True
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 = [ signals = [
("dev1", "sig1", {"describe": {"signal_info": {"ndim": 1}}}), ("dev1", "sig1", {"describe": {"signal_info": {"ndim": 1}}}),
("dev1", "sig2", {"describe": {"signal_info": {"ndim": 2}}}), ("dev1", "sig2", {"describe": {"signal_info": {"ndim": 2}}}),
] ]
mocked_client.device_manager.get_bec_signals = lambda _filters: signals mocked_client.device_manager.get_bec_signals = lambda _filters: signals
out = FilterIO.update_with_signal_class(
widget=dap_mock.fit_model_combobox, out = get_bec_signals_for_classes(
signal_class_filter=["AsyncSignal"], client=mocked_client, signal_class_filter=["AsyncSignal"], ndim_filter=1
client=mocked_client,
ndim_filter=1,
) )
assert out == [("dev1", "sig1", {"describe": {"signal_info": {"ndim": 1}}})] assert out == [("dev1", "sig1", {"describe": {"signal_info": {"ndim": 1}}})]
def test_update_with_signal_class_line_edit_passthrough(line_edit_mock, mocked_client): def test_replace_combobox_items_empty(qtbot):
signals = [("dev1", "sig1", {"describe": {"signal_info": {"ndim": 1}}})] widget = QComboBox()
mocked_client.device_manager.get_bec_signals = lambda _filters: signals qtbot.addWidget(widget)
out = FilterIO.update_with_signal_class( widget.addItem("old")
widget=line_edit_mock,
signal_class_filter=["AsyncSignal"], replace_combobox_items(widget, [])
client=mocked_client,
ndim_filter=1, assert widget.count() == 0
)
assert out == signals
+3 -5
View File
@@ -12,9 +12,7 @@ from bec_widgets.widgets.control.device_control.positioner_box import (
PositionerBox, PositionerBox,
PositionerControlLine, PositionerControlLine,
) )
from bec_widgets.widgets.control.device_input.device_line_edit.device_line_edit import ( from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox
DeviceLineEdit,
)
from .client_mocks import mocked_client from .client_mocks import mocked_client
from .conftest import create_widget from .conftest import create_widget
@@ -164,8 +162,8 @@ def test_positioner_box_open_dialog_selection(qtbot, positioner_box):
# pylint: disable=protected-access # pylint: disable=protected-access
assert positioner_box._dialog is not None assert positioner_box._dialog is not None
qtbot.waitUntil(lambda: positioner_box._dialog.isVisible() is True, timeout=1000) qtbot.waitUntil(lambda: positioner_box._dialog.isVisible() is True, timeout=1000)
line_edit = positioner_box._dialog.findChild(DeviceLineEdit) line_edit = positioner_box._dialog.findChild(DeviceComboBox)
line_edit.setText("samy") line_edit.setCurrentText("samy")
close_button = positioner_box._dialog.findChild(QPushButton) close_button = positioner_box._dialog.findChild(QPushButton)
assert close_button.text() == "Close" assert close_button.text() == "Close"
qtbot.mouseClick(close_button, Qt.LeftButton) qtbot.mouseClick(close_button, Qt.LeftButton)
+8
View File
@@ -9,6 +9,7 @@ from qtpy.QtCore import QModelIndex, Qt
from bec_widgets.utils.forms_from_types.items import StrFormItem from bec_widgets.utils.forms_from_types.items import StrFormItem
from bec_widgets.utils.widget_io import WidgetIO from bec_widgets.utils.widget_io import WidgetIO
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox
from bec_widgets.widgets.control.scan_control import ScanControl from bec_widgets.widgets.control.scan_control import ScanControl
from .client_mocks import mocked_client from .client_mocks import mocked_client
@@ -304,6 +305,13 @@ def test_on_scan_selected(scan_control, scan_name):
assert widget is not None # Confirm that a widget exists assert widget is not None # Confirm that a widget exists
expected_widget_type = scan_control.arg_box.WIDGET_HANDLER.get(arg_value, None) expected_widget_type = scan_control.arg_box.WIDGET_HANDLER.get(arg_value, None)
assert isinstance(widget, expected_widget_type) # Confirm the widget type matches assert isinstance(widget, expected_widget_type) # Confirm the widget type matches
if isinstance(widget, DeviceComboBox):
assert widget.currentText() == ""
assert widget.autocomplete is True
assert "samx" in widget.devices
assert (
"async_device" in widget.devices
) # async device should also be present in the device list
# Check kwargs boxes # Check kwargs boxes
kwargs_group = [param for param in expected_scan_info["gui_config"]["kwarg_groups"]] kwargs_group = [param for param in expected_scan_info["gui_config"]["kwarg_groups"]]
+10 -7
View File
@@ -6,8 +6,8 @@ import pytest
from qtpy import QtCore from qtpy import QtCore
from qtpy.QtWidgets import QDialogButtonBox, QLabel from qtpy.QtWidgets import QDialogButtonBox, QLabel
from bec_widgets.widgets.control.device_input.base_classes.device_signal_input_base import ( from bec_widgets.widgets.control.device_input.signal_combobox.signal_combobox import (
DeviceSignalInputBaseConfig, SignalComboBoxConfig,
) )
from bec_widgets.widgets.utility.signal_label.signal_label import ChoiceDialog, SignalLabel from bec_widgets.widgets.utility.signal_label.signal_label import ChoiceDialog, SignalLabel
@@ -61,7 +61,7 @@ SAMX_INFO_DICT = {
@pytest.fixture @pytest.fixture
def signal_label(qtbot, mocked_client: MagicMock): def signal_label(qtbot, mocked_client: MagicMock):
with patch.object(mocked_client.device_manager.devices.samx, "_info", SAMX_INFO_DICT): with patch.object(mocked_client.device_manager.devices.samx, "_info", SAMX_INFO_DICT):
config = DeviceSignalInputBaseConfig(device="samx", default="samx") config = SignalComboBoxConfig(device="samx", default="samx")
widget = SignalLabel( widget = SignalLabel(
config=config, custom_label="Test Label", custom_units="m/s", client=mocked_client config=config, custom_label="Test Label", custom_units="m/s", client=mocked_client
) )
@@ -149,7 +149,8 @@ def test_choose_signal_dialog_sends_choices(signal_label: SignalLabel, qtbot):
dialog = signal_label.show_choice_dialog() dialog = signal_label.show_choice_dialog()
qtbot.waitUntil(dialog.button_box.button(QDialogButtonBox.Ok).isVisible, timeout=500) qtbot.waitUntil(dialog.button_box.button(QDialogButtonBox.Ok).isVisible, timeout=500)
dialog._device_field.dev["test device"] = MagicMock() dialog._device_field.dev["test device"] = MagicMock()
dialog._device_field.setText("test device") dialog._device_field.devices = ["test device"]
dialog._device_field.setCurrentText("test device")
dialog._signal_field._signals = [("test signal", {"component_name": "test signal"})] dialog._signal_field._signals = [("test signal", {"component_name": "test signal"})]
dialog._signal_field.addItem("test signal") dialog._signal_field.addItem("test signal")
dialog._signal_field.setCurrentIndex(0) dialog._signal_field.setCurrentIndex(0)
@@ -162,7 +163,8 @@ def test_dialog_handler_updates_devices(signal_label: SignalLabel, qtbot):
dialog = signal_label.show_choice_dialog() dialog = signal_label.show_choice_dialog()
qtbot.waitUntil(dialog.button_box.button(QDialogButtonBox.Ok).isVisible, timeout=500) qtbot.waitUntil(dialog.button_box.button(QDialogButtonBox.Ok).isVisible, timeout=500)
dialog._device_field.dev["flux_capacitor"] = MagicMock() dialog._device_field.dev["flux_capacitor"] = MagicMock()
dialog._device_field.setText("flux_capacitor") dialog._device_field.devices = ["flux_capacitor"]
dialog._device_field.setCurrentText("flux_capacitor")
dialog._signal_field._signals = [("spin_speed", {"component_name": "spin_speed"})] dialog._signal_field._signals = [("spin_speed", {"component_name": "spin_speed"})]
dialog._signal_field.addItem("spin_speed") dialog._signal_field.addItem("spin_speed")
dialog._signal_field.setCurrentIndex(0) dialog._signal_field.setCurrentIndex(0)
@@ -176,7 +178,7 @@ def test_choose_signal_dialog_invalid_device(signal_label: SignalLabel, qtbot):
signal_label._process_dialog = MagicMock() signal_label._process_dialog = MagicMock()
dialog = signal_label.show_choice_dialog() dialog = signal_label.show_choice_dialog()
qtbot.waitUntil(dialog.button_box.button(QDialogButtonBox.Ok).isVisible, timeout=500) qtbot.waitUntil(dialog.button_box.button(QDialogButtonBox.Ok).isVisible, timeout=500)
dialog._device_field.setText("invalid device") dialog._device_field.setCurrentText("invalid device")
dialog._signal_field.addItem("test signal") dialog._signal_field.addItem("test signal")
dialog._signal_field.setCurrentIndex(0) dialog._signal_field.setCurrentIndex(0)
qtbot.mouseClick(dialog.button_box.button(QDialogButtonBox.Ok), QtCore.Qt.LeftButton) qtbot.mouseClick(dialog.button_box.button(QDialogButtonBox.Ok), QtCore.Qt.LeftButton)
@@ -206,7 +208,8 @@ def test_dialog_has_signals(signal_label: SignalLabel, qtbot):
"signals": {"signal 1": {"kind_str": "hinted"}, "signal 2": {"kind_str": "normal"}} "signals": {"signal 1": {"kind_str": "hinted"}, "signal 2": {"kind_str": "normal"}}
} }
dialog._device_field.setText("test device") dialog._device_field.devices = ["test device"]
dialog._device_field.setCurrentText("test device")
assert dialog._signal_field.count() == 2 # the actual signal and the category label assert dialog._signal_field.count() == 2 # the actual signal and the category label
assert dialog._signal_field.currentText() == "signal 1" assert dialog._signal_field.currentText() == "signal 1"