bec/bec_widgets
1
0 Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-10-31 00:31:28 +01:00
Code Activity

refactor(device_browser): remove the DeviceBrowser widget

Browse Source
This commit is contained in:
wyzula_j authored and Jan Wyzula committed 2026-10-20 14:28:10 +02:00
1 parent ca63c614a6
commit 6ebaee8648
20 files changed
+4 -1553
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

bec_widgets/applications/views/device_manager_view/device_manager_display_widget.py
+3 -3
View File
@@ -46,6 +46,9 @@ from bec_widgets.widgets.control.device_manager.components import (
OphydValidation,
)
from bec_widgets.widgets.control.device_manager.components._util import SharedSelectionSignal
from bec_widgets.widgets.control.device_manager.components.config_communicator import (
CommunicateConfigAction,
)
from bec_widgets.widgets.control.device_manager.components.ophyd_validation.ophyd_validation_utils import (
ConfigStatus,
ConnectionStatus,
@@ -53,9 +56,6 @@ from bec_widgets.widgets.control.device_manager.components.ophyd_validation.ophy
from bec_widgets.widgets.progress.device_initialization_progress_bar.device_initialization_progress_bar import (
DeviceInitializationProgressBar,
)
from bec_widgets.widgets.services.device_browser.device_item.config_communicator import (
CommunicateConfigAction,
)
from bec_widgets.widgets.utility.spinner.spinner import SpinnerWidget
if TYPE_CHECKING: # pragma: no cover
bec_widgets/cli/client.py
-25
View File
@@ -35,7 +35,6 @@ _Widgets = {
"BeamlineStateManager": "BeamlineStateManager",
"BecConsole": "BecConsole",
"DapComboBox": "DapComboBox",
"DeviceBrowser": "DeviceBrowser",
"Heatmap": "Heatmap",
"Image": "Image",
"LogPanel": "LogPanel",
@@ -1136,30 +1135,6 @@ class DeveloperView(RPCBase):
"""
class DeviceBrowser(RPCBase):
"""DeviceBrowser is a widget that displays all available devices in the current BEC session."""
_IMPORT_MODULE = "bec_widgets.widgets.services.device_browser.device_browser"
@rpc_call
def remove(self):
"""
Cleanup the BECConnector
"""
@rpc_call
def attach(self):
"""
None
"""
@rpc_call
def detach(self):
"""
Detach the widget from its parent dock widget (if widget is in the dock), making it a floating widget.
"""
class DeviceInitializationProgressBar(RPCBase):
"""A progress bar that displays the progress of device initialization."""
bec_widgets/cli/designer_plugins.py
-5
View File
@@ -33,10 +33,6 @@ designer_plugins = {
"bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button",
"DarkModeButton",
),
"DeviceBrowser": (
"bec_widgets.widgets.services.device_browser.device_browser",
"DeviceBrowser",
),
"DeviceComboBox": (
"bec_widgets.widgets.control.device_input.device_combobox.device_combobox",
"DeviceComboBox",
@@ -130,7 +126,6 @@ widget_icons = {
"ColorButtonNative": "colors",
"DapComboBox": "data_exploration",
"DarkModeButton": "dark_mode",
"DeviceBrowser": "lists",
"DeviceComboBox": "list_alt",
"FeedbackDialog": "feedback",
"Heatmap": "dataset",
bec_widgets/widgets/services/device_browser/device_item/config_communicator.py → bec_widgets/widgets/control/device_manager/components/config_communicator.py
View File
File renamed without changes.
bec_widgets/widgets/services/device_browser/__init__.py
View File
Whitespace-only changes.
bec_widgets/widgets/services/device_browser/device_browser.py
-221
View File
@@ -1,221 +0,0 @@
import os
from typing import Callable
import bec_lib
from bec_lib.callback_handler import EventType
from bec_lib.config_helper import ConfigHelper
from bec_lib.endpoints import MessageEndpoints
from bec_lib.logger import bec_logger
from bec_lib.messages import ConfigAction, ScanStatusMessage
from bec_qthemes import material_icon
from pyqtgraph import SignalProxy
from qtpy.QtCore import QThreadPool, Signal
from qtpy.QtWidgets import QFileDialog, QListWidget, QToolButton, QVBoxLayout, QWidget
from bec_widgets.utils.bec_signal_proxy import cleanup_signal_proxy
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.list_of_expandable_frames import ListOfExpandableFrames
from bec_widgets.utils.rpc_register import RPCRegister
from bec_widgets.utils.ui_loader import UILoader
from bec_widgets.widgets.services.device_browser.device_item import DeviceItem
from bec_widgets.widgets.services.device_browser.device_item.device_config_dialog import (
DirectUpdateDeviceConfigDialog,
)
from bec_widgets.widgets.services.device_browser.util import map_device_type_to_icon
logger = bec_logger.logger
class DeviceBrowser(BECWidget, QWidget):
"""
DeviceBrowser is a widget that displays all available devices in the current BEC session.
"""
devices_changed: Signal = Signal()
editing_enabled: Signal = Signal(bool)
device_update: Signal = Signal(str, dict)
PLUGIN = True
ICON_NAME = "lists"
def __init__(
self,
parent: QWidget | None = None,
config=None,
client=None,
gui_id: str | None = None,
**kwargs,
) -> None:
super().__init__(parent=parent, client=client, gui_id=gui_id, config=config, **kwargs)
self.get_bec_shortcuts()
self._config_helper = ConfigHelper(
self.client.connector, self.client._service_name, self.client.device_manager
)
self._q_threadpool = QThreadPool()
self.ui = None
self.init_ui()
self.dev_list = ListOfExpandableFrames(self, DeviceItem)
self.ui.verticalLayout.addWidget(self.dev_list)
self.dev_list.setVerticalScrollMode(QListWidget.ScrollMode.ScrollPerPixel)
self.proxy_device_update = SignalProxy(
self.ui.filter_input.textChanged, rateLimit=500, slot=self.update_device_list
)
self._device_update_callback_id = self.bec_dispatcher.client.callbacks.register(
EventType.DEVICE_UPDATE, self.on_device_update
)
self._scan_status_callback_id = self.bec_dispatcher.client.callbacks.register(
EventType.SCAN_STATUS, self.scan_status_changed
)
self._default_config_dir = os.path.abspath(
os.path.join(os.path.dirname(bec_lib.__file__), "./configs/")
)
self.devices_changed.connect(self.update_device_list)
self.init_warning_label()
self.init_tool_buttons()
self.init_device_list()
self.update_device_list()
def init_ui(self) -> None:
"""
Initialize the UI by loading the UI file and setting the layout.
"""
layout = QVBoxLayout()
ui_file_path = os.path.join(os.path.dirname(__file__), "device_browser.ui")
self.ui = UILoader(self).loader(ui_file_path)
layout.addWidget(self.ui)
self.setLayout(layout)
def init_warning_label(self):
self.ui.scan_running_warning.setText("Warning: editing disabled while scan is running!")
self.ui.scan_running_warning.setStyleSheet(
"background-color: #fcba03; color: rgb(0, 0, 0);"
)
scan_status = self.bec_dispatcher.client.connector.get(MessageEndpoints.scan_status())
initial_status = scan_status.status if scan_status is not None else "closed"
self.set_editing_mode(initial_status not in ["open", "paused"])
def init_tool_buttons(self):
def _setup_button(button: QToolButton, icon: str, slot: Callable, tooltip: str = ""):
button.clicked.connect(slot)
button.setIcon(material_icon(icon, size=(20, 20), convert_to_pixmap=False))
button.setToolTip(tooltip)
_setup_button(self.ui.add_button, "add", self._create_add_dialog, "add new device")
_setup_button(self.ui.save_button, "save", self._save_to_file, "save config to file")
_setup_button(
self.ui.import_button, "input", self._load_from_file, "append/merge config from file"
)
def _create_add_dialog(self):
dialog = DirectUpdateDeviceConfigDialog(parent=self, device=None, action="add")
dialog.open()
def on_device_update(self, action: ConfigAction, content: dict) -> None:
"""
Callback for device update events. Triggers the device_update signal.
Args:
action (str): The action that triggered the event.
content (dict): The content of the config update.
"""
if action in ["add", "remove", "reload"]:
self.devices_changed.emit()
if action in ["update", "reload"]:
self.device_update.emit(action, content)
def init_device_list(self):
self.dev_list.clear()
with RPCRegister.delayed_broadcast():
for device, device_obj in self.dev.items():
self._add_item_to_list(device, device_obj)
def _add_item_to_list(self, device: str, device_obj):
_, device_item = self.dev_list.add_item(
id=device,
parent=self,
device=device,
devices=self.dev,
icon=map_device_type_to_icon(device_obj),
config_helper=self._config_helper,
q_threadpool=self._q_threadpool,
)
self.editing_enabled.connect(device_item.set_editable)
self.device_update.connect(device_item.config_update)
tooltip = self.dev[device]._config.get("description", "")
device_item.setToolTip(tooltip)
@SafeSlot(dict, dict)
def scan_status_changed(self, scan_info: dict, _: dict):
"""disable editing when scans are running and enable editing when they are finished"""
msg = ScanStatusMessage.model_validate(scan_info)
self.set_editing_mode(msg.status not in ["open", "paused"])
def set_editing_mode(self, enabled: bool):
self.ui.add_button.setEnabled(enabled)
self.ui.scan_running_warning.setHidden(enabled)
self.editing_enabled.emit(enabled)
@SafeSlot()
def reset_device_list(self) -> None:
self.init_device_list()
self.update_device_list()
@SafeSlot()
@SafeSlot(str)
def update_device_list(self, *_) -> None:
"""
Update the device list based on the filter input.
There are two ways to trigger this function:
1. By changing the text in the filter input.
2. By emitting the device_update signal.
Either way, the function will filter the devices based on the filter input text and update the device list.
"""
for device in self.dev:
if device not in self.dev_list:
# it is possible the device has just been added to the config
self._add_item_to_list(device, self.dev[device])
self.dev_list.update_filter(self.ui.filter_input.text())
@SafeSlot()
def _load_from_file(self):
file_path, _ = QFileDialog.getOpenFileName(
self, "Update config from file", self._default_config_dir, "Config files (*.yml *.yaml)"
)
if file_path:
self._config_helper.update_session_with_file(file_path)
@SafeSlot()
def _save_to_file(self):
file_path, _ = QFileDialog.getSaveFileName(
self, "Save config to file", self._default_config_dir, "Config files (*.yml *.yaml)"
)
if file_path:
self._config_helper.save_current_session(file_path)
def cleanup(self):
# Stop the filter proxy first, so no queued filter change reaches the closed browser.
cleanup_signal_proxy(self.proxy_device_update)
super().cleanup()
self.bec_dispatcher.client.callbacks.remove(self._scan_status_callback_id)
self.bec_dispatcher.client.callbacks.remove(self._device_update_callback_id)
if __name__ == "__main__": # pragma: no cover
import sys
from qtpy.QtWidgets import QApplication
from bec_widgets.utils.colors import apply_theme
app = QApplication(sys.argv)
apply_theme("light")
widget = DeviceBrowser()
widget.show()
sys.exit(app.exec_())
bec_widgets/widgets/services/device_browser/device_browser.pyproject
-1
View File
@@ -1 +0,0 @@
{'files': ['device_browser.py']}
bec_widgets/widgets/services/device_browser/device_browser.ui
-90
View File
@@ -1,90 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<ui version="4.0">
<class>Form</class>
<widget class="QWidget" name="Form">
<property name="geometry">
<rect>
<x>0</x>
<y>0</y>
<width>406</width>
<height>500</height>
</rect>
</property>
<property name="windowTitle">
<string>Form</string>
</property>
<layout class="QVBoxLayout" name="verticalLayout">
<item>
<widget class="QGroupBox" name="browser_group_box">
<property name="title">
<string>Device Browser</string>
</property>
<layout class="QVBoxLayout" name="verticalLayout_2">
<item>
<layout class="QHBoxLayout" name="filter_layout">
<item>
<widget class="QLineEdit" name="filter_input">
<property name="placeholderText">
<string>Filter</string>
</property>
</widget>
</item>
<item>
<widget class="QGroupBox" name="button_box">
<layout class="QHBoxLayout" name="horizontalLayout">
<property name="leftMargin">
<number>0</number>
</property>
<property name="topMargin">
<number>0</number>
</property>
<property name="rightMargin">
<number>0</number>
</property>
<property name="bottomMargin">
<number>0</number>
</property>
<item>
<widget class="QToolButton" name="add_button">
<property name="text">
<string>...</string>
</property>
</widget>
</item>
<item>
<widget class="QToolButton" name="save_button">
<property name="text">
<string>...</string>
</property>
</widget>
</item>
<item>
<widget class="QToolButton" name="import_button">
<property name="text">
<string>...</string>
</property>
</widget>
</item>
</layout>
</widget>
</item>
</layout>
</item>
<item>
<widget class="QLabel" name="scan_running_warning">
<property name="styleSheet">
<string notr="true" />
</property>
<property name="text">
<string>warning</string>
</property>
</widget>
</item>
</layout>
</widget>
</item>
</layout>
</widget>
<resources />
<connections />
</ui>
bec_widgets/widgets/services/device_browser/device_browser_plugin.py
-57
View File
@@ -1,57 +0,0 @@
# Copyright (C) 2022 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
from qtpy.QtDesigner import QDesignerCustomWidgetInterface
from qtpy.QtWidgets import QWidget
from bec_widgets.utils.bec_designer import designer_material_icon
from bec_widgets.widgets.services.device_browser.device_browser import DeviceBrowser
DOM_XML = """
<ui language='c++'>
<widget class='DeviceBrowser' name='device_browser'>
</widget>
</ui>
"""
class DeviceBrowserPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
def __init__(self):
super().__init__()
self._form_editor = None
def createWidget(self, parent):
if parent is None:
return QWidget()
t = DeviceBrowser(parent)
return t
def domXml(self):
return DOM_XML
def group(self):
return "BEC Services"
def icon(self):
return designer_material_icon(DeviceBrowser.ICON_NAME)
def includeFile(self):
return "device_browser"
def initialize(self, form_editor):
self._form_editor = form_editor
def isContainer(self):
return False
def isInitialized(self):
return self._form_editor is not None
def name(self):
return "DeviceBrowser"
def toolTip(self):
return "DeviceBrowser"
def whatsThis(self):
return self.toolTip()
bec_widgets/widgets/services/device_browser/device_item/__init__.py
-1
View File
@@ -1 +0,0 @@
from .device_item import DeviceItem
bec_widgets/widgets/services/device_browser/device_item/device_config_dialog.py
-321
View File
@@ -1,321 +0,0 @@
from ast import literal_eval
from typing import Literal
from bec_lib.atlas_models import Device as DeviceConfigModel
from bec_lib.config_helper import CONF as DEVICE_CONF_KEYS
from bec_lib.config_helper import ConfigHelper
from bec_lib.logger import bec_logger
from pydantic import field_validator
from qtpy.QtCore import QSize, Qt, QThreadPool, Signal # type: ignore
from qtpy.QtWidgets import (
QApplication,
QDialog,
QDialogButtonBox,
QLabel,
QStackedLayout,
QVBoxLayout,
QWidget,
)
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.forms_from_types.items import DynamicFormItem
from bec_widgets.widgets.services.device_browser.device_item.config_communicator import (
CommunicateConfigAction,
)
from bec_widgets.widgets.services.device_browser.device_item.device_config_form import (
DeviceConfigForm,
)
from bec_widgets.widgets.utility.spinner.spinner import SpinnerWidget
logger = bec_logger.logger
_StdBtn = QDialogButtonBox.StandardButton
def _try_literal_eval(value: str):
if value == "":
return ""
try:
return literal_eval(value)
except SyntaxError as e:
raise ValueError(f"Entered config value {value} is not a valid python value!") from e
class DeviceConfigDialog(QDialog):
RPC = False
applied = Signal()
accepted_data = Signal(dict)
def __init__(
self, *, parent=None, class_deviceconfig_item: type[DynamicFormItem] | None = None, **kwargs
):
self._initial_config = {}
self._class_deviceconfig_item = class_deviceconfig_item
super().__init__(parent=parent, **kwargs)
self._container = QStackedLayout()
self._container.setStackingMode(QStackedLayout.StackingMode.StackAll)
self._layout = QVBoxLayout()
self._data = {}
self._add_form()
self._add_overlay()
self._add_buttons()
self.setWindowTitle("Add new device")
self.setLayout(self._container)
self._overlay_widget.setVisible(False)
self._form._validity.setVisible(True)
self._connect_form()
def _connect_form(self):
self._form.validity_proc.connect(self.enable_buttons_for_validity)
self._form.validate_form()
def _add_form(self):
self._form_widget = QWidget()
self._form_widget.setLayout(self._layout)
self._form = DeviceConfigForm()
self._layout.addWidget(self._form)
self._container.addWidget(self._form_widget)
def _add_overlay(self):
self._overlay_widget = QWidget()
self._overlay_widget.setStyleSheet("background-color:rgba(128,128,128,128);")
self._overlay_widget.setAutoFillBackground(True)
self._overlay_layout = QVBoxLayout()
self._overlay_layout.setAlignment(Qt.AlignmentFlag.AlignCenter)
self._overlay_widget.setLayout(self._overlay_layout)
self._spinner = SpinnerWidget(parent=self)
self._spinner.setMinimumSize(QSize(100, 100))
self._overlay_layout.addWidget(self._spinner)
self._container.addWidget(self._overlay_widget)
def _add_buttons(self):
self.button_box = QDialogButtonBox(_StdBtn.Apply | _StdBtn.Ok | _StdBtn.Cancel)
self.button_box.button(_StdBtn.Apply).clicked.connect(self.apply)
self.button_box.accepted.connect(self.accept)
self.button_box.rejected.connect(self.reject)
self._layout.addWidget(self.button_box)
def _fill_form(self):
self._form.set_data(DeviceConfigModel.model_validate(self._initial_config))
def updated_config(self):
new_config = self._form.get_form_data()
diff = {
k: v for k, v in new_config.items() if self._initial_config.get(k) != new_config.get(k)
}
if self._initial_config.get("deviceConfig") in [{}, None] and new_config.get(
"deviceConfig"
) in [{}, None]:
diff.pop("deviceConfig", None)
if diff.get("deviceConfig") is not None:
# TODO: special cased in some parts of device manager but not others, should
# be removed in config update as with below issue
diff["deviceConfig"].pop("device_access", None)
# TODO: replace when https://github.com/bec-project/bec/issues/528 https://github.com/bec-project/bec/issues/547 are resolved
diff["deviceConfig"] = {
k: _try_literal_eval(str(v)) for k, v in diff["deviceConfig"].items() if k != ""
}
# Due to above issues, if deviceConfig changes we must remove and recreate the device - so we need the whole config
if "deviceConfig" in diff:
new_config["deviceConfig"] = diff["deviceConfig"]
return new_config
return diff
@SafeSlot(bool)
def enable_buttons_for_validity(self, valid: bool):
# TODO: replace when https://github.com/bec-project/bec/issues/528 https://github.com/bec-project/bec/issues/547 are resolved
for button in [self.button_box.button(b) for b in [_StdBtn.Apply, _StdBtn.Ok]]:
button.setEnabled(valid)
button.setToolTip(self._form._validity_message.toolTip())
def _process_action(self):
self.accepted_data.emit(self._form.get_form_data())
def get_data(self):
return self._data
@SafeSlot(popup_error=True)
def apply(self):
self._process_action()
self.applied.emit()
@SafeSlot(popup_error=True)
def accept(self):
self._process_action()
return super().accept()
class DirectUpdateDeviceConfigDialog(BECWidget, DeviceConfigDialog):
def __init__(
self,
*,
parent=None,
device: str | None = None,
config_helper: ConfigHelper | None = None,
action: Literal["update"] | Literal["add"] = "update",
threadpool: QThreadPool | None = None,
**kwargs,
):
"""A dialog to edit the configuration of a device in BEC. Generated from the pydantic model
for device specification in bec_lib.atlas_models.
Args:
parent (QObject): the parent QObject
device (str | None): the name of the device. used with the "update" action to prefill the dialog and validate entries.
config_helper (ConfigHelper | None): a ConfigHelper object for communication with Redis, will be created if necessary.
action (Literal["update", "add"]): the action which the form should perform on application or acceptance.
"""
self._device = device
self._q_threadpool = threadpool or QThreadPool()
self._config_helper = config_helper or ConfigHelper(
self.client.connector, self.client._service_name
)
super().__init__(parent=parent, **kwargs)
self.get_bec_shortcuts()
self._action: Literal["update", "add"] = action
user_warning = QLabel(
"Warning: edit items here at your own risk - minimal validation is applied to the entered values.\n"
"Items in the deviceConfig dictionary should correspond to python literals, e.g. numbers, lists, strings (including quotes), etc."
)
user_warning.setWordWrap(True)
user_warning.setStyleSheet("QLabel { color: red; }")
self._layout.insertWidget(0, user_warning)
self.setWindowTitle(
f"Edit config for: {device}" if action == "update" else "Add new device"
)
if self._action == "update":
self._modify_for_update()
self._form.validity_proc.disconnect(self.enable_buttons_for_validity)
else:
self._set_schema_to_check_devices()
# TODO: replace when https://github.com/bec-project/bec/issues/528 https://github.com/bec-project/bec/issues/547 are resolved
# self._form._validity.setVisible(True)
def _modify_for_update(self):
for row in self._form.enumerate_form_widgets():
if row.label.property("_model_field_name") in DEVICE_CONF_KEYS.NON_UPDATABLE:
row.widget._set_pretty_display()
if self._device in self.dev:
self._fetch_config()
self._fill_form()
self._form._validity.setVisible(False)
def _set_schema_to_check_devices(self):
class _NameValidatedConfigModel(DeviceConfigModel):
@field_validator("name")
@staticmethod
def _validate_name(value: str, *_):
if not value.isidentifier():
raise ValueError(
f"Invalid device name: {value}. Device names must be valid Python identifiers."
)
if value in self.dev:
raise ValueError(f"A device with name {value} already exists!")
return value
self._form.set_schema(_NameValidatedConfigModel)
def _fetch_config(self):
if self.dev is not None and (device := self.dev.get(self._device)) is not None: # type: ignore
self._initial_config = device._config
def _process_action(self):
updated_config = self.updated_config()
if self._action == "add":
if self.dev is not None and (name := updated_config.get("name")) in self.dev:
raise ValueError(
f"Can't create a new device with the same name as already existing device {name}!"
)
self._proc_device_config_change(updated_config)
else:
if updated_config == {}:
logger.info("No changes made to device config")
return
self._proc_device_config_change(updated_config)
def _proc_device_config_change(self, config: dict):
logger.info(f"Sending request to {self._action} device config: {config}")
self._start_waiting_display()
communicate_update = CommunicateConfigAction(
self._config_helper, self._device, config, self._action
)
communicate_update.signals.error.connect(self.update_error)
communicate_update.signals.done.connect(self.update_done)
self._q_threadpool.start(communicate_update)
@SafeSlot()
def update_done(self):
self._stop_waiting_display()
if self._action == "update":
self._fetch_config()
self._fill_form()
@SafeSlot(Exception, popup_error=True)
def update_error(self, e: Exception):
self._stop_waiting_display()
if self._action == "update":
self._fetch_config()
self._fill_form()
raise e
def _start_waiting_display(self):
self._overlay_widget.setVisible(True)
self._spinner.start()
def _stop_waiting_display(self):
self._overlay_widget.setVisible(False)
self._spinner.stop()
def main(): # pragma: no cover
import sys
from qtpy.QtWidgets import QApplication, QLineEdit, QPushButton, QWidget
from bec_widgets.utils.colors import apply_theme
dialog = None
app = QApplication(sys.argv)
apply_theme("light")
widget = QWidget()
widget.setLayout(layout := QVBoxLayout())
device = QLineEdit()
layout.addWidget(device)
def _destroy_dialog(*_):
nonlocal dialog
dialog = None
def accept(*args):
logger.success(f"submitted device config form {dialog} {args}")
_destroy_dialog()
def _show_dialog(*_):
nonlocal dialog
if dialog is None:
kwargs = {"device": dev} if (dev := device.text()) else {"action": "add"}
dialog = DirectUpdateDeviceConfigDialog(**kwargs) # type: ignore
dialog.accepted.connect(accept)
dialog.rejected.connect(_destroy_dialog)
dialog.open()
button = QPushButton("Show device dialog")
layout.addWidget(button)
button.clicked.connect(_show_dialog)
widget.show()
sys.exit(app.exec_())
if __name__ == "__main__":
main()
bec_widgets/widgets/services/device_browser/device_item/device_config_form.py
-76
View File
@@ -1,76 +0,0 @@
from __future__ import annotations
from functools import partial
from bec_lib.atlas_models import Device as DeviceConfigModel
from pydantic import BaseModel
from bec_widgets.utils.colors import get_theme_name
from bec_widgets.utils.forms_from_types import styles
from bec_widgets.utils.forms_from_types.forms import PydanticModelForm, PydanticModelFormItem
from bec_widgets.utils.forms_from_types.items import (
DEFAULT_WIDGET_TYPES,
BoolFormItem,
BoolToggleFormItem,
DictFormItem,
FormItemSpec,
)
class DeviceConfigForm(PydanticModelForm):
RPC = False
PLUGIN = False
def __init__(
self,
parent=None,
client=None,
pretty_display=False,
class_deviceconfig_item: type[BaseModel] | None = None,
**kwargs,
):
super().__init__(
parent=parent,
data_model=DeviceConfigModel,
pretty_display=pretty_display,
client=client,
**kwargs,
)
self._class_deviceconfig_item: type[BaseModel] | None = class_deviceconfig_item
self._widget_types = DEFAULT_WIDGET_TYPES.copy()
self._widget_types["bool"] = (lambda spec: spec.item_type is bool, BoolToggleFormItem)
self._widget_types["optional_bool"] = (
lambda spec: spec.item_type == bool | None,
BoolFormItem,
)
pred, _ = self._widget_types["dict"]
self._widget_types["dict"] = pred, self._custom_device_config_item
self._validity.setVisible(True)
self.populate()
def _post_init(self): ...
def _custom_device_config_item(self, spec: FormItemSpec):
if spec.name != "deviceConfig":
return DictFormItem
if self._class_deviceconfig_item is not None:
return partial(PydanticModelFormItem, model=self._class_deviceconfig_item)
return DictFormItem
def set_pretty_display_theme(self, theme: str | None = None):
if theme is None:
theme = get_theme_name()
self.setStyleSheet(styles.pretty_display_theme(theme))
self._validity.setVisible(False)
def get_form_data(self):
"""Get the entered metadata as a dict."""
return self._md_schema.model_validate(super().get_form_data()).model_dump()
def set_schema(self, schema: type[BaseModel]):
if not issubclass(schema, DeviceConfigModel):
raise TypeError("This class doesn't support changing the schema")
super().set_schema(schema)
def set_data(self, data: DeviceConfigModel): # type: ignore # This class locks the type
super().set_data(data)
bec_widgets/widgets/services/device_browser/device_item/device_item.py
-220
View File
@@ -1,220 +0,0 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from bec_lib.atlas_models import Device as DeviceConfigModel
from bec_lib.config_helper import ConfigHelper
from bec_lib.devicemanager import DeviceContainer
from bec_lib.logger import bec_logger
from bec_lib.messages import ConfigAction
from bec_qthemes import material_icon
from qtpy.QtCore import QMimeData, QSize, Qt, QThreadPool, Signal
from qtpy.QtGui import QDrag
from qtpy.QtWidgets import QApplication, QHBoxLayout, QTabWidget, QToolButton, QVBoxLayout, QWidget
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.expandable_frame import ExpandableGroupFrame
from bec_widgets.widgets.services.device_browser.device_item.config_communicator import (
CommunicateConfigAction,
)
from bec_widgets.widgets.services.device_browser.device_item.device_config_dialog import (
DirectUpdateDeviceConfigDialog,
)
from bec_widgets.widgets.services.device_browser.device_item.device_config_form import (
DeviceConfigForm,
)
from bec_widgets.widgets.services.device_browser.device_item.device_signal_display import (
SignalDisplay,
)
if TYPE_CHECKING: # pragma: no cover
from qtpy.QtGui import QMouseEvent
logger = bec_logger.logger
class DeviceItem(ExpandableGroupFrame):
RPC = False
def __init__(
self,
*,
parent,
device: str,
devices: DeviceContainer,
icon: str = "",
config_helper: ConfigHelper,
q_threadpool: QThreadPool | None = None,
) -> None:
super().__init__(parent, title=device, expanded=False, icon=icon)
self.dev = devices
self._drag_pos = None
self._expanded_first_time = False
self._data = None
self.device = device
self._layout = QHBoxLayout()
self._layout.setContentsMargins(0, 0, 0, 0)
self._tab_widget = QTabWidget(tabShape=QTabWidget.TabShape.Rounded)
self._tab_widget.setDocumentMode(True)
self._layout.addWidget(self._tab_widget)
self._form_page = QWidget(parent=self)
self._form_page_layout = QVBoxLayout()
self._form_page.setLayout(self._form_page_layout)
self._signal_page = QWidget(parent=self)
self._signal_page_layout = QVBoxLayout()
self._signal_page.setLayout(self._signal_page_layout)
self._tab_widget.addTab(self._form_page, "Configuration")
self._tab_widget.addTab(self._signal_page, "Signals")
self._config_helper = config_helper
self._q_threadpool = q_threadpool or QThreadPool()
self.set_layout(self._layout)
self.adjustSize()
def _create_title_layout(self, title: str, icon: str):
super()._create_title_layout(title, icon)
self.edit_button = QToolButton()
self.edit_button.setIcon(material_icon(icon_name="edit", size=(15, 15)))
self._title_layout.insertWidget(self._title_layout.count() - 1, self.edit_button)
self.edit_button.clicked.connect(self._create_edit_dialog)
self.delete_button = QToolButton()
self.delete_button.setIcon(material_icon(icon_name="delete", size=(15, 15)))
self._title_layout.insertWidget(self._title_layout.count() - 1, self.delete_button)
self.delete_button.clicked.connect(self._delete_device)
@SafeSlot()
def _create_edit_dialog(self):
dialog = DirectUpdateDeviceConfigDialog(
parent=self,
device=self.device,
config_helper=self._config_helper,
threadpool=self._q_threadpool,
)
dialog.accepted.connect(self._reload_config)
dialog.applied.connect(self._reload_config)
dialog.open()
@SafeSlot()
def _delete_device(self):
self.expanded = False
deleter = CommunicateConfigAction(self._config_helper, self.device, None, "remove")
deleter.signals.error.connect(self._deletion_error)
deleter.signals.done.connect(self._deletion_done)
self._q_threadpool.start(deleter)
@SafeSlot(Exception, popup_error=True)
def _deletion_error(self, e: Exception):
raise e
@SafeSlot()
def _deletion_done(self):
self.imminent_deletion.emit()
self.deleteLater()
@SafeSlot()
def switch_expanded_state(self):
if not self.expanded and not self._expanded_first_time:
self._expanded_first_time = True
self.form = DeviceConfigForm(parent=self, pretty_display=True)
self._form_page_layout.addWidget(self.form)
self.signals = SignalDisplay(parent=self, device=self.device)
self._signal_page_layout.addWidget(self.signals)
self._reload_config()
self.broadcast_size_hint.emit(self.sizeHint())
super().switch_expanded_state()
if self._expanded_first_time:
self.form.adjustSize()
self.updateGeometry()
if self._expanded:
self.form.set_pretty_display_theme()
self.adjustSize()
self.broadcast_size_hint.emit(self.sizeHint())
@SafeSlot(bool)
def set_editable(self, enabled: bool):
self.edit_button.setEnabled(enabled)
self.delete_button.setEnabled(enabled)
@SafeSlot(str, dict)
def config_update(self, action: ConfigAction, content: dict) -> None:
if self.device in content:
self._reload_config()
@SafeSlot(popup_error=True)
def _reload_config(self, *_):
self.set_display_config(self.dev[self.device]._config)
def set_display_config(self, config_dict: dict):
"""Set the displayed information from a device config dict, which must conform to the
bec_lib.atlas_models.Device config model."""
self._data = DeviceConfigModel.model_validate(config_dict)
if self._expanded_first_time:
self.form.set_data(self._data)
def mousePressEvent(self, event: QMouseEvent) -> None:
super().mousePressEvent(event)
if event.button() == Qt.LeftButton:
self._drag_pos = event.pos()
def mouseMoveEvent(self, event: QMouseEvent) -> None:
if not (event.buttons() and Qt.LeftButton):
return
if (event.pos() - self._drag_pos).manhattanLength() < QApplication.startDragDistance():
return
drag = QDrag(self)
mime_data = QMimeData()
mime_data.setText(self.device)
drag.setMimeData(mime_data)
drag.exec_(Qt.MoveAction)
def mouseDoubleClickEvent(self, event: QMouseEvent) -> None:
logger.debug("Double Clicked")
# TODO: Implement double click action for opening the device properties dialog
return super().mouseDoubleClickEvent(event)
if __name__ == "__main__": # pragma: no cover
import sys
from unittest.mock import MagicMock
from qtpy.QtWidgets import QApplication
from bec_widgets.widgets.services.device_browser.device_item.device_config_form import (
DeviceConfigForm,
)
from bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button import DarkModeButton
app = QApplication(sys.argv)
widget = QWidget()
layout = QHBoxLayout()
widget.setLayout(layout)
mock_config = {
"name": "Test Device",
"enabled": True,
"deviceClass": "FakeDeviceClass",
"deviceConfig": {"kwarg1": "value1"},
"readoutPriority": "baseline",
"description": "A device for testing out a widget",
"readOnly": True,
"softwareTrigger": False,
"deviceTags": {"tag1", "tag2", "tag3"},
"userParameter": {"some_setting": "some_ value"},
}
item = DeviceItem(
parent=widget,
device="Device",
devices={"Device": MagicMock(enabled=True, _config=mock_config)}, # type: ignore
config_helper=ConfigHelper(MagicMock()),
)
layout.addWidget(DarkModeButton())
layout.addWidget(item)
widget.show()
sys.exit(app.exec_())
bec_widgets/widgets/services/device_browser/device_item/device_signal_display.py
-119
View File
@@ -1,119 +0,0 @@
from bec_lib.device import Device
from bec_qthemes import material_icon
from qtpy.QtCore import Qt
from qtpy.QtWidgets import QHBoxLayout, QLabel, QToolButton, QVBoxLayout, QWidget
from bec_widgets.utils.bec_connector import ConnectionConfig
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
from bec_widgets.utils.ophyd_kind_util import Kind
from bec_widgets.widgets.utility.signal_label.signal_label import SignalLabel
class SignalDisplay(BECWidget, QWidget):
RPC = False
def __init__(
self,
parent=None,
client=None,
device: str = "",
config: ConnectionConfig = None,
gui_id: str | None = None,
**kwargs,
):
"""A widget to display all the signals from a given device, and allow getting
a fresh reading."""
super().__init__(parent=parent, client=client, config=config, gui_id=gui_id, **kwargs)
self.get_bec_shortcuts()
self._layout = QVBoxLayout()
self.setLayout(self._layout)
self._content = QWidget()
self._layout.addWidget(self._content)
self._device = device
self.device = device
@SafeSlot()
def _refresh(self):
if (dev := self.dev.get(self.device)) is not None:
dev.read(cached=True)
dev.read_configuration(cached=True)
def _add_refresh_button(self):
button_holder = QWidget()
button_holder.setLayout(QHBoxLayout())
button_holder.layout().setAlignment(Qt.AlignmentFlag.AlignRight)
button_holder.layout().setContentsMargins(0, 0, 0, 0)
refresh_button = QToolButton()
refresh_button.setIcon(
material_icon(icon_name="refresh", size=(20, 20), convert_to_pixmap=False)
)
refresh_button.clicked.connect(self._refresh)
button_holder.layout().addWidget(refresh_button)
self._content_layout.addWidget(button_holder)
def _populate(self):
self._content.deleteLater()
self._content = QWidget()
self._layout.addWidget(self._content)
self._content_layout = QVBoxLayout()
self._content_layout.setContentsMargins(0, 0, 0, 0)
self._content.setLayout(self._content_layout)
self._add_refresh_button()
if self._device in self.dev:
if isinstance(self.dev[self.device], Device):
for sig, info in self.dev[self.device]._info.get("signals", {}).items():
if info.get("kind_str") in [
Kind.hinted.name,
Kind.normal.name,
Kind.config.name,
]:
self._content_layout.addWidget(
SignalLabel(
parent=self,
device=self._device,
signal=sig,
show_select_button=False,
show_default_units=True,
)
)
else:
self._content_layout.addWidget(
SignalLabel(
parent=self,
device=self._device,
signal=self._device,
show_select_button=False,
show_default_units=True,
)
)
self._content_layout.addStretch(1)
else:
self._content_layout.addWidget(
QLabel(f"Device {self.device} not found in device manager!")
)
@SafeProperty(str)
def device(self):
return self._device
@device.setter
def device(self, value: str):
self._device = value
self._populate()
if __name__ == "__main__": # pragma: no cover
import sys
from qtpy.QtWidgets import QApplication
from bec_widgets.utils.colors import apply_theme
app = QApplication(sys.argv)
apply_theme("light")
widget = SignalDisplay(device="samx")
widget.show()
sys.exit(app.exec_())
bec_widgets/widgets/services/device_browser/register_device_browser.py
-17
View File
@@ -1,17 +0,0 @@
def main(): # pragma: no cover
from qtpy import PYSIDE6
if not PYSIDE6:
print("PYSIDE6 is not available in the environment. Cannot patch designer.")
return
from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection
from bec_widgets.widgets.services.device_browser.device_browser_plugin import (
DeviceBrowserPlugin,
)
QPyDesignerCustomWidgetCollection.addCustomWidget(DeviceBrowserPlugin())
if __name__ == "__main__": # pragma: no cover
main()
bec_widgets/widgets/services/device_browser/util.py
-11
View File
@@ -1,11 +0,0 @@
from bec_lib.device import Device
def map_device_type_to_icon(device_obj: Device) -> str:
"""Associate device types with material icon names"""
match device_obj._info.get("device_base_class", "").lower():
case "positioner":
return "precision_manufacturing"
case "signal":
return "vital_signs"
return "deployed_code"
tests/end-2-end/user_interaction/test_user_interaction_e2e.py
-15
View File
@@ -203,21 +203,6 @@ def test_widgets_e2e_dap_combo_box(qtbot, connected_client_gui_obj, random_gener
maybe_remove_dock_area(qtbot, gui=gui, random_int_gen=random_generator_from_seed)
@pytest.mark.timeout(PYTEST_TIMEOUT)
def test_widgets_e2e_device_browser(qtbot, connected_client_gui_obj, random_generator_from_seed):
"""Test the DeviceBrowser widget."""
gui = connected_client_gui_obj
bec = gui._client
# Create dock_area and widget
widget = create_widget(qtbot, gui, gui.available_widgets.DeviceBrowser)
widget: client.DeviceBrowser
# No rpc calls yet to check
# Test removing the widget, or leaving it open for the next test
maybe_remove_dock_area(qtbot, gui=gui, random_int_gen=random_generator_from_seed)
@pytest.mark.timeout(PYTEST_TIMEOUT)
def test_widgets_e2e_image(qtbot, connected_client_gui_obj, random_generator_from_seed):
"""Test the Image widget."""
tests/unit_tests/test_config_communicator.py
+1 -1
View File
@@ -3,7 +3,7 @@ from unittest.mock import ANY, MagicMock, call
import pytest
from bec_lib.config_helper import ConfigHelper
from bec_widgets.widgets.services.device_browser.device_item.config_communicator import (
from bec_widgets.widgets.control.device_manager.components.config_communicator import (
CommunicateConfigAction,
)
tests/unit_tests/test_device_browser.py
-201
View File
@@ -1,201 +0,0 @@
from typing import TYPE_CHECKING
from unittest import mock
import pytest
from bec_lib.device import Device
from qtpy.QtCore import QPoint, Qt
from qtpy.QtWidgets import QTabWidget
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.services.device_browser.device_browser import DeviceBrowser
from bec_widgets.widgets.services.device_browser.device_item.device_config_form import (
DeviceConfigForm,
)
from bec_widgets.widgets.services.device_browser.device_item.device_signal_display import (
SignalDisplay,
)
if TYPE_CHECKING: # pragma: no cover
from qtpy.QtWidgets import QListWidgetItem
from bec_widgets.widgets.services.device_browser.device_item import DeviceItem
# pylint: disable=no-member
# pylint: disable=missing-function-docstring
# pylint: disable=redefined-outer-name
# pylint: disable=protected-access
@pytest.fixture
def device_browser(qtbot, mocked_client):
dev_browser = DeviceBrowser(client=mocked_client)
dev_browser.dev["samx"].read_configuration = mock.MagicMock()
qtbot.addWidget(dev_browser)
qtbot.waitExposed(dev_browser)
yield dev_browser
def test_device_browser_init_with_devices(device_browser: DeviceBrowser):
"""
Test that the device browser is initialized with the correct number of devices.
"""
device_list = device_browser.dev_list
assert device_list.count() == len(device_browser.dev)
@pytest.mark.parametrize(
["search_term", "expected_num_visible"],
[("sam", 3), ("nonexistent", 0), ("", -1), (r"(\)", -1)],
)
def test_device_browser_filtering(
qtbot, device_browser, search_term: str, expected_num_visible: int
):
"""
Test that the device browser is able to filter the device list.
"""
expected = expected_num_visible if expected_num_visible >= 0 else len(device_browser.dev)
def num_visible(item_dict):
return len(list(filter(lambda i: not i.widget.isHidden(), item_dict.values())))
device_browser.ui.filter_input.setText(search_term)
qtbot.wait(100)
assert num_visible(device_browser.dev_list._item_dict) == expected
def test_device_item_mouse_press_event(device_browser, qtbot):
"""
Test that the mousePressEvent is triggered correctly.
"""
# Simulate a left mouse press event on the device item
device_item: QListWidgetItem = device_browser.dev_list.itemAt(0, 0)
widget: DeviceItem = device_browser.dev_list.itemWidget(device_item)
qtbot.mouseClick(widget._title, Qt.MouseButton.LeftButton)
def test_update_event_captured(device_browser, qtbot):
device_browser.update_device_list = mock.MagicMock()
device_browser.update_device_list.assert_not_called()
device_browser.on_device_update("remove", {})
device_browser.update_device_list.assert_called_once()
device_browser.on_device_update("", {})
def test_device_item_expansion(device_browser, qtbot):
"""
Test that the form is displayed when the item is expanded, and that the expansion is triggered
by clicking on the expansion button, the title, or the device icon
"""
device_item: QListWidgetItem = device_browser.dev_list.itemAt(0, 0)
widget: DeviceItem = device_browser.dev_list.itemWidget(device_item)
qtbot.mouseClick(widget._expansion_button, Qt.MouseButton.LeftButton)
tab_widget: QTabWidget = widget._contents.layout().itemAt(0).widget()
qtbot.waitUntil(lambda: tab_widget.widget(0) is not None, timeout=100)
qtbot.waitUntil(
lambda: isinstance(tab_widget.widget(0).layout().itemAt(0).widget(), DeviceConfigForm),
timeout=100,
)
form = tab_widget.widget(0).layout().itemAt(0).widget()
assert widget.expanded
assert (name_field := form.widget_dict.get("name")) is not None
qtbot.waitUntil(lambda: name_field.getValue() == "aptrx", timeout=500)
qtbot.mouseClick(widget._expansion_button, Qt.MouseButton.LeftButton)
assert not widget.expanded
qtbot.mouseClick(widget._title, Qt.MouseButton.LeftButton)
qtbot.waitUntil(lambda: widget.expanded, timeout=500)
qtbot.mouseClick(widget._title_icon, Qt.MouseButton.LeftButton)
qtbot.waitUntil(lambda: not widget.expanded, timeout=500)
def test_device_item_mouse_press_and_move_events_creates_drag(device_browser, qtbot):
"""
Test that the mousePressEvent is triggered correctly and initiates a drag.
"""
device_item: QListWidgetItem = device_browser.dev_list.itemAt(0, 0)
widget: DeviceItem = device_browser.dev_list.itemWidget(device_item)
device_name = widget.device
with mock.patch("qtpy.QtGui.QDrag.exec_") as mock_exec:
with mock.patch("qtpy.QtGui.QDrag.setMimeData") as mock_set_mimedata:
qtbot.mousePress(widget._title, Qt.MouseButton.LeftButton, pos=QPoint(0, 0))
qtbot.mouseMove(widget, pos=QPoint(10, 10))
qtbot.mouseRelease(widget, Qt.MouseButton.LeftButton)
mock_set_mimedata.assert_called_once()
mock_exec.assert_called_once()
assert mock_set_mimedata.call_args[0][0].text() == device_name
def test_device_item_double_click_event(device_browser, qtbot):
"""
Test that the mouseDoubleClickEvent is triggered correctly.
"""
# Simulate a left mouse press event on the device item
device_item: QListWidgetItem = device_browser.dev_list.itemAt(0, 0)
widget: DeviceItem = device_browser.dev_list.itemWidget(device_item)
qtbot.mouseDClick(widget, Qt.LeftButton)
def test_device_deletion(device_browser, qtbot):
device_item: QListWidgetItem = device_browser.dev_list.itemAt(0, 0)
widget: DeviceItem = device_browser.dev_list.itemWidget(device_item)
widget._config_helper = mock.MagicMock()
assert widget.device in device_browser.dev_list._item_dict
qtbot.mouseClick(widget.delete_button, Qt.LeftButton)
qtbot.waitUntil(lambda: widget.device not in device_browser.dev_list._item_dict, timeout=10000)
def test_signal_display(mocked_client, qtbot):
signal_display = SignalDisplay(client=mocked_client, device="test_device")
qtbot.addWidget(signal_display)
device_mock = mock.MagicMock()
signal_display.dev = {"test_device": device_mock}
signal_display._refresh()
device_mock.read.assert_called()
device_mock.read_configuration.assert_called()
def test_signal_display_no_device(mocked_client, qtbot):
device_mock = mock.MagicMock()
mocked_client.device_manager.devices = {"test_device_1": device_mock}
signal_display = SignalDisplay(client=mocked_client, device="test_device_2")
qtbot.addWidget(signal_display)
assert (
signal_display._content_layout.itemAt(1).widget().text()
== "Device test_device_2 not found in device manager!"
)
signal_display._refresh()
device_mock.read.assert_not_called()
device_mock.read_configuration.assert_not_called()
def test_signal_display_omitted_not_added(mocked_client, qtbot):
device_mock = mock.MagicMock(spec=Device)
device_mock._info = {"signals": {"signal_1": {"kind_str": "omitted"}}}
signal_display = SignalDisplay(client=mocked_client, device="test_device_1")
signal_display.dev = {"test_device_1": device_mock}
signal_display._populate()
qtbot.addWidget(signal_display)
assert signal_display._content_layout.itemAt(1).widget() is None
def test_device_browser_cleanup_detaches_filter_proxy(device_browser: DeviceBrowser, qtbot):
"""
A filter change still queued in the rate-limited filter proxy when the browser is
cleaned up, or made afterwards, must not reach update_device_list of the closed browser.
"""
proxy = device_browser.proxy_device_update
with mock.patch.object(device_browser.dev_list, "update_filter") as update_filter:
device_browser.ui.filter_input.setText("sam") # queued in the proxy
assert proxy.timer.timer.isActive() # pg's ThreadsafeTimer wraps a QTimer
device_browser.close() # closeEvent -> cleanup()
device_browser.ui.filter_input.setText("samx") # changed after cleanup
qtbot.wait(600)
update_filter.assert_not_called()
assert not proxy.timer.timer.isActive()
assert proxy.args is None
tests/unit_tests/test_device_config_form_dialog.py
-169
View File
@@ -1,169 +0,0 @@
from unittest.mock import MagicMock, patch
import pytest
from bec_lib.atlas_models import Device as DeviceConfigModel
from qtpy.QtWidgets import QDialogButtonBox, QPushButton
from bec_widgets.utils.forms_from_types.items import StrFormItem
from bec_widgets.widgets.services.device_browser.device_item.device_config_dialog import (
DirectUpdateDeviceConfigDialog,
_try_literal_eval,
)
_BASIC_CONFIG = {
"name": "test_device",
"enabled": True,
"deviceClass": "TestDevice",
"readoutPriority": "monitored",
}
@pytest.fixture
def mock_client():
mock_device = MagicMock(_config=DeviceConfigModel.model_validate(_BASIC_CONFIG).model_dump())
mock_client = MagicMock()
mock_client.device_manager.devices = {"test_device": mock_device}
return mock_client
@pytest.fixture
def update_dialog(mock_client, qtbot):
"""Fixture to create a DeviceConfigDialog instance."""
update_dialog = DirectUpdateDeviceConfigDialog(
device="test_device", config_helper=MagicMock(), client=mock_client
)
qtbot.addWidget(update_dialog)
return update_dialog
@pytest.fixture
def add_dialog(mock_client, qtbot):
"""Fixture to create a DeviceConfigDialog instance."""
add_dialog = DirectUpdateDeviceConfigDialog(
device=None, config_helper=MagicMock(), client=mock_client, action="add"
)
qtbot.addWidget(add_dialog)
return add_dialog
def test_initialization(update_dialog):
assert update_dialog._device == "test_device"
assert update_dialog._container.count() == 2
def test_fill_form(update_dialog):
with patch.object(update_dialog._form, "set_data") as mock_set_data:
update_dialog._fill_form()
mock_set_data.assert_called_once_with(DeviceConfigModel.model_validate(_BASIC_CONFIG))
def test_updated_config(update_dialog):
"""Test that updated_config returns the correct changes."""
update_dialog._initial_config = {"key1": "value1", "key2": "value2"}
with patch.object(
update_dialog._form, "get_form_data", return_value={"key1": "value1", "key2": "new_value"}
):
updated = update_dialog.updated_config()
assert updated == {"key2": "new_value"}
def test_apply(update_dialog):
with patch.object(update_dialog, "_process_action") as mock_process_update:
update_dialog.apply()
mock_process_update.assert_called_once()
def test_accept(update_dialog):
with (
patch.object(update_dialog, "_process_action") as mock_process_update,
patch("qtpy.QtWidgets.QDialog.accept") as mock_parent_accept,
):
update_dialog.accept()
mock_process_update.assert_called_once()
mock_parent_accept.assert_called_once()
def test_waiting_display(update_dialog, qtbot):
with (
patch.object(update_dialog._spinner, "start") as mock_spinner_start,
patch.object(update_dialog._spinner, "stop") as mock_spinner_stop,
):
update_dialog.show()
update_dialog._start_waiting_display()
qtbot.waitUntil(update_dialog._overlay_widget.isVisible, timeout=100)
mock_spinner_start.assert_called_once()
mock_spinner_stop.assert_not_called()
update_dialog._stop_waiting_display()
qtbot.waitUntil(lambda: not update_dialog._overlay_widget.isVisible(), timeout=100)
mock_spinner_stop.assert_called_once()
def test_update_cycle(update_dialog):
update = {"enabled": False, "readoutPriority": "baseline", "deviceTags": {"tag"}}
def _mock_send(action="update", config=None, wait_for_response=True, timeout_s=None):
device = update_dialog.client.device_manager.devices["test_device"]
device._config = {**device._config, **config["test_device"]} # type: ignore
update_dialog._q_threadpool = MagicMock()
update_dialog._q_threadpool.start.side_effect = lambda runnable: runnable.run()
update_dialog._config_helper.send_config_request = MagicMock(side_effect=_mock_send)
for item in update_dialog._form.enumerate_form_widgets():
if (val := update.get(item.label.property("_model_field_name"))) is not None:
item.widget.setValue(val)
assert update_dialog.updated_config() == update
update_dialog.apply()
update_dialog._q_threadpool.start.assert_called_once()
update_dialog._config_helper.send_config_request.assert_called_with(
action="update", config={"test_device": update}, wait_for_response=False
)
@pytest.mark.parametrize(
["changes", "result"],
[
({}, {}),
({"readOnly": True}, {"readOnly": True}),
({"readOnly": False}, {}),
({"readOnly": True, "description": "test"}, {"readOnly": True, "description": "test"}),
(
{"deviceConfig": {"param1": "'val1'"}},
DeviceConfigModel(
enabled=True,
deviceClass="TestDevice",
deviceConfig={"param1": "val1"},
readoutPriority="monitored",
name="test_device",
).model_dump(),
),
({"deviceConfig": {}}, {}),
],
)
def test_update_with_modified_deviceconfig(update_dialog, changes, result):
for k, v in changes.items():
update_dialog._form.widget_dict[k].setValue(v)
assert update_dialog.updated_config() == result
def test_add_form_init_without_name(add_dialog, qtbot):
assert (name_widget := add_dialog._form.widget_dict.get("name")) is not None
assert isinstance(name_widget, StrFormItem)
assert name_widget.getValue() is None
def test_add_form_validates_and_disables_on_init(add_dialog, qtbot):
assert (ok_button := add_dialog.button_box.button(QDialogButtonBox.Ok)) is not None
assert isinstance(ok_button, QPushButton)
assert not ok_button.isEnabled()
def test_try_literal_eval():
assert _try_literal_eval("") == ""
assert _try_literal_eval("[1, 2, 3]") == [1, 2, 3]
assert _try_literal_eval('"[,,]"') == "[,,]"
with pytest.raises(ValueError) as e:
_try_literal_eval("[,,]")
assert e.match("Entered config value [,,]")