Compare commits

...

38 Commits

Author SHA1 Message Date
copilot-swe-agent[bot] d44eb69a92 fix: prevent segfault and CDockManager teardown crashes with PySide6 6.11.0
- bec_widget.py: set _destroyed=True BEFORE calling cleanup() so the flag
  is available throughout the entire teardown phase
- bec_console.py: in _parking_parent, skip console.window() as parking
  target when the window is marked as _destroyed, preventing setParent()
  into a widget mid-destruction (segfault in PySide6 6.11.0)
- basic_dock_area.py: add DockAreaWidget.cleanup() that calls delete_all()
  first to release CDockWidgets via the Qt ADS API before BECWidget.cleanup()
  runs, fixing teardown crashes with PySide6-QtAds 4.5.x
2026-05-30 13:40:20 +00:00
copilot-swe-agent[bot] f4d51b6287 feat: update PySide6 to 6.11.0 and PySide6-QtAds to 4.5.0.4 2026-05-29 12:19:12 +00:00
semantic-release c9fc0a82b9 3.13.3
Automatically generated by python-semantic-release
2026-05-22 12:30:17 +00:00
wakonig_k 668b1bd9cd fix(tests): rename description attribute to _description in FakeDevice 2026-05-22 14:29:28 +02:00
semantic-release 1a6c8bf30f 3.13.2
Automatically generated by python-semantic-release
2026-05-22 08:57:04 +00:00
wakonig_k c346bd0f18 fix(tests): rename description attribute to _description in FakePositioner 2026-05-22 10:56:05 +02:00
semantic-release 5f86e41a03 3.13.1
Automatically generated by python-semantic-release
2026-05-21 14:41:40 +00:00
wakonig_k f7a48b5f6a fix(gui): replace window.show() with window.raise_window() and add hide() method 2026-05-21 16:40:51 +02:00
wakonig_k b4beb274da fix: use .show instead of .start 2026-05-21 16:40:51 +02:00
semantic-release 80694d151f 3.13.0
Automatically generated by python-semantic-release
2026-05-21 14:20:49 +00:00
wakonig_k f03a5d9e85 feat(rpc-base): set default RPC timeout and allow customization 2026-05-21 16:19:48 +02:00
semantic-release 5e8f0e8083 3.12.2
Automatically generated by python-semantic-release
2026-05-21 13:29:11 +00:00
wyzula_j 9eb05416ab fix(toggle): disable styling implemented 2026-05-21 15:28:17 +02:00
semantic-release ab6a1aecc1 3.12.1
Automatically generated by python-semantic-release
2026-05-21 11:40:06 +00:00
wyzula_j d99db7d042 fix(device_input): ensure callback is removed after cleanup 2026-05-21 13:39:19 +02:00
wyzula_j a976837cff fix(signal_combobox): signature matched for update_signals_from_filters 2026-05-21 13:39:19 +02:00
wyzula_j 56427a7f0c fix(device_input): correct cleanup unsubscribe 2026-05-21 13:39:19 +02:00
semantic-release c4d4b78846 3.12.0
Automatically generated by python-semantic-release
2026-05-20 09:04:47 +00:00
wakonig_k 2dc0227d38 fix(scan-control): filter out private scans from allowed scans 2026-05-20 11:04:00 +02:00
wyzula_j 2d8e1eed4d fix(scan-control): hide hidden scan arguments 2026-05-20 11:04:00 +02:00
wyzula_j 3b579e740f fix(scan-control): reject unsupported scan input types 2026-05-20 11:04:00 +02:00
wyzula_j b8740c9594 fix(scan-control): skip duplicate visible scan kwargs 2026-05-20 11:04:00 +02:00
wakonig_k d5bf10e216 feat: add support for new scan signatures including units 2026-05-20 11:04:00 +02:00
semantic-release 3a165b26ed 3.11.1
Automatically generated by python-semantic-release
2026-05-18 19:59:50 +00:00
wakonig_k faa200bf5c fix(scan progressbar): fix device subscription cleanup 2026-05-18 21:59:02 +02:00
semantic-release b0fc0d325e 3.11.0
Automatically generated by python-semantic-release
2026-05-15 11:14:18 +00:00
wyzula_j daa1ba020c refactor(device_input): consolidation of device/signal combobox logic; docsrtings added 2026-05-15 13:13:31 +02:00
wyzula_j 3d934a8c38 feat(device_input): comboboxes can have line edit like autocomplete 2026-05-15 13:13:31 +02:00
wyzula_j c47b246a9f fix(scan_control): ScanGroupBox enforce correct device combobox type in correct order 2026-05-15 13:13:31 +02:00
wyzula_j bb6c0bb08f fix: remove device/signal line edit and abstraction layer for combobox/lineEdit 2026-05-15 13:13:31 +02:00
semantic-release ce456572d7 3.10.0
Automatically generated by python-semantic-release
2026-05-13 15:36:26 +00:00
wyzula_j e22ab7e4c1 feat: BL plugin menu in BECDockArea 2026-05-13 17:35:34 +02:00
wakonig_k 0cd000dfa1 docs: fix link to doc page 2026-05-13 09:18:42 +02:00
semantic-release 1057db9d76 3.9.1
Automatically generated by python-semantic-release
2026-05-12 17:46:15 +00:00
wyzula_j be35e249f9 wip further opt 2026-05-12 19:45:22 +02:00
wyzula_j cdd833dfc2 tests(scan_control): tests extended and optimized 2026-05-12 19:45:22 +02:00
wyzula_j 3c7834b492 fix: logpanel fixture overwriting xread 2026-05-12 19:45:22 +02:00
wyzula_j acd35a2786 fix(scan_control): restore scan parameters from history are fetched on demand with button 2026-05-12 19:45:22 +02:00
60 changed files with 3207 additions and 2482 deletions
+132
View File
@@ -1,6 +1,138 @@
# CHANGELOG
## v3.13.3 (2026-05-22)
### Bug Fixes
- **tests**: Rename description attribute to _description in FakeDevice
([`668b1bd`](https://github.com/bec-project/bec_widgets/commit/668b1bd9cd158fc12cff2c340d7317f30a212121))
## v3.13.2 (2026-05-22)
### Bug Fixes
- **tests**: Rename description attribute to _description in FakePositioner
([`c346bd0`](https://github.com/bec-project/bec_widgets/commit/c346bd0f18ce873ff5ca6c59150c9581c9edca8d))
## v3.13.1 (2026-05-21)
### Bug Fixes
- Use .show instead of .start
([`b4beb27`](https://github.com/bec-project/bec_widgets/commit/b4beb274da745da618f9b37ec241cd0109c088f1))
- **gui**: Replace window.show() with window.raise_window() and add hide() method
([`f7a48b5`](https://github.com/bec-project/bec_widgets/commit/f7a48b5f6a51d391dca26ca42d03bad4f278ff22))
## v3.13.0 (2026-05-21)
### Features
- **rpc-base**: Set default RPC timeout and allow customization
([`f03a5d9`](https://github.com/bec-project/bec_widgets/commit/f03a5d9e853bd62b8ec1bad1c1e112fe01befe70))
## v3.12.2 (2026-05-21)
### Bug Fixes
- **toggle**: Disable styling implemented
([`9eb0541`](https://github.com/bec-project/bec_widgets/commit/9eb05416ab68dcb88732dca8974c665030d34e0b))
## v3.12.1 (2026-05-21)
### Bug Fixes
- **device_input**: Correct cleanup unsubscribe
([`56427a7`](https://github.com/bec-project/bec_widgets/commit/56427a7f0c3a89fe847d415c8b45212e663434c4))
- **device_input**: Ensure callback is removed after cleanup
([`d99db7d`](https://github.com/bec-project/bec_widgets/commit/d99db7d04208945b86a39d65022b211ba093caed))
- **signal_combobox**: Signature matched for update_signals_from_filters
([`a976837`](https://github.com/bec-project/bec_widgets/commit/a976837cff612349f2a3f17900903c203bc3d250))
## v3.12.0 (2026-05-20)
### Bug Fixes
- **scan-control**: Filter out private scans from allowed scans
([`2dc0227`](https://github.com/bec-project/bec_widgets/commit/2dc0227d38f0e217e252a5e5751bafd60363a5a4))
- **scan-control**: Hide hidden scan arguments
([`2d8e1ee`](https://github.com/bec-project/bec_widgets/commit/2d8e1eed4d6503c42a38c8de910ddaa54132405d))
- **scan-control**: Reject unsupported scan input types
([`3b579e7`](https://github.com/bec-project/bec_widgets/commit/3b579e740f36c60c3635681a9b2c35b518498f58))
- **scan-control**: Skip duplicate visible scan kwargs
([`b8740c9`](https://github.com/bec-project/bec_widgets/commit/b8740c95941d36102f07a51d74a50e6f262a6646))
### Features
- Add support for new scan signatures including units
([`d5bf10e`](https://github.com/bec-project/bec_widgets/commit/d5bf10e21682ae8270078c7858a036bafbabf10e))
## v3.11.1 (2026-05-18)
### Bug Fixes
- **scan progressbar**: Fix device subscription cleanup
([`faa200b`](https://github.com/bec-project/bec_widgets/commit/faa200bf5c3cf0c5bebb9858700106899f583695))
## v3.11.0 (2026-05-15)
### Bug Fixes
- Remove device/signal line edit and abstraction layer for combobox/lineEdit
([`bb6c0bb`](https://github.com/bec-project/bec_widgets/commit/bb6c0bb08fc9802bec0d6b9994a76a5bcf2a3a81))
- **scan_control**: Scangroupbox enforce correct device combobox type in correct order
([`c47b246`](https://github.com/bec-project/bec_widgets/commit/c47b246a9fd5c9aff2512c2744b8ff19c87e6e03))
### Features
- **device_input**: Comboboxes can have line edit like autocomplete
([`3d934a8`](https://github.com/bec-project/bec_widgets/commit/3d934a8c3825b17319c3cb99750b96042e0bc230))
### Refactoring
- **device_input**: Consolidation of device/signal combobox logic; docsrtings added
([`daa1ba0`](https://github.com/bec-project/bec_widgets/commit/daa1ba020ce6d05800186d2467a496c1024e8aa5))
## v3.10.0 (2026-05-13)
### Documentation
- Fix link to doc page
([`0cd000d`](https://github.com/bec-project/bec_widgets/commit/0cd000dfa1bb6f1b4d286e5aab30299361f436f6))
### Features
- Bl plugin menu in BECDockArea
([`e22ab7e`](https://github.com/bec-project/bec_widgets/commit/e22ab7e4c10552e22aaaa9dbc30d098fbfa9c49c))
## v3.9.1 (2026-05-12)
### Bug Fixes
- Logpanel fixture overwriting xread
([`3c7834b`](https://github.com/bec-project/bec_widgets/commit/3c7834b492a5d2da13689f58b20caf38dda9ac1d))
- **scan_control**: Restore scan parameters from history are fetched on demand with button
([`acd35a2`](https://github.com/bec-project/bec_widgets/commit/acd35a278660ce4962167af6237b5d12007f0774))
## v3.9.0 (2026-05-12)
### Bug Fixes
+1 -2
View File
@@ -192,8 +192,7 @@ Positioner boxes and tweak controls handle precise moves, homing, and calibratio
## Documentation
Documentation of BEC Widgets can be found [here](https://bec-widgets.readthedocs.io/en/latest/). The documentation of
the BEC can be found [here](https://bec.readthedocs.io/en/latest/).
The documentation can be found [here](https://bec.readthedocs.io/).
## License
-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):
"""A view for users to manage devices within the application."""
+14 -3
View File
@@ -222,6 +222,7 @@ class BECGuiClient(RPCBase):
self._ipython_registry: dict[str, RPCReference] = {}
self.available_widgets = AvailableWidgetsNamespace()
register_serializer_extension()
self._rpc_timeout = 5
####################
#### Client API ####
@@ -232,6 +233,16 @@ class BECGuiClient(RPCBase):
"""The launcher object."""
return RPCBase(gui_id=f"{self._gui_id}:launcher", parent=self, object_name="launcher")
def set_rpc_timeout(self, timeout: float):
"""Set the timeout for RPC calls to the GUI server.
Args:
timeout(float): The timeout in seconds.
"""
if not isinstance(timeout, (int, float)) or timeout < 0:
raise ValueError("Timeout must be a non-negative number.")
self._rpc_timeout = timeout
def _safe_register_stream(self, endpoint: EndpointInfo, cb: Callable, **kwargs):
"""Check if already registered for registration in idempotent functions."""
if not self._client.connector.any_stream_is_registered(endpoint, cb=cb):
@@ -358,7 +369,7 @@ class BECGuiClient(RPCBase):
)
if not self._check_if_server_is_alive():
self.start(wait=True)
self.show(wait=True)
if wait:
with wait_for_server(self):
return self._new_impl(
@@ -550,7 +561,7 @@ class BECGuiClient(RPCBase):
if self.launcher and len(self._top_level) == 0:
self.launcher._run_rpc("show") # pylint: disable=protected-access
for window in self._top_level.values():
window.show()
window.raise_window()
def _show_all(self):
with wait_for_server(self):
@@ -569,7 +580,7 @@ class BECGuiClient(RPCBase):
if self.launcher and len(self._top_level) == 0:
self.launcher._run_rpc("raise") # pylint: disable=protected-access
for window in self._top_level.values():
window._run_rpc("raise") # type: ignore[attr-defined]
window.raise_window()
def _raise_all(self):
with wait_for_server(self):
-10
View File
@@ -42,10 +42,6 @@ designer_plugins = {
"bec_widgets.widgets.control.device_input.device_combobox.device_combobox",
"DeviceComboBox",
),
"DeviceLineEdit": (
"bec_widgets.widgets.control.device_input.device_line_edit.device_line_edit",
"DeviceLineEdit",
),
"Heatmap": ("bec_widgets.widgets.plots.heatmap.heatmap", "Heatmap"),
"IDEExplorer": ("bec_widgets.widgets.utility.ide_explorer.ide_explorer", "IDEExplorer"),
"Image": ("bec_widgets.widgets.plots.image.image", "Image"),
@@ -101,10 +97,6 @@ designer_plugins = {
"SignalComboBox",
),
"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"),
"StopButton": ("bec_widgets.widgets.control.buttons.stop_button.stop_button", "StopButton"),
"TextBox": ("bec_widgets.widgets.editors.text_box.text_box", "TextBox"),
@@ -134,7 +126,6 @@ widget_icons = {
"DarkModeButton": "dark_mode",
"DeviceBrowser": "lists",
"DeviceComboBox": "list_alt",
"DeviceLineEdit": "edit_note",
"Heatmap": "dataset",
"IDEExplorer": "widgets",
"Image": "image",
@@ -160,7 +151,6 @@ widget_icons = {
"ScatterWaveform": "scatter_plot",
"SignalComboBox": "list_alt",
"SignalLabel": "scoreboard",
"SignalLineEdit": "vital_signs",
"SpinnerWidget": "progress_activity",
"StopButton": "dangerous",
"TextBox": "chat",
+13 -3
View File
@@ -24,6 +24,8 @@ else:
# pylint: disable=protected-access
_DEFAULT_RPC_TIMEOUT = object()
def _name_arg(arg):
if isinstance(arg, DeviceBaseWithConfig):
@@ -154,6 +156,7 @@ class RPCReference:
class RPCBase:
def __init__(
self,
gui_id: str | None = None,
@@ -207,12 +210,16 @@ class RPCBase:
# Use explicit call to ensure action name is 'raise' (not 'raise_')
return self._run_rpc("raise")
def hide(self):
"""Hide this widget (or its container)."""
return self._run_rpc("hide")
def _run_rpc(
self,
method,
*args,
wait_for_rpc_response=True,
timeout=5,
wait_for_rpc_response: bool = True,
timeout: float | None | object = _DEFAULT_RPC_TIMEOUT,
gui_id: str | None = None,
**kwargs,
) -> Any:
@@ -223,13 +230,16 @@ class RPCBase:
method: The method to call.
args: The arguments to pass to the method.
wait_for_rpc_response: Whether to wait for the RPC response.
timeout: The timeout for the RPC response.
timeout: The timeout for the RPC response. If omitted, the client's default RPC
timeout is used. If explicitly set to None, wait indefinitely.
gui_id: The GUI ID to use for the RPC call. If None, the default GUI ID is used.
kwargs: The keyword arguments to pass to the method.
Returns:
The result of the RPC call.
"""
if timeout is _DEFAULT_RPC_TIMEOUT:
timeout = self._root._rpc_timeout
if method in ["show", "hide", "raise"] and gui_id is None:
obj = self._root._server_registry.get(self._gui_id)
if obj is None:
+4 -4
View File
@@ -15,7 +15,7 @@ class FakeDevice(BECDevice):
super().__init__(name=name)
self._enabled = enabled
self.signals = {self.name: {"value": 1.0}}
self.description = {self.name: {"source": self.name, "dtype": "number", "shape": []}}
self._description = {self.name: {"source": self.name, "dtype": "number", "shape": []}}
self._readout_priority = readout_priority
self._config = {
"readoutPriority": "baseline",
@@ -74,7 +74,7 @@ class FakeDevice(BECDevice):
Returns:
dict: Description of the device
"""
return self.description
return self._description
class FakePositioner(BECPositioner):
@@ -96,7 +96,7 @@ class FakePositioner(BECPositioner):
self._limits = limits
self._readout_priority = readout_priority
self.signals = {self.name: {"value": 1.0}}
self.description = {self.name: {"source": self.name, "dtype": "number", "shape": []}}
self._description = {self.name: {"source": self.name, "dtype": "number", "shape": []}}
self._config = {
"readoutPriority": "baseline",
"deviceClass": "ophyd_devices.SimPositioner",
@@ -176,7 +176,7 @@ class FakePositioner(BECPositioner):
Returns:
dict: Description of the device
"""
return self.description
return self._description
@property
def precision(self):
+9
View File
@@ -143,6 +143,15 @@ def get_plugin_designer_registry() -> dict[str, tuple[str, str]]:
return {}
@lru_cache
def get_plugin_widget_icons() -> dict[str, str]:
"""If there is a plugin repository installed, return the designer widget icon registry."""
designer_module = get_plugin_designer_module()
if designer_module and hasattr(designer_module, "widget_icons"):
return designer_module.widget_icons
return {}
def get_all_plugin_widgets() -> BECClassContainer:
"""If there is a plugin repository installed, load all widgets from it."""
if plugin := user_widget_plugin():
+1 -1
View File
@@ -362,7 +362,7 @@ class BECWidget(BECConnector):
"""Wrap the close even to ensure the rpc_register is cleaned up."""
try:
if not self._destroyed:
self.cleanup()
self._destroyed = True
self.cleanup()
finally:
super().closeEvent(event) # pylint: disable=no-member
+90 -317
View File
@@ -1,12 +1,12 @@
"""Module for handling filter I/O operations in BEC Widgets for input fields.
These operations include filtering device/signal names and/or device types.
"""
"""Small helpers for populating editable combo boxes used by device inputs."""
from abc import ABC, abstractmethod
from __future__ import annotations
from contextlib import nullcontext
from bec_lib.logger import bec_logger
from qtpy.QtCore import QStringListModel
from qtpy.QtWidgets import QComboBox, QCompleter, QLineEdit
from qtpy.QtCore import QSignalBlocker
from qtpy.QtWidgets import QComboBox
from typeguard import TypeCheckError
from bec_widgets.utils.ophyd_kind_util import Kind
@@ -14,329 +14,102 @@ from bec_widgets.utils.ophyd_kind_util import Kind
logger = bec_logger.logger
class WidgetFilterHandler(ABC):
"""Abstract base class for widget filter handlers"""
def replace_combobox_items(
combo_box: QComboBox,
items: list[str | tuple],
*,
preserve_current_text: bool = False,
block_signals: bool = False,
) -> None:
"""Replace all combobox entries.
@abstractmethod
def set_selection(self, widget, selection: list[str | tuple]) -> None:
"""Set the filtered_selection for the widget
Args:
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):
"""Handler for QLineEdit widget"""
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):
"""Handler for QComboBox widget"""
def set_selection(self, widget: QComboBox, selection: list[str | tuple]) -> None:
"""Set the selection for the widget to the completer model
Args:
widget (QComboBox): The QComboBox widget
selection (list[str | tuple]): Filtered selection of items. If tuple, it contains (text, data) pairs.
"""
widget.clear()
if len(selection) == 0:
return
for element in selection:
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:
"""Public interface to set filters for input widgets.
It supports the list of widgets stored in class attribute _handlers.
Args:
combo_box: Combobox whose entries should be replaced.
items: Entries to add. String entries are added as display text. Tuple entries are
passed to ``QComboBox.addItem`` as ``(text, data)``.
preserve_current_text: If True, restore the combobox text after replacing the items.
block_signals: If True, block combobox signals while the items are replaced.
"""
current_text = combo_box.currentText()
signal_blocker = QSignalBlocker(combo_box) if block_signals else nullcontext()
with signal_blocker:
combo_box.clear()
for item in items:
if isinstance(item, str):
combo_box.addItem(item)
else:
combo_box.addItem(*item)
if preserve_current_text:
combo_box.setCurrentText(current_text)
_handlers = {QLineEdit: LineEditFilterHandler, QComboBox: ComboBoxFilterHandler}
@staticmethod
def set_selection(widget, selection: list[str | tuple], ignore_errors=True):
"""
Retrieve value from the widget instance.
def signal_items_for_kind(
*, 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.
Args:
widget: Widget instance.
selection (list[str | tuple]): Filtered selection of items.
If tuple, it contains (text, data) pairs.
ignore_errors(bool, optional): Whether to ignore if no handler is found.
"""
handler_class = FilterIO._find_handler(widget)
if handler_class:
return handler_class().set_selection(widget=widget, selection=selection)
if not ignore_errors:
raise ValueError(
f"No matching handler for widget type: {type(widget)} in handler list {FilterIO._handlers}"
)
return None
Args:
kind: Signal kind to collect.
signal_filter: Enabled signal kinds.
device_info: Signal metadata from the BEC device info dictionary.
device_name: Name of the device owning the signals.
@staticmethod
def check_input(widget, text: str, ignore_errors=True):
"""
Check if the input text is in the filtered selection.
Returns:
Combobox entries as ``(display_text, signal_info)`` tuples.
"""
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
Args:
widget: Widget instance.
text(str): Input text.
ignore_errors(bool, optional): Whether to ignore if no handler is found.
obj_name = signal_info.get("obj_name", "")
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
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
if (
signal_without_device != signal_name
and component_name.replace(".", "_") != signal_without_device
):
items.append((f"{signal_without_device} ({signal_name})", signal_info))
else:
items.append((signal_name, signal_info))
return items
@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.
def get_bec_signals_for_classes(
*, client, signal_class_filter: str | list[str], ndim_filter: int | list[int] | None = None
) -> list[tuple[str, str, dict]]:
"""Return BEC signals filtered by signal class and optional dimensionality.
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}"
)
Args:
client: BEC client that provides ``device_manager.get_bec_signals``.
signal_class_filter: Signal class name or class names passed to the device manager.
ndim_filter: Optional dimensionality filter. If provided, only signals whose
``describe.signal_info.ndim`` is in this value are returned.
@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.
Returns:
Tuples of ``(device_name, signal_name, signal_config)`` for matching signals.
"""
if not client or not hasattr(client, "device_manager"):
return []
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.
try:
signals = client.device_manager.get_bec_signals(signal_class_filter)
except TypeCheckError as exc:
logger.warning(f"Error retrieving signals: {exc}")
return []
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}"
)
if ndim_filter is None:
return signals
@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
accepted_ndim = [ndim_filter] if isinstance(ndim_filter, int) else ndim_filter
filtered_signals: list[tuple[str, str, dict]] = []
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 accepted_ndim:
filtered_signals.append((device_name, signal_name, signal_config))
return filtered_signals
+4 -2
View File
@@ -27,8 +27,10 @@ from qtpy.QtWidgets import (
import bec_widgets
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 DeviceComboBox
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import (
BECDeviceFilter,
DeviceComboBox,
)
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:
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):
widget.setCurrentIndex(value)
@@ -1465,6 +1465,16 @@ class DockAreaWidget(BECWidget, QWidget):
for dock in self.dock_list():
self._delete_dock(dock)
def cleanup(self):
"""Tear down all docks via the Qt ADS API before the base BECWidget cleanup runs.
Explicitly releasing dock widgets through the CDockManager API first prevents crashes
in PySide6 6.11.0 / PySide6-QtAds 4.5.x where the CDockManager destructor interacts
badly with dock widgets that are deleted outside of it.
"""
self.delete_all()
super().cleanup()
if __name__ == "__main__": # pragma: no cover
import sys
@@ -1,6 +1,7 @@
from __future__ import annotations
import os
from functools import lru_cache
from typing import Literal, Mapping, Sequence
import slugify
@@ -20,7 +21,12 @@ from qtpy.QtWidgets import (
import bec_widgets.widgets.containers.qt_ads as QtAds
from bec_widgets import BECWidget, SafeProperty, SafeSlot
from bec_widgets.cli.designer_plugins import widget_icons
from bec_widgets.utils.bec_plugin_helper import (
get_plugin_rpc_widget_registry,
get_plugin_widget_icons,
)
from bec_widgets.utils.colors import apply_theme
from bec_widgets.utils.plugin_utils import get_rpc_widget_registry
from bec_widgets.utils.rpc_decorator import rpc_timeout
from bec_widgets.utils.rpc_widget_handler import widget_handler
from bec_widgets.utils.toolbars.actions import (
@@ -74,6 +80,19 @@ PROFILE_STATE_KEYS = {key: SETTINGS_KEYS[key] for key in ("geom", "state", "ads_
StartupProfile = Literal["restore", "skip"] | str | None
@lru_cache
def _plugin_toolbar_actions() -> dict[str, tuple[str, str, str]]:
plugin_registry = get_plugin_rpc_widget_registry()
internal_registry = get_rpc_widget_registry()
plugin_icons = get_plugin_widget_icons()
return {
widget_name: (plugin_icons.get(widget_name, "widgets"), f"Add {widget_name}", widget_name)
for widget_name in sorted(plugin_registry)
if widget_name not in internal_registry
}
class BECDockArea(DockAreaWidget):
RPC = True
PLUGIN = False
@@ -390,6 +409,10 @@ class BECDockArea(DockAreaWidget):
_build_menu("menu_devices", "Add Device Control ", device_actions)
_build_menu("menu_utils", "Add Utils ", util_actions)
plugin_actions = _plugin_toolbar_actions()
if plugin_actions:
_build_menu("menu_plugins", "Add Plugins ", plugin_actions)
# Create flat toolbar bundles for each widget type
def _build_flat_bundles(category: str, mapping: dict[str, tuple[str, str, str]]):
bundle = ToolbarBundle(f"flat_{category}", self.toolbar.components)
@@ -460,14 +483,16 @@ class BECDockArea(DockAreaWidget):
bda.add_action("dark_mode")
self.toolbar.add_bundle(bda)
self._apply_toolbar_layout()
# Store mappings on self for use in _hook_toolbar
# Store mappings on self for use in _hook_toolbar and _apply_toolbar_layout
self._ACTION_MAPPINGS = {
"menu_plots": plot_actions,
"menu_devices": device_actions,
"menu_utils": util_actions,
}
if plugin_actions:
self._ACTION_MAPPINGS["menu_plugins"] = plugin_actions
self._apply_toolbar_layout()
def _hook_toolbar(self):
def _connect_menu(menu_key: str):
@@ -476,7 +501,8 @@ class BECDockArea(DockAreaWidget):
# first two items not needed for this part
for key, (_, _, widget_type) in mapping.items():
act = menu.actions[key].action
toolbar_action = menu.actions[key]
act = toolbar_action.action
if key == "terminal":
act.triggered.connect(
lambda _, t=widget_type: self.new(widget=t, closable=True, startup_cmd=None)
@@ -487,12 +513,18 @@ class BECDockArea(DockAreaWidget):
widget=t, closable=True, show_settings_action=False
)
)
elif menu_key == "menu_plugins":
act.triggered.connect(
lambda _, t=widget_type, a=toolbar_action: self._new_plugin_widget(t, a)
)
else:
act.triggered.connect(lambda _, t=widget_type: self.new(widget=t))
_connect_menu("menu_plots")
_connect_menu("menu_devices")
_connect_menu("menu_utils")
if "menu_plugins" in self._ACTION_MAPPINGS:
_connect_menu("menu_plugins")
def _connect_flat_actions(mapping: dict[str, tuple[str, str, str]]):
for action_id, (_, _, widget_type) in mapping.items():
@@ -507,6 +539,10 @@ class BECDockArea(DockAreaWidget):
self.toolbar.components.get_action("attach_all").action.triggered.connect(self.attach_all)
self.toolbar.components.get_action("screenshot").action.triggered.connect(self.screenshot)
def _new_plugin_widget(self, widget_type: str, toolbar_action: MaterialIconAction) -> None:
# Created as helper method for simple tests
self.new(widget=widget_type, dock_icon=toolbar_action.get_icon())
def _set_editable(self, editable: bool) -> None:
self.workspace_is_locked = not editable
self._editable = editable
@@ -1108,14 +1144,10 @@ class BECDockArea(DockAreaWidget):
if mode_key == "user":
bundles = ["spacer_bundle", "workspace", "dock_actions"]
elif mode_key == "creator":
bundles = [
"menu_plots",
"menu_devices",
"menu_utils",
"spacer_bundle",
"workspace",
"dock_actions",
]
bundles = ["menu_plots", "menu_devices", "menu_utils"]
if "menu_plugins" in getattr(self, "_ACTION_MAPPINGS", {}):
bundles.append("menu_plugins")
bundles += ["spacer_bundle", "workspace", "dock_actions"]
elif mode_key == "plot":
bundles = ["flat_plots", "spacer_bundle", "workspace", "dock_actions"]
elif mode_key == "device":
@@ -21,9 +21,9 @@ from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.widgets.control.device_control.position_indicator.position_indicator import (
PositionIndicator,
)
from bec_widgets.widgets.control.device_input.base_classes.device_input_base import BECDeviceFilter
from bec_widgets.widgets.control.device_input.device_line_edit.device_line_edit import (
DeviceLineEdit,
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import (
BECDeviceFilter,
DeviceComboBox,
)
from bec_widgets.widgets.utility.spinner.spinner import SpinnerWidget
@@ -257,10 +257,10 @@ class PositionerBoxBase(BECWidget, QWidget):
self._dialog = QDialog(self)
self._dialog.setWindowTitle("Positioner Selection")
layout = QVBoxLayout()
line_edit = DeviceLineEdit(
line_edit = DeviceComboBox(
self, client=self.client, device_filter=[BECDeviceFilter.POSITIONER]
)
line_edit.textChanged.connect(set_positioner)
line_edit.currentTextChanged.connect(set_positioner)
layout.addWidget(line_edit)
close_button = QPushButton("Close")
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,121 @@
"""Editable combobox for selecting BEC devices."""
from __future__ import annotations
import enum
from bec_lib.callback_handler import EventType
from bec_lib.device import ReadoutPriority
from qtpy.QtCore import QSize, Signal, Slot
from qtpy.QtWidgets import QComboBox, QSizePolicy
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, field_validator
from qtpy.QtCore import QSize, QStringListModel, Signal, Slot
from qtpy.QtWidgets import QComboBox, QCompleter, QSizePolicy
from bec_widgets.utils.colors import get_accent_colors
from bec_widgets.utils.error_popups import SafeProperty
from bec_widgets.widgets.control.device_input.base_classes.device_input_base import (
BECDeviceFilter,
DeviceInputBase,
DeviceInputConfig,
)
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 get_bec_signals_for_classes, replace_combobox_items
logger = bec_logger.logger
class DeviceComboBox(DeviceInputBase, QComboBox):
class BECDeviceFilter(enum.Enum):
"""Device class filters accepted by :class:`DeviceComboBox`."""
DEVICE = "Device"
POSITIONER = "Positioner"
SIGNAL = "Signal"
COMPUTED_SIGNAL = "ComputedSignal"
class DeviceInputConfig(ConnectionConfig):
"""Serializable configuration for :class:`DeviceComboBox`.
Attributes:
device_filter: Enabled device class filters as ``BECDeviceFilter.value`` strings.
readout_filter: Enabled readout priority filters as ``ReadoutPriority.value`` strings.
devices: Explicit device names shown by the combobox.
default: Device selected by default.
arg_name: Optional argument name used by scan/input widgets.
apply_filter: Whether the combobox should refresh devices from the BEC device manager.
signal_class_filter: Signal class names used to restrict listed devices.
autocomplete: Whether to use the explicit completer model instead of Qt's default
editable-combobox completer.
"""
Combobox widget for device input with autocomplete for device names.
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):
"""Validate configured device class filters.
Args:
value: Device class filter values from the persisted widget configuration.
Returns:
The validated filter values.
Raises:
ValueError: If any configured filter is not a valid ``BECDeviceFilter`` 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):
"""Validate configured readout priority filters.
Args:
value: Readout priority filter values from the persisted widget configuration.
Returns:
The validated filter values.
Raises:
ValueError: If any configured filter is not a valid ``ReadoutPriority`` 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):
"""Editable combobox for selecting a BEC device.
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.
signal_class_filter: List of signal classes to filter the devices by. Only devices with signals of these classes will be shown.
parent: Optional parent widget.
client: Optional BEC client object.
config: Device input configuration as a ``DeviceInputConfig`` instance or dictionary.
gui_id: Optional GUI identifier.
device_filter: Device class filter or filters from ``BECDeviceFilter``.
readout_priority_filter: Readout priority filter or filters from ``ReadoutPriority``.
available_devices: Explicit device names to show. Passing this disables automatic
BEC filtering.
default: Device name selected during initialization.
arg_name: Optional argument name used by scan/input widgets.
signal_class_filter: Signal class names used to restrict listed devices.
autocomplete: If True, use the explicit line-edit style completer. If False, keep
Qt's default editable-combobox completion behavior.
**kwargs: Additional keyword arguments passed to ``BECWidget``.
"""
ICON_NAME = "list_alt"
@@ -37,62 +126,96 @@ class DeviceComboBox(DeviceInputBase, QComboBox):
device_reset = Signal()
device_config_update = Signal()
_device_handler = {
BECDeviceFilter.DEVICE: Device,
BECDeviceFilter.POSITIONER: Positioner,
BECDeviceFilter.SIGNAL: BECSignal,
BECDeviceFilter.COMPUTED_SIGNAL: ComputedSignal,
}
def __init__(
self,
parent=None,
client=None,
config: DeviceInputConfig = None,
config: DeviceInputConfig | dict | None = None,
gui_id: str | None = None,
device_filter: BECDeviceFilter | list[BECDeviceFilter] | None = None,
readout_priority_filter: (
str | ReadoutPriority | list[str] | list[ReadoutPriority] | None
) = None,
device_filter: BECDeviceFilter | str | list[BECDeviceFilter | str] | None = None,
readout_priority_filter: str | ReadoutPriority | list[str | ReadoutPriority] | None = None,
available_devices: list[str] | None = None,
default: str | None = None,
arg_name: str | None = None,
signal_class_filter: list[str] | None = None,
autocomplete: bool | None = None,
**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,
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._is_valid_input = 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
self.setSizePolicy(QSizePolicy.Preferred, QSizePolicy.Fixed)
self.setMinimumSize(QSize(100, 0))
self._callback_id = None
self._is_valid_input = False
self._accent_colors = get_accent_colors()
self._set_first_element_as_empty = False
# 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 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
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
self.set_readout_priority_filter(
readout_priority_filter
or [
ReadoutPriority.MONITORED,
ReadoutPriority.BASELINE,
ReadoutPriority.ASYNC,
ReadoutPriority.CONTINUOUS,
ReadoutPriority.ON_REQUEST,
]
)
if device_filter is not None:
self.set_device_filter(device_filter)
if signal_class_filter is not None:
self.signal_class_filter = signal_class_filter
# Set default device if passed
if default is not None:
self.set_device(default)
else:
self.setCurrentText("")
self._callback_id = self.bec_dispatcher.client.callbacks.register(
EventType.DEVICE_UPDATE, self.on_device_update
)
@@ -100,100 +223,414 @@ class DeviceComboBox(DeviceInputBase, QComboBox):
self.currentTextChanged.connect(self.check_validity)
self.check_validity(self.currentText())
@staticmethod
def _process_config(config: DeviceInputConfig | dict | None) -> DeviceInputConfig:
"""Normalize user-provided configuration.
Args:
config: Existing configuration, configuration dictionary, or None.
Returns:
A validated ``DeviceInputConfig`` instance.
"""
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.
Args:
device: Device name to select.
"""
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."""
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]
@SafeSlot(list)
def set_available_devices(self, devices: list[str]):
"""Use an explicit device list and disable automatic BEC filtering.
Args:
devices: Device names to show in the combobox.
"""
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)
def set_first_element_as_empty(self) -> bool:
"""
Whether the first element in the combobox should be empty.
This is useful to allow the user to select a device from the list.
"""
"""Whether an empty choice is inserted as the first item."""
return self._set_first_element_as_empty
@set_first_element_as_empty.setter
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
if self._set_first_element_as_empty:
self.insertItem(0, "")
current_text = self.currentText()
if value:
if self.count() == 0 or self.itemText(0) != "":
self.insertItem(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:
self.setCompleter(QCompleter(self._completer_model, self))
else:
if self.count() > 0 and self.itemText(0) == "":
self.removeItem(0)
self.setCompleter(QCompleter(self.model(), self))
def on_device_update(self, action: str, content: dict) -> None:
"""
Callback for device update events. Triggers the device_update signal.
@property
def device_filter(self) -> list[BECDeviceFilter]:
"""Device class filters."""
return self._device_filter
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.currentText()
return self.get_device_object(dev_name)
@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)
self.setStyleSheet("border: 1px solid transparent;")
else:
self._is_valid_input = False
self.device_reset.emit()
if self.isEnabled():
self.setStyleSheet("border: 1px solid red;")
def validate_device(self, device: str) -> bool: # type: ignore[override]
"""
Extend validation so that previewsignal pseudodevices (labels like
``"eiger_preview"``) are accepted as valid choices.
The validation run only on device not on the previewsignal.
Args:
device: The text currently entered/selected.
Returns:
True if the device is a genuine BEC device *or* one of the
whitelisted previewsignal entries.
"""
idx = self.findText(device)
if idx >= 0 and isinstance(self.itemData(idx), tuple):
device = self.itemData(idx)[0] # type: ignore[assignment]
return super().validate_device(device)
@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 setEnabled(self, enabled: bool) -> None: # noqa: N802
super().setEnabled(enabled)
self._update_validity_style(self._is_valid_input)
def set_device_filter(
self, filter_selection: BECDeviceFilter | str | list[BECDeviceFilter | str]
):
"""Enable one or more device class filters.
Args:
filter_selection: Filter or filters to enable. Strings must match
``BECDeviceFilter.value``.
"""
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.
Args:
filter_selection: Readout priority filter or filters to enable. Strings must match
``ReadoutPriority.value``.
"""
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:
"""Refresh filters when the BEC device configuration changes.
Args:
action: Device update action emitted by BEC.
content: Device update payload. Currently unused.
"""
if self._callback_id is None or getattr(self, "_destroyed", False):
return
if action in ["add", "remove", "reload"]:
self.device_config_update.emit()
def cleanup(self):
"""Cleanup the widget."""
if self._callback_id is not None:
callback_id = self._callback_id
self._callback_id = None
self.bec_dispatcher.client.callbacks.remove(callback_id)
super().cleanup()
def get_current_device(self) -> object:
"""Return the current BEC device object.
Returns:
Device object for the current combobox text.
"""
return self.get_device_object(self._device_name_from_text(self.currentText()))
@Slot(str)
def check_validity(self, input_text: str) -> None:
"""Validate current text and update visual state.
Args:
input_text: Current combobox text.
"""
if self.validate_device(input_text):
self._is_valid_input = True
self.device_selected.emit(input_text)
else:
self._is_valid_input = False
self.device_reset.emit()
self._update_validity_style(self._is_valid_input)
def validate_device(self, device: str | None) -> bool:
"""Validate a device against the current filtered device selection.
Args:
device: Device name or displayed device text to validate.
Returns:
True if the device exists in the current BEC device manager and is present in the
filtered combobox list.
"""
if not device:
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
def get_device_object(self, device: str) -> object:
"""Return a device object by name.
Args:
device: Device name.
Returns:
BEC device object.
Raises:
ValueError: If the device is not available in the device manager.
"""
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
@staticmethod
def _as_list(value):
return value if isinstance(value, list) else [value]
@staticmethod
def _normalize_device_filter(value: BECDeviceFilter | str) -> BECDeviceFilter | None:
if isinstance(value, BECDeviceFilter):
return value
return BECDeviceFilter._value2member_map_.get(value)
@staticmethod
def _normalize_readout_filter(value: ReadoutPriority | str) -> ReadoutPriority | None:
if isinstance(value, ReadoutPriority):
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 all(isinstance(device, self._device_handler[entry]) for entry in self.device_filter)
def _check_readout_filter(
self, device: Device | BECSignal | ComputedSignal | Positioner
) -> bool:
return device.readout_priority in self.readout_filter
def _update_validity_style(self, is_valid: bool) -> None:
border_color = "transparent" if is_valid or not self.isEnabled() else "red"
self.setStyleSheet(f"border: 1px solid {border_color};")
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]):
items = [""] + devices if self._set_first_element_as_empty else devices
replace_combobox_items(self, items, preserve_current_text=True, block_signals=True)
self._completer_model.setStringList(devices)
self.check_validity(self.currentText())
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
# pylint: disable=import-outside-toplevel
from qtpy.QtWidgets import (
QApplication,
QCheckBox,
@@ -235,10 +672,7 @@ if __name__ == "__main__": # pragma: no cover
def _apply_filters():
raw = class_input.text().strip()
if raw:
combo.signal_class_filter = [entry.strip() for entry in raw.split(",") if entry.strip()]
else:
combo.signal_class_filter = []
combo.signal_class_filter = [entry.strip() for entry in raw.split(",") if entry.strip()]
combo.filter_to_device = filter_device.isChecked()
combo.filter_to_positioner = filter_positioner.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,72 @@
"""Editable combobox for selecting BEC device signals."""
from __future__ import annotations
from qtpy.QtCore import QSize, Qt, Signal
from qtpy.QtWidgets import QComboBox, QSizePolicy
from bec_lib.callback_handler import EventType
from bec_lib.device import Signal as BECSignal
from bec_lib.logger import bec_logger
from pydantic import Field
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.filter_io import ComboBoxFilterHandler, FilterIO
from bec_widgets.utils.ophyd_kind_util import Kind
from bec_widgets.widgets.control.device_input.base_classes.device_signal_input_base import (
DeviceSignalInputBase,
DeviceSignalInputBaseConfig,
from bec_widgets.utils.filter_io import (
get_bec_signals_for_classes,
replace_combobox_items,
signal_items_for_kind,
)
from bec_widgets.utils.ophyd_kind_util import Kind
logger = bec_logger.logger
class SignalComboBox(DeviceSignalInputBase, QComboBox):
class SignalComboBoxConfig(ConnectionConfig):
"""Serializable configuration for :class:`SignalComboBox`.
Attributes:
signal_filter: Enabled signal kind filters as ``Kind.name`` strings.
signal_class_filter: Signal class names used to build the signal list from BEC.
ndim_filter: Optional dimensionality filter for class-based signal lists.
default: Signal selected by default.
arg_name: Optional argument name used by scan/input widgets.
device: Device name used to scope kind-based signal filtering.
signals: Signal names available after filtering.
autocomplete: Whether to use the explicit completer model instead of Qt's default
editable-combobox completer.
"""
Line edit widget for device input with autocomplete for device names.
signal_filter: list[str] = Field(default_factory=list)
signal_class_filter: list[str] = Field(default_factory=list)
ndim_filter: int | list[int] | None = None
default: str | None = None
arg_name: str | None = None
device: str | None = None
signals: list[str] = Field(default_factory=list)
autocomplete: bool = False
class SignalComboBox(BECWidget, QComboBox):
"""Editable combobox for selecting a signal from a BEC device.
Args:
parent: Parent widget.
client: BEC client object.
config: Device input configuration.
gui_id: GUI ID.
device: Device name to filter signals from.
signal_filter: Signal filter, list of signal kinds from ophyd Kind enum. Check DeviceSignalInputBase for more details.
signal_class_filter: List of signal classes to filter the signals by. Only signals of these classes will be shown.
ndim_filter: Dimensionality filter, int or list of ints to filter signals by their number of dimensions. If signal do not support ndim, it will be included in the selection anyway.
default: Default device name.
arg_name: Argument name, can be used for the other widgets which has to call some other function in bec using correct argument names.
store_signal_config: Whether to store the full signal config in the combobox item data.
require_device: If True, signals are only shown/validated when a device is set.
Signals:
device_signal_changed: Emitted when the current text represents a valid signal selection.
signal_reset: Emitted when validation fails and the selection should be treated as cleared.
parent: Optional parent widget.
client: Optional BEC client object.
config: Signal combobox configuration as a ``SignalComboBoxConfig`` instance or
dictionary.
gui_id: Optional GUI identifier.
device: Device name used to scope kind-based signal filtering.
signal_filter: Signal kind filter or filters from ``Kind``.
signal_class_filter: Signal class names used to build a class-based signal list.
ndim_filter: Dimensionality filter for class-based signal lists.
default: Signal selected during initialization.
arg_name: Optional argument name used by scan/input widgets.
store_signal_config: Whether to store each signal config in the item data.
require_device: If True, class-based signal filtering requires a valid selected device.
autocomplete: If True, use the explicit line-edit style completer. If False, keep
Qt's default editable-combobox completion behavior.
**kwargs: Additional keyword arguments passed to ``BECWidget``.
"""
USER_ACCESS = ["set_signal", "set_device", "signals", "get_signal_name"]
@@ -47,289 +82,453 @@ class SignalComboBox(DeviceSignalInputBase, QComboBox):
self,
parent=None,
client=None,
config: DeviceSignalInputBaseConfig | None = None,
config: SignalComboBoxConfig | dict | None = None,
gui_id: 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,
ndim_filter: int | list[int] | None = None,
default: str | None = None,
arg_name: str | None = None,
store_signal_config: bool = True,
require_device: bool = False,
autocomplete: bool | None = None,
**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:
self.config.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.setMinimumSize(QSize(100, 0))
self._set_first_element_as_empty = True
self._signal_class_filter = signal_class_filter or []
self._store_signal_config = store_signal_config
self.config.ndim_filter = ndim_filter or None
self._require_device = require_device
self._is_valid_input = False
if self.config.autocomplete:
self.autocomplete = True
# Note: Runtime arguments (e.g. device, default, arg_name) intentionally take
# precedence over values from the passed-in config. Full reconciliation and
# restoration of state between designer-provided config and runtime arguments
# is not yet implemented, as earlier attempts caused issues with QtDesigner.
self._device_update_register = self.bec_dispatcher.client.callbacks.register(
EventType.DEVICE_UPDATE, self.update_signals_from_filters
)
self.currentTextChanged.connect(self.on_text_changed)
# Kind filtering is always applied; class filtering is additive. If signal_filter is None,
# we default to hinted+normal, even when signal_class_filter is empty or None. To disable
# kinds, pass an explicit signal_filter or toggle include_* after init.
if signal_filter is not None:
self.set_filter(signal_filter)
else:
self.set_filter([Kind.hinted, Kind.normal, Kind.config])
self.set_filter(signal_filter or [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.check_validity(self.currentText())
@staticmethod
def _process_config(config: SignalComboBoxConfig | dict | None) -> SignalComboBoxConfig:
"""Normalize user-provided configuration.
Args:
config: Existing configuration, configuration dictionary, or None.
Returns:
A validated ``SignalComboBoxConfig`` instance.
"""
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.
Args:
signal: Signal display text, object name, or component name to select.
"""
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)
def set_device(self, device: str | None):
"""
Set the device. When signal_class_filter is active, ensures base-class
logic runs and then refreshes the signal list to show only signals from
that device matching the signal class filter.
"""Set the device that scopes kind-based signal filtering.
Args:
device(str): device name.
device: Device name to use for signal filtering. Invalid or empty values clear
the current device and signal selection.
"""
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()
previous_device = self._device
valid_device = device if self.validate_device(device) else None
self._device = valid_device
self.config.device = self._device
if valid_device is None or valid_device != previous_device:
self.setCurrentText("")
self.update_signals_from_filters()
@SafeSlot()
@SafeSlot(dict, dict)
def update_signals_from_filters(
self, content: dict | None = None, metadata: dict | None = None
):
"""Update the filters for the combobox.
When signal_class_filter is active, skip the normal Kind-based filtering.
@SafeSlot(str, dict)
def update_signals_from_filters(self, action: str | None = None, content: dict | None = None):
"""Refresh available signals from the current device and filters.
Args:
content (dict | None): Content dictionary from BEC event.
metadata (dict | None): Metadata dictionary from BEC event.
action: Optional BEC device update action. If provided, only device list changing
actions trigger a refresh.
content: Optional callback payload from BEC device updates. Currently unused.
"""
super().update_signals_from_filters(content, metadata)
if self._device_update_register is None or getattr(self, "_destroyed", False):
return
if action is not None and action not in ["add", "remove", "reload"]:
return
self.config.signal_filter = [kind.name for kind in self.signal_filter]
if self._signal_class_filter:
self.update_signals_from_signal_classes()
return
# pylint: disable=protected-access
if FilterIO._find_handler(self) is ComboBoxFilterHandler:
if len(self._config_signals) > 0:
self.insertItem(
len(self._hinted_signals) + len(self._normal_signals), "Config Signals"
)
self.model().item(len(self._hinted_signals) + len(self._normal_signals)).setEnabled(
False
)
if len(self._normal_signals) > 0:
self.insertItem(len(self._hinted_signals), "Normal Signals")
self.model().item(len(self._hinted_signals)).setEnabled(False)
if len(self._hinted_signals) > 0:
self.insertItem(0, "Hinted Signals")
self.model().item(0).setEnabled(False)
if not self.validate_device(self._device):
self._device = None
self.config.device = None
self._set_signal_groups([], [], [])
return
device = self.get_device_object(self._device)
device_info = device._info.get("signals", {})
if isinstance(device, BECSignal):
self._set_signal_groups([(self._device, {})], [], [])
return
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)
def set_first_element_as_empty(self) -> bool:
"""
Whether the first element in the combobox should be empty.
This is useful to allow the user to select a device from the list.
"""
"""Whether an empty choice is inserted as the first item."""
return self._set_first_element_as_empty
@set_first_element_as_empty.setter
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
if self._set_first_element_as_empty:
self.insertItem(0, "")
if value:
if self.count() == 0 or self.itemText(0) != "":
self.insertItem(0, "")
self.setCurrentIndex(0)
else:
if self.count() > 0 and self.itemText(0) == "":
self.removeItem(0)
elif self.count() > 0 and self.itemText(0) == "":
self.removeItem(0)
@SafeProperty("QStringList")
def signal_class_filter(self) -> list[str]:
"""
Get the list of signal classes to filter.
Returns:
list[str]: List of signal class names to filter.
"""
"""Signal class names used to build the signal list."""
return self._signal_class_filter
@signal_class_filter.setter
def signal_class_filter(self, value: list[str] | None):
"""
Set the signal class filter.
Args:
value (list[str] | None): List of signal class names to filter, or None/empty
to disable class-based filtering and revert to the default behavior.
"""
normalized_value = value or []
self._signal_class_filter = normalized_value
self.config.signal_class_filter = normalized_value
if self._signal_class_filter:
self.update_signals_from_signal_classes()
else:
self.update_signals_from_filters()
self._signal_class_filter = value or []
self.config.signal_class_filter = self._signal_class_filter
self.update_signals_from_filters()
@SafeProperty(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
@ndim_filter.setter
def ndim_filter(self, value: int):
self.config.ndim_filter = None if value < 0 else value
if self._signal_class_filter:
self.update_signals_from_signal_classes(ndim_filter=self.config.ndim_filter)
self.update_signals_from_filters()
@SafeProperty(bool)
def require_device(self) -> bool:
"""
If True, signals are only shown/validated when a device is set.
Note:
This property affects list rebuilding only when a signal_class_filter
is active. Without a signal class filter, the available signals are
managed by the standard Kind-based filtering.
"""
"""Whether validation/listing requires a selected device."""
return self._require_device
@require_device.setter
def require_device(self, value: bool):
self._require_device = value
# Rebuild list when toggled, but only when using signal_class_filter
if self._signal_class_filter:
self.update_signals_from_signal_classes()
self.update_signals_from_filters()
def set_to_obj_name(self, obj_name: str) -> bool:
"""
Set the combobox to the object name of the signal.
@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:
self.setCompleter(QCompleter(self._completer_model, self))
else:
self.setCompleter(QCompleter(self.model(), self))
@property
def signals(self) -> list[str | tuple[str, dict]]:
"""Available signals after filtering."""
return self._signals
@signals.setter
def signals(self, value: list[str | tuple[str, dict]]):
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
def setEnabled(self, enabled: bool) -> None: # noqa: N802
super().setEnabled(enabled)
self._update_validity_style(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.
Args:
obj_name (str): Object name of the signal.
filter_selection: Filter or filters to enable. Strings must match ``Kind`` member
names.
"""
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_device_object(self, device: str) -> object | None:
"""Return a BEC device object by name.
Args:
device: Device name.
Returns:
bool: True if the object name was found and set, False otherwise.
Device object if it exists in the device manager, otherwise None.
"""
for i in range(self.count()):
signal_data = self.itemData(i)
if signal_data and signal_data.get("obj_name") == obj_name:
self.setCurrentIndex(i)
return True
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.
Args:
device: Device name to validate.
raise_on_false: If True, raise instead of returning False for missing devices.
Returns:
True if the device exists in the current device manager.
Raises:
ValueError: If ``raise_on_false`` is True and the device is missing.
"""
if device in self.dev:
return True
if raise_on_false:
raise ValueError(f"Device {device} not found in devicemanager.")
return False
def set_to_first_enabled(self) -> bool:
"""
Set the combobox to the first enabled item.
def validate_signal(self, signal: str) -> bool:
"""Validate a signal by display text, object name, or component name.
Args:
signal: Signal display text, object name, or component name.
Returns:
bool: True if an enabled item was found and set, False otherwise.
True if the signal is present in the current filtered signal list.
"""
for i in range(self.count()):
if self.model().item(i).isEnabled():
self.setCurrentIndex(i)
if not signal:
return False
return self._display_text_for_signal(signal) is not None
def set_to_obj_name(self, obj_name: str) -> bool:
"""Select the item whose signal config has the given object name.
Args:
obj_name: Signal object name to select.
Returns:
True if a matching signal was selected.
"""
index = self._find_signal_index(obj_name)
if index < 0:
return False
self.setCurrentIndex(index)
return True
def set_to_first_enabled(self) -> bool:
"""Select the first enabled item.
Returns:
True if an enabled item was found and selected.
"""
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 False
def get_signal_name(self) -> str:
"""
Get the signal name from the combobox.
"""Return the selected signal object name when available.
Returns:
str: The signal name.
Signal object name from item data, or the current display text when no item data
is available.
"""
signal_name = self.currentText()
index = self.findText(signal_name)
if index == -1:
return signal_name
current_text = self.currentText()
index = self._find_signal_index(current_text)
if index < 0:
return current_text
signal_info = self.itemData(index)
if signal_info:
signal_name = signal_info.get("obj_name", signal_name)
return signal_name if signal_name else ""
if isinstance(signal_info, dict):
return signal_info.get("obj_name") or current_text
return current_text
def get_signal_config(self) -> dict | None:
"""
Get the signal config from the combobox for the currently selected signal.
"""Return the selected signal config if item-data storage is enabled.
Returns:
dict | None: The signal configuration dictionary or None if not available.
Selected signal configuration dictionary, or None when item-data storage is disabled
or the current item has no configuration.
"""
if not self._store_signal_config:
return None
index = self.currentIndex()
if index == -1:
return None
signal_info = self.itemData(index)
return signal_info if signal_info else None
signal_info = self.itemData(self.currentIndex())
return signal_info if isinstance(signal_info, dict) else None
def update_signals_from_signal_classes(self, ndim_filter: int | list[int] | None = None):
"""
Update the combobox with signals filtered by signal classes and optionally by ndim.
Uses device_manager.get_bec_signals() to retrieve signals.
If a device is set, only shows signals from that device.
"""Refresh signals from ``device_manager.get_bec_signals`` for class-based filtering.
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.
ndim_filter: Optional dimensionality filter overriding the configured value for
this refresh.
"""
if not self._signal_class_filter:
return
if self._require_device and not self._device:
self.clear()
self._signals = []
FilterIO.set_selection(widget=self, selection=self._signals)
self.signals = []
return
# Update stored ndim_filter if a new one is provided
if ndim_filter is not None:
self.config.ndim_filter = ndim_filter
self.clear()
# Get signals with ndim filtering applied at the FilterIO level
signals = FilterIO.update_with_signal_class(
widget=self,
signal_class_filter=self._signal_class_filter,
signals = get_bec_signals_for_classes(
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._signals = []
combo_items: list[str | tuple[str, dict]] = []
item_tooltips: dict[int, str] = {}
for device_name, signal_name, signal_config in signals:
# Filter by device if one is set
if self._device and device_name != self._device:
continue
if self._signal_filter:
@@ -339,76 +538,157 @@ class SignalComboBox(DeviceSignalInputBase, QComboBox):
}:
continue
# Get storage_name for tooltip
storage_name = signal_config.get("storage_name", "")
# Store the full signal config as item data if requested
if self._store_signal_config:
self.addItem(signal_name, signal_config)
combo_items.append((signal_name, signal_config))
else:
self.addItem(signal_name)
combo_items.append(signal_name)
# Track for validation
self._signals.append(signal_name)
# Set tooltip to storage_name (Qt.ToolTipRole = 3)
storage_name = signal_config.get("storage_name", "")
if storage_name:
self.setItemData(self.count() - 1, storage_name, Qt.ItemDataRole.ToolTipRole)
item_tooltips[len(combo_items) - 1] = storage_name
# Keep FilterIO selection in sync for validate_signal
FilterIO.set_selection(widget=self, selection=self._signals)
self.signals = combo_items
tooltip_offset = 1 if self._set_first_element_as_empty and self.count() else 0
for item_index, tooltip in item_tooltips.items():
self.setItemData(item_index + tooltip_offset, tooltip, Qt.ItemDataRole.ToolTipRole)
self.check_validity(self.currentText())
@SafeSlot()
def reset_selection(self):
"""Reset the selection of the combobox."""
self.clear()
self.setItemText(0, "Select a device")
"""Reset the current selection and refresh available signals."""
self.setCurrentText("")
self.update_signals_from_filters()
self.device_signal_changed.emit("")
@SafeSlot(str)
def on_text_changed(self, text: str):
"""Validate and emit only when the signal is valid.
For a positioner, the readback value has to be renamed to the device name.
When using signal_class_filter, device validation is skipped.
"""Validate the current text when edited or selected.
Args:
text: Current combobox text.
"""
self.check_validity(text)
@Slot(str)
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.
Args:
input_text: Current combobox text.
"""
if self._signal_class_filter:
if self._require_device and (not self._device or not input_text):
is_valid = False
else:
is_valid = self.validate_signal(input_text)
is_valid = not (self._require_device and not self._device) and self.validate_signal(
input_text
)
else:
if self._require_device and not self.validate_device(self._device):
is_valid = False
else:
is_valid = self.validate_device(self._device) and self.validate_signal(input_text)
is_valid = self.validate_device(self._device) and self.validate_signal(input_text)
if is_valid:
self._is_valid_input = True
self.device_signal_changed.emit(input_text)
self.setStyleSheet("border: 1px solid transparent;")
else:
self._is_valid_input = False
self.signal_reset.emit()
if self.isEnabled():
self.setStyleSheet("border: 1px solid red;")
self._update_validity_style(self._is_valid_input)
@property
def selected_signal_comp_name(self) -> str:
return dict(self.signals).get(self.currentText(), {}).get("component_name", "")
def cleanup(self):
"""Cleanup the widget."""
if self._device_update_register is not None:
callback_id = self._device_update_register
self._device_update_register = None
self.bec_dispatcher.client.callbacks.remove(callback_id)
super().cleanup()
@property
def is_valid_input(self) -> bool:
"""Whether the current text represents a valid signal selection."""
return self._is_valid_input
@staticmethod
def _normalize_kind(value: Kind | str) -> Kind | None:
if isinstance(value, Kind):
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()
self.check_validity(self.currentText())
def _update_validity_style(self, is_valid: bool) -> None:
border_color = "transparent" if is_valid or not self.isEnabled() else "red"
self.setStyleSheet(f"border: 1px solid {border_color};")
def _replace_signal_items(self, items: list[str | tuple[str, dict]] | None = None):
combo_items = self._signals if items is None else items
display_items = [""] + combo_items if self._set_first_element_as_empty else combo_items
replace_combobox_items(
self, display_items, preserve_current_text=bool(self.currentText()), block_signals=True
)
self._completer_model.setStringList(
[entry if isinstance(entry, str) else entry[0] for entry in combo_items]
)
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:
if not signal:
return False
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
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
@@ -417,16 +697,14 @@ if __name__ == "__main__": # pragma: no cover
apply_theme("dark")
widget = QWidget()
widget.setFixedSize(200, 200)
layout = QVBoxLayout()
widget.setLayout(layout)
layout = QVBoxLayout(widget)
box = SignalComboBox(
device="waveform",
signal_class_filter=["AsyncSignal", "AsyncMultiSignal"],
ndim_filter=[1, 2],
store_signal_config=True,
signal_filter=[Kind.hinted, Kind.normal, Kind.config],
) # change signal filter class to test
box.setEditable(True)
)
layout.addWidget(box)
widget.show()
app.exec_()
@@ -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()
@@ -14,7 +14,6 @@ from qtpy.QtWidgets import (
QLabel,
QPushButton,
QSizePolicy,
QSpacerItem,
QVBoxLayout,
QWidget,
)
@@ -25,8 +24,8 @@ from bec_widgets.utils.colors import apply_theme, get_accent_colors
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
from bec_widgets.widgets.control.buttons.stop_button.stop_button import StopButton
from bec_widgets.widgets.control.scan_control.scan_group_box import ScanGroupBox
from bec_widgets.widgets.control.scan_control.scan_info_adapter import ScanInfoAdapter
from bec_widgets.widgets.editors.scan_metadata.scan_metadata import ScanMetadata
from bec_widgets.widgets.utility.toggle.toggle import ToggleSwitch
class ScanParameterConfig(BaseModel):
@@ -84,7 +83,6 @@ class ScanControl(BECWidget, QWidget):
self.kwarg_boxes = []
self.expert_mode = False # TODO implement in the future versions
self.previous_scan = None
self.last_scan_found = None
# Widget Default Parameters
self.config.default_scan = default_scan
@@ -97,6 +95,7 @@ class ScanControl(BECWidget, QWidget):
self._hide_scan_control_buttons = False
self._hide_metadata = False
self._hide_scan_selection_combobox = False
self._scan_info_adapter = ScanInfoAdapter()
# Create and set main layout
self._init_UI()
@@ -123,17 +122,12 @@ class ScanControl(BECWidget, QWidget):
scan_selection_layout.addWidget(self.comboBox_scan_selection, 1)
self.scan_selection_group.layout().addLayout(scan_selection_layout)
# Label to reload the last scan parameters within scan selection group box
self.toggle_layout = QHBoxLayout()
self.toggle_layout.addSpacerItem(
QSpacerItem(0, 0, QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed)
# Button to reload the last scan parameters on demand.
self.last_scan_button = QPushButton(
"Restore last scan parameters", self.scan_selection_group
)
self.last_scan_label = QLabel("Restore last scan parameters", self.scan_selection_group)
self.toggle = ToggleSwitch(parent=self.scan_selection_group, checked=False)
self.toggle.enabled.connect(self.request_last_executed_scan_parameters)
self.toggle_layout.addWidget(self.last_scan_label)
self.toggle_layout.addWidget(self.toggle)
self.scan_selection_group.layout().addLayout(self.toggle_layout)
self.last_scan_button.clicked.connect(self.request_last_executed_scan_parameters)
self.scan_selection_group.layout().addWidget(self.last_scan_button)
self.scan_selection_group.setSizePolicy(
QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Fixed
)
@@ -191,12 +185,17 @@ class ScanControl(BECWidget, QWidget):
MessageEndpoints.available_scans()
).resource
if self.config.allowed_scans is None:
supported_scans = ["ScanBase", "SyncFlyScanBase", "AsyncFlyScanBase"]
allowed_scans = [
scan_name
for scan_name, scan_info in self.available_scans.items()
if scan_info["base_class"] in supported_scans and len(scan_info["gui_config"]) > 0
]
supported_scans = ["ScanBase", "SyncFlyScanBase", "AsyncFlyScanBase", "ScanBaseV4"]
def _is_scan_supported(scan_name):
scan_info = self.available_scans[scan_name]
return (
scan_info.get("base_class") in supported_scans
and self._scan_info_adapter.has_scan_ui_config(scan_info)
and not scan_name.startswith("_")
)
allowed_scans = filter(_is_scan_supported, self.available_scans.keys())
else:
allowed_scans = self.config.allowed_scans
@@ -206,7 +205,6 @@ class ScanControl(BECWidget, QWidget):
"""Callback for scan selection combo box"""
selected_scan_name = self.comboBox_scan_selection.currentText()
self.scan_selected.emit(selected_scan_name)
self.request_last_executed_scan_parameters()
self.restore_scan_parameters(selected_scan_name)
@SafeSlot()
@@ -215,10 +213,6 @@ class ScanControl(BECWidget, QWidget):
"""
Requests the last executed scan parameters from BEC and restores them to the scan control widget.
"""
self.last_scan_found = False
if not self.toggle.checked:
return
current_scan = self.comboBox_scan_selection.currentText()
history = (
self.client.connector.xread(
@@ -246,8 +240,6 @@ class ScanControl(BECWidget, QWidget):
if merged and self.kwarg_boxes:
for box in self.kwarg_boxes:
box.set_parameters(merged)
self.last_scan_found = True
break
@SafeProperty(str)
@@ -390,14 +382,14 @@ class ScanControl(BECWidget, QWidget):
self.reset_layout()
selected_scan_info = self.available_scans.get(scan_name, {})
gui_config = selected_scan_info.get("gui_config", {})
self.arg_group = gui_config.get("arg_group", None)
self.kwarg_groups = gui_config.get("kwarg_groups", None)
gui_config = self._scan_info_adapter.build_scan_ui_config(selected_scan_info)
arg_group = gui_config.get("arg_group", None)
kwarg_groups = gui_config.get("kwarg_groups", [])
if bool(self.arg_group["arg_inputs"]):
self.add_arg_group(self.arg_group)
if len(self.kwarg_groups) > 0:
self.add_kwargs_boxes(self.kwarg_groups)
if arg_group and bool(arg_group.get("arg_inputs")):
self.add_arg_group(arg_group)
if kwarg_groups:
self.add_kwargs_boxes(kwarg_groups)
self.update()
self.adjustSize()
@@ -428,6 +420,7 @@ class ScanControl(BECWidget, QWidget):
position = self.ARG_BOX_POSITION + (1 if self.arg_box is not None else 0)
for group in groups:
box = ScanGroupBox(box_type="kwargs", config=group)
box.reference_units_changed.connect(self._apply_reference_units_to_other_boxes)
box.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Fixed)
self.layout.insertWidget(position + len(self.kwarg_boxes), box)
self.kwarg_boxes.append(box)
@@ -441,11 +434,30 @@ class ScanControl(BECWidget, QWidget):
"""
self.arg_box = ScanGroupBox(box_type="args", config=group)
self.arg_box.device_selected.connect(self.emit_device_selected)
self.arg_box.reference_units_changed.connect(self._apply_reference_units_to_other_boxes)
self.arg_box.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Fixed)
self.arg_box.hide_add_remove_buttons = self._hide_add_remove_buttons
self.layout.insertWidget(self.ARG_BOX_POSITION, self.arg_box)
self.arg_box.setVisible(not self._hide_arg_box)
def _scan_group_boxes(self) -> list[ScanGroupBox]:
boxes = []
if self.arg_box is not None:
boxes.append(self.arg_box)
boxes.extend(self.kwarg_boxes)
return boxes
def _apply_reference_units_to_other_boxes(
self, source_box: ScanGroupBox, reference_name: str, units: str | None
) -> None:
"""
Propagate device-derived units to scan fields that reference a device in another group.
"""
for box in self._scan_group_boxes():
if box is source_box:
continue
box.apply_reference_units(reference_name, units)
@SafeSlot(str)
def emit_device_selected(self, dev_names):
"""
@@ -496,8 +508,6 @@ class ScanControl(BECWidget, QWidget):
Args:
scan_name(str): Name of the scan to restore the parameters for.
"""
if self.last_scan_found is True:
return
scan_params = self.config.scans.get(scan_name, None)
if scan_params is None and self.previous_scan is None:
return
@@ -21,9 +21,9 @@ from qtpy.QtWidgets import (
)
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_line_edit.device_line_edit import (
DeviceLineEdit,
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import (
BECDeviceFilter,
DeviceComboBox,
)
logger = bec_logger.logger
@@ -164,8 +164,8 @@ class ScanCheckBox(QCheckBox):
class ScanGroupBox(QGroupBox):
WIDGET_HANDLER = {
ScanArgType.DEVICE: DeviceLineEdit,
ScanArgType.DEVICEBASE: DeviceLineEdit,
ScanArgType.DEVICE: DeviceComboBox,
ScanArgType.DEVICEBASE: DeviceComboBox,
ScanArgType.FLOAT: ScanDoubleSpinBox,
ScanArgType.INT: ScanSpinBox,
ScanArgType.BOOL: ScanCheckBox,
@@ -174,6 +174,7 @@ class ScanGroupBox(QGroupBox):
}
device_selected = Signal(str)
reference_units_changed = Signal(object, str, object)
def __init__(
self,
@@ -209,6 +210,8 @@ class ScanGroupBox(QGroupBox):
self.labels = []
self.widgets = []
self._widget_configs = {}
self._column_labels = {}
self.selected_devices = {}
self.init_box(self.config)
@@ -247,6 +250,7 @@ class ScanGroupBox(QGroupBox):
label = QLabel(text=display_name)
self.layout.addWidget(label, row, column_index)
self.labels.append(label)
self._column_labels[column_index] = label
def add_input_widgets(self, group_inputs: dict, row) -> None:
"""
@@ -271,22 +275,41 @@ class ScanGroupBox(QGroupBox):
continue
if default == "_empty":
default = None
widget = widget_class(parent=self.parent(), arg_name=arg_name, default=default)
if isinstance(widget, DeviceLineEdit):
widget.set_device_filter(BECDeviceFilter.DEVICE)
if widget_class is DeviceComboBox:
widget = widget_class(
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)
self._apply_numeric_precision(widget, item)
self._apply_numeric_limits(widget, item)
if isinstance(widget, DeviceComboBox):
self.selected_devices[widget] = ""
widget.device_selected.connect(self.emit_device_selected)
widget.currentTextChanged.connect(
lambda text, device_widget=widget: self._handle_device_text_changed(
device_widget, text
)
)
if isinstance(widget, ScanLiteralsComboBox):
widget.set_literals(item["type"].get("Literal", []))
tooltip = item.get("tooltip", None)
if tooltip is not None:
widget.setToolTip(item["tooltip"])
self._widget_configs[widget] = item
self._apply_unit_metadata(widget, item)
self.layout.addWidget(widget, row, column_index)
self.widgets.append(widget)
@Slot(str)
def emit_device_selected(self, device_name):
self.selected_devices[self.sender()] = device_name.strip()
sender = self.sender()
self.selected_devices[sender] = device_name.strip()
if isinstance(sender, DeviceComboBox):
units = self._device_units(sender.get_current_device())
self._update_reference_units(sender, units)
self._emit_reference_units_changed(sender, units)
selected_devices_str = " ".join(self.selected_devices.values())
self.device_selected.emit(selected_devices_str)
@@ -311,8 +334,9 @@ class ScanGroupBox(QGroupBox):
return
for widget in self.widgets[-len(self.inputs) :]:
if isinstance(widget, DeviceLineEdit):
if isinstance(widget, DeviceComboBox):
self.selected_devices[widget] = ""
self._widget_configs.pop(widget, None)
widget.close()
widget.deleteLater()
self.widgets = self.widgets[: -len(self.inputs)]
@@ -323,8 +347,9 @@ class ScanGroupBox(QGroupBox):
def remove_all_widget_bundles(self):
"""Remove every widget bundle from the scan control layout."""
for widget in list(self.widgets):
if isinstance(widget, DeviceLineEdit):
if isinstance(widget, DeviceComboBox):
self.selected_devices.pop(widget, None)
self._widget_configs.pop(widget, None)
widget.close()
widget.deleteLater()
self.layout.removeWidget(widget)
@@ -360,8 +385,10 @@ class ScanGroupBox(QGroupBox):
for j in range(self.layout.columnCount()):
try: # In case that the bundle size changes
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()
elif isinstance(widget, DeviceComboBox):
value = widget.currentText()
else:
value = WidgetIO.get_value(widget)
args.append(value)
@@ -373,8 +400,10 @@ class ScanGroupBox(QGroupBox):
kwargs = {}
for i in range(self.layout.columnCount()):
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
elif isinstance(widget, DeviceComboBox):
value = widget.currentText()
elif isinstance(widget, ScanLiteralsComboBox):
value = widget.get_value()
else:
@@ -390,7 +419,7 @@ class ScanGroupBox(QGroupBox):
if item is not None:
widget = item.widget()
if widget is not None:
if isinstance(widget, DeviceLineEdit):
if isinstance(widget, DeviceComboBox):
widget_rows += 1
return widget_rows
@@ -423,3 +452,159 @@ class ScanGroupBox(QGroupBox):
if widget.arg_name == key:
WidgetIO.set_value(widget, value)
break
@staticmethod
def _unit_tooltip(item: dict, units: str | None = None) -> str | None:
tooltip = item.get("tooltip", None)
reference_units = item.get("reference_units", None)
units = units or item.get("units", None)
tooltip_parts = [tooltip] if tooltip else []
if units:
tooltip_parts.append(f"Units: {units}")
elif reference_units:
tooltip_parts.append(f"Units from: {reference_units}")
if tooltip_parts:
return "\n".join(tooltip_parts)
return None
def _apply_unit_metadata(self, widget, item: dict, units: str | None = None) -> None:
units = units or item.get("units", None)
tooltip = self._unit_tooltip(item, units)
existing_tooltip = widget.toolTip()
if existing_tooltip:
# strip the existing unit info from the tooltip if it exists
# to avoid tooltip bloat on multiple updates
existing_tooltip = "\n".join(
line
for line in existing_tooltip.splitlines()
if not (line.startswith("Units:") or line.startswith("Units from:"))
).strip()
if tooltip:
if existing_tooltip:
widget.setToolTip(f"{existing_tooltip}\n{tooltip}")
else:
widget.setToolTip(tooltip)
if hasattr(widget, "setSuffix"):
widget.setSuffix(f" {units}" if units else "")
def _refresh_column_label(self, column: int, item: dict) -> None:
if column not in self._column_labels:
return
self._column_labels[column].setText(item.get("display_name", item.get("name", None)))
@staticmethod
def _device_units(device) -> str | None:
egu = getattr(device, "egu", None)
if not callable(egu):
return None
try:
return egu()
except Exception:
logger.exception("Failed to fetch engineering units from device %s", device)
return None
def _widget_position(self, widget) -> tuple[int, int] | None:
for row in range(self.layout.rowCount()):
for column in range(self.layout.columnCount()):
item = self.layout.itemAtPosition(row, column)
if item is not None and item.widget() is widget:
return row, column
return None
def _update_reference_units(self, device_widget: DeviceComboBox, units: str | None) -> None:
position = self._widget_position(device_widget)
if position is None:
return
source_row, _ = position
source_name = device_widget.arg_name
for widget in self.widgets:
item = self._widget_configs.get(widget, {})
if item.get("reference_units") != source_name:
continue
widget_position = self._widget_position(widget)
if widget_position is None:
continue
row, column = widget_position
if self.box_type == "args" and row != source_row:
continue
self._apply_unit_metadata(widget, item, units)
self._refresh_column_label(column, item)
def apply_reference_units(self, reference_name: str, units: str | None) -> None:
"""
Apply units to widgets that reference an argument owned by another group box.
Cross-box references only have one widget row, so row scoping is intentionally handled by
the source group before this method is called.
"""
for widget in self.widgets:
item = self._widget_configs.get(widget, {})
if item.get("reference_units") != reference_name:
continue
self._apply_unit_metadata(widget, item, units)
position = self._widget_position(widget)
if position is not None:
_, column = position
self._refresh_column_label(column, item)
def _emit_reference_units_changed(
self, device_widget: DeviceComboBox, units: str | None
) -> None:
reference_name = getattr(device_widget, "arg_name", None)
if not reference_name:
return
self.reference_units_changed.emit(self, reference_name, units)
def _handle_device_text_changed(self, device_widget: DeviceComboBox, device_name: str) -> None:
if not device_widget.validate_device(device_name):
self.selected_devices[device_widget] = ""
self._update_reference_units(device_widget, None)
self._emit_reference_units_changed(device_widget, None)
@staticmethod
def _apply_numeric_precision(widget: ScanDoubleSpinBox, item: dict) -> None:
if not isinstance(widget, ScanDoubleSpinBox):
return
precision = item.get("precision")
if precision is None:
return
try:
widget.setDecimals(max(0, int(precision)))
except (TypeError, ValueError):
logger.warning(
"Ignoring invalid precision %r for parameter %s", precision, item.get("name")
)
@staticmethod
def _apply_numeric_limits(widget: ScanDoubleSpinBox | ScanSpinBox, item: dict) -> None:
if isinstance(widget, ScanSpinBox):
minimum = -2147483647 # largest int which qt allows
maximum = 2147483647
if item.get("ge") is not None:
minimum = int(item["ge"])
if item.get("gt") is not None:
minimum = int(item["gt"]) + 1
if item.get("le") is not None:
maximum = int(item["le"])
if item.get("lt") is not None:
maximum = int(item["lt"]) - 1
widget.setRange(minimum, maximum)
return
if isinstance(widget, ScanDoubleSpinBox):
minimum = -float("inf")
maximum = float("inf")
step = 10 ** (-widget.decimals())
if item.get("ge") is not None:
minimum = float(item["ge"])
if item.get("gt") is not None:
minimum = float(item["gt"]) + step
if item.get("le") is not None:
maximum = float(item["le"])
if item.get("lt") is not None:
maximum = float(item["lt"]) - step
widget.setRange(minimum, maximum)
@@ -0,0 +1,287 @@
"""Helpers for translating BEC scan metadata into ScanControl UI configuration."""
from __future__ import annotations
import re
from typing import Any
AnnotationValue = str | dict[str, Any] | list[Any] | None
ScanArgumentMetadata = dict[str, Any]
SignatureEntry = dict[str, Any]
ScanInputConfig = dict[str, Any]
ScanInfo = dict[str, Any]
ScanUIConfig = dict[str, Any]
SUPPORTED_SCAN_INPUT_TYPES = {"device", "DeviceBase", "float", "int", "bool", "str"}
class ScanInfoAdapter:
"""Normalize available-scan payloads into the structure consumed by ``ScanControl``."""
@staticmethod
def has_scan_ui_config(scan_info: ScanInfo) -> bool:
"""Check whether a scan exposes enough metadata to build a UI.
Args:
scan_info (ScanInfo): Available-scan payload for one scan.
Returns:
bool: ``True`` when a supported GUI metadata field is present.
"""
if not (
scan_info.get("gui_visibility")
or scan_info.get("gui_config")
or scan_info.get("gui_visualization")
or scan_info.get("signature")
):
return False
gui_config = ScanInfoAdapter().build_scan_ui_config(scan_info)
return not ScanInfoAdapter.unsupported_inputs(gui_config)
@staticmethod
def is_supported_input_type(input_type: AnnotationValue) -> bool:
"""Return whether ``ScanGroupBox`` has a widget for this serialized type."""
return (
isinstance(input_type, str)
and input_type in SUPPORTED_SCAN_INPUT_TYPES
or isinstance(input_type, dict)
and "Literal" in input_type
)
@staticmethod
def unsupported_inputs(gui_config: ScanUIConfig) -> list[ScanInputConfig]:
"""Return input configs that cannot be rendered by ``ScanGroupBox``."""
inputs = []
arg_group = gui_config.get("arg_group")
if arg_group:
inputs.extend(arg_group.get("inputs", []))
for group in gui_config.get("kwarg_groups", []):
inputs.extend(group.get("inputs", []))
return [
input_config
for input_config in inputs
if not ScanInfoAdapter.is_supported_input_type(input_config.get("type"))
]
@staticmethod
def format_display_name(name: str) -> str:
"""Convert a parameter name into a user-facing label.
Args:
name (str): Raw parameter name.
Returns:
str: Formatted display label such as ``Exp Time``.
"""
parts = re.split(r"(_|\d+)", name)
return " ".join(part.capitalize() for part in parts if part.isalnum()).strip()
@staticmethod
def resolve_tooltip(scan_argument: ScanArgumentMetadata) -> str | None:
"""Resolve the tooltip text from parsed ``ScanArgument`` metadata.
Args:
scan_argument (ScanArgumentMetadata): Parsed ``ScanArgument`` metadata.
Returns:
str | None: Explicit tooltip text if provided, otherwise the description fallback.
"""
return scan_argument.get("tooltip") or scan_argument.get("description")
@staticmethod
def parse_annotation(
annotation: AnnotationValue,
) -> tuple[AnnotationValue, ScanArgumentMetadata]:
"""Extract the serialized base annotation and ``ScanArgument`` metadata.
Args:
annotation (AnnotationValue): Serialized annotation payload from BEC.
Returns:
tuple[AnnotationValue, ScanArgumentMetadata]: The unwrapped annotation and parsed
``ScanArgument`` metadata.
"""
scan_argument: ScanArgumentMetadata = {}
if isinstance(annotation, list):
annotation = next(
(entry for entry in annotation if entry != "NoneType"),
annotation[0] if annotation else "_empty",
)
if isinstance(annotation, dict) and "Annotated" in annotation:
annotated = annotation["Annotated"]
annotation = annotated.get("type", "_empty")
scan_argument = annotated.get("metadata", {}).get("ScanArgument", {}) or {}
return annotation, scan_argument
@staticmethod
def scan_arg_type_from_annotation(annotation: AnnotationValue) -> AnnotationValue:
"""Normalize an annotation value to the widget type expected by ``ScanControl``.
Args:
annotation (AnnotationValue): Serialized or parsed annotation value.
Returns:
AnnotationValue: The normalized type identifier used by the widget layer.
"""
if isinstance(annotation, dict):
return annotation
if annotation in ("_empty", None):
return "str"
return annotation
def scan_input_from_signature(
self, param: SignatureEntry, arg: bool = False
) -> ScanInputConfig:
"""Build one ScanControl input description from a signature entry.
Args:
param (SignatureEntry): Serialized signature entry.
arg (bool): Whether the parameter belongs to the positional arg bundle.
Returns:
ScanInputConfig: Normalized input configuration for ``ScanControl``.
"""
annotation, scan_argument = self.parse_annotation(param.get("annotation"))
return self._build_scan_input(
name=param["name"],
annotation=annotation,
scan_argument=scan_argument,
arg=arg,
default=None if arg else param.get("default", None),
)
def scan_input_from_arg_input(
self, name: str, item_type: AnnotationValue, signature_by_name: dict[str, SignatureEntry]
) -> ScanInputConfig:
"""Build one arg-bundle input description from ``arg_input`` metadata.
Args:
name (str): Argument name from ``arg_input``.
item_type (AnnotationValue): Serialized argument type from ``arg_input``.
signature_by_name (dict[str, SignatureEntry]): Signature entries indexed by
parameter name.
Returns:
ScanInputConfig: Normalized input configuration for one arg-bundle field.
"""
if name in signature_by_name:
scan_input = self.scan_input_from_signature(signature_by_name[name], arg=True)
scan_input["type"] = self.scan_arg_type_from_annotation(
self.parse_annotation(signature_by_name[name].get("annotation"))[0]
)
else:
annotation, scan_argument = self.parse_annotation(item_type)
scan_input = self._build_scan_input(
name=name,
annotation=annotation,
scan_argument=scan_argument,
arg=True,
default=None,
)
if scan_input["type"] in ("_empty", None):
scan_input["type"] = item_type
return scan_input
def _build_scan_input(
self,
name: str,
annotation: AnnotationValue,
scan_argument: ScanArgumentMetadata,
*,
arg: bool,
default: Any,
) -> ScanInputConfig:
"""Build one normalized ScanControl input configuration.
Args:
name (str): Parameter name.
annotation (AnnotationValue): Parsed annotation value.
scan_argument (ScanArgumentMetadata): Parsed ``ScanArgument`` metadata.
arg (bool): Whether the parameter belongs to the positional arg bundle.
default (Any): Default value for the parameter.
Returns:
ScanInputConfig: Normalized input configuration.
"""
return {
"arg": arg,
"name": name,
"type": self.scan_arg_type_from_annotation(annotation),
"display_name": scan_argument.get("display_name") or self.format_display_name(name),
"tooltip": self.resolve_tooltip(scan_argument),
"default": default,
"expert": scan_argument.get("expert", False),
"hidden": scan_argument.get("hidden", False),
"precision": scan_argument.get("precision"),
"units": scan_argument.get("units"),
"reference_units": scan_argument.get("reference_units"),
"gt": scan_argument.get("gt"),
"ge": scan_argument.get("ge"),
"lt": scan_argument.get("lt"),
"le": scan_argument.get("le"),
"alternative_group": scan_argument.get("alternative_group"),
}
def build_scan_ui_config(self, scan_info: ScanInfo) -> ScanUIConfig:
"""Normalize one available-scan entry into the widget UI configuration.
Args:
scan_info (ScanInfo): Available-scan payload for one scan.
Returns:
ScanUIConfig: Legacy group structure consumed by ``ScanControl`` and
``ScanGroupBox``.
"""
gui_visualization = (
scan_info.get("gui_visualization") or scan_info.get("gui_visibility") or {}
)
if not gui_visualization and scan_info.get("gui_config"):
return scan_info["gui_config"]
signature = scan_info.get("signature", [])
signature_by_name = {entry["name"]: entry for entry in signature}
arg_group = None
arg_input = scan_info.get("arg_input", {})
if isinstance(arg_input, dict) and arg_input:
bundle_size = scan_info.get("arg_bundle_size", {})
inputs = [
self.scan_input_from_arg_input(name, item_type, signature_by_name)
for name, item_type in arg_input.items()
]
arg_group = {
"name": "Scan Arguments",
"bundle": bundle_size.get("bundle"),
"arg_inputs": arg_input,
"inputs": inputs,
"min": bundle_size.get("min"),
"max": bundle_size.get("max"),
}
kwarg_groups = []
arg_names = set(arg_input) if isinstance(arg_input, dict) else set()
visible_kwarg_names = set()
for group_name, input_names in gui_visualization.items():
inputs = []
for input_name in input_names:
if input_name in arg_names or input_name not in signature_by_name:
continue
if input_name in visible_kwarg_names:
continue
param = signature_by_name[input_name]
if param.get("kind") in ("VAR_POSITIONAL", "VAR_KEYWORD"):
continue
scan_input = self.scan_input_from_signature(param)
if scan_input.get("hidden"):
continue
inputs.append(scan_input)
visible_kwarg_names.add(input_name)
if inputs:
kwarg_groups.append({"name": group_name, "inputs": inputs})
return {
"scan_class_name": scan_info.get("class"),
"arg_group": arg_group,
"kwarg_groups": kwarg_groups,
}
@@ -120,7 +120,12 @@ class BecConsoleRegistry:
return None
window = console.window()
if window is not None and window is not console and self._is_valid_qobject(window):
if (
window is not None
and window is not console
and self._is_valid_qobject(window)
and not getattr(window, "_destroyed", False)
):
return window
if not avoid_console:
@@ -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.bundles import ToolbarBundle, ToolbarComponents
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 DeviceComboBox
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import (
BECDeviceFilter,
DeviceComboBox,
)
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.bundles import ToolbarComponents
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 DeviceComboBox
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import (
BECDeviceFilter,
DeviceComboBox,
)
from bec_widgets.widgets.utility.visual.colormap_widget.colormap_widget import BECColorMapWidget
@@ -58,7 +58,7 @@
</widget>
</item>
<item row="0" column="1">
<widget class="DeviceLineEdit" name="device_x"/>
<widget class="DeviceComboBox" name="device_x"/>
</item>
<item row="1" column="0">
<widget class="QLabel" name="label_2">
@@ -87,7 +87,7 @@
</widget>
</item>
<item row="0" column="1">
<widget class="DeviceLineEdit" name="device_y"/>
<widget class="DeviceComboBox" name="device_y"/>
</item>
<item row="1" column="0">
<widget class="QLabel" name="label_4">
@@ -116,7 +116,7 @@
</widget>
</item>
<item row="0" column="1">
<widget class="DeviceLineEdit" name="device_z"/>
<widget class="DeviceComboBox" name="device_z"/>
</item>
<item row="1" column="0">
<widget class="QLabel" name="label_6">
@@ -135,9 +135,9 @@
</widget>
<customwidgets>
<customwidget>
<class>DeviceLineEdit</class>
<extends>QLineEdit</extends>
<header>device_line_edit</header>
<class>DeviceComboBox</class>
<extends>QComboBox</extends>
<header>device_combo_box</header>
</customwidget>
<customwidget>
<class>ToggleSwitch</class>
@@ -154,7 +154,7 @@
<connections>
<connection>
<sender>device_x</sender>
<signal>textChanged(QString)</signal>
<signal>currentTextChanged(QString)</signal>
<receiver>signal_x</receiver>
<slot>clear()</slot>
<hints>
@@ -170,7 +170,7 @@
</connection>
<connection>
<sender>device_y</sender>
<signal>textChanged(QString)</signal>
<signal>currentTextChanged(QString)</signal>
<receiver>signal_y</receiver>
<slot>clear()</slot>
<hints>
@@ -186,7 +186,7 @@
</connection>
<connection>
<sender>device_z</sender>
<signal>textChanged(QString)</signal>
<signal>currentTextChanged(QString)</signal>
<receiver>signal_z</receiver>
<slot>clear()</slot>
<hints>
@@ -79,7 +79,7 @@ class CurveRow(QTreeWidgetItem):
Columns:
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
4: Style QComboBox
5: Pen width QSpinBox
@@ -152,6 +152,7 @@ class ScanProgressBar(BECWidget, QWidget):
self.connect_to_queue()
self._progress_source = None
self._progress_device = None
self.task = None
self.scan_number = None
self.progress_started.connect(lambda: print("Scan progress started"))
@@ -166,7 +167,7 @@ class ScanProgressBar(BECWidget, QWidget):
"""
Set the source of the progress.
"""
if self._progress_source == source:
if self._progress_source == source and self._progress_device == device:
self.update_source_label(source, device=device)
return
if self._progress_source is not None:
@@ -175,10 +176,11 @@ class ScanProgressBar(BECWidget, QWidget):
(
MessageEndpoints.scan_progress()
if self._progress_source == ProgressSource.SCAN_PROGRESS
else MessageEndpoints.device_progress(device=device)
else MessageEndpoints.device_progress(device=self._progress_device)
),
)
self._progress_source = source
self._progress_device = None if source == ProgressSource.SCAN_PROGRESS else device
self.bec_dispatcher.connect_slot(
self.on_progress_update,
(
@@ -316,11 +318,16 @@ class ScanProgressBar(BECWidget, QWidget):
(
MessageEndpoints.scan_progress()
if self._progress_source == ProgressSource.SCAN_PROGRESS
else MessageEndpoints.device_progress(device=self._progress_source.value)
else MessageEndpoints.device_progress(device=self._progress_device)
),
)
self._progress_source = None
self._progress_device = None
self.progressbar.close()
self.progressbar.deleteLater()
self.bec_dispatcher.disconnect_slot(
self.on_queue_update, MessageEndpoints.scan_queue_status()
)
super().cleanup()
@@ -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.error_popups import SafeProperty, SafeSlot
from bec_widgets.utils.ophyd_kind_util import Kind
from bec_widgets.widgets.control.device_input.device_line_edit.device_line_edit import (
DeviceLineEdit,
)
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
if TYPE_CHECKING:
@@ -58,7 +56,7 @@ class ChoiceDialog(QDialog):
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)
layout.addWidget(self._device_field)
layout.addWidget(self._signal_field)
@@ -73,10 +71,13 @@ class ChoiceDialog(QDialog):
self._signal_field.include_config_signals = show_config
self.setLayout(layout)
self._device_field.textChanged.connect(self._update_device)
self._device_field.currentTextChanged.connect(self._update_device)
if 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)
def _display_error(self):
@@ -97,19 +98,19 @@ class ChoiceDialog(QDialog):
self._device_field.set_device(device)
self._signal_field.set_device(device)
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)
else:
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._signal_field.clear()
def accept(self):
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()
return super().accept()
@@ -170,7 +171,7 @@ class SignalLabel(BECWidget, QWidget):
client (BECClient, optional): The BEC client. Defaults to None.
device (str, optional): The device 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_default_units (bool, optional): Whether to show default units. Defaults to False.
custom_label (str, optional): Custom label for the widget. Defaults to "".
+35 -9
View File
@@ -1,6 +1,6 @@
import sys
from qtpy.QtCore import Property, QEasingCurve, QPointF, QPropertyAnimation, Qt, Signal
from qtpy.QtCore import Property, QEasingCurve, QEvent, QPointF, QPropertyAnimation, Qt, Signal
from qtpy.QtGui import QColor, QPainter
from qtpy.QtWidgets import QApplication, QWidget
@@ -41,10 +41,22 @@ class ToggleSwitch(QWidget):
theme = getattr(QApplication.instance(), "theme", None)
colors = theme.colors if theme else {}
self._active_track_color = colors.get("PRIMARY", QColor(33, 150, 243))
self._active_thumb_color = colors.get("ON_PRIMARY", QColor(255, 255, 255))
self._inactive_track_color = colors.get("SEPARATOR", QColor(200, 200, 200))
self._inactive_thumb_color = colors.get("ON_PRIMARY", QColor(255, 255, 255))
self._active_track_color = self._theme_color(colors, "PRIMARY", QColor(33, 150, 243))
self._active_thumb_color = self._theme_color(colors, "ON_PRIMARY", QColor(255, 255, 255))
self._inactive_track_color = self._theme_color(colors, "SEPARATOR", QColor(200, 200, 200))
self._inactive_thumb_color = self._theme_color(colors, "ON_PRIMARY", QColor(255, 255, 255))
self._disabled_track_color = self._theme_color(colors, "DISABLED_BG", QColor(220, 220, 220))
self._disabled_thumb_color = self._theme_color(colors, "DISABLED_FG", QColor(150, 150, 150))
self._disabled_border_color = self._theme_color(
colors, "DISABLED_BORDER", QColor(170, 170, 170)
)
if hasattr(self, "_checked"):
self.update_colors()
@staticmethod
def _theme_color(colors: dict, key: str, fallback: QColor) -> QColor:
color = colors.get(key, fallback)
return color if isinstance(color, QColor) else QColor(color)
@Property(bool)
def checked(self):
@@ -119,29 +131,40 @@ class ToggleSwitch(QWidget):
def paintEvent(self, event):
painter = QPainter(self)
painter.setRenderHint(QPainter.Antialiasing)
painter.setRenderHint(QPainter.RenderHint.Antialiasing)
# Draw track
painter.setBrush(self._track_color)
painter.setPen(Qt.NoPen)
painter.setPen(self._disabled_border_color if not self.isEnabled() else Qt.PenStyle.NoPen)
painter.drawRoundedRect(
0, 0, self.width(), self.height(), self.height() / 2, self.height() / 2
)
# Draw thumb
painter.setBrush(self._thumb_color)
painter.setPen(Qt.PenStyle.NoPen)
diameter = int(self.height() * 0.8)
painter.drawEllipse(int(self._thumb_pos.x()), int(self._thumb_pos.y()), diameter, diameter)
def mousePressEvent(self, event):
if event.button() == Qt.LeftButton:
if self.isEnabled() and event.button() == Qt.MouseButton.LeftButton:
self.checked = not self.checked
def update_colors(self):
if not self.isEnabled():
self._thumb_color = self._disabled_thumb_color
self._track_color = self._disabled_track_color
return
self._thumb_color = self.active_thumb_color if self._checked else self.inactive_thumb_color
self._track_color = self.active_track_color if self._checked else self.inactive_track_color
def changeEvent(self, event):
if event.type() == QEvent.Type.EnabledChange:
self.update_colors()
self.update()
super().changeEvent(event)
def get_thumb_pos(self, checked):
return QPointF(self.width() - self.height() + 3, 2) if checked else QPointF(3, 2)
@@ -167,7 +190,7 @@ class ToggleSwitch(QWidget):
if __name__ == "__main__": # pragma: no cover
from qtpy.QtWidgets import QHBoxLayout, QVBoxLayout, QWidget
from qtpy.QtWidgets import QHBoxLayout, QWidget
from bec_widgets.utils.colors import apply_theme
from bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button import DarkModeButton
@@ -177,9 +200,12 @@ if __name__ == "__main__": # pragma: no cover
widget = QWidget()
layout = QHBoxLayout(widget)
toggle = ToggleSwitch()
toggle_disabled = ToggleSwitch()
dark_mode_btn = DarkModeButton()
layout.addWidget(toggle)
layout.addWidget(toggle_disabled)
layout.addWidget(dark_mode_btn)
toggle_disabled.setEnabled(False)
window = QWidget()
window.setLayout(layout)
window.show()
+14 -3
View File
@@ -1,6 +1,6 @@
[project]
name = "bec_widgets"
version = "3.9.0"
version = "3.13.3"
description = "BEC Widgets"
requires-python = ">=3.11"
classifiers = [
@@ -10,8 +10,8 @@ classifiers = [
]
dependencies = [
"PyJWT~=2.9",
"PySide6==6.9.0",
"PySide6-QtAds==4.4.0",
"PySide6==6.11.0",
"PySide6-QtAds==4.5.0.4",
"bec_ipython_client~=3.107,>=3.107.2", # needed for jupyter console
"bec_lib~=3.107,>=3.107.2",
"bec_qthemes~=1.0, >=1.3.4",
@@ -60,6 +60,17 @@ qtermwidget = ["pyside6_qtermwidget"]
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
+1 -1
View File
@@ -45,7 +45,7 @@ def connected_client_gui_obj(qtbot, gui_id, bec_client_lib):
"""
gui = BECGuiClient(gui_id=gui_id)
try:
gui.start(wait=True)
gui.show(wait=True)
qtbot.waitUntil(lambda: hasattr(gui, "bec"), timeout=5000)
gui.bec.delete_all() # ensure clean state
qtbot.waitUntil(lambda: len(gui.bec.widget_list()) == 0, timeout=10000)
+4 -4
View File
@@ -143,11 +143,11 @@ def test_rpc_gui_obj(connected_client_gui_obj: BECGuiClient, qtbot):
qtbot.wait(500)
gui.kill_server()
assert not gui._gui_is_alive()
gui.start(wait=True)
gui.show(wait=True)
assert gui._gui_is_alive()
# calling start multiple times should not change anything
gui.start(wait=True)
gui.start(wait=True)
# calling show multiple times should not change anything
gui.show(wait=True)
gui.show(wait=True)
def wait_for_gui_started():
return "bec" in gui.windows
+1 -1
View File
@@ -75,7 +75,7 @@ def connected_client_gui_obj(qtbot_scope_module, gui_id, bec_client_lib):
"""
gui = BECGuiClient(gui_id=gui_id)
try:
gui.start(wait=True)
gui.show(wait=True)
qtbot_scope_module.waitUntil(lambda: hasattr(gui, "bec"), timeout=5000)
gui.bec.delete_all() # ensure clean state
qtbot_scope_module.waitUntil(lambda: len(gui.bec.widget_list()) == 0, timeout=10000)
+30 -1
View File
@@ -2,7 +2,10 @@ from importlib.machinery import FileFinder, SourceFileLoader
from types import ModuleType
from unittest import mock
from bec_widgets.utils.bec_plugin_helper import _all_widgets_from_all_submods
from bec_widgets.utils.bec_plugin_helper import (
_all_widgets_from_all_submods,
get_plugin_widget_icons,
)
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.plugin_utils import BECClassContainer, BECClassInfo
@@ -69,3 +72,29 @@ def test_all_widgets_from_module_no_widgets():
widgets = _all_widgets_from_all_submods(module).as_dict()
assert widgets == {}
def test_get_plugin_widget_icons_from_designer_module():
designer_module = mock.MagicMock(spec=ModuleType)
designer_module.widget_icons = {"PluginWidget": "star"}
get_plugin_widget_icons.cache_clear()
try:
with mock.patch(
"bec_widgets.utils.bec_plugin_helper.get_plugin_designer_module",
return_value=designer_module,
):
assert get_plugin_widget_icons() == {"PluginWidget": "star"}
finally:
get_plugin_widget_icons.cache_clear()
def test_get_plugin_widget_icons_without_designer_module():
get_plugin_widget_icons.cache_clear()
try:
with mock.patch(
"bec_widgets.utils.bec_plugin_helper.get_plugin_designer_module", return_value=None
):
assert get_plugin_widget_icons() == {}
finally:
get_plugin_widget_icons.cache_clear()
+10 -1
View File
@@ -5,6 +5,7 @@ import pytest
from bec_widgets.cli.client import BECDockArea
from bec_widgets.cli.client_utils import BECGuiClient, _start_plot_process
from bec_widgets.cli.rpc.rpc_base import RPCBase, RPCResponseTimeoutError, rpc_timeout
@pytest.fixture
@@ -220,7 +221,7 @@ def test_client_utils_new_starts_server_when_not_alive():
with mock.patch("bec_widgets.cli.client_utils.wait_for_server", _no_wait_for_server):
with (
mock.patch.object(gui, "_check_if_server_is_alive", return_value=False),
mock.patch.object(gui, "start") as mock_start,
mock.patch.object(gui, "show") as mock_start,
):
gui.new(wait=False, startup_profile=None)
@@ -257,3 +258,11 @@ def test_client_utils_delete_falls_back_to_direct_close():
gui.delete("dock")
widget._run_rpc.assert_called_once_with("close")
def test_client_utils_gui_client_set_rpc_timeout():
gui = BECGuiClient()
assert gui._rpc_timeout == 5
gui.set_rpc_timeout(10)
assert gui._rpc_timeout == 10
+4
View File
@@ -58,6 +58,10 @@ def test_curve_setting_init(curve_setting_fixture):
# Check that there's a curve_manager inside y_axis_box
assert hasattr(curve_setting, "curve_manager")
assert curve_setting.y_axis_box.layout.count() > 0
assert not curve_setting.device_x.isEnabled()
assert not curve_setting.signal_x.isEnabled()
assert "red" not in curve_setting.device_x.styleSheet()
assert "red" not in curve_setting.signal_x.styleSheet()
def test_curve_setting_accept_changes(curve_setting_fixture, qtbot):
+70 -116
View File
@@ -1,152 +1,106 @@
from unittest import mock
import pytest
from bec_lib.device import ReadoutPriority
from qtpy.QtWidgets import QWidget
from bec_lib.device import Positioner, ReadoutPriority
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,
DeviceInputBase,
DeviceComboBox,
DeviceInputConfig,
)
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox
from .client_mocks import mocked_client
from .conftest import create_widget
# DeviceInputBase is meant to be mixed in a QWidget
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"""
def test_device_combobox_init_with_config(qtbot, mocked_client):
config = {
"widget_class": "DeviceInputWidget",
"widget_class": "DeviceComboBox",
"gui_id": "test_gui_id",
"device_filter": [BECDeviceFilter.POSITIONER],
"device_filter": [BECDeviceFilter.POSITIONER.value],
"default": "samx",
}
widget = DeviceInputWidget(client=mocked_client, config=config)
widget2 = DeviceInputWidget(
client=mocked_client, config=DeviceInputConfig.model_validate(config)
widget = create_widget(
qtbot=qtbot,
widget=DeviceComboBox,
client=mocked_client,
config=DeviceInputConfig.model_validate(config),
)
qtbot.addWidget(widget)
qtbot.addWidget(widget2)
qtbot.waitExposed(widget)
qtbot.waitExposed(widget2)
for w in [widget, widget2]:
assert w.config.gui_id == "test_gui_id"
assert w.config.device_filter == ["Positioner"]
assert w.config.default == "samx"
assert widget.config.gui_id == "test_gui_id"
assert widget.config.device_filter == ["Positioner"]
assert widget.config.default == "samx"
def test_device_input_base_set_device_filter(device_input_base):
"""Test device filter setter."""
device_input_base.set_device_filter(BECDeviceFilter.POSITIONER)
assert device_input_base.config.device_filter == ["Positioner"]
def test_device_combobox_set_device_filter(qtbot, mocked_client):
widget = create_widget(qtbot=qtbot, widget=DeviceComboBox, client=mocked_client)
widget.set_device_filter(BECDeviceFilter.POSITIONER)
assert widget.config.device_filter == ["Positioner"]
def test_device_input_base_set_device_filter_error(device_input_base):
"""Test set_device_filter with Noneexisting class. This should not raise. It writes a log message entry."""
device_input_base.set_device_filter("NonExistingClass")
assert device_input_base.device_filter == []
def test_device_combobox_set_device_filter_error(qtbot, mocked_client):
widget = create_widget(qtbot=qtbot, widget=DeviceComboBox, client=mocked_client)
widget.set_device_filter("NonExistingClass")
assert widget.device_filter == []
def test_device_input_base_set_default_device(device_input_base):
"""Test setting the default device. Also tests the update_devices method."""
device_input_base.set_device("samx")
assert device_input_base.config.default == None
device_input_base.set_device_filter(BECDeviceFilter.POSITIONER)
device_input_base.set_readout_priority_filter(ReadoutPriority.MONITORED)
device_input_base.set_device("samx")
assert device_input_base.config.default == "samx"
def test_device_combobox_set_default_device(qtbot, mocked_client):
widget = create_widget(qtbot=qtbot, widget=DeviceComboBox, client=mocked_client)
widget.set_device("samx")
assert widget.config.default == "samx"
def test_device_input_base_get_filters(device_input_base):
"""Test getting the available filters."""
filters = device_input_base.get_available_filters()
selection = [
BECDeviceFilter.POSITIONER,
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_combobox_properties(qtbot, mocked_client):
widget = create_widget(qtbot=qtbot, widget=DeviceComboBox, client=mocked_client)
def test_device_input_base_properties(device_input_base):
"""Test setting the properties of the device input base."""
assert device_input_base.device_filter == []
device_input_base.filter_to_device = True
assert device_input_base.device_filter == [BECDeviceFilter.DEVICE]
device_input_base.filter_to_positioner = True
assert device_input_base.device_filter == [BECDeviceFilter.DEVICE, BECDeviceFilter.POSITIONER]
device_input_base.filter_to_computed_signal = True
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 == [
widget.filter_to_device = True
widget.filter_to_positioner = True
widget.filter_to_computed_signal = True
widget.filter_to_signal = True
assert widget.device_filter == [
BECDeviceFilter.DEVICE,
BECDeviceFilter.POSITIONER,
BECDeviceFilter.COMPUTED_SIGNAL,
BECDeviceFilter.SIGNAL,
]
assert device_input_base.readout_filter == []
device_input_base.readout_async = True
assert device_input_base.readout_filter == [ReadoutPriority.ASYNC]
device_input_base.readout_baseline = True
assert device_input_base.readout_filter == [ReadoutPriority.ASYNC, ReadoutPriority.BASELINE]
device_input_base.readout_monitored = True
assert device_input_base.readout_filter == [
ReadoutPriority.ASYNC,
ReadoutPriority.BASELINE,
ReadoutPriority.MONITORED,
]
device_input_base.readout_on_request = True
assert device_input_base.readout_filter == [
ReadoutPriority.ASYNC,
ReadoutPriority.BASELINE,
ReadoutPriority.MONITORED,
ReadoutPriority.ON_REQUEST,
]
widget.readout_async = False
assert ReadoutPriority.ASYNC not in widget.readout_filter
widget.readout_async = True
assert ReadoutPriority.ASYNC in widget.readout_filter
def test_device_combobox_multiple_device_filters_match_main_intersection(qtbot, mocked_client):
widget = create_widget(qtbot=qtbot, widget=DeviceComboBox, client=mocked_client)
widget.filter_to_device = True
widget.filter_to_positioner = True
assert widget.devices
assert all(isinstance(getattr(widget.dev, device), Positioner) for device in widget.devices)
assert "eiger" not in widget.devices
def test_device_combobox_empty_readout_filter_matches_main_empty_selection(qtbot, mocked_client):
widget = create_widget(qtbot=qtbot, widget=DeviceComboBox, client=mocked_client)
widget.readout_monitored = False
widget.readout_baseline = False
widget.readout_async = False
widget.readout_continuous = False
widget.readout_on_request = False
assert widget.readout_filter == []
assert widget.devices == []
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(
return_value=[
("samx", "async_signal", {"signal_class": "AsyncSignal"}),
+95 -76
View File
@@ -1,10 +1,11 @@
from unittest import mock
import pytest
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 DeviceComboBox
from bec_widgets.widgets.control.device_input.device_line_edit.device_line_edit import (
DeviceLineEdit,
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import (
BECDeviceFilter,
DeviceComboBox,
)
from .client_mocks import mocked_client
@@ -37,6 +38,21 @@ def test_device_input_combobox_init(device_input_combobox):
assert device_input_combobox.client is not None
assert isinstance(device_input_combobox, 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 == [
"samx",
"samy",
@@ -65,81 +81,84 @@ 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"
def test_device_input_combobox_autocomplete(qtbot, mocked_client):
widget = DeviceComboBox(client=mocked_client, autocomplete=True)
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
line_edit = widget.lineEdit()
text_changes: list[str] = []
line_edit.setPlaceholderText("Select Device")
line_edit.textChanged.connect(text_changes.append)
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()
assert widget.lineEdit().placeholderText() == "Select Device"
widget.lineEdit().setText("manual_device")
assert text_changes[-1] == "manual_device"
def test_device_input_combobox_refresh_preserves_manual_text(device_input_combobox):
device_input_combobox.setCurrentText("manual_device")
device_input_combobox.update_devices_from_filters()
assert device_input_combobox.currentText() == "manual_device"
assert device_input_combobox.is_valid_input is False
def test_device_input_combobox_disabled_invalid_has_neutral_border(device_input_combobox):
device_input_combobox.setCurrentText("manual_device")
assert "red" in device_input_combobox.styleSheet()
device_input_combobox.setEnabled(False)
assert "transparent" in device_input_combobox.styleSheet()
device_input_combobox.setEnabled(True)
assert "red" in device_input_combobox.styleSheet()
def test_device_input_combobox_cleanup_unregisters_callback(qtbot, mocked_client):
with mock.patch.object(mocked_client.callbacks, "remove"):
widget = DeviceComboBox(client=mocked_client)
qtbot.addWidget(widget)
callback_id = widget._callback_id
widget.close()
widget.deleteLater()
mocked_client.callbacks.remove.assert_called_once_with(callback_id)
assert widget._callback_id is None
def test_device_input_combobox_cleanup_clears_callback_before_unregister(qtbot, mocked_client):
widget = DeviceComboBox(client=mocked_client)
qtbot.addWidget(widget)
callback_id = widget._callback_id
def assert_callback_cleared(removed_callback_id):
assert removed_callback_id == callback_id
assert widget._callback_id is None
with mock.patch.object(
mocked_client.callbacks, "remove", side_effect=assert_callback_cleared
) as remove_mock:
widget.cleanup()
remove_mock.assert_called_once_with(callback_id)
def test_get_device_from_input_combobox_init(device_input_combobox):
device_input_combobox.setCurrentIndex(0)
device_text = device_input_combobox.currentText()
current_device = device_input_combobox.get_current_device()
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
+259 -77
View File
@@ -2,17 +2,15 @@ from unittest import mock
import pytest
from bec_lib.device import Signal
from qtpy.QtWidgets import QWidget
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.base_classes.device_signal_input_base import (
DeviceSignalInputBase,
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import (
BECDeviceFilter,
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_line_edit.signal_line_edit import (
SignalLineEdit,
from bec_widgets.widgets.control.device_input.signal_combobox.signal_combobox import (
SignalComboBox,
SignalComboBoxConfig,
)
from .client_mocks import mocked_client
@@ -20,36 +18,19 @@ from .conftest import create_widget
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
def signal_names(signals):
return [entry[0] if isinstance(entry, tuple) else entry for entry in signals]
@pytest.fixture
def device_signal_combobox(qtbot, mocked_client):
"""Fixture with mocked FilterIO and WidgetIO"""
widget = create_widget(qtbot=qtbot, widget=SignalComboBox, client=mocked_client)
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
def test_device_signal_combo(qtbot, mocked_client):
"""Fixture to create a SignalComboBox widget and a DeviceInputWidget widget"""
@@ -63,34 +44,82 @@ def test_device_signal_combo(qtbot, mocked_client):
yield input, signal
def test_device_signal_base_init(device_signal_base):
"""Test if the DeviceSignalInputBase is initialized correctly"""
assert device_signal_base._device is None
assert device_signal_base._signal_filter == set()
assert device_signal_base._signals == []
assert device_signal_base._hinted_signals == []
assert device_signal_base._normal_signals == []
assert device_signal_base._config_signals == []
def test_signal_combobox_init(device_signal_combobox):
assert device_signal_combobox._device is None
assert device_signal_combobox._signal_filter == {Kind.config, Kind.normal, Kind.hinted}
assert device_signal_combobox._signals == []
assert device_signal_combobox._hinted_signals == []
assert device_signal_combobox._normal_signals == []
assert device_signal_combobox._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):
"""Test if the DeviceSignalInputBase has the correct QProperties"""
assert device_signal_base._signal_filter == set()
device_signal_base.include_config_signals = False
device_signal_base.include_normal_signals = False
assert device_signal_base._signal_filter == set()
device_signal_base.include_config_signals = True
assert device_signal_base._signal_filter == {Kind.config}
device_signal_base.include_normal_signals = True
assert device_signal_base._signal_filter == {Kind.config, Kind.normal}
device_signal_base.include_hinted_signals = True
assert device_signal_base._signal_filter == {Kind.config, Kind.normal, Kind.hinted}
device_signal_base.include_hinted_signals = True
assert device_signal_base._signal_filter == {Kind.config, Kind.normal, Kind.hinted}
device_signal_base.include_hinted_signals = True
assert device_signal_base._signal_filter == {Kind.config, Kind.normal, Kind.hinted}
device_signal_base.include_hinted_signals = False
assert device_signal_base._signal_filter == {Kind.config, Kind.normal}
def test_signal_combobox_config_defaults_are_independent_lists():
config_a = SignalComboBoxConfig(widget_class="SignalComboBox")
config_b = SignalComboBoxConfig(widget_class="SignalComboBox")
config_a.signal_filter.append("hinted")
config_a.signal_class_filter.append("AsyncSignal")
config_a.signals.append("sig")
assert config_b.signal_filter == []
assert config_b.signal_class_filter == []
assert config_b.signals == []
def test_signal_combobox_autocomplete(qtbot, mocked_client):
widget = create_widget(
qtbot=qtbot, widget=SignalComboBox, client=mocked_client, autocomplete=True
)
line_edit = widget.lineEdit()
text_changes: list[str] = []
line_edit.setPlaceholderText("Select Signal")
line_edit.textChanged.connect(text_changes.append)
widget.set_device("samx")
assert widget.autocomplete is True
assert widget.completer() is not None
assert widget.completer().model().stringList() == ["samx (readback)", "setpoint", "velocity"]
assert widget.completer().model() != widget.model()
widget.autocomplete = False
assert widget.completer() is not None
assert widget.completer().model() == widget.model()
assert widget.lineEdit().placeholderText() == "Select Signal"
widget.lineEdit().setText("manual_signal")
assert text_changes[-1] == "manual_signal"
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_disabled_invalid_has_neutral_border(device_signal_combobox):
device_signal_combobox.setCurrentText("manual_signal")
assert "red" in device_signal_combobox.styleSheet()
device_signal_combobox.setEnabled(False)
assert "transparent" in device_signal_combobox.styleSheet()
device_signal_combobox.setEnabled(True)
assert "red" in device_signal_combobox.styleSheet()
def test_signal_combobox(qtbot, device_signal_combobox):
@@ -128,30 +157,87 @@ def test_signal_combobox(qtbot, device_signal_combobox):
assert device_signal_combobox._hinted_signals == [("fake_signal", {})]
def test_signal_lineedit(device_signal_line_edit):
"""Test the signal_combobox"""
def test_linked_device_combobox_updates_signal_combobox_on_each_text_change(
qtbot, test_device_signal_combo
):
device, signal = test_device_signal_combo
device.currentTextChanged.connect(signal.set_device)
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
emitted_device_texts: list[str] = []
device.currentTextChanged.connect(emitted_device_texts.append)
device.setCurrentText("samx")
assert signal.device == "samx"
assert signal.currentText() == "samx (readback)"
device.setCurrentText("sa")
assert emitted_device_texts[-1] == "sa"
assert signal.device == ""
assert signal.signals == []
assert signal.currentText() == ""
assert signal.is_valid_input is False
device.setCurrentText("samx")
assert emitted_device_texts[-1] == "samx"
assert signal.device == "samx"
assert [entry[0] for entry in signal.signals] == ["samx (readback)", "setpoint", "velocity"]
def test_device_signal_input_base_cleanup(qtbot, mocked_client):
with mock.patch.object(mocked_client.callbacks, "remove"):
widget = DeviceInputWidget(client=mocked_client)
widget = SignalComboBox(client=mocked_client)
callback_id = widget._device_update_register
widget.close()
widget.deleteLater()
mocked_client.callbacks.remove.assert_called_once_with(widget._device_update_register)
mocked_client.callbacks.remove.assert_called_once_with(callback_id)
assert widget._device_update_register is None
def test_signal_combobox_cleanup_clears_callback_before_unregister(qtbot, mocked_client):
widget = create_widget(qtbot=qtbot, widget=SignalComboBox, client=mocked_client)
callback_id = widget._device_update_register
def assert_callback_cleared(removed_callback_id):
assert removed_callback_id == callback_id
assert widget._device_update_register is None
with mock.patch.object(
mocked_client.callbacks, "remove", side_effect=assert_callback_cleared
) as remove_mock:
widget.cleanup()
remove_mock.assert_called_once_with(callback_id)
def test_signal_combobox_cleanup_blocks_in_flight_device_update(qtbot, mocked_client):
widget = create_widget(qtbot=qtbot, widget=SignalComboBox, client=mocked_client)
callback_id = widget._device_update_register
def trigger_in_flight_update(_):
widget.update_signals_from_filters("reload", {})
with (
mock.patch.object(
mocked_client.callbacks, "remove", side_effect=trigger_in_flight_update
) as remove_mock,
mock.patch.object(widget, "_set_signal_groups") as set_signal_groups,
):
widget.cleanup()
remove_mock.assert_called_once_with(callback_id)
set_signal_groups.assert_not_called()
def test_signal_combobox_device_update_ignores_update_action(qtbot, mocked_client):
widget = create_widget(qtbot=qtbot, widget=SignalComboBox, client=mocked_client)
with mock.patch.object(widget, "_set_signal_groups") as set_signal_groups:
widget.update_signals_from_filters("update", {})
set_signal_groups.assert_not_called()
def test_signal_combobox_get_signal_name_with_item_data(qtbot, device_signal_combobox):
@@ -252,11 +338,107 @@ def test_signal_combobox_signal_class_filter_by_device(qtbot, mocked_client):
device="samx",
)
assert widget.signals == ["samx_readback_async"]
assert signal_names(widget.signals) == ["samx_readback_async"]
assert widget.signal_class_filter == ["AsyncSignal"]
widget.set_device("samy")
assert widget.signals == ["samy_readback_async"]
assert signal_names(widget.signals) == ["samy_readback_async"]
def test_signal_combobox_signal_class_filter_selects_by_metadata(qtbot, mocked_client):
"""Class-based signal lists should support obj_name/component_name lookup."""
mocked_client.device_manager.get_bec_signals = mock.MagicMock(
return_value=[
(
"eiger",
"image",
{
"obj_name": "eiger_image",
"component_name": "det.image",
"signal_class": "PreviewSignal",
"describe": {"signal_info": {"ndim": 2}},
},
)
]
)
widget = create_widget(
qtbot=qtbot,
widget=SignalComboBox,
client=mocked_client,
signal_class_filter=["PreviewSignal"],
ndim_filter=[2],
device="eiger",
)
assert widget.validate_signal("eiger_image") is True
assert widget.validate_signal("det.image") is True
assert widget.set_to_obj_name("eiger_image") is True
assert widget.currentText() == "image"
widget.set_signal("det.image")
assert widget.currentText() == "image"
def test_signal_combobox_signal_class_update_revalidates_selected_signal(qtbot, mocked_client):
"""Signal-class rebuilds should validate after items and signal metadata are in sync."""
mocked_client.device_manager.get_bec_signals = mock.MagicMock(
return_value=[
(
"eiger",
"img",
{
"obj_name": "img",
"signal_class": "PreviewSignal",
"describe": {"signal_info": {"ndim": 2}},
},
)
]
)
widget = create_widget(
qtbot=qtbot,
widget=SignalComboBox,
client=mocked_client,
signal_class_filter=["PreviewSignal"],
ndim_filter=[2],
require_device=True,
)
widget.set_device("eiger")
assert widget.currentText() == "img"
assert widget.is_valid_input is True
def test_signal_combobox_signal_class_refresh_preserves_manual_text(qtbot, mocked_client):
mocked_client.device_manager.get_bec_signals = mock.MagicMock(
return_value=[
(
"eiger",
"img",
{
"obj_name": "img",
"signal_class": "PreviewSignal",
"describe": {"signal_info": {"ndim": 2}},
},
)
]
)
widget = create_widget(
qtbot=qtbot,
widget=SignalComboBox,
client=mocked_client,
signal_class_filter=["PreviewSignal"],
ndim_filter=[2],
require_device=True,
)
widget.set_device("eiger")
widget.setCurrentText("manual_signal")
widget.update_signals_from_signal_classes()
assert widget.currentText() == "manual_signal"
assert widget.is_valid_input is False
def test_signal_class_filter_setter_clears_to_kind_filters(qtbot, mocked_client):
@@ -271,7 +453,7 @@ def test_signal_class_filter_setter_clears_to_kind_filters(qtbot, mocked_client)
signal_class_filter=["AsyncSignal"],
device="samx",
)
assert widget.signals == ["samx_readback_async"]
assert signal_names(widget.signals) == ["samx_readback_async"]
widget.signal_class_filter = []
samx = widget.dev.samx
@@ -294,7 +476,7 @@ def test_signal_class_filter_setter_none_reverts_to_kind_filters(qtbot, mocked_c
signal_class_filter=["AsyncSignal"],
device="samx",
)
assert widget.signals == ["samx_readback_async"]
assert signal_names(widget.signals) == ["samx_readback_async"]
widget.signal_class_filter = None
samx = widget.dev.samx
@@ -361,12 +543,12 @@ def test_signal_combobox_class_kind_ndim_filters(qtbot, mocked_client):
)
# Default kinds are hinted + normal, ndim=1, device=samx
assert widget.signals == ["sig1"]
assert signal_names(widget.signals) == ["sig1"]
# Enable config kinds and widen ndim to include sig2
widget.include_config_signals = True
widget.ndim_filter = 2
assert widget.signals == ["sig2"]
assert signal_names(widget.signals) == ["sig2"]
def test_signal_combobox_require_device_validation(qtbot, mocked_client):
@@ -394,7 +576,7 @@ def test_signal_combobox_require_device_validation(qtbot, mocked_client):
assert widget.signals == []
widget.set_device("samx")
assert widget.signals == ["sig1"]
assert signal_names(widget.signals) == ["sig1"]
resets: list[str] = []
widget.signal_reset.connect(lambda: resets.append("reset"))
+66
View File
@@ -11,6 +11,7 @@ from qtpy.QtGui import QPixmap
from qtpy.QtWidgets import QDialog, QMessageBox, QWidget
import bec_widgets.widgets.containers.dock_area.basic_dock_area as basic_dock_module
import bec_widgets.widgets.containers.dock_area.dock_area as dock_area_module
import bec_widgets.widgets.containers.dock_area.profile_utils as profile_utils
from bec_widgets.widgets.containers.dock_area.basic_dock_area import (
DockAreaWidget,
@@ -68,6 +69,13 @@ def temp_profile_dir():
return os.environ["BECWIDGETS_PROFILE_DIR"]
@pytest.fixture
def clear_plugin_toolbar_actions_cache():
dock_area_module._plugin_toolbar_actions.cache_clear()
yield
dock_area_module._plugin_toolbar_actions.cache_clear()
@pytest.fixture
def module_profile_factory(monkeypatch, tmp_path):
"""Provide a helper to create synthetic module-level (read-only) profiles."""
@@ -985,6 +993,64 @@ class TestToolbarFunctionality:
# Verify save was called with the filename
mock_screenshot.save.assert_called_once_with(str(screenshot_path))
def test_plugin_toolbar_actions_empty_when_no_plugins(self, clear_plugin_toolbar_actions_cache):
"""Test that no plugin toolbar actions are produced when no plugin widgets exist."""
with patch(
"bec_widgets.widgets.containers.dock_area.dock_area.get_plugin_rpc_widget_registry",
return_value={},
):
plugin_actions = dock_area_module._plugin_toolbar_actions()
assert plugin_actions == {}
def test_plugin_toolbar_actions_include_available_plugins(
self, clear_plugin_toolbar_actions_cache
):
"""Test that plugin toolbar actions are built from RPC widgets and generated icons."""
plugin_registry = {
"FakePluginWidget": ("fake_plugin.widgets.fake_plugin_widget", "FakePluginWidget")
}
with (
patch(
"bec_widgets.widgets.containers.dock_area.dock_area.get_plugin_rpc_widget_registry",
return_value=plugin_registry,
),
patch(
"bec_widgets.widgets.containers.dock_area.dock_area.get_plugin_widget_icons",
return_value={"FakePluginWidget": "star"},
),
):
plugin_actions = dock_area_module._plugin_toolbar_actions()
assert plugin_actions == {
"FakePluginWidget": ("star", "Add FakePluginWidget", "FakePluginWidget")
}
def test_plugin_toolbar_actions_ignore_builtin_name_collisions(
self, clear_plugin_toolbar_actions_cache
):
"""Test that plugin widgets shadowed by built-ins are not added to the plugin menu."""
plugin_registry = {"Waveform": ("fake_plugin.widgets.waveform", "Waveform")}
with patch(
"bec_widgets.widgets.containers.dock_area.dock_area.get_plugin_rpc_widget_registry",
return_value=plugin_registry,
):
plugin_actions = dock_area_module._plugin_toolbar_actions()
assert plugin_actions == {}
def test_new_plugin_widget_passes_toolbar_icon_to_new(self):
"""Test that plugin widget creation passes the toolbar icon to dock creation."""
dock_area = MagicMock()
toolbar_action = MagicMock()
dock_icon = object()
toolbar_action.get_icon.return_value = dock_icon
BECDockArea._new_plugin_widget(dock_area, "FakePluginWidget", toolbar_action)
toolbar_action.get_icon.assert_called_once_with()
dock_area.new.assert_called_once_with(widget="FakePluginWidget", dock_icon=dock_icon)
class TestDockSettingsDialog:
"""Test dock settings dialog functionality."""
+39 -55
View File
@@ -1,74 +1,58 @@
import pytest
from qtpy.QtWidgets import QComboBox
from bec_widgets.utils.filter_io import FilterIO
from bec_widgets.widgets.control.device_input.device_line_edit.device_line_edit import (
DeviceLineEdit,
)
from bec_widgets.utils.filter_io import get_bec_signals_for_classes, replace_combobox_items
from bec_widgets.widgets.dap.dap_combo_box.dap_combo_box import DapComboBox
from .client_mocks import mocked_client
from .conftest import create_widget
@pytest.fixture(scope="function")
def dap_mock(qtbot, mocked_client):
"""Fixture for QLineEdit widget"""
models = ["GaussianModel", "LorentzModel", "SineModel"]
mocked_client.dap._available_dap_plugins.keys.return_value = models
def test_replace_combobox_items(qtbot, 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}
@pytest.fixture(scope="function")
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):
def test_get_bec_signals_for_classes_ndim_filter(mocked_client):
signals = [
("dev1", "sig1", {"describe": {"signal_info": {"ndim": 1}}}),
("dev1", "sig2", {"describe": {"signal_info": {"ndim": 2}}}),
]
mocked_client.device_manager.get_bec_signals = lambda _filters: signals
out = FilterIO.update_with_signal_class(
widget=dap_mock.fit_model_combobox,
signal_class_filter=["AsyncSignal"],
client=mocked_client,
ndim_filter=1,
out = get_bec_signals_for_classes(
client=mocked_client, signal_class_filter=["AsyncSignal"], ndim_filter=1
)
assert out == [("dev1", "sig1", {"describe": {"signal_info": {"ndim": 1}}})]
def test_update_with_signal_class_line_edit_passthrough(line_edit_mock, mocked_client):
signals = [("dev1", "sig1", {"describe": {"signal_info": {"ndim": 1}}})]
mocked_client.device_manager.get_bec_signals = lambda _filters: signals
out = FilterIO.update_with_signal_class(
widget=line_edit_mock,
signal_class_filter=["AsyncSignal"],
client=mocked_client,
ndim_filter=1,
)
assert out == signals
def test_replace_combobox_items_empty(qtbot):
widget = QComboBox()
qtbot.addWidget(widget)
widget.addItem("old")
replace_combobox_items(widget, [])
assert widget.count() == 0
def test_replace_combobox_items_preserves_text_and_blocks_signals(qtbot):
widget = QComboBox()
qtbot.addWidget(widget)
widget.setEditable(True)
widget.addItems(["old", "other"])
widget.setCurrentText("typed")
emitted: list[str] = []
widget.currentTextChanged.connect(emitted.append)
replace_combobox_items(widget, ["new"], preserve_current_text=True, block_signals=True)
assert widget.currentText() == "typed"
assert widget.itemText(0) == "new"
assert emitted == []
+2 -2
View File
@@ -62,8 +62,8 @@ TEST_LOG_MESSAGES = [
@pytest.fixture
def log_panel(qtbot, mocked_client):
mocked_client.connector.xread = lambda *_, **__: TEST_LOG_MESSAGES
def log_panel(qtbot, mocked_client, monkeypatch):
monkeypatch.setattr(mocked_client.connector, "xread", lambda *_, **__: TEST_LOG_MESSAGES)
widget = LogPanel()
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
+3 -5
View File
@@ -12,9 +12,7 @@ from bec_widgets.widgets.control.device_control.positioner_box import (
PositionerBox,
PositionerControlLine,
)
from bec_widgets.widgets.control.device_input.device_line_edit.device_line_edit import (
DeviceLineEdit,
)
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox
from .client_mocks import mocked_client
from .conftest import create_widget
@@ -164,8 +162,8 @@ def test_positioner_box_open_dialog_selection(qtbot, positioner_box):
# pylint: disable=protected-access
assert positioner_box._dialog is not None
qtbot.waitUntil(lambda: positioner_box._dialog.isVisible() is True, timeout=1000)
line_edit = positioner_box._dialog.findChild(DeviceLineEdit)
line_edit.setText("samy")
line_edit = positioner_box._dialog.findChild(DeviceComboBox)
line_edit.setCurrentText("samy")
close_button = positioner_box._dialog.findChild(QPushButton)
assert close_button.text() == "Close"
qtbot.mouseClick(close_button, Qt.LeftButton)
+402 -7
View File
@@ -9,7 +9,9 @@ from qtpy.QtCore import QModelIndex, Qt
from bec_widgets.utils.forms_from_types.items import StrFormItem
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.scan_info_adapter import ScanInfoAdapter
from .client_mocks import mocked_client
@@ -279,6 +281,359 @@ def test_populate_scans(scan_control, mocked_client):
assert sorted(items) == sorted(expected_scans)
def test_scan_control_uses_gui_visibility_and_signature(qtbot, mocked_client):
scan_info = {
"class": "AnnotatedScan",
"base_class": "ScanBase",
"arg_input": {
"device": "DeviceBase",
"start": {
"Annotated": {
"type": "float",
"metadata": {
"ScanArgument": {
"display_name": "Start Position",
"description": "Start position",
"tooltip": "Custom start tooltip",
"expert": False,
"alternative_group": None,
"units": None,
"reference_units": "device",
}
},
}
},
"stop": {
"Annotated": {
"type": "float",
"metadata": {
"ScanArgument": {
"display_name": None,
"description": "Stop position",
"tooltip": None,
"expert": False,
"alternative_group": None,
"units": None,
"reference_units": "device",
}
},
}
},
},
"arg_bundle_size": {"bundle": 3, "min": 1, "max": None},
"gui_visibility": {
"Movement Parameters": ["steps", "step_size"],
"Acquisition Parameters": ["exp_time", "relative"],
},
"required_kwargs": [],
"signature": [
{"name": "args", "kind": "VAR_POSITIONAL", "default": "_empty", "annotation": "_empty"},
{"name": "steps", "kind": "KEYWORD_ONLY", "default": 10, "annotation": "int"},
{
"name": "step_size",
"kind": "KEYWORD_ONLY",
"default": None,
"annotation": {
"Annotated": {
"type": "float",
"metadata": {
"ScanArgument": {
"display_name": "Step Size Custom",
"description": "Step size",
"tooltip": "Custom step tooltip",
"expert": False,
"alternative_group": "scan_resolution",
"units": "mm",
"reference_units": None,
}
},
}
},
},
{
"name": "exp_time",
"kind": "KEYWORD_ONLY",
"default": 0,
"annotation": {
"Annotated": {
"type": "float",
"metadata": {
"ScanArgument": {
"display_name": None,
"description": None,
"tooltip": "Exposure time",
"expert": False,
"alternative_group": None,
"units": "s",
"reference_units": None,
}
},
}
},
},
{"name": "relative", "kind": "KEYWORD_ONLY", "default": False, "annotation": "bool"},
{"name": "kwargs", "kind": "VAR_KEYWORD", "default": "_empty", "annotation": "_empty"},
],
}
mocked_client.connector.set_and_publish(
MessageEndpoints.available_scans(),
AvailableResourceMessage(resource={"annotated_scan": scan_info}),
)
widget = ScanControl(client=mocked_client)
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
widget.comboBox_scan_selection.setCurrentText("annotated_scan")
assert widget.comboBox_scan_selection.count() == 1
assert widget.arg_box.layout.itemAtPosition(0, 1).widget().text() == "Start Position"
assert "Custom start tooltip\nUnits from: device" in widget.arg_box.widgets[1].toolTip()
with patch.object(mocked_client.device_manager.devices.samx, "egu", return_value="mm"):
WidgetIO.set_value(widget.arg_box.widgets[0], "samx")
assert widget.arg_box.layout.itemAtPosition(0, 1).widget().text() == "Start Position"
assert widget.arg_box.widgets[1].suffix() == " mm"
assert "Custom start tooltip\nUnits: mm" in widget.arg_box.widgets[1].toolTip()
widget.arg_box.widgets[0].setCurrentText("not_a_device")
assert widget.arg_box.layout.itemAtPosition(0, 1).widget().text() == "Start Position"
assert widget.arg_box.widgets[1].suffix() == ""
assert "Custom start tooltip\nUnits from: device" in widget.arg_box.widgets[1].toolTip()
assert [box.title() for box in widget.kwarg_boxes] == [
"Movement Parameters",
"Acquisition Parameters",
]
assert widget.kwarg_boxes[0].layout.itemAtPosition(0, 1).widget().text() == "Step Size Custom"
assert widget.kwarg_boxes[0].widgets[1].suffix() == " mm"
assert "Custom step tooltip\nUnits: mm" in widget.kwarg_boxes[0].widgets[1].toolTip()
assert widget.kwarg_boxes[1].layout.itemAtPosition(0, 0).widget().text() == "Exp Time"
assert "Exposure time\nUnits: s" in widget.kwarg_boxes[1].widgets[0].toolTip()
def test_scan_info_adapter_skips_duplicate_visible_kwargs():
scan_info = {
"class": "DuplicateScan",
"base_class": "ScanBaseV4",
"arg_input": {},
"arg_bundle_size": {"bundle": 0, "min": None, "max": None},
"gui_visibility": {
"Scan Parameters": ["relative", "burst_at_each_point"],
"Acquisition Parameters": ["exp_time", "burst_at_each_point"],
},
"signature": [
{"name": "relative", "kind": "KEYWORD_ONLY", "default": False, "annotation": "bool"},
{
"name": "burst_at_each_point",
"kind": "KEYWORD_ONLY",
"default": 1,
"annotation": "int",
},
{"name": "exp_time", "kind": "KEYWORD_ONLY", "default": 0, "annotation": "float"},
],
}
gui_config = ScanInfoAdapter().build_scan_ui_config(scan_info)
groups = {
group["name"]: [input_spec["name"] for input_spec in group["inputs"]]
for group in gui_config["kwarg_groups"]
}
assert groups == {
"Scan Parameters": ["relative", "burst_at_each_point"],
"Acquisition Parameters": ["exp_time"],
}
def test_scan_info_adapter_rejects_unsupported_visible_inputs():
scan_info = {
"class": "UnsupportedScan",
"base_class": "ScanBaseV4",
"arg_input": {},
"arg_bundle_size": {"bundle": 0, "min": None, "max": None},
"gui_visibility": {"Regions": ["regions"]},
"signature": [
{
"name": "regions",
"kind": "KEYWORD_ONLY",
"default": "_empty",
"annotation": {
"Generic": {
"origin": "list",
"args": [
{"Generic": {"origin": "tuple", "args": ["float", "float", "int"]}}
],
}
},
}
],
}
gui_config = ScanInfoAdapter().build_scan_ui_config(scan_info)
unsupported_inputs = ScanInfoAdapter.unsupported_inputs(gui_config)
assert [input_spec["name"] for input_spec in unsupported_inputs] == ["regions"]
assert ScanInfoAdapter.has_scan_ui_config(scan_info) is False
def test_scan_info_adapter_skips_hidden_visible_kwargs():
scan_info = {
"class": "HiddenScan",
"base_class": "ScanBaseV4",
"arg_input": {},
"arg_bundle_size": {"bundle": 0, "min": None, "max": None},
"gui_visibility": {"Acquisition": ["exp_time", "internal_token"]},
"signature": [
{"name": "exp_time", "kind": "KEYWORD_ONLY", "default": 0, "annotation": "float"},
{
"name": "internal_token",
"kind": "KEYWORD_ONLY",
"default": None,
"annotation": {
"Annotated": {
"type": "str",
"metadata": {
"ScanArgument": {"display_name": "Internal Token", "hidden": True}
},
}
},
},
],
}
gui_config = ScanInfoAdapter().build_scan_ui_config(scan_info)
assert [input_spec["name"] for input_spec in gui_config["kwarg_groups"][0]["inputs"]] == [
"exp_time"
]
def test_scan_control_propagates_reference_units_across_kwarg_groups(qtbot, mocked_client):
scan_info = {
"class": "RoundScan",
"base_class": "ScanBaseV4",
"arg_input": {},
"arg_bundle_size": {"bundle": 0, "min": None, "max": None},
"gui_visibility": {
"Motors": ["motor_1", "motor_2"],
"Ring Parameters": ["inner_radius", "outer_radius", "center_1", "center_2"],
},
"required_kwargs": [],
"signature": [
{
"name": "motor_1",
"kind": "POSITIONAL_OR_KEYWORD",
"default": "_empty",
"annotation": "DeviceBase",
},
{
"name": "motor_2",
"kind": "POSITIONAL_OR_KEYWORD",
"default": "_empty",
"annotation": "DeviceBase",
},
{
"name": "inner_radius",
"kind": "POSITIONAL_OR_KEYWORD",
"default": "_empty",
"annotation": {
"Annotated": {
"type": "float",
"metadata": {
"ScanArgument": {
"display_name": "Inner Radius",
"units": None,
"reference_units": "motor_1",
"ge": 0,
}
},
}
},
},
{
"name": "outer_radius",
"kind": "POSITIONAL_OR_KEYWORD",
"default": "_empty",
"annotation": {
"Annotated": {
"type": "float",
"metadata": {
"ScanArgument": {
"display_name": "Outer Radius",
"units": None,
"reference_units": "motor_1",
"ge": 0,
}
},
}
},
},
{
"name": "center_1",
"kind": "KEYWORD_ONLY",
"default": 0,
"annotation": {
"Annotated": {
"type": "float",
"metadata": {
"ScanArgument": {
"display_name": "Center Motor 1",
"units": None,
"reference_units": "motor_1",
}
},
}
},
},
{
"name": "center_2",
"kind": "KEYWORD_ONLY",
"default": 0,
"annotation": {
"Annotated": {
"type": "float",
"metadata": {
"ScanArgument": {
"display_name": "Center Motor 2",
"units": None,
"reference_units": "motor_2",
}
},
}
},
},
],
}
mocked_client.connector.set_and_publish(
MessageEndpoints.available_scans(),
AvailableResourceMessage(resource={"round_scan": scan_info}),
)
widget = ScanControl(client=mocked_client)
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
widget.comboBox_scan_selection.setCurrentText("round_scan")
motor_box = widget.kwarg_boxes[0]
ring_box = widget.kwarg_boxes[1]
assert "Units from: motor_1" in ring_box.widgets[0].toolTip()
assert ring_box.widgets[0].suffix() == ""
with patch.object(mocked_client.device_manager.devices.samx, "egu", return_value="mm"):
WidgetIO.set_value(motor_box.widgets[0], "samx")
assert ring_box.widgets[0].suffix() == " mm"
assert ring_box.widgets[1].suffix() == " mm"
assert ring_box.widgets[2].suffix() == " mm"
assert ring_box.widgets[3].suffix() == ""
assert "Units: mm" in ring_box.widgets[0].toolTip()
motor_box.widgets[0].setCurrentText("not_a_device")
assert ring_box.widgets[0].suffix() == ""
assert ring_box.widgets[1].suffix() == ""
assert ring_box.widgets[2].suffix() == ""
assert "Units from: motor_1" in ring_box.widgets[0].toolTip()
def test_current_scan(scan_control, mocked_client):
current_scan = scan_control.current_scan
wrong_scan = "error_scan"
@@ -304,6 +659,13 @@ def test_on_scan_selected(scan_control, scan_name):
assert widget is not None # Confirm that a widget exists
expected_widget_type = scan_control.arg_box.WIDGET_HANDLER.get(arg_value, None)
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
kwargs_group = [param for param in expected_scan_info["gui_config"]["kwarg_groups"]]
@@ -503,12 +865,47 @@ def test_changing_scans_remember_parameters(scan_control, mocked_client):
assert grid_kwargs["burst_at_each_point"] == kwargs["burst_at_each_point"]
@pytest.mark.skip(reason="Unreliable - GH issue #1134")
def test_get_scan_parameters_from_redis(scan_control, mocked_client):
def test_scan_selection_does_not_fetch_last_scan_parameters(
scan_control, mocked_client, monkeypatch
):
xread = MagicMock(wraps=mocked_client.connector.xread)
monkeypatch.setattr(mocked_client.connector, "xread", xread)
scan_control.comboBox_scan_selection.setCurrentText("line_scan")
assert scan_control.comboBox_scan_selection.currentText() == "line_scan"
scan_control.comboBox_scan_selection.setCurrentText("grid_scan")
xread.assert_not_called()
def test_restore_last_scan_parameters_button_fetches_on_demand(
scan_control, mocked_client, monkeypatch
):
xread = MagicMock(wraps=mocked_client.connector.xread)
monkeypatch.setattr(mocked_client.connector, "xread", xread)
scan_control.comboBox_scan_selection.setCurrentText("grid_scan")
scan_control.comboBox_scan_selection.setCurrentText("line_scan")
xread.assert_not_called()
scan_control.last_scan_button.click()
xread.assert_called_once_with(
MessageEndpoints.scan_history(), from_start=True, user_id=scan_control.object_name
)
args, kwargs = scan_control.get_scan_parameters(bec_object=False)
assert args == ["samx", 0.0, 2.0]
assert kwargs["steps"] == 10
assert kwargs["relative"] is False
assert kwargs["exp_time"] == 2
def test_get_scan_parameters_from_redis(scan_control):
scan_name = "line_scan"
scan_control.comboBox_scan_selection.setCurrentText(scan_name)
scan_control.toggle.checked = True
scan_control.last_scan_button.click()
args, kwargs = scan_control.get_scan_parameters(bec_object=False)
@@ -588,8 +985,7 @@ def test_scan_metadata_is_passed_to_scan_function(scan_control: ScanControl):
scans.grid_scan.assert_called_once_with(metadata=TEST_MD)
@pytest.mark.skip(reason="Unreliable - GH issue #1134")
def test_restore_parameters_with_fewer_arg_bundles(scan_control, qtbot):
def test_restore_parameters_with_fewer_arg_bundles(scan_control):
"""
Ensure that when more argument bundles are present than exist in the
stored history, restoring parameters regenerates the arg box to the
@@ -605,8 +1001,7 @@ def test_restore_parameters_with_fewer_arg_bundles(scan_control, qtbot):
assert scan_control.arg_box.count_arg_rows() == 3
# Trigger restore of parameters from history
scan_control.toggle.checked = True
qtbot.wait(200)
scan_control.last_scan_button.click()
# After restore, arg_box should have only one bundle (the history size)
assert scan_control.arg_box.count_arg_rows() == 1
@@ -67,28 +67,28 @@ def test_kwarg_box(qtbot):
assert kwarg_box.widgets[0].__class__.__name__ == "ScanDoubleSpinBox"
assert kwarg_box.widgets[0].arg_name == "exp_time"
assert WidgetIO.get_value(kwarg_box.widgets[0]) == 0
assert kwarg_box.widgets[0].toolTip() == "Exposure time in seconds"
assert "Exposure time in seconds" in kwarg_box.widgets[0].toolTip()
# Widget 1
assert kwarg_box.widgets[1].__class__.__name__ == "ScanSpinBox"
assert kwarg_box.widgets[1].arg_name == "num_points"
assert WidgetIO.get_value(kwarg_box.widgets[1]) == 1
assert kwarg_box.widgets[1].toolTip() == "Number of points"
assert "Number of points" in kwarg_box.widgets[1].toolTip()
# Widget 2
assert kwarg_box.widgets[2].__class__.__name__ == "ScanCheckBox"
assert kwarg_box.widgets[2].arg_name == "relative"
assert WidgetIO.get_value(kwarg_box.widgets[2]) == False
assert (
kwarg_box.widgets[2].toolTip()
== "If True, the motors will be moved relative to their current position"
"If True, the motors will be moved relative to their current position"
in kwarg_box.widgets[2].toolTip()
)
# Widget 3
assert kwarg_box.widgets[3].__class__.__name__ == "ScanLineEdit"
assert kwarg_box.widgets[3].arg_name == "scan_type"
assert WidgetIO.get_value(kwarg_box.widgets[3]) == "line"
assert kwarg_box.widgets[3].toolTip() == "Type of scan"
assert "Type of scan" in kwarg_box.widgets[3].toolTip()
parameters = kwarg_box.get_parameters()
assert parameters == {"exp_time": 0, "num_points": 1, "relative": False, "scan_type": "line"}
@@ -146,14 +146,92 @@ def test_arg_box(qtbot):
assert arg_box.widgets[0].__class__.__name__ == "ScanLineEdit"
assert arg_box.widgets[0].arg_name == "device"
assert WidgetIO.get_value(arg_box.widgets[0]) == "samx"
assert arg_box.widgets[0].toolTip() == "Device to scan"
assert "Device to scan" in arg_box.widgets[0].toolTip()
# Widget 1
assert arg_box.widgets[1].__class__.__name__ == "ScanDoubleSpinBox"
assert arg_box.widgets[1].arg_name == "start"
assert WidgetIO.get_value(arg_box.widgets[1]) == 0
assert arg_box.widgets[1].toolTip() == "Start position"
assert "Start position" in arg_box.widgets[1].toolTip()
# Widget 2
assert arg_box.widgets[2].__class__.__name__ == "ScanSpinBox"
assert arg_box.widgets[2].arg_name
def test_spinbox_limits_from_scan_info(qtbot):
group_input = {
"name": "Kwarg Test",
"inputs": [
{
"arg": False,
"name": "exp_time",
"type": "float",
"display_name": "Exp Time",
"tooltip": "Exposure time in seconds",
"default": 2.0,
"expert": False,
"precision": 3,
"gt": 1.5,
"ge": None,
"lt": 5.0,
"le": None,
},
{
"arg": False,
"name": "num_points",
"type": "int",
"display_name": "Num Points",
"tooltip": "Number of points",
"default": 4,
"expert": False,
"gt": None,
"ge": 3,
"lt": 9,
"le": None,
},
{
"arg": False,
"name": "settling_time",
"type": "float",
"display_name": "Settling Time",
"tooltip": "Settling time in seconds",
"default": 0.5,
"expert": False,
"gt": None,
"ge": 0.2,
"lt": None,
"le": 3.5,
},
{
"arg": False,
"name": "steps",
"type": "int",
"display_name": "Steps",
"tooltip": "Number of steps",
"default": 4,
"expert": False,
"gt": 0,
"ge": None,
"lt": None,
"le": 10,
},
],
}
kwarg_box = ScanGroupBox(box_type="kwargs", config=group_input)
exp_time = kwarg_box.widgets[0]
num_points = kwarg_box.widgets[1]
settling_time = kwarg_box.widgets[2]
steps = kwarg_box.widgets[3]
assert exp_time.decimals() == 3
assert exp_time.minimum() == 1.501
assert exp_time.maximum() == 4.999
assert num_points.minimum() == 3
assert num_points.maximum() == 8
assert settling_time.minimum() == 0.2
assert settling_time.maximum() == 3.5
assert steps.minimum() == 1
assert steps.maximum() == 10
+65 -1
View File
@@ -3,7 +3,9 @@ from unittest import mock
import numpy as np
import pytest
from bec_lib import messages
from bec_lib.endpoints import MessageEndpoints
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.widgets.progress.bec_progressbar.bec_progressbar import (
BECProgressBar,
ProgressState,
@@ -148,7 +150,6 @@ def test_source_label_updates(scan_progressbar):
def test_set_progress_source_connections(scan_progressbar, monkeypatch):
""" """
from bec_lib.endpoints import MessageEndpoints
connect_calls = []
disconnect_calls = []
@@ -187,6 +188,69 @@ def test_set_progress_source_connections(scan_progressbar, monkeypatch):
assert len(connect_calls) == prev_connect_count, "No extra connect made for same source"
def test_set_progress_source_disconnects_previous_device_subscription(
scan_progressbar, monkeypatch
):
disconnect_calls = []
monkeypatch.setattr(scan_progressbar.bec_dispatcher, "connect_slot", lambda *args: None)
monkeypatch.setattr(
scan_progressbar.bec_dispatcher,
"disconnect_slot",
lambda slot, endpoint: disconnect_calls.append(endpoint),
)
scan_progressbar.set_progress_source(ProgressSource.DEVICE_PROGRESS, device="motor1")
scan_progressbar.set_progress_source(ProgressSource.DEVICE_PROGRESS, device="motor2")
assert disconnect_calls == [MessageEndpoints.device_progress(device="motor1")]
def test_set_progress_source_disconnects_device_when_switching_to_scan(
scan_progressbar, monkeypatch
):
disconnect_calls = []
monkeypatch.setattr(scan_progressbar.bec_dispatcher, "connect_slot", lambda *args: None)
monkeypatch.setattr(
scan_progressbar.bec_dispatcher,
"disconnect_slot",
lambda slot, endpoint: disconnect_calls.append(endpoint),
)
scan_progressbar.set_progress_source(ProgressSource.DEVICE_PROGRESS, device="motor1")
scan_progressbar.set_progress_source(ProgressSource.SCAN_PROGRESS)
assert disconnect_calls == [MessageEndpoints.device_progress(device="motor1")]
def test_cleanup_disconnects_active_device_subscription(scan_progressbar, monkeypatch):
disconnect_calls = []
monkeypatch.setattr(scan_progressbar.bec_dispatcher, "connect_slot", lambda *args: None)
monkeypatch.setattr(
scan_progressbar.bec_dispatcher,
"disconnect_slot",
lambda slot, endpoint: disconnect_calls.append(endpoint),
)
monkeypatch.setattr(scan_progressbar.progressbar, "close", lambda: None)
monkeypatch.setattr(scan_progressbar.progressbar, "deleteLater", lambda: None)
monkeypatch.setattr(BECWidget, "cleanup", lambda self: None)
scan_progressbar.set_progress_source(ProgressSource.DEVICE_PROGRESS, device="motor1")
ScanProgressBar.cleanup(scan_progressbar)
assert disconnect_calls == [
MessageEndpoints.device_progress(device="motor1"),
MessageEndpoints.scan_queue_status(),
]
assert scan_progressbar._progress_source is None
assert scan_progressbar._progress_device is None
def test_progressbar_queue_update(scan_progressbar):
"""
Test that an empty queue update does not change the progress source.
+10 -7
View File
@@ -6,8 +6,8 @@ import pytest
from qtpy import QtCore
from qtpy.QtWidgets import QDialogButtonBox, QLabel
from bec_widgets.widgets.control.device_input.base_classes.device_signal_input_base import (
DeviceSignalInputBaseConfig,
from bec_widgets.widgets.control.device_input.signal_combobox.signal_combobox import (
SignalComboBoxConfig,
)
from bec_widgets.widgets.utility.signal_label.signal_label import ChoiceDialog, SignalLabel
@@ -61,7 +61,7 @@ SAMX_INFO_DICT = {
@pytest.fixture
def signal_label(qtbot, mocked_client: MagicMock):
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(
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()
qtbot.waitUntil(dialog.button_box.button(QDialogButtonBox.Ok).isVisible, timeout=500)
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.addItem("test signal")
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()
qtbot.waitUntil(dialog.button_box.button(QDialogButtonBox.Ok).isVisible, timeout=500)
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.addItem("spin_speed")
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()
dialog = signal_label.show_choice_dialog()
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.setCurrentIndex(0)
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"}}
}
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.currentText() == "signal 1"
+23
View File
@@ -36,3 +36,26 @@ def test_toggle_click(qtbot, toggle):
qtbot.mouseClick(toggle, Qt.LeftButton)
toggle.paintEvent(None)
assert toggle.checked is not init_state
def test_toggle_disabled_state_blocks_clicks_and_restores_colors(qtbot, toggle):
toggle.checked = True
assert toggle._track_color == toggle.active_track_color
assert toggle._thumb_color == toggle.active_thumb_color
toggle.setEnabled(False)
assert toggle._track_color == toggle._disabled_track_color
assert toggle._thumb_color == toggle._disabled_thumb_color
qtbot.mouseClick(toggle, Qt.LeftButton)
assert toggle.checked is True
assert toggle._track_color == toggle._disabled_track_color
assert toggle._thumb_color == toggle._disabled_thumb_color
toggle.setEnabled(True)
assert toggle.checked is True
assert toggle._track_color == toggle.active_track_color
assert toggle._thumb_color == toggle.active_thumb_color