mirror of
https://github.com/bec-project/bec_widgets.git
synced 2026-04-10 18:50:55 +02:00
Compare commits
6 Commits
feat/dm_ma
...
feat/avail
| Author | SHA1 | Date | |
|---|---|---|---|
| fb452fe9b0 | |||
| 495405240e | |||
| 42d93f9446 | |||
| c06eb36701 | |||
| 91988c4fef | |||
| 787bfe09ef |
@@ -1,67 +0,0 @@
|
||||
from qtpy import QtCore, QtWidgets
|
||||
|
||||
from bec_widgets.examples.device_manager_view.device_manager_view import DeviceManagerView
|
||||
from bec_widgets.widgets.containers.advanced_dock_area.advanced_dock_area import AdvancedDockArea
|
||||
|
||||
|
||||
class BECMainApp(QtWidgets.QWidget):
|
||||
def __init__(self, parent=None):
|
||||
super().__init__(parent)
|
||||
|
||||
# Main layout
|
||||
layout = QtWidgets.QVBoxLayout(self)
|
||||
layout.setContentsMargins(0, 0, 0, 0)
|
||||
layout.setSpacing(0)
|
||||
|
||||
# Tab widget as central area
|
||||
self.tabs = QtWidgets.QTabWidget(self)
|
||||
self.tabs.setContentsMargins(0, 0, 0, 0)
|
||||
self.tabs.setTabPosition(QtWidgets.QTabWidget.West) # Tabs on the left side
|
||||
|
||||
layout.addWidget(self.tabs)
|
||||
# Add DM
|
||||
self._add_device_manager_view()
|
||||
|
||||
# Add Plot area
|
||||
self._add_ad_dockarea()
|
||||
|
||||
# Adjust size of tab bar
|
||||
# TODO not yet properly working, tabs a spread across the full length, to be checked!
|
||||
tab_bar = self.tabs.tabBar()
|
||||
tab_bar.setFixedWidth(tab_bar.sizeHint().width())
|
||||
|
||||
def _add_device_manager_view(self) -> None:
|
||||
self.device_manager_view = DeviceManagerView(parent=self)
|
||||
self.add_tab(self.device_manager_view, "Device Manager")
|
||||
|
||||
def _add_ad_dockarea(self) -> None:
|
||||
self.advanced_dock_area = AdvancedDockArea(parent=self)
|
||||
self.add_tab(self.advanced_dock_area, "Plot Area")
|
||||
|
||||
def add_tab(self, widget: QtWidgets.QWidget, title: str):
|
||||
"""Add a custom QWidget as a tab."""
|
||||
tab_container = QtWidgets.QWidget()
|
||||
tab_layout = QtWidgets.QVBoxLayout(tab_container)
|
||||
tab_layout.setContentsMargins(0, 0, 0, 0)
|
||||
tab_layout.setSpacing(0)
|
||||
|
||||
tab_layout.addWidget(widget)
|
||||
self.tabs.addTab(tab_container, title)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
|
||||
from bec_lib.bec_yaml_loader import yaml_load
|
||||
from bec_qthemes import apply_theme
|
||||
|
||||
app = QtWidgets.QApplication(sys.argv)
|
||||
apply_theme("light")
|
||||
win = BECMainApp()
|
||||
config_path = "/Users/appel_c/work_psi_awi/bec_workspace/csaxs_bec/csaxs_bec/device_configs/first_light.yaml"
|
||||
cfg = yaml_load(config_path)
|
||||
cfg.update({"device_will_fail": {"name": "device_will_fail", "some_param": 1}})
|
||||
win.device_manager_view.device_table_view.set_device_config(cfg)
|
||||
win.resize(1920, 1080)
|
||||
win.show()
|
||||
sys.exit(app.exec_())
|
||||
@@ -1,39 +1,35 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from typing import TYPE_CHECKING, List
|
||||
from typing import List
|
||||
|
||||
import PySide6QtAds as QtAds
|
||||
import yaml
|
||||
from bec_lib.bec_yaml_loader import yaml_load
|
||||
from bec_lib.file_utils import DeviceConfigWriter
|
||||
from bec_lib.logger import bec_logger
|
||||
from bec_lib.plugin_helper import plugin_package_name, plugin_repo_path
|
||||
from bec_qthemes import apply_theme
|
||||
from bec_qthemes import material_icon
|
||||
from PySide6QtAds import CDockManager, CDockWidget
|
||||
from qtpy.QtCore import Qt, QTimer
|
||||
from qtpy.QtWidgets import QFileDialog, QMessageBox, QSplitter, QVBoxLayout, QWidget
|
||||
from qtpy.QtWidgets import (
|
||||
QPushButton,
|
||||
QSizePolicy,
|
||||
QSplitter,
|
||||
QStackedLayout,
|
||||
QTreeWidget,
|
||||
QVBoxLayout,
|
||||
QWidget,
|
||||
)
|
||||
|
||||
from bec_widgets import BECWidget
|
||||
from bec_widgets.utils.error_popups import SafeSlot
|
||||
from bec_widgets.utils.toolbars.actions import MaterialIconAction
|
||||
from bec_widgets.utils.toolbars.bundles import ToolbarBundle
|
||||
from bec_widgets.utils.toolbars.toolbar import ModularToolBar
|
||||
from bec_widgets.widgets.containers.advanced_dock_area.advanced_dock_area import AdvancedDockArea
|
||||
from bec_widgets.widgets.control.device_manager.components import (
|
||||
DeviceTableView,
|
||||
DMConfigView,
|
||||
DMOphydTest,
|
||||
DocstringView,
|
||||
)
|
||||
from bec_widgets.widgets.control.device_manager.components.available_device_resources.available_device_resources import (
|
||||
AvailableDeviceResources,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from bec_lib.client import BECClient
|
||||
|
||||
logger = bec_logger.logger
|
||||
from bec_widgets.widgets.control.device_manager.components.device_table_view import DeviceTableView
|
||||
from bec_widgets.widgets.control.device_manager.components.dm_config_view import DMConfigView
|
||||
from bec_widgets.widgets.control.device_manager.components.dm_ophyd_test import (
|
||||
DeviceManagerOphydTest,
|
||||
)
|
||||
from bec_widgets.widgets.editors.monaco.monaco_widget import MonacoWidget
|
||||
from bec_widgets.widgets.editors.web_console.web_console import WebConsole
|
||||
from bec_widgets.widgets.services.bec_status_box.bec_status_box import BECStatusBox
|
||||
from bec_widgets.widgets.utility.ide_explorer.ide_explorer import IDEExplorer
|
||||
|
||||
|
||||
def set_splitter_weights(splitter: QSplitter, weights: List[float]) -> None:
|
||||
@@ -73,7 +69,7 @@ def set_splitter_weights(splitter: QSplitter, weights: List[float]) -> None:
|
||||
class DeviceManagerView(BECWidget, QWidget):
|
||||
|
||||
def __init__(self, parent=None, *args, **kwargs):
|
||||
super().__init__(parent=parent, client=None, *args, **kwargs)
|
||||
super().__init__(parent, *args, **kwargs)
|
||||
|
||||
# Top-level layout hosting a toolbar and the dock manager
|
||||
self._root_layout = QVBoxLayout(self)
|
||||
@@ -82,52 +78,40 @@ class DeviceManagerView(BECWidget, QWidget):
|
||||
self.dock_manager = CDockManager(self)
|
||||
self._root_layout.addWidget(self.dock_manager)
|
||||
|
||||
# Available Resources Widget
|
||||
# Initialize the widgets
|
||||
self.available_devices = AvailableDeviceResources(self)
|
||||
self.available_devices_dock = QtAds.CDockWidget("Available Devices", self)
|
||||
self.device_table_view = DeviceTableView(self)
|
||||
# Placeholder
|
||||
self.dm_config_view = DMConfigView(self)
|
||||
|
||||
# Placeholder for ophyd test
|
||||
WebConsole.startup_cmd = "ipython"
|
||||
self.ophyd_test = DeviceManagerOphydTest(self)
|
||||
self.ophyd_test_dock = QtAds.CDockWidget("Ophyd Test", self)
|
||||
self.ophyd_test_dock.setWidget(self.ophyd_test)
|
||||
|
||||
# Create the dock widgets
|
||||
self.available_devices_dock = QtAds.CDockWidget("Explorer", self)
|
||||
self.available_devices_dock.setWidget(self.available_devices)
|
||||
|
||||
# Device Table View widget
|
||||
self.device_table_view = DeviceTableView(self)
|
||||
self.device_table_view_dock = QtAds.CDockWidget("Device Table", self)
|
||||
self.device_table_view_dock.setWidget(self.device_table_view)
|
||||
|
||||
# Device Config View widget
|
||||
self.dm_config_view = DMConfigView(self)
|
||||
self.dm_config_view_dock = QtAds.CDockWidget("Device Config View", self)
|
||||
# Device Table will be central widget
|
||||
self.dock_manager.setCentralWidget(self.device_table_view_dock)
|
||||
|
||||
self.dm_config_view_dock = QtAds.CDockWidget("YAML Editor", self)
|
||||
self.dm_config_view_dock.setWidget(self.dm_config_view)
|
||||
|
||||
# Docstring View
|
||||
self.dm_docs_view = DocstringView(self)
|
||||
self.dm_docs_view_dock = QtAds.CDockWidget("Docstring View", self)
|
||||
self.dm_docs_view_dock.setWidget(self.dm_docs_view)
|
||||
|
||||
# Ophyd Test view
|
||||
self.ophyd_test_view = DMOphydTest(self)
|
||||
self.ophyd_test_dock_view = QtAds.CDockWidget("Ophyd Test View", self)
|
||||
self.ophyd_test_dock_view.setWidget(self.ophyd_test_view)
|
||||
|
||||
# Arrange widgets within the QtAds dock manager
|
||||
|
||||
# Central widget area
|
||||
self.central_dock_area = self.dock_manager.setCentralWidget(self.device_table_view_dock)
|
||||
# Add the dock widgets to the dock manager
|
||||
self.dock_manager.addDockWidget(
|
||||
QtAds.DockWidgetArea.BottomDockWidgetArea,
|
||||
self.dm_docs_view_dock,
|
||||
self.central_dock_area,
|
||||
)
|
||||
|
||||
# Left Area
|
||||
self.left_dock_area = self.dock_manager.addDockWidget(
|
||||
QtAds.DockWidgetArea.LeftDockWidgetArea, self.available_devices_dock
|
||||
)
|
||||
self.dock_manager.addDockWidget(
|
||||
QtAds.DockWidgetArea.BottomDockWidgetArea, self.dm_config_view_dock, self.left_dock_area
|
||||
monaco_yaml_area = self.dock_manager.addDockWidget(
|
||||
QtAds.DockWidgetArea.RightDockWidgetArea, self.dm_config_view_dock
|
||||
)
|
||||
|
||||
# Right area
|
||||
self.dock_manager.addDockWidget(
|
||||
QtAds.DockWidgetArea.RightDockWidgetArea, self.ophyd_test_dock_view
|
||||
QtAds.DockWidgetArea.BottomDockWidgetArea, self.ophyd_test_dock, monaco_yaml_area
|
||||
)
|
||||
|
||||
for dock in self.dock_manager.dockWidgets():
|
||||
@@ -143,251 +127,12 @@ class DeviceManagerView(BECWidget, QWidget):
|
||||
area.titleBar().setVisible(False)
|
||||
|
||||
# Apply stretch after the layout is done
|
||||
self.set_default_view([2, 8, 2], [3, 1])
|
||||
# self.set_default_view([2, 8, 2], [2, 2, 4])
|
||||
self.set_default_view([2, 5, 3], [5, 5])
|
||||
|
||||
# Connect slots
|
||||
self.device_table_view.selected_device.connect(self.dm_config_view.on_select_config)
|
||||
self.device_table_view.selected_device.connect(self.dm_docs_view.on_select_config)
|
||||
self.ophyd_test_view.device_validated.connect(
|
||||
self.device_table_view.update_device_validation
|
||||
)
|
||||
self.device_table_view.device_configs_added.connect(self.ophyd_test_view.add_device_configs)
|
||||
|
||||
self._add_toolbar()
|
||||
|
||||
def _add_toolbar(self):
|
||||
self.toolbar = ModularToolBar(self)
|
||||
|
||||
# Add IO actions
|
||||
self._add_io_actions()
|
||||
self._add_table_actions()
|
||||
self.toolbar.show_bundles(["IO", "Table"])
|
||||
self._root_layout.insertWidget(0, self.toolbar)
|
||||
|
||||
def _add_io_actions(self):
|
||||
# Create IO bundle
|
||||
io_bundle = ToolbarBundle("IO", self.toolbar.components)
|
||||
|
||||
# Add load config from plugin dir
|
||||
self.toolbar.add_bundle(io_bundle)
|
||||
|
||||
load = MaterialIconAction(
|
||||
icon_name="file_open", parent=self, tooltip="Load configuration file from disk"
|
||||
)
|
||||
self.toolbar.components.add_safe("load", load)
|
||||
load.action.triggered.connect(self._load_file_action)
|
||||
io_bundle.add_action("load")
|
||||
|
||||
# Add safe to disk
|
||||
safe_to_disk = MaterialIconAction(
|
||||
icon_name="file_save", parent=self, tooltip="Save config to disk"
|
||||
)
|
||||
self.toolbar.components.add_safe("safe_to_disk", safe_to_disk)
|
||||
safe_to_disk.action.triggered.connect(self._safe_to_disk_action)
|
||||
io_bundle.add_action("safe_to_disk")
|
||||
|
||||
# Add load config from redis
|
||||
load_redis = MaterialIconAction(
|
||||
icon_name="cached", parent=self, tooltip="Load current config from Redis"
|
||||
)
|
||||
load_redis.action.triggered.connect(self._load_redis_action)
|
||||
self.toolbar.components.add_safe("load_redis", load_redis)
|
||||
io_bundle.add_action("load_redis")
|
||||
|
||||
# Update config action
|
||||
update_config_redis = MaterialIconAction(
|
||||
icon_name="cloud_upload", parent=self, tooltip="Update current config in Redis"
|
||||
)
|
||||
update_config_redis.action.triggered.connect(self._update_redis_action)
|
||||
self.toolbar.components.add_safe("update_config_redis", update_config_redis)
|
||||
io_bundle.add_action("update_config_redis")
|
||||
|
||||
# Table actions
|
||||
|
||||
def _add_table_actions(self) -> None:
|
||||
table_bundle = ToolbarBundle("Table", self.toolbar.components)
|
||||
|
||||
# Add load config from plugin dir
|
||||
self.toolbar.add_bundle(table_bundle)
|
||||
|
||||
# Reset composed view
|
||||
reset_composed = MaterialIconAction(
|
||||
icon_name="delete_sweep", parent=self, tooltip="Reset current composed config view"
|
||||
)
|
||||
reset_composed.action.triggered.connect(self._reset_composed_view)
|
||||
self.toolbar.components.add_safe("reset_composed", reset_composed)
|
||||
table_bundle.add_action("reset_composed")
|
||||
|
||||
# Add device
|
||||
add_device = MaterialIconAction(icon_name="add", parent=self, tooltip="Add new device")
|
||||
add_device.action.triggered.connect(self._add_device_action)
|
||||
self.toolbar.components.add_safe("add_device", add_device)
|
||||
table_bundle.add_action("add_device")
|
||||
|
||||
# Remove device
|
||||
remove_device = MaterialIconAction(icon_name="remove", parent=self, tooltip="Remove device")
|
||||
remove_device.action.triggered.connect(self._remove_device_action)
|
||||
self.toolbar.components.add_safe("remove_device", remove_device)
|
||||
table_bundle.add_action("remove_device")
|
||||
|
||||
# Rerun validation
|
||||
rerun_validation = MaterialIconAction(
|
||||
icon_name="checklist", parent=self, tooltip="Run device validation on selected devices"
|
||||
)
|
||||
rerun_validation.action.triggered.connect(self._rerun_validation_action)
|
||||
self.toolbar.components.add_safe("rerun_validation", rerun_validation)
|
||||
table_bundle.add_action("rerun_validation")
|
||||
|
||||
# Most likly, no actions on available devices
|
||||
# Actions (vielleicht bundle fuer available devices )
|
||||
# - reset composed view
|
||||
# - add new device (EpicsMotor, EpicsMotorECMC, EpicsSignal, CustomDevice)
|
||||
# - remove device
|
||||
# - rerun validation (with/without connect)
|
||||
|
||||
# IO actions
|
||||
|
||||
@SafeSlot()
|
||||
def _load_file_action(self):
|
||||
"""Action for the 'load' action to load a config from disk for the io_bundle of the toolbar."""
|
||||
# Check if plugin repo is installed...
|
||||
try:
|
||||
plugin_path = plugin_repo_path()
|
||||
plugin_name = plugin_package_name()
|
||||
config_path = os.path.join(plugin_path, plugin_name, "device_configs")
|
||||
except ValueError:
|
||||
# Get the recovery config path as fallback
|
||||
config_path = self._get_recovery_config_path()
|
||||
logger.warning(
|
||||
f"No plugin repository installed, fallback to recovery config path: {config_path}"
|
||||
)
|
||||
|
||||
# Implement the file loading logic here
|
||||
start_dir = os.path.abspath(config_path)
|
||||
file_path, _ = QFileDialog.getOpenFileName(
|
||||
self, caption="Select Config File", dir=start_dir
|
||||
)
|
||||
if file_path:
|
||||
try:
|
||||
config = yaml_load(file_path)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to load config from file {file_path}. Error: {e}")
|
||||
return
|
||||
self.device_table_view.set_device_config(
|
||||
config
|
||||
) # TODO ADD QDialog with 'replace', 'add' & 'cancel'
|
||||
|
||||
# TODO would we ever like to add the current config to an existing composition
|
||||
@SafeSlot()
|
||||
def _load_redis_action(self):
|
||||
"""Action for the 'load_redis' action to load the current config from Redis for the io_bundle of the toolbar."""
|
||||
reply = QMessageBox.question(
|
||||
self,
|
||||
"Load currently active config",
|
||||
"Do you really want to flush the current config and reload?",
|
||||
QMessageBox.Yes | QMessageBox.No,
|
||||
QMessageBox.No,
|
||||
)
|
||||
if reply == QMessageBox.Yes:
|
||||
cfg = {}
|
||||
config_list = self.client.device_manager._get_redis_device_config()
|
||||
for item in config_list:
|
||||
k = item["name"]
|
||||
item.pop("name")
|
||||
cfg[k] = item
|
||||
self.device_table_view.set_device_config(cfg)
|
||||
else:
|
||||
return
|
||||
|
||||
@SafeSlot()
|
||||
def _safe_to_disk_action(self):
|
||||
"""Action for the 'safe_to_disk' action to save the current config to disk."""
|
||||
# Check if plugin repo is installed...
|
||||
try:
|
||||
config_path = self._get_recovery_config_path()
|
||||
except ValueError:
|
||||
# Get the recovery config path as fallback
|
||||
config_path = os.path.abspath(os.path.expanduser("~"))
|
||||
logger.warning(f"Failed to find recovery config path, fallback to: {config_path}")
|
||||
|
||||
# Implement the file loading logic here
|
||||
file_path, _ = QFileDialog.getSaveFileName(
|
||||
self, caption="Save Config File", dir=config_path
|
||||
)
|
||||
if file_path:
|
||||
config = self.device_table_view.get_device_config()
|
||||
with open(file_path, "w") as file:
|
||||
file.write(yaml.dump(config))
|
||||
|
||||
# TODO add here logic, should be asyncronous, but probably block UI, and show a loading spinner. If failed, it should report..
|
||||
@SafeSlot()
|
||||
def _update_redis_action(self):
|
||||
"""Action for the 'update_redis' action to update the current config in Redis."""
|
||||
config = self.device_table_view.get_device_config()
|
||||
reply = QMessageBox.question(
|
||||
self,
|
||||
"Not implemented yet",
|
||||
"This feature has not been implemented yet, will be coming soon...!!",
|
||||
QMessageBox.Cancel,
|
||||
QMessageBox.Cancel,
|
||||
)
|
||||
|
||||
# Table actions
|
||||
|
||||
@SafeSlot()
|
||||
def _reset_composed_view(self):
|
||||
"""Action for the 'reset_composed_view' action to reset the composed view."""
|
||||
reply = QMessageBox.question(
|
||||
self,
|
||||
"Clear View",
|
||||
"You are about to clear the current composed config view, please confirm...",
|
||||
QMessageBox.Yes | QMessageBox.No,
|
||||
QMessageBox.No,
|
||||
)
|
||||
if reply == QMessageBox.Yes:
|
||||
self.device_table_view.clear_device_configs()
|
||||
|
||||
# TODO Here we would like to implement a custom popup view, that allows to add new devices
|
||||
# We want to have a combobox to choose from EpicsMotor, EpicsMotorECMC, EpicsSignal, EpicsSignalRO, and maybe EpicsSignalWithRBV and custom Device
|
||||
# For all default Epics devices, we would like to preselect relevant fields, and prompt them with the proper deviceConfig args already, i.e. 'prefix', 'read_pv', 'write_pv' etc..
|
||||
# For custom Device, they should receive all options. It might be cool to get a side panel with docstring view of the class upon inspecting it to make it easier in case deviceConfig entries are required..
|
||||
@SafeSlot()
|
||||
def _add_device_action(self):
|
||||
"""Action for the 'add_device' action to add a new device."""
|
||||
# Implement the logic to add a new device
|
||||
reply = QMessageBox.question(
|
||||
self,
|
||||
"Not implemented yet",
|
||||
"This feature has not been implemented yet, will be coming soon...!!",
|
||||
QMessageBox.Cancel,
|
||||
QMessageBox.Cancel,
|
||||
)
|
||||
|
||||
# TODO fix the device table remove actions. This is currently not working properly...
|
||||
@SafeSlot()
|
||||
def _remove_device_action(self):
|
||||
"""Action for the 'remove_device' action to remove a device."""
|
||||
reply = QMessageBox.question(
|
||||
self,
|
||||
"Not implemented yet",
|
||||
"This feature has not been implemented yet, will be coming soon...!!",
|
||||
QMessageBox.Cancel,
|
||||
QMessageBox.Cancel,
|
||||
)
|
||||
|
||||
# TODO implement proper logic for validation. We should also carefully review how these jobs update the table, and how we can cancel pending validations
|
||||
# in case they are no longer relevant. We might want to 'block' the interactivity on the items for which validation runs with 'connect'!
|
||||
@SafeSlot()
|
||||
def _rerun_validation_action(self):
|
||||
"""Action for the 'rerun_validation' action to rerun validation on selected devices."""
|
||||
# Implement the logic to rerun validation on selected devices
|
||||
reply = QMessageBox.question(
|
||||
self,
|
||||
"Not implemented yet",
|
||||
"This feature has not been implemented yet, will be coming soon...!!",
|
||||
QMessageBox.Cancel,
|
||||
QMessageBox.Cancel,
|
||||
self.device_table_view.model.devices_reset.connect(
|
||||
self.available_devices.update_devices_state
|
||||
)
|
||||
|
||||
####### Default view has to be done with setting up splitters ########
|
||||
@@ -452,40 +197,18 @@ class DeviceManagerView(BECWidget, QWidget):
|
||||
v = [1, 1]
|
||||
self.set_default_view(h, v)
|
||||
|
||||
def _get_recovery_config_path(self) -> str:
|
||||
"""Get the recovery config path from the log_writer config."""
|
||||
# pylint: disable=protected-access
|
||||
log_writer_config: BECClient = self.client._service_config.config.get("log_writer", {})
|
||||
writer = DeviceConfigWriter(service_config=log_writer_config)
|
||||
return os.path.abspath(os.path.expanduser(writer.get_recovery_directory()))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
from copy import deepcopy
|
||||
|
||||
from bec_lib.bec_yaml_loader import yaml_load
|
||||
from qtpy.QtWidgets import QApplication
|
||||
|
||||
from bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button import DarkModeButton
|
||||
|
||||
app = QApplication(sys.argv)
|
||||
w = QWidget()
|
||||
l = QVBoxLayout()
|
||||
w.setLayout(l)
|
||||
apply_theme("dark")
|
||||
button = DarkModeButton()
|
||||
l.addWidget(button)
|
||||
device_manager_view = DeviceManagerView()
|
||||
l.addWidget(device_manager_view)
|
||||
config_path = "/Users/appel_c/work_psi_awi/bec_workspace/csaxs_bec/csaxs_bec/device_configs/first_light.yaml"
|
||||
cfg = yaml_load(config_path)
|
||||
cfg.update({"device_will_fail": {"name": "device_will_fail", "some_param": 1}})
|
||||
|
||||
# config = device_manager_view.client.device_manager._get_redis_device_config()
|
||||
device_manager_view.device_table_view.set_device_config(cfg)
|
||||
w.show()
|
||||
w.setWindowTitle("Device Manager View")
|
||||
w.resize(1920, 1080)
|
||||
config = device_manager_view.client.device_manager._get_redis_device_config()
|
||||
device_manager_view.device_table_view.set_device_config(config)
|
||||
device_manager_view.show()
|
||||
device_manager_view.setWindowTitle("Device Manager View")
|
||||
device_manager_view.resize(1200, 800)
|
||||
# developer_view.set_stretch(horizontal=[1, 3, 2], vertical=[5, 5]) #can be set during runtime
|
||||
sys.exit(app.exec_())
|
||||
|
||||
@@ -2,10 +2,6 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
|
||||
from bec_lib.bec_yaml_loader import yaml_load
|
||||
from bec_lib.logger import bec_logger
|
||||
from bec_qthemes import material_icon
|
||||
from qtpy import QtCore, QtWidgets
|
||||
|
||||
@@ -13,8 +9,6 @@ from bec_widgets.examples.device_manager_view.device_manager_view import DeviceM
|
||||
from bec_widgets.utils.bec_widget import BECWidget
|
||||
from bec_widgets.utils.error_popups import SafeSlot
|
||||
|
||||
logger = bec_logger.logger
|
||||
|
||||
|
||||
class DeviceManagerWidget(BECWidget, QtWidgets.QWidget):
|
||||
|
||||
@@ -47,50 +41,19 @@ class DeviceManagerWidget(BECWidget, QtWidgets.QWidget):
|
||||
self._overlay_widget.setSizePolicy(
|
||||
QtWidgets.QSizePolicy.Policy.Expanding, QtWidgets.QSizePolicy.Policy.Expanding
|
||||
)
|
||||
# Load current config
|
||||
self.button_load_current_config = QtWidgets.QPushButton("Load Current Config")
|
||||
icon = material_icon(icon_name="database", size=(24, 24), convert_to_pixmap=False)
|
||||
self.button_load_current_config.setIcon(icon)
|
||||
self._overlay_layout.addWidget(self.button_load_current_config)
|
||||
self.button_load_current_config.clicked.connect(self._load_config_clicked)
|
||||
# Load config from disk
|
||||
self.button_load_config_from_file = QtWidgets.QPushButton("Load Config From File")
|
||||
icon = material_icon(icon_name="folder", size=(24, 24), convert_to_pixmap=False)
|
||||
self.button_load_config_from_file.setIcon(icon)
|
||||
self._overlay_layout.addWidget(self.button_load_config_from_file)
|
||||
self.button_load_config_from_file.clicked.connect(self._load_config_from_file_clicked)
|
||||
self._overlay_widget.setVisible(True)
|
||||
|
||||
def _load_config_from_file_clicked(self):
|
||||
"""Handle click on 'Load Config From File' button."""
|
||||
start_dir = os.path.expanduser("~")
|
||||
file_path, _ = QtWidgets.QFileDialog.getOpenFileName(
|
||||
self, caption="Select Config File", dir=start_dir
|
||||
)
|
||||
if file_path:
|
||||
self._load_config_from_file(file_path)
|
||||
|
||||
def _load_config_from_file(self, file_path: str):
|
||||
try:
|
||||
config = yaml_load(file_path)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to load config from file {file_path}. Error: {e}")
|
||||
return
|
||||
config_list = []
|
||||
for name, cfg in config.items():
|
||||
config_list.append(cfg)
|
||||
config_list[-1]["name"] = name
|
||||
self.device_manager_view.device_table_view.set_device_config(config_list)
|
||||
# self.device_manager_view.ophyd_test.on_device_config_update(config)
|
||||
self.stacked_layout.setCurrentWidget(self.device_manager_view)
|
||||
|
||||
@SafeSlot()
|
||||
def _load_config_clicked(self):
|
||||
"""Handle click on 'Load Current Config' button."""
|
||||
config = self.client.device_manager._get_redis_device_config()
|
||||
config.append({"name": "wrong_device", "some_value": 1})
|
||||
self.device_manager_view.device_table_view.set_device_config(config)
|
||||
# self.device_manager_view.ophyd_test.on_device_config_update(config)
|
||||
self.device_manager_view.ophyd_test.on_device_config_update(config)
|
||||
self.stacked_layout.setCurrentWidget(self.device_manager_view)
|
||||
|
||||
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from bec_qthemes import material_icon
|
||||
from qtpy.QtCore import Signal
|
||||
from qtpy.QtCore import QSize, Signal
|
||||
from qtpy.QtWidgets import (
|
||||
QApplication,
|
||||
QFrame,
|
||||
@@ -19,7 +19,8 @@ from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
|
||||
|
||||
|
||||
class ExpandableGroupFrame(QFrame):
|
||||
|
||||
broadcast_size_hint = Signal(QSize)
|
||||
imminent_deletion = Signal()
|
||||
expansion_state_changed = Signal()
|
||||
|
||||
EXPANDED_ICON_NAME: str = "collapse_all"
|
||||
@@ -31,6 +32,7 @@ class ExpandableGroupFrame(QFrame):
|
||||
super().__init__(parent=parent)
|
||||
self._expanded = expanded
|
||||
|
||||
self._title_text = f"<b>{title}</b>"
|
||||
self.setFrameStyle(QFrame.Shape.StyledPanel | QFrame.Shadow.Plain)
|
||||
self.setSizePolicy(QSizePolicy.Policy.Minimum, QSizePolicy.Policy.Minimum)
|
||||
self._layout = QVBoxLayout()
|
||||
@@ -49,21 +51,27 @@ class ExpandableGroupFrame(QFrame):
|
||||
def _create_title_layout(self, title: str, icon: str):
|
||||
self._title_layout = QHBoxLayout()
|
||||
self._layout.addLayout(self._title_layout)
|
||||
self._internal_title_layout = QHBoxLayout()
|
||||
self._title_layout.addLayout(self._internal_title_layout)
|
||||
|
||||
self._title = ClickableLabel(f"<b>{title}</b>")
|
||||
self._title = ClickableLabel()
|
||||
self._set_title_text(self._title_text)
|
||||
self._title_icon = ClickableLabel()
|
||||
self._title_layout.addWidget(self._title_icon)
|
||||
self._title_layout.addWidget(self._title)
|
||||
self._internal_title_layout.addWidget(self._title_icon)
|
||||
self._internal_title_layout.addWidget(self._title)
|
||||
self.icon_name = icon
|
||||
self._title.clicked.connect(self.switch_expanded_state)
|
||||
self._title_icon.clicked.connect(self.switch_expanded_state)
|
||||
|
||||
self._title_layout.addStretch(1)
|
||||
self._internal_title_layout.addStretch(1)
|
||||
|
||||
self._expansion_button = QToolButton()
|
||||
self._update_expansion_icon()
|
||||
self._title_layout.addWidget(self._expansion_button, stretch=1)
|
||||
|
||||
def get_title_layout(self) -> QHBoxLayout:
|
||||
return self._internal_title_layout
|
||||
|
||||
def set_layout(self, layout: QLayout) -> None:
|
||||
self._contents.setLayout(layout)
|
||||
self._contents.layout().setContentsMargins(0, 0, 0, 0) # type: ignore
|
||||
@@ -112,6 +120,18 @@ class ExpandableGroupFrame(QFrame):
|
||||
else:
|
||||
self._title_icon.setVisible(False)
|
||||
|
||||
@SafeProperty(str)
|
||||
def title_text(self): # type: ignore
|
||||
return self._title_text
|
||||
|
||||
@title_text.setter
|
||||
def title_text(self, title_text: str):
|
||||
self._title_text = title_text
|
||||
self._set_title_text(self._title_text)
|
||||
|
||||
def _set_title_text(self, title_text: str):
|
||||
self._title.setText(title_text)
|
||||
|
||||
|
||||
# Application example
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
|
||||
133
bec_widgets/utils/list_of_expandable_frames.py
Normal file
133
bec_widgets/utils/list_of_expandable_frames.py
Normal file
@@ -0,0 +1,133 @@
|
||||
import re
|
||||
from functools import partial
|
||||
from re import Pattern
|
||||
from typing import Generic, Iterable, NamedTuple, TypeVar
|
||||
|
||||
from bec_lib.logger import bec_logger
|
||||
from more_itertools import consume
|
||||
from PySide6.QtGui import QFont
|
||||
from PySide6.QtWidgets import QListWidgetItem, QWidget
|
||||
from qtpy.QtCore import QSize, Qt
|
||||
from qtpy.QtWidgets import QListWidget
|
||||
|
||||
from bec_widgets.utils.error_popups import SafeSlot
|
||||
from bec_widgets.utils.expandable_frame import ExpandableGroupFrame
|
||||
|
||||
logger = bec_logger.logger
|
||||
|
||||
_SORT_KEY_ROLE = 117
|
||||
|
||||
_EF = TypeVar("_EF", bound=ExpandableGroupFrame)
|
||||
|
||||
|
||||
class ListOfExpandableFrames(QListWidget, Generic[_EF]):
|
||||
def __init__(
|
||||
self, /, parent: QWidget | None = None, item_class: type[_EF] = ExpandableGroupFrame
|
||||
) -> None:
|
||||
super().__init__(parent)
|
||||
_Items = NamedTuple("_Items", (("item", QListWidgetItem), ("widget", _EF)))
|
||||
self.item_tuple = _Items
|
||||
self._item_class = item_class
|
||||
self._item_dict: dict[str, _Items] = {}
|
||||
|
||||
def __contains__(self, id: str):
|
||||
return id in self._item_dict
|
||||
|
||||
def clear(self) -> None:
|
||||
self._item_dict = {}
|
||||
return super().clear()
|
||||
|
||||
def add_item(self, id: str, *args, **kwargs) -> _EF:
|
||||
"""Adds the specified type of widget as an item. args and kwargs are passed to the constructor.
|
||||
|
||||
Args:
|
||||
id (str): the key under which to store the list item in the internal dict
|
||||
|
||||
Returns:
|
||||
The widget created in the addition process
|
||||
"""
|
||||
|
||||
def _remove_item(item: QListWidgetItem):
|
||||
self.takeItem(self.row(item))
|
||||
del self._item_dict[id]
|
||||
self.sortItems()
|
||||
|
||||
def _updatesize(item: QListWidgetItem, item_widget: _EF):
|
||||
item_widget.adjustSize()
|
||||
item.setSizeHint(QSize(item_widget.width(), item_widget.height()))
|
||||
|
||||
item = QListWidgetItem(self)
|
||||
item.setData(_SORT_KEY_ROLE, id) # used for sorting
|
||||
|
||||
item_widget = self._item_class(*args, **kwargs)
|
||||
item_widget.expansion_state_changed.connect(partial(_updatesize, item, item_widget))
|
||||
item_widget.imminent_deletion.connect(partial(_remove_item, item))
|
||||
item_widget.broadcast_size_hint.connect(item.setSizeHint)
|
||||
|
||||
self.addItem(item)
|
||||
self.setItemWidget(item, item_widget)
|
||||
self._item_dict[id] = self.item_tuple(item, item_widget)
|
||||
|
||||
item.setSizeHint(item_widget.sizeHint())
|
||||
return item_widget
|
||||
|
||||
def sort_by_key(self, role=_SORT_KEY_ROLE, order=Qt.SortOrder.AscendingOrder):
|
||||
items = [self.takeItem(0) for i in range(self.count())]
|
||||
items.sort(key=lambda it: it.data(role), reverse=(order == Qt.SortOrder.DescendingOrder))
|
||||
|
||||
for it in items:
|
||||
self.addItem(it)
|
||||
# reattach its custom widget
|
||||
widget = self.itemWidget(it)
|
||||
if widget:
|
||||
self.setItemWidget(it, widget)
|
||||
|
||||
def item_widget_pairs(self):
|
||||
return self._item_dict.values()
|
||||
|
||||
def widgets(self):
|
||||
return (i.widget for i in self._item_dict.values())
|
||||
|
||||
def get_item_widget(self, id: str):
|
||||
if (item := self._item_dict.get(id)) is None:
|
||||
return None
|
||||
return item
|
||||
|
||||
def set_hidden_pattern(self, pattern: Pattern):
|
||||
self.hide_all()
|
||||
self._set_hidden(filter(pattern.search, self._item_dict.keys()), False)
|
||||
|
||||
def set_hidden(self, ids: Iterable[str]):
|
||||
self._set_hidden(ids, True)
|
||||
|
||||
def _set_hidden(self, ids: Iterable[str], hidden: bool):
|
||||
for id in ids:
|
||||
if (_item := self._item_dict.get(id)) is not None:
|
||||
_item.item.setHidden(hidden)
|
||||
_item.widget.setHidden(hidden)
|
||||
else:
|
||||
logger.warning(
|
||||
f"List {self.__qualname__} does not have an item with ID {id} to hide!"
|
||||
)
|
||||
self.sortItems()
|
||||
|
||||
def hide_all(self):
|
||||
self.set_hidden_state_on_all(True)
|
||||
|
||||
def unhide_all(self):
|
||||
self.set_hidden_state_on_all(False)
|
||||
|
||||
def set_hidden_state_on_all(self, hidden: bool):
|
||||
for _item in self._item_dict.values():
|
||||
_item.item.setHidden(hidden)
|
||||
_item.widget.setHidden(hidden)
|
||||
self.sortItems()
|
||||
|
||||
@SafeSlot(str)
|
||||
def update_filter(self, value: str):
|
||||
if value == "":
|
||||
return self.unhide_all()
|
||||
try:
|
||||
self.set_hidden_pattern(re.compile(value, re.IGNORECASE))
|
||||
except Exception:
|
||||
self.unhide_all()
|
||||
File diff suppressed because one or more lines are too long
@@ -1,4 +0,0 @@
|
||||
from .device_table_view import DeviceTableView
|
||||
from .dm_config_view import DMConfigView
|
||||
from .dm_docstring_view import DocstringView
|
||||
from .dm_ophyd_test import DMOphydTest
|
||||
|
||||
@@ -0,0 +1,12 @@
|
||||
from typing import Any, Callable, Generator, Iterable, TypeVar
|
||||
|
||||
_T = TypeVar("_T")
|
||||
_RT = TypeVar("_RT")
|
||||
|
||||
|
||||
def yield_only_passing(fn: Callable[[_T], _RT], vals: Iterable[_T]) -> Generator[_RT, Any, None]:
|
||||
for v in vals:
|
||||
try:
|
||||
yield fn(v)
|
||||
except BaseException:
|
||||
pass
|
||||
@@ -1,39 +1,34 @@
|
||||
from textwrap import dedent
|
||||
from typing import NamedTuple
|
||||
|
||||
from bec_qthemes import material_icon
|
||||
from qtpy.QtCore import QSize
|
||||
from qtpy.QtWidgets import QFrame, QHBoxLayout, QLabel, QListWidgetItem, QVBoxLayout, QWidget
|
||||
|
||||
from bec_widgets.utils.expandable_frame import ExpandableGroupFrame
|
||||
from bec_widgets.widgets.control.device_manager.components.available_device_resources.available_device_group_ui import (
|
||||
Ui_AvailableDeviceGroup,
|
||||
)
|
||||
from bec_widgets.widgets.control.device_manager.components.available_device_resources.device_resource_backend import (
|
||||
HashableDevice,
|
||||
)
|
||||
from bec_widgets.widgets.control.device_manager.components.available_device_resources.device_tag_group_item_ui import (
|
||||
Ui_DeviceTagGroup,
|
||||
)
|
||||
|
||||
DEVICE_HASH_ROLE = 101
|
||||
|
||||
|
||||
def _warning_string(spec: HashableDevice):
|
||||
|
||||
names_str = "\n ".join(spec.names)
|
||||
msg = (
|
||||
f"Device defined with multiple names! Please check:\n {names_str}\n"
|
||||
name_warning = (
|
||||
f"Device defined with multiple names! Please check:\n {'\n '.join(spec.names)}\n"
|
||||
if len(spec.names) > 1
|
||||
else ""
|
||||
)
|
||||
|
||||
source_str = "\n ".join(spec.source_files)
|
||||
source_warning = (
|
||||
f"Device found in multiple source files! Please check:\n {source_str}"
|
||||
if len(spec.source_files) > 1
|
||||
f"Device found in multiple source files! Please check:\n {'\n '.join(spec._source_files)}"
|
||||
if len(spec._source_files) > 1
|
||||
else ""
|
||||
)
|
||||
return f"{msg}{source_warning}"
|
||||
return f"{name_warning}{source_warning}"
|
||||
|
||||
|
||||
class _DeviceEntryWidget(QFrame):
|
||||
_grid_size = QSize(120, 80)
|
||||
|
||||
def __init__(self, device_spec: HashableDevice, parent=None, **kwargs):
|
||||
super().__init__(parent, **kwargs)
|
||||
@@ -44,19 +39,26 @@ class _DeviceEntryWidget(QFrame):
|
||||
self.setFrameShadow(QFrame.Shadow.Raised)
|
||||
|
||||
self._layout = QVBoxLayout()
|
||||
self._layout.setContentsMargins(5, 5, 5, 5)
|
||||
self._layout.setContentsMargins(0, 0, 0, 0)
|
||||
self.setLayout(self._layout)
|
||||
self.setMinimumSize(self._grid_size)
|
||||
|
||||
self.setup_title_layout(device_spec)
|
||||
self.check_and_display_warning()
|
||||
|
||||
self.setToolTip(device_spec.rich_text())
|
||||
self.setToolTip(self._rich_text())
|
||||
|
||||
self.details = QLabel(f"Tags:\n{', '.join(device_spec.deviceTags)}")
|
||||
self.details.setStyleSheet("QLabel { font-size: 8pt; }")
|
||||
self.details.setWordWrap(True)
|
||||
self._layout.addWidget(self.details)
|
||||
def _rich_text(self):
|
||||
return dedent(
|
||||
f"""
|
||||
<b><u><h2> {self._device_spec.name}: </h2></u></b>
|
||||
<table>
|
||||
<tr><td> description: </td><td><i> {self._device_spec.description} </i></td></tr>
|
||||
<tr><td> config: </td><td><i> {self._device_spec.deviceConfig} </i></td></tr>
|
||||
<tr><td> enabled: </td><td><i> {self._device_spec.enabled} </i></td></tr>
|
||||
<tr><td> read only: </td><td><i> {self._device_spec.readOnly} </i></td></tr>
|
||||
</table>
|
||||
"""
|
||||
)
|
||||
|
||||
def setup_title_layout(self, device_spec: HashableDevice):
|
||||
self._title_layout = QHBoxLayout()
|
||||
@@ -72,10 +74,11 @@ class _DeviceEntryWidget(QFrame):
|
||||
self.title.setStyleSheet(self.title_style("#FF0000"))
|
||||
self._title_layout.addWidget(self.title)
|
||||
|
||||
self._title_layout.addStretch(1)
|
||||
self._layout.addWidget(self._title_container)
|
||||
|
||||
def check_and_display_warning(self):
|
||||
if len(self._device_spec.names) == 1 and len(self._device_spec.source_files) == 1:
|
||||
if len(self._device_spec.names) == 1 and len(self._device_spec._source_files) == 1:
|
||||
self._warning_label.setText("")
|
||||
self._warning_label.setToolTip("")
|
||||
else:
|
||||
@@ -102,18 +105,18 @@ class _DeviceEntry(NamedTuple):
|
||||
widget: _DeviceEntryWidget
|
||||
|
||||
|
||||
class DeviceTagGroup(QWidget, Ui_DeviceTagGroup):
|
||||
class AvailableDeviceGroup(ExpandableGroupFrame, Ui_AvailableDeviceGroup):
|
||||
def __init__(
|
||||
self, parent=None, name: str = "TagGroupTitle", data: set[HashableDevice] = set(), **kwargs
|
||||
):
|
||||
super().__init__(parent=parent, **kwargs)
|
||||
self.setupUi(self)
|
||||
self.device_list.setGridSize(_DeviceEntryWidget._grid_size)
|
||||
self.title.setText(name)
|
||||
self.title_text = name # type: ignore
|
||||
self._devices: dict[str, _DeviceEntry] = {}
|
||||
for device in data:
|
||||
self._add_item(device)
|
||||
self.device_list.sortItems()
|
||||
self.setMinimumSize(self.device_list.sizeHint())
|
||||
self._update_num_included()
|
||||
|
||||
self.add_to_composition_button.clicked.connect(self.test)
|
||||
@@ -148,6 +151,14 @@ class DeviceTagGroup(QWidget, Ui_DeviceTagGroup):
|
||||
self.n_included.setText(f"{n_included} / {len(self._devices)}")
|
||||
self.n_included.setStyleSheet(f"QLabel {{ color: {color}; }}")
|
||||
|
||||
def sizeHint(self) -> QSize:
|
||||
if not getattr(self, "device_list", None) or not self.expanded:
|
||||
return super().sizeHint()
|
||||
return QSize(
|
||||
max(150, self.device_list.viewport().width()),
|
||||
self.device_list.sizeHintForRow(0) * self.device_list.count() + 50,
|
||||
)
|
||||
|
||||
def resizeEvent(self, event):
|
||||
super().resizeEvent(event)
|
||||
self.setMinimumHeight(self.sizeHint().height())
|
||||
@@ -162,7 +173,7 @@ class DeviceTagGroup(QWidget, Ui_DeviceTagGroup):
|
||||
print(self.get_selection())
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"{self.__class__.__name__}: {self.title.text()}"
|
||||
return f"{self.__class__.__name__}: {self.title_text}"
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
@@ -171,7 +182,7 @@ if __name__ == "__main__":
|
||||
from qtpy.QtWidgets import QApplication
|
||||
|
||||
app = QApplication(sys.argv)
|
||||
widget = DeviceTagGroup(name="Tag group 1")
|
||||
widget = AvailableDeviceGroup(name="Tag group 1")
|
||||
for item in [
|
||||
HashableDevice(
|
||||
**{
|
||||
@@ -0,0 +1,66 @@
|
||||
from functools import partial
|
||||
|
||||
from bec_qthemes import material_icon
|
||||
from PySide6.QtWidgets import QFrame
|
||||
from qtpy.QtCore import QMetaObject
|
||||
from qtpy.QtWidgets import QLabel, QListWidget, QToolButton, QVBoxLayout
|
||||
|
||||
|
||||
class Ui_AvailableDeviceGroup(object):
|
||||
def setupUi(self, AvailableDeviceGroup):
|
||||
if not AvailableDeviceGroup.objectName():
|
||||
AvailableDeviceGroup.setObjectName("AvailableDeviceGroup")
|
||||
AvailableDeviceGroup.setMinimumWidth(150)
|
||||
self.verticalLayout = QVBoxLayout()
|
||||
self.verticalLayout.setObjectName("verticalLayout")
|
||||
AvailableDeviceGroup.set_layout(self.verticalLayout)
|
||||
|
||||
title_layout = AvailableDeviceGroup.get_title_layout()
|
||||
|
||||
self.n_included = QLabel(AvailableDeviceGroup, text="...")
|
||||
self.n_included.setObjectName("n_included")
|
||||
title_layout.addWidget(self.n_included)
|
||||
|
||||
self.delete_tag_button = QToolButton(AvailableDeviceGroup)
|
||||
self.delete_tag_button.setObjectName("delete_tag_button")
|
||||
title_layout.addWidget(self.delete_tag_button)
|
||||
|
||||
self.remove_from_composition_button = QToolButton(AvailableDeviceGroup)
|
||||
self.remove_from_composition_button.setObjectName("remove_from_composition_button")
|
||||
title_layout.addWidget(self.remove_from_composition_button)
|
||||
|
||||
self.add_to_composition_button = QToolButton(AvailableDeviceGroup)
|
||||
self.add_to_composition_button.setObjectName("add_to_composition_button")
|
||||
title_layout.addWidget(self.add_to_composition_button)
|
||||
|
||||
self.remove_all_button = QToolButton(AvailableDeviceGroup)
|
||||
self.remove_all_button.setObjectName("remove_all_from_composition_button")
|
||||
title_layout.addWidget(self.remove_all_button)
|
||||
|
||||
self.add_all_button = QToolButton(AvailableDeviceGroup)
|
||||
self.add_all_button.setObjectName("add_all_to_composition_button")
|
||||
title_layout.addWidget(self.add_all_button)
|
||||
|
||||
self.device_list = QListWidget(AvailableDeviceGroup)
|
||||
self.device_list.setSelectionMode(QListWidget.SelectionMode.ExtendedSelection)
|
||||
self.device_list.setObjectName("device_list")
|
||||
self.device_list.setFrameStyle(0)
|
||||
|
||||
self.verticalLayout.addWidget(self.device_list)
|
||||
|
||||
self.set_icons()
|
||||
|
||||
QMetaObject.connectSlotsByName(AvailableDeviceGroup)
|
||||
|
||||
def set_icons(self):
|
||||
icon = partial(material_icon, size=(15, 15), convert_to_pixmap=False)
|
||||
self.delete_tag_button.setIcon(icon("delete"))
|
||||
self.delete_tag_button.setToolTip("Delete tag group")
|
||||
self.remove_from_composition_button.setIcon(icon("remove"))
|
||||
self.remove_from_composition_button.setToolTip("Remove selected from composition")
|
||||
self.add_to_composition_button.setIcon(icon("add"))
|
||||
self.add_to_composition_button.setToolTip("Add selected to composition")
|
||||
self.remove_all_button.setIcon(icon("chips"))
|
||||
self.remove_all_button.setToolTip("Remove all with this tag from composition")
|
||||
self.add_all_button.setIcon(icon("add_box"))
|
||||
self.add_all_button.setToolTip("Add all with this tag to composition")
|
||||
@@ -1,11 +1,13 @@
|
||||
from random import randint
|
||||
from typing import Any, Callable, Generator, Iterable, TypeVar
|
||||
from typing import Any, Iterable
|
||||
|
||||
from qtpy.QtCore import QSize
|
||||
from qtpy.QtWidgets import QListWidgetItem, QWidget
|
||||
from qtpy.QtWidgets import QWidget
|
||||
|
||||
from bec_widgets.utils.bec_widget import BECWidget
|
||||
from bec_widgets.utils.error_popups import SafeSlot
|
||||
from bec_widgets.widgets.control.device_manager.components.available_device_resources._util import (
|
||||
yield_only_passing,
|
||||
)
|
||||
from bec_widgets.widgets.control.device_manager.components.available_device_resources.available_device_resources_ui import (
|
||||
Ui_availableDeviceResources,
|
||||
)
|
||||
@@ -13,20 +15,6 @@ from bec_widgets.widgets.control.device_manager.components.available_device_reso
|
||||
HashableDevice,
|
||||
get_backend,
|
||||
)
|
||||
from bec_widgets.widgets.control.device_manager.components.available_device_resources.device_tag_group import (
|
||||
DeviceTagGroup,
|
||||
)
|
||||
|
||||
_T = TypeVar("_T")
|
||||
_RT = TypeVar("_RT")
|
||||
|
||||
|
||||
def _yield_only_passing(fn: Callable[[_T], _RT], vals: Iterable[_T]) -> Generator[_RT, Any, None]:
|
||||
for v in vals:
|
||||
try:
|
||||
yield fn(v)
|
||||
except BaseException:
|
||||
pass
|
||||
|
||||
|
||||
class AvailableDeviceResources(BECWidget, QWidget, Ui_availableDeviceResources):
|
||||
@@ -34,43 +22,51 @@ class AvailableDeviceResources(BECWidget, QWidget, Ui_availableDeviceResources):
|
||||
super().__init__(parent=parent, **kwargs)
|
||||
self.setupUi(self)
|
||||
self._backend = get_backend()
|
||||
self._items: dict[str, tuple[QListWidgetItem, DeviceTagGroup]] = {}
|
||||
self.refresh_full_list()
|
||||
self.grouping_selector.addItem("deviceTags")
|
||||
self.grouping_selector.addItems(self._backend.allowed_sort_keys)
|
||||
self._grouping_selection_changed("deviceTags")
|
||||
self.grouping_selector.currentTextChanged.connect(self._grouping_selection_changed)
|
||||
self.search_box.textChanged.connect(self.device_groups_list.update_filter)
|
||||
|
||||
def refresh_full_list(self):
|
||||
self.tag_groups_list.clear()
|
||||
self._items = {}
|
||||
for tag_group, devices in self._backend.tag_groups.items():
|
||||
self._add_tag_group(tag_group, devices)
|
||||
self._add_tag_group("Untagged devices", self._backend.untagged_devices)
|
||||
def refresh_full_list(self, device_groups: dict[str, set[HashableDevice]]):
|
||||
self.device_groups_list.clear()
|
||||
for device_group, devices in device_groups.items():
|
||||
self._add_device_group(device_group, devices)
|
||||
if self.grouping_selector.currentText == "deviceTags":
|
||||
self._add_device_group("Untagged devices", self._backend.untagged_devices)
|
||||
self.device_groups_list.sort_by_key()
|
||||
|
||||
def _add_tag_group(self, tag_group: str, devices: set[HashableDevice]):
|
||||
item = QListWidgetItem(self.tag_groups_list)
|
||||
tag_group_widget = DeviceTagGroup(self.tag_groups_list, tag_group, devices)
|
||||
self.tag_groups_list.setItemWidget(item, tag_group_widget)
|
||||
self.tag_groups_list.addItem(item)
|
||||
self._items[tag_group] = (item, tag_group_widget)
|
||||
item.setSizeHint(QSize(tag_group_widget.width(), tag_group_widget.height()))
|
||||
def _add_device_group(self, device_group: str, devices: set[HashableDevice]):
|
||||
self.device_groups_list.add_item(
|
||||
device_group, self.device_groups_list, device_group, devices, expanded=False
|
||||
)
|
||||
|
||||
def _reset_devices_state(self):
|
||||
for _, tag_group in self._items.values():
|
||||
tag_group.reset_devices_state()
|
||||
for device_group in self.device_groups_list.widgets():
|
||||
device_group.reset_devices_state()
|
||||
|
||||
def set_devices_state(self, devices: Iterable[HashableDevice], included: bool):
|
||||
for device in devices:
|
||||
for _, tag_group in self._items.values():
|
||||
tag_group.set_item_state(hash(device), included)
|
||||
for device_group in self.device_groups_list.widgets():
|
||||
device_group.set_item_state(hash(device), included)
|
||||
|
||||
def resizeEvent(self, event):
|
||||
super().resizeEvent(event)
|
||||
for list_item, tag_group_widget in self._items.values():
|
||||
list_item.setSizeHint(tag_group_widget.sizeHint())
|
||||
for list_item, device_group_widget in self.device_groups_list.item_widget_pairs():
|
||||
list_item.setSizeHint(device_group_widget.sizeHint())
|
||||
|
||||
@SafeSlot(list)
|
||||
def update_devices_state(self, config_list: list[dict[str, Any]]):
|
||||
self.set_devices_state(
|
||||
_yield_only_passing(HashableDevice.model_validate, config_list), True
|
||||
)
|
||||
self.set_devices_state(yield_only_passing(HashableDevice.model_validate, config_list), True)
|
||||
|
||||
@SafeSlot(str)
|
||||
def _grouping_selection_changed(self, sort_key: str):
|
||||
self.search_box.setText("")
|
||||
if sort_key == "deviceTags":
|
||||
device_groups = self._backend.tag_groups
|
||||
else:
|
||||
device_groups = self._backend.group_by_key(sort_key)
|
||||
self.refresh_full_list(device_groups)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -1,5 +1,19 @@
|
||||
from qtpy.QtCore import QMetaObject, Qt
|
||||
from qtpy.QtWidgets import QAbstractItemView, QListView, QListWidget, QVBoxLayout
|
||||
from qtpy.QtWidgets import (
|
||||
QAbstractItemView,
|
||||
QComboBox,
|
||||
QHBoxLayout,
|
||||
QLabel,
|
||||
QLineEdit,
|
||||
QListView,
|
||||
QListWidget,
|
||||
QVBoxLayout,
|
||||
)
|
||||
|
||||
from bec_widgets.utils.list_of_expandable_frames import ListOfExpandableFrames
|
||||
from bec_widgets.widgets.control.device_manager.components.available_device_resources.available_device_group import (
|
||||
AvailableDeviceGroup,
|
||||
)
|
||||
|
||||
|
||||
class Ui_availableDeviceResources(object):
|
||||
@@ -8,20 +22,34 @@ class Ui_availableDeviceResources(object):
|
||||
availableDeviceResources.setObjectName("availableDeviceResources")
|
||||
self.verticalLayout = QVBoxLayout(availableDeviceResources)
|
||||
self.verticalLayout.setObjectName("verticalLayout")
|
||||
self.tag_groups_list = QListWidget(availableDeviceResources)
|
||||
self.tag_groups_list.setObjectName("tag_groups_list")
|
||||
self.tag_groups_list.setSelectionMode(QAbstractItemView.SelectionMode.NoSelection)
|
||||
self.tag_groups_list.setVerticalScrollMode(QAbstractItemView.ScrollMode.ScrollPerPixel)
|
||||
self.tag_groups_list.setHorizontalScrollMode(QAbstractItemView.ScrollMode.ScrollPerPixel)
|
||||
self.tag_groups_list.setMovement(QListView.Movement.Static)
|
||||
self.tag_groups_list.setSpacing(2)
|
||||
self.tag_groups_list.setDragDropMode(QListWidget.DragDropMode.DragOnly)
|
||||
self.tag_groups_list.setDragEnabled(True)
|
||||
self.tag_groups_list.setAcceptDrops(False)
|
||||
self.tag_groups_list.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
|
||||
|
||||
self.search_layout = QHBoxLayout()
|
||||
self.verticalLayout.addLayout(self.search_layout)
|
||||
self.search_layout.addWidget(QLabel("Filter groups: "))
|
||||
self.search_box = QLineEdit()
|
||||
self.search_layout.addWidget(self.search_box)
|
||||
self.search_layout.addWidget(QLabel("Group by: "))
|
||||
self.grouping_selector = QComboBox()
|
||||
self.search_layout.addWidget(self.grouping_selector)
|
||||
|
||||
self.device_groups_list = ListOfExpandableFrames(
|
||||
availableDeviceResources, AvailableDeviceGroup
|
||||
)
|
||||
self.device_groups_list.setObjectName("device_groups_list")
|
||||
self.device_groups_list.setSelectionMode(QAbstractItemView.SelectionMode.NoSelection)
|
||||
self.device_groups_list.setVerticalScrollMode(QAbstractItemView.ScrollMode.ScrollPerPixel)
|
||||
self.device_groups_list.setHorizontalScrollMode(QAbstractItemView.ScrollMode.ScrollPerPixel)
|
||||
self.device_groups_list.setMovement(QListView.Movement.Static)
|
||||
self.device_groups_list.setSpacing(2)
|
||||
self.device_groups_list.setDragDropMode(QListWidget.DragDropMode.DragOnly)
|
||||
self.device_groups_list.setSelectionBehavior(QListWidget.SelectionBehavior.SelectItems)
|
||||
self.device_groups_list.setSelectionMode(QListWidget.SelectionMode.ExtendedSelection)
|
||||
self.device_groups_list.setDragEnabled(True)
|
||||
self.device_groups_list.setAcceptDrops(False)
|
||||
self.device_groups_list.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
|
||||
availableDeviceResources.setMinimumWidth(250)
|
||||
availableDeviceResources.resize(250, availableDeviceResources.height())
|
||||
|
||||
self.verticalLayout.addWidget(self.tag_groups_list)
|
||||
self.verticalLayout.addWidget(self.device_groups_list)
|
||||
|
||||
QMetaObject.connectSlotsByName(availableDeviceResources)
|
||||
|
||||
@@ -1,79 +1,28 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import operator
|
||||
from functools import reduce
|
||||
import os
|
||||
from enum import Enum, auto
|
||||
from functools import partial, reduce
|
||||
from glob import glob
|
||||
from pathlib import Path
|
||||
from textwrap import dedent
|
||||
from typing import AbstractSet, Protocol
|
||||
from typing import Protocol
|
||||
|
||||
from bec_lib.atlas_models import Device
|
||||
import bec_lib
|
||||
from bec_lib.atlas_models import HashableDevice, HashableDeviceSet
|
||||
from bec_lib.bec_yaml_loader import yaml_load
|
||||
from bec_lib.logger import bec_logger
|
||||
from bec_lib.plugin_helper import plugin_package_name, plugin_repo_path
|
||||
from pydantic import model_validator
|
||||
|
||||
logger = bec_logger.logger
|
||||
|
||||
_BASE_REPO_PATH = Path(os.path.dirname(bec_lib.__file__)) / "../.."
|
||||
|
||||
|
||||
class HashableDevice(Device):
|
||||
source_files: set[str] = set()
|
||||
names: set[str] = set()
|
||||
|
||||
@model_validator(mode="after")
|
||||
def add_name(self) -> HashableDevice:
|
||||
self.names.add(self.name)
|
||||
return self
|
||||
|
||||
def as_normal_device(self):
|
||||
return Device.model_validate(self)
|
||||
|
||||
def __hash__(self) -> int:
|
||||
config_values = sorted(
|
||||
(str(kv) for kv in self.deviceConfig.items()) if self.deviceConfig else []
|
||||
)
|
||||
return (reduce(operator.add, (self.name, self.deviceClass, *config_values))).__hash__()
|
||||
|
||||
def __eq__(self, value: object) -> bool:
|
||||
if not isinstance(value, self.__class__):
|
||||
return False
|
||||
if hash(self) == hash(value):
|
||||
return True
|
||||
return False
|
||||
|
||||
def rich_text(self) -> str:
|
||||
return dedent(
|
||||
f"""
|
||||
<b><u><h2> {self.name}: </h2></u></b>
|
||||
<table>
|
||||
<tr><td> description: </td><td><i> {self.description} </i></td></tr>
|
||||
<tr><td> config: </td><td><i> {self.deviceConfig} </i></td></tr>
|
||||
<tr><td> enabled: </td><td><i> {self.enabled} </i></td></tr>
|
||||
<tr><td> read only: </td><td><i> {self.readOnly} </i></td></tr>
|
||||
</table>
|
||||
"""
|
||||
)
|
||||
|
||||
def add_sources(self, other: HashableDevice):
|
||||
self.source_files.update(other.source_files)
|
||||
|
||||
def add_tags(self, other: HashableDevice):
|
||||
self.deviceTags.update(other.deviceTags)
|
||||
|
||||
def add_names(self, other: HashableDevice):
|
||||
self.names.update(other.names)
|
||||
|
||||
|
||||
class _HashableDeviceSet(set):
|
||||
def __or__(self, value: AbstractSet) -> _HashableDeviceSet:
|
||||
for item in self:
|
||||
if item in value:
|
||||
for other_item in value:
|
||||
if other_item == item:
|
||||
item.add_sources(other_item)
|
||||
item.add_tags(other_item)
|
||||
item.add_names(other_item)
|
||||
for other_item in value:
|
||||
if other_item not in self:
|
||||
self.add(other_item)
|
||||
return self
|
||||
class HashModel(str, Enum):
|
||||
DEFAULT = auto()
|
||||
DEFAULT_DEVICECONFIG = auto()
|
||||
DEFAULT_EPICS = auto()
|
||||
|
||||
|
||||
class DeviceResourceBackend(Protocol):
|
||||
@@ -93,6 +42,11 @@ class DeviceResourceBackend(Protocol):
|
||||
"""A set of all untagged devices. The same device may not appear more than once."""
|
||||
...
|
||||
|
||||
@property
|
||||
def allowed_sort_keys(self) -> set[str]:
|
||||
"""A set of all fields which you may group devices by"""
|
||||
...
|
||||
|
||||
def tags(self) -> set[str]:
|
||||
"""Returns a set of all the tags in all available devices."""
|
||||
...
|
||||
@@ -101,10 +55,15 @@ class DeviceResourceBackend(Protocol):
|
||||
"""Returns a set of the devices in the tag group with the given key."""
|
||||
...
|
||||
|
||||
def group_by_key(self, key: str) -> dict[str, set[HashableDevice]]:
|
||||
"""Return a dict of all devices, organised by the specified key, which must be one of
|
||||
the string keys in the Device model."""
|
||||
...
|
||||
|
||||
|
||||
def _devices_from_file(file: str, include_source: bool = True):
|
||||
data = yaml_load(file, process_includes=False)
|
||||
return _HashableDeviceSet(
|
||||
return HashableDeviceSet(
|
||||
HashableDevice.model_validate(
|
||||
dev | {"name": name, "source_files": {file} if include_source else set()}
|
||||
)
|
||||
@@ -116,16 +75,18 @@ class _ConfigFileBackend(DeviceResourceBackend):
|
||||
def __init__(self) -> None:
|
||||
self._raw_device_set: set[
|
||||
HashableDevice
|
||||
] = self._get_config_from_backup_file() or self._get_configs_from_plugin_files(
|
||||
] = self._get_config_from_backup_files() | self._get_configs_from_plugin_files(
|
||||
Path(plugin_repo_path()) / plugin_package_name() / "device_configs/"
|
||||
)
|
||||
self._tag_groups = self._get_tag_groups()
|
||||
self._device_groups = self._get_tag_groups()
|
||||
|
||||
def _get_config_from_backup_file(self):
|
||||
return None
|
||||
# return _devices_from_file(
|
||||
# "/home/perl_d/Development/bec/bec/logs/device_configs/recovery_configs/recovery_config_2025-08-22_14-02-29.yaml"
|
||||
# )
|
||||
def _get_config_from_backup_files(self):
|
||||
dir = _BASE_REPO_PATH / "logs/device_configs/recovery_configs"
|
||||
files = glob("*.yaml", root_dir=dir)
|
||||
return reduce(
|
||||
operator.or_,
|
||||
map(partial(_devices_from_file, include_source=False), (str(dir / f) for f in files)),
|
||||
)
|
||||
|
||||
def _get_configs_from_plugin_files(self, dir: Path):
|
||||
files = glob("*.yaml", root_dir=dir, recursive=True)
|
||||
@@ -139,7 +100,7 @@ class _ConfigFileBackend(DeviceResourceBackend):
|
||||
|
||||
@property
|
||||
def tag_groups(self):
|
||||
return self._tag_groups
|
||||
return self._device_groups
|
||||
|
||||
@property
|
||||
def all_devices(self):
|
||||
@@ -149,12 +110,22 @@ class _ConfigFileBackend(DeviceResourceBackend):
|
||||
def untagged_devices(self):
|
||||
return {d for d in self._raw_device_set if d.deviceTags == set()}
|
||||
|
||||
@property
|
||||
def allowed_sort_keys(self) -> set[str]:
|
||||
return {n for n, info in HashableDevice.model_fields.items() if info.annotation is str}
|
||||
|
||||
def tags(self) -> set[str]:
|
||||
return reduce(operator.or_, (dev.deviceTags for dev in self._raw_device_set))
|
||||
|
||||
def tag_group(self, tag: str) -> set[HashableDevice]:
|
||||
return self.tag_groups[tag]
|
||||
|
||||
def group_by_key(self, key: str) -> dict[str, set[HashableDevice]]:
|
||||
if key not in self.allowed_sort_keys:
|
||||
raise ValueError(f"Cannot group available devices by model key {key}")
|
||||
group_names: set[str] = {getattr(item, key) for item in self._raw_device_set}
|
||||
return {g: {d for d in self._raw_device_set if getattr(d, key) == g} for g in group_names}
|
||||
|
||||
|
||||
def get_backend() -> DeviceResourceBackend:
|
||||
return _ConfigFileBackend()
|
||||
|
||||
@@ -1,135 +0,0 @@
|
||||
import math
|
||||
from functools import partial
|
||||
|
||||
from bec_qthemes import material_icon
|
||||
from qtpy.QtCore import QMetaObject, QSize, Qt
|
||||
from qtpy.QtWidgets import (
|
||||
QAbstractItemView,
|
||||
QFrame,
|
||||
QHBoxLayout,
|
||||
QLabel,
|
||||
QListView,
|
||||
QListWidget,
|
||||
QSizePolicy,
|
||||
QSpacerItem,
|
||||
QToolButton,
|
||||
QVBoxLayout,
|
||||
)
|
||||
|
||||
|
||||
class AutoHeightListWidget(QListWidget):
|
||||
def __init__(self, parent=None):
|
||||
super().__init__(parent)
|
||||
self.setViewMode(QListView.ViewMode.IconMode)
|
||||
self.setResizeMode(QListView.ResizeMode.Adjust)
|
||||
self.setWrapping(True)
|
||||
self.setUniformItemSizes(True)
|
||||
self.setMovement(QListView.Movement.Static)
|
||||
self.setAcceptDrops(False)
|
||||
self.setDragEnabled(True)
|
||||
self.setSelectionMode(QAbstractItemView.SelectionMode.ExtendedSelection)
|
||||
self.setSpacing(5)
|
||||
|
||||
self.setVerticalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
|
||||
self.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
|
||||
|
||||
def resizeEvent(self, event):
|
||||
super().resizeEvent(event)
|
||||
self.setMinimumHeight(self._calcSize().height())
|
||||
self.setMaximumHeight(self._calcSize().height())
|
||||
|
||||
def sizeHint(self):
|
||||
return self._calcSize()
|
||||
|
||||
def minimumSizeHint(self):
|
||||
return self._calcSize()
|
||||
|
||||
def _calcSize(self):
|
||||
if self.count() == 0:
|
||||
return super().sizeHint()
|
||||
|
||||
grid = self.gridSize()
|
||||
if not grid.isValid():
|
||||
grid = QSize(100, 100) # fallback
|
||||
|
||||
items_per_row = max(1, self.viewport().width() // grid.width())
|
||||
rows = math.ceil(self.count() / items_per_row)
|
||||
|
||||
height = rows * grid.height() + 2 * self.frameWidth()
|
||||
return QSize(self.viewport().width(), height)
|
||||
|
||||
|
||||
class Ui_DeviceTagGroup(object):
|
||||
def setupUi(self, DeviceTagGroup):
|
||||
if not DeviceTagGroup.objectName():
|
||||
DeviceTagGroup.setObjectName("DeviceTagGroup")
|
||||
DeviceTagGroup.setMinimumWidth(150)
|
||||
self.verticalLayout = QVBoxLayout(DeviceTagGroup)
|
||||
self.verticalLayout.setObjectName("verticalLayout")
|
||||
self.frame = QFrame(DeviceTagGroup)
|
||||
self.frame.setObjectName("frame")
|
||||
self.frame.setFrameShape(QFrame.Shape.StyledPanel)
|
||||
self.frame.setFrameShadow(QFrame.Shadow.Raised)
|
||||
self.verticalLayout_2 = QVBoxLayout(self.frame)
|
||||
self.verticalLayout_2.setObjectName("verticalLayout_2")
|
||||
self.horizontalLayout = QHBoxLayout()
|
||||
self.horizontalLayout.setObjectName("horizontalLayout")
|
||||
|
||||
self.title = QLabel(self.frame)
|
||||
self.title.setObjectName("title")
|
||||
self.horizontalLayout.addWidget(self.title)
|
||||
|
||||
self.n_included = QLabel(self.frame, text="...")
|
||||
self.n_included.setObjectName("n_included")
|
||||
self.horizontalLayout.addWidget(self.n_included)
|
||||
|
||||
self.horizontalSpacer = QSpacerItem(
|
||||
40, 20, QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Minimum
|
||||
)
|
||||
self.horizontalLayout.addItem(self.horizontalSpacer)
|
||||
|
||||
self.delete_tag_button = QToolButton(self.frame)
|
||||
self.delete_tag_button.setObjectName("delete_tag_button")
|
||||
self.horizontalLayout.addWidget(self.delete_tag_button)
|
||||
|
||||
self.remove_from_composition_button = QToolButton(self.frame)
|
||||
self.remove_from_composition_button.setObjectName("remove_from_composition_button")
|
||||
self.horizontalLayout.addWidget(self.remove_from_composition_button)
|
||||
|
||||
self.add_to_composition_button = QToolButton(self.frame)
|
||||
self.add_to_composition_button.setObjectName("add_to_composition_button")
|
||||
self.horizontalLayout.addWidget(self.add_to_composition_button)
|
||||
|
||||
self.remove_all_button = QToolButton(self.frame)
|
||||
self.remove_all_button.setObjectName("remove_all_from_composition_button")
|
||||
self.horizontalLayout.addWidget(self.remove_all_button)
|
||||
|
||||
self.add_all_button = QToolButton(self.frame)
|
||||
self.add_all_button.setObjectName("add_all_to_composition_button")
|
||||
self.horizontalLayout.addWidget(self.add_all_button)
|
||||
|
||||
self.verticalLayout_2.addLayout(self.horizontalLayout)
|
||||
|
||||
self.device_list = AutoHeightListWidget(self.frame)
|
||||
self.device_list.setObjectName("device_list")
|
||||
|
||||
self.verticalLayout_2.addWidget(self.device_list)
|
||||
|
||||
self.verticalLayout.addWidget(self.frame)
|
||||
|
||||
self.set_icons()
|
||||
|
||||
QMetaObject.connectSlotsByName(DeviceTagGroup)
|
||||
|
||||
def set_icons(self):
|
||||
icon = partial(material_icon, size=(15, 15), convert_to_pixmap=False)
|
||||
self.delete_tag_button.setIcon(icon("delete"))
|
||||
self.delete_tag_button.setToolTip("Delete tag group")
|
||||
self.remove_from_composition_button.setIcon(icon("remove"))
|
||||
self.remove_from_composition_button.setToolTip("Remove selected from composition")
|
||||
self.add_to_composition_button.setIcon(icon("add"))
|
||||
self.add_to_composition_button.setToolTip("Add selected to composition")
|
||||
self.remove_all_button.setIcon(icon("chips"))
|
||||
self.remove_all_button.setToolTip("Remove all with this tag from composition")
|
||||
self.add_all_button.setIcon(icon("add_box"))
|
||||
self.add_all_button.setToolTip("Add all with this tag to composition")
|
||||
@@ -3,18 +3,16 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import time
|
||||
import json
|
||||
|
||||
from bec_lib.logger import bec_logger
|
||||
from bec_qthemes import material_icon
|
||||
from qtpy import QtCore, QtGui, QtWidgets
|
||||
from thefuzz import fuzz
|
||||
|
||||
from bec_widgets.utils.bec_signal_proxy import BECSignalProxy
|
||||
from bec_widgets.utils.bec_widget import BECWidget
|
||||
from bec_widgets.utils.colors import get_accent_colors, get_theme_palette
|
||||
from bec_widgets.utils.colors import get_accent_colors
|
||||
from bec_widgets.utils.error_popups import SafeSlot
|
||||
from bec_widgets.widgets.control.device_manager.components.dm_ophyd_test import ValidationStatus
|
||||
|
||||
logger = bec_logger.logger
|
||||
|
||||
@@ -32,25 +30,21 @@ class DictToolTipDelegate(QtWidgets.QStyledItemDelegate):
|
||||
model: DeviceFilterProxyModel = index.model()
|
||||
model_index = model.mapToSource(index)
|
||||
row_dict = model.sourceModel().get_row_data(model_index)
|
||||
description = row_dict.get("description", "")
|
||||
QtWidgets.QToolTip.showText(event.globalPos(), description, view)
|
||||
QtWidgets.QToolTip.showText(event.globalPos(), row_dict["description"], view)
|
||||
return True
|
||||
|
||||
|
||||
class CenterCheckBoxDelegate(DictToolTipDelegate):
|
||||
"""Custom checkbox delegate to center checkboxes in table cells."""
|
||||
|
||||
def __init__(self, parent=None, colors=None):
|
||||
def __init__(self, parent=None):
|
||||
super().__init__(parent)
|
||||
self._colors = colors if colors else get_accent_colors()
|
||||
colors = get_accent_colors()
|
||||
self._icon_checked = material_icon(
|
||||
"check_box", size=QtCore.QSize(16, 16), color=self._colors.default, filled=True
|
||||
"check_box", size=QtCore.QSize(16, 16), color=colors.default, filled=True
|
||||
)
|
||||
self._icon_unchecked = material_icon(
|
||||
"check_box_outline_blank",
|
||||
size=QtCore.QSize(16, 16),
|
||||
color=self._colors.default,
|
||||
filled=True,
|
||||
"check_box_outline_blank", size=QtCore.QSize(16, 16), color=colors.default, filled=True
|
||||
)
|
||||
|
||||
def apply_theme(self, theme: str | None = None):
|
||||
@@ -81,51 +75,9 @@ class CenterCheckBoxDelegate(DictToolTipDelegate):
|
||||
return model.setData(index, new_state, QtCore.Qt.CheckStateRole)
|
||||
|
||||
|
||||
class DeviceValidatedDelegate(DictToolTipDelegate):
|
||||
"""Custom delegate for displaying validated device configurations."""
|
||||
|
||||
def __init__(self, parent=None, colors=None):
|
||||
super().__init__(parent)
|
||||
self._colors = colors if colors else get_accent_colors()
|
||||
self._icons = {
|
||||
ValidationStatus.PENDING: material_icon(
|
||||
icon_name="circle", size=(12, 12), color=self._colors.default, filled=True
|
||||
),
|
||||
ValidationStatus.VALID: material_icon(
|
||||
icon_name="circle", size=(12, 12), color=self._colors.success, filled=True
|
||||
),
|
||||
ValidationStatus.FAILED: material_icon(
|
||||
icon_name="circle", size=(12, 12), color=self._colors.emergency, filled=True
|
||||
),
|
||||
}
|
||||
|
||||
def apply_theme(self, theme: str | None = None):
|
||||
colors = get_accent_colors()
|
||||
for status, icon in self._icons.items():
|
||||
icon.setColor(colors[status])
|
||||
|
||||
def paint(self, painter, option, index):
|
||||
status = index.model().data(index, QtCore.Qt.DisplayRole)
|
||||
if status is None:
|
||||
return super().paint(painter, option, index)
|
||||
|
||||
pixmap = self._icons.get(status)
|
||||
if pixmap:
|
||||
rect = option.rect
|
||||
pix_rect = pixmap.rect()
|
||||
pix_rect.moveCenter(rect.center())
|
||||
painter.drawPixmap(pix_rect.topLeft(), pixmap)
|
||||
|
||||
super().paint(painter, option, index)
|
||||
|
||||
|
||||
class WrappingTextDelegate(DictToolTipDelegate):
|
||||
"""Custom delegate for wrapping text in table cells."""
|
||||
|
||||
def __init__(self, table: BECTableView, parent=None):
|
||||
super().__init__(parent)
|
||||
self._table = table
|
||||
|
||||
def paint(self, painter, option, index):
|
||||
text = index.model().data(index, QtCore.Qt.DisplayRole)
|
||||
if not text:
|
||||
@@ -139,14 +91,12 @@ class WrappingTextDelegate(DictToolTipDelegate):
|
||||
|
||||
def sizeHint(self, option, index):
|
||||
text = str(index.model().data(index, QtCore.Qt.DisplayRole) or "")
|
||||
column_width = self._table.columnWidth(index.column()) - 8 # -4 & 4
|
||||
# if not text:
|
||||
# return super().sizeHint(option, index)
|
||||
|
||||
# Avoid pathological heights for too-narrow columns
|
||||
min_width = option.fontMetrics.averageCharWidth() * 4
|
||||
if column_width < min_width:
|
||||
fm = QtGui.QFontMetrics(option.font)
|
||||
elided = fm.elidedText(text, QtCore.Qt.ElideRight, column_width)
|
||||
return QtCore.QSize(column_width, fm.height() + 4)
|
||||
# Use the actual column width
|
||||
table = index.model().parent() # or store reference to QTableView
|
||||
column_width = table.columnWidth(index.column()) # - 8
|
||||
|
||||
doc = QtGui.QTextDocument()
|
||||
doc.setDefaultFont(option.font)
|
||||
@@ -154,25 +104,8 @@ class WrappingTextDelegate(DictToolTipDelegate):
|
||||
doc.setPlainText(text)
|
||||
|
||||
layout_height = doc.documentLayout().documentSize().height()
|
||||
return QtCore.QSize(column_width, int(layout_height) + 4)
|
||||
|
||||
# def sizeHint(self, option, index):
|
||||
# text = str(index.model().data(index, QtCore.Qt.DisplayRole) or "")
|
||||
# # if not text:
|
||||
# # return super().sizeHint(option, index)
|
||||
|
||||
# # Use the actual column width
|
||||
# table = index.model().parent() # or store reference to QTableView
|
||||
# column_width = table.columnWidth(index.column()) # - 8
|
||||
|
||||
# doc = QtGui.QTextDocument()
|
||||
# doc.setDefaultFont(option.font)
|
||||
# doc.setTextWidth(column_width)
|
||||
# doc.setPlainText(text)
|
||||
|
||||
# layout_height = doc.documentLayout().documentSize().height()
|
||||
# height = int(layout_height) + 4 # Needs some extra padding, otherwise it gets cut off
|
||||
# return QtCore.QSize(column_width, height)
|
||||
height = int(layout_height) + 4 # Needs some extra padding, otherwise it gets cut off
|
||||
return QtCore.QSize(column_width, height)
|
||||
|
||||
|
||||
class DeviceTableModel(QtCore.QAbstractTableModel):
|
||||
@@ -182,16 +115,13 @@ class DeviceTableModel(QtCore.QAbstractTableModel):
|
||||
Sort logic is implemented directly on the data of the table view.
|
||||
"""
|
||||
|
||||
device_configs_added = QtCore.Signal(dict) # Dict[str, dict] of configs that were added
|
||||
devices_removed = QtCore.Signal(list) # List of strings with device names that were removed
|
||||
device_added = QtCore.Signal(dict)
|
||||
devices_reset = QtCore.Signal(list)
|
||||
|
||||
def __init__(self, parent=None):
|
||||
def __init__(self, device_config: list[dict] | None = None, parent=None):
|
||||
super().__init__(parent)
|
||||
self._device_config: dict[str, dict] = {}
|
||||
self._list_items: list[dict] = []
|
||||
self._validation_status: dict[str, ValidationStatus] = {}
|
||||
self._device_config = device_config or []
|
||||
self.headers = [
|
||||
"",
|
||||
"name",
|
||||
"deviceClass",
|
||||
"readoutPriority",
|
||||
@@ -206,7 +136,7 @@ class DeviceTableModel(QtCore.QAbstractTableModel):
|
||||
###############################################
|
||||
|
||||
def rowCount(self, parent=QtCore.QModelIndex()) -> int:
|
||||
return len(self._list_items)
|
||||
return len(self._device_config)
|
||||
|
||||
def columnCount(self, parent=QtCore.QModelIndex()) -> int:
|
||||
return len(self.headers)
|
||||
@@ -220,20 +150,15 @@ class DeviceTableModel(QtCore.QAbstractTableModel):
|
||||
"""Return the row data for the given index."""
|
||||
if not index.isValid():
|
||||
return {}
|
||||
return copy.deepcopy(self._list_items[index.row()])
|
||||
return copy.deepcopy(self._device_config[index.row()])
|
||||
|
||||
def data(self, index, role=QtCore.Qt.DisplayRole):
|
||||
"""Return data for the given index and role."""
|
||||
if not index.isValid():
|
||||
return None
|
||||
row, col = index.row(), index.column()
|
||||
|
||||
if col == 0 and role == QtCore.Qt.DisplayRole: # QtCore.Qt.DisplayRole:
|
||||
dev_name = self._list_items[row].get("name", "")
|
||||
return self._validation_status.get(dev_name, ValidationStatus.PENDING)
|
||||
|
||||
key = self.headers[col]
|
||||
value = self._list_items[row].get(key)
|
||||
value = self._device_config[row].get(key)
|
||||
|
||||
if role == QtCore.Qt.DisplayRole:
|
||||
if key in ("enabled", "readOnly"):
|
||||
@@ -288,7 +213,7 @@ class DeviceTableModel(QtCore.QAbstractTableModel):
|
||||
if key in ("enabled", "readOnly") and role == QtCore.Qt.CheckStateRole:
|
||||
if not self._checkable_columns_enabled.get(key, True):
|
||||
return False # ignore changes if column is disabled
|
||||
self._list_items[row][key] = value == QtCore.Qt.Checked
|
||||
self._device_config[row][key] = value == QtCore.Qt.Checked
|
||||
self.dataChanged.emit(index, index, [QtCore.Qt.CheckStateRole])
|
||||
return True
|
||||
return False
|
||||
@@ -297,115 +222,88 @@ class DeviceTableModel(QtCore.QAbstractTableModel):
|
||||
############ Public methods ########
|
||||
####################################
|
||||
|
||||
def get_device_config(self) -> dict[str, dict]:
|
||||
"""Method to get the device configuration."""
|
||||
def get_device_config(self) -> list[dict]:
|
||||
"""Return the current device config (with checkbox updates applied)."""
|
||||
return self._device_config
|
||||
|
||||
def add_device_configs(self, device_configs: dict[str, dict]):
|
||||
def set_checkbox_enabled(self, column_name: str, enabled: bool):
|
||||
"""
|
||||
Add devices to the model.
|
||||
Enable/Disable the checkbox column.
|
||||
|
||||
Args:
|
||||
device_configs (dict[str, dict]): A dictionary of device configurations to add.
|
||||
column_name (str): The name of the column to modify.
|
||||
enabled (bool): Whether the checkbox should be enabled or disabled.
|
||||
"""
|
||||
already_in_list = []
|
||||
for k, cfg in device_configs.items():
|
||||
if k in self._device_config:
|
||||
logger.warning(f"Device {k} already exists in the model.")
|
||||
already_in_list.append(k)
|
||||
continue
|
||||
self._device_config[k] = cfg
|
||||
new_list_cfg = copy.deepcopy(cfg)
|
||||
new_list_cfg["name"] = k
|
||||
row = len(self._list_items)
|
||||
self.beginInsertRows(QtCore.QModelIndex(), row, row)
|
||||
self._list_items.append(new_list_cfg)
|
||||
self.endInsertRows()
|
||||
for k in already_in_list:
|
||||
device_configs.pop(k)
|
||||
self.device_configs_added.emit(device_configs)
|
||||
if column_name in self._checkable_columns_enabled:
|
||||
self._checkable_columns_enabled[column_name] = enabled
|
||||
col = self.headers.index(column_name)
|
||||
top_left = self.index(0, col)
|
||||
bottom_right = self.index(self.rowCount() - 1, col)
|
||||
self.dataChanged.emit(
|
||||
top_left, bottom_right, [QtCore.Qt.CheckStateRole, QtCore.Qt.DisplayRole]
|
||||
)
|
||||
|
||||
def set_device_config(self, device_configs: dict[str, dict]):
|
||||
def set_device_config(self, device_config: list[dict]):
|
||||
"""
|
||||
Replace the device config.
|
||||
|
||||
Args:
|
||||
device_config (dict[str, dict]): The new device config to set.
|
||||
device_config (list[dict]): The new device config to set.
|
||||
"""
|
||||
diff_names = set(device_configs.keys()) - set(self._device_config.keys())
|
||||
self.beginResetModel()
|
||||
self._device_config.clear()
|
||||
self._list_items.clear()
|
||||
for k, cfg in device_configs.items():
|
||||
self._device_config[k] = cfg
|
||||
new_list_cfg = copy.deepcopy(cfg)
|
||||
new_list_cfg["name"] = k
|
||||
self._list_items.append(new_list_cfg)
|
||||
self._device_config = list(device_config)
|
||||
self.endResetModel()
|
||||
self.devices_removed.emit(diff_names)
|
||||
self.device_configs_added.emit(device_configs)
|
||||
self.devices_reset.emit(self._device_config)
|
||||
|
||||
def remove_device_configs(self, device_configs: dict[str, dict]):
|
||||
@SafeSlot(dict)
|
||||
def add_device(self, device: dict):
|
||||
"""
|
||||
Remove devices from the model.
|
||||
Add an extra device to the device config at the bottom.
|
||||
|
||||
Args:
|
||||
device_configs (dict[str, dict]): A dictionary of device configurations to remove.
|
||||
device (dict): The device configuration to add.
|
||||
"""
|
||||
removed = []
|
||||
for k in device_configs.keys():
|
||||
if k not in self._device_config:
|
||||
logger.warning(f"Device {k} does not exist in the model.")
|
||||
continue
|
||||
new_cfg = self._device_config.pop(k)
|
||||
new_cfg["name"] = k
|
||||
row = self._list_items.index(new_cfg)
|
||||
row = len(self._device_config)
|
||||
self.beginInsertRows(QtCore.QModelIndex(), row, row)
|
||||
self._device_config.append(device)
|
||||
self.endInsertRows()
|
||||
|
||||
@SafeSlot(int)
|
||||
def remove_device_by_row(self, row: int):
|
||||
"""
|
||||
Remove one device row by index. This maps to the row to the source of the data model
|
||||
|
||||
Args:
|
||||
row (int): The index of the device row to remove.
|
||||
"""
|
||||
if 0 <= row < len(self._device_config):
|
||||
self.beginRemoveRows(QtCore.QModelIndex(), row, row)
|
||||
self._list_items.pop(row)
|
||||
self._device_config.pop(row)
|
||||
self.endRemoveRows()
|
||||
removed.append(k)
|
||||
self.devices_removed.emit(removed)
|
||||
|
||||
def clear_table(self):
|
||||
@SafeSlot(list)
|
||||
def remove_devices_by_rows(self, rows: list[int]):
|
||||
"""
|
||||
Clear the table.
|
||||
"""
|
||||
device_names = list(self._device_config.keys())
|
||||
self.beginResetModel()
|
||||
self._device_config.clear()
|
||||
self._list_items.clear()
|
||||
self.endResetModel()
|
||||
self.devices_removed.emit(device_names)
|
||||
|
||||
def update_validation_status(self, device_name: str, status: int | ValidationStatus):
|
||||
"""
|
||||
Handle device status changes.
|
||||
Remove multiple device rows by their indices.
|
||||
|
||||
Args:
|
||||
device_name (str): The name of the device.
|
||||
status (int): The new status of the device.
|
||||
rows (list[int]): The indices of the device rows to remove.
|
||||
"""
|
||||
if isinstance(status, int):
|
||||
status = ValidationStatus(status)
|
||||
if device_name not in self._device_config:
|
||||
logger.warning(
|
||||
f"Device {device_name} not found in device_config dict {self._device_config}"
|
||||
)
|
||||
return
|
||||
self._validation_status[device_name] = status
|
||||
row = None
|
||||
for ii, item in enumerate(self._list_items):
|
||||
if item["name"] == device_name:
|
||||
row = ii
|
||||
for row in sorted(rows, reverse=True):
|
||||
self.remove_device_by_row(row)
|
||||
|
||||
@SafeSlot(str)
|
||||
def remove_device_by_name(self, name: str):
|
||||
"""
|
||||
Remove one device row by name.
|
||||
|
||||
Args:
|
||||
name (str): The name of the device to remove.
|
||||
"""
|
||||
for row, device in enumerate(self._device_config):
|
||||
if device.get("name") == name:
|
||||
self.remove_device_by_row(row)
|
||||
break
|
||||
if row is None:
|
||||
logger.warning(
|
||||
f"Device {device_name} not found in device_status dict {self._validation_status}"
|
||||
)
|
||||
return
|
||||
# Emit dataChanged for column 0 (status column)
|
||||
index = self.index(row, 0)
|
||||
self.dataChanged.emit(index, index, [QtCore.Qt.DisplayRole])
|
||||
|
||||
|
||||
class BECTableView(QtWidgets.QTableView):
|
||||
@@ -425,7 +323,12 @@ class BECTableView(QtWidgets.QTableView):
|
||||
if not proxy_indexes:
|
||||
return
|
||||
|
||||
source_rows = self._get_source_rows(proxy_indexes)
|
||||
# Get unique rows (proxy indices) in reverse order so removal indexes stay valid
|
||||
proxy_rows = sorted({idx.row() for idx in proxy_indexes}, reverse=True)
|
||||
# Map to source model rows
|
||||
source_rows = [
|
||||
self.model().mapToSource(self.model().index(row, 0)).row() for row in proxy_rows
|
||||
]
|
||||
|
||||
model: DeviceTableModel = self.model().sourceModel() # access underlying model
|
||||
# Delegate confirmation and removal to helper
|
||||
@@ -433,28 +336,14 @@ class BECTableView(QtWidgets.QTableView):
|
||||
if not removed:
|
||||
return
|
||||
|
||||
def _get_source_rows(self, proxy_indexes: list[QtWidgets.QModelIndex]) -> list[int]:
|
||||
"""
|
||||
Map proxy model indices to source model row indices.
|
||||
|
||||
Args:
|
||||
proxy_indexes (list[QModelIndex]): List of proxy model indices.
|
||||
|
||||
Returns:
|
||||
list[int]: List of source model row indices.
|
||||
"""
|
||||
proxy_rows = sorted({idx for idx in proxy_indexes}, reverse=True)
|
||||
source_rows = [self.model().mapToSource(idx).row() for idx in proxy_rows]
|
||||
return list(set(source_rows))
|
||||
|
||||
def _confirm_and_remove_rows(self, model: DeviceTableModel, source_rows: list[int]) -> bool:
|
||||
"""
|
||||
Prompt the user to confirm removal of rows and remove them from the model if accepted.
|
||||
|
||||
Returns True if rows were removed, False otherwise.
|
||||
"""
|
||||
configs = [model._list_items[r] for r in sorted(source_rows)]
|
||||
names = [cfg.get("name", "<unknown>") for cfg in configs]
|
||||
cfg = model.get_device_config()
|
||||
names = [str(cfg[r].get("name", "<unknown>")) for r in sorted(source_rows)]
|
||||
|
||||
msg = QtWidgets.QMessageBox(self)
|
||||
msg.setIcon(QtWidgets.QMessageBox.Warning)
|
||||
@@ -469,8 +358,8 @@ class BECTableView(QtWidgets.QTableView):
|
||||
|
||||
res = msg.exec_()
|
||||
if res == QtWidgets.QMessageBox.Ok:
|
||||
configs_to_be_removed = {model._device_config[name] for name in names}
|
||||
model.remove_device_configs(configs_to_be_removed)
|
||||
model.remove_devices_by_rows(source_rows)
|
||||
# TODO add signal for removed devices
|
||||
return True
|
||||
return False
|
||||
|
||||
@@ -482,7 +371,7 @@ class DeviceFilterProxyModel(QtCore.QSortFilterProxyModel):
|
||||
self._hidden_rows = set()
|
||||
self._filter_text = ""
|
||||
self._enable_fuzzy = True
|
||||
self._filter_columns = [1, 2] # name and deviceClass for search
|
||||
self._filter_columns = [0, 1] # name and deviceClass for search
|
||||
|
||||
def hide_rows(self, row_indices: list[int]):
|
||||
"""
|
||||
@@ -546,12 +435,11 @@ class DeviceFilterProxyModel(QtCore.QSortFilterProxyModel):
|
||||
class DeviceTableView(BECWidget, QtWidgets.QWidget):
|
||||
"""Device Table View for the device manager."""
|
||||
|
||||
selected_device = QtCore.Signal(dict) # Selected device configuration dict[str,dict]
|
||||
device_configs_added = QtCore.Signal(dict) # Dict[str, dict] of configs that were added
|
||||
devices_removed = QtCore.Signal(list) # List of strings with device names that were removed
|
||||
selected_device = QtCore.Signal(dict)
|
||||
|
||||
RPC = False
|
||||
PLUGIN = False
|
||||
devices_removed = QtCore.Signal(list)
|
||||
|
||||
def __init__(self, parent=None, client=None):
|
||||
super().__init__(client=client, parent=parent, theme_update=True)
|
||||
@@ -568,10 +456,6 @@ class DeviceTableView(BECWidget, QtWidgets.QWidget):
|
||||
self.layout.addLayout(self.search_controls)
|
||||
self.layout.addWidget(self.table)
|
||||
|
||||
# Connect signals
|
||||
self._model.devices_removed.connect(self.devices_removed.emit)
|
||||
self._model.device_configs_added.connect(self.device_configs_added.emit)
|
||||
|
||||
def _setup_search(self):
|
||||
"""Create components related to the search functionality"""
|
||||
|
||||
@@ -612,48 +496,43 @@ class DeviceTableView(BECWidget, QtWidgets.QWidget):
|
||||
"""Setup the table view."""
|
||||
# Model + Proxy
|
||||
self.table = BECTableView(self)
|
||||
self._model = DeviceTableModel(parent=self.table)
|
||||
self.model = DeviceTableModel(parent=self.table)
|
||||
self.proxy = DeviceFilterProxyModel(parent=self.table)
|
||||
self.proxy.setSourceModel(self._model)
|
||||
self.proxy.setSourceModel(self.model)
|
||||
self.table.setModel(self.proxy)
|
||||
self.table.setSortingEnabled(True)
|
||||
|
||||
# Delegates
|
||||
colors = get_accent_colors()
|
||||
self.checkbox_delegate = CenterCheckBoxDelegate(self.table, colors=colors)
|
||||
self.checkbox_delegate = CenterCheckBoxDelegate(self.table)
|
||||
self.wrap_delegate = WrappingTextDelegate(self.table)
|
||||
self.tool_tip_delegate = DictToolTipDelegate(self.table)
|
||||
self.validated_delegate = DeviceValidatedDelegate(self.table, colors=colors)
|
||||
self.table.setItemDelegateForColumn(0, self.validated_delegate) # ValidationStatus
|
||||
self.table.setItemDelegateForColumn(1, self.tool_tip_delegate) # name
|
||||
self.table.setItemDelegateForColumn(2, self.tool_tip_delegate) # deviceClass
|
||||
self.table.setItemDelegateForColumn(3, self.tool_tip_delegate) # readoutPriority
|
||||
self.table.setItemDelegateForColumn(4, self.wrap_delegate) # deviceTags
|
||||
self.table.setItemDelegateForColumn(5, self.checkbox_delegate) # enabled
|
||||
self.table.setItemDelegateForColumn(6, self.checkbox_delegate) # readOnly
|
||||
self.table.setItemDelegateForColumn(0, self.tool_tip_delegate) # name
|
||||
self.table.setItemDelegateForColumn(1, self.tool_tip_delegate) # deviceClass
|
||||
self.table.setItemDelegateForColumn(2, self.tool_tip_delegate) # readoutPriority
|
||||
self.table.setItemDelegateForColumn(3, self.wrap_delegate) # deviceTags
|
||||
self.table.setItemDelegateForColumn(4, self.checkbox_delegate) # enabled
|
||||
self.table.setItemDelegateForColumn(5, self.checkbox_delegate) # readOnly
|
||||
|
||||
# Column resize policies
|
||||
# TODO maybe we need here a flexible header options as deviceClass
|
||||
# may get quite long for beamlines plugin repos
|
||||
header = self.table.horizontalHeader()
|
||||
header.setSectionResizeMode(0, QtWidgets.QHeaderView.Fixed) # ValidationStatus
|
||||
header.setSectionResizeMode(1, QtWidgets.QHeaderView.ResizeToContents) # name
|
||||
header.setSectionResizeMode(2, QtWidgets.QHeaderView.ResizeToContents) # deviceClass
|
||||
header.setSectionResizeMode(3, QtWidgets.QHeaderView.ResizeToContents) # readoutPriority
|
||||
header.setSectionResizeMode(4, QtWidgets.QHeaderView.Stretch) # deviceTags
|
||||
header.setSectionResizeMode(5, QtWidgets.QHeaderView.Fixed) # enabled
|
||||
header.setSectionResizeMode(6, QtWidgets.QHeaderView.Fixed) # readOnly
|
||||
header.setSectionResizeMode(0, QtWidgets.QHeaderView.ResizeToContents) # name
|
||||
header.setSectionResizeMode(1, QtWidgets.QHeaderView.ResizeToContents) # deviceClass
|
||||
header.setSectionResizeMode(2, QtWidgets.QHeaderView.ResizeToContents) # readoutPriority
|
||||
header.setSectionResizeMode(3, QtWidgets.QHeaderView.Stretch) # deviceTags
|
||||
header.setSectionResizeMode(4, QtWidgets.QHeaderView.Fixed) # enabled
|
||||
header.setSectionResizeMode(5, QtWidgets.QHeaderView.Fixed) # readOnly
|
||||
|
||||
self.table.setColumnWidth(0, 25)
|
||||
self.table.setColumnWidth(5, 70)
|
||||
self.table.setColumnWidth(6, 70)
|
||||
self.table.setColumnWidth(3, 70)
|
||||
self.table.setColumnWidth(4, 70)
|
||||
|
||||
# Ensure column widths stay fixed
|
||||
header.setMinimumSectionSize(25)
|
||||
header.setMinimumSectionSize(70)
|
||||
header.setDefaultSectionSize(90)
|
||||
|
||||
# Enable resizing of column
|
||||
self._geometry_resize_proxy = BECSignalProxy(
|
||||
header.geometriesChanged, rateLimit=10, slot=self._on_table_resized
|
||||
)
|
||||
header.sectionResized.connect(self.on_table_resized)
|
||||
|
||||
# Selection behavior
|
||||
self.table.setSelectionBehavior(QtWidgets.QAbstractItemView.SelectRows)
|
||||
@@ -664,26 +543,28 @@ class DeviceTableView(BECWidget, QtWidgets.QWidget):
|
||||
|
||||
# QtCore.QTimer.singleShot(0, lambda: header.sectionResized.emit(0, 0, 0))
|
||||
|
||||
def get_device_config(self) -> dict[str, dict]:
|
||||
def device_config(self) -> list[dict]:
|
||||
"""Get the device config."""
|
||||
return self._model.get_device_config()
|
||||
return self.model.get_device_config()
|
||||
|
||||
def apply_theme(self, theme: str | None = None):
|
||||
self.checkbox_delegate.apply_theme(theme)
|
||||
self.validated_delegate.apply_theme(theme)
|
||||
|
||||
######################################
|
||||
########### Slot API #################
|
||||
######################################
|
||||
|
||||
@SafeSlot()
|
||||
def _on_table_resized(self, *args):
|
||||
@SafeSlot(int, int, int)
|
||||
def on_table_resized(self, column, old_width, new_width):
|
||||
"""Handle changes to the table column resizing."""
|
||||
option = QtWidgets.QStyleOptionViewItem()
|
||||
model = self.table.model()
|
||||
for row in range(model.rowCount()):
|
||||
index = model.index(row, 4)
|
||||
height = self.wrap_delegate.sizeHint(option, index).height()
|
||||
if column != len(self.model.headers) - 1:
|
||||
return
|
||||
|
||||
for row in range(self.table.model().rowCount()):
|
||||
index = self.table.model().index(row, column)
|
||||
delegate = self.table.itemDelegate(index)
|
||||
option = QtWidgets.QStyleOptionViewItem()
|
||||
height = delegate.sizeHint(option, index).height()
|
||||
self.table.setRowHeight(row, height)
|
||||
|
||||
@SafeSlot(QtCore.QItemSelection, QtCore.QItemSelection)
|
||||
@@ -705,106 +586,86 @@ class DeviceTableView(BECWidget, QtWidgets.QWidget):
|
||||
|
||||
source_indexes = [self.proxy.mapToSource(idx) for idx in selected_indexes]
|
||||
source_rows = {idx.row() for idx in source_indexes}
|
||||
configs = [copy.deepcopy(self._model._list_items[r]) for r in sorted(source_rows)]
|
||||
names = [cfg.pop("name") for cfg in configs]
|
||||
selected_cfgs = {name: cfg for name, cfg in zip(names, configs)}
|
||||
self.selected_device.emit(selected_cfgs)
|
||||
# Ignore if multiple are selected
|
||||
if len(source_rows) > 1:
|
||||
self.selected_device.emit({})
|
||||
return
|
||||
|
||||
# Get the single row
|
||||
(row,) = source_rows
|
||||
source_index = self.model.index(row, 0) # pick column 0 or whichever
|
||||
device = self.model.get_row_data(source_index)
|
||||
self.selected_device.emit(device)
|
||||
|
||||
@SafeSlot(QtCore.QModelIndex)
|
||||
def _on_row_selected(self, index: QtCore.QModelIndex):
|
||||
"""Handle row selection in the device table."""
|
||||
if not index.isValid():
|
||||
return
|
||||
source_index = self.proxy.mapToSource(index)
|
||||
device = self.model.get_device_at_index(source_index)
|
||||
self.selected_device.emit(device)
|
||||
|
||||
######################################
|
||||
##### Ext. Slot API #################
|
||||
######################################
|
||||
|
||||
@SafeSlot(dict)
|
||||
def set_device_config(self, device_configs: dict[str, dict]):
|
||||
@SafeSlot(list)
|
||||
def set_device_config(self, config: list[dict]):
|
||||
"""
|
||||
Set the device config.
|
||||
|
||||
Args:
|
||||
config (dict[str,dict]): The device config to set.
|
||||
config (list[dict]): The device config to set.
|
||||
"""
|
||||
self._model.set_device_config(device_configs)
|
||||
self.model.set_device_config(config)
|
||||
|
||||
@SafeSlot()
|
||||
def clear_device_configs(self):
|
||||
"""Clear the device configs."""
|
||||
self._model.clear_table()
|
||||
def clear_device_config(self):
|
||||
"""
|
||||
Clear the device config.
|
||||
"""
|
||||
self.model.set_device_config([])
|
||||
|
||||
@SafeSlot(dict)
|
||||
def add_device_configs(self, device_configs: dict[str, dict]):
|
||||
def add_device(self, device: dict):
|
||||
"""
|
||||
Add devices to the config.
|
||||
Add a device to the config.
|
||||
|
||||
Args:
|
||||
device_configs (dict[str, dict]): The device configs to add.
|
||||
device (dict): The device to add.
|
||||
"""
|
||||
self._model.add_device_configs(device_configs)
|
||||
|
||||
@SafeSlot(dict)
|
||||
def remove_device_configs(self, device_configs: dict[str, dict]):
|
||||
"""
|
||||
Remove devices from the config.
|
||||
|
||||
Args:
|
||||
device_configs (dict[str, dict]): The device configs to remove.
|
||||
"""
|
||||
self._model.remove_device_configs(device_configs)
|
||||
self.model.add_device(device)
|
||||
|
||||
@SafeSlot(int)
|
||||
@SafeSlot(str)
|
||||
def remove_device(self, device_name: str):
|
||||
def remove_device(self, dev: int | str):
|
||||
"""
|
||||
Remove a device from the config.
|
||||
Remove the device from the config either by row id, or device name.
|
||||
|
||||
Args:
|
||||
device_name (str): The name of the device to remove.
|
||||
dev (int | str): The device to remove, either by row id or device name.
|
||||
"""
|
||||
cfg = self._model._device_config.get(device_name, None)
|
||||
if cfg is None:
|
||||
logger.warning(f"Device {device_name} not found in device_config dict")
|
||||
if isinstance(dev, int):
|
||||
# TODO test this properly, check with proxy index and source index
|
||||
# Use the proxy model to map to the correct row
|
||||
model_source_index = self.table.model().mapToSource(self.table.model().index(dev, 0))
|
||||
self.model.remove_device_by_row(model_source_index.row())
|
||||
return
|
||||
if isinstance(dev, str):
|
||||
self.model.remove_device_by_name(dev)
|
||||
return
|
||||
self._model.remove_device_configs({device_name: cfg})
|
||||
|
||||
@SafeSlot(str, int)
|
||||
def update_device_validation(
|
||||
self, device_name: str, validation_status: int | ValidationStatus
|
||||
) -> None:
|
||||
"""
|
||||
Update the validation status of a device.
|
||||
|
||||
Args:
|
||||
device_name (str): The name of the device.
|
||||
validation_status (int | ValidationStatus): The new validation status.
|
||||
"""
|
||||
self._model.update_validation_status(device_name, validation_status)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
|
||||
import numpy as np
|
||||
from qtpy.QtWidgets import QApplication
|
||||
|
||||
app = QApplication(sys.argv)
|
||||
widget = QtWidgets.QWidget()
|
||||
layout = QtWidgets.QVBoxLayout(widget)
|
||||
layout.setContentsMargins(0, 0, 0, 0)
|
||||
window = DeviceTableView()
|
||||
layout.addWidget(window)
|
||||
# QPushButton
|
||||
button = QtWidgets.QPushButton("Test status_update")
|
||||
layout.addWidget(button)
|
||||
|
||||
def _button_clicked():
|
||||
names = list(window._model._device_config.keys())
|
||||
for name in names:
|
||||
window.update_device_validation(
|
||||
name, ValidationStatus.VALID if np.random.rand() > 0.5 else ValidationStatus.FAILED
|
||||
)
|
||||
|
||||
button.clicked.connect(_button_clicked)
|
||||
# pylint: disable=protected-access
|
||||
config = window.client.device_manager._get_redis_device_config()
|
||||
names = [cfg.pop("name") for cfg in config]
|
||||
config_dict = {name: cfg for name, cfg in zip(names, config)}
|
||||
window.set_device_config(config_dict)
|
||||
widget.show()
|
||||
window.set_device_config(config)
|
||||
window.show()
|
||||
sys.exit(app.exec_())
|
||||
|
||||
@@ -2,23 +2,17 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import traceback
|
||||
|
||||
import yaml
|
||||
from bec_lib.logger import bec_logger
|
||||
from qtpy import QtCore, QtWidgets
|
||||
|
||||
from bec_widgets.utils.bec_widget import BECWidget
|
||||
from bec_widgets.utils.colors import get_accent_colors, get_theme_palette
|
||||
from bec_widgets.utils.error_popups import SafeSlot
|
||||
from bec_widgets.widgets.editors.monaco.monaco_widget import MonacoWidget
|
||||
|
||||
logger = bec_logger.logger
|
||||
|
||||
|
||||
class DMConfigView(BECWidget, QtWidgets.QWidget):
|
||||
def __init__(self, parent=None, client=None):
|
||||
super().__init__(client=client, parent=parent, theme_update=True)
|
||||
super().__init__(client=client, parent=parent)
|
||||
self.stacked_layout = QtWidgets.QStackedLayout()
|
||||
self.stacked_layout.setContentsMargins(0, 0, 0, 0)
|
||||
self.stacked_layout.setSpacing(0)
|
||||
@@ -41,11 +35,12 @@ class DMConfigView(BECWidget, QtWidgets.QWidget):
|
||||
self.monaco_editor.set_minimap_enabled(False)
|
||||
# self.monaco_editor.setFixedHeight(600)
|
||||
self.monaco_editor.set_readonly(True)
|
||||
self.monaco_editor.editor.set_scroll_beyond_last_line_enabled(False)
|
||||
self.monaco_editor.editor.set_line_numbers_mode("off")
|
||||
|
||||
def _customize_overlay(self):
|
||||
self._overlay_widget.setAlignment(QtCore.Qt.AlignmentFlag.AlignCenter)
|
||||
self._overlay_widget.setStyleSheet(
|
||||
"background: qlineargradient(x1:0, y1:0, x2:0, y2:1,stop:0 #ffffff, stop:1 #e0e0e0);"
|
||||
)
|
||||
self._overlay_widget.setAutoFillBackground(True)
|
||||
self._overlay_widget.setSizePolicy(
|
||||
QtWidgets.QSizePolicy.Policy.Expanding, QtWidgets.QSizePolicy.Policy.Expanding
|
||||
@@ -54,20 +49,13 @@ class DMConfigView(BECWidget, QtWidgets.QWidget):
|
||||
@SafeSlot(dict)
|
||||
def on_select_config(self, device: dict):
|
||||
"""Handle selection of a device from the device table."""
|
||||
if len(device) != 1:
|
||||
if not device:
|
||||
text = ""
|
||||
self.stacked_layout.setCurrentWidget(self._overlay_widget)
|
||||
else:
|
||||
try:
|
||||
text = yaml.dump(device, default_flow_style=False)
|
||||
self.stacked_layout.setCurrentWidget(self.monaco_editor)
|
||||
except Exception:
|
||||
content = traceback.format_exc()
|
||||
logger.error(f"Error converting device to YAML:\n{content}")
|
||||
text = ""
|
||||
self.stacked_layout.setCurrentWidget(self._overlay_widget)
|
||||
text = yaml.dump(device, default_flow_style=False)
|
||||
self.stacked_layout.setCurrentWidget(self.monaco_editor)
|
||||
self.monaco_editor.set_readonly(False) # Enable editing
|
||||
text = text.rstrip()
|
||||
self.monaco_editor.set_text(text)
|
||||
self.monaco_editor.set_readonly(True) # Disable editing again
|
||||
|
||||
|
||||
@@ -1,128 +0,0 @@
|
||||
"""Module to visualize the docstring of a device class."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import inspect
|
||||
import re
|
||||
import traceback
|
||||
|
||||
from bec_lib.logger import bec_logger
|
||||
from bec_lib.plugin_helper import get_plugin_class, plugin_package_name
|
||||
from bec_lib.utils.rpc_utils import rgetattr
|
||||
from qtpy import QtCore, QtWidgets
|
||||
|
||||
from bec_widgets.utils.error_popups import SafeSlot
|
||||
|
||||
logger = bec_logger.logger
|
||||
|
||||
try:
|
||||
import ophyd
|
||||
import ophyd_devices
|
||||
|
||||
READY_TO_VIEW = True
|
||||
except ImportError:
|
||||
logger.warning(f"Optional dependencies not available: {ImportError}")
|
||||
ophyd_devices = None
|
||||
ophyd = None
|
||||
|
||||
|
||||
class DocstringView(QtWidgets.QTextEdit):
|
||||
def __init__(self, parent: QtWidgets.QWidget | None = None):
|
||||
super().__init__(parent)
|
||||
self.setReadOnly(True)
|
||||
self.setFocusPolicy(QtCore.Qt.NoFocus)
|
||||
if not READY_TO_VIEW:
|
||||
self._set_text("Ophyd or ophyd_devices not installed, cannot show docstrings.")
|
||||
self.setEnabled(False)
|
||||
return
|
||||
|
||||
def _format_docstring(self, doc: str | None) -> str:
|
||||
if not doc:
|
||||
return "<i>No docstring available.</i>"
|
||||
|
||||
# Escape HTML
|
||||
doc = doc.replace("&", "&").replace("<", "<").replace(">", ">")
|
||||
|
||||
# Remove leading/trailing blank lines from the entire docstring
|
||||
lines = [line.rstrip() for line in doc.splitlines()]
|
||||
while lines and lines[0].strip() == "":
|
||||
lines.pop(0)
|
||||
while lines and lines[-1].strip() == "":
|
||||
lines.pop()
|
||||
doc = "\n".join(lines)
|
||||
|
||||
# Improved regex: match section header + all following indented lines
|
||||
section_regex = re.compile(
|
||||
r"(?m)^(Parameters|Args|Returns|Examples|Attributes|Raises)\b(?:\n([ \t]+.*))*",
|
||||
re.MULTILINE,
|
||||
)
|
||||
|
||||
def strip_section(match: re.Match) -> str:
|
||||
# Capture all lines in the match
|
||||
block = match.group(0)
|
||||
lines = block.splitlines()
|
||||
# Remove leading/trailing empty lines within the section
|
||||
lines = [line for line in lines if line.strip() != ""]
|
||||
return "\n".join(lines)
|
||||
|
||||
doc = section_regex.sub(strip_section, doc)
|
||||
|
||||
# Highlight section titles
|
||||
doc = re.sub(
|
||||
r"(?m)^(Parameters|Args|Returns|Examples|Attributes|Raises)\b", r"<b>\1</b>", doc
|
||||
)
|
||||
|
||||
# Convert indented blocks to <pre> and strip leading/trailing newlines
|
||||
def pre_block(match: re.Match) -> str:
|
||||
text = match.group(0).strip("\n")
|
||||
return f"<pre>{text}</pre>"
|
||||
|
||||
doc = re.sub(r"(?m)(?:\n[ \t]+.*)+", pre_block, doc)
|
||||
|
||||
# Replace remaining newlines with <br> and collapse multiple <br>
|
||||
doc = doc.replace("\n", "<br>")
|
||||
doc = re.sub(r"(<br>)+", r"<br>", doc)
|
||||
doc = doc.strip("<br>")
|
||||
|
||||
return f"<div style='font-family: sans-serif; font-size: 12pt;'>{doc}</div>"
|
||||
|
||||
def _set_text(self, text: str):
|
||||
self.setReadOnly(False)
|
||||
self.setMarkdown(text)
|
||||
# self.setHtml(self._format_docstring(text))
|
||||
self.setReadOnly(True)
|
||||
|
||||
@SafeSlot(dict)
|
||||
def on_select_config(self, device: dict):
|
||||
if len(device) != 1:
|
||||
self._set_text("")
|
||||
return
|
||||
k = next(iter(device))
|
||||
device_class = device[k].get("deviceClass", "")
|
||||
self.set_device_class(device_class)
|
||||
|
||||
@SafeSlot(str)
|
||||
def set_device_class(self, device_class_str: str) -> None:
|
||||
docstring = ""
|
||||
if not READY_TO_VIEW:
|
||||
return
|
||||
try:
|
||||
module_cls = get_plugin_class(device_class_str, [ophyd_devices, ophyd])
|
||||
docstring = inspect.getdoc(module_cls)
|
||||
self._set_text(docstring or "No docstring available.")
|
||||
except Exception:
|
||||
content = traceback.format_exc()
|
||||
logger.error(f"Error retrieving docstring for {device_class_str}: {content}")
|
||||
self._set_text(f"Error retrieving docstring for {device_class_str}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
|
||||
from qtpy.QtWidgets import QApplication
|
||||
|
||||
app = QApplication(sys.argv)
|
||||
config_view = DocstringView()
|
||||
config_view.set_device_class("ophyd_devices.sim.sim_camera.SimCamera")
|
||||
config_view.show()
|
||||
sys.exit(app.exec_())
|
||||
@@ -1,24 +1,18 @@
|
||||
"""Module to run a static tests for devices from a yaml config."""
|
||||
"""Module to run a static test for the current config and see if it is valid."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import enum
|
||||
import re
|
||||
import traceback
|
||||
from html import escape
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import bec_lib
|
||||
from bec_lib.logger import bec_logger
|
||||
from bec_qthemes import material_icon
|
||||
from ophyd import status
|
||||
from qtpy import QtCore, QtGui, QtWidgets
|
||||
|
||||
from bec_widgets.utils.bec_widget import BECWidget
|
||||
from bec_widgets.utils.colors import get_accent_colors
|
||||
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
|
||||
from bec_widgets.widgets.editors.web_console.web_console import WebConsole
|
||||
from bec_widgets.widgets.utility.spinner.spinner import SpinnerWidget
|
||||
|
||||
READY_TO_TEST = False
|
||||
|
||||
@@ -34,380 +28,305 @@ except ImportError:
|
||||
ophyd_devices = None
|
||||
bec_server = None
|
||||
|
||||
if TYPE_CHECKING: # pragma no cover
|
||||
try:
|
||||
from ophyd_devices.utils.static_device_test import StaticDeviceTest
|
||||
except ImportError:
|
||||
StaticDeviceTest = None
|
||||
|
||||
|
||||
class ValidationStatus(int, enum.Enum):
|
||||
"""Validation status for device configurations."""
|
||||
|
||||
PENDING = 0 # colors.default
|
||||
VALID = 1 # colors.highlight
|
||||
FAILED = 2 # colors.emergency
|
||||
UNKNOWN = 0 # colors.default
|
||||
ERROR = 1 # colors.emergency
|
||||
VALID = 2 # colors.highlight
|
||||
CANT_CONNECT = 3 # colors.warning
|
||||
CONNECTED = 4 # colors.success
|
||||
|
||||
|
||||
class DeviceValidationResult(QtCore.QObject):
|
||||
"""Simple object to inject validation signals into QRunnable."""
|
||||
class DeviceValidationListItem(QtWidgets.QWidget):
|
||||
"""Custom list item widget showing device name and validation status."""
|
||||
|
||||
# Device validation signal, device_name, ValidationStatus as int, error message or ''
|
||||
device_validated = QtCore.Signal(str, bool, str)
|
||||
|
||||
|
||||
class DeviceValidationRunnable(QtCore.QRunnable):
|
||||
"""Runnable for validating a device configuration."""
|
||||
status_changed = QtCore.Signal(int) # Signal emitted when status changes -> ValidationStatus
|
||||
# Signal emitted when device was validated with name, success, msg
|
||||
device_validated = QtCore.Signal(str, str)
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
device_name: str,
|
||||
config: dict,
|
||||
static_device_test: StaticDeviceTest | None,
|
||||
connect: bool = False,
|
||||
device_config: dict[str, dict],
|
||||
status: ValidationStatus,
|
||||
status_icons: dict[ValidationStatus, QtGui.QPixmap],
|
||||
validate_icon: QtGui.QPixmap,
|
||||
parent=None,
|
||||
static_device_test=None,
|
||||
):
|
||||
"""
|
||||
Initialize the device validation runnable.
|
||||
|
||||
Args:
|
||||
device_name (str): The name of the device to validate.
|
||||
config (dict): The configuration dictionary for the device.
|
||||
static_device_test (StaticDeviceTest): The static device test instance.
|
||||
connect (bool, optional): Whether to connect to the device. Defaults to False.
|
||||
"""
|
||||
super().__init__()
|
||||
self.device_name = device_name
|
||||
self.config = config
|
||||
self._connect = connect
|
||||
self._static_device_test = static_device_test
|
||||
self.signals = DeviceValidationResult()
|
||||
|
||||
def run(self):
|
||||
"""Run method for device validation."""
|
||||
if self._static_device_test is None:
|
||||
logger.error(
|
||||
f"Ophyd devices or bec_server not available, cannot run validation for device {self.device_name}."
|
||||
)
|
||||
return
|
||||
try:
|
||||
self._static_device_test.config = {self.device_name: self.config}
|
||||
results = self._static_device_test.run_with_list_output(connect=self._connect)
|
||||
success = results[0].success
|
||||
msg = results[0].message
|
||||
self.signals.device_validated.emit(self.device_name, success, msg)
|
||||
except Exception:
|
||||
content = traceback.format_exc()
|
||||
logger.error(f"Validation failed for device {self.device_name}. Exception: {content}")
|
||||
self.signals.device_validated.emit(self.device_name, False, content)
|
||||
|
||||
|
||||
class ValidationListItem(QtWidgets.QWidget):
|
||||
"""Custom list item widget showing device name and validation status."""
|
||||
|
||||
def __init__(self, device_name: str, device_config: dict, parent=None):
|
||||
"""
|
||||
Initialize the validation list item.
|
||||
|
||||
Args:
|
||||
device_name (str): The name of the device.
|
||||
device_config (dict): The configuration of the device.
|
||||
validation_colors (dict[ValidationStatus, QtGui.QColor]): The colors for each validation status.
|
||||
parent (QtWidgets.QWidget, optional): The parent widget.
|
||||
"""
|
||||
super().__init__(parent)
|
||||
self.main_layout = QtWidgets.QHBoxLayout(self)
|
||||
self.main_layout.setContentsMargins(2, 2, 2, 2)
|
||||
self.main_layout.setSpacing(4)
|
||||
self.device_name = device_name
|
||||
if len(device_config.keys()) > 1:
|
||||
logger.warning(
|
||||
f"Multiple devices found for config: {list(device_config.keys())}, using first one"
|
||||
)
|
||||
self._static_device_test = static_device_test
|
||||
self.device_name = list(device_config.keys())[0]
|
||||
self.device_config = device_config
|
||||
self.validation_msg = "Validation in progress..."
|
||||
self.status: ValidationStatus = status
|
||||
colors = get_accent_colors()
|
||||
self._status_icon = status_icons
|
||||
self._validate_icon = validate_icon
|
||||
self._setup_ui()
|
||||
self._update_status_indicator()
|
||||
|
||||
def _setup_ui(self):
|
||||
"""Setup the UI for the list item."""
|
||||
label = QtWidgets.QLabel(self.device_name)
|
||||
self.main_layout.addWidget(label)
|
||||
self.main_layout.addStretch()
|
||||
self._spinner = SpinnerWidget(parent=self)
|
||||
self._spinner.speed = 80
|
||||
self._spinner.setFixedSize(24, 24)
|
||||
self.main_layout.addWidget(self._spinner)
|
||||
self._base_style = "font-weight: bold;"
|
||||
self.setStyleSheet(self._base_style)
|
||||
self._start_spinner()
|
||||
layout = QtWidgets.QHBoxLayout(self)
|
||||
layout.setContentsMargins(4, 4, 4, 4)
|
||||
|
||||
def _start_spinner(self):
|
||||
"""Start the spinner animation."""
|
||||
self._spinner.start()
|
||||
QtWidgets.QApplication.processEvents()
|
||||
# Device name label
|
||||
self.name_label = QtWidgets.QLabel(self.device_name)
|
||||
self.name_label.setStyleSheet("font-weight: bold;")
|
||||
layout.addWidget(self.name_label)
|
||||
|
||||
def _stop_spinner(self):
|
||||
"""Stop the spinner animation."""
|
||||
self._spinner.stop()
|
||||
self._spinner.setVisible(False)
|
||||
# Make sure status is on the right
|
||||
layout.addStretch()
|
||||
self.request_validation_button = QtWidgets.QPushButton("Validate")
|
||||
self.request_validation_button.setIcon(self._validate_icon)
|
||||
if self._static_device_test is None:
|
||||
self.request_validation_button.setDisabled(True)
|
||||
else:
|
||||
self.request_validation_button.clicked.connect(self.on_request_validation)
|
||||
# self.request_validation_button.setVisible(False) -> Hide it??
|
||||
layout.addWidget(self.request_validation_button)
|
||||
# Status indicator
|
||||
self.status_indicator = QtWidgets.QLabel()
|
||||
self._update_status_indicator()
|
||||
layout.addWidget(self.status_indicator)
|
||||
|
||||
@SafeSlot()
|
||||
def on_validation_restart(self):
|
||||
"""Handle validation restart."""
|
||||
self.validation_msg = ""
|
||||
self._start_spinner()
|
||||
self.setStyleSheet("") # Check if this works as expected
|
||||
def on_request_validation(self):
|
||||
"""Handle validate button click."""
|
||||
if self._static_device_test is None:
|
||||
logger.warning("Static device test not available.")
|
||||
return
|
||||
self._static_device_test.config = self.device_config
|
||||
# TODO logic if connect is allowed
|
||||
ret = self._static_device_test.run_with_list_output(connect=False)[0]
|
||||
if ret.success:
|
||||
self.set_status(ValidationStatus.VALID)
|
||||
else:
|
||||
self.set_status(ValidationStatus.ERROR)
|
||||
self.device_validated.emit(ret.name, ret.message)
|
||||
|
||||
@SafeSlot(str)
|
||||
def on_validation_failed(self, error_msg: str):
|
||||
"""Handle validation failure."""
|
||||
self.validation_msg = error_msg
|
||||
colors = get_accent_colors()
|
||||
self._stop_spinner()
|
||||
self.main_layout.removeWidget(self._spinner)
|
||||
self._spinner.deleteLater()
|
||||
label = QtWidgets.QLabel("")
|
||||
icon = material_icon("error", color=colors.emergency, size=(24, 24))
|
||||
label.setPixmap(icon)
|
||||
self.main_layout.addWidget(label)
|
||||
def _update_status_indicator(self):
|
||||
"""Update the status indicator color based on validation status."""
|
||||
self.status_indicator.setPixmap(self._status_icon[self.status])
|
||||
|
||||
def set_status(self, status: ValidationStatus):
|
||||
"""Update the validation status."""
|
||||
self.status = status
|
||||
self._update_status_indicator()
|
||||
self.status_changed.emit(self.status)
|
||||
|
||||
def get_status(self) -> ValidationStatus:
|
||||
"""Get the current validation status."""
|
||||
return self.status
|
||||
|
||||
|
||||
class DMOphydTest(BECWidget, QtWidgets.QWidget):
|
||||
"""Widget to test device configurations using ophyd devices."""
|
||||
class DeviceManagerOphydTest(BECWidget, QtWidgets.QWidget):
|
||||
|
||||
# Signal to emit the validation status of a device
|
||||
device_validated = QtCore.Signal(str, int)
|
||||
config_changed = QtCore.Signal(
|
||||
dict, dict
|
||||
) # Signal emitted when the device config changed, new_config, old_config
|
||||
|
||||
def __init__(self, parent=None, client=None):
|
||||
super().__init__(parent=parent, client=client)
|
||||
if not READY_TO_TEST:
|
||||
self.setDisabled(True)
|
||||
self.static_device_test = None
|
||||
self._set_disabled()
|
||||
static_device_test = None
|
||||
else:
|
||||
from ophyd_devices.utils.static_device_test import StaticDeviceTest
|
||||
|
||||
self.static_device_test = StaticDeviceTest(config_dict={})
|
||||
self._device_list_items: dict[str, QtWidgets.QListWidgetItem] = {}
|
||||
self._thread_pool = QtCore.QThreadPool.globalInstance()
|
||||
|
||||
static_device_test = StaticDeviceTest(config_dict={})
|
||||
self._static_device_test = static_device_test
|
||||
self._device_config: dict[str, dict] = {}
|
||||
self._main_layout = QtWidgets.QVBoxLayout(self)
|
||||
self._main_layout.setContentsMargins(0, 0, 0, 0)
|
||||
self._main_layout.setSpacing(4)
|
||||
|
||||
# We add a splitter between the list and the text box
|
||||
# Setup icons
|
||||
colors = get_accent_colors()
|
||||
self._validate_icon = material_icon(
|
||||
icon_name="play_arrow", color=colors.default, filled=True
|
||||
)
|
||||
self._status_icons = {
|
||||
ValidationStatus.UNKNOWN: material_icon(
|
||||
icon_name="circle", size=(12, 12), color=colors.default, filled=True
|
||||
),
|
||||
ValidationStatus.ERROR: material_icon(
|
||||
icon_name="circle", size=(12, 12), color=colors.emergency, filled=True
|
||||
),
|
||||
ValidationStatus.VALID: material_icon(
|
||||
icon_name="circle", size=(12, 12), color=colors.highlight, filled=True
|
||||
),
|
||||
ValidationStatus.CANT_CONNECT: material_icon(
|
||||
icon_name="circle", size=(12, 12), color=colors.warning, filled=True
|
||||
),
|
||||
ValidationStatus.CONNECTED: material_icon(
|
||||
icon_name="circle", size=(12, 12), color=colors.success, filled=True
|
||||
),
|
||||
}
|
||||
|
||||
self.setLayout(self._main_layout)
|
||||
|
||||
# splitter
|
||||
self.splitter = QtWidgets.QSplitter(QtCore.Qt.Orientation.Vertical)
|
||||
self._main_layout.addWidget(self.splitter)
|
||||
|
||||
self._setup_list_ui()
|
||||
self._setup_textbox_ui()
|
||||
# Add custom list
|
||||
self.setup_device_validation_list()
|
||||
|
||||
# Setup text box
|
||||
self.setup_text_box()
|
||||
|
||||
def _setup_list_ui(self):
|
||||
"""Setup the list UI."""
|
||||
self._list_widget = QtWidgets.QListWidget(self)
|
||||
self._list_widget.setSelectionMode(QtWidgets.QAbstractItemView.SingleSelection)
|
||||
self.splitter.addWidget(self._list_widget)
|
||||
# Connect signals
|
||||
self._list_widget.currentItemChanged.connect(self._on_current_item_changed)
|
||||
self.config_changed.connect(self.on_config_updated)
|
||||
|
||||
def _setup_textbox_ui(self):
|
||||
"""Setup the text box UI."""
|
||||
self._text_box = QtWidgets.QTextEdit(self)
|
||||
self._text_box.setReadOnly(True)
|
||||
self._text_box.setFocusPolicy(QtCore.Qt.NoFocus)
|
||||
self.splitter.addWidget(self._text_box)
|
||||
@SafeSlot(list)
|
||||
def on_device_config_update(self, config: list[dict]):
|
||||
old_cfg = self._device_config
|
||||
self._device_config = self._compile_device_config_list(config)
|
||||
self.config_changed.emit(self._device_config, old_cfg)
|
||||
|
||||
@SafeSlot(dict)
|
||||
def add_device_configs(self, device_configs: dict[str, dict]) -> None:
|
||||
"""Receive an update with device configs.
|
||||
def _compile_device_config_list(self, config: list[dict]) -> dict[str, dict]:
|
||||
return {dev["name"]: {k: v for k, v in dev.items() if k != "name"} for dev in config}
|
||||
|
||||
@SafeSlot(dict, dict)
|
||||
def on_config_updated(self, new_config: dict, old_config: dict):
|
||||
"""Handle config updates and refresh the validation list."""
|
||||
# Find differences for potential re-validation
|
||||
diffs = self._find_diffs(new_config, old_config)
|
||||
# Check diff first
|
||||
for diff in diffs:
|
||||
if not diff:
|
||||
continue
|
||||
if len(diff) > 1:
|
||||
logger.warning(f"Multiple devices found in diff: {diff}, using first one")
|
||||
name = list(diff.keys())[0]
|
||||
if name in self.client.device_manager.devices:
|
||||
status = ValidationStatus.CONNECTED
|
||||
else:
|
||||
status = ValidationStatus.UNKNOWN
|
||||
if self.get_device_status(diff) is None:
|
||||
self.add_device(diff, status)
|
||||
else:
|
||||
self.update_device_status(diff, status)
|
||||
|
||||
def _find_diffs(self, new_config: dict, old_config: dict) -> list[dict]:
|
||||
"""
|
||||
Return list of keys/paths where d1 and d2 differ. This goes recursively through the dictionary.
|
||||
|
||||
Args:
|
||||
device_configs (dict[str, dict]): The updated device configurations.
|
||||
new_config: The first dictionary to compare.
|
||||
old_config: The second dictionary to compare.
|
||||
"""
|
||||
for device_name, device_config in device_configs.items():
|
||||
if device_name in self._device_list_items:
|
||||
logger.error(f"Device {device_name} is already in the list.")
|
||||
return
|
||||
item = QtWidgets.QListWidgetItem(self._list_widget)
|
||||
widget = ValidationListItem(device_name=device_name, device_config=device_config)
|
||||
diffs = []
|
||||
keys = set(new_config.keys()) | set(old_config.keys())
|
||||
for k in keys:
|
||||
if k not in old_config: # New device
|
||||
diffs.append({k: new_config[k]})
|
||||
continue
|
||||
if k not in new_config: # Removed device
|
||||
diffs.append({k: old_config[k]})
|
||||
continue
|
||||
# Compare device config if exists in both
|
||||
v1, v2 = old_config[k], new_config[k]
|
||||
if isinstance(v1, dict) and isinstance(v2, dict):
|
||||
if self._find_diffs(v2, v1): # recurse: something inside changed
|
||||
diffs.append({k: new_config[k]})
|
||||
elif v1 != v2:
|
||||
diffs.append({k: new_config[k]})
|
||||
return diffs
|
||||
|
||||
# wrap it in a QListWidgetItem
|
||||
item.setSizeHint(widget.sizeHint())
|
||||
self._list_widget.addItem(item)
|
||||
self._list_widget.setItemWidget(item, widget)
|
||||
self._device_list_items[device_name] = item
|
||||
self._run_device_validation(widget)
|
||||
def setup_device_validation_list(self):
|
||||
"""Setup the device validation list."""
|
||||
# Create the custom validation list widget
|
||||
self.validation_list = QtWidgets.QListWidget()
|
||||
self.validation_list.setSelectionMode(QtWidgets.QAbstractItemView.SingleSelection)
|
||||
self.splitter.addWidget(self.validation_list)
|
||||
# self._main_layout.addWidget(self.validation_list)
|
||||
|
||||
@SafeSlot(dict)
|
||||
def remove_device_configs(self, device_configs: dict[str, dict]) -> None:
|
||||
"""Remove device configs from the list.
|
||||
def setup_text_box(self):
|
||||
"""Setup the text box for device validation messages."""
|
||||
self.validation_text_box = QtWidgets.QTextEdit()
|
||||
self.validation_text_box.setReadOnly(True)
|
||||
self.splitter.addWidget(self.validation_text_box)
|
||||
# self._main_layout.addWidget(self.validation_text_box)
|
||||
|
||||
Args:
|
||||
device_name (str): The name of the device to remove.
|
||||
"""
|
||||
for device_name in device_configs.keys():
|
||||
if device_name not in self._device_list_items:
|
||||
logger.warning(f"Device {device_name} not found in list.")
|
||||
return
|
||||
self._remove_list_item(device_name)
|
||||
@SafeSlot(str, str)
|
||||
def on_device_validated(self, device_name: str, message: str):
|
||||
"""Handle device validation results."""
|
||||
text = f"Device {device_name} was validated. Message: {message}"
|
||||
self.validation_text_box.setText(text)
|
||||
|
||||
def _remove_list_item(self, device_name: str):
|
||||
"""Remove a device from the list."""
|
||||
# Get the list item
|
||||
item = self._device_list_items.pop(device_name)
|
||||
def _set_disabled(self) -> None:
|
||||
"""Disable the full view"""
|
||||
self.setDisabled(True)
|
||||
|
||||
# Retrieve the custom widget attached to the item
|
||||
widget = self._list_widget.itemWidget(item)
|
||||
if widget is not None:
|
||||
widget.deleteLater() # clean up custom widget
|
||||
|
||||
# Remove the item from the QListWidget
|
||||
row = self._list_widget.row(item)
|
||||
self._list_widget.takeItem(row)
|
||||
|
||||
def _run_device_validation(self, widget: ValidationListItem):
|
||||
"""
|
||||
Run the device validation in a separate thread.
|
||||
|
||||
Args:
|
||||
widget (ValidationListItem): The widget to validate.
|
||||
"""
|
||||
if not READY_TO_TEST:
|
||||
logger.error("Ophyd devices or bec_server not available, cannot run validation.")
|
||||
return
|
||||
if (
|
||||
widget.device_name in self.client.device_manager.devices
|
||||
): # TODO and config has to be exact the same..
|
||||
self._on_device_validated(
|
||||
widget.device_name,
|
||||
ValidationStatus.VALID,
|
||||
f"Device {widget.device_name} is already in active config",
|
||||
)
|
||||
return
|
||||
runnable = DeviceValidationRunnable(
|
||||
device_name=widget.device_name,
|
||||
config=widget.device_config,
|
||||
static_device_test=self.static_device_test,
|
||||
connect=False,
|
||||
)
|
||||
runnable.signals.device_validated.connect(self._on_device_validated)
|
||||
self._thread_pool.start(runnable)
|
||||
|
||||
@SafeSlot(str, bool, str)
|
||||
def _on_device_validated(self, device_name: str, success: bool, message: str):
|
||||
"""Handle the device validation result.
|
||||
|
||||
Args:
|
||||
device_name (str): The name of the device.
|
||||
success (bool): Whether the validation was successful.
|
||||
message (str): The validation message.
|
||||
"""
|
||||
logger.info(f"Device {device_name} validation result: {success}, message: {message}")
|
||||
item = self._device_list_items.get(device_name, None)
|
||||
if not item:
|
||||
logger.error(f"Device {device_name} not found in the list.")
|
||||
return
|
||||
if success:
|
||||
self._remove_list_item(device_name=device_name)
|
||||
self.device_validated.emit(device_name, ValidationStatus.VALID.value)
|
||||
else:
|
||||
widget: ValidationListItem = self._list_widget.itemWidget(item)
|
||||
widget.on_validation_failed(message)
|
||||
self.device_validated.emit(device_name, ValidationStatus.FAILED.value)
|
||||
|
||||
def _on_current_item_changed(
|
||||
self, current: QtWidgets.QListWidgetItem, previous: QtWidgets.QListWidgetItem
|
||||
def add_device(
|
||||
self, device_config: dict[str, dict], status: ValidationStatus = ValidationStatus.UNKNOWN
|
||||
):
|
||||
"""Handle the current item change in the list widget.
|
||||
|
||||
Args:
|
||||
current (QListWidgetItem): The currently selected item.
|
||||
previous (QListWidgetItem): The previously selected item.
|
||||
"""
|
||||
widget: ValidationListItem = self._list_widget.itemWidget(current)
|
||||
if widget:
|
||||
try:
|
||||
formatted_html = self._format_validation_message(widget.validation_msg)
|
||||
self._text_box.setHtml(formatted_html)
|
||||
except Exception as e:
|
||||
logger.error(f"Error formatting validation message: {e}")
|
||||
self._text_box.setPlainText(widget.validation_msg)
|
||||
|
||||
def _format_validation_message(self, raw_msg: str) -> str:
|
||||
"""Simple HTML formatting for validation messages, wrapping text naturally."""
|
||||
if not raw_msg.strip():
|
||||
return "<i>Validation in progress...</i>"
|
||||
if raw_msg == "Validation in progress...":
|
||||
return "<i>Validation in progress...</i>"
|
||||
|
||||
raw_msg = escape(raw_msg)
|
||||
|
||||
# Split into lines
|
||||
lines = raw_msg.splitlines()
|
||||
summary = lines[0] if lines else "Validation Result"
|
||||
rest = "\n".join(lines[1:]).strip()
|
||||
|
||||
# Split traceback / final ERROR
|
||||
tb_match = re.search(r"(Traceback.*|ERROR:.*)$", rest, re.DOTALL | re.MULTILINE)
|
||||
if tb_match:
|
||||
main_text = rest[: tb_match.start()].strip()
|
||||
error_detail = tb_match.group().strip()
|
||||
else:
|
||||
main_text = rest
|
||||
error_detail = ""
|
||||
|
||||
# Highlight field names in orange (simple regex for word: Field)
|
||||
main_text_html = re.sub(
|
||||
r"(\b\w+\b)(?=: Field required)",
|
||||
r'<span style="color:#FF8C00; font-weight:bold;">\1</span>',
|
||||
main_text,
|
||||
)
|
||||
# Wrap in div for monospace, allowing wrapping
|
||||
main_text_html = (
|
||||
f'<div style="white-space: pre-wrap;">{main_text_html}</div>' if main_text_html else ""
|
||||
"""Add a device to the validation list."""
|
||||
# Create the custom widget
|
||||
item_widget = DeviceValidationListItem(
|
||||
device_config=device_config,
|
||||
status=status,
|
||||
status_icons=self._status_icons,
|
||||
validate_icon=self._validate_icon,
|
||||
static_device_test=self._static_device_test,
|
||||
)
|
||||
|
||||
# Traceback / error in red
|
||||
error_html = (
|
||||
f'<div style="white-space: pre-wrap; color:#A00000;">{error_detail}</div>'
|
||||
if error_detail
|
||||
else ""
|
||||
)
|
||||
# Create a list widget item
|
||||
list_item = QtWidgets.QListWidgetItem()
|
||||
list_item.setSizeHint(item_widget.sizeHint())
|
||||
|
||||
# Summary at top, dark red
|
||||
html = (
|
||||
f'<div style="font-family: monospace; font-size:13px; white-space: pre-wrap;">'
|
||||
f'<div style="font-weight:bold; color:#8B0000; margin-bottom:4px;">{summary}</div>'
|
||||
f"{main_text_html}"
|
||||
f"{error_html}"
|
||||
f"</div>"
|
||||
)
|
||||
return html
|
||||
# Add item to list and set custom widget
|
||||
self.validation_list.addItem(list_item)
|
||||
self.validation_list.setItemWidget(list_item, item_widget)
|
||||
item_widget.device_validated.connect(self.on_device_validated)
|
||||
|
||||
@SafeSlot()
|
||||
def clear_list(self):
|
||||
"""Clear the device list."""
|
||||
self._thread_pool.clear()
|
||||
if self._thread_pool.waitForDone(2000) is False: # Wait for threads to finish
|
||||
logger.error("Failed to wait for threads to finish. Removing items from the list.")
|
||||
self._device_list_items.clear()
|
||||
self._list_widget.clear()
|
||||
def update_device_status(self, device_config: dict[str, dict], status: ValidationStatus):
|
||||
"""Update the validation status for a specific device."""
|
||||
for i in range(self.validation_list.count()):
|
||||
item = self.validation_list.item(i)
|
||||
widget = self.validation_list.itemWidget(item)
|
||||
if (
|
||||
isinstance(widget, DeviceValidationListItem)
|
||||
and widget.device_config == device_config
|
||||
):
|
||||
widget.set_status(status)
|
||||
break
|
||||
|
||||
def remove_device(self, device_name: str):
|
||||
"""Remove a device from the list."""
|
||||
item = self._device_list_items.pop(device_name, None)
|
||||
if item:
|
||||
self._list_widget.removeItemWidget(item)
|
||||
def clear_devices(self):
|
||||
"""Clear all devices from the list."""
|
||||
self.validation_list.clear()
|
||||
|
||||
def get_device_status(self, device_config: dict[str, dict]) -> ValidationStatus | None:
|
||||
"""Get the validation status for a specific device."""
|
||||
for i in range(self.validation_list.count()):
|
||||
item = self.validation_list.item(i)
|
||||
widget = self.validation_list.itemWidget(item)
|
||||
if (
|
||||
isinstance(widget, DeviceValidationListItem)
|
||||
and widget.device_config == device_config
|
||||
):
|
||||
return widget.get_status()
|
||||
return None
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
|
||||
from bec_lib.bec_yaml_loader import yaml_load
|
||||
|
||||
# pylint: disable=ungrouped-imports
|
||||
from qtpy.QtWidgets import QApplication
|
||||
|
||||
app = QApplication(sys.argv)
|
||||
device_manager_ophyd_test = DMOphydTest()
|
||||
config_path = "/Users/appel_c/work_psi_awi/bec_workspace/csaxs_bec/csaxs_bec/device_configs/endstation.yaml"
|
||||
cfg = yaml_load(config_path)
|
||||
cfg.update({"device_will_fail": {"name": "device_will_fail", "some_param": 1}})
|
||||
device_manager_ophyd_test.add_device_configs(cfg)
|
||||
device_manager_ophyd_test = DeviceManagerOphydTest()
|
||||
cfg = device_manager_ophyd_test.client.device_manager._get_redis_device_config()
|
||||
cfg.append({"name": "Wrong_Device", "type": "test"})
|
||||
device_manager_ophyd_test.on_device_config_update(cfg)
|
||||
device_manager_ophyd_test.show()
|
||||
device_manager_ophyd_test.setWindowTitle("Device Manager Ophyd Test")
|
||||
device_manager_ophyd_test.resize(800, 600)
|
||||
|
||||
@@ -11,19 +11,13 @@ 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 QSize, QThreadPool, Signal
|
||||
from qtpy.QtWidgets import (
|
||||
QFileDialog,
|
||||
QListWidget,
|
||||
QListWidgetItem,
|
||||
QToolButton,
|
||||
QVBoxLayout,
|
||||
QWidget,
|
||||
)
|
||||
from qtpy.QtCore import QThreadPool, Signal
|
||||
from qtpy.QtWidgets import QFileDialog, QListWidget, QToolButton, QVBoxLayout, QWidget
|
||||
|
||||
from bec_widgets.cli.rpc.rpc_register import RPCRegister
|
||||
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.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 (
|
||||
@@ -59,7 +53,8 @@ class DeviceBrowser(BECWidget, QWidget):
|
||||
self._q_threadpool = QThreadPool()
|
||||
self.ui = None
|
||||
self.init_ui()
|
||||
self.dev_list: QListWidget = self.ui.device_list
|
||||
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
|
||||
@@ -132,25 +127,15 @@ class DeviceBrowser(BECWidget, QWidget):
|
||||
|
||||
def init_device_list(self):
|
||||
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()}")
|
||||
|
||||
def _remove_item(item: QListWidgetItem):
|
||||
self.dev_list.takeItem(self.dev_list.row(item))
|
||||
del self._device_items[device]
|
||||
self.dev_list.sortItems()
|
||||
|
||||
item = QListWidgetItem(self.dev_list)
|
||||
device_item = DeviceItem(
|
||||
device_item = self.dev_list.add_item(
|
||||
id=device,
|
||||
parent=self,
|
||||
device=device,
|
||||
devices=self.dev,
|
||||
@@ -158,18 +143,11 @@ class DeviceBrowser(BECWidget, QWidget):
|
||||
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))
|
||||
|
||||
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)
|
||||
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(dict, dict)
|
||||
def scan_status_changed(self, scan_info: dict, _: dict):
|
||||
@@ -198,20 +176,11 @@ 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:
|
||||
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])
|
||||
try:
|
||||
self.regex = re.compile(filter_text, re.IGNORECASE)
|
||||
except re.error:
|
||||
self.regex = None # Invalid regex, disable filtering
|
||||
for device in self.dev:
|
||||
self._device_items[device].setHidden(False)
|
||||
return
|
||||
for device in self.dev:
|
||||
self._device_items[device].setHidden(not self.regex.search(device))
|
||||
self.dev_list.update_filter(self.ui.filter_input.text())
|
||||
|
||||
@SafeSlot()
|
||||
def _load_from_file(self):
|
||||
|
||||
@@ -1,93 +1,90 @@
|
||||
<?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"/>
|
||||
<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="text">
|
||||
<string>warning</string>
|
||||
<property name="windowTitle">
|
||||
<string>Form</string>
|
||||
</property>
|
||||
</widget>
|
||||
</item>
|
||||
<item>
|
||||
<widget class="QListWidget" name="device_list"/>
|
||||
</item>
|
||||
</layout>
|
||||
<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>
|
||||
</item>
|
||||
</layout>
|
||||
</widget>
|
||||
<resources/>
|
||||
<connections/>
|
||||
</ui>
|
||||
<resources />
|
||||
<connections />
|
||||
</ui>
|
||||
@@ -35,9 +35,6 @@ logger = bec_logger.logger
|
||||
|
||||
|
||||
class DeviceItem(ExpandableGroupFrame):
|
||||
broadcast_size_hint = Signal(QSize)
|
||||
imminent_deletion = Signal()
|
||||
|
||||
RPC = False
|
||||
|
||||
def __init__(
|
||||
|
||||
@@ -37,11 +37,11 @@ def device_browser(qtbot, mocked_client):
|
||||
yield dev_browser
|
||||
|
||||
|
||||
def test_device_browser_init_with_devices(device_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.ui.device_list
|
||||
device_list = device_browser.dev_list
|
||||
assert device_list.count() == len(device_browser.dev)
|
||||
|
||||
|
||||
@@ -58,11 +58,11 @@ def test_device_browser_filtering(
|
||||
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.isHidden(), item_dict.values())))
|
||||
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._device_items) == expected
|
||||
assert num_visible(device_browser.dev_list._item_dict) == expected
|
||||
|
||||
|
||||
def test_device_item_mouse_press_event(device_browser, qtbot):
|
||||
@@ -70,8 +70,8 @@ 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.ui.device_list.itemAt(0, 0)
|
||||
widget: DeviceItem = device_browser.ui.device_list.itemWidget(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)
|
||||
|
||||
|
||||
@@ -88,8 +88,8 @@ 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.ui.device_list.itemAt(0, 0)
|
||||
widget: DeviceItem = device_browser.ui.device_list.itemWidget(device_item)
|
||||
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)
|
||||
@@ -115,8 +115,8 @@ def test_device_item_mouse_press_and_move_events_creates_drag(device_browser, qt
|
||||
"""
|
||||
Test that the mousePressEvent is triggered correctly and initiates a drag.
|
||||
"""
|
||||
device_item: QListWidgetItem = device_browser.ui.device_list.itemAt(0, 0)
|
||||
widget: DeviceItem = device_browser.ui.device_list.itemWidget(device_item)
|
||||
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:
|
||||
@@ -133,19 +133,19 @@ 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.ui.device_list.itemAt(0, 0)
|
||||
widget: DeviceItem = device_browser.ui.device_list.itemWidget(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.ui.device_list.itemAt(0, 0)
|
||||
widget: DeviceItem = device_browser.ui.device_list.itemWidget(device_item)
|
||||
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._device_items
|
||||
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._device_items, timeout=10000)
|
||||
qtbot.waitUntil(lambda: widget.device not in device_browser.dev_list._item_dict, timeout=10000)
|
||||
|
||||
|
||||
def test_signal_display(mocked_client, qtbot):
|
||||
|
||||
Reference in New Issue
Block a user