mirror of
https://github.com/bec-project/bec_widgets.git
synced 2026-06-10 23:28:49 +02:00
Compare commits
25 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| b119c5ad76 | |||
| 9a58dba414 | |||
| c9fc0a82b9 | |||
| 668b1bd9cd | |||
| 1a6c8bf30f | |||
| c346bd0f18 | |||
| 5f86e41a03 | |||
| f7a48b5f6a | |||
| b4beb274da | |||
| 80694d151f | |||
| f03a5d9e85 | |||
| 5e8f0e8083 | |||
| 9eb05416ab | |||
| ab6a1aecc1 | |||
| d99db7d042 | |||
| a976837cff | |||
| 56427a7f0c | |||
| c4d4b78846 | |||
| 2dc0227d38 | |||
| 2d8e1eed4d | |||
| 3b579e740f | |||
| b8740c9594 | |||
| d5bf10e216 | |||
| 3a165b26ed | |||
| faa200bf5c |
@@ -1,6 +1,101 @@
|
||||
# CHANGELOG
|
||||
|
||||
|
||||
## v3.13.4 (2026-05-29)
|
||||
|
||||
### Bug Fixes
|
||||
|
||||
- **positioner_box**: Fix STOP button
|
||||
([`9a58dba`](https://github.com/bec-project/bec_widgets/commit/9a58dba414d9eec32fd7de7fc64c97c38f020b84))
|
||||
|
||||
|
||||
## 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
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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):
|
||||
|
||||
+1
-1
@@ -429,7 +429,7 @@ class PositionerBox2D(PositionerBoxBase):
|
||||
|
||||
@SafeSlot()
|
||||
def on_stop(self):
|
||||
self._stop_device(f"{self.device_hor} or {self.device_ver}")
|
||||
self._stop_device([self.device_hor, self.device_ver])
|
||||
|
||||
@SafeProperty(float)
|
||||
def step_size_hor(self):
|
||||
|
||||
@@ -1,11 +1,10 @@
|
||||
import uuid
|
||||
from abc import abstractmethod
|
||||
from typing import Callable, TypedDict
|
||||
from typing import Callable, Sequence, TypedDict
|
||||
|
||||
from bec_lib.device import Positioner
|
||||
from bec_lib.endpoints import MessageEndpoints
|
||||
from bec_lib.logger import bec_logger
|
||||
from bec_lib.messages import ScanQueueMessage
|
||||
from bec_lib.messages import VariableMessage
|
||||
from qtpy.QtWidgets import (
|
||||
QDialog,
|
||||
QDoubleSpinBox,
|
||||
@@ -116,17 +115,16 @@ class PositionerBoxBase(BECWidget, QWidget):
|
||||
else:
|
||||
ui["units"].setVisible(False)
|
||||
|
||||
def _stop_device(self, device: str):
|
||||
def _stop_device(self, device: str | Sequence[str]):
|
||||
"""Stop call"""
|
||||
request_id = str(uuid.uuid4())
|
||||
params = {"device": device, "rpc_id": request_id, "func": "stop", "args": [], "kwargs": {}}
|
||||
msg = ScanQueueMessage(
|
||||
scan_type="device_rpc",
|
||||
parameter=params,
|
||||
queue="emergency",
|
||||
metadata={"RID": request_id, "response": False},
|
||||
)
|
||||
self.client.connector.send(MessageEndpoints.scan_queue_request(self.client.username), msg)
|
||||
devices = [device] if isinstance(device, str) else list(device)
|
||||
devices = [dev for dev in devices if dev]
|
||||
if not devices:
|
||||
logger.warning("Stop requested without a valid device.")
|
||||
return
|
||||
|
||||
msg = VariableMessage(value=devices)
|
||||
self.client.connector.send(MessageEndpoints.stop_devices(), msg)
|
||||
|
||||
# pylint: disable=unused-argument
|
||||
def _on_device_readback(
|
||||
|
||||
@@ -489,13 +489,17 @@ class DeviceComboBox(BECWidget, QComboBox):
|
||||
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:
|
||||
self.bec_dispatcher.client.callbacks.remove(self._callback_id)
|
||||
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:
|
||||
|
||||
@@ -197,16 +197,21 @@ class SignalComboBox(BECWidget, QComboBox):
|
||||
self.update_signals_from_filters()
|
||||
|
||||
@SafeSlot()
|
||||
@SafeSlot(dict, dict)
|
||||
def update_signals_from_filters(
|
||||
self, content: dict | None = None, metadata: dict | None = None
|
||||
):
|
||||
@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:
|
||||
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.
|
||||
metadata: Optional callback metadata from BEC device updates. Currently unused.
|
||||
"""
|
||||
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:
|
||||
@@ -588,7 +593,10 @@ class SignalComboBox(BECWidget, QComboBox):
|
||||
|
||||
def cleanup(self):
|
||||
"""Cleanup the widget."""
|
||||
self.bec_dispatcher.client.callbacks.remove(self._device_update_register)
|
||||
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()
|
||||
|
||||
@staticmethod
|
||||
|
||||
@@ -14,7 +14,6 @@ from qtpy.QtWidgets import (
|
||||
QLabel,
|
||||
QPushButton,
|
||||
QSizePolicy,
|
||||
QSpacerItem,
|
||||
QVBoxLayout,
|
||||
QWidget,
|
||||
)
|
||||
@@ -25,6 +24,7 @@ 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
|
||||
|
||||
|
||||
@@ -95,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()
|
||||
@@ -184,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
|
||||
@@ -376,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()
|
||||
@@ -414,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)
|
||||
@@ -427,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):
|
||||
"""
|
||||
|
||||
@@ -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:
|
||||
"""
|
||||
@@ -281,20 +285,31 @@ class ScanGroupBox(QGroupBox):
|
||||
)
|
||||
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)
|
||||
|
||||
@@ -321,6 +336,7 @@ class ScanGroupBox(QGroupBox):
|
||||
for widget in self.widgets[-len(self.inputs) :]:
|
||||
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)]
|
||||
@@ -333,6 +349,7 @@ class ScanGroupBox(QGroupBox):
|
||||
for widget in list(self.widgets):
|
||||
if isinstance(widget, DeviceComboBox):
|
||||
self.selected_devices.pop(widget, None)
|
||||
self._widget_configs.pop(widget, None)
|
||||
widget.close()
|
||||
widget.deleteLater()
|
||||
self.layout.removeWidget(widget)
|
||||
@@ -435,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,
|
||||
}
|
||||
@@ -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()
|
||||
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
+10
-1
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "bec_widgets"
|
||||
version = "3.11.0"
|
||||
version = "3.13.4"
|
||||
description = "BEC Widgets"
|
||||
requires-python = ">=3.11"
|
||||
classifiers = [
|
||||
@@ -63,6 +63,15 @@ qtermwidget = ["pyside6_qtermwidget"]
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
[build-system]
|
||||
requires = ["hatchling"]
|
||||
build-backend = "hatchling.build"
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
from bec_lib.device import ReadoutPriority
|
||||
|
||||
@@ -124,6 +126,36 @@ def test_device_input_combobox_disabled_invalid_has_neutral_border(device_input_
|
||||
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()
|
||||
|
||||
@@ -188,10 +188,56 @@ def test_linked_device_combobox_updates_signal_combobox_on_each_text_change(
|
||||
def test_device_signal_input_base_cleanup(qtbot, mocked_client):
|
||||
with mock.patch.object(mocked_client.callbacks, "remove"):
|
||||
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):
|
||||
|
||||
@@ -2,7 +2,7 @@ from unittest import mock
|
||||
|
||||
import pytest
|
||||
from bec_lib.endpoints import MessageEndpoints
|
||||
from bec_lib.messages import ScanQueueMessage
|
||||
from bec_lib.messages import VariableMessage
|
||||
from qtpy.QtCore import Qt, QTimer
|
||||
from qtpy.QtGui import QValidator
|
||||
from qtpy.QtWidgets import QPushButton
|
||||
@@ -34,15 +34,11 @@ class PositionerWithoutPrecision(Positioner):
|
||||
def positioner_box(qtbot, mocked_client):
|
||||
"""Fixture for PositionerBox widget"""
|
||||
with mock.patch(
|
||||
"bec_widgets.widgets.control.device_control.positioner_box.positioner_box_base.uuid.uuid4"
|
||||
) as mock_uuid:
|
||||
mock_uuid.return_value = "fake_uuid"
|
||||
with mock.patch(
|
||||
"bec_widgets.widgets.control.device_control.positioner_box.positioner_box_base.PositionerBoxBase._check_device_is_valid",
|
||||
return_value=True,
|
||||
):
|
||||
db = create_widget(qtbot, PositionerBox, device="samx", client=mocked_client)
|
||||
yield db
|
||||
"bec_widgets.widgets.control.device_control.positioner_box.positioner_box_base.PositionerBoxBase._check_device_is_valid",
|
||||
return_value=True,
|
||||
):
|
||||
db = create_widget(qtbot, PositionerBox, device="samx", client=mocked_client)
|
||||
yield db
|
||||
|
||||
|
||||
def test_positioner_box(positioner_box):
|
||||
@@ -89,16 +85,8 @@ def test_positioner_box_on_stop(positioner_box):
|
||||
"""Test on stop button"""
|
||||
with mock.patch.object(positioner_box.client.connector, "send") as mock_send:
|
||||
positioner_box.on_stop()
|
||||
params = {"device": "samx", "rpc_id": "fake_uuid", "func": "stop", "args": [], "kwargs": {}}
|
||||
msg = ScanQueueMessage(
|
||||
scan_type="device_rpc",
|
||||
parameter=params,
|
||||
queue="emergency",
|
||||
metadata={"RID": "fake_uuid", "response": False},
|
||||
)
|
||||
mock_send.assert_called_once_with(
|
||||
MessageEndpoints.scan_queue_request(positioner_box.client.username), msg
|
||||
)
|
||||
msg = VariableMessage(value=["samx"])
|
||||
mock_send.assert_called_once_with(MessageEndpoints.stop_devices(), msg)
|
||||
|
||||
|
||||
def test_positioner_box_setpoint_change(positioner_box):
|
||||
@@ -139,19 +127,15 @@ def test_positioner_control_line(qtbot, mocked_client):
|
||||
Inherits from PositionerBox, but the layout is changed. Check dimensions only
|
||||
"""
|
||||
with mock.patch(
|
||||
"bec_widgets.widgets.control.device_control.positioner_box.positioner_box_base.uuid.uuid4"
|
||||
) as mock_uuid:
|
||||
mock_uuid.return_value = "fake_uuid"
|
||||
with mock.patch(
|
||||
"bec_widgets.widgets.control.device_control.positioner_box.positioner_box.positioner_box.PositionerBox._check_device_is_valid",
|
||||
return_value=True,
|
||||
):
|
||||
db = PositionerControlLine(device="samx", client=mocked_client)
|
||||
qtbot.addWidget(db)
|
||||
"bec_widgets.widgets.control.device_control.positioner_box.positioner_box.positioner_box.PositionerBox._check_device_is_valid",
|
||||
return_value=True,
|
||||
):
|
||||
db = PositionerControlLine(device="samx", client=mocked_client)
|
||||
qtbot.addWidget(db)
|
||||
|
||||
assert db.ui.device_box.height() == db.height()
|
||||
assert db.ui.device_box.height() >= db.dimensions[0]
|
||||
assert db.ui.device_box.width() == 600
|
||||
assert db.ui.device_box.height() == db.height()
|
||||
assert db.ui.device_box.height() >= db.dimensions[0]
|
||||
assert db.ui.device_box.width() == 600
|
||||
|
||||
|
||||
def test_positioner_box_open_dialog_selection(qtbot, positioner_box):
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
from bec_lib.endpoints import MessageEndpoints
|
||||
from bec_lib.messages import VariableMessage
|
||||
|
||||
from bec_widgets.widgets.control.device_control.positioner_box import PositionerBox2D
|
||||
|
||||
@@ -12,17 +14,13 @@ from .conftest import create_widget
|
||||
def positioner_box_2d(qtbot, mocked_client):
|
||||
"""Fixture for PositionerBox widget"""
|
||||
with mock.patch(
|
||||
"bec_widgets.widgets.control.device_control.positioner_box.positioner_box_base.uuid.uuid4"
|
||||
) as mock_uuid:
|
||||
mock_uuid.return_value = "fake_uuid"
|
||||
with mock.patch(
|
||||
"bec_widgets.widgets.control.device_control.positioner_box.positioner_box_base.PositionerBoxBase._check_device_is_valid",
|
||||
return_value=True,
|
||||
):
|
||||
db = create_widget(
|
||||
qtbot, PositionerBox2D, device_hor="samx", device_ver="samy", client=mocked_client
|
||||
)
|
||||
yield db
|
||||
"bec_widgets.widgets.control.device_control.positioner_box.positioner_box_base.PositionerBoxBase._check_device_is_valid",
|
||||
return_value=True,
|
||||
):
|
||||
db = create_widget(
|
||||
qtbot, PositionerBox2D, device_hor="samx", device_ver="samy", client=mocked_client
|
||||
)
|
||||
yield db
|
||||
|
||||
|
||||
def test_positioner_box_2d(positioner_box_2d):
|
||||
@@ -82,6 +80,14 @@ def test_positioner_box_setpoint_changes(positioner_box_2d: PositionerBox2D):
|
||||
mock_move.assert_called_once_with(100, relative=False)
|
||||
|
||||
|
||||
def test_positioner_box_2d_on_stop(positioner_box_2d: PositionerBox2D):
|
||||
"""Stop button sends both positioners to the immediate stop endpoint."""
|
||||
with mock.patch.object(positioner_box_2d.client.connector, "send") as mock_send:
|
||||
positioner_box_2d.on_stop()
|
||||
msg = VariableMessage(value=["samx", "samy"])
|
||||
mock_send.assert_called_once_with(MessageEndpoints.stop_devices(), msg)
|
||||
|
||||
|
||||
def _hor_buttons(widget: PositionerBox2D):
|
||||
return [
|
||||
widget.ui.tweak_increase_hor,
|
||||
|
||||
@@ -11,6 +11,7 @@ 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
|
||||
|
||||
@@ -280,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"
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user