diff --git a/bec_widgets/tests/fixtures.py b/bec_widgets/tests/fixtures.py index 7078d1db..78c46351 100644 --- a/bec_widgets/tests/fixtures.py +++ b/bec_widgets/tests/fixtures.py @@ -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) diff --git a/bec_widgets/utils/error_popups.py b/bec_widgets/utils/error_popups.py index 993e2f36..6157a19d 100644 --- a/bec_widgets/utils/error_popups.py +++ b/bec_widgets/utils/error_popups.py @@ -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 diff --git a/bec_widgets/widgets/containers/main_window/addons/notification_center/notification_demo.py b/bec_widgets/widgets/containers/main_window/addons/notification_center/notification_demo.py new file mode 100644 index 00000000..c9e174dc --- /dev/null +++ b/bec_widgets/widgets/containers/main_window/addons/notification_center/notification_demo.py @@ -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() diff --git a/bec_widgets/widgets/containers/main_window/addons/notification_center/notification_qml.py b/bec_widgets/widgets/containers/main_window/addons/notification_center/notification_qml.py new file mode 100644 index 00000000..7160fe14 --- /dev/null +++ b/bec_widgets/widgets/containers/main_window/addons/notification_center/notification_qml.py @@ -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() diff --git a/bec_widgets/widgets/containers/main_window/addons/notification_center/notification_qwidget.py b/bec_widgets/widgets/containers/main_window/addons/notification_center/notification_qwidget.py new file mode 100644 index 00000000..8fba4ded --- /dev/null +++ b/bec_widgets/widgets/containers/main_window/addons/notification_center/notification_qwidget.py @@ -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( + "" + + "".join( + f"" + f"" + for k, v in facts + ) + + "
{k}{v}
" + ) + 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() diff --git a/bec_widgets/widgets/containers/main_window/addons/notification_center/notification_ux_common.py b/bec_widgets/widgets/containers/main_window/addons/notification_center/notification_ux_common.py new file mode 100644 index 00000000..ccc010f7 --- /dev/null +++ b/bec_widgets/widgets/containers/main_window/addons/notification_center/notification_ux_common.py @@ -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.""" diff --git a/bec_widgets/widgets/containers/main_window/addons/notification_center/qml/NotificationBell.qml b/bec_widgets/widgets/containers/main_window/addons/notification_center/qml/NotificationBell.qml new file mode 100644 index 00000000..d254d813 --- /dev/null +++ b/bec_widgets/widgets/containers/main_window/addons/notification_center/qml/NotificationBell.qml @@ -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" +} diff --git a/bec_widgets/widgets/containers/main_window/addons/notification_center/qml/NotificationDrawer.qml b/bec_widgets/widgets/containers/main_window/addons/notification_center/qml/NotificationDrawer.qml new file mode 100644 index 00000000..bf234e44 --- /dev/null +++ b/bec_widgets/widgets/containers/main_window/addons/notification_center/qml/NotificationDrawer.qml @@ -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) + } + } + } + } + } +} diff --git a/bec_widgets/widgets/containers/main_window/addons/notification_center/qml/NotificationToasts.qml b/bec_widgets/widgets/containers/main_window/addons/notification_center/qml/NotificationToasts.qml new file mode 100644 index 00000000..711d2354 --- /dev/null +++ b/bec_widgets/widgets/containers/main_window/addons/notification_center/qml/NotificationToasts.qml @@ -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) + } + } + } + } + } + } + } + } +} diff --git a/bec_widgets/widgets/containers/main_window/main_window.py b/bec_widgets/widgets/containers/main_window/main_window.py index 6753c845..1682596c 100644 --- a/bec_widgets/widgets/containers/main_window/main_window.py +++ b/bec_widgets/widgets/containers/main_window/main_window.py @@ -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() diff --git a/tests/unit_tests/test_notification_ports.py b/tests/unit_tests/test_notification_ports.py new file mode 100644 index 00000000..2f8541b6 --- /dev/null +++ b/tests/unit_tests/test_notification_ports.py @@ -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'")