bec/bec_widgets
1
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-03-05 00:12:49 +01:00
Code Activity

refactor: cleanup

Browse Source
This commit is contained in:
appel_c
2025-09-30 08:18:16 +02:00
committed by wyzula-jan
parent d8900a579f
commit 12a454f762
6 changed files with 181 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
bec_widgets/examples/device_manager_view/device_manager_view.py
View File

@@ -96,6 +96,7 @@ class DeviceManagerView(BECWidget, QWidget):
self._root_layout.setContentsMargins(0, 0, 0, 0)
self._root_layout.setSpacing(0)
self.dock_manager = CDockManager(self)
self.dock_manager.setStyleSheet("")
self._root_layout.addWidget(self.dock_manager)
# Available Resources Widget
@@ -277,7 +278,9 @@ class DeviceManagerView(BECWidget, QWidget):
# Rerun validation
rerun_validation = MaterialIconAction(
icon_name="checklist", parent=self, tooltip="Run device validation on selected devices"
icon_name="checklist",
parent=self,
tooltip="Run device validation with 'connect' on selected devices",
)
rerun_validation.action.triggered.connect(self._rerun_validation_action)
self.toolbar.components.add_safe("rerun_validation", rerun_validation)
@@ -433,8 +436,8 @@ class DeviceManagerView(BECWidget, QWidget):
@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 = self._coming_soon()
configs = self.device_table_view.table.selected_configs()
self.ophyd_test_view.change_device_configs(configs, True, True)
####### Default view has to be done with setting up splitters ########
def set_default_view(self, horizontal_weights: list, vertical_weights: list):

16
bec_widgets/examples/device_manager_view/device_manager_widget.py
View File

@@ -100,10 +100,24 @@ if __name__ == "__main__":
from qtpy.QtWidgets import QApplication
app = QApplication(sys.argv)
from bec_widgets.utils.colors import apply_theme
apply_theme("light")
widget = QtWidgets.QWidget()
layout = QtWidgets.QVBoxLayout(widget)
widget.setLayout(layout)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(0)
device_manager = DeviceManagerWidget()
# config = device_manager.client.device_manager._get_redis_device_config()
# device_manager.device_table_view.set_device_config(config)
device_manager.show()
layout.addWidget(device_manager)
from bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button import DarkModeButton
dark_mode_button = DarkModeButton()
layout.addWidget(dark_mode_button)
widget.show()
device_manager.setWindowTitle("Device Manager View")
device_manager.resize(1600, 1200)
# developer_view.set_stretch(horizontal=[1, 3, 2], vertical=[5, 5]) #can be set during runtime

32
bec_widgets/widgets/control/device_manager/components/device_table_view.py
View File

@@ -7,22 +7,13 @@ import json
from contextlib import contextmanager
from functools import partial
from typing import TYPE_CHECKING, Any, Iterable, List
from unittest.mock import MagicMock, patch
from uuid import uuid4
from bec_lib.logger import bec_logger
from bec_qthemes import material_icon
from qtpy import QtCore, QtGui, QtWidgets
from qtpy.QtCore import QModelIndex, QPersistentModelIndex, QPoint, QRect, QSize, Qt, QTimer
from qtpy.QtWidgets import (
QAbstractItemView,
QHeaderView,
QMessageBox,
QStyle,
QStyleOption,
QStyleOptionViewItem,
QWidget,
)
from qtpy.QtCore import QModelIndex, QPersistentModelIndex, Qt, QTimer
from qtpy.QtWidgets import QAbstractItemView, QHeaderView, QMessageBox
from thefuzz import fuzz
from bec_widgets.utils.bec_signal_proxy import BECSignalProxy
@@ -91,8 +82,8 @@ class CenterCheckBoxDelegate(CustomDisplayDelegate):
def __init__(self, parent=None, colors=None):
super().__init__(parent)
self._colors: AccentColors = colors if colors else get_accent_colors() # type: ignore
_icon = partial(material_icon, size=(16, 16), color=self._colors.default, filled=True)
colors: AccentColors = colors if colors else get_accent_colors() # type: ignore
_icon = partial(material_icon, size=(16, 16), color=colors.default, filled=True)
self._icon_checked = _icon("check_box")
self._icon_unchecked = _icon("check_box_outline_blank")
@@ -122,12 +113,12 @@ class DeviceValidatedDelegate(CustomDisplayDelegate):
def __init__(self, parent=None, colors=None):
super().__init__(parent)
self._colors = colors if colors else get_accent_colors()
colors = colors if colors else get_accent_colors()
_icon = partial(material_icon, icon_name="circle", size=(12, 12), filled=True)
self._icons = {
ValidationStatus.PENDING: _icon(color=self._colors.default),
ValidationStatus.VALID: _icon(color=self._colors.success),
ValidationStatus.FAILED: _icon(color=self._colors.emergency),
ValidationStatus.PENDING: _icon(color=colors.default),
ValidationStatus.VALID: _icon(color=colors.success),
ValidationStatus.FAILED: _icon(color=colors.emergency),
}
def apply_theme(self, theme: str | None = None):
@@ -750,6 +741,7 @@ class DeviceTableView(BECWidget, QtWidgets.QWidget):
########### Slot API #################
######################################
# TODO RESIZING IS not working as it should be !!
@SafeSlot()
def _on_table_resized(self, *args):
"""Handle changes to the table column resizing."""
@@ -870,8 +862,8 @@ if __name__ == "__main__":
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)
# names = [cfg.pop("name") for cfg in config]
# config_dict = {name: cfg for name, cfg in zip(names, config)}
window.set_device_config(config)
widget.show()
sys.exit(app.exec_())

21
bec_widgets/widgets/control/device_manager/components/dm_config_view.py
View File

@@ -9,7 +9,6 @@ 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
@@ -78,6 +77,24 @@ if __name__ == "__main__":
from qtpy.QtWidgets import QApplication
app = QApplication(sys.argv)
widget = QtWidgets.QWidget()
layout = QtWidgets.QVBoxLayout(widget)
widget.setLayout(layout)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(0)
config_view = DMConfigView()
config_view.show()
layout.addWidget(config_view)
combo_box = QtWidgets.QComboBox()
config = config_view.client.device_manager._get_redis_device_config()
combo_box.addItems([""] + [str(v) for v, item in enumerate(config)])
def on_select(text):
if text == "":
config_view.on_select_config([])
else:
config_view.on_select_config([config[int(text)]])
combo_box.currentTextChanged.connect(on_select)
layout.addWidget(combo_box)
widget.show()
sys.exit(app.exec_())

122
bec_widgets/widgets/control/device_manager/components/dm_docstring_view.py
View File

@@ -4,6 +4,7 @@ from __future__ import annotations
import inspect
import re
import textwrap
import traceback
from bec_lib.logger import bec_logger
@@ -26,6 +27,45 @@ except ImportError:
ophyd = None
def docstring_to_markdown(obj) -> str:
"""
Convert a Python docstring to Markdown suitable for QTextEdit.setMarkdown.
"""
raw = inspect.getdoc(obj) or "*No docstring available.*"
# Dedent and normalize newlines
text = textwrap.dedent(raw).strip()
md = ""
if hasattr(obj, "__name__"):
md += f"# {obj.__name__}\n\n"
# Highlight section headers for Markdown
headers = ["Parameters", "Args", "Returns", "Raises", "Attributes", "Examples", "Notes"]
for h in headers:
doc = re.sub(rf"(?m)^({h})\s*:?\s*$", rf"### \1", text)
# Preserve code blocks (4+ space indented lines)
def fence_code(match: re.Match) -> str:
block = re.sub(r"^ {4}", "", match.group(0), flags=re.M)
return f"```\n{block}\n```"
doc = re.sub(r"(?m)(^ {4,}.*(\n {4,}.*)*)", fence_code, text)
# Preserve normal line breaks for Markdown
lines = doc.splitlines()
processed_lines = []
for line in lines:
if line.strip() == "":
processed_lines.append("")
else:
processed_lines.append(line + " ")
doc = "\n".join(processed_lines)
md += doc
return md
class DocstringView(QtWidgets.QTextEdit):
def __init__(self, parent: QtWidgets.QWidget | None = None):
super().__init__(parent)
@@ -36,60 +76,9 @@ class DocstringView(QtWidgets.QTextEdit):
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("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
# 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(list)
@@ -102,17 +91,15 @@ class DocstringView(QtWidgets.QTextEdit):
@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.")
markdown = docstring_to_markdown(module_cls)
self._set_text(markdown)
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}")
logger.exception("Error retrieving docstring")
self._set_text(f"*Error retrieving docstring for `{device_class_str}`*")
if __name__ == "__main__":
@@ -121,7 +108,26 @@ if __name__ == "__main__":
from qtpy.QtWidgets import QApplication
app = QApplication(sys.argv)
widget = QtWidgets.QWidget()
layout = QtWidgets.QVBoxLayout(widget)
widget.setLayout(layout)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(0)
config_view = DocstringView()
config_view.set_device_class("ophyd_devices.sim.sim_camera.SimCamera")
config_view.show()
layout.addWidget(config_view)
combo = QtWidgets.QComboBox()
combo.addItems(
[
"",
"ophyd_devices.sim.sim_camera.SimCamera",
"ophyd.EpicsSignalWithRBV",
"ophyd.EpicsMotor",
"csaxs_bec.devices.epics.mcs_card.mcs_card_csaxs.MCSCardCSAXS",
]
)
combo.currentTextChanged.connect(config_view.set_device_class)
layout.addWidget(combo)
widget.show()
sys.exit(app.exec_())

135
bec_widgets/widgets/control/device_manager/components/dm_ophyd_test.py
View File

@@ -83,23 +83,23 @@ class DeviceTester(QtCore.QRunnable):
continue
with self._lock:
if len(self._pending_queue) > 0:
item, cfg = self._pending_queue.pop()
item, cfg, connect = self._pending_queue.pop()
self._active.add(item)
fut = self._test_executor.submit(self._run_test, item, {item: cfg})
fut = self._test_executor.submit(self._run_test, item, {item: cfg}, connect)
fut.__dict__["__device_name"] = item
fut.add_done_callback(self._done_cb)
self._safe_check_and_clear()
self._cleanup()
def submit(self, devices: Iterable[tuple[str, dict]]):
def submit(self, devices: Iterable[tuple[str, dict, bool]]):
with self._lock:
self._pending_queue.extend(devices)
self._pending_event.set()
@staticmethod
def _run_test(name: str, config: dict) -> tuple[str, bool, str]:
def _run_test(name: str, config: dict, connect: bool) -> tuple[str, bool, str]:
tester = StaticDeviceTest(config_dict=config) # type: ignore # we exit early if it is None
results = tester.run_with_list_output(connect=False)
results = tester.run_with_list_output(connect=connect)
return name, results[0].success, results[0].message
def _safe_check_and_clear(self):
@@ -164,7 +164,6 @@ class ValidationListItem(QtWidgets.QWidget):
def _start_spinner(self):
"""Start the spinner animation."""
self._spinner.start()
QtWidgets.QApplication.processEvents()
def _stop_spinner(self):
"""Stop the spinner animation."""
@@ -197,6 +196,8 @@ class DMOphydTest(BECWidget, QtWidgets.QWidget):
# Signal to emit the validation status of a device
device_validated = QtCore.Signal(str, int)
# validation_msg in markdown format
validation_msg_md = QtCore.Signal(str)
def __init__(self, parent=None, client=None):
super().__init__(parent=parent, client=client)
@@ -208,18 +209,18 @@ class DMOphydTest(BECWidget, QtWidgets.QWidget):
self.tester.signals.device_validated.connect(self._on_device_validated)
QtCore.QThreadPool.globalInstance().start(self.tester)
self._device_list_items: dict[str, QtWidgets.QListWidgetItem] = {}
self._thread_pool = QtCore.QThreadPool(maxThreadCount=1)
# TODO Consider using the thread pool from BECConnector instead of fetching the global instance!
self._thread_pool = QtCore.QThreadPool.globalInstance()
self._main_layout = QtWidgets.QVBoxLayout(self)
self._main_layout.setContentsMargins(0, 0, 0, 0)
self._main_layout.setSpacing(4)
self._main_layout.setSpacing(0)
# We add a splitter between the list and the text box
self.splitter = QtWidgets.QSplitter(QtCore.Qt.Orientation.Vertical)
self._main_layout.addWidget(self.splitter)
self._setup_list_ui()
self._setup_textbox_ui()
def _setup_list_ui(self):
"""Setup the list UI."""
@@ -229,15 +230,11 @@ class DMOphydTest(BECWidget, QtWidgets.QWidget):
# Connect signals
self._list_widget.currentItemChanged.connect(self._on_current_item_changed)
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(dict)
def change_device_configs(self, device_configs: list[dict[str, Any]], added: bool) -> None:
@SafeSlot(list, bool)
@SafeSlot(list, bool, bool)
def change_device_configs(
self, device_configs: list[dict[str, Any]], added: bool, connect: bool = False
) -> None:
"""Receive an update with device configs.
Args:
@@ -250,7 +247,7 @@ class DMOphydTest(BECWidget, QtWidgets.QWidget):
continue
if self.tester:
self._add_device(name, cfg)
self.tester.submit([(name, cfg)])
self.tester.submit([(name, cfg, connect)])
continue
if name not in self._device_list_items:
continue
@@ -314,62 +311,39 @@ class DMOphydTest(BECWidget, QtWidgets.QWidget):
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)
formatted_md = self._format_markdown_text(widget.device_name, widget.validation_msg)
self.validation_msg_md.emit(formatted_md)
except Exception as e:
logger.error(f"Error formatting validation message: {e}")
self._text_box.setPlainText(widget.validation_msg)
logger.error(
f"##Error formatting validation message for device {widget.device_name}:\n{e}"
)
self.validation_msg_md.emit(widget.validation_msg)
else:
self.validation_msg_md.emit("")
def _format_validation_message(self, raw_msg: str) -> str:
def _format_markdown_text(self, device_name: str, raw_msg: str) -> str:
"""Simple HTML formatting for validation messages, wrapping text naturally."""
if not raw_msg.strip():
return "<i>Validation in progress...</i>"
return f"### Validation in progress for {device_name}... \n\n"
if raw_msg == "Validation in progress...":
return "<i>Validation in progress...</i>"
return f"### Validation in progress for {device_name}... \n\n"
raw_msg = escape(raw_msg)
m = re.search(r"ERROR:\s*([^\s]+)\s+is not valid:\s*(.+?errors?)", raw_msg)
device, summary = m.group(1), m.group(2)
lines = [f"## Error for '{device}'", f"'{device}' is not valid: {summary}"]
# 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 ""
# Find each field block: \n<field>\n Field required ...
field_pat = re.compile(
r"\n(?P<field>\w+)\n\s+(?P<rest>Field required.*?(?=\n\w+\n|$))", re.DOTALL
)
# Traceback / error in red
error_html = (
f'<div style="white-space: pre-wrap; color:#A00000;">{error_detail}</div>'
if error_detail
else ""
)
for m in field_pat.finditer(raw_msg):
field = m.group("field")
rest = m.group("rest").rstrip()
lines.append(f"### {field}")
lines.append(rest)
# 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
return "\n".join(lines)
def validation_running(self):
return self._device_list_items != {}
@@ -382,6 +356,7 @@ class DMOphydTest(BECWidget, QtWidgets.QWidget):
logger.error("Failed to wait for threads to finish. Removing items from the list.")
self._device_list_items.clear()
self._list_widget.clear()
self.validation_msg_md.emit("")
def remove_device(self, device_name: str):
"""Remove a device from the list."""
@@ -404,12 +379,32 @@ if __name__ == "__main__":
from qtpy.QtWidgets import QApplication
app = QApplication(sys.argv)
wid = QtWidgets.QWidget()
layout = QtWidgets.QVBoxLayout(wid)
wid.setLayout(layout)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(0)
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.show()
try:
config_path = "/Users/appel_c/work_psi_awi/bec_workspace/csaxs_bec/csaxs_bec/device_configs/endstation.yaml"
config = [{"name": k, **v} for k, v in yaml_load(config_path).items()]
except Exception as e:
logger.error(f"Error loading config: {e}")
import os
import bec_lib
config_path = os.path.join(os.path.dirname(bec_lib.__file__), "configs", "demo_config.yaml")
config = [{"name": k, **v} for k, v in yaml_load(config_path).items()]
config.append({"name": "non_existing_device", "type": "NonExistingDevice"})
device_manager_ophyd_test.change_device_configs(config, True, True)
layout.addWidget(device_manager_ophyd_test)
device_manager_ophyd_test.setWindowTitle("Device Manager Ophyd Test")
device_manager_ophyd_test.resize(800, 600)
text_box = QtWidgets.QTextEdit()
text_box.setReadOnly(True)
layout.addWidget(text_box)
device_manager_ophyd_test.validation_msg_md.connect(text_box.setMarkdown)
wid.show()
sys.exit(app.exec_())