Compare commits

..

25 Commits

Author SHA1 Message Date
semantic-release b119c5ad76 3.13.4
Automatically generated by python-semantic-release
2026-05-29 17:29:10 +00:00
wyzula_j 9a58dba414 fix(positioner_box): fix STOP button 2026-05-29 19:28:15 +02: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
26 changed files with 1371 additions and 121 deletions
+95
View File
@@ -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
+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):
+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):
@@ -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()
+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()
+10 -1
View File
@@ -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"
+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)
+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
@@ -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()
+47 -1
View File
@@ -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):
+16 -32
View File
@@ -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):
+17 -11
View File
@@ -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,
+354
View File
@@ -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
+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.
+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