1
0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-04-09 18:20:55 +02:00

Compare commits

..

9 Commits

Author SHA1 Message Date
semantic-release
bdff736aa2 2.29.0
Automatically generated by python-semantic-release
2025-07-22 11:39:06 +00:00
7cda2ed846 refactor(notification_banner): BECNotificationBroker done as singleton to sync all windows in the session 2025-07-22 13:38:23 +02:00
cd9d22d0b4 feat(notification_banner): notification centre for alarms implemented into BECMainWindow 2025-07-22 13:38:23 +02:00
semantic-release
37b80e16a0 2.28.0
Automatically generated by python-semantic-release
2025-07-21 12:23:48 +00:00
7f0098f153 feat: save and load config from devicebrowser 2025-07-21 14:23:01 +02:00
8489ef4a69 feat: remove and readd device for config changes 2025-07-21 14:23:01 +02:00
13976557fb feat: disable editing while scan active 2025-07-21 14:23:01 +02:00
semantic-release
06ad87ce0a 2.27.1
Automatically generated by python-semantic-release
2025-07-17 13:22:03 +00:00
00e3713181 fix(image_roi_tree): rois signals are disconnected when roi tree widget is closed 2025-07-17 15:21:11 +02:00
17 changed files with 1934 additions and 286 deletions

View File

@@ -1,6 +1,42 @@
# CHANGELOG
## v2.29.0 (2025-07-22)
### Features
- **notification_banner**: Notification centre for alarms implemented into BECMainWindow
([`cd9d22d`](https://github.com/bec-project/bec_widgets/commit/cd9d22d0b40d633af76cb1188b57feb7b6a5dbf2))
### Refactoring
- **notification_banner**: Becnotificationbroker done as singleton to sync all windows in the
session
([`7cda2ed`](https://github.com/bec-project/bec_widgets/commit/7cda2ed846d3c27799f4f15f6c5c667631b1ca55))
## v2.28.0 (2025-07-21)
### Features
- Disable editing while scan active
([`1397655`](https://github.com/bec-project/bec_widgets/commit/13976557fbdb71a1161029521d81a655d25dd134))
- Remove and readd device for config changes
([`8489ef4`](https://github.com/bec-project/bec_widgets/commit/8489ef4a69d69b39648b1a9270012f14f95c6121))
- Save and load config from devicebrowser
([`7f0098f`](https://github.com/bec-project/bec_widgets/commit/7f0098f1533d419cc75801c4d6cbea485c7bbf94))
## v2.27.1 (2025-07-17)
### Bug Fixes
- **image_roi_tree**: Rois signals are disconnected when roi tree widget is closed
([`00e3713`](https://github.com/bec-project/bec_widgets/commit/00e3713181916a432e4e9dec8a0d80205914cf77))
## v2.27.0 (2025-07-17)
### Features

View File

@@ -1,194 +0,0 @@
import sys
import uuid
import pyqtgraph as pg
from bec_lib.endpoints import MessageEndpoints
from bec_lib.logger import bec_logger
from qtpy.QtCore import Qt
from qtpy.QtWidgets import QApplication, QFileDialog, QFrame, QSplitter, QVBoxLayout, QWidget
from bec_widgets.utils.bec_widget 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.editors.monaco.monaco_widget import MonacoWidget
from bec_widgets.widgets.editors.web_console.web_console import WebConsole
logger = bec_logger.logger
class ScriptInterface(BECWidget, QWidget):
"""
A simple script interface widget that allows interaction with Monaco editor and Web Console.
"""
PLUGIN = True
ICON_NAME = "terminal"
def __init__(self, parent=None, config=None, client=None, gui_id=None, **kwargs):
super().__init__(
parent=parent, client=client, gui_id=gui_id, config=config, theme_update=True, **kwargs
)
self.current_script_id = ""
layout = QVBoxLayout()
layout.setContentsMargins(0, 0, 0, 0)
self.toolbar = ModularToolBar(parent=self, orientation="horizontal")
self.splitter = QSplitter(self)
self.splitter.setObjectName("splitter")
self.splitter.setFrameShape(QFrame.Shape.NoFrame)
self.splitter.setOrientation(Qt.Orientation.Vertical)
self.splitter.setChildrenCollapsible(True)
self.monaco_editor = MonacoWidget(self)
self.splitter.addWidget(self.monaco_editor)
self.web_console = WebConsole(self)
self.splitter.addWidget(self.web_console)
layout.addWidget(self.toolbar)
layout.addWidget(self.splitter)
self.setLayout(layout)
self.toolbar.components.add_safe(
"new_script", MaterialIconAction("add", "New Script", parent=self)
)
self.toolbar.components.add_safe(
"open", MaterialIconAction("folder_open", "Open Script", parent=self)
)
self.toolbar.components.add_safe(
"save", MaterialIconAction("save", "Save Script", parent=self)
)
self.toolbar.components.add_safe(
"run", MaterialIconAction("play_arrow", "Run Script", parent=self)
)
self.toolbar.components.add_safe(
"stop", MaterialIconAction("stop", "Stop Script", parent=self)
)
bundle = ToolbarBundle("file_io", self.toolbar.components)
bundle.add_action("new_script")
bundle.add_action("open")
bundle.add_action("save")
self.toolbar.add_bundle(bundle)
bundle = ToolbarBundle("script_execution", self.toolbar.components)
bundle.add_action("run")
bundle.add_action("stop")
self.toolbar.add_bundle(bundle)
self.toolbar.components.get_action("open").action.triggered.connect(self.open_file_dialog)
self.toolbar.components.get_action("run").action.triggered.connect(self.run_script)
self.toolbar.components.get_action("stop").action.triggered.connect(
self.web_console.send_ctrl_c
)
self.set_save_button_enabled(False)
self.toolbar.show_bundles(["file_io", "script_execution"])
self.web_console.set_readonly(True)
self._init_file_content = ""
self._text_changed_proxy = pg.SignalProxy(
self.monaco_editor.text_changed, rateLimit=1, slot=self._on_text_changed
)
@SafeSlot(str)
def _on_text_changed(self, text: str):
"""
Handle text changes in the Monaco editor.
"""
text = text[0]
if text != self._init_file_content:
self.set_save_button_enabled(True)
else:
self.set_save_button_enabled(False)
@property
def current_script_id(self):
return self._current_script_id
@current_script_id.setter
def current_script_id(self, value):
if not isinstance(value, str):
raise ValueError("Script ID must be a string.")
self._current_script_id = value
self._update_subscription()
def _update_subscription(self):
if self.current_script_id:
self.bec_dispatcher.connect_slot(
self.on_script_execution_info,
MessageEndpoints.script_execution_info(self.current_script_id),
)
else:
self.bec_dispatcher.disconnect_slot(
self.on_script_execution_info,
MessageEndpoints.script_execution_info(self.current_script_id),
)
@SafeSlot(dict, dict)
def on_script_execution_info(self, content: dict, metadata: dict):
print(f"Script execution info: {content}")
current_lines = content.get("current_lines")
if not current_lines:
self.monaco_editor.clear_highlighted_lines()
return
line_number = current_lines[0]
self.monaco_editor.clear_highlighted_lines()
self.monaco_editor.set_highlighted_lines(line_number, line_number)
def open_file_dialog(self):
"""
Open a file dialog to select a script file.
"""
start_dir = "./"
dialog = QFileDialog(self)
dialog.setDirectory(start_dir)
dialog.setNameFilter("Python Files (*.py);;All Files (*)")
dialog.setFileMode(QFileDialog.FileMode.ExistingFile)
if dialog.exec():
selected_files = dialog.selectedFiles()
if not selected_files:
return
file_path = selected_files[0]
with open(file_path, "r", encoding="utf-8") as file:
content = file.read()
self.monaco_editor.set_text(content)
self._init_file_content = content
logger.info(f"Selected files: {selected_files}")
def set_save_button_enabled(self, enabled: bool):
"""
Set the save button enabled state.
"""
action = self.toolbar.components.get_action("save")
if action:
action.action.setEnabled(enabled)
def run_script(self):
print("Running script...")
script_id = str(uuid.uuid4())
self.current_script_id = script_id
script_text = self.monaco_editor.get_text()
script_text = f'bec._run_script("{script_id}", """{script_text}""")'
script_text = script_text.replace("\n", "\\n").replace("'", "\\'").strip()
if not script_text.endswith("\n"):
script_text += "\\n"
self.web_console.write(script_text)
if __name__ == "__main__":
import sys
from qtpy.QtWidgets import QApplication
app = QApplication(sys.argv)
script_interface = ScriptInterface()
script_interface.resize(800, 600)
script_interface.show()
sys.exit(app.exec_())

View File

@@ -19,10 +19,15 @@ from qtpy.QtWidgets import (
import bec_widgets
from bec_widgets.utils import UILoader
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.colors import apply_theme
from bec_widgets.utils.colors import apply_theme, set_theme
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.widget_io import WidgetHierarchy
from bec_widgets.widgets.containers.main_window.addons.hover_widget import HoverWidget
from bec_widgets.widgets.containers.main_window.addons.notification_center.notification_banner import (
BECNotificationBroker,
NotificationCentre,
NotificationIndicator,
)
from bec_widgets.widgets.containers.main_window.addons.scroll_label import ScrollLabel
from bec_widgets.widgets.containers.main_window.addons.web_links import BECWebLinksMixin
from bec_widgets.widgets.progress.scan_progressbar.scan_progressbar import ScanProgressBar
@@ -50,6 +55,14 @@ class BECMainWindow(BECWidget, QMainWindow):
self.app = QApplication.instance()
self.status_bar = self.statusBar()
self.setWindowTitle(window_title)
# Notification Centre overlay
self.notification_centre = NotificationCentre(parent=self) # Notification layer
self.notification_broker = BECNotificationBroker()
self._nc_margin = 16
self._position_notification_centre()
# Init ui
self._init_ui()
self._connect_to_theme_change()
@@ -58,6 +71,34 @@ class BECMainWindow(BECWidget, QMainWindow):
self.display_client_message, MessageEndpoints.client_info()
)
def setCentralWidget(self, widget: QWidget, qt_default: bool = False): # type: ignore[override]
"""
Reimplement QMainWindow.setCentralWidget so that the *main content*
widget always lives on the lower layer of the stacked layout that
hosts our notification overlays.
Args:
widget: The widget that should become the new central content.
qt_default: When *True* the call is forwarded to the base class so
that Qt behaves exactly as the original implementation (used
during __init__ when we first install ``self._full_content``).
"""
super().setCentralWidget(widget)
self.notification_centre.raise_()
self.statusBar().raise_()
def resizeEvent(self, event):
super().resizeEvent(event)
self._position_notification_centre()
def _position_notification_centre(self):
"""Keep the notification panel at a fixed margin top-right."""
if not hasattr(self, "notification_centre"):
return
margin = getattr(self, "_nc_margin", 16) # px
nc = self.notification_centre
nc.move(self.width() - nc.width() - margin, margin)
################################################################################
# MainWindow Elements Initialization
################################################################################
@@ -94,6 +135,26 @@ class BECMainWindow(BECWidget, QMainWindow):
# Add scan_progress bar with display logic
self._add_scan_progress_bar()
# Setup NotificationIndicator to bottom right of the status bar
self._add_notification_indicator()
################################################################################
# Notification indicator and Notification Centre helpers
def _add_notification_indicator(self):
"""
Add the notification indicator to the status bar and hook the signals.
"""
# Add the notification indicator to the status bar
self.notification_indicator = NotificationIndicator(self)
self.status_bar.addPermanentWidget(self.notification_indicator)
# Connect the notification broker to the indicator
self.notification_centre.counts_updated.connect(self.notification_indicator.update_counts)
self.notification_indicator.filter_changed.connect(self.notification_centre.apply_filter)
self.notification_indicator.show_all_requested.connect(self.notification_centre.show_all)
self.notification_indicator.hide_all_requested.connect(self.notification_centre.hide_all)
################################################################################
# Client message status bar widget helpers
@@ -379,12 +440,12 @@ class BECMainWindow(BECWidget, QMainWindow):
@SafeSlot(str)
def change_theme(self, theme: str):
"""
Change the theme of the application.
Change the theme of the application and propagate it to widgets.
Args:
theme(str): The theme to apply, either "light" or "dark".
theme(str): Either "light" or "dark".
"""
apply_theme(theme)
set_theme(theme) # emits theme_updated and applies palette globally
def event(self, event):
if event.type() == QEvent.Type.StatusTip:

View File

@@ -1,7 +1,6 @@
from typing import Literal
import qtmonaco
from qtpy.QtCore import Signal
from qtpy.QtWidgets import QApplication, QVBoxLayout, QWidget
from bec_widgets.utils.bec_widget import BECWidget
@@ -13,7 +12,6 @@ class MonacoWidget(BECWidget, QWidget):
A simple Monaco editor widget
"""
text_changed = Signal(str)
PLUGIN = True
ICON_NAME = "code"
USER_ACCESS = [
@@ -38,7 +36,6 @@ class MonacoWidget(BECWidget, QWidget):
self.editor = qtmonaco.Monaco(self)
layout.addWidget(self.editor)
self.setLayout(layout)
self.editor.text_changed.connect(self.text_changed.emit)
self.editor.initialized.connect(self.apply_theme)
def apply_theme(self, theme: str | None = None) -> None:

View File

@@ -6,12 +6,11 @@ import time
from bec_lib.logger import bec_logger
from louie.saferef import safe_ref
from qtpy.QtCore import QTimer, QUrl, Signal, qInstallMessageHandler
from qtpy.QtCore import QUrl, qInstallMessageHandler
from qtpy.QtWebEngineWidgets import QWebEnginePage, QWebEngineView
from qtpy.QtWidgets import QApplication, QVBoxLayout, QWidget
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.error_popups import SafeProperty
logger = bec_logger.logger
@@ -166,16 +165,11 @@ class WebConsole(BECWidget, QWidget):
A simple widget to display a website
"""
_js_callback = Signal(bool)
initialized = Signal()
PLUGIN = True
ICON_NAME = "terminal"
def __init__(self, parent=None, config=None, client=None, gui_id=None, **kwargs):
super().__init__(parent=parent, client=client, gui_id=gui_id, config=config, **kwargs)
self._startup_cmd = "bec --nogui"
self._is_initialized = False
_web_console_registry.register(self)
self._token = _web_console_registry._token
layout = QVBoxLayout()
@@ -187,48 +181,6 @@ class WebConsole(BECWidget, QWidget):
layout.addWidget(self.browser)
self.setLayout(layout)
self.page.setUrl(QUrl(f"http://localhost:{_web_console_registry._server_port}"))
self._startup_timer = QTimer()
self._startup_timer.setInterval(1000)
self._startup_timer.timeout.connect(self._check_page_ready)
self._startup_timer.start()
self._js_callback.connect(self._on_js_callback)
def _check_page_ready(self):
"""
Check if the page is ready and stop the timer if it is.
"""
if self.page.isLoading():
return
self.page.runJavaScript("window.term !== undefined", self._js_callback.emit)
def _on_js_callback(self, ready: bool):
"""
Callback for when the JavaScript is ready.
"""
if not ready:
return
self._is_initialized = True
self._startup_timer.stop()
if self._startup_cmd:
self.write(self._startup_cmd)
self.initialized.emit()
@SafeProperty(str)
def startup_cmd(self):
"""
Get the startup command for the web console.
"""
return self._startup_cmd
@startup_cmd.setter
def startup_cmd(self, cmd: str):
"""
Set the startup command for the web console.
"""
if not isinstance(cmd, str):
raise ValueError("Startup command must be a string.")
self._startup_cmd = cmd
def write(self, data: str, send_return: bool = True):
"""
@@ -261,19 +213,10 @@ class WebConsole(BECWidget, QWidget):
"document.querySelector('textarea.xterm-helper-textarea').dispatchEvent(new KeyboardEvent('keypress', {charCode: 3}))"
)
def set_readonly(self, readonly: bool):
"""
Set the web console to read-only mode.
"""
if not isinstance(readonly, bool):
raise ValueError("Readonly must be a boolean.")
self.setEnabled(not readonly)
def cleanup(self):
"""
Clean up the registry by removing any instances that are no longer valid.
"""
self._startup_timer.stop()
_web_console_registry.unregister(self)
super().cleanup()

View File

@@ -3,6 +3,7 @@ from __future__ import annotations
import math
from typing import TYPE_CHECKING
from bec_lib import bec_logger
from bec_qthemes import material_icon
from qtpy.QtCore import QEvent, Qt
from qtpy.QtGui import QColor
@@ -39,6 +40,9 @@ if TYPE_CHECKING:
from bec_widgets.widgets.plots.image.image import Image
logger = bec_logger.logger
class ROILockButton(QToolButton):
"""Keeps its icon and checked state in sync with a single ROI."""
@@ -447,6 +451,18 @@ class ROIPropertyTree(BECWidget, QWidget):
def cleanup(self):
self.cmap.close()
self.cmap.deleteLater()
if self.controller and hasattr(self.controller, "rois"):
for roi in self.controller.rois: # disconnect all signals from ROIs
try:
if isinstance(roi, RectangularROI):
roi.edgesChanged.disconnect()
else:
roi.centerChanged.disconnect()
roi.penChanged.disconnect()
roi.nameChanged.disconnect()
except (RuntimeError, TypeError) as e:
logger.error(f"Failed to disconnect roi qt signal: {e}")
super().cleanup()

View File

@@ -1,15 +1,25 @@
import os
import re
from functools import partial
from typing import Callable
import bec_lib
from bec_lib.callback_handler import EventType
from bec_lib.config_helper import ConfigHelper
from bec_lib.endpoints import MessageEndpoints
from bec_lib.logger import bec_logger
from bec_lib.messages import ConfigAction
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 QListWidget, QListWidgetItem, QVBoxLayout, QWidget
from qtpy.QtWidgets import (
QFileDialog,
QListWidget,
QListWidgetItem,
QToolButton,
QVBoxLayout,
QWidget,
)
from bec_widgets.cli.rpc.rpc_register import RPCRegister
from bec_widgets.utils.bec_widget import BECWidget
@@ -30,6 +40,7 @@ class DeviceBrowser(BECWidget, QWidget):
"""
devices_changed: Signal = Signal()
editing_enabled: Signal = Signal(bool)
device_update: Signal = Signal(str, dict)
PLUGIN = True
ICON_NAME = "lists"
@@ -47,7 +58,7 @@ class DeviceBrowser(BECWidget, QWidget):
self._config_helper = ConfigHelper(self.client.connector, self.client._service_name)
self._q_threadpool = QThreadPool()
self.ui = None
self.ini_ui()
self.init_ui()
self.dev_list: QListWidget = self.ui.device_list
self.dev_list.setVerticalScrollMode(QListWidget.ScrollMode.ScrollPerPixel)
self.proxy_device_update = SignalProxy(
@@ -56,14 +67,22 @@ class DeviceBrowser(BECWidget, QWidget):
self.bec_dispatcher.client.callbacks.register(
EventType.DEVICE_UPDATE, self.on_device_update
)
self.bec_dispatcher.client.callbacks.register(
EventType.SCAN_STATUS, self.scan_status_changed
)
self._default_config_dir = os.path.abspath(
os.path.join(os.path.dirname(bec_lib.__file__), "./configs/")
)
self.devices_changed.connect(self.update_device_list)
self.ui.add_button.clicked.connect(self._create_add_dialog)
self.ui.add_button.setIcon(material_icon("add", size=(20, 20), convert_to_pixmap=False))
self.init_warning_label()
self.init_tool_buttons()
self.init_device_list()
self.update_device_list()
def ini_ui(self) -> None:
def init_ui(self) -> None:
"""
Initialize the UI by loading the UI file and setting the layout.
"""
@@ -73,6 +92,27 @@ class DeviceBrowser(BECWidget, QWidget):
layout.addWidget(self.ui)
self.setLayout(layout)
def init_warning_label(self):
self.ui.scan_running_warning.setText("Warning: editing diabled while scan is running!")
self.ui.scan_running_warning.setStyleSheet(
"background-color: #fcba03; color: rgb(0, 0, 0);"
)
scan_status = self.bec_dispatcher.client.connector.get(MessageEndpoints.scan_status())
initial_status = scan_status.status if scan_status is not None else "closed"
self.set_editing_mode(initial_status not in ["open", "paused"])
def init_tool_buttons(self):
def _setup_button(button: QToolButton, icon: str, slot: Callable, tooltip: str = ""):
button.clicked.connect(slot)
button.setIcon(material_icon(icon, size=(20, 20), convert_to_pixmap=False))
button.setToolTip(tooltip)
_setup_button(self.ui.add_button, "add", self._create_add_dialog, "add new device")
_setup_button(self.ui.save_button, "save", self._save_to_file, "save config to file")
_setup_button(
self.ui.import_button, "input", self._load_from_file, "append/merge config from file"
)
def _create_add_dialog(self):
dialog = DeviceConfigDialog(parent=self, device=None, action="add")
dialog.open()
@@ -120,6 +160,7 @@ class DeviceBrowser(BECWidget, QWidget):
)
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)
@@ -130,6 +171,17 @@ class DeviceBrowser(BECWidget, QWidget):
self.dev_list.addItem(item)
self._device_items[device] = item
@SafeSlot(dict, dict)
def scan_status_changed(self, scan_info: dict, _: dict):
"""disable editing when scans are running and enable editing when they are finished"""
msg = ScanStatusMessage.model_validate(scan_info)
self.set_editing_mode(msg.status not in ["open", "paused"])
def set_editing_mode(self, enabled: bool):
self.ui.add_button.setEnabled(enabled)
self.ui.scan_running_warning.setHidden(enabled)
self.editing_enabled.emit(enabled)
@SafeSlot()
def reset_device_list(self) -> None:
self.init_device_list()
@@ -161,6 +213,22 @@ class DeviceBrowser(BECWidget, QWidget):
for device in self.dev:
self._device_items[device].setHidden(not self.regex.search(device))
@SafeSlot()
def _load_from_file(self):
file_path, _ = QFileDialog.getOpenFileName(
self, "Update config from file", self._default_config_dir, "Config files (*.yml *.yaml)"
)
if file_path:
self._config_helper.update_session_with_file(file_path)
@SafeSlot()
def _save_to_file(self):
file_path, _ = QFileDialog.getSaveFileName(
self, "Save config to file", self._default_config_dir, "Config files (*.yml *.yaml)"
)
if file_path:
self._config_helper.save_current_session(file_path)
if __name__ == "__main__": # pragma: no cover
import sys

View File

@@ -51,11 +51,35 @@
</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>
<item>
<widget class="QListWidget" name="device_list"/>
</item>

View File

@@ -24,8 +24,10 @@ class CommunicateConfigAction(QRunnable):
) -> None:
super().__init__()
self.config_helper = config_helper
if action in ["add", "update"] and config is None:
raise ValueError("Must supply config to add or update a device")
self.device = device
self.config = config
self.config = config or {}
self.action = action
self.signals = _CommSignals()
@@ -37,24 +39,35 @@ class CommunicateConfigAction(QRunnable):
raise ValueError(
"Must be updating a device or be supplied a name for a new device"
)
req_args = {
"action": self.action,
"config": {dev_name: self.config},
"wait_for_response": False,
}
timeout = (
self.config_helper.suggested_timeout_s(self.config)
if self.config is not None
else 20
)
RID = self.config_helper.send_config_request(**req_args)
logger.info("Waiting for config reply")
reply = self.config_helper.wait_for_config_reply(RID, timeout=timeout)
self.config_helper.handle_update_reply(reply, RID, timeout)
logger.info("Done updating config!")
if "deviceConfig" not in self.config or self.action in ["add", "remove"]:
self.process_simple_action(dev_name)
else:
# updating an existing device, but need to recreate it for this change
self.process_remove_readd(dev_name)
else:
raise ValueError(f"action {self.action} is not supported")
except Exception as e:
self.signals.error.emit(e)
else:
self.signals.done.emit()
def process_simple_action(self, dev_name: str, action: ConfigAction | None = None):
req_args = {
"action": action or self.action,
"config": {dev_name: self.config},
"wait_for_response": False,
}
timeout = (
self.config_helper.suggested_timeout_s(self.config) if self.config is not None else 20
)
RID = self.config_helper.send_config_request(**req_args)
logger.info("Waiting for config reply")
reply = self.config_helper.wait_for_config_reply(RID, timeout=timeout)
self.config_helper.handle_update_reply(reply, RID, timeout)
logger.info("Done updating config!")
def process_remove_readd(self, dev_name: str):
logger.info(f"Removing and readding device: {dev_name}")
self.process_simple_action(dev_name, "remove")
self.process_simple_action(dev_name, "add")
logger.info(f"Reinstated {dev_name} successfully!")

View File

@@ -68,7 +68,7 @@ class DeviceConfigDialog(BECWidget, QDialog):
self.client.connector, self.client._service_name
)
self._device = device
self._action = action
self._action: Literal["update", "add"] = action
self._q_threadpool = threadpool or QThreadPool()
self.setWindowTitle(f"Edit config for: {device}")
self._container = QStackedLayout()
@@ -168,6 +168,10 @@ class DeviceConfigDialog(BECWidget, QDialog):
diff = {
k: v for k, v in new_config.items() if self._initial_config.get(k) != new_config.get(k)
}
if self._initial_config.get("deviceConfig") in [{}, None] and new_config.get(
"deviceConfig"
) in [{}, None]:
diff.pop("deviceConfig", None)
if diff.get("deviceConfig") is not None:
# TODO: special cased in some parts of device manager but not others, should
# be removed in config update as with below issue
@@ -176,6 +180,11 @@ class DeviceConfigDialog(BECWidget, QDialog):
diff["deviceConfig"] = {
k: _try_literal_eval(str(v)) for k, v in diff["deviceConfig"].items() if k != ""
}
# Due to above issues, if deviceConfig changes we must remove and recreate the device - so we need the whole config
if "deviceConfig" in diff:
new_config["deviceConfig"] = diff["deviceConfig"]
return new_config
return diff
@SafeSlot(bool)
@@ -212,7 +221,7 @@ class DeviceConfigDialog(BECWidget, QDialog):
self._proc_device_config_change(updated_config)
def _proc_device_config_change(self, config: dict):
logger.info(f"Sending request to update device config: {config}")
logger.info(f"Sending request to {self._action} device config: {config}")
self._start_waiting_display()
communicate_update = CommunicateConfigAction(

View File

@@ -140,6 +140,11 @@ class DeviceItem(ExpandableGroupFrame):
self.adjustSize()
self.broadcast_size_hint.emit(self.sizeHint())
@SafeSlot(bool)
def set_editable(self, enabled: bool):
self.edit_button.setEnabled(enabled)
self.delete_button.setEnabled(enabled)
@SafeSlot(str, dict)
def config_update(self, action: ConfigAction, content: dict) -> None:
if self.device in content:

View File

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "bec_widgets"
version = "2.27.0"
version = "2.29.0"
description = "BEC Widgets"
requires-python = ">=3.10"
classifiers = [

View File

@@ -1,5 +1,6 @@
from unittest.mock import ANY, MagicMock
from unittest.mock import ANY, MagicMock, call
import pytest
from bec_lib.config_helper import ConfigHelper
from bec_widgets.widgets.services.device_browser.device_item.config_communicator import (
@@ -20,9 +21,41 @@ def test_must_have_a_name(qtbot):
qtbot.waitUntil(lambda: error_occurred, timeout=100)
@pytest.mark.parametrize(
["action", "config", "error"],
[("update", None, True), ("remove", None, False), ("add", {}, False)],
)
def test_action_config_match(action, config, error):
def init():
return CommunicateConfigAction(
ConfigHelper(MagicMock()), device="test", config=config, action=action
)
if error:
with pytest.raises(ValueError):
init()
else:
assert init()
def test_wait_for_reply_on_RID():
ch = MagicMock(spec=ConfigHelper)
ch.send_config_request.return_value = "abcde"
cca = CommunicateConfigAction(config_helper=ch, device="samx", config={}, action="update")
cca.run()
ch.wait_for_config_reply.assert_called_with("abcde", timeout=ANY)
ch.wait_for_config_reply.assert_called_once_with("abcde", timeout=ANY)
def test_remove_readd_with_device_config(qtbot):
ch = MagicMock(spec=ConfigHelper)
ch.send_config_request.return_value = "abcde"
cca = CommunicateConfigAction(
config_helper=ch, device="samx", config={"deviceConfig": {"arg": "val"}}, action="update"
)
cca.run()
ch.send_config_request.assert_has_calls(
[
call(action="remove", config=ANY, wait_for_response=False),
call(action="add", config=ANY, wait_for_response=False),
]
)

View File

@@ -120,6 +120,37 @@ def test_update_cycle(update_dialog, qtbot):
)
@pytest.mark.parametrize(
["changes", "result"],
[
({}, {}),
({"readOnly": True}, {"readOnly": True}),
({"readOnly": False}, {}),
({"readOnly": True, "description": "test"}, {"readOnly": True, "description": "test"}),
(
{"deviceConfig": {"param1": "'val1'"}},
{
"enabled": True,
"deviceClass": "TestDevice",
"deviceConfig": {"param1": "val1"},
"readoutPriority": "monitored",
"description": None,
"readOnly": False,
"softwareTrigger": False,
"deviceTags": set(),
"userParameter": {},
"name": "test_device",
},
),
({"deviceConfig": {}}, {}),
],
)
def test_update_with_modified_deviceconfig(update_dialog, changes, result):
for k, v in changes.items():
update_dialog._form.widget_dict[k].setValue(v)
assert update_dialog.updated_config() == result
def test_add_form_init_without_name(add_dialog, qtbot):
assert (name_widget := add_dialog._form.widget_dict.get("name")) is not None
assert isinstance(name_widget, StrFormItem)

View File

@@ -399,3 +399,35 @@ def test_new_roi_respects_global_lock(roi_tree, image_widget, qtbot):
assert not roi.movable
# Disable global lock again
roi_tree.lock_all_action.action.setChecked(False)
def test_cleanup_disconnect_signals(roi_tree, image_widget):
"""Test that cleanup disconnects ROI signals so further changes do not update the tree."""
# Add a rectangular ROI
roi = image_widget.add_roi(kind="rect", name="cleanup_test", pos=(10, 10), size=(20, 20))
item = roi_tree.roi_items[roi]
# Test that signals are connected before cleanup
pre_name = item.text(roi_tree.COL_ROI)
pre_coord = item.child(2).text(roi_tree.COL_PROPS)
# Change ROI properties to see updates
roi.label = "connected_name"
roi.setPos(30, 30)
# Verify that the tree item updated
assert item.text(roi_tree.COL_ROI) == "connected_name"
assert item.child(2).text(roi_tree.COL_PROPS) != pre_coord
# Perform cleanup to disconnect signals
roi_tree.cleanup()
# Store initial state
initial_name = item.text(roi_tree.COL_ROI)
initial_coord = item.child(2).text(roi_tree.COL_PROPS)
# Change ROI properties after cleanup
roi.label = "changed_name"
roi.setPos(50, 50)
# Verify that the tree item was not updated
assert item.text(roi_tree.COL_ROI) == initial_name
assert item.child(2).text(roi_tree.COL_PROPS) == initial_coord

View File

@@ -0,0 +1,340 @@
import pytest
from qtpy import QtCore, QtGui, QtWidgets
from bec_widgets.utils.error_popups import ErrorPopupUtility
from bec_widgets.widgets.containers.main_window.addons.notification_center.notification_banner import (
DARK_PALETTE,
LIGHT_PALETTE,
SEVERITY,
BECNotificationBroker,
NotificationCentre,
NotificationIndicator,
NotificationToast,
SeverityKind,
)
from .client_mocks import mocked_client
@pytest.fixture
def toast(qtbot):
"""Return a NotificationToast with a very short lifetime (50 ms) for fast tests."""
t = NotificationToast(
title="Test Title", body="Test Body", kind=SeverityKind.WARNING, lifetime_ms=50 # 0.05 s
)
qtbot.addWidget(t)
qtbot.waitExposed(t)
return t
def test_initial_state(toast):
"""Constructor should correctly propagate title / body / kind."""
assert toast.title == "Test Title"
assert toast.body == "Test Body"
assert toast.kind == SeverityKind.WARNING
# progress bar height fixed at 4 px
assert toast.progress.maximumHeight() == 4
def test_apply_theme_updates_colours(qtbot, toast):
"""apply_theme("light") should inject LIGHT palette colours into stylesheets."""
toast.apply_theme("light")
assert LIGHT_PALETTE["title"] in toast._title_lbl.styleSheet()
toast.apply_theme("dark")
assert DARK_PALETTE["title"] in toast._title_lbl.styleSheet()
def test_expired_signal(qtbot, toast):
"""Toast must emit expired once its lifetime finishes."""
with qtbot.waitSignal(toast.expired, timeout=1000):
pass
assert toast._expired
def test_closed_signal(qtbot, toast):
"""Calling close() must emit closed."""
with qtbot.waitSignal(toast.closed, timeout=1000):
toast.close()
def test_property_setters_update_ui(qtbot, toast):
"""Changing properties through setters should update both state and label text."""
# title
toast.title = "New Title"
assert toast.title == "New Title"
assert toast._title_lbl.text() == "New Title"
# body
toast.body = "New Body"
assert toast.body == "New Body"
assert toast._body_lbl.text() == "New Body"
# kind
toast.kind = SeverityKind.MINOR
assert toast.kind == SeverityKind.MINOR
expected_color = SEVERITY["minor"]["color"]
assert toast._accent_color.name() == QtGui.QColor(expected_color).name()
# traceback
new_tb = "Traceback: divide by zero"
toast.traceback = new_tb
assert toast.traceback == new_tb
assert toast.trace_view.toPlainText() == new_tb
def _make_enter_event(widget):
"""Utility: synthetic QEnterEvent centred on *widget*."""
centre = widget.rect().center()
local = QtCore.QPointF(centre)
scene = QtCore.QPointF(widget.mapTo(widget.window(), centre))
global_ = QtCore.QPointF(widget.mapToGlobal(centre))
return QtGui.QEnterEvent(local, scene, global_)
def test_time_label_toggle_absolute(qtbot, toast):
"""Hovering time-label switches between relative and absolute timestamp."""
rel_text = toast.time_lbl.text()
# Enter
QtWidgets.QApplication.sendEvent(toast.time_lbl, _make_enter_event(toast.time_lbl))
qtbot.wait(100)
abs_text = toast.time_lbl.text()
assert abs_text != rel_text and "-" in abs_text and ":" in abs_text
# Leave
QtWidgets.QApplication.sendEvent(toast.time_lbl, QtCore.QEvent(QtCore.QEvent.Leave))
qtbot.wait(100)
assert toast.time_lbl.text() != abs_text
def test_hover_pauses_and_resumes_expiry(qtbot):
"""Countdown must pause on hover and resume on leave."""
t = NotificationToast(title="Hover", body="x", kind=SeverityKind.INFO, lifetime_ms=200)
qtbot.addWidget(t)
qtbot.waitExposed(t)
qtbot.wait(50) # allow animation to begin
# Pause
QtWidgets.QApplication.sendEvent(t, _make_enter_event(t))
qtbot.wait(250) # longer than lifetime, but hover keeps it alive
assert not t._expired
# Resume
QtWidgets.QApplication.sendEvent(t, QtCore.QEvent(QtCore.QEvent.Leave))
with qtbot.waitSignal(t.expired, timeout=500):
pass
assert t._expired
def test_toast_paint_event(qtbot):
"""
Grabbing the widget as a pixmap forces paintEvent to execute.
The test passes if no exceptions occur and the resulting pixmap is valid.
"""
t = NotificationToast(title="Paint", body="Check", kind=SeverityKind.INFO, lifetime_ms=0)
qtbot.addWidget(t)
t.resize(420, 160)
t.show()
qtbot.waitExposed(t)
pix = t.grab()
assert not pix.isNull()
# ------------------------------------------------------------------------
# NotificationCentre tests
# ------------------------------------------------------------------------
@pytest.fixture
def centre(qtbot, mocked_client):
"""NotificationCentre embedded in a live parent widget kept alive for the test."""
parent = QtWidgets.QWidget()
parent.resize(600, 400)
ctr = NotificationCentre(parent=parent, fixed_width=300, margin=8)
broker = BECNotificationBroker(client=mocked_client)
layout = QtWidgets.QVBoxLayout(parent)
layout.setContentsMargins(0, 0, 0, 0)
layout.addWidget(ctr)
# Keep a Python reference so GC doesn't drop the parent (and cascadedelete centre)
ctr._test_parent_ref = parent # type: ignore[attr-defined]
qtbot.addWidget(parent)
qtbot.addWidget(ctr)
parent.show()
qtbot.waitExposed(parent)
yield ctr
broker.reset_singleton()
def _post(ctr: NotificationCentre, kind=SeverityKind.INFO, title="T", body="B"):
"""Convenience wrapper that posts a toast and returns it."""
return ctr.add_notification(title=title, body=body, kind=kind, lifetime_ms=0)
# ------------------------------------------------------------------------
# Tests
# ------------------------------------------------------------------------
def test_add_notification_emits_signal(qtbot, centre):
"""Adding a toast emits toast_added and makes centre visible."""
with qtbot.waitSignal(centre.toast_added, timeout=500) as sig:
toast = _post(centre, SeverityKind.INFO)
assert toast in centre.toasts
assert sig.args == [SeverityKind.INFO.value]
def test_counts_updated(qtbot, centre):
"""counts_updated reflects current per-kind counts."""
seen = []
centre.counts_updated.connect(lambda d: seen.append(d.copy()))
_post(centre, SeverityKind.INFO)
_post(centre, SeverityKind.WARNING)
qtbot.wait(100)
assert seen[-1][SeverityKind.INFO] == 1
assert seen[-1][SeverityKind.WARNING] == 1
centre.clear_all()
qtbot.wait(100)
assert seen[-1][SeverityKind.INFO] == 0
assert seen[-1][SeverityKind.WARNING] == 0
def test_filtering_hides_unrelated_toasts(centre):
info = _post(centre, SeverityKind.INFO)
warn = _post(centre, SeverityKind.WARNING)
centre.apply_filter({SeverityKind.INFO})
assert info.isVisible()
assert not warn.isVisible()
centre.apply_filter(None)
assert warn.isVisible()
def test_hide_show_all(qtbot, centre):
_post(centre, SeverityKind.MINOR)
centre.hide_all()
assert not centre.isVisible()
centre.show_all()
assert centre.isVisible()
assert all(t.isVisible() for t in centre.toasts)
def test_clear_all(qtbot, centre):
_post(centre, SeverityKind.INFO)
_post(centre, SeverityKind.WARNING)
# expect two toast_removed emissions
for _ in range(2):
qtbot.waitSignal(centre.toast_removed, timeout=500, raising=False)
centre.clear_all()
assert not centre.toasts
assert not centre.isVisible()
def test_theme_propagation(qtbot, centre):
toast = _post(centre, SeverityKind.INFO)
centre.apply_theme("light")
assert LIGHT_PALETTE["title"] in toast._title_lbl.styleSheet()
# ------------------------------------------------------------------------
# NotificationIndicator tests
# ------------------------------------------------------------------------
@pytest.fixture
def indicator(qtbot, centre):
"""Indicator wired to the same centre used in centre fixture."""
ind = NotificationIndicator()
qtbot.addWidget(ind)
# wire signals
centre.counts_updated.connect(ind.update_counts)
ind.filter_changed.connect(centre.apply_filter)
ind.show_all_requested.connect(centre.show_all)
ind.hide_all_requested.connect(centre.hide_all)
return ind
def _emit_counts(centre: NotificationCentre, info=0, warn=0, minor=0, major=0):
"""Helper to create dummy toasts and update counts."""
for _ in range(info):
_post(centre, SeverityKind.INFO)
for _ in range(warn):
_post(centre, SeverityKind.WARNING)
for _ in range(minor):
_post(centre, SeverityKind.MINOR)
for _ in range(major):
_post(centre, SeverityKind.MAJOR)
def test_indicator_updates_visibility(qtbot, centre, indicator):
"""Indicator shows/hides buttons based on counts."""
_emit_counts(centre, info=1)
qtbot.wait(50)
# "info" button visible, others hidden
assert indicator._btn[SeverityKind.INFO].isVisible()
assert not indicator._btn[SeverityKind.WARNING].isVisible()
# add warning toast → warning button appears
_emit_counts(centre, warn=1)
qtbot.wait(50)
assert indicator._btn[SeverityKind.WARNING].isVisible()
# clear all → indicator hides itself
centre.clear_all()
qtbot.wait(50)
assert not indicator.isVisible()
def test_indicator_filter_buttons(qtbot, centre, indicator):
"""Toggling buttons emits appropriate filter signals."""
# add two kinds so indicator is visible
_emit_counts(centre, info=1, warn=1)
qtbot.wait(200)
# click INFO button
with qtbot.waitSignal(indicator.filter_changed, timeout=500) as sig:
qtbot.mouseClick(indicator._btn[SeverityKind.INFO], QtCore.Qt.LeftButton)
assert sig.args[0] == {SeverityKind.INFO}
def test_broker_posts_notification(qtbot, centre, mocked_client):
"""post_notification should create a toast in the centre with correct data."""
broker = BECNotificationBroker(parent=None, client=mocked_client, centre=centre)
broker._err_util = ErrorPopupUtility()
msg = {
"alarm_type": "ValueError",
"msg": "test alarm",
"severity": 2, # MAJOR
"source": {"device": "samx", "source": "async_file_writer"},
}
broker.post_notification(msg, meta={})
qtbot.wait(200) # allow toast to be posted
# One toast should now exist
assert len(centre.toasts) == 1
toast = centre.toasts[0]
assert toast.title == "ValueError"
assert "Error occurred. See details." in toast.body
assert toast.kind == SeverityKind.MAJOR
assert toast._lifetime == 0