feat(notifications): QML and QWidget toasts with history drawer, folded repeats and sticky critical errors instead of modal error dialogs

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Kns5hgJ1MEXNfFeCuTESFP
This commit is contained in:
Claude committed 2026-10-09 21:29:59 +00:00
1 parent d2d7486930
commit 4ac2e629aa
11 files changed
+3390 -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
@@ -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'")