merge: bring in notifications-errors-01xgs8 for the system dock

This commit is contained in:
Claude committed 2026-10-09 21:34:44 +00:00
commit aa74b3befd
25 files changed
+4556 -12

No files matched your search

+8
View File
@@ -21,6 +21,7 @@ Requires ``bec_widgets[dev]`` (pytest-qt, fakeredis) and bec_lib's ``threads_che
# pylint: disable=redefined-outer-name,unused-argument,wrong-import-position
import os
import sys
# Keep ophyd on its dummy control layer, so importing it does not create a real EPICS CA context
os.environ.setdefault("OPHYD_CONTROL_LAYER", "dummy")
@@ -159,6 +160,13 @@ def notification_broker_reset(bec_dispatcher):
"""Tear down the app-wide notification broker before the dispatcher it is bound to."""
yield
BECNotificationBroker.reset_singleton()
# the reworked notification hub is only reset if some test imported it
hub_module = sys.modules.get(
"bec_widgets.widgets.containers.main_window.addons.notification_center."
"notification_ux_common"
)
if hub_module is not None:
hub_module.NotificationHub.reset_instance()
@pytest.fixture(autouse=True)
+41 -7
View File
@@ -111,7 +111,9 @@ def SafeProperty(
error_msg = traceback.format_exc()
if popup_error:
ErrorPopupUtility().custom_exception_hook(*sys.exc_info(), popup_error=True)
ErrorPopupUtility().custom_exception_hook(
*sys.exc_info(), popup_error=True, source=py_getter.__qualname__
)
logger.error(f"SafeProperty error in GETTER of '{prop_name}':\n{error_msg}")
return default
@@ -173,7 +175,7 @@ def SafeProperty(
if popup_error:
ErrorPopupUtility().custom_exception_hook(
*sys.exc_info(), popup_error=True
*sys.exc_info(), popup_error=True, source=setter_func.__qualname__
)
logger.error(f"SafeProperty error in SETTER of '{prop_name}':\n{error_msg}")
return
@@ -298,7 +300,9 @@ def SafeSlot(*slot_args, **slot_kwargs): # pylint: disable=invalid-name
slot_name = f"{method.__module__}.{method.__qualname__}"
error_msg = traceback.format_exc()
if call_params["popup_error"]:
ErrorPopupUtility().custom_exception_hook(*sys.exc_info(), popup_error=True)
ErrorPopupUtility().custom_exception_hook(
*sys.exc_info(), popup_error=True, source=method.__qualname__
)
logger.error(f"SafeSlot error in slot '{slot_name}':\n{error_msg}")
if call_params["raise_error"]:
raise
@@ -308,6 +312,26 @@ def SafeSlot(*slot_args, **slot_kwargs): # pylint: disable=invalid-name
return error_managed
def _notification_hub():
"""Return the notification hub if a window shows notifications, else None.
When the reworked notification UI is attached to a window, errors and warnings become
toasts with a history entry instead of blocking message boxes.
"""
# only look the module up: if nobody imported it, no window can show notifications, and
# importing it here would pull Qt Quick into every widget that reports an error
module = sys.modules.get(
"bec_widgets.widgets.containers.main_window.addons.notification_center."
"notification_ux_common"
)
if module is None:
return None
hub = module.NotificationHub._instance # pylint: disable=protected-access
if hub is None or not shiboken6.isValid(hub) or not hub.has_hosts:
return None
return hub
class WarningPopupUtility(QObject):
"""
Utility class to show warning popups in the application.
@@ -315,6 +339,11 @@ class WarningPopupUtility(QObject):
@SafeSlot(str, str, str, QWidget)
def show_warning_message(self, title, message, detailed_text, widget):
hub = _notification_hub()
if hub is not None:
source = type(widget).__name__ if widget is not None else ""
hub.notify(title, message, "warning", details=detailed_text, source=source)
return
msg = QMessageBox(widget)
msg.setIcon(QMessageBox.Warning)
msg.setWindowTitle(title)
@@ -415,12 +444,17 @@ class _ErrorPopupUtility(QObject):
def get_error_message(self, exctype, value, tb):
return "".join(traceback.format_exception(exctype, value, tb))
def custom_exception_hook(self, exctype, value, tb, popup_error=False):
def custom_exception_hook(self, exctype, value, tb, popup_error=False, source=None):
if popup_error or self.enable_error_popup:
title = "Method error" if popup_error else "Application Error"
hub = _notification_hub()
if hub is not None:
hub.report_exception(
title, self.get_error_message(exctype, value, tb), source=source or title
)
return
self.error_occurred.emit(
"Method error" if popup_error else "Application Error",
self.get_error_message(exctype, value, tb),
self.parent(),
title, self.get_error_message(exctype, value, tb), self.parent()
)
else:
sys.__excepthook__(exctype, value, tb) # Call the original excepthook
+26
View File
@@ -0,0 +1,26 @@
"""Shared plumbing for widgets that render their content with Qt Quick (QML).
The QML ports keep the usual ``BECWidget`` shell and public API and only swap the rendering
layer: a :class:`~qtpy.QtQuickWidgets.QQuickWidget` is embedded into the existing widget and
driven by a small Python backend object.
"""
from bec_widgets.utils.quick.host import (
QML_IMPORT_PATH,
MaterialIconProvider,
QmlTheme,
create_quick_widget,
quick_engine,
release_quick_widget,
)
from bec_widgets.utils.quick.list_model import DictListModel
__all__ = [
"QML_IMPORT_PATH",
"DictListModel",
"MaterialIconProvider",
"QmlTheme",
"create_quick_widget",
"quick_engine",
"release_quick_widget",
]
+233
View File
@@ -0,0 +1,233 @@
"""Hosting helpers for QML views embedded into BEC widgets.
All QML views share one :class:`QQmlEngine` per application. The engine carries the
application-wide ``theme`` context property (a :class:`QmlTheme` bridging ``app.theme`` from
``bec_qthemes``), an ``image://material/<name>`` provider for Material icons and the import path
of the shared ``BecUi`` control module. Per-widget state is handed to the view through initial
properties of the root item, typically a single ``backend`` QObject.
"""
from __future__ import annotations
from pathlib import Path
from urllib.parse import parse_qs
from bec_lib.logger import bec_logger
from bec_qthemes import material_icon
from qtpy.QtCore import Property, QObject, QSize, Qt, QUrl, Signal
from qtpy.QtGui import QColor, QPixmap
from qtpy.QtQml import QQmlEngine
from qtpy.QtQuick import QQuickImageProvider
from qtpy.QtQuickWidgets import QQuickWidget
from qtpy.QtWidgets import QApplication, QWidget
logger = bec_logger.logger
QML_IMPORT_PATH = Path(__file__).parent / "qml"
_DARK_FALLBACK = {
"PRIMARY": "#3b82f6",
"ON_PRIMARY": "#ffffff",
"BG": "#0f1115",
"CARD_BG": "#171a21",
"FIELD_BG": "#10131a",
"BORDER": "#2a2f3a",
"FG": "#e8ebf1",
"ACCENT_DEFAULT": "#8ab4f7",
"ACCENT_HIGHLIGHT": "#B53565",
"ACCENT_WARNING": "#EAC435",
"ACCENT_EMERGENCY": "#CC181E",
"ACCENT_SUCCESS": "#2CA58D",
}
def _blend(base: QColor, top: QColor, amount: float) -> QColor:
"""Return ``top`` mixed into ``base`` by ``amount`` (0 = base, 1 = top)."""
return QColor.fromRgbF(
base.redF() + (top.redF() - base.redF()) * amount,
base.greenF() + (top.greenF() - base.greenF()) * amount,
base.blueF() + (top.blueF() - base.blueF()) * amount,
1.0,
)
class ThemeTokens:
"""Resolved colour tokens of the active BEC theme, shared by the QML and QWidget ports."""
def __init__(self):
app = QApplication.instance()
theme = getattr(app, "theme", None) if app is not None else None
name = getattr(theme, "theme", "dark") if theme is not None else "dark"
def color(key: str) -> QColor:
fallback = _DARK_FALLBACK.get(key, "#808080")
if theme is None:
return QColor(fallback)
return QColor(theme.color(key, fallback))
self.name: str = str(name)
self.bg = color("BG")
self.card = color("CARD_BG")
self.field = color("FIELD_BG")
self.border = color("BORDER")
self.fg = color("FG")
self.primary = color("PRIMARY")
self.on_primary = color("ON_PRIMARY")
self.accent = color("ACCENT_DEFAULT")
self.highlight = color("ACCENT_HIGHLIGHT")
self.warning = color("ACCENT_WARNING")
self.danger = color("ACCENT_EMERGENCY")
self.success = color("ACCENT_SUCCESS")
self.dark = self.bg.lightnessF() < 0.5
self.fg_muted = _blend(self.card, self.fg, 0.62)
self.fg_subtle = _blend(self.card, self.fg, 0.4)
self.hover = _blend(self.card, self.fg, 0.06)
self.pressed = _blend(self.card, self.fg, 0.12)
self.track = _blend(self.card, self.fg, 0.1)
def soft(self, color: QColor, amount: float = 0.16) -> QColor:
"""Tint ``color`` onto the card background, e.g. for chips and banners."""
return _blend(self.card, color, amount)
class QmlTheme(QObject):
"""Expose the BEC theme colours to QML as ``theme.<token>`` and follow theme changes."""
changed = Signal()
def __init__(self, parent: QObject | None = None):
super().__init__(parent)
self._tokens = ThemeTokens()
self._connected_theme = None
self._connect_theme()
def _connect_theme(self) -> None:
app = QApplication.instance()
theme = getattr(app, "theme", None) if app is not None else None
if theme is None or theme is self._connected_theme:
return
theme.theme_changed.connect(self.refresh)
self._connected_theme = theme
def refresh(self, *_args) -> None:
"""Re-read the tokens from the application theme."""
self._connect_theme()
self._tokens = ThemeTokens()
self.changed.emit()
# pylint: disable=missing-function-docstring
def _get(self, name: str) -> QColor:
return getattr(self._tokens, name)
name = Property(str, lambda self: self._tokens.name, notify=changed)
dark = Property(bool, lambda self: self._tokens.dark, notify=changed)
bg = Property(QColor, lambda self: self._get("bg"), notify=changed)
card = Property(QColor, lambda self: self._get("card"), notify=changed)
field = Property(QColor, lambda self: self._get("field"), notify=changed)
border = Property(QColor, lambda self: self._get("border"), notify=changed)
fg = Property(QColor, lambda self: self._get("fg"), notify=changed)
fgMuted = Property(QColor, lambda self: self._get("fg_muted"), notify=changed)
fgSubtle = Property(QColor, lambda self: self._get("fg_subtle"), notify=changed)
hover = Property(QColor, lambda self: self._get("hover"), notify=changed)
pressed = Property(QColor, lambda self: self._get("pressed"), notify=changed)
track = Property(QColor, lambda self: self._get("track"), notify=changed)
primary = Property(QColor, lambda self: self._get("primary"), notify=changed)
onPrimary = Property(QColor, lambda self: self._get("on_primary"), notify=changed)
accent = Property(QColor, lambda self: self._get("accent"), notify=changed)
highlight = Property(QColor, lambda self: self._get("highlight"), notify=changed)
warning = Property(QColor, lambda self: self._get("warning"), notify=changed)
danger = Property(QColor, lambda self: self._get("danger"), notify=changed)
success = Property(QColor, lambda self: self._get("success"), notify=changed)
class MaterialIconProvider(QQuickImageProvider):
"""Serve Material icons to QML as ``image://material/<name>?color=%23rrggbb&filled=1``."""
def __init__(self):
super().__init__(QQuickImageProvider.ImageType.Pixmap)
def requestPixmap(self, icon_id: str, size: QSize, requested_size: QSize) -> QPixmap:
name, _, query = icon_id.partition("?")
options = parse_qs(query)
color = options.get("color", [None])[0]
filled = options.get("filled", ["0"])[0] in ("1", "true")
target = requested_size if requested_size.isValid() else QSize(24, 24)
if target.width() <= 0 or target.height() <= 0:
target = QSize(24, 24)
pixmap = material_icon(name, size=target, color=color, filled=filled)
if size is not None:
size.setWidth(pixmap.width())
size.setHeight(pixmap.height())
return pixmap
_ENGINE: QQmlEngine | None = None
_THEME: QmlTheme | None = None
def quick_engine() -> QQmlEngine:
"""Return the application-wide QML engine used by all BEC QML views."""
global _ENGINE, _THEME # pylint: disable=global-statement
if _ENGINE is not None:
if _THEME._connected_theme is None: # pylint: disable=protected-access
# the theme was applied after the engine was created
_THEME.refresh()
return _ENGINE
app = QApplication.instance()
_ENGINE = QQmlEngine(app)
_ENGINE.addImportPath(str(QML_IMPORT_PATH))
_ENGINE.addImageProvider("material", MaterialIconProvider())
_THEME = QmlTheme(_ENGINE)
_ENGINE.rootContext().setContextProperty("theme", _THEME)
_ENGINE.warnings.connect(
lambda warnings: [logger.warning(f"QML: {w.toString()}") for w in warnings]
)
if app is not None:
app.aboutToQuit.connect(_drop_engine)
return _ENGINE
def _drop_engine() -> None:
global _ENGINE, _THEME # pylint: disable=global-statement
_ENGINE = None
_THEME = None
def create_quick_widget(
parent: QWidget, qml_file: str | Path, properties: dict | None = None
) -> QQuickWidget:
"""Create a QQuickWidget showing ``qml_file`` on the shared engine.
Args:
parent(QWidget): Parent widget, usually the BEC widget shell.
qml_file(str | Path): Path of the root QML file.
properties(dict | None): Initial properties of the root item, e.g. ``{"backend": obj}``.
Returns:
QQuickWidget: The view, already loaded. Load errors are logged.
"""
view = QQuickWidget(quick_engine(), parent)
view.setResizeMode(QQuickWidget.ResizeMode.SizeRootObjectToView)
view.setAttribute(Qt.WidgetAttribute.WA_AlwaysStackOnTop, False)
view.setClearColor(parent.palette().window().color())
if properties:
view.setInitialProperties(properties)
view.setSource(QUrl.fromLocalFile(str(qml_file)))
for error in view.errors():
logger.error(f"QML error in {qml_file}: {error.toString()}")
return view
def release_quick_widget(view: QQuickWidget | None) -> None:
"""Unload the QML scene of ``view`` before its backend objects are destroyed.
Without this, bindings of a still-loaded scene are re-evaluated against already deleted
backends during teardown and log ``TypeError`` warnings.
"""
if view is None:
return
try:
view.setSource(QUrl())
except RuntimeError:
# the C++ object is already gone
return
+83
View File
@@ -0,0 +1,83 @@
"""A small list model that feeds lists of dictionaries to QML views."""
from __future__ import annotations
from typing import Any, Sequence
from qtpy.QtCore import Property, QAbstractListModel, QByteArray, QModelIndex, Qt, Signal
class DictListModel(QAbstractListModel):
"""List model whose rows are plain dictionaries with a fixed set of keys.
Each key becomes a QML role. :meth:`set_items` updates rows in place when the row count is
unchanged, so QML delegates are kept alive and ``Behavior`` animations run on value changes
instead of the delegates being rebuilt.
Args:
roles(Sequence[str]): Keys of the row dictionaries exposed as roles.
parent: Parent QObject.
"""
count_changed = Signal()
def __init__(self, roles: Sequence[str], parent=None):
super().__init__(parent)
self._keys = list(roles)
self._roles = {Qt.ItemDataRole.UserRole + 1 + i: key for i, key in enumerate(self._keys)}
self._items: list[dict[str, Any]] = []
# pylint: disable=invalid-name
def roleNames(self) -> dict[int, QByteArray]:
return {role: QByteArray(key.encode()) for role, key in self._roles.items()}
def rowCount(self, parent: QModelIndex = QModelIndex()) -> int:
if parent.isValid():
return 0
return len(self._items)
def data(self, index: QModelIndex, role: int = Qt.ItemDataRole.DisplayRole):
if not index.isValid() or not 0 <= index.row() < len(self._items):
return None
key = self._roles.get(role)
if key is None:
return None
return self._items[index.row()].get(key)
@property
def items(self) -> list[dict[str, Any]]:
"""Copy of the current rows."""
return [dict(item) for item in self._items]
def set_items(self, items: Sequence[dict[str, Any]]) -> None:
"""Replace the rows, reusing existing rows where possible.
Args:
items(Sequence[dict]): New rows.
"""
items = [dict(item) for item in items]
old_count = len(self._items)
new_count = len(items)
common = min(old_count, new_count)
for row in range(common):
changed = [
role
for role, key in self._roles.items()
if self._items[row].get(key) != items[row].get(key)
]
self._items[row] = items[row]
if changed:
index = self.index(row, 0)
self.dataChanged.emit(index, index, changed)
if new_count > old_count:
self.beginInsertRows(QModelIndex(), old_count, new_count - 1)
self._items.extend(items[old_count:])
self.endInsertRows()
elif new_count < old_count:
self.beginRemoveRows(QModelIndex(), new_count, old_count - 1)
del self._items[new_count:]
self.endRemoveRows()
if new_count != old_count:
self.count_changed.emit()
count = Property(int, lambda self: len(self._items), notify=count_changed)
@@ -0,0 +1,56 @@
import QtQuick
import QtQuick.Layouts
// Rounded surface with an optional title row. Children go into the content column.
Rectangle {
id: root
property string title: ""
property string subtitle: ""
default property alias content: column.data
property alias headerExtras: extras.data
property int padding: 12
property alias spacing: column.spacing
// Let the content column take the remaining height, e.g. for a list that fills the card.
property bool fillContent: false
color: theme.card
radius: 10
border.color: theme.border
border.width: 1
implicitHeight: outer.implicitHeight + 2 * padding
implicitWidth: outer.implicitWidth + 2 * padding
ColumnLayout {
id: outer
anchors.fill: parent
anchors.margins: root.padding
spacing: 10
RowLayout {
visible: root.title !== "" || extras.children.length > 0
Layout.fillWidth: true
spacing: 8
Text {
text: root.title
color: theme.fg
font.pixelSize: 13
font.weight: Font.DemiBold
}
Text {
text: root.subtitle
visible: text !== ""
color: theme.fgSubtle
font.pixelSize: 12
Layout.fillWidth: true
elide: Text.ElideRight
}
Item { Layout.fillWidth: root.subtitle === "" }
Row { id: extras; spacing: 4 }
}
ColumnLayout {
id: column
Layout.fillWidth: true
Layout.fillHeight: root.fillContent
spacing: 8
}
}
}
@@ -0,0 +1,12 @@
import QtQuick
// Background used by all input controls: field colour, border, focus and error states.
Rectangle {
property bool focused: false
property bool invalid: false
property bool hovered: false
radius: 6
color: theme.field
border.width: focused || invalid ? 2 : 1
border.color: invalid ? theme.danger : focused ? theme.primary : hovered ? theme.fgSubtle : theme.border
}
@@ -0,0 +1,25 @@
import QtQuick
// Material icon served by the "material" image provider, tinted with the given colour.
Item {
id: root
property string name: ""
property color color: theme.fg
property bool filled: false
property int size: 18
implicitWidth: size
implicitHeight: size
Image {
anchors.centerIn: parent
width: root.size
height: root.size
sourceSize.width: root.size * 2
sourceSize.height: root.size * 2
smooth: true
fillMode: Image.PreserveAspectFit
source: root.name === "" ? "" : "image://material/" + root.name + "?color="
+ encodeURIComponent(root.color.toString()) + (root.filled ? "&filled=1" : "")
}
}
@@ -0,0 +1,36 @@
import QtQuick
import QtQuick.Controls.Basic
// Flat, square icon button with hover and pressed feedback and a tooltip.
AbstractButton {
id: root
property string iconName: ""
property color iconColor: theme.fgMuted
property string tip: ""
property int iconSize: 18
implicitWidth: 30
implicitHeight: 30
hoverEnabled: true
focusPolicy: Qt.TabFocus
opacity: enabled ? 1.0 : 0.4
Accessible.name: tip
background: Rectangle {
radius: 6
color: root.down ? theme.pressed : (root.hovered || root.checked ? theme.hover : "transparent")
border.width: root.visualFocus ? 1 : 0
border.color: theme.primary
}
contentItem: Item {
Icon {
anchors.centerIn: parent
name: root.iconName
size: root.iconSize
color: root.checked ? theme.primary : root.iconColor
}
}
ToolTip.visible: tip !== "" && hovered
ToolTip.delay: 500
ToolTip.text: tip
}
@@ -0,0 +1,38 @@
import QtQuick
import QtQuick.Controls.Basic
// Single-line text input with optional unit suffix and error state.
TextField {
id: root
property bool invalid: false
property string suffix: ""
implicitHeight: 32
implicitWidth: 120
color: theme.fg
placeholderTextColor: theme.fgSubtle
selectionColor: theme.primary
selectedTextColor: theme.onPrimary
font.pixelSize: 13
leftPadding: 9
rightPadding: suffixLabel.visible ? suffixLabel.implicitWidth + 14 : 9
selectByMouse: true
hoverEnabled: true
opacity: enabled ? 1.0 : 0.5
background: FieldFrame {
focused: root.activeFocus
invalid: root.invalid
hovered: root.hovered
}
Text {
id: suffixLabel
visible: root.suffix !== ""
text: root.suffix
color: theme.fgSubtle
font.pixelSize: 12
anchors.right: parent.right
anchors.rightMargin: 9
anchors.verticalCenter: parent.verticalCenter
}
}
@@ -0,0 +1,88 @@
import QtQuick
import QtQuick.Controls.Basic
// Combo box styled like the input fields. With editable: true it filters while typing.
ComboBox {
id: root
property bool invalid: false
property string placeholder: ""
implicitHeight: 32
implicitWidth: 140
font.pixelSize: 13
hoverEnabled: true
opacity: enabled ? 1.0 : 0.5
background: FieldFrame {
focused: root.activeFocus || root.popup.visible
invalid: root.invalid
hovered: root.hovered
}
contentItem: TextField {
leftPadding: 9
rightPadding: 4
text: root.editable ? root.editText : root.displayText
readOnly: !root.editable
placeholderText: root.placeholder
placeholderTextColor: theme.fgSubtle
enabled: root.enabled
color: theme.fg
selectionColor: theme.primary
selectedTextColor: theme.onPrimary
font: root.font
verticalAlignment: Text.AlignVCenter
background: null
selectByMouse: true
// forward clicks to the combo when it is not editable
TapHandler { enabled: !root.editable; onTapped: root.popup.visible ? root.popup.close() : root.popup.open() }
}
indicator: Icon {
x: root.width - width - 6
y: (root.height - height) / 2
name: "expand_more"
size: 18
color: theme.fgMuted
}
delegate: ItemDelegate {
id: option
required property int index
required property var modelData
width: ListView.view ? ListView.view.width : root.width
height: 30
highlighted: root.highlightedIndex === index
contentItem: Text {
text: option.modelData === "" ? "None" : option.modelData
color: option.modelData === "" ? theme.fgSubtle : theme.fg
font.pixelSize: 13
elide: Text.ElideRight
verticalAlignment: Text.AlignVCenter
}
background: Rectangle {
color: option.highlighted ? theme.hover : "transparent"
Rectangle {
width: 3; height: parent.height - 10; radius: 1.5
anchors.verticalCenter: parent.verticalCenter
color: theme.primary
visible: root.currentIndex === option.index
}
}
}
popup: Popup {
y: root.height + 2
width: root.width
implicitHeight: Math.min(contentItem.implicitHeight + 8, 280)
padding: 4
contentItem: ListView {
clip: true
implicitHeight: contentHeight
model: root.popup.visible ? root.delegateModel : null
currentIndex: root.highlightedIndex
ScrollIndicator.vertical: ScrollIndicator {}
}
background: Rectangle {
color: theme.card
border.color: theme.border
radius: 8
}
}
}
@@ -0,0 +1,40 @@
import QtQuick
// Small rounded status chip with a coloured dot. Set pulse for live states such as "Moving".
Rectangle {
id: root
property string text: ""
property color tone: theme.fgMuted
property bool pulse: false
implicitHeight: 22
implicitWidth: label.implicitWidth + 26
radius: height / 2
color: Qt.rgba(tone.r, tone.g, tone.b, 0.16)
Rectangle {
id: dot
width: 7; height: 7; radius: 3.5
color: root.tone
anchors.left: parent.left
anchors.leftMargin: 9
anchors.verticalCenter: parent.verticalCenter
SequentialAnimation on opacity {
running: root.pulse
loops: Animation.Infinite
alwaysRunToEnd: true
NumberAnimation { to: 0.25; duration: 550; easing.type: Easing.InOutQuad }
NumberAnimation { to: 1.0; duration: 550; easing.type: Easing.InOutQuad }
}
}
Text {
id: label
anchors.left: dot.right
anchors.leftMargin: 6
anchors.verticalCenter: parent.verticalCenter
text: root.text
color: root.tone
font.pixelSize: 12
font.weight: Font.DemiBold
}
}
@@ -0,0 +1,34 @@
import QtQuick
import QtQuick.Controls.Basic
// Toggle switch for boolean parameters.
Switch {
id: root
implicitHeight: 32
hoverEnabled: true
padding: 0
indicator: Rectangle {
implicitWidth: 38
implicitHeight: 22
x: root.leftPadding
y: (root.height - height) / 2
radius: 11
color: root.checked ? theme.primary : theme.track
border.color: root.visualFocus ? theme.primary : (root.checked ? theme.primary : theme.border)
Behavior on color { ColorAnimation { duration: 120 } }
Rectangle {
x: root.checked ? parent.width - width - 3 : 3
y: 3
width: 16; height: 16; radius: 8
color: root.checked ? theme.onPrimary : theme.fgMuted
Behavior on x { NumberAnimation { duration: 120; easing.type: Easing.OutCubic } }
}
}
contentItem: Text {
leftPadding: root.indicator.width + 8
text: root.text
color: theme.fg
font.pixelSize: 13
verticalAlignment: Text.AlignVCenter
}
}
@@ -0,0 +1,65 @@
import QtQuick
import QtQuick.Controls.Basic
import QtQuick.Layouts
// Labelled button. variant: "primary", "danger", "success", "neutral" or "ghost".
AbstractButton {
id: root
property string variant: "neutral"
property string iconName: ""
property string tip: ""
property bool busy: false
readonly property color baseColor: variant === "primary" ? theme.primary
: variant === "danger" ? theme.danger
: variant === "success" ? theme.success
: "transparent"
readonly property bool solid: variant === "primary" || variant === "danger" || variant === "success"
readonly property color labelColor: solid ? "white" : theme.fg
implicitHeight: 32
implicitWidth: Math.max(64, row.implicitWidth + 24)
hoverEnabled: true
opacity: enabled ? 1.0 : 0.45
Accessible.name: text
background: Rectangle {
radius: 6
color: root.solid ? (root.down ? Qt.darker(root.baseColor, 1.25) : root.hovered ? Qt.lighter(root.baseColor, 1.1) : root.baseColor)
: (root.down ? theme.pressed : root.hovered ? theme.hover : "transparent")
border.width: root.variant === "neutral" || root.visualFocus ? 1 : 0
border.color: root.visualFocus ? theme.primary : theme.border
Behavior on color { ColorAnimation { duration: 90 } }
}
contentItem: Item {
implicitWidth: row.implicitWidth
implicitHeight: row.implicitHeight
RowLayout {
id: row
anchors.centerIn: parent
spacing: 6
BusyIndicator {
visible: root.busy
running: root.busy
Layout.preferredWidth: 16
Layout.preferredHeight: 16
}
Icon {
visible: root.iconName !== "" && !root.busy
name: root.iconName
color: root.labelColor
size: 16
}
Text {
text: root.text
color: root.labelColor
font.pixelSize: 13
font.weight: root.solid ? Font.DemiBold : Font.Medium
elide: Text.ElideRight
}
}
}
ToolTip.visible: tip !== "" && hovered
ToolTip.delay: 500
ToolTip.text: tip
}
+10
View File
@@ -0,0 +1,10 @@
module BecUi
Icon 1.0 Icon.qml
IconButton 1.0 IconButton.qml
TextButton 1.0 TextButton.qml
StatusPill 1.0 StatusPill.qml
Card 1.0 Card.qml
FieldFrame 1.0 FieldFrame.qml
InputField 1.0 InputField.qml
SelectField 1.0 SelectField.qml
SwitchField 1.0 SwitchField.qml
+420
View File
@@ -0,0 +1,420 @@
"""Small themed QWidget controls used by the QWidget twins of the QML widget ports.
They mirror the ``BecUi`` QML controls in ``bec_widgets/utils/quick/qml`` so that the QML and
QWidget versions of a widget look and behave the same and can be compared side by side. Colours
come from the same :class:`~bec_widgets.utils.quick.host.ThemeTokens`.
"""
from __future__ import annotations
from bec_qthemes import material_icon
from qtpy.QtCore import Property, QEasingCurve, QPropertyAnimation, QRectF, QSize, Qt
from qtpy.QtGui import QColor, QFont, QPainter
from qtpy.QtWidgets import (
QFrame,
QHBoxLayout,
QLabel,
QLineEdit,
QPushButton,
QSizePolicy,
QToolButton,
QVBoxLayout,
QWidget,
)
from bec_widgets.utils.quick.host import ThemeTokens
def rgba(color: QColor, alpha: float) -> str:
"""Return a QSS ``rgba()`` string for ``color`` with alpha in 0..1."""
return f"rgba({color.red()}, {color.green()}, {color.blue()}, {int(alpha * 255)})"
def refresh_kit_theme(root: QWidget) -> ThemeTokens:
"""Re-theme every kit control below ``root`` and return the current tokens."""
tokens = ThemeTokens()
for widget in [root, *root.findChildren(QWidget)]:
refresh = getattr(widget, "refresh_theme", None)
if callable(refresh) and widget is not root:
refresh(tokens)
return tokens
def field_qss(tokens: ThemeTokens) -> str:
"""Stylesheet for inputs inside kit cards; ``invalid=true`` marks a field with an error."""
return f"""
QLineEdit, QComboBox, QSpinBox, QDoubleSpinBox {{
background: {tokens.field.name()};
color: {tokens.fg.name()};
border: 1px solid {tokens.border.name()};
border-radius: 6px;
padding: 0px 8px;
min-height: 30px;
font-size: 13px;
selection-background-color: {tokens.primary.name()};
}}
QLineEdit:hover, QComboBox:hover, QSpinBox:hover, QDoubleSpinBox:hover {{
border-color: {tokens.fg_subtle.name()};
}}
QLineEdit:focus, QComboBox:focus, QSpinBox:focus, QDoubleSpinBox:focus {{
border: 2px solid {tokens.primary.name()};
padding: 0px 7px;
}}
QLineEdit[invalid="true"], QComboBox[invalid="true"] {{
border: 2px solid {tokens.danger.name()};
padding: 0px 7px;
}}
QLineEdit:disabled, QComboBox:disabled {{ color: {tokens.fg_subtle.name()}; }}
QComboBox::drop-down {{ border: none; width: 22px; }}
"""
def set_invalid(widget: QWidget, invalid: bool) -> None:
"""Toggle the ``invalid`` style of a field styled with :func:`field_qss`."""
if bool(widget.property("invalid")) == invalid:
return
widget.setProperty("invalid", invalid)
widget.style().unpolish(widget)
widget.style().polish(widget)
class IconButton(QToolButton):
"""Flat square icon button with hover feedback, matching ``BecUi.IconButton``."""
def __init__(self, icon_name: str, tip: str = "", parent: QWidget | None = None, size=18):
super().__init__(parent)
self._icon_name = icon_name
self._icon_size = size
self.setToolTip(tip)
self.setAccessibleName(tip)
self.setAutoRaise(True)
self.setFixedSize(30, 30)
self.setIconSize(QSize(size, size))
self.setCursor(Qt.CursorShape.PointingHandCursor)
self.refresh_theme(ThemeTokens())
def set_icon_name(self, icon_name: str) -> None:
"""Change the Material icon."""
self._icon_name = icon_name
self.refresh_theme(ThemeTokens())
def refresh_theme(self, tokens: ThemeTokens) -> None:
"""Apply the theme tokens."""
color = tokens.primary if self.isChecked() else tokens.fg_muted
self.setIcon(
material_icon(self._icon_name, size=(48, 48), color=color, convert_to_pixmap=False)
)
self.setStyleSheet(
f"QToolButton {{ border: none; border-radius: 6px; background: transparent; }}"
f"QToolButton:hover, QToolButton:checked {{ background: {tokens.hover.name()}; }}"
f"QToolButton:pressed {{ background: {tokens.pressed.name()}; }}"
)
class TextButton(QPushButton):
"""Labelled button; variant is ``primary``, ``danger``, ``success``, ``neutral`` or ``ghost``."""
def __init__(
self,
text: str,
variant: str = "neutral",
icon_name: str = "",
parent: QWidget | None = None,
):
super().__init__(text, parent)
self._variant = variant
self._icon_name = icon_name
self.setMinimumHeight(32)
self.setCursor(Qt.CursorShape.PointingHandCursor)
self.refresh_theme(ThemeTokens())
def set_variant(self, variant: str) -> None:
"""Switch the colour variant."""
self._variant = variant
self.refresh_theme(ThemeTokens())
def refresh_theme(self, tokens: ThemeTokens) -> None:
"""Apply the theme tokens."""
base = {"primary": tokens.primary, "danger": tokens.danger, "success": tokens.success}.get(
self._variant
)
if base is not None:
label = QColor("white")
qss = (
f"QPushButton {{ background: {base.name()}; color: white; border: none;"
f" border-radius: 6px; padding: 0 12px; font-size: 13px; font-weight: 600; }}"
f"QPushButton:hover {{ background: {base.lighter(110).name()}; }}"
f"QPushButton:pressed {{ background: {base.darker(125).name()}; }}"
f"QPushButton:disabled {{ background: {rgba(base, 0.45)};"
f" color: {rgba(label, 0.7)}; }}"
)
else:
label = tokens.fg
border = tokens.border.name() if self._variant == "neutral" else "transparent"
qss = (
f"QPushButton {{ background: transparent; color: {tokens.fg.name()};"
f" border: 1px solid {border}; border-radius: 6px; padding: 0 12px;"
f" font-size: 13px; font-weight: 500; }}"
f"QPushButton:hover {{ background: {tokens.hover.name()}; }}"
f"QPushButton:pressed {{ background: {tokens.pressed.name()}; }}"
f"QPushButton:disabled {{ color: {tokens.fg_subtle.name()}; }}"
)
self.setStyleSheet(qss)
if self._icon_name:
self.setIcon(
material_icon(self._icon_name, size=(32, 32), color=label, convert_to_pixmap=False)
)
self.setIconSize(QSize(16, 16))
class StatusPill(QWidget):
"""Rounded status chip with a coloured dot, matching ``BecUi.StatusPill``."""
def __init__(self, parent: QWidget | None = None):
super().__init__(parent)
self._text = ""
self._tone = QColor("#808080")
self._dot_opacity = 1.0
self._pulse = QPropertyAnimation(self, b"dot_opacity", self)
self._pulse.setDuration(1100)
self._pulse.setKeyValueAt(0.0, 1.0)
self._pulse.setKeyValueAt(0.5, 0.25)
self._pulse.setKeyValueAt(1.0, 1.0)
self._pulse.setEasingCurve(QEasingCurve.Type.InOutQuad)
self._pulse.setLoopCount(-1)
self.setSizePolicy(QSizePolicy.Policy.Fixed, QSizePolicy.Policy.Fixed)
font = self.font()
font.setPixelSize(12)
font.setWeight(QFont.Weight.DemiBold)
self.setFont(font)
def set_status(self, text: str, tone: QColor, pulse: bool = False) -> None:
"""Set label, colour and whether the dot pulses."""
self._text = text
self._tone = QColor(tone)
if pulse and self._pulse.state() != QPropertyAnimation.State.Running:
self._pulse.start()
elif not pulse:
self._pulse.stop()
self._dot_opacity = 1.0
self.setToolTip(text)
self.updateGeometry()
self.update()
@property
def text(self) -> str:
"""Current label."""
return self._text
@property
def pulsing(self) -> bool:
"""Whether the dot animation runs."""
return self._pulse.state() == QPropertyAnimation.State.Running
def _get_dot_opacity(self) -> float:
return self._dot_opacity
def _set_dot_opacity(self, value: float) -> None:
self._dot_opacity = value
self.update()
dot_opacity = Property(float, _get_dot_opacity, _set_dot_opacity)
def sizeHint(self) -> QSize: # pylint: disable=invalid-name
return QSize(self.fontMetrics().horizontalAdvance(self._text) + 26, 22)
def paintEvent(self, _event): # pylint: disable=invalid-name
painter = QPainter(self)
painter.setRenderHint(QPainter.RenderHint.Antialiasing)
rect = QRectF(self.rect())
background = QColor(self._tone)
background.setAlphaF(0.16)
painter.setPen(Qt.PenStyle.NoPen)
painter.setBrush(background)
painter.drawRoundedRect(rect, rect.height() / 2, rect.height() / 2)
dot = QColor(self._tone)
dot.setAlphaF(self._dot_opacity)
painter.setBrush(dot)
painter.drawEllipse(QRectF(9, rect.height() / 2 - 3.5, 7, 7))
painter.setPen(self._tone)
painter.drawText(rect.adjusted(22, 0, 0, 0), Qt.AlignmentFlag.AlignVCenter, self._text)
painter.end()
def cleanup(self) -> None:
"""Stop the pulse animation."""
self._pulse.stop()
class Card(QFrame):
"""Rounded surface with an optional title row, matching ``BecUi.Card``."""
def __init__(self, title: str = "", parent: QWidget | None = None, padding: int = 12):
super().__init__(parent)
self.setObjectName("uxCard")
outer = QVBoxLayout(self)
outer.setContentsMargins(padding, padding, padding, padding)
outer.setSpacing(10)
self.header = QHBoxLayout()
self.header.setSpacing(8)
self.title_label = QLabel(title, self)
self.subtitle_label = QLabel("", self)
self.subtitle_label.setVisible(False)
self.header.addWidget(self.title_label)
self.header.addWidget(self.subtitle_label, 1)
self.header_extras = QHBoxLayout()
self.header_extras.setSpacing(4)
self.header.addLayout(self.header_extras)
outer.addLayout(self.header)
self.body = QVBoxLayout()
self.body.setSpacing(8)
outer.addLayout(self.body)
self._has_title = bool(title)
self._update_header_visibility()
self.refresh_theme(ThemeTokens())
def set_title(self, title: str, subtitle: str = "") -> None:
"""Set the title and an optional muted subtitle."""
self._has_title = bool(title)
self.title_label.setText(title)
self.subtitle_label.setText(subtitle)
self.subtitle_label.setVisible(bool(subtitle))
self._update_header_visibility()
def add_header_widget(self, widget: QWidget) -> None:
"""Add a widget to the right end of the title row."""
self.header_extras.addWidget(widget)
self._has_title = True
self._update_header_visibility()
def _update_header_visibility(self) -> None:
self.title_label.setVisible(bool(self.title_label.text()))
def refresh_theme(self, tokens: ThemeTokens) -> None:
"""Apply the theme tokens."""
self.setStyleSheet(
f"QFrame#uxCard {{ background: {tokens.card.name()};"
f" border: 1px solid {tokens.border.name()}; border-radius: 10px; }}"
)
self.title_label.setStyleSheet(
f"color: {tokens.fg.name()}; font-size: 13px; font-weight: 600; background: none;"
)
self.subtitle_label.setStyleSheet(
f"color: {tokens.fg_subtle.name()}; font-size: 12px; background: none;"
)
class PortedPropertiesMixin:
"""Make a port export and load the ``SafeProperty`` settings of the class it ports.
``BECConnector`` only collects the properties defined directly on ``type(self)``, so a
subclass would silently drop the inherited ones (e.g. ``ring_json``) from saved profiles.
Set ``PORTED_FROM`` to the original class; its properties and those of every class in
between are collected.
"""
PORTED_FROM: type | None = None
def _get_bec_meta_objects(self) -> dict:
objects = {}
for cls in type(self).__mro__:
for name, attr in vars(cls).items():
if name in objects or not isinstance(attr, Property):
continue
if getattr(attr.fget, "__is_safe_getter__", False):
objects[name] = attr
if cls is self.PORTED_FROM:
break
return objects
class ToggleSwitch(QPushButton):
"""Checkable switch painted like ``BecUi.SwitchField``; the label shows On or Off."""
def __init__(self, parent: QWidget | None = None):
super().__init__(parent)
self.setCheckable(True)
self.setCursor(Qt.CursorShape.PointingHandCursor)
self.setMinimumHeight(32)
self.setFixedWidth(84)
self._knob = 0.0
self._animation = QPropertyAnimation(self, b"knob", self)
self._animation.setDuration(120)
self._animation.setEasingCurve(QEasingCurve.Type.OutCubic)
self.toggled.connect(self._animate)
def _animate(self, checked: bool) -> None:
self._animation.stop()
self._animation.setStartValue(self._knob)
self._animation.setEndValue(1.0 if checked else 0.0)
self._animation.start()
def setChecked(self, checked: bool) -> None: # pylint: disable=invalid-name
"""Set the state without animating."""
self.blockSignals(True)
super().setChecked(checked)
self.blockSignals(False)
self._animation.stop()
self._knob = 1.0 if checked else 0.0
self.update()
def _get_knob(self) -> float:
return self._knob
def _set_knob(self, value: float) -> None:
self._knob = value
self.update()
knob = Property(float, _get_knob, _set_knob)
def paintEvent(self, _event): # pylint: disable=invalid-name
tokens = ThemeTokens()
painter = QPainter(self)
painter.setRenderHint(QPainter.RenderHint.Antialiasing)
track = QRectF(0, (self.height() - 22) / 2, 38, 22)
painter.setPen(tokens.primary if self.isChecked() else tokens.border)
painter.setBrush(tokens.primary if self.isChecked() else tokens.track)
painter.drawRoundedRect(track, 11, 11)
painter.setPen(Qt.PenStyle.NoPen)
painter.setBrush(tokens.on_primary if self.isChecked() else tokens.fg_muted)
painter.drawEllipse(QRectF(track.left() + 3 + self._knob * 16, track.top() + 3, 16, 16))
painter.setPen(tokens.fg)
font = self.font()
font.setPixelSize(13)
painter.setFont(font)
painter.drawText(
QRectF(track.right() + 8, 0, self.width() - track.right() - 8, self.height()),
Qt.AlignmentFlag.AlignVCenter,
"On" if self.isChecked() else "Off",
)
painter.end()
class SuffixLineEdit(QLineEdit):
"""Line edit that paints a unit suffix inside its right edge, like ``BecUi.InputField``."""
def __init__(self, parent: QWidget | None = None):
super().__init__(parent)
self._suffix = ""
def set_suffix(self, suffix: str) -> None:
"""Set the unit shown at the right edge."""
self._suffix = suffix or ""
width = self.fontMetrics().horizontalAdvance(self._suffix) + 10 if self._suffix else 0
self.setTextMargins(0, 0, width, 0)
self.update()
def paintEvent(self, event): # pylint: disable=invalid-name
super().paintEvent(event)
if not self._suffix:
return
painter = QPainter(self)
painter.setPen(ThemeTokens().fg_subtle)
font = self.font()
font.setPixelSize(12)
painter.setFont(font)
painter.drawText(
self.rect().adjusted(0, 0, -9, 0),
Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter,
self._suffix,
)
painter.end()
@@ -0,0 +1,205 @@
"""Demo window for the reworked notifications, used for screenshots and side-by-side checks.
Run this module with ``--variant qml`` or ``--variant qwidget`` (``python -m
bec_widgets.widgets.containers.main_window.addons.notification_center.notification_demo``).
No BEC server is needed: the buttons post sample
notifications and raise errors inside ``SafeSlot(popup_error=True)`` slots.
"""
from __future__ import annotations
import argparse
import sys
import traceback
import numpy as np
import pyqtgraph as pg
from qtpy.QtWidgets import (
QApplication,
QHBoxLayout,
QLabel,
QMainWindow,
QPushButton,
QToolBar,
QVBoxLayout,
QWidget,
)
from bec_widgets.utils.colors import apply_theme
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.widgets.containers.main_window.addons.notification_center.notification_ux_common import (
NotificationHub,
Severity,
)
SAMPLE_TRACE = """Traceback (most recent call last):
File "/opt/bec/bec_server/device_server/devices/devicemanager.py", line 412, in stage
status = obj.stage()
File "/opt/bec/ophyd_devices/devices/eiger9m.py", line 233, in on_stage
self.cam.acquire.set(1).wait(timeout=10)
File "/opt/bec/ophyd/status.py", line 336, in wait
raise WaitTimeoutError(f"Status {self!r} has not completed yet.")
ophyd.utils.errors.WaitTimeoutError: Status DeviceStatus(device=eiger9m_cam_acquire) has not completed yet.
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/opt/bec/bec_server/scan_server/scan_worker.py", line 289, in _stage_devices
self.device_manager.stage(devices, timeout=10)
bec_lib.alarm_handler.DeviceTimeoutError: Device eiger9m did not respond within 10 s while staging."""
def post_samples(hub: NotificationHub) -> None:
"""Fill the hub with a realistic mix of notifications for screenshots."""
hub.notify(
"Scan 1042 finished",
"line_scan over samx, 101 points in 2 min 13 s.",
Severity.SUCCESS,
source="Scan queue",
scan_number=1042,
)
hub.notify(
"Profile saved",
"Workspace 'alignment' was saved to the beamline profiles.",
Severity.INFO,
source="Workspace",
)
hub.notify(
"Ring current low",
"Ring current dropped to 180 mA (threshold 200 mA). Data may be noisy.",
Severity.WARNING,
source="Beamline states",
)
trace = ""
try:
raise ValueError("Target 12.5 is outside the soft limits [-10, 10] of samx.")
except ValueError:
trace = traceback.format_exc()
for _ in range(3):
hub.report_exception("Method error", trace, source="PositionerBox.move")
hub.post_alarm(
{
"severity": 2,
"alarm_type": "DeviceTimeoutError",
"msg": SAMPLE_TRACE,
"info": {
"id": "demo-critical",
"error_message": SAMPLE_TRACE,
"compact_error_message": "Device eiger9m did not respond within 10 s while"
" staging. The scan was aborted.",
"exception_type": "DeviceTimeoutError",
"device": "eiger9m",
},
},
{"scan_number": 1043, "scan_id": "4f0d6c1e-7a51-4b0c-9a59-2f9a3d1c2b77"},
)
class DemoWindow(QMainWindow):
"""Window that looks like the BEC app, with buttons that produce notifications."""
def __init__(self, variant: str = "qwidget"):
super().__init__()
self.setWindowTitle(f"BEC notifications ({variant})")
self.resize(1180, 720)
self.menuBar().addMenu("File")
self.menuBar().addMenu("View")
self.menuBar().addMenu("Help")
toolbar = QToolBar("Main", self)
for label in ("Dock area", "Device manager", "Scan control"):
toolbar.addAction(label)
self.addToolBar(toolbar)
central = QWidget(self)
layout = QHBoxLayout(central)
controls = QVBoxLayout()
self.hub = NotificationHub.instance()
for label, slot in (
("Info", lambda: self.hub.notify("Profile saved", "Workspace saved.", "info")),
("Success", lambda: self.hub.notify("Scan finished", "101 points.", "success")),
("Warning", lambda: self.hub.notify("Ring current low", "180 mA.", "warning")),
("Error in a slot", self.failing_slot),
("Critical alarm", self.critical),
("Burst of 20 errors", self.burst),
("Toggle theme", self.toggle_theme),
):
button = QPushButton(label, central)
button.clicked.connect(slot)
controls.addWidget(button)
controls.addStretch(1)
layout.addLayout(controls)
plot = pg.PlotWidget(central)
x = np.linspace(-5, 5, 400)
plot.plot(x, np.exp(-(x**2) / 2) + 0.05 * np.sin(9 * x), pen=pg.mkPen("#3b82f6", width=2))
plot.showGrid(x=True, y=True, alpha=0.2)
layout.addWidget(plot, 1)
self.setCentralWidget(central)
self.statusBar().addWidget(QLabel("Ready · account p12345"))
if variant == "qml":
from bec_widgets.widgets.containers.main_window.addons.notification_center.notification_qml import (
NotificationHostQML as Host,
)
else:
from bec_widgets.widgets.containers.main_window.addons.notification_center.notification_qwidget import (
NotificationHostQWidget as Host,
)
self.notifications = Host(self)
@SafeSlot(popup_error=True)
def failing_slot(self):
"""Raise inside a SafeSlot: shows a toast instead of a modal dialog."""
raise ValueError("Target 12.5 is outside the soft limits [-10, 10] of samx.")
def critical(self):
"""Post a critical alarm with a long traceback."""
self.hub.post_alarm(
{
"severity": 2,
"alarm_type": "DeviceTimeoutError",
"msg": SAMPLE_TRACE,
"info": {
"id": None,
"error_message": SAMPLE_TRACE,
"compact_error_message": "Device eiger9m did not respond within 10 s.",
"exception_type": "DeviceTimeoutError",
"device": "eiger9m",
},
},
{"scan_number": 1043},
)
def burst(self):
"""Post many different errors at once; the stack shows three and counts the rest."""
for i in range(20):
self.hub.notify(f"Readback failed for motor{i}", "Timeout after 2 s.", "error")
def toggle_theme(self):
"""Switch between the light and dark theme."""
theme = getattr(QApplication.instance(), "theme", None)
apply_theme("light" if theme is not None and theme.theme == "dark" else "dark")
def closeEvent(self, event): # pylint: disable=invalid-name
"""Detach the notification UI."""
self.notifications.cleanup()
super().closeEvent(event)
def main(variant: str | None = None) -> None: # pragma: no cover
"""Start the demo."""
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--variant", choices=("qml", "qwidget"), default=variant or "qwidget")
parser.add_argument("--theme", choices=("dark", "light"), default="dark")
parser.add_argument("--samples", action="store_true", help="start with sample notifications")
args = parser.parse_args()
app = QApplication(sys.argv)
apply_theme(args.theme)
window = DemoWindow(args.variant)
window.show()
if args.samples:
post_samples(window.hub)
sys.exit(app.exec())
if __name__ == "__main__": # pragma: no cover
main()
@@ -0,0 +1,177 @@
"""Notification toasts, history drawer and bell rendered with Qt Quick (QML).
Same UX as :mod:`notification_qwidget`; the state comes from
:class:`~.notification_ux_common.NotificationHostBase`.
"""
from __future__ import annotations
from pathlib import Path
from qtpy.QtCore import Property, QObject, Qt, Signal, Slot
from qtpy.QtQuickWidgets import QQuickWidget
from bec_widgets.utils.quick import DictListModel, create_quick_widget, release_quick_widget
from bec_widgets.widgets.containers.main_window.addons.notification_center.notification_ux_common import (
NotificationHostBase,
)
QML_DIR = Path(__file__).parent / "qml"
ROW_ROLES = [
"entryId",
"severity",
"severityLabel",
"icon",
"color",
"tint",
"title",
"body",
"details",
"hasDetails",
"source",
"meta",
"time",
"timeAbs",
"firstAbs",
"scan",
"count",
"unread",
"needsAck",
"critical",
]
class NotificationBackend(QObject):
"""Bridge between :class:`NotificationHostQML` and its three QML views."""
changed = Signal()
def __init__(self, host: NotificationHostQML):
super().__init__(host)
self._host = host
self._state: dict = {}
self._toasts = DictListModel(ROW_ROLES, self)
self._rows = DictListModel(ROW_ROLES, self)
def update_from(self, state: dict) -> None:
"""Take a new view state from the host."""
state = dict(state)
self._toasts.set_items(state.pop("toasts"))
self._rows.set_items(state.pop("rows"))
state["selected"] = state["selected"] or {}
self._state = state
self.changed.emit()
# pylint: disable=invalid-name,missing-function-docstring
@Slot()
def toggleDrawer(self) -> None:
self._host.toggle_drawer()
@Slot(bool)
def setDrawerOpen(self, open_: bool) -> None:
self._host.set_drawer_open(open_)
@Slot(str)
def openDetails(self, entry_id: str) -> None:
self._host.open_details(entry_id)
@Slot(str)
def select(self, entry_id: str) -> None:
self._host.select(entry_id or None)
@Slot(str)
def setFilter(self, key: str) -> None:
self._host.set_filter(key)
@Slot(str)
def setSearch(self, text: str) -> None:
self._host.set_search(text)
@Slot(str)
def acknowledge(self, entry_id: str) -> None:
self._host.acknowledge(entry_id)
@Slot(str)
def remove(self, entry_id: str) -> None:
self._host.remove(entry_id)
@Slot(str)
def dismissToast(self, entry_id: str) -> None:
self._host.dismiss_toast(entry_id)
@Slot(str, result=bool)
def copyDetails(self, entry_id: str) -> bool:
return self._host.copy_details(entry_id)
@Slot()
def markAllRead(self) -> None:
self._host.mark_all_read()
@Slot()
def clearHistory(self) -> None:
self._host.clear_history()
@Slot(bool)
def setHovered(self, hovered: bool) -> None:
self._host.set_toasts_hovered(hovered)
state = Property("QVariantMap", lambda self: self._state, notify=changed)
toasts = Property(QObject, lambda self: self._toasts, constant=True)
rows = Property(QObject, lambda self: self._rows, constant=True)
class NotificationHostQML(NotificationHostBase):
"""Notification UI rendered in QML. See :class:`NotificationHostBase`."""
def _create_views(self) -> None:
self.backend = NotificationBackend(self)
props = {"backend": self.backend}
# toasts float over the window content: a transparent view stacked on top
self.toasts = create_quick_widget(self.window, QML_DIR / "NotificationToasts.qml", props)
self.toasts.setAttribute(Qt.WidgetAttribute.WA_AlwaysStackOnTop, True)
self.toasts.setAttribute(Qt.WidgetAttribute.WA_TranslucentBackground, True)
self.toasts.setClearColor(Qt.GlobalColor.transparent)
self.toasts.hide()
self.drawer = create_quick_widget(self.window, QML_DIR / "NotificationDrawer.qml", props)
self.drawer.hide()
self.bell = None
if self.status_bar is not None:
self.bell = create_quick_widget(
self.status_bar, QML_DIR / "NotificationBell.qml", props
)
self.bell.setFixedSize(36, 24)
self.bell.setResizeMode(QQuickWidget.ResizeMode.SizeRootObjectToView)
self.status_bar.addPermanentWidget(self.bell)
root = self.toasts.rootObject()
if root is not None:
root.implicitHeightChanged.connect(self._place_views)
def _sync_views(self, state: dict) -> None:
self.backend.update_from(state)
self.toasts.setVisible(bool(state["toasts"]))
self.drawer.setVisible(state["drawerOpen"])
def _place_views(self, *_args) -> None:
if self.drawer.isVisible():
self.drawer.setGeometry(self.drawer_geometry())
self.drawer.raise_()
if self.toasts.isVisible():
root = self.toasts.rootObject()
height = int(root.property("implicitHeight")) if root is not None else 0
rect = self.toast_geometry(max(0, height - 16))
self.toasts.setGeometry(rect.adjusted(-8, -8, 8, 8) & self.window.rect())
self.toasts.raise_()
def _apply_theme(self) -> None:
if self.bell is not None:
self.bell.setClearColor(self.status_bar.palette().window().color())
self.drawer.setClearColor(self.window.palette().window().color())
def _destroy_views(self) -> None:
for view in (self.toasts, self.drawer, self.bell):
if view is None:
continue
release_quick_widget(view)
view.hide()
view.deleteLater()
@@ -0,0 +1,740 @@
"""Notification toasts, history drawer and bell built with QWidgets.
Same UX as :mod:`notification_qml`; the state comes from
:class:`~.notification_ux_common.NotificationHostBase`.
"""
from __future__ import annotations
from bec_qthemes import material_icon
from qtpy.QtCore import QAbstractListModel, QModelIndex, QRectF, QSize, Qt
from qtpy.QtGui import QColor, QFont, QFontDatabase, QPainter, QPainterPath, QPen
from qtpy.QtWidgets import (
QFrame,
QGraphicsDropShadowEffect,
QHBoxLayout,
QLabel,
QLineEdit,
QListView,
QPlainTextEdit,
QPushButton,
QScrollArea,
QStackedWidget,
QStyle,
QStyledItemDelegate,
QToolButton,
QVBoxLayout,
QWidget,
)
from bec_widgets.utils.quick.host import ThemeTokens
from bec_widgets.utils.ux_kit import IconButton, TextButton, field_qss, rgba
from bec_widgets.widgets.containers.main_window.addons.notification_center.notification_ux_common import (
NotificationHostBase,
)
def _icon_badge(painter: QPainter, rect: QRectF, icon: str, color: QColor, tint: QColor):
"""Paint a round tinted badge with a Material icon, as used by toasts and rows."""
painter.setPen(Qt.PenStyle.NoPen)
painter.setBrush(tint)
painter.drawEllipse(rect)
size = int(rect.width() * 0.6)
pixmap = material_icon(icon, size=(size * 2, size * 2), color=color, filled=True)
target = QRectF(rect.center().x() - size / 2, rect.center().y() - size / 2, size, size).toRect()
painter.drawPixmap(target, pixmap)
class _Badge(QWidget):
"""Round severity icon painted on a tinted circle."""
def __init__(self, parent: QWidget | None = None, size: int = 32):
super().__init__(parent)
self.setFixedSize(size, size)
self._icon = "info"
self._color = QColor("#888")
self._tint = QColor("#333")
def set_icon(self, icon: str, color: str, tint: str) -> None:
"""Set icon name and colours."""
self._icon, self._color, self._tint = icon, QColor(color), QColor(tint)
self.update()
def paintEvent(self, _event): # pylint: disable=invalid-name
painter = QPainter(self)
painter.setRenderHint(QPainter.RenderHint.Antialiasing)
_icon_badge(painter, QRectF(self.rect()), self._icon, self._color, self._tint)
painter.end()
class ToastCard(QFrame):
"""One toast: badge, title, message, meta line and actions."""
def __init__(self, host: NotificationHostBase, parent: QWidget | None = None):
super().__init__(parent)
self.host = host
self.entry_id = ""
self._row: dict = {}
self.setObjectName("toastCard")
self.setAttribute(Qt.WidgetAttribute.WA_StyledBackground, False)
self.badge = _Badge(self, 32)
self.title = QLabel(self)
self.title.setTextFormat(Qt.TextFormat.PlainText)
self.count = QLabel(self)
self.close_btn = IconButton("close", "Dismiss", self, size=16)
self.close_btn.setFixedSize(24, 24)
self.close_btn.clicked.connect(lambda: self.host.dismiss_toast(self.entry_id))
self.body = QLabel(self)
self.body.setWordWrap(True)
self.body.setTextFormat(Qt.TextFormat.PlainText)
self.meta = QLabel(self)
self.details_btn = TextButton("Details", "ghost", "open_in_new", self)
self.details_btn.setMinimumHeight(28)
self.details_btn.clicked.connect(lambda: self.host.open_details(self.entry_id))
self.copy_btn = IconButton("content_copy", "Copy report", self, size=16)
self.copy_btn.setFixedSize(28, 28)
self.copy_btn.clicked.connect(lambda: self.host.copy_details(self.entry_id))
self.ack_btn = TextButton("Acknowledge", "danger", "done", self)
self.ack_btn.setMinimumHeight(28)
self.ack_btn.clicked.connect(lambda: self.host.acknowledge(self.entry_id))
head = QHBoxLayout()
head.setSpacing(6)
head.addWidget(self.title, 1)
head.addWidget(self.count)
head.addWidget(self.close_btn)
text = QVBoxLayout()
text.setSpacing(3)
text.addLayout(head)
text.addWidget(self.body)
text.addWidget(self.meta)
self.actions = QHBoxLayout()
self.actions.setSpacing(6)
self.actions.addStretch(1)
self.actions.addWidget(self.copy_btn)
self.actions.addWidget(self.details_btn)
self.actions.addWidget(self.ack_btn)
text.addLayout(self.actions)
row = QHBoxLayout(self)
row.setContentsMargins(14, 12, 10, 10)
row.setSpacing(12)
row.addWidget(self.badge, 0, Qt.AlignmentFlag.AlignTop)
row.addLayout(text, 1)
shadow = QGraphicsDropShadowEffect(self)
shadow.setBlurRadius(24)
shadow.setOffset(0, 4)
shadow.setColor(QColor(0, 0, 0, 90))
self.setGraphicsEffect(shadow)
def set_row(self, row: dict, tokens: ThemeTokens) -> None:
"""Show the entry described by ``row``."""
self._row = row
self.entry_id = row["id"]
self.badge.set_icon(row["icon"], row["color"], row["tint"])
self.title.setText(row["title"])
self.title.setToolTip(row["title"])
self.count.setText(f"×{row['count']}")
self.count.setVisible(row["count"] > 1)
body = row["body"] or ""
self.body.setText(body if len(body) <= 220 else body[:217] + "…")
self.body.setToolTip(body if len(body) > 220 else "")
self.body.setVisible(bool(body))
self.meta.setText(row["meta"])
self.close_btn.setVisible(not row["needsAck"])
self.ack_btn.setVisible(row["needsAck"])
self.details_btn.setVisible(row["hasDetails"] or row["needsAck"])
self.copy_btn.setVisible(row["severity"] in ("error", "critical"))
self._style(tokens)
self.update()
def _style(self, tokens: ThemeTokens) -> None:
self.title.setStyleSheet(
f"color: {tokens.fg.name()}; font-size: 13px; font-weight: 600; background: none;"
)
self.body.setStyleSheet(
f"color: {tokens.fg_muted.name()}; font-size: 12px; background: none;"
)
self.meta.setStyleSheet(
f"color: {tokens.fg_subtle.name()}; font-size: 11px; background: none;"
)
self.count.setStyleSheet(
f"color: {self._row.get('color', tokens.fg.name())}; font-size: 11px;"
f" font-weight: 700; background: {self._row.get('tint', 'transparent')};"
" border-radius: 8px; padding: 0px 6px;"
)
self.count.setFixedHeight(16)
for button in (self.close_btn, self.details_btn, self.copy_btn, self.ack_btn):
button.refresh_theme(tokens)
def paintEvent(self, _event): # pylint: disable=invalid-name
tokens = ThemeTokens()
painter = QPainter(self)
painter.setRenderHint(QPainter.RenderHint.Antialiasing)
rect = QRectF(self.rect()).adjusted(0.5, 0.5, -0.5, -0.5)
path = QPainterPath()
path.addRoundedRect(rect, 10, 10)
painter.fillPath(path, tokens.card)
accent = QColor(self._row.get("color", tokens.border.name()))
critical = self._row.get("needsAck", False)
painter.setPen(QPen(accent if critical else tokens.border, 1.5 if critical else 1))
painter.drawPath(path)
# severity stripe along the straight part of the left edge
painter.setPen(Qt.PenStyle.NoPen)
painter.setBrush(accent)
painter.drawRoundedRect(QRectF(rect.left(), rect.top() + 10, 4, rect.height() - 20), 2, 2)
painter.end()
class ToastStack(QWidget):
"""Bottom-right column of toasts with an overflow link; hovering pauses the countdowns."""
def __init__(self, host: NotificationHostBase, parent: QWidget):
super().__init__(parent)
self.host = host
self.setAttribute(Qt.WidgetAttribute.WA_TranslucentBackground, True)
self._cards: dict[str, ToastCard] = {}
self._layout = QVBoxLayout(self)
self._layout.setContentsMargins(8, 8, 8, 8) # room for the shadow
self._layout.setSpacing(10)
self.more_btn = TextButton("", "neutral", "expand_less", self)
self.more_btn.clicked.connect(lambda: self.host.set_drawer_open(True))
self._layout.addWidget(self.more_btn, 0, Qt.AlignmentFlag.AlignRight)
def set_state(self, state: dict, tokens: ThemeTokens) -> None:
"""Show the toasts of ``state``; cards are reused by entry id."""
rows = state["toasts"]
wanted = [row["id"] for row in rows]
for entry_id in list(self._cards):
if entry_id not in wanted:
card = self._cards.pop(entry_id)
self._layout.removeWidget(card)
card.deleteLater()
# newest at the bottom, nearest to the status bar
for position, row in enumerate(rows, start=1):
card = self._cards.get(row["id"])
if card is None:
card = ToastCard(self.host, self)
self._cards[row["id"]] = card
card.set_row(row, tokens)
if self._layout.indexOf(card) != position:
self._layout.removeWidget(card)
self._layout.insertWidget(position, card)
overflow = state["overflow"]
self.more_btn.setText(f"{overflow} more in notifications")
self.more_btn.setVisible(overflow > 0)
self.more_btn.refresh_theme(tokens)
# solid like a card: the link floats over plots
self.more_btn.setStyleSheet(
self.more_btn.styleSheet()
+ f"QPushButton {{ background: {tokens.card.name()}; }}"
+ f"QPushButton:hover {{ background: {tokens.hover.name()}; }}"
)
self.setVisible(bool(rows))
@property
def cards(self) -> list[ToastCard]:
"""Cards on screen, oldest first."""
return [
self._layout.itemAt(i).widget()
for i in range(self._layout.count())
if isinstance(self._layout.itemAt(i).widget(), ToastCard)
]
def enterEvent(self, event): # pylint: disable=invalid-name
self.host.set_toasts_hovered(True)
super().enterEvent(event)
def leaveEvent(self, event): # pylint: disable=invalid-name
self.host.set_toasts_hovered(False)
super().leaveEvent(event)
class _RowsModel(QAbstractListModel):
"""Rows of the history list; each row is the dictionary from ``entry_row``."""
def __init__(self, parent=None):
super().__init__(parent)
self.rows: list[dict] = []
def set_rows(self, rows: list[dict]) -> None:
"""Replace the rows."""
self.beginResetModel()
self.rows = rows
self.endResetModel()
def rowCount(self, parent: QModelIndex = QModelIndex()) -> int: # pylint: disable=invalid-name
return 0 if parent.isValid() else len(self.rows)
def data(self, index: QModelIndex, role: int = Qt.ItemDataRole.DisplayRole):
if not index.isValid():
return None
row = self.rows[index.row()]
if role == Qt.ItemDataRole.UserRole:
return row
if role == Qt.ItemDataRole.DisplayRole:
return row["title"]
if role == Qt.ItemDataRole.ToolTipRole:
return row["timeAbs"]
return None
class _RowDelegate(QStyledItemDelegate):
"""Paints a history row: stripe, badge, title, time, message and meta line."""
HEIGHT = 78
def sizeHint(self, _option, _index): # pylint: disable=invalid-name
return QSize(200, self.HEIGHT)
def paint(self, painter: QPainter, option, index):
row = index.data(Qt.ItemDataRole.UserRole)
if row is None:
return
tokens = ThemeTokens()
painter.save()
painter.setRenderHint(QPainter.RenderHint.Antialiasing)
rect = QRectF(option.rect).adjusted(8, 3, -8, -3)
hovered = bool(option.state & QStyle.StateFlag.State_MouseOver)
background = QColor(row["tint"]) if row["needsAck"] else tokens.card
if hovered:
background = tokens.hover if not row["needsAck"] else QColor(row["tint"]).lighter(110)
path = QPainterPath()
path.addRoundedRect(rect, 8, 8)
painter.fillPath(path, background)
painter.setPen(QPen(QColor(row["color"]) if row["needsAck"] else tokens.border, 1))
painter.drawPath(path)
_icon_badge(
painter,
QRectF(rect.left() + 10, rect.top() + 10, 28, 28),
row["icon"],
QColor(row["color"]),
QColor(row["tint"]) if not row["needsAck"] else tokens.card,
)
left = rect.left() + 48
right = rect.right() - 10
font = QFont(option.font)
font.setPixelSize(11)
painter.setFont(font)
time_w = painter.fontMetrics().horizontalAdvance(row["time"])
painter.setPen(tokens.fg_subtle)
painter.drawText(
QRectF(right - time_w, rect.top() + 8, time_w, 18),
Qt.AlignmentFlag.AlignVCenter,
row["time"],
)
title_right = right - time_w - 8
if row["unread"]:
painter.setPen(Qt.PenStyle.NoPen)
painter.setBrush(tokens.primary)
painter.drawEllipse(QRectF(title_right - 7, rect.top() + 14, 7, 7))
title_right -= 14
if row["count"] > 1:
badge = f"×{row['count']}"
badge_w = painter.fontMetrics().horizontalAdvance(badge) + 12
badge_rect = QRectF(title_right - badge_w, rect.top() + 9, badge_w, 16)
painter.setPen(Qt.PenStyle.NoPen)
painter.setBrush(QColor(row["tint"]))
painter.drawRoundedRect(badge_rect, 8, 8)
painter.setPen(QColor(row["color"]))
painter.drawText(badge_rect, Qt.AlignmentFlag.AlignCenter, badge)
title_right -= badge_w + 6
font.setPixelSize(13)
font.setWeight(QFont.Weight.DemiBold)
painter.setFont(font)
painter.setPen(tokens.fg)
title = painter.fontMetrics().elidedText(
row["title"], Qt.TextElideMode.ElideRight, int(title_right - left)
)
painter.drawText(
QRectF(left, rect.top() + 8, title_right - left, 18),
Qt.AlignmentFlag.AlignVCenter,
title,
)
font.setPixelSize(12)
font.setWeight(QFont.Weight.Normal)
painter.setFont(font)
painter.setPen(tokens.fg_muted)
body = painter.fontMetrics().elidedText(
row["body"].replace("\n", " "), Qt.TextElideMode.ElideRight, int(right - left)
)
painter.drawText(
QRectF(left, rect.top() + 28, right - left, 18), Qt.AlignmentFlag.AlignVCenter, body
)
font.setPixelSize(11)
painter.setFont(font)
meta_color = QColor(row["color"]) if row["needsAck"] else tokens.fg_subtle
painter.setPen(meta_color)
meta = ("Needs acknowledgement · " if row["needsAck"] else "") + " · ".join(
part for part in (row["severityLabel"], row["source"], row["scan"]) if part
)
meta = painter.fontMetrics().elidedText(
meta, Qt.TextElideMode.ElideRight, int(right - left)
)
painter.drawText(
QRectF(left, rect.top() + 48, right - left, 16), Qt.AlignmentFlag.AlignVCenter, meta
)
painter.restore()
class _DetailsPage(QWidget):
"""Full view of one entry: message, metadata, details text and actions."""
def __init__(self, host: NotificationHostBase, parent: QWidget | None = None):
super().__init__(parent)
self.host = host
self.entry_id = ""
self.back_btn = TextButton("All notifications", "ghost", "arrow_back", self)
self.back_btn.clicked.connect(lambda: self.host.select(None))
self.badge = _Badge(self, 36)
self.title = QLabel(self)
self.title.setWordWrap(True)
self.title.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
self.pill = QLabel(self)
self.pill.setFixedHeight(18)
self.body = QLabel(self)
self.body.setWordWrap(True)
self.body.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
self.facts = QLabel(self)
self.facts.setTextFormat(Qt.TextFormat.RichText)
self.details_label = QLabel("Details", self)
self.details = QPlainTextEdit(self)
self.details.setReadOnly(True)
self.details.setLineWrapMode(QPlainTextEdit.LineWrapMode.NoWrap)
mono = QFontDatabase.systemFont(QFontDatabase.SystemFont.FixedFont)
mono.setPixelSize(11)
self.details.setFont(mono)
self.copy_btn = TextButton("Copy report", "neutral", "content_copy", self)
self.copy_btn.clicked.connect(self._copy)
self.ack_btn = TextButton("Acknowledge", "danger", "done", self)
self.ack_btn.clicked.connect(lambda: self.host.acknowledge(self.entry_id))
self.remove_btn = TextButton("Remove", "ghost", "delete", self)
self.remove_btn.clicked.connect(lambda: self.host.remove(self.entry_id))
head = QHBoxLayout()
head.setSpacing(12)
head.addWidget(self.badge, 0, Qt.AlignmentFlag.AlignTop)
title_col = QVBoxLayout()
title_col.setSpacing(4)
title_col.addWidget(self.pill, 0, Qt.AlignmentFlag.AlignLeft)
title_col.addWidget(self.title)
head.addLayout(title_col, 1)
actions = QHBoxLayout()
actions.setSpacing(6)
actions.addWidget(self.copy_btn)
actions.addWidget(self.ack_btn)
actions.addStretch(1)
actions.addWidget(self.remove_btn)
layout = QVBoxLayout(self)
layout.setContentsMargins(12, 4, 12, 12)
layout.setSpacing(10)
layout.addWidget(self.back_btn, 0, Qt.AlignmentFlag.AlignLeft)
layout.addLayout(head)
layout.addWidget(self.body)
layout.addWidget(self.facts)
layout.addWidget(self.details_label)
layout.addWidget(self.details, 1)
layout.addStretch(0)
layout.addLayout(actions)
def _copy(self) -> None:
if self.host.copy_details(self.entry_id):
self.copy_btn.setText("Copied")
def set_row(self, row: dict, tokens: ThemeTokens) -> None:
"""Show ``row``."""
if row["id"] != self.entry_id:
self.copy_btn.setText("Copy report")
self.entry_id = row["id"]
self.badge.set_icon(row["icon"], row["color"], row["tint"])
self.title.setText(row["title"])
label = row["severityLabel"] + (" · needs acknowledgement" if row["needsAck"] else "")
self.pill.setText(label)
self.body.setText(row["body"] or "No message.")
facts = [("When", row["timeAbs"])]
if row["count"] > 1:
facts.append(("Repeated", f"{row['count']} times since {row['firstAbs']}"))
if row["source"]:
facts.append(("Source", row["source"]))
if row["scan"]:
facts.append(("Scan", row["scan"]))
self.facts.setText(
"<table cellspacing='0' cellpadding='2'>"
+ "".join(
f"<tr><td style='color:{tokens.fg_subtle.name()}; padding-right:12px'>{k}</td>"
f"<td style='color:{tokens.fg.name()}'>{v}</td></tr>"
for k, v in facts
)
+ "</table>"
)
if self.details.toPlainText() != row["details"]:
self.details.setPlainText(row["details"])
self.details.setVisible(row["hasDetails"])
self.details_label.setVisible(row["hasDetails"])
self.ack_btn.setVisible(row["needsAck"])
self.title.setStyleSheet(
f"color: {tokens.fg.name()}; font-size: 15px; font-weight: 600; background: none;"
)
self.pill.setStyleSheet(
f"color: {row['color']}; background: {row['tint']}; border-radius: 9px;"
" padding: 1px 8px; font-size: 11px; font-weight: 600;"
)
self.body.setStyleSheet(f"color: {tokens.fg.name()}; font-size: 13px; background: none;")
self.facts.setStyleSheet("font-size: 12px; background: none;")
self.details_label.setStyleSheet(
f"color: {tokens.fg_subtle.name()}; font-size: 11px; font-weight: 600;"
" text-transform: uppercase; background: none;"
)
self.details.setStyleSheet(
f"QPlainTextEdit {{ background: {tokens.field.name()}; color: {tokens.fg.name()};"
f" border: 1px solid {tokens.border.name()}; border-radius: 8px; padding: 6px; }}"
)
for button in (self.back_btn, self.copy_btn, self.ack_btn, self.remove_btn):
button.refresh_theme(tokens)
class HistoryDrawer(QFrame):
"""Right-edge panel with filters, search, the notification history and details."""
def __init__(self, host: NotificationHostBase, parent: QWidget):
super().__init__(parent)
self.host = host
self.setObjectName("notificationDrawer")
self.title = QLabel("Notifications", self)
self.subtitle = QLabel(self)
self.read_btn = IconButton("done_all", "Mark all as read", self)
self.read_btn.clicked.connect(self.host.mark_all_read)
self.clear_btn = IconButton("clear_all", "Clear history", self)
self.clear_btn.clicked.connect(self.host.clear_history)
self.close_btn = IconButton("close", "Close", self)
self.close_btn.clicked.connect(lambda: self.host.set_drawer_open(False))
self.chips: dict[str, QPushButton] = {}
chips_row = QHBoxLayout()
chips_row.setSpacing(6)
for key in ("all", "errors", "warnings", "info"):
chip = QPushButton(self)
chip.setCheckable(True)
chip.setFixedHeight(26)
chip.setCursor(Qt.CursorShape.PointingHandCursor)
chip.clicked.connect(lambda _=False, k=key: self.host.set_filter(k))
self.chips[key] = chip
chips_row.addWidget(chip)
chips_row.addStretch(1)
self.search = QLineEdit(self)
self.search.setPlaceholderText("Search notifications")
self.search.setClearButtonEnabled(True)
self.search.textChanged.connect(self.host.set_search)
self.model = _RowsModel(self)
self.list = QListView(self)
self.list.setModel(self.model)
self.list.setItemDelegate(_RowDelegate(self.list))
self.list.setMouseTracking(True)
self.list.setUniformItemSizes(True)
self.list.setVerticalScrollMode(QListView.ScrollMode.ScrollPerPixel)
self.list.setFrameShape(QFrame.Shape.NoFrame)
self.list.clicked.connect(
lambda index: self.host.select(index.data(Qt.ItemDataRole.UserRole)["id"])
)
self.empty = QLabel(self)
self.empty.setAlignment(Qt.AlignmentFlag.AlignCenter)
self.empty.setWordWrap(True)
self.details = _DetailsPage(host, self)
details_scroll = QScrollArea(self)
details_scroll.setWidgetResizable(True)
details_scroll.setFrameShape(QFrame.Shape.NoFrame)
details_scroll.setWidget(self.details)
self.pages = QStackedWidget(self)
self.pages.addWidget(self.list)
self.pages.addWidget(self.empty)
self.pages.addWidget(details_scroll)
header = QHBoxLayout()
header.setSpacing(4)
title_col = QVBoxLayout()
title_col.setSpacing(0)
title_col.addWidget(self.title)
title_col.addWidget(self.subtitle)
header.addLayout(title_col, 1)
header.addWidget(self.read_btn)
header.addWidget(self.clear_btn)
header.addWidget(self.close_btn)
self.list_tools = QWidget(self)
tools = QVBoxLayout(self.list_tools)
tools.setContentsMargins(0, 0, 0, 0)
tools.setSpacing(8)
tools.addLayout(chips_row)
tools.addWidget(self.search)
top = QVBoxLayout()
top.setContentsMargins(16, 14, 12, 8)
top.setSpacing(10)
top.addLayout(header)
top.addWidget(self.list_tools)
layout = QVBoxLayout(self)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(0)
layout.addLayout(top)
layout.addWidget(self.pages, 1)
shadow = QGraphicsDropShadowEffect(self)
shadow.setBlurRadius(28)
shadow.setOffset(-4, 0)
shadow.setColor(QColor(0, 0, 0, 80))
self.setGraphicsEffect(shadow)
def set_state(self, state: dict, tokens: ThemeTokens) -> None:
"""Render the drawer part of ``state``."""
self.setVisible(state["drawerOpen"])
if not state["drawerOpen"]:
return
self._style(tokens)
unread = state["openCriticals"]
self.subtitle.setText(
f"{state['total']} in history" + (f" · {unread} critical open" if unread else "")
)
for item in state["filters"]:
chip = self.chips[item["key"]]
chip.setText(f"{item['label']} {item['count']}")
chip.setChecked(item["active"])
if self.search.text() != state["search"]:
self.search.blockSignals(True)
self.search.setText(state["search"])
self.search.blockSignals(False)
self.model.set_rows(state["rows"])
selected = state["selected"]
self.list_tools.setVisible(selected is None)
if selected is not None:
self.details.set_row(selected, tokens)
self.pages.setCurrentIndex(2)
elif state["rows"]:
self.pages.setCurrentIndex(0)
else:
if state["total"]:
self.empty.setText("Nothing matches this filter.")
else:
self.empty.setText("You're all caught up.\nNew messages and errors appear here.")
self.pages.setCurrentIndex(1)
def _style(self, tokens: ThemeTokens) -> None:
chip_qss = (
f"QPushButton {{ background: transparent; color: {tokens.fg_muted.name()};"
f" border: 1px solid {tokens.border.name()}; border-radius: 13px;"
f" padding: 3px 10px; font-size: 12px; }}"
f"QPushButton:hover {{ background: {tokens.hover.name()}; }}"
f"QPushButton:checked {{ background: {rgba(tokens.primary, 0.18)};"
f" color: {tokens.fg.name()}; border-color: {tokens.primary.name()};"
f" font-weight: 600; }}"
)
self.setStyleSheet(
f"QFrame#notificationDrawer {{ background: {tokens.bg.name()};"
f" border-left: 1px solid {tokens.border.name()}; }}"
f"QListView, QScrollArea, QStackedWidget, QScrollArea > QWidget > QWidget"
f" {{ background: {tokens.bg.name()}; border: none; }}" + field_qss(tokens)
)
for chip in self.chips.values():
chip.setStyleSheet(chip_qss)
self.title.setStyleSheet(
f"color: {tokens.fg.name()}; font-size: 15px; font-weight: 600; background: none;"
)
self.subtitle.setStyleSheet(
f"color: {tokens.fg_subtle.name()}; font-size: 11px; background: none;"
)
self.empty.setStyleSheet(
f"color: {tokens.fg_subtle.name()}; font-size: 13px; background: none;"
)
for button in (self.read_btn, self.clear_btn, self.close_btn):
button.refresh_theme(tokens)
class BellButton(QToolButton):
"""Status-bar bell with a badge counting unread notifications."""
def __init__(self, host: NotificationHostBase, parent: QWidget | None = None):
super().__init__(parent)
self.host = host
self._state: dict = {}
self.setAutoRaise(True)
self.setCheckable(True)
self.setFixedSize(36, 24)
self.setIconSize(QSize(18, 18))
self.setCursor(Qt.CursorShape.PointingHandCursor)
self.setAccessibleName("Notifications")
self.clicked.connect(self.host.toggle_drawer)
def set_state(self, state: dict, tokens: ThemeTokens) -> None:
"""Update icon, badge and tooltip."""
self._state = state
self.setChecked(state["drawerOpen"])
color = state["bellColor"] if state["unread"] or state["openCriticals"] else tokens.fg_muted
self.setIcon(
material_icon(state["bellIcon"], size=(36, 36), color=color, convert_to_pixmap=False)
)
self.setToolTip(state["bellTip"])
self.setStyleSheet(
f"QToolButton {{ border: none; border-radius: 6px; background: transparent; }}"
f"QToolButton:hover, QToolButton:checked {{ background: {tokens.hover.name()}; }}"
)
self.update()
def paintEvent(self, event): # pylint: disable=invalid-name
super().paintEvent(event)
unread = self._state.get("unread", 0)
if not unread:
return
text = "99+" if unread > 99 else str(unread)
painter = QPainter(self)
painter.setRenderHint(QPainter.RenderHint.Antialiasing)
font = self.font()
font.setPixelSize(9)
font.setWeight(QFont.Weight.Bold)
painter.setFont(font)
width = max(14, painter.fontMetrics().horizontalAdvance(text) + 8)
rect = QRectF(self.width() - width - 1, 1, width, 14)
painter.setPen(Qt.PenStyle.NoPen)
painter.setBrush(QColor(self._state["bellColor"]))
painter.drawRoundedRect(rect, 7, 7)
painter.setPen(QColor("white"))
painter.drawText(rect, Qt.AlignmentFlag.AlignCenter, text)
painter.end()
class NotificationHostQWidget(NotificationHostBase):
"""Notification UI drawn with QWidgets. See :class:`NotificationHostBase`."""
def _create_views(self) -> None:
self.toasts = ToastStack(self, self.window)
self.drawer = HistoryDrawer(self, self.window)
self.drawer.hide()
self.toasts.hide()
self.bell = BellButton(self)
if self.status_bar is not None:
self.status_bar.addPermanentWidget(self.bell)
def _sync_views(self, state: dict) -> None:
tokens = ThemeTokens()
self.toasts.set_state(state, tokens)
self.drawer.set_state(state, tokens)
self.bell.set_state(state, tokens)
def _place_views(self) -> None:
if self.drawer.isVisible():
self.drawer.setGeometry(self.drawer_geometry())
self.drawer.raise_()
if self.toasts.isVisible():
# the stack has an 8 px margin for the shadow, so the cards line up with MARGIN
width = self.toast_geometry(0).width() + 16
layout = self.toasts.layout()
height = (
layout.totalHeightForWidth(width)
if layout.hasHeightForWidth()
else layout.totalSizeHint().height()
)
rect = self.toast_geometry(height - 16)
self.toasts.setGeometry(rect.adjusted(-8, -8, 8, 8) & self.window.rect())
self.toasts.raise_()
def _destroy_views(self) -> None:
for widget in (self.toasts, self.drawer, self.bell):
widget.hide()
widget.deleteLater()
@@ -0,0 +1,962 @@
"""Toolkit-independent core of the reworked notification and error UX.
The QML and QWidget versions of the notification UI share everything that is not drawing:
- :class:`NotificationHub` is the application-wide history of notifications. BEC alarms,
errors caught by ``SafeSlot(popup_error=True)``, uncaught exceptions and calls to
:meth:`NotificationHub.notify` all end up here. Repeated messages are folded into one entry
with a counter instead of flooding the screen.
- :class:`ToastQueue` decides which entries are shown as short toasts: at most a few at a time,
auto-dismissed after a severity dependent time (paused while hovered) except for critical
errors, which stay until they are acknowledged.
- :class:`NotificationHostBase` attaches the UI to a window: a toast stack in the bottom right
corner, a history drawer on the right edge and a bell button with an unread badge for the
status bar. Subclasses only create and place their views and render :meth:`view_state`.
"""
from __future__ import annotations
import time
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any
from uuid import uuid4
import shiboken6
from bec_lib.endpoints import MessageEndpoints
from bec_lib.logger import bec_logger
from bec_lib.messages import ErrorInfo
from qtpy.QtCore import QEvent, QObject, QRect, Qt, QThread, QTimer, Signal
from qtpy.QtGui import QColor
from qtpy.QtWidgets import QApplication, QMainWindow, QStatusBar, QWidget
from bec_widgets.utils.quick.host import ThemeTokens, _blend
logger = bec_logger.logger
class Severity(str, Enum):
"""Severity of a notification, from least to most serious."""
INFO = "info"
SUCCESS = "success"
WARNING = "warning"
ERROR = "error"
CRITICAL = "critical"
# label, Material icon, toast lifetime in ms (0 = until acknowledged) and rank for sorting
SEVERITY_META: dict[Severity, dict[str, Any]] = {
Severity.INFO: {"label": "Info", "icon": "info", "lifetime": 4000, "rank": 0},
Severity.SUCCESS: {"label": "Done", "icon": "check_circle", "lifetime": 4000, "rank": 1},
Severity.WARNING: {"label": "Warning", "icon": "warning", "lifetime": 8000, "rank": 2},
Severity.ERROR: {"label": "Error", "icon": "error", "lifetime": 12000, "rank": 3},
Severity.CRITICAL: {"label": "Critical", "icon": "dangerous", "lifetime": 0, "rank": 4},
}
# legacy banner kinds and BEC alarm levels mapped onto the new severities
_LEGACY_KINDS = {
"info": Severity.INFO,
"warning": Severity.WARNING,
"minor": Severity.ERROR,
"major": Severity.CRITICAL,
}
_ALARM_LEVELS = {0: Severity.WARNING, 1: Severity.ERROR, 2: Severity.CRITICAL}
# history filters: key -> (label, severities)
FILTERS: dict[str, tuple[str, tuple[Severity, ...]]] = {
"all": ("All", tuple(Severity)),
"errors": ("Errors", (Severity.ERROR, Severity.CRITICAL)),
"warnings": ("Warnings", (Severity.WARNING,)),
"info": ("Info", (Severity.INFO, Severity.SUCCESS)),
}
def to_severity(value: Any) -> Severity:
"""Normalise a severity given as :class:`Severity`, legacy banner kind or alarm level.
Args:
value: ``Severity``, its value, a legacy kind (``info``/``warning``/``minor``/``major``)
or an integer BEC alarm level (0, 1, 2).
Returns:
Severity: The matching severity; unknown values become ``WARNING``.
"""
if isinstance(value, Severity):
return value
raw = getattr(value, "value", value)
if isinstance(raw, str):
raw = raw.lower()
if raw in _LEGACY_KINDS:
return _LEGACY_KINDS[raw]
try:
return Severity(raw)
except ValueError:
return Severity.WARNING
if isinstance(raw, int) and not isinstance(raw, bool):
return _ALARM_LEVELS.get(raw, Severity.WARNING)
return Severity.WARNING
def severity_color(tokens: ThemeTokens, severity: Severity) -> QColor:
"""Theme colour used for a severity: icon, accent stripe and badge."""
if severity == Severity.INFO:
return QColor(tokens.accent)
if severity == Severity.SUCCESS:
return QColor(tokens.success)
if severity == Severity.WARNING:
return QColor(tokens.warning)
if severity == Severity.ERROR:
# orange: between warning and critical so the two error levels read differently
return _blend(tokens.warning, tokens.danger, 0.55)
return QColor(tokens.danger)
def split_traceback(text: str) -> tuple[str, str]:
"""Return ``(exception type, message)`` from the last line of a formatted traceback."""
lines = [line.strip() for line in (text or "").strip().splitlines() if line.strip()]
if not lines:
return "Error", ""
last = lines[-1]
head, sep, tail = last.partition(":")
if sep and head and " " not in head.strip():
return head.strip().rsplit(".", 1)[-1], tail.strip()
return "Error", last
def relative_time(timestamp: float, now: float | None = None) -> str:
"""Short human readable age of ``timestamp`` such as ``2 min ago`` or ``13:04``."""
now = time.time() if now is None else now
seconds = max(0, int(now - timestamp))
if seconds < 30:
return "just now"
if seconds < 3600:
return f"{max(1, seconds // 60)} min ago"
stamp = datetime.fromtimestamp(timestamp)
if stamp.date() == datetime.fromtimestamp(now).date():
return stamp.strftime("%H:%M")
return stamp.strftime("%b %d, %H:%M")
def absolute_time(timestamp: float) -> str:
"""Full local timestamp."""
return datetime.fromtimestamp(timestamp).strftime("%Y-%m-%d %H:%M:%S")
@dataclass
class NotificationEntry:
"""One notification in the history."""
id: str
severity: Severity
title: str
body: str = ""
details: str = ""
source: str = ""
scan_number: int | None = None
created: float = field(default_factory=time.time)
updated: float = field(default_factory=time.time)
count: int = 1
read: bool = False
acknowledged: bool = False
@property
def needs_ack(self) -> bool:
"""A critical entry stays pinned and on screen until somebody acknowledges it."""
return self.severity == Severity.CRITICAL and not self.acknowledged
def copy_text(self) -> str:
"""Plain-text report used by the Copy action, ready to paste into a logbook or issue."""
lines = [f"[{SEVERITY_META[self.severity]['label']}] {self.title}"]
lines.append(f"Time: {absolute_time(self.updated)}")
if self.count > 1:
lines.append(f"Occurrences: {self.count} (first at {absolute_time(self.created)})")
if self.source:
lines.append(f"Source: {self.source}")
if self.scan_number is not None:
lines.append(f"Scan: #{self.scan_number}")
if self.body:
lines.extend(["", self.body])
if self.details:
lines.extend(["", self.details])
return "\n".join(lines)
class NotificationHub(QObject):
"""Application-wide notification history shared by every window and both UI versions.
Signals:
posted(str): A notification was added, or a repeated one was folded into an entry.
changed(): Anything in the history changed.
scan_started(): A new scan opened; hosts hide the non-critical toasts.
"""
MAX_ENTRIES = 500
DEDUP_WINDOW_S = 120.0
posted = Signal(str)
changed = Signal()
scan_started = Signal()
_post_requested = Signal(dict)
_instance: NotificationHub | None = None
def __init__(self, parent: QObject | None = None):
super().__init__(parent)
self._entries: list[NotificationEntry] = []
self._hosts = 0
self._dispatcher = None
self._post_requested.connect(self._post, Qt.ConnectionType.QueuedConnection)
@classmethod
def instance(cls) -> NotificationHub:
"""Return the hub of this application, creating it on first use."""
inst = cls._instance
if inst is None or not shiboken6.isValid(inst):
inst = cls(QApplication.instance())
cls._instance = inst
return inst
@classmethod
def reset_instance(cls) -> None:
"""Drop the hub, e.g. between tests. Its BEC subscriptions are disconnected."""
inst = cls._instance
cls._instance = None
if inst is not None and shiboken6.isValid(inst):
inst.disconnect_bec()
shiboken6.delete(inst)
# ------------------------------------------------------------------ public API
def notify(
self,
title: str,
body: str = "",
severity: Severity | str | int = Severity.INFO,
*,
details: str = "",
source: str = "",
scan_number: int | None = None,
entry_id: str | None = None,
) -> str:
"""Post a notification. Safe to call from any thread.
Args:
title(str): Short headline, e.g. ``Motor samx not reachable``.
body(str): One or two sentences for the toast and the history row.
severity(Severity | str | int): Severity, legacy kind or alarm level.
details(str): Optional long text such as a traceback, shown in the details view.
source(str): Where it came from, e.g. ``Scan server`` or a widget name.
scan_number(int | None): Scan the notification refers to.
entry_id(str | None): Stable id; a second post with the same id is ignored.
Returns:
str: The id of the (possibly folded) entry.
"""
payload = {
"title": str(title or "Notification"),
"body": str(body or ""),
"severity": to_severity(severity),
"details": str(details or ""),
"source": str(source or ""),
"scan_number": scan_number,
"entry_id": entry_id or uuid4().hex,
}
if QThread.currentThread() is not self.thread():
self._post_requested.emit(payload)
return payload["entry_id"]
return self._post(payload)
def report_exception(self, title: str, traceback_text: str, source: str = "") -> str:
"""Post an error from a formatted traceback; used instead of a modal error dialog."""
exc_type, message = split_traceback(traceback_text)
return self.notify(
exc_type,
message or "An unexpected error occurred.",
Severity.ERROR,
details=traceback_text.strip(),
source=source or title,
)
def post_alarm(self, msg: dict, meta: dict | None = None) -> str | None:
"""Translate a BEC alarm message into a notification.
Args:
msg(dict): Alarm content with ``severity``, ``alarm_type``, ``msg`` and ``info``.
meta(dict | None): Message metadata, may hold ``scan_id`` and ``scan_number``.
Returns:
str | None: Id of the entry, or None if the alarm was already posted.
"""
msg = msg or {}
meta = meta or {}
info = msg.get("info")
alarm_id = info.get("id") if isinstance(info, dict) else getattr(info, "id", None)
if isinstance(info, dict):
try:
info = ErrorInfo(**info)
except Exception: # pylint: disable=broad-except
# incomplete payload: keep what is there
info = type("PartialInfo", (), dict(info))()
if alarm_id and self.get(alarm_id) is not None:
return None
trace = getattr(info, "error_message", None) or msg.get("msg") or ""
compact = getattr(info, "compact_error_message", None)
exc_type, last_line = split_traceback(trace)
title = msg.get("alarm_type") or getattr(info, "exception_type", None) or exc_type
body = compact or last_line or str(msg.get("msg") or "")
details = trace if trace.strip() and trace.strip() != body.strip() else ""
if meta.get("scan_id") and details:
details = f"Scan id: {meta['scan_id']}\n\n{details}"
device = getattr(info, "device", None)
source = f"Device {device}" if device else "BEC server"
return self.notify(
title,
body,
msg.get("severity", 0),
details=details,
source=source,
scan_number=meta.get("scan_number"),
entry_id=alarm_id,
)
@property
def entries(self) -> list[NotificationEntry]:
"""History, newest first."""
return list(self._entries)
def get(self, entry_id: str) -> NotificationEntry | None:
"""Return the entry with ``entry_id`` or None."""
for entry in self._entries:
if entry.id == entry_id:
return entry
return None
def acknowledge(self, entry_id: str) -> None:
"""Acknowledge a (critical) entry; it is unpinned and its toast closes."""
entry = self.get(entry_id)
if entry is None or entry.acknowledged:
return
entry.acknowledged = True
entry.read = True
self.changed.emit()
def remove(self, entry_id: str) -> None:
"""Remove one entry from the history."""
entry = self.get(entry_id)
if entry is not None:
self._entries.remove(entry)
self.changed.emit()
def clear(self) -> None:
"""Clear the history but keep critical errors nobody has acknowledged yet."""
kept = [entry for entry in self._entries if entry.needs_ack]
if len(kept) != len(self._entries):
self._entries = kept
self.changed.emit()
def mark_read(self, entry_ids: list[str] | None = None) -> None:
"""Mark the given entries, or all of them, as read."""
touched = False
for entry in self._entries:
if not entry.read and (entry_ids is None or entry.id in entry_ids):
entry.read = True
touched = True
if touched:
self.changed.emit()
def unread_count(self) -> int:
"""Number of unread entries."""
return sum(1 for entry in self._entries if not entry.read)
def top_unread_severity(self) -> Severity | None:
"""Most serious severity among unread or unacknowledged entries."""
open_entries = [e for e in self._entries if not e.read or e.needs_ack]
if not open_entries:
return None
return max(open_entries, key=lambda e: SEVERITY_META[e.severity]["rank"]).severity
# ------------------------------------------------------------------ hosts and BEC feed
def register_host(self) -> None:
"""Called by a host when it attaches; errors are then routed here, not to dialogs."""
self._hosts += 1
def unregister_host(self) -> None:
"""Called by a host when it detaches."""
self._hosts = max(0, self._hosts - 1)
@property
def has_hosts(self) -> bool:
"""Whether any window currently shows notifications."""
return self._hosts > 0
def connect_bec(self, dispatcher) -> None:
"""Subscribe to BEC alarms and scan status once per application."""
if self._dispatcher is not None or dispatcher is None:
return
self._dispatcher = dispatcher
dispatcher.connect_slot(self._on_alarm, MessageEndpoints.alarm())
dispatcher.connect_slot(self._on_scan_status, MessageEndpoints.scan_status())
def disconnect_bec(self) -> None:
"""Undo :meth:`connect_bec`."""
dispatcher, self._dispatcher = self._dispatcher, None
if dispatcher is None:
return
try:
dispatcher.disconnect_slot(self._on_alarm, MessageEndpoints.alarm())
dispatcher.disconnect_slot(self._on_scan_status, MessageEndpoints.scan_status())
except Exception as exc: # pylint: disable=broad-except
logger.warning(f"NotificationHub could not disconnect from BEC: {exc}")
def _on_alarm(self, msg: dict, meta: dict) -> None:
try:
self.post_alarm(msg, meta)
except Exception as exc: # pylint: disable=broad-except
logger.error(f"Could not turn alarm into a notification: {exc}")
def _on_scan_status(self, msg: dict, _meta: dict) -> None:
if (msg or {}).get("status") == "open":
self.scan_started.emit()
# ------------------------------------------------------------------ internals
def _post(self, payload: dict) -> str:
entry_id = payload["entry_id"]
if self.get(entry_id) is not None:
return entry_id
now = time.time()
for entry in self._entries:
if now - entry.updated > self.DEDUP_WINDOW_S:
break
if (
entry.severity == payload["severity"]
and entry.title == payload["title"]
and entry.body == payload["body"]
and entry.source == payload["source"]
):
entry.count += 1
entry.updated = now
entry.read = False
entry.acknowledged = False
entry.details = payload["details"] or entry.details
self._entries.remove(entry)
self._entries.insert(0, entry)
self.changed.emit()
self.posted.emit(entry.id)
return entry.id
entry = NotificationEntry(
id=entry_id,
severity=payload["severity"],
title=payload["title"],
body=payload["body"],
details=payload["details"],
source=payload["source"],
scan_number=payload["scan_number"],
created=now,
updated=now,
)
self._entries.insert(0, entry)
self._trim()
self.changed.emit()
self.posted.emit(entry.id)
return entry.id
def _trim(self) -> None:
excess = len(self._entries) - self.MAX_ENTRIES
if excess <= 0:
return
for entry in reversed(list(self._entries)):
if excess <= 0:
break
if not entry.needs_ack:
self._entries.remove(entry)
excess -= 1
class ToastQueue(QObject):
"""Decides which notifications are on screen as toasts and for how long.
At most :attr:`MAX_VISIBLE` toasts are shown; newer ones replace the oldest non-critical
toast and the rest are counted as overflow ("+3 more"), which stays reachable in the
history. Hovering the stack pauses every countdown.
Signals:
changed(): The visible toasts or the overflow count changed.
"""
MAX_VISIBLE = 3
changed = Signal()
def __init__(self, hub: NotificationHub, parent: QObject | None = None):
super().__init__(parent)
self._hub = hub
self._visible: list[str] = [] # oldest first
self._timers: dict[str, QTimer] = {}
self._remaining: dict[str, int] = {}
self._started: dict[str, float] = {}
self._paused = False
self._suspended = False
self.overflow = 0
hub.posted.connect(self._on_posted)
hub.changed.connect(self._on_hub_changed)
hub.scan_started.connect(self.dismiss_transient)
@property
def visible_ids(self) -> list[str]:
"""Ids of the toasts on screen, oldest first."""
return list(self._visible)
def set_suspended(self, suspended: bool) -> None:
"""Hide all toasts while the history drawer is open; they stay in the history."""
self._suspended = suspended
if suspended:
self.dismiss_all(keep_critical=False)
def set_paused(self, paused: bool) -> None:
"""Pause or resume all countdowns, e.g. while the pointer is over the stack."""
if paused == self._paused:
return
self._paused = paused
now = time.monotonic()
for entry_id, timer in self._timers.items():
if paused:
elapsed = int((now - self._started[entry_id]) * 1000)
self._remaining[entry_id] = max(500, self._remaining[entry_id] - elapsed)
timer.stop()
else:
self._started[entry_id] = now
timer.start(self._remaining[entry_id])
def dismiss(self, entry_id: str) -> None:
"""Close one toast; the entry stays in the history."""
if entry_id not in self._visible:
return
self._visible.remove(entry_id)
self._stop_timer(entry_id)
self.changed.emit()
def dismiss_transient(self) -> None:
"""Close all toasts except unacknowledged critical errors (used when a scan starts)."""
self.dismiss_all(keep_critical=True)
def dismiss_all(self, keep_critical: bool = True) -> None:
"""Close all toasts, optionally keeping the critical ones."""
keep = []
for entry_id in self._visible:
entry = self._hub.get(entry_id)
if keep_critical and entry is not None and entry.needs_ack:
keep.append(entry_id)
else:
self._stop_timer(entry_id)
changed = keep != self._visible or self.overflow
self._visible = keep
self.overflow = 0
if changed:
self.changed.emit()
def cleanup(self) -> None:
"""Stop all timers and disconnect from the hub."""
for entry_id in list(self._timers):
self._stop_timer(entry_id)
for signal, slot in (
(self._hub.posted, self._on_posted),
(self._hub.changed, self._on_hub_changed),
(self._hub.scan_started, self.dismiss_transient),
):
try:
signal.disconnect(slot)
except (RuntimeError, TypeError):
pass
# ------------------------------------------------------------------ internals
def _on_posted(self, entry_id: str) -> None:
if self._suspended:
return
entry = self._hub.get(entry_id)
if entry is None:
return
if entry_id in self._visible:
# a repeat: move it to the newest position and restart its countdown
self._visible.remove(entry_id)
self._visible.append(entry_id)
self._start_timer(entry)
self.changed.emit()
return
if len(self._visible) >= self.MAX_VISIBLE:
victim = next(
(i for i in self._visible if not getattr(self._hub.get(i), "needs_ack", False)),
None,
)
if victim is None and not entry.needs_ack:
self.overflow += 1
self.changed.emit()
return
victim = victim or self._visible[0]
self._visible.remove(victim)
self._stop_timer(victim)
self.overflow += 1
self._visible.append(entry_id)
self._start_timer(entry)
self.changed.emit()
def _on_hub_changed(self) -> None:
gone = [
i
for i in self._visible
if (entry := self._hub.get(i)) is None
or (entry.severity == Severity.CRITICAL and entry.acknowledged)
]
for entry_id in gone:
self._visible.remove(entry_id)
self._stop_timer(entry_id)
if gone:
self.changed.emit()
def _start_timer(self, entry: NotificationEntry) -> None:
self._stop_timer(entry.id)
lifetime = SEVERITY_META[entry.severity]["lifetime"]
if lifetime <= 0:
return
timer = QTimer(self)
timer.setSingleShot(True)
timer.timeout.connect(lambda entry_id=entry.id: self.dismiss(entry_id))
self._timers[entry.id] = timer
self._remaining[entry.id] = lifetime
self._started[entry.id] = time.monotonic()
if not self._paused:
timer.start(lifetime)
def _stop_timer(self, entry_id: str) -> None:
timer = self._timers.pop(entry_id, None)
self._remaining.pop(entry_id, None)
self._started.pop(entry_id, None)
if timer is not None:
timer.stop()
timer.deleteLater()
def entry_row(entry: NotificationEntry, tokens: ThemeTokens, now: float | None = None) -> dict:
"""Flat dictionary describing one entry, consumed by both views."""
meta = SEVERITY_META[entry.severity]
color = severity_color(tokens, entry.severity)
subtitle = [part for part in (entry.source, relative_time(entry.updated, now)) if part]
if entry.scan_number is not None:
subtitle.insert(1 if entry.source else 0, f"Scan #{entry.scan_number}")
return {
"id": entry.id,
"entryId": entry.id,
"severity": entry.severity.value,
"severityLabel": meta["label"],
"icon": meta["icon"],
"color": color.name(),
"tint": tokens.soft(color, 0.14).name(),
"title": entry.title,
"body": entry.body,
"details": entry.details,
"hasDetails": bool(entry.details),
"source": entry.source,
"meta": " · ".join(subtitle),
"time": relative_time(entry.updated, now),
"timeAbs": absolute_time(entry.updated),
"firstAbs": absolute_time(entry.created),
"scan": "" if entry.scan_number is None else f"#{entry.scan_number}",
"count": entry.count,
"unread": not entry.read,
"needsAck": entry.needs_ack,
"critical": entry.severity == Severity.CRITICAL,
}
class NotificationHostBase(QObject):
"""Attach the notification UI (toasts, history drawer, bell) to a window.
Subclasses implement :meth:`_create_views`, :meth:`_sync_views` and :meth:`_place_views`.
Args:
window(QWidget): Window to overlay, usually a ``QMainWindow``.
status_bar(QStatusBar | None): Where to put the bell; defaults to the window's status
bar if it is a ``QMainWindow``. Pass ``None`` on a plain widget to skip the bell.
dispatcher: ``BECDispatcher`` used to receive BEC alarms; optional.
"""
TOAST_WIDTH = 380
DRAWER_WIDTH = 400
MARGIN = 16
def __init__(self, window: QWidget, status_bar: QStatusBar | None = None, dispatcher=None):
super().__init__(window)
self.window = window
self.hub = NotificationHub.instance()
self.hub.register_host()
self.hub.connect_bec(dispatcher)
self.queue = ToastQueue(self.hub, self)
self.drawer_open = False
self.filter_key = "all"
self.search = ""
self.selected_id: str | None = None
self._sync_pending = False
if status_bar is None and isinstance(window, QMainWindow):
status_bar = window.statusBar()
self.status_bar = status_bar
self._create_views()
self.hub.changed.connect(self.sync)
self.queue.changed.connect(self.sync)
self._tick = QTimer(self)
self._tick.setInterval(30_000)
self._tick.timeout.connect(self.sync)
self._tick.start()
app = QApplication.instance()
theme = getattr(app, "theme", None)
if theme is not None:
theme.theme_changed.connect(self._on_theme_changed)
window.installEventFilter(self)
self._cleaned = False
self.sync_now()
# ------------------------------------------------------------------ actions
def toggle_drawer(self) -> None:
"""Open or close the history drawer."""
self.set_drawer_open(not self.drawer_open)
def set_drawer_open(self, open_: bool) -> None:
"""Open or close the history drawer. Opening marks everything as read."""
self.drawer_open = bool(open_)
self.queue.set_suspended(self.drawer_open)
if self.drawer_open:
self.hub.mark_read()
else:
self.selected_id = None
self.sync()
def open_details(self, entry_id: str) -> None:
"""Open the drawer on the details of one entry (the toast's Details action)."""
self.selected_id = entry_id
self.set_drawer_open(True)
def select(self, entry_id: str | None) -> None:
"""Show the details of ``entry_id``; None goes back to the list."""
self.selected_id = entry_id or None
self.sync()
def set_filter(self, key: str) -> None:
"""Filter the history by one of :data:`FILTERS`."""
self.filter_key = key if key in FILTERS else "all"
self.sync()
def set_search(self, text: str) -> None:
"""Filter the history by text in title, message, source or details."""
self.search = text or ""
self.sync()
def acknowledge(self, entry_id: str) -> None:
"""Acknowledge a critical error."""
self.hub.acknowledge(entry_id)
def remove(self, entry_id: str) -> None:
"""Remove an entry from the history."""
if self.selected_id == entry_id:
self.selected_id = None
self.queue.dismiss(entry_id)
self.hub.remove(entry_id)
def dismiss_toast(self, entry_id: str) -> None:
"""Close a toast without touching the history."""
entry = self.hub.get(entry_id)
if entry is not None and entry.needs_ack:
self.hub.acknowledge(entry_id)
self.queue.dismiss(entry_id)
if entry is not None:
self.hub.mark_read([entry_id])
def copy_details(self, entry_id: str) -> bool:
"""Copy a plain-text report of the entry to the clipboard."""
entry = self.hub.get(entry_id)
if entry is None:
return False
QApplication.clipboard().setText(entry.copy_text())
return True
def mark_all_read(self) -> None:
"""Mark the whole history as read."""
self.hub.mark_read()
def clear_history(self) -> None:
"""Clear the history; unacknowledged critical errors stay."""
self.selected_id = None
self.hub.clear()
def set_toasts_hovered(self, hovered: bool) -> None:
"""Pause the toast countdowns while the pointer is over them."""
self.queue.set_paused(hovered)
# ------------------------------------------------------------------ state
def filtered_entries(self) -> list[NotificationEntry]:
"""History entries matching the filter and search, critical ones pinned on top."""
severities = FILTERS[self.filter_key][1]
needle = self.search.strip().lower()
rows = []
for entry in self.hub.entries:
if entry.severity not in severities:
continue
if needle and not any(
needle in text.lower()
for text in (entry.title, entry.body, entry.source, entry.details)
):
continue
rows.append(entry)
rows.sort(key=lambda e: not e.needs_ack) # stable: keeps newest first otherwise
return rows
def view_state(self) -> dict:
"""Everything the views need to render, as plain Python data."""
tokens = ThemeTokens()
now = time.time()
entries = self.hub.entries
counts = {
key: sum(1 for e in entries if e.severity in severities)
for key, (_label, severities) in FILTERS.items()
}
toasts = []
if not self.drawer_open:
for entry_id in self.queue.visible_ids:
entry = self.hub.get(entry_id)
if entry is not None:
toasts.append(entry_row(entry, tokens, now))
selected = self.hub.get(self.selected_id) if self.selected_id else None
top = self.hub.top_unread_severity()
unread = self.hub.unread_count()
open_criticals = sum(1 for e in entries if e.needs_ack)
if open_criticals:
bell_tip = f"{open_criticals} critical error(s) need attention"
elif unread:
bell_tip = f"{unread} unread notification(s)"
else:
bell_tip = "Notifications"
return {
"drawerOpen": self.drawer_open,
"toasts": toasts,
"overflow": self.queue.overflow if toasts else 0,
# the list is only built while the drawer is visible
"rows": (
[entry_row(e, tokens, now) for e in self.filtered_entries()]
if self.drawer_open
else []
),
"filters": [
{"key": key, "label": label, "count": counts[key], "active": key == self.filter_key}
for key, (label, _sev) in FILTERS.items()
],
"filterKey": self.filter_key,
"search": self.search,
"total": len(entries),
"selected": entry_row(selected, tokens, now) if selected is not None else None,
"unread": unread,
"openCriticals": open_criticals,
"bellColor": (
severity_color(tokens, top).name() if top is not None else tokens.fg_muted.name()
),
"bellIcon": "notifications_active" if unread or open_criticals else "notifications",
"bellTip": bell_tip,
}
def sync(self, *_args) -> None:
"""Schedule a view update; bursts of changes are folded into one update."""
if self._cleaned or self._sync_pending:
return
self._sync_pending = True
QTimer.singleShot(0, self.sync_now)
def sync_now(self) -> None:
"""Push the current state into the views and re-place them right away."""
self._sync_pending = False
if self._cleaned:
return
state = self.view_state()
self._sync_views(state)
self._place_views()
def content_rect(self) -> QRect:
"""Area of the window the overlays may cover: below menu and tool bars, above status."""
rect = self.window.rect()
top = 0
bottom = rect.bottom()
if isinstance(self.window, QMainWindow):
menu = self.window.menuWidget()
if menu is not None and menu.isVisible():
top = menu.geometry().bottom() + 1
central = self.window.centralWidget()
if central is not None:
top = max(top, central.geometry().top())
status = self.window.statusBar() if self.window.statusBar() else None
if status is not None and status.isVisible():
bottom = status.geometry().top() - 1
return QRect(rect.left(), top, rect.width(), max(0, bottom - top + 1))
def toast_geometry(self, content_height: int) -> QRect:
"""Bottom-right rectangle for a toast stack of ``content_height`` pixels."""
area = self.content_rect()
width = min(self.TOAST_WIDTH, max(200, area.width() - 2 * self.MARGIN))
height = min(content_height, max(0, area.height() - 2 * self.MARGIN))
return QRect(
area.right() - self.MARGIN - width + 1,
area.bottom() - self.MARGIN - height + 1,
width,
height,
)
def drawer_geometry(self) -> QRect:
"""Right-edge rectangle of the history drawer."""
area = self.content_rect()
width = min(self.DRAWER_WIDTH, area.width())
return QRect(area.right() - width + 1, area.top(), width, area.height())
# ------------------------------------------------------------------ plumbing
def eventFilter(self, watched, event): # pylint: disable=invalid-name
if watched is self.window and event.type() in (QEvent.Type.Resize, QEvent.Type.Show):
self._place_views()
return False
def _on_theme_changed(self, *_args) -> None:
self._apply_theme()
self.sync()
def cleanup(self) -> None:
"""Detach from the window and the hub."""
if self._cleaned:
return
self._cleaned = True
self._tick.stop()
self.queue.cleanup()
for signal, slot in ((self.hub.changed, self.sync), (self.queue.changed, self.sync)):
try:
signal.disconnect(slot)
except (RuntimeError, TypeError):
pass
app = QApplication.instance()
theme = getattr(app, "theme", None)
if theme is not None:
try:
theme.theme_changed.disconnect(self._on_theme_changed)
except (RuntimeError, TypeError):
pass
if shiboken6.isValid(self.window):
self.window.removeEventFilter(self)
self.hub.unregister_host()
self._destroy_views()
# ------------------------------------------------------------------ for subclasses
def _create_views(self) -> None:
raise NotImplementedError
def _sync_views(self, state: dict) -> None:
raise NotImplementedError
def _place_views(self) -> None:
raise NotImplementedError
def _apply_theme(self) -> None:
"""Re-style the views after a theme change; QML follows the theme by itself."""
def _destroy_views(self) -> None:
"""Release the views."""
@@ -0,0 +1,54 @@
import QtQuick
import QtQuick.Controls.Basic
import BecUi
// Status-bar bell. The badge counts unread notifications and takes the colour of the most
// serious one; a critical error that is not acknowledged keeps the bell coloured.
AbstractButton {
id: root
required property var backend
readonly property var st: backend.state
readonly property int unread: st.unread || 0
readonly property bool alert: unread > 0 || (st.openCriticals || 0) > 0
implicitWidth: 36
implicitHeight: 24
hoverEnabled: true
checked: st.drawerOpen || false
Accessible.name: "Notifications"
onClicked: backend.toggleDrawer()
background: Rectangle {
radius: 6
color: root.down ? theme.pressed : (root.hovered || root.checked ? theme.hover : "transparent")
}
contentItem: Item {
Icon {
anchors.centerIn: parent
name: root.st.bellIcon || "notifications"
color: root.alert ? (root.st.bellColor || theme.fgMuted) : theme.fgMuted
size: 18
}
}
Rectangle {
visible: root.unread > 0
anchors.right: parent.right
anchors.top: parent.top
anchors.margins: 1
height: 14
width: Math.max(14, badgeText.implicitWidth + 8)
radius: 7
color: root.st.bellColor || theme.primary
Text {
id: badgeText
anchors.centerIn: parent
text: root.unread > 99 ? "99+" : root.unread
color: "white"
font.pixelSize: 9
font.weight: Font.Bold
}
}
ToolTip.visible: hovered
ToolTip.delay: 400
ToolTip.text: st.bellTip || "Notifications"
}
@@ -0,0 +1,414 @@
import QtQuick
import QtQuick.Controls.Basic
import QtQuick.Layouts
import BecUi
// Right-edge history drawer: filter chips, search, the list of notifications and the
// details of one entry with its full text, metadata and actions.
Rectangle {
id: root
required property var backend
readonly property var st: backend.state
readonly property var sel: st.selected || ({})
readonly property bool hasSelection: sel.entryId !== undefined
color: theme.bg
// shadow towards the content and a hairline border
Rectangle {
anchors.right: parent.left
width: 8; height: parent.height
gradient: Gradient {
orientation: Gradient.Horizontal
GradientStop { position: 0.0; color: "transparent" }
GradientStop { position: 1.0; color: Qt.rgba(0, 0, 0, theme.dark ? 0.3 : 0.08) }
}
}
Rectangle { anchors.left: parent.left; width: 1; height: parent.height; color: theme.border }
ColumnLayout {
anchors.fill: parent
anchors.leftMargin: 1
spacing: 0
// ---------------------------------------------------------------- header
ColumnLayout {
Layout.fillWidth: true
Layout.leftMargin: 16
Layout.rightMargin: 12
Layout.topMargin: 14
Layout.bottomMargin: 8
spacing: 10
RowLayout {
Layout.fillWidth: true
spacing: 4
ColumnLayout {
Layout.fillWidth: true
spacing: 0
Text {
Layout.fillWidth: true
text: "Notifications"
color: theme.fg
font.pixelSize: 15
font.weight: Font.DemiBold
}
Text {
Layout.fillWidth: true
text: (root.st.total || 0) + " in history"
+ ((root.st.openCriticals || 0) > 0 ? " · " + root.st.openCriticals + " critical open" : "")
color: theme.fgSubtle
font.pixelSize: 11
}
}
IconButton { iconName: "done_all"; tip: "Mark all as read"; onClicked: root.backend.markAllRead() }
IconButton { iconName: "clear_all"; tip: "Clear history"; onClicked: root.backend.clearHistory() }
IconButton { iconName: "close"; tip: "Close"; onClicked: root.backend.setDrawerOpen(false) }
}
Flow {
Layout.fillWidth: true
visible: !root.hasSelection
spacing: 6
Repeater {
model: root.st.filters || []
delegate: AbstractButton {
id: chip
required property var modelData
implicitHeight: 26
implicitWidth: chipLabel.implicitWidth + 22
hoverEnabled: true
onClicked: root.backend.setFilter(modelData.key)
background: Rectangle {
radius: 13
color: chip.modelData.active ? Qt.rgba(theme.primary.r, theme.primary.g, theme.primary.b, 0.18)
: chip.hovered ? theme.hover : "transparent"
border.color: chip.modelData.active ? theme.primary : theme.border
}
contentItem: Text {
id: chipLabel
text: chip.modelData.label + " " + chip.modelData.count
color: chip.modelData.active ? theme.fg : theme.fgMuted
font.pixelSize: 12
font.weight: chip.modelData.active ? Font.DemiBold : Font.Normal
horizontalAlignment: Text.AlignHCenter
verticalAlignment: Text.AlignVCenter
}
}
}
}
InputField {
id: search
Layout.fillWidth: true
visible: !root.hasSelection
placeholderText: "Search notifications"
text: root.st.search || ""
onTextEdited: root.backend.setSearch(text)
}
}
// ---------------------------------------------------------------- list
ListView {
id: list
Layout.fillWidth: true
Layout.fillHeight: true
visible: !root.hasSelection && count > 0
clip: true
model: root.backend.rows
spacing: 0
boundsBehavior: Flickable.StopAtBounds
ScrollBar.vertical: ScrollBar { policy: ScrollBar.AsNeeded }
delegate: Item {
id: row
required property string entryId
required property string icon
required property color color
required property color tint
required property string title
required property string body
required property string time
required property string timeAbs
required property string severityLabel
required property string source
required property string scan
required property int count
required property bool unread
required property bool needsAck
width: ListView.view.width
height: 78
Rectangle {
anchors.fill: parent
anchors.leftMargin: 8
anchors.rightMargin: 8
anchors.topMargin: 3
anchors.bottomMargin: 3
radius: 8
color: row.needsAck ? (rowMouse.containsMouse ? Qt.lighter(row.tint, 1.1) : row.tint)
: (rowMouse.containsMouse ? theme.hover : theme.card)
border.color: row.needsAck ? row.color : theme.border
Rectangle {
x: 10; y: 10
width: 28; height: 28; radius: 14
color: row.needsAck ? theme.card : row.tint
Icon { anchors.centerIn: parent; name: row.icon; color: row.color; filled: true; size: 17 }
}
ColumnLayout {
anchors.fill: parent
anchors.leftMargin: 48
anchors.rightMargin: 10
anchors.topMargin: 6
anchors.bottomMargin: 6
spacing: 2
RowLayout {
Layout.fillWidth: true
spacing: 6
Text {
Layout.fillWidth: true
text: row.title
textFormat: Text.PlainText
color: theme.fg
font.pixelSize: 13
font.weight: Font.DemiBold
elide: Text.ElideRight
}
Rectangle {
visible: row.count > 1
implicitWidth: rowCount.implicitWidth + 12
implicitHeight: 16
radius: 8
color: row.tint
Text {
id: rowCount
anchors.centerIn: parent
text: "×" + row.count
color: row.color
font.pixelSize: 11
}
}
Rectangle {
visible: row.unread
width: 7; height: 7; radius: 3.5
color: theme.primary
}
Text {
text: row.time
color: theme.fgSubtle
font.pixelSize: 11
}
}
Text {
Layout.fillWidth: true
text: row.body.replace(/\n/g, " ")
textFormat: Text.PlainText
color: theme.fgMuted
font.pixelSize: 12
elide: Text.ElideRight
}
Text {
Layout.fillWidth: true
text: (row.needsAck ? "Needs acknowledgement · " : "")
+ [row.severityLabel, row.source, row.scan].filter(function (p) { return p !== "" }).join(" · ")
color: row.needsAck ? row.color : theme.fgSubtle
font.pixelSize: 11
elide: Text.ElideRight
}
}
MouseArea {
id: rowMouse
anchors.fill: parent
hoverEnabled: true
cursorShape: Qt.PointingHandCursor
onClicked: root.backend.select(row.entryId)
ToolTip.visible: containsMouse
ToolTip.delay: 800
ToolTip.text: row.timeAbs
}
}
}
}
// ---------------------------------------------------------------- empty state
Item {
Layout.fillWidth: true
Layout.fillHeight: true
visible: !root.hasSelection && list.count === 0
Column {
anchors.centerIn: parent
spacing: 8
Icon {
anchors.horizontalCenter: parent.horizontalCenter
name: (root.st.total || 0) > 0 ? "filter_alt_off" : "inbox"
color: theme.fgSubtle
size: 32
}
Text {
anchors.horizontalCenter: parent.horizontalCenter
horizontalAlignment: Text.AlignHCenter
text: (root.st.total || 0) > 0 ? "Nothing matches this filter."
: "You're all caught up.\nNew messages and errors appear here."
color: theme.fgSubtle
font.pixelSize: 13
}
}
}
// ---------------------------------------------------------------- details
Flickable {
id: detailsView
Layout.fillWidth: true
Layout.fillHeight: true
visible: root.hasSelection
clip: true
contentHeight: details.implicitHeight + 16
boundsBehavior: Flickable.StopAtBounds
ScrollBar.vertical: ScrollBar { policy: ScrollBar.AsNeeded }
ColumnLayout {
id: details
width: detailsView.width - 24
x: 12
spacing: 10
TextButton {
text: "All notifications"
iconName: "arrow_back"
variant: "ghost"
onClicked: root.backend.select("")
}
RowLayout {
Layout.fillWidth: true
spacing: 12
Rectangle {
Layout.alignment: Qt.AlignTop
width: 36; height: 36; radius: 18
color: root.sel.tint || theme.hover
Icon { anchors.centerIn: parent; name: root.sel.icon || ""; color: root.sel.color || theme.fg; filled: true; size: 21 }
}
ColumnLayout {
Layout.fillWidth: true
spacing: 4
Rectangle {
implicitHeight: 18
implicitWidth: pillText.implicitWidth + 16
radius: 9
color: root.sel.tint || "transparent"
Text {
id: pillText
anchors.centerIn: parent
text: (root.sel.severityLabel || "") + (root.sel.needsAck ? " · needs acknowledgement" : "")
color: root.sel.color || theme.fg
font.pixelSize: 11
font.weight: Font.DemiBold
}
}
TextEdit {
Layout.fillWidth: true
text: root.sel.title || ""
textFormat: TextEdit.PlainText
readOnly: true
selectByMouse: true
wrapMode: TextEdit.Wrap
color: theme.fg
selectionColor: theme.primary
font.pixelSize: 15
font.weight: Font.DemiBold
}
}
}
TextEdit {
Layout.fillWidth: true
text: root.sel.body || "No message."
textFormat: TextEdit.PlainText
readOnly: true
selectByMouse: true
wrapMode: TextEdit.Wrap
color: theme.fg
selectionColor: theme.primary
font.pixelSize: 13
}
GridLayout {
columns: 2
columnSpacing: 12
rowSpacing: 4
Text { text: "When"; color: theme.fgSubtle; font.pixelSize: 12 }
Text { text: root.sel.timeAbs || ""; color: theme.fg; font.pixelSize: 12 }
Text { visible: (root.sel.count || 0) > 1; text: "Repeated"; color: theme.fgSubtle; font.pixelSize: 12 }
Text { visible: (root.sel.count || 0) > 1; text: (root.sel.count || 0) + " times since " + (root.sel.firstAbs || ""); color: theme.fg; font.pixelSize: 12 }
Text { visible: (root.sel.source || "") !== ""; text: "Source"; color: theme.fgSubtle; font.pixelSize: 12 }
Text { visible: (root.sel.source || "") !== ""; text: root.sel.source || ""; color: theme.fg; font.pixelSize: 12 }
Text { visible: (root.sel.scan || "") !== ""; text: "Scan"; color: theme.fgSubtle; font.pixelSize: 12 }
Text { visible: (root.sel.scan || "") !== ""; text: root.sel.scan || ""; color: theme.fg; font.pixelSize: 12 }
}
Text {
visible: root.sel.hasDetails || false
text: "DETAILS"
color: theme.fgSubtle
font.pixelSize: 11
font.weight: Font.DemiBold
}
Rectangle {
Layout.fillWidth: true
Layout.preferredHeight: Math.min(320, traceText.contentHeight + 16)
visible: root.sel.hasDetails || false
radius: 8
color: theme.field
border.color: theme.border
Flickable {
id: traceFlick
anchors.fill: parent
anchors.margins: 8
clip: true
contentWidth: traceText.contentWidth
contentHeight: traceText.contentHeight
boundsBehavior: Flickable.StopAtBounds
ScrollBar.vertical: ScrollBar { policy: ScrollBar.AsNeeded }
ScrollBar.horizontal: ScrollBar { policy: ScrollBar.AsNeeded }
TextEdit {
id: traceText
text: root.sel.details || ""
textFormat: TextEdit.PlainText
readOnly: true
selectByMouse: true
color: theme.fg
selectionColor: theme.primary
font.family: "monospace"
font.pixelSize: 11
}
}
}
RowLayout {
Layout.fillWidth: true
spacing: 6
TextButton {
id: copyReport
property string copiedId: ""
text: copiedId !== "" && copiedId === root.sel.entryId ? "Copied" : "Copy report"
iconName: "content_copy"
onClicked: if (root.backend.copyDetails(root.sel.entryId)) copiedId = root.sel.entryId
}
TextButton {
visible: root.sel.needsAck || false
text: "Acknowledge"
iconName: "done"
variant: "danger"
onClicked: root.backend.acknowledge(root.sel.entryId)
}
Item { Layout.fillWidth: true }
TextButton {
text: "Remove"
iconName: "delete"
variant: "ghost"
onClicked: root.backend.remove(root.sel.entryId)
}
}
}
}
}
}
@@ -0,0 +1,203 @@
import QtQuick
import QtQuick.Layouts
import BecUi
// Bottom-right toast stack. Newest toast at the bottom, nearest to the status bar.
// Hovering the stack pauses every countdown. Critical errors stay until acknowledged.
Item {
id: root
required property var backend
readonly property var st: backend.state
implicitWidth: 396
implicitHeight: stack.implicitHeight + 16
HoverHandler {
onHoveredChanged: root.backend.setHovered(hovered)
}
Column {
id: stack
anchors.left: parent.left
anchors.right: parent.right
anchors.bottom: parent.bottom
anchors.margins: 8
spacing: 10
move: Transition { NumberAnimation { properties: "y"; duration: 140; easing.type: Easing.OutCubic } }
add: Transition {
NumberAnimation { property: "opacity"; from: 0; to: 1; duration: 160 }
NumberAnimation { property: "x"; from: 40; to: 0; duration: 180; easing.type: Easing.OutCubic }
}
Item {
width: stack.width
height: moreButton.visible ? moreButton.implicitHeight : 0
visible: (root.st.overflow || 0) > 0
TextButton {
id: moreButton
anchors.right: parent.right
visible: parent.visible
text: (root.st.overflow || 0) + " more in notifications"
iconName: "expand_less"
background: Rectangle {
radius: 6
color: moreButton.hovered ? theme.hover : theme.card
border.color: theme.border
}
onClicked: root.backend.setDrawerOpen(true)
}
}
Repeater {
model: root.backend.toasts
delegate: Item {
id: toast
required property string entryId
required property string icon
required property color color
required property color tint
required property string title
required property string body
required property string meta
required property string severity
required property int count
required property bool needsAck
required property bool hasDetails
width: stack.width
height: card.height
// soft shadow
Rectangle {
x: 0; y: 3
width: card.width; height: card.height
radius: 11
color: Qt.rgba(0, 0, 0, theme.dark ? 0.35 : 0.07)
}
Rectangle {
id: card
width: parent.width
height: content.implicitHeight + 22
radius: 10
color: theme.card
border.width: toast.needsAck ? 1.5 : 1
border.color: toast.needsAck ? toast.color : theme.border
Rectangle {
x: 0; y: card.radius
width: 4; height: parent.height - 2 * card.radius
radius: 2
color: toast.color
}
RowLayout {
id: content
anchors.left: parent.left
anchors.right: parent.right
anchors.top: parent.top
anchors.leftMargin: 14
anchors.rightMargin: 10
anchors.topMargin: 12
spacing: 12
Rectangle {
Layout.alignment: Qt.AlignTop
width: 32; height: 32; radius: 16
color: toast.tint
Icon { anchors.centerIn: parent; name: toast.icon; color: toast.color; filled: true; size: 19 }
}
ColumnLayout {
Layout.fillWidth: true
spacing: 3
RowLayout {
Layout.fillWidth: true
spacing: 6
Text {
Layout.fillWidth: true
text: toast.title
textFormat: Text.PlainText
color: theme.fg
font.pixelSize: 13
font.weight: Font.DemiBold
elide: Text.ElideRight
}
Rectangle {
visible: toast.count > 1
implicitWidth: countLabel.implicitWidth + 12
implicitHeight: 16
radius: 8
color: toast.tint
Text {
id: countLabel
anchors.centerIn: parent
text: "×" + toast.count
color: toast.color
font.pixelSize: 11
font.weight: Font.Bold
}
}
IconButton {
visible: !toast.needsAck
implicitWidth: 24; implicitHeight: 24
iconName: "close"; iconSize: 16
tip: "Dismiss"
onClicked: root.backend.dismissToast(toast.entryId)
}
}
Text {
Layout.fillWidth: true
visible: toast.body !== ""
text: toast.body
textFormat: Text.PlainText
color: theme.fgMuted
font.pixelSize: 12
wrapMode: Text.Wrap
maximumLineCount: 4
elide: Text.ElideRight
}
Text {
Layout.fillWidth: true
text: toast.meta
color: theme.fgSubtle
font.pixelSize: 11
elide: Text.ElideRight
}
RowLayout {
Layout.fillWidth: true
spacing: 6
visible: copyButton.visible || detailsButton.visible || ackButton.visible
Item { Layout.fillWidth: true }
IconButton {
id: copyButton
visible: toast.severity === "error" || toast.severity === "critical"
implicitWidth: 28; implicitHeight: 28
iconName: "content_copy"; iconSize: 16
tip: "Copy report"
onClicked: root.backend.copyDetails(toast.entryId)
}
TextButton {
id: detailsButton
visible: toast.hasDetails || toast.needsAck
implicitHeight: 28
text: "Details"; iconName: "open_in_new"; variant: "ghost"
onClicked: root.backend.openDetails(toast.entryId)
}
TextButton {
id: ackButton
visible: toast.needsAck
implicitHeight: 28
text: "Acknowledge"; iconName: "done"; variant: "danger"
onClicked: root.backend.acknowledge(toast.entryId)
}
}
}
}
}
}
}
}
}
@@ -64,9 +64,16 @@ class BECMainWindow(BECWidget, QMainWindow):
self.setWindowTitle(window_title)
self.setAttribute(Qt.WidgetAttribute.WA_DeleteOnClose, True)
# Notification Centre overlay
self.notification_centre = NotificationCentre(parent=self) # Notification layer
self.notification_broker = BECNotificationBroker(parent=self)
# Notification Centre overlay. BEC_NOTIFICATION_UI=qml|qwidget opts into the reworked
# toasts and history drawer, which also replace the modal error dialogs.
self._notification_ui = os.environ.get("BEC_NOTIFICATION_UI", "legacy").lower()
self.notifications = None
if self._notification_ui in ("qml", "qwidget"):
self.notification_centre = None
self.notification_broker = None
else:
self.notification_centre = NotificationCentre(parent=self) # Notification layer
self.notification_broker = BECNotificationBroker(parent=self)
self._nc_margin = 16
self._position_notification_centre()
self._widget_hierarchy_dialog: WidgetHierarchyDialog | None = None
@@ -95,7 +102,10 @@ class BECMainWindow(BECWidget, QMainWindow):
during __init__ when we first install ``self._full_content``).
"""
super().setCentralWidget(widget)
self.notification_centre.raise_()
if self.notification_centre is not None:
self.notification_centre.raise_()
if self.notifications is not None:
self.notifications.sync()
self.statusBar().raise_()
def resizeEvent(self, event):
@@ -104,7 +114,7 @@ class BECMainWindow(BECWidget, QMainWindow):
def _position_notification_centre(self):
"""Keep the notification panel at a fixed margin top-right."""
if not hasattr(self, "notification_centre"):
if getattr(self, "notification_centre", None) is None:
return
margin = getattr(self, "_nc_margin", 16) # px
nc = self.notification_centre
@@ -158,6 +168,9 @@ class BECMainWindow(BECWidget, QMainWindow):
"""
Add the notification indicator to the status bar and hook the signals.
"""
if self.notification_centre is None:
self._add_reworked_notifications()
return
# Add the notification indicator to the status bar
self.notification_indicator = NotificationIndicator(self)
self.status_bar.addPermanentWidget(self.notification_indicator)
@@ -168,6 +181,20 @@ class BECMainWindow(BECWidget, QMainWindow):
self.notification_indicator.show_all_requested.connect(self.notification_centre.show_all)
self.notification_indicator.hide_all_requested.connect(self.notification_centre.hide_all)
def _add_reworked_notifications(self):
"""Attach the reworked toasts, history drawer and bell (QML or QWidget version)."""
if self._notification_ui == "qml":
from bec_widgets.widgets.containers.main_window.addons.notification_center.notification_qml import (
NotificationHostQML as Host,
)
else:
from bec_widgets.widgets.containers.main_window.addons.notification_center.notification_qwidget import (
NotificationHostQWidget as Host,
)
self.notifications = Host(self, status_bar=self.status_bar, dispatcher=self.bec_dispatcher)
# the guided tour points at the bell
self.notification_indicator = self.notifications.bell
################################################################################
# Client message status bar widget helpers
@@ -539,6 +566,9 @@ class BECMainWindow(BECWidget, QMainWindow):
self._scan_progress_bar_full.deleteLater()
self._scan_progress_hover.close()
self._scan_progress_hover.deleteLater()
if self.notifications is not None:
self.notifications.cleanup()
self.notifications = None
super().cleanup()
+551
View File
@@ -0,0 +1,551 @@
# pylint: disable=missing-function-docstring,redefined-outer-name,protected-access
import threading
import time
import traceback
from unittest import mock
import pytest
from qtpy.QtWidgets import QApplication, QLabel, QMainWindow, QMessageBox, QWidget
from bec_widgets.utils.error_popups import ErrorPopupUtility, SafeSlot, WarningPopupUtility
from bec_widgets.widgets.containers.main_window.addons.notification_center import (
notification_ux_common as common,
)
from bec_widgets.widgets.containers.main_window.addons.notification_center.notification_qml import (
NotificationHostQML,
)
from bec_widgets.widgets.containers.main_window.addons.notification_center.notification_qwidget import (
NotificationHostQWidget,
)
from bec_widgets.widgets.containers.main_window.addons.notification_center.notification_ux_common import (
NotificationHub,
Severity,
ToastQueue,
)
TRACE = """Traceback (most recent call last):
File "x.py", line 1, in move
raise ValueError("Target 12.5 is outside the limits.")
ValueError: Target 12.5 is outside the limits."""
@pytest.fixture
def hub():
yield NotificationHub.instance()
NotificationHub.reset_instance()
@pytest.fixture
def window(qtbot):
win = QMainWindow()
win.setCentralWidget(QLabel("content"))
win.statusBar().addWidget(QLabel("status"))
win.resize(1000, 700)
qtbot.addWidget(win)
win.show()
qtbot.waitExposed(win)
yield win
@pytest.fixture(params=["qwidget", "qml"])
def host(request, window, hub, qtbot):
cls = NotificationHostQWidget if request.param == "qwidget" else NotificationHostQML
attached = cls(window)
yield attached
attached.cleanup()
def flush(qtbot, host):
"""Run the coalesced view update."""
host.sync_now()
qtbot.wait(10)
def toast_count(host) -> int:
if isinstance(host, NotificationHostQML):
return host.backend.toasts.rowCount()
return len(host.toasts.cards)
def row_count(host) -> int:
if isinstance(host, NotificationHostQML):
return host.backend.rows.rowCount()
return host.drawer.model.rowCount()
# ---------------------------------------------------------------------------- pure helpers
@pytest.mark.parametrize(
"value, expected",
[
("info", Severity.INFO),
("minor", Severity.ERROR),
("major", Severity.CRITICAL),
(0, Severity.WARNING),
(1, Severity.ERROR),
(2, Severity.CRITICAL),
(Severity.SUCCESS, Severity.SUCCESS),
("nonsense", Severity.WARNING),
(None, Severity.WARNING),
],
)
def test_to_severity(value, expected):
assert common.to_severity(value) == expected
def test_split_traceback():
assert common.split_traceback(TRACE) == ("ValueError", "Target 12.5 is outside the limits.")
assert common.split_traceback("") == ("Error", "")
assert common.split_traceback("something odd happened") == ("Error", "something odd happened")
assert common.split_traceback("ophyd.utils.errors.WaitTimeoutError: late")[0] == (
"WaitTimeoutError"
)
def test_relative_time():
now = time.time()
assert common.relative_time(now - 5, now) == "just now"
assert common.relative_time(now - 300, now) == "5 min ago"
# ---------------------------------------------------------------------------- hub
def test_hub_is_singleton(hub):
assert NotificationHub.instance() is hub
def test_notify_adds_entry_newest_first(hub):
first = hub.notify("A", "one")
second = hub.notify("B", "two", "warning", source="x")
assert [e.id for e in hub.entries] == [second, first]
assert hub.get(second).severity == Severity.WARNING
assert hub.unread_count() == 2
def test_repeated_notifications_fold_into_one_entry(hub):
first = hub.notify("Readback failed", "timeout", "error", source="samx")
hub.notify("Other", "x")
again = hub.notify("Readback failed", "timeout", "error", source="samx")
assert again == first
assert len(hub.entries) == 2
assert hub.entries[0].id == first # moved to the top
assert hub.get(first).count == 2
# a different source is a different problem
hub.notify("Readback failed", "timeout", "error", source="samy")
assert len(hub.entries) == 3
def test_same_entry_id_is_posted_once(hub):
hub.notify("A", entry_id="fixed")
hub.notify("A changed", entry_id="fixed")
assert len(hub.entries) == 1
def test_history_is_trimmed_but_keeps_open_critical(hub, monkeypatch):
monkeypatch.setattr(NotificationHub, "MAX_ENTRIES", 5)
critical = hub.notify("Beam lost", severity="critical")
for i in range(10):
hub.notify(f"info {i}")
assert len(hub.entries) == 5
assert hub.get(critical) is not None
def test_clear_keeps_unacknowledged_critical(hub):
critical = hub.notify("Beam lost", severity="critical")
hub.notify("hello")
hub.clear()
assert [e.id for e in hub.entries] == [critical]
hub.acknowledge(critical)
hub.clear()
assert hub.entries == []
def test_acknowledge_and_top_severity(hub):
hub.notify("w", severity="warning")
critical = hub.notify("c", severity="critical")
assert hub.top_unread_severity() == Severity.CRITICAL
hub.mark_read()
# still open: a critical error needs an acknowledgement, reading it is not enough
assert hub.top_unread_severity() == Severity.CRITICAL
hub.acknowledge(critical)
assert hub.top_unread_severity() is None
def test_post_alarm_parses_error_info(hub):
entry_id = hub.post_alarm(
{
"severity": 2,
"alarm_type": "DeviceTimeoutError",
"info": {
"id": "abc",
"error_message": TRACE,
"compact_error_message": "Device eiger did not respond.",
"exception_type": "DeviceTimeoutError",
"device": "eiger",
},
},
{"scan_number": 12, "scan_id": "sid"},
)
entry = hub.get(entry_id)
assert entry_id == "abc"
assert entry.severity == Severity.CRITICAL
assert entry.title == "DeviceTimeoutError"
assert entry.body == "Device eiger did not respond."
assert entry.source == "Device eiger"
assert entry.scan_number == 12
assert "sid" in entry.details and "Traceback" in entry.details
# the same alarm arriving again is ignored
assert hub.post_alarm({"severity": 2, "info": {"id": "abc", "error_message": "x"}}) is None
def test_post_alarm_without_info(hub):
entry_id = hub.post_alarm({"severity": 0, "alarm_type": "Warning", "msg": "low beam"})
entry = hub.get(entry_id)
assert entry.severity == Severity.WARNING
assert entry.body == "low beam"
assert entry.details == ""
def test_report_exception(hub):
entry = hub.get(hub.report_exception("Method error", TRACE, source="Box.move"))
assert entry.title == "ValueError"
assert entry.body == "Target 12.5 is outside the limits."
assert entry.source == "Box.move"
assert entry.severity == Severity.ERROR
assert "Traceback" in entry.details
report = entry.copy_text()
assert "[Error] ValueError" in report and "Source: Box.move" in report and TRACE in report
def test_notify_from_worker_thread_is_queued(hub, qtbot):
thread = threading.Thread(target=lambda: hub.notify("from thread"))
thread.start()
thread.join()
qtbot.waitUntil(lambda: len(hub.entries) == 1, timeout=1000)
assert hub.entries[0].title == "from thread"
def test_bec_alarm_subscription(hub, bec_dispatcher):
with mock.patch.object(bec_dispatcher, "connect_slot") as connect:
hub.connect_bec(bec_dispatcher)
hub.connect_bec(bec_dispatcher) # only once
assert connect.call_count == 2
with mock.patch.object(bec_dispatcher, "disconnect_slot") as disconnect:
hub.disconnect_bec()
assert disconnect.call_count == 2
def test_scan_start_signal(hub, qtbot):
with qtbot.waitSignal(hub.scan_started, timeout=500):
hub._on_scan_status({"status": "open"}, {})
with qtbot.assertNotEmitted(hub.scan_started):
hub._on_scan_status({"status": "closed"}, {})
# ---------------------------------------------------------------------------- toast queue
@pytest.fixture
def queue(hub):
toast_queue = ToastQueue(hub)
yield toast_queue
toast_queue.cleanup()
toast_queue.deleteLater()
def test_queue_limits_visible_toasts(hub, queue):
ids = [hub.notify(f"n{i}", severity="error") for i in range(5)]
assert queue.visible_ids == ids[-3:]
assert queue.overflow == 2
def test_queue_critical_evicts_oldest_transient(hub, queue):
critical = [hub.notify(f"c{i}", severity="critical") for i in range(2)]
info = hub.notify("info")
new_critical = hub.notify("c-new", severity="critical")
assert info not in queue.visible_ids
assert queue.visible_ids == critical + [new_critical]
# all slots held by critical errors: a transient toast only counts as overflow
hub.notify("more info")
assert queue.visible_ids == critical + [new_critical]
def test_queue_auto_dismiss_and_pause(hub, queue, qtbot, monkeypatch):
monkeypatch.setitem(common.SEVERITY_META[Severity.INFO], "lifetime", 80)
entry_id = hub.notify("short")
queue.set_paused(True)
qtbot.wait(150)
assert entry_id in queue.visible_ids
queue.set_paused(False)
qtbot.waitUntil(lambda: entry_id not in queue.visible_ids, timeout=2000)
# the entry stays in the history
assert hub.get(entry_id) is not None
def test_queue_critical_stays_until_acknowledged(hub, queue):
critical = hub.notify("c", severity="critical")
assert critical not in queue._timers
hub.acknowledge(critical)
assert critical not in queue.visible_ids
def test_queue_scan_start_keeps_critical(hub, queue):
hub.notify("info")
critical = hub.notify("c", severity="critical")
hub.scan_started.emit()
assert queue.visible_ids == [critical]
def test_queue_repeat_moves_toast_to_newest(hub, queue):
first = hub.notify("a", "x", "error")
hub.notify("b", "y", "error")
hub.notify("a", "x", "error")
assert queue.visible_ids[-1] == first
assert len(queue.visible_ids) == 2
def test_queue_suspended_shows_nothing(hub, queue):
hub.notify("a")
queue.set_suspended(True)
assert queue.visible_ids == []
hub.notify("b")
assert queue.visible_ids == []
# ---------------------------------------------------------------------------- hosts (both views)
def test_host_shows_toasts_and_bell(host, hub, qtbot):
hub.notify("Hello", "world")
hub.notify("Broken", "it failed", "critical")
flush(qtbot, host)
assert toast_count(host) == 2
assert host.toasts.isVisible()
state = host.view_state()
assert state["unread"] == 2
assert state["openCriticals"] == 1
assert "critical" in state["bellTip"]
assert host.toasts.geometry().right() <= host.window.width()
assert host.toasts.geometry().bottom() < host.window.statusBar().geometry().top()
def test_host_drawer_lists_filters_and_details(host, hub, qtbot):
hub.notify("Info one", "a")
error = hub.report_exception("Method error", TRACE, source="Box.move")
hub.notify("Warn", "low beam", "warning")
host.set_drawer_open(True)
flush(qtbot, host)
assert host.drawer.isVisible()
assert toast_count(host) == 0 # toasts give way to the drawer
assert hub.unread_count() == 0 # opening the drawer reads everything
assert row_count(host) == 3
host.set_filter("errors")
flush(qtbot, host)
assert row_count(host) == 1
host.set_filter("all")
host.set_search("low beam")
flush(qtbot, host)
assert row_count(host) == 1
host.set_search("")
host.select(error)
flush(qtbot, host)
assert host.view_state()["selected"]["entryId"] == error
assert host.copy_details(error)
assert "ValueError" in QApplication.clipboard().text()
host.remove(error)
flush(qtbot, host)
assert hub.get(error) is None
assert host.view_state()["selected"] is None
host.set_drawer_open(False)
flush(qtbot, host)
assert not host.drawer.isVisible()
def test_host_critical_pinned_on_top(host, hub, qtbot):
critical = hub.notify("Beam lost", severity="critical")
hub.notify("newer info")
host.set_drawer_open(True)
flush(qtbot, host)
assert host.view_state()["rows"][0]["entryId"] == critical
host.acknowledge(critical)
flush(qtbot, host)
assert host.view_state()["rows"][0]["entryId"] != critical
def test_host_open_details_from_toast(host, hub, qtbot):
entry_id = hub.notify("Broken", "x", "error", details="trace")
flush(qtbot, host)
host.open_details(entry_id)
flush(qtbot, host)
assert host.drawer_open
assert host.view_state()["selected"]["details"] == "trace"
def test_host_dismiss_critical_toast_acknowledges(host, hub, qtbot):
critical = hub.notify("Beam lost", severity="critical")
flush(qtbot, host)
host.dismiss_toast(critical)
flush(qtbot, host)
assert hub.get(critical).acknowledged
assert toast_count(host) == 0
def test_host_follows_theme(host, hub, qtbot):
from bec_widgets.utils.colors import apply_theme
hub.notify("x", severity="warning")
flush(qtbot, host)
apply_theme("dark")
flush(qtbot, host)
apply_theme("light")
flush(qtbot, host)
assert toast_count(host) == 1
def test_host_cleanup_unregisters(window, hub):
attached = NotificationHostQWidget(window)
assert hub.has_hosts
attached.cleanup()
attached.cleanup() # idempotent
assert not hub.has_hosts
def test_qml_views_load_without_errors(window, hub):
attached = NotificationHostQML(window)
try:
for view in (attached.toasts, attached.drawer, attached.bell):
assert view.errors() == []
assert view.rootObject() is not None
finally:
attached.cleanup()
def test_qwidget_toast_buttons(window, hub, qtbot):
attached = NotificationHostQWidget(window)
try:
hub.notify("Broken", "x", "critical", details="trace")
flush(qtbot, attached)
card = attached.toasts.cards[0]
assert card.ack_btn.isVisibleTo(card)
assert not card.close_btn.isVisibleTo(card)
card.details_btn.click()
flush(qtbot, attached)
assert attached.drawer_open
attached.drawer.details.ack_btn.click()
flush(qtbot, attached)
assert hub.entries[0].acknowledged
finally:
attached.cleanup()
# ---------------------------------------------------------------------------- error routing
class _Failing(QWidget):
@SafeSlot(popup_error=True)
def fail(self):
raise ValueError("Target 12.5 is outside the limits.")
def test_safeslot_error_becomes_notification_when_host_attached(window, hub, qtbot):
attached = NotificationHostQWidget(window)
widget = _Failing()
qtbot.addWidget(widget)
try:
with (
mock.patch.object(QMessageBox, "exec_") as exec_,
qtbot.assertNotEmitted(ErrorPopupUtility().error_occurred),
):
widget.fail()
exec_.assert_not_called()
entry = hub.entries[0]
assert entry.title == "ValueError"
assert entry.source == "_Failing.fail"
assert "Traceback" in entry.details
finally:
attached.cleanup()
def test_safeslot_error_uses_dialog_without_host(hub, qtbot):
widget = _Failing()
qtbot.addWidget(widget)
with qtbot.waitSignal(ErrorPopupUtility().error_occurred, timeout=1000):
widget.fail()
assert hub.entries == []
def test_warning_popup_becomes_notification(window, hub):
attached = NotificationHostQWidget(window)
try:
with mock.patch.object(QMessageBox, "exec_") as exec_:
WarningPopupUtility().show_warning("Export", "No data to export.", "details", window)
exec_.assert_not_called()
entry = hub.entries[0]
assert (entry.title, entry.body, entry.severity) == (
"Export",
"No data to export.",
Severity.WARNING,
)
assert entry.details == "details"
finally:
attached.cleanup()
def test_application_error_routed(window, hub):
attached = NotificationHostQWidget(window)
try:
util = ErrorPopupUtility()
util.enable_global_error_popups(True)
try:
raise RuntimeError("boom")
except RuntimeError as exc:
util.custom_exception_hook(type(exc), exc, exc.__traceback__)
assert hub.entries[0].title == "RuntimeError"
assert hub.entries[0].source == "Application Error"
finally:
util.enable_global_error_popups(False)
attached.cleanup()
# ---------------------------------------------------------------------------- main window opt-in
@pytest.mark.parametrize("variant", ["qwidget", "qml"])
def test_main_window_opt_in(qtbot, monkeypatch, variant, mocked_client):
from bec_widgets.widgets.containers.main_window.main_window import BECMainWindow
monkeypatch.setenv("BEC_NOTIFICATION_UI", variant)
win = BECMainWindow(client=mocked_client)
qtbot.addWidget(win)
try:
assert win.notification_centre is None
assert win.notifications is not None
assert win.notification_indicator is win.notifications.bell
win.notifications.hub.notify("hello")
win.notifications.sync_now()
assert win.notifications.view_state()["unread"] == 1
finally:
win.close()
def test_main_window_default_is_legacy(qtbot, mocked_client, monkeypatch):
from bec_widgets.widgets.containers.main_window.main_window import BECMainWindow
monkeypatch.delenv("BEC_NOTIFICATION_UI", raising=False)
win = BECMainWindow(client=mocked_client)
qtbot.addWidget(win)
try:
assert win.notification_centre is not None
assert win.notifications is None
finally:
win.close()
def test_traceback_helper_matches_python():
try:
raise KeyError("missing")
except KeyError:
text = traceback.format_exc()
assert common.split_traceback(text) == ("KeyError", "'missing'")