mirror of
https://github.com/bec-project/bec_widgets.git
synced 2025-07-12 18:51:50 +02:00
feat(device_browser): device deletion from config
This commit is contained in:
@ -3,11 +3,12 @@ import re
|
||||
from functools import partial
|
||||
|
||||
from bec_lib.callback_handler import EventType
|
||||
from bec_lib.config_helper import ConfigHelper
|
||||
from bec_lib.logger import bec_logger
|
||||
from bec_lib.messages import ConfigAction
|
||||
from bec_qthemes import material_icon
|
||||
from pyqtgraph import SignalProxy
|
||||
from qtpy.QtCore import QSize, Signal
|
||||
from qtpy.QtCore import QSize, QThreadPool, Signal
|
||||
from qtpy.QtWidgets import QListWidget, QListWidgetItem, QVBoxLayout, QWidget
|
||||
|
||||
from bec_widgets.cli.rpc.rpc_register import RPCRegister
|
||||
@ -42,6 +43,8 @@ class DeviceBrowser(BECWidget, QWidget):
|
||||
) -> 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._q_threadpool = QThreadPool()
|
||||
self.ui = None
|
||||
self.ini_ui()
|
||||
self.dev_list: QListWidget = self.ui.device_list
|
||||
@ -88,29 +91,40 @@ class DeviceBrowser(BECWidget, QWidget):
|
||||
self.dev_list.clear()
|
||||
self._device_items: dict[str, QListWidgetItem] = {}
|
||||
|
||||
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):
|
||||
def _updatesize(item: QListWidgetItem, device_item: DeviceItem):
|
||||
device_item.adjustSize()
|
||||
item.setSizeHint(QSize(device_item.width(), device_item.height()))
|
||||
logger.debug(f"Adjusting {item} size to {device_item.width(), device_item.height()}")
|
||||
|
||||
with RPCRegister.delayed_broadcast():
|
||||
for device, device_obj in self.dev.items():
|
||||
item = QListWidgetItem(self.dev_list)
|
||||
device_item = DeviceItem(
|
||||
parent=self,
|
||||
device=device,
|
||||
devices=self.dev,
|
||||
icon=map_device_type_to_icon(device_obj),
|
||||
)
|
||||
device_item.expansion_state_changed.connect(partial(_updatesize, item, device_item))
|
||||
tooltip = self.dev[device]._config.get("description", "")
|
||||
device_item.setToolTip(tooltip)
|
||||
device_item.broadcast_size_hint.connect(item.setSizeHint)
|
||||
item.setSizeHint(device_item.sizeHint())
|
||||
def _remove_item(item: QListWidgetItem):
|
||||
self.dev_list.takeItem(self.dev_list.row(item))
|
||||
del self._device_items[device]
|
||||
self.dev_list.sortItems()
|
||||
|
||||
self.dev_list.setItemWidget(item, device_item)
|
||||
self.dev_list.addItem(item)
|
||||
self._device_items[device] = item
|
||||
item = QListWidgetItem(self.dev_list)
|
||||
device_item = DeviceItem(
|
||||
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,
|
||||
)
|
||||
device_item.expansion_state_changed.connect(partial(_updatesize, item, device_item))
|
||||
device_item.imminent_deletion.connect(partial(_remove_item, item))
|
||||
tooltip = self.dev[device]._config.get("description", "")
|
||||
device_item.setToolTip(tooltip)
|
||||
device_item.broadcast_size_hint.connect(item.setSizeHint)
|
||||
item.setSizeHint(device_item.sizeHint())
|
||||
|
||||
self.dev_list.setItemWidget(item, device_item)
|
||||
self.dev_list.addItem(item)
|
||||
self._device_items[device] = item
|
||||
|
||||
@SafeSlot()
|
||||
def reset_device_list(self) -> None:
|
||||
@ -129,6 +143,10 @@ class DeviceBrowser(BECWidget, QWidget):
|
||||
Either way, the function will filter the devices based on the filter input text and update the device list.
|
||||
"""
|
||||
filter_text = self.ui.filter_input.text()
|
||||
for device in self.dev:
|
||||
if device not in self._device_items:
|
||||
# it is possible the device has just been added to the config
|
||||
self._add_item_to_list(device, self.dev[device])
|
||||
try:
|
||||
self.regex = re.compile(filter_text, re.IGNORECASE)
|
||||
except re.error:
|
||||
|
||||
@ -0,0 +1,60 @@
|
||||
from bec_lib.config_helper import ConfigHelper
|
||||
from bec_lib.logger import bec_logger
|
||||
from bec_lib.messages import ConfigAction
|
||||
from qtpy.QtCore import QObject, QRunnable, Signal
|
||||
|
||||
from bec_widgets.utils.error_popups import SafeSlot
|
||||
|
||||
logger = bec_logger.logger
|
||||
|
||||
|
||||
class _CommSignals(QObject):
|
||||
error = Signal(Exception)
|
||||
done = Signal()
|
||||
|
||||
|
||||
class CommunicateConfigAction(QRunnable):
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
config_helper: ConfigHelper,
|
||||
device: str | None,
|
||||
config: dict | None,
|
||||
action: ConfigAction,
|
||||
) -> None:
|
||||
super().__init__()
|
||||
self.config_helper = config_helper
|
||||
self.device = device
|
||||
self.config = config
|
||||
self.action = action
|
||||
self.signals = _CommSignals()
|
||||
|
||||
@SafeSlot()
|
||||
def run(self):
|
||||
try:
|
||||
if self.action in ["add", "update", "remove"]:
|
||||
if (dev_name := self.device or self.config.get("name")) is None:
|
||||
raise ValueError(
|
||||
"Must be updating a device or be supplied a name for a new device"
|
||||
)
|
||||
req_args = {
|
||||
"action": self.action,
|
||||
"config": {dev_name: self.config},
|
||||
"wait_for_response": False,
|
||||
}
|
||||
timeout = (
|
||||
self.config_helper.suggested_timeout_s(self.config)
|
||||
if self.config is not None
|
||||
else 20
|
||||
)
|
||||
RID = self.config_helper.send_config_request(**req_args)
|
||||
logger.info("Waiting for config reply")
|
||||
reply = self.config_helper.wait_for_config_reply(RID, timeout=timeout)
|
||||
self.config_helper.handle_update_reply(reply, RID, timeout)
|
||||
logger.info("Done updating config!")
|
||||
else:
|
||||
raise ValueError(f"action {self.action} is not supported")
|
||||
except Exception as e:
|
||||
self.signals.error.emit(e)
|
||||
else:
|
||||
self.signals.done.emit()
|
||||
@ -5,7 +5,7 @@ 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 qtpy.QtCore import QObject, QRunnable, QSize, Qt, QThreadPool, Signal
|
||||
from qtpy.QtCore import QSize, Qt, QThreadPool, Signal
|
||||
from qtpy.QtWidgets import (
|
||||
QApplication,
|
||||
QDialog,
|
||||
@ -18,6 +18,9 @@ from qtpy.QtWidgets import (
|
||||
|
||||
from bec_widgets.utils.bec_widget import BECWidget
|
||||
from bec_widgets.utils.error_popups import SafeSlot
|
||||
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,
|
||||
)
|
||||
@ -26,53 +29,18 @@ from bec_widgets.widgets.utility.spinner.spinner import SpinnerWidget
|
||||
logger = bec_logger.logger
|
||||
|
||||
|
||||
class _CommSignals(QObject):
|
||||
error = Signal(Exception)
|
||||
done = Signal()
|
||||
|
||||
|
||||
class _CommunicateUpdate(QRunnable):
|
||||
|
||||
def __init__(self, config_helper: ConfigHelper, device: str, config: dict, action: str) -> None:
|
||||
super().__init__()
|
||||
self.config_helper = config_helper
|
||||
self.device = device
|
||||
self.config = config
|
||||
self.action = action
|
||||
self.signals = _CommSignals()
|
||||
|
||||
@SafeSlot()
|
||||
def run(self):
|
||||
try:
|
||||
if (dev_name := self.device or self.config.get("name")) is None:
|
||||
raise ValueError("Must be updating a device or be supplied a name for a new device")
|
||||
req_args = {
|
||||
"action": self.action,
|
||||
"config": {dev_name: self.config},
|
||||
"wait_for_response": False,
|
||||
}
|
||||
timeout = self.config_helper.suggested_timeout_s(self.config)
|
||||
RID = self.config_helper.send_config_request(**req_args)
|
||||
logger.info("Waiting for config reply")
|
||||
reply = self.config_helper.wait_for_config_reply(RID, timeout=timeout)
|
||||
self.config_helper.handle_update_reply(reply, RID, timeout)
|
||||
logger.info("Done updating config!")
|
||||
except Exception as e:
|
||||
self.signals.error.emit(e)
|
||||
finally:
|
||||
self.signals.done.emit()
|
||||
|
||||
|
||||
class DeviceConfigDialog(BECWidget, QDialog):
|
||||
RPC = False
|
||||
applied = Signal()
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
parent=None,
|
||||
device: str | None = None,
|
||||
config_helper: ConfigHelper | None = None,
|
||||
action: Literal["update", "add"] = "update",
|
||||
threadpool: QThreadPool | None = None,
|
||||
**kwargs,
|
||||
):
|
||||
"""A dialog to edit the configuration of a device in BEC. Generated from the pydantic model
|
||||
@ -89,9 +57,9 @@ class DeviceConfigDialog(BECWidget, QDialog):
|
||||
self._config_helper = config_helper or ConfigHelper(
|
||||
self.client.connector, self.client._service_name
|
||||
)
|
||||
self.threadpool = QThreadPool()
|
||||
self._device = device
|
||||
self._action = action
|
||||
self._q_threadpool = threadpool or QThreadPool()
|
||||
self.setWindowTitle(f"Edit config for: {device}")
|
||||
self._container = QStackedLayout()
|
||||
self._container.setStackingMode(QStackedLayout.StackAll)
|
||||
@ -216,12 +184,12 @@ class DeviceConfigDialog(BECWidget, QDialog):
|
||||
logger.info(f"Sending request to update device config: {config}")
|
||||
|
||||
self._start_waiting_display()
|
||||
communicate_update = _CommunicateUpdate(
|
||||
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.threadpool.start(communicate_update)
|
||||
self._q_threadpool.start(communicate_update)
|
||||
|
||||
@SafeSlot()
|
||||
def update_done(self):
|
||||
|
||||
@ -3,15 +3,19 @@ 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_qthemes import material_icon
|
||||
from qtpy.QtCore import QMimeData, QSize, Qt, Signal
|
||||
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 (
|
||||
DeviceConfigDialog,
|
||||
)
|
||||
@ -31,10 +35,20 @@ logger = bec_logger.logger
|
||||
|
||||
class DeviceItem(ExpandableGroupFrame):
|
||||
broadcast_size_hint = Signal(QSize)
|
||||
imminent_deletion = Signal()
|
||||
|
||||
RPC = False
|
||||
|
||||
def __init__(self, parent, device: str, devices: DeviceContainer, icon: str = "") -> None:
|
||||
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
|
||||
@ -48,35 +62,64 @@ class DeviceItem(ExpandableGroupFrame):
|
||||
self._tab_widget.setDocumentMode(True)
|
||||
self._layout.addWidget(self._tab_widget)
|
||||
|
||||
self.set_layout(self._layout)
|
||||
|
||||
self._form_page = QWidget()
|
||||
self._form_page = QWidget(parent=self)
|
||||
self._form_page_layout = QVBoxLayout()
|
||||
self._form_page.setLayout(self._form_page_layout)
|
||||
|
||||
self._signal_page = QWidget()
|
||||
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=(10, 10), convert_to_pixmap=False)
|
||||
)
|
||||
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 = DeviceConfigDialog(parent=self, device=self.device)
|
||||
dialog = DeviceConfigDialog(
|
||||
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 RuntimeError(f"Failed to delete device {self.device}") from 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:
|
||||
@ -157,7 +200,12 @@ if __name__ == "__main__": # pragma: no cover
|
||||
"deviceTags": {"tag1", "tag2", "tag3"},
|
||||
"userParameter": {"some_setting": "some_ value"},
|
||||
}
|
||||
item = DeviceItem(widget, "Device", {"Device": MagicMock(enabled=True, _config=mock_config)})
|
||||
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()
|
||||
|
||||
28
tests/unit_tests/test_config_communicator.py
Normal file
28
tests/unit_tests/test_config_communicator.py
Normal file
@ -0,0 +1,28 @@
|
||||
from unittest.mock import ANY, MagicMock
|
||||
|
||||
from bec_lib.config_helper import ConfigHelper
|
||||
|
||||
from bec_widgets.widgets.services.device_browser.device_item.config_communicator import (
|
||||
CommunicateConfigAction,
|
||||
)
|
||||
|
||||
|
||||
def test_must_have_a_name(qtbot):
|
||||
error_occurred = False
|
||||
|
||||
def oops():
|
||||
nonlocal error_occurred
|
||||
error_occurred = True
|
||||
|
||||
c = CommunicateConfigAction(ConfigHelper(MagicMock()), device=None, config={}, action="add")
|
||||
c.signals.error.connect(oops)
|
||||
c.run()
|
||||
qtbot.waitUntil(lambda: error_occurred, timeout=100)
|
||||
|
||||
|
||||
def test_wait_for_reply_on_RID():
|
||||
ch = MagicMock(spec=ConfigHelper)
|
||||
ch.send_config_request.return_value = "abcde"
|
||||
cca = CommunicateConfigAction(config_helper=ch, device="samx", config={}, action="update")
|
||||
cca.run()
|
||||
ch.wait_for_config_reply.assert_called_with("abcde", timeout=ANY)
|
||||
@ -132,3 +132,13 @@ def test_device_item_double_click_event(device_browser, qtbot):
|
||||
device_item: QListWidgetItem = device_browser.ui.device_list.itemAt(0, 0)
|
||||
widget: DeviceItem = device_browser.ui.device_list.itemWidget(device_item)
|
||||
qtbot.mouseDClick(widget, Qt.LeftButton)
|
||||
|
||||
|
||||
def test_device_deletion(device_browser, qtbot):
|
||||
device_item: QListWidgetItem = device_browser.ui.device_list.itemAt(0, 0)
|
||||
widget: DeviceItem = device_browser.ui.device_list.itemWidget(device_item)
|
||||
widget._config_helper = mock.MagicMock()
|
||||
|
||||
assert widget.device in device_browser._device_items
|
||||
qtbot.mouseClick(widget.delete_button, Qt.LeftButton)
|
||||
qtbot.waitUntil(lambda: widget.device not in device_browser._device_items, timeout=10000)
|
||||
|
||||
@ -2,8 +2,7 @@ from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from bec_lib.atlas_models import Device as DeviceConfigModel
|
||||
from PySide6.QtWidgets import QPushButton
|
||||
from qtpy.QtWidgets import QDialogButtonBox, QLineEdit
|
||||
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 (
|
||||
|
||||
Reference in New Issue
Block a user