From 49be3ece8c77485aca1011b5dd8242fbd80c05e3 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 9 Oct 2026 20:29:22 +0000 Subject: [PATCH 1/2] feat(bec_status_box): QML and QWidget views with overall state, grouped services and inline details Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_015TeMnCZ9FjLpwiDkPu6K9j --- bec_widgets/cli/client.py | 2 +- bec_widgets/utils/qml_host.py | 173 ++++++ .../services/bec_status_box/bec_status_box.py | 272 ++++----- .../services/bec_status_box/qml/StatusBox.qml | 286 ++++++++++ .../bec_status_box/status_box_widgets.py | 515 ++++++++++++++++++ .../services/bec_status_box/status_model.py | 419 ++++++++++++++ tests/unit_tests/test_bec_status_box.py | 308 ++++++++--- 7 files changed, 1729 insertions(+), 246 deletions(-) create mode 100644 bec_widgets/utils/qml_host.py create mode 100644 bec_widgets/widgets/services/bec_status_box/qml/StatusBox.qml create mode 100644 bec_widgets/widgets/services/bec_status_box/status_box_widgets.py create mode 100644 bec_widgets/widgets/services/bec_status_box/status_model.py diff --git a/bec_widgets/cli/client.py b/bec_widgets/cli/client.py index 56b0f853..da899bb0 100644 --- a/bec_widgets/cli/client.py +++ b/bec_widgets/cli/client.py @@ -558,7 +558,7 @@ class BECStatusBox(RPCBase): @rpc_call def get_server_state(self) -> "str": """ - Get the state ("RUNNING", "BUSY", "IDLE", "ERROR") of the BEC server + Get the state ("RUNNING", "BUSY", "IDLE", "ERROR", "NOTCONNECTED") of the BEC server """ @rpc_call diff --git a/bec_widgets/utils/qml_host.py b/bec_widgets/utils/qml_host.py new file mode 100644 index 00000000..7e44065d --- /dev/null +++ b/bec_widgets/utils/qml_host.py @@ -0,0 +1,173 @@ +"""Helpers for hosting QML views inside BEC QWidgets. + +The helpers expose the active BEC theme to QML (``theme`` context property), serve Material icons +from ``bec_qthemes`` through the ``image://material/?color=&filled=1`` URL scheme, and +create a ``QQuickWidget`` that resizes with its host widget. +""" + +from __future__ import annotations + +from pathlib import Path +from urllib.parse import parse_qs + +from bec_qthemes import material_icon +from qtpy.QtCore import Property, QObject, QSize, QUrl, Signal +from qtpy.QtGui import QColor, QPixmap +from qtpy.QtQuick import QQuickImageProvider +from qtpy.QtQuickWidgets import QQuickWidget +from qtpy.QtWidgets import QApplication, QWidget + +from bec_widgets.utils.colors import get_theme_name + +_FALLBACK_COLORS = { + "dark": { + "BG": "#1e1f22", + "CARD_BG": "#2b2c2f", + "FG": "#dddddd", + "BORDER": "#464749", + "DISABLED_FG": "#6f6f71", + "PRIMARY": "#6089ef", + "ACCENT_DEFAULT": "#40b6e0", + "ACCENT_WARNING": "#eda200", + "ACCENT_EMERGENCY": "#e0534b", + "ACCENT_SUCCESS": "#59a869", + }, + "light": { + "BG": "#ffffff", + "CARD_BG": "#f4f5f7", + "FG": "#1e1f22", + "BORDER": "#d0d3d9", + "DISABLED_FG": "#8a8d93", + "PRIMARY": "#0a60ff", + "ACCENT_DEFAULT": "#0a60ff", + "ACCENT_WARNING": "#c88600", + "ACCENT_EMERGENCY": "#d32f2f", + "ACCENT_SUCCESS": "#2e7d32", + }, +} + + +class QmlTheme(QObject): + """Expose the active BEC theme colors to QML and follow theme changes.""" + + changed = Signal() + + def __init__(self, parent: QObject | None = None): + super().__init__(parent) + app = QApplication.instance() + theme = getattr(app, "theme", None) + if theme is not None: + theme.theme_changed.connect(self._on_theme_changed) + + def _on_theme_changed(self, _theme: str): + self.changed.emit() + + def _color(self, key: str) -> QColor: + name = get_theme_name() + fallback = _FALLBACK_COLORS.get(name, _FALLBACK_COLORS["dark"])[key] + theme = getattr(QApplication.instance(), "theme", None) + if theme is None: + return QColor(fallback) + return theme.color(key, fallback) + + @Property(bool, notify=changed) + def dark(self) -> bool: + """Whether the dark theme is active.""" + return get_theme_name() != "light" + + @Property(QColor, notify=changed) + def background(self) -> QColor: + """Window background color.""" + return self._color("BG") + + @Property(QColor, notify=changed) + def card(self) -> QColor: + """Card/surface background color.""" + return self._color("CARD_BG") + + @Property(QColor, notify=changed) + def text(self) -> QColor: + """Primary text color.""" + return self._color("FG") + + @Property(QColor, notify=changed) + def muted(self) -> QColor: + """Secondary text color.""" + return self._color("DISABLED_FG") + + @Property(QColor, notify=changed) + def border(self) -> QColor: + """Border and divider color.""" + return self._color("BORDER") + + @Property(QColor, notify=changed) + def primary(self) -> QColor: + """Primary accent color.""" + return self._color("PRIMARY") + + @Property(QColor, notify=changed) + def success(self) -> QColor: + """Success accent color.""" + return self._color("ACCENT_SUCCESS") + + @Property(QColor, notify=changed) + def warning(self) -> QColor: + """Warning accent color.""" + return self._color("ACCENT_WARNING") + + @Property(QColor, notify=changed) + def emergency(self) -> QColor: + """Emergency accent color.""" + return self._color("ACCENT_EMERGENCY") + + +class MaterialIconProvider(QQuickImageProvider): + """Serve ``bec_qthemes`` Material icons to QML as ``image://material/?color=..``.""" + + def __init__(self): + super().__init__(QQuickImageProvider.ImageType.Pixmap) + + def requestPixmap(self, icon_id: str, size: QSize, requested_size: QSize) -> QPixmap: + name, _, query = icon_id.partition("?") + params = {key: values[-1] for key, values in parse_qs(query).items()} + side = max(requested_size.width(), requested_size.height(), 0) or 24 + pixmap = material_icon( + name, + size=(side, side), + color=params.get("color"), + filled=params.get("filled") == "1", + convert_to_pixmap=True, + ) + if size is not None: + size.setWidth(pixmap.width()) + size.setHeight(pixmap.height()) + return pixmap + + +def create_quick_widget( + parent: QWidget, qml_file: str | Path, context: dict[str, QObject] | None = None +) -> QQuickWidget: + """Create a QQuickWidget for a QML file with the BEC theme and icon provider installed. + + Args: + parent (QWidget): The host widget. + qml_file (str | Path): Path to the root QML file. + context (dict[str, QObject] | None): Extra context properties exposed to QML. + + Returns: + QQuickWidget: The widget, already loaded. Errors are raised as RuntimeError. + """ + quick = QQuickWidget(parent) + quick.setResizeMode(QQuickWidget.ResizeMode.SizeRootObjectToView) + quick.setClearColor(QColor(0, 0, 0, 0)) + engine = quick.engine() + engine.addImageProvider("material", MaterialIconProvider()) + theme = QmlTheme(quick) + quick.rootContext().setContextProperty("theme", theme) + for name, obj in (context or {}).items(): + quick.rootContext().setContextProperty(name, obj) + quick.setSource(QUrl.fromLocalFile(str(Path(qml_file).resolve()))) + if quick.status() == QQuickWidget.Status.Error: + errors = "; ".join(err.toString() for err in quick.errors()) + raise RuntimeError(f"Failed to load {qml_file}: {errors}") + return quick diff --git a/bec_widgets/widgets/services/bec_status_box/bec_status_box.py b/bec_widgets/widgets/services/bec_status_box/bec_status_box.py index dd284e64..ae8734de 100644 --- a/bec_widgets/widgets/services/bec_status_box/bec_status_box.py +++ b/bec_widgets/widgets/services/bec_status_box/bec_status_box.py @@ -1,21 +1,25 @@ -"""This module contains the BECStatusBox widget, which displays the status of different BEC services in a collapsible tree widget. +"""This module contains the BECStatusBox widget, which displays the status of different BEC services. The widget automatically updates the status of all running BEC services, and displays their status. """ from __future__ import annotations -from collections import defaultdict +import os from dataclasses import dataclass +from pathlib import Path from typing import TYPE_CHECKING +from bec_lib.logger import bec_logger from bec_lib.utils.import_utils import lazy_import_from from pydantic import BaseModel -from qtpy.QtCore import QObject, QTimer, Signal, Slot -from qtpy.QtWidgets import QHBoxLayout, QTreeWidget, QTreeWidgetItem +from qtpy.QtCore import QObject, QTimer, Signal +from qtpy.QtWidgets import QHBoxLayout, QWidget from bec_widgets.utils.bec_widget import BECWidget from bec_widgets.utils.compact_popup import CompactPopupWidget -from bec_widgets.widgets.services.bec_status_box.status_item import StatusItem +from bec_widgets.utils.error_popups import SafeSlot +from bec_widgets.widgets.services.bec_status_box.status_box_widgets import StatusBoxWidgetView +from bec_widgets.widgets.services.bec_status_box.status_model import ServiceStatusModel, build_rows if TYPE_CHECKING: # pragma: no cover from bec_lib.client import BECClient @@ -24,6 +28,10 @@ else: # TODO : Put normal imports back when Pydantic gets faster BECStatus = lazy_import_from("bec_lib.messages", ("BECStatus",)) +logger = bec_logger.logger + +QML_FILE = Path(__file__).parent / "qml" / "StatusBox.qml" + @dataclass class BECServiceInfoContainer: @@ -68,11 +76,17 @@ class BECServiceStatusMixin(QObject): class BECStatusBox(BECWidget, CompactPopupWidget): """An autonomous widget to display the status of BEC services. + The header shows the overall state of BEC, the list groups the core services and the other + services (clients, data processing). Clicking a service shows its host, user, uptime, load and + versions inline, with a button to copy them. + Args: parent Optional : The parent widget for the BECStatusBox. Defaults to None. box_name Optional(str): The name of the top service label. Defaults to "BEC Server". client Optional(BECClient): The client object to connect to the BEC server. Defaults to None - config Optional(BECStatusBoxConfig | dict): The configuration for the status box. Defaults to None. + bec_service_status_mixin Optional(BECServiceStatusMixin): Source of the service updates. + view Optional(str): "qml" or "widgets". Defaults to the BEC_STATUS_BOX_VIEW environment + variable, or "qml". gui_id Optional(str): The unique id for the widget. Defaults to None. """ @@ -89,13 +103,16 @@ class BECStatusBox(BECWidget, CompactPopupWidget): box_name: str = "BEC Servers", client: BECClient = None, bec_service_status_mixin: BECServiceStatusMixin = None, + view: str | None = None, gui_id: str = None, **kwargs, ): super().__init__(parent=parent, layout=QHBoxLayout, client=client, gui_id=gui_id, **kwargs) self.box_name = box_name - self.status_container = defaultdict(lambda: {"info": None, "item": None, "widget": None}) + self.status_container: dict[str, BECServiceInfoContainer] = {} + self.core_state = "IDLE" + self.model = ServiceStatusModel(self) if not bec_service_status_mixin: bec_service_status_mixin = BECServiceStatusMixin(self, client=self.client) @@ -103,219 +120,162 @@ class BECStatusBox(BECWidget, CompactPopupWidget): self.label = box_name self.tooltip = "BEC servers health status" - self.init_ui() + self.view_kind = (view or os.environ.get("BEC_STATUS_BOX_VIEW") or "qml").lower() + self.view = self._create_view() + self.layout.setContentsMargins(0, 0, 0, 0) + self.addWidget(self.view) + self.model.summaryChanged.connect(self._update_compact_state) self.bec_service_status.services_update.connect(self.update_service_status) self.bec_core_state.connect(self.update_top_item_status) - self.tree.itemDoubleClicked.connect(self.on_tree_item_double_clicked) - self.addWidget(self.tree) + self._update_compact_state() - def init_ui(self) -> None: - """Init the UI for the BECStatusBox widget, should only take place once.""" - self.init_ui_tree_widget() - top_label = self._create_status_widget(self.box_name, status=BECStatus.IDLE) - tree_item = QTreeWidgetItem() - tree_item.setExpanded(True) - tree_item.setDisabled(True) - self.status_container[self.box_name].update({"item": tree_item, "widget": top_label}) - self.tree.addTopLevelItem(tree_item) - self.tree.setItemWidget(tree_item, 0, top_label) - self.service_update.connect(top_label.update_config) - self._initialized = True + def _create_view(self) -> QWidget: + """Create the QML or QWidget view on the shared model.""" + if self.view_kind == "qml": + try: + # pylint: disable=import-outside-toplevel + from bec_widgets.utils.qml_host import create_quick_widget - def init_ui_tree_widget(self) -> None: - """Initialise the tree widget for the status box.""" - self.tree = QTreeWidget(self) - self.tree.setHeaderHidden(True) - # TODO probably here is a problem still with setting the stylesheet - self.tree.setStyleSheet( - "QTreeWidget::item:!selected " - "{ " - "border: 1px solid gainsboro; " - "border-left: none; " - "border-top: none; " - "}" - "QTreeWidget::item:selected {}" - ) + quick = create_quick_widget(self, QML_FILE, context={"statusModel": self.model}) + quick.setMinimumSize(280, 220) + return quick + except Exception as exc: # pylint: disable=broad-except + logger.warning(f"QML status box unavailable, using the QWidget view: {exc}") + self.view_kind = "widgets" + view = StatusBoxWidgetView(self.model, self) + view.setMinimumSize(280, 220) + return view def get_server_state(self) -> str: - """Get the state ("RUNNING", "BUSY", "IDLE", "ERROR") of the BEC server""" - return self.status_container[self.box_name]["info"].status + """Get the state ("RUNNING", "BUSY", "IDLE", "ERROR", "NOTCONNECTED") of the BEC server""" + return self.core_state - def _create_status_widget( - self, service_name: str, status=BECStatus, info: dict = None, metrics: dict = None - ) -> StatusItem: - """Creates a StatusItem (QWidget) for the given service, and stores all relevant - information about the service in the status_container. + @SafeSlot(str) + def update_top_item_status(self, status: str) -> None: + """Update the combined state of the core services. Args: - service_name (str): The name of the service. - status (BECStatus): The status of the service. - info Optional(dict): The information about the service. Default is {} - metric Optional(dict): Metrics for the respective service. Default is None - - Returns: - StatusItem: The status item widget. + status (str): The state of the core services. """ - if info is None: - info = {} - self._update_status_container(service_name, status, info, metrics) - item = StatusItem(parent=self, config=self.status_container[service_name]["info"]) - return item - - @Slot(str) - def update_top_item_status(self, status: BECStatus) -> None: - """Method to update the status of the top item in the tree widget. - Gets the status from the Signal 'bec_core_state' and updates the StatusItem via the signal 'service_update'. - - Args: - status (BECStatus): The state of the core services. - """ - self.status_container[self.box_name]["info"].status = status - self.set_global_state("emergency" if status == "NOTCONNECTED" else "success") - self.service_update.emit(self.status_container[self.box_name]["info"]) + if status == self.core_state: + return + self.core_state = status + self._refresh_model() def _update_status_container( - self, service_name: str, status: BECStatus, info: dict, metrics: dict = None + self, service_name: str, status: BECStatus | str, info: dict, metrics: dict = None ) -> None: - """Update the status_container with the newest status and metrics for the BEC service. - If information about the service already exists, it will create a new entry. + """Store the newest status and metrics of a BEC service. Args: service_name (str): The name of the service. - status (BECStatus): The status of the service. + status (BECStatus | str): The status of the service. info (dict): The information about the service. metrics (dict): The metrics of the service. """ - container = self.status_container[service_name].get("info", None) - + status = status.name if isinstance(status, BECStatus) else status + container = self.status_container.get(service_name) if container: - container.status = status.name + container.status = status container.info = info container.metrics = metrics - return - service_info_item = BECServiceInfoContainer( - service_name=service_name, - status=status.name if isinstance(status, BECStatus) else status, - info=info, - metrics=metrics, - ) - self.status_container[service_name].update({"info": service_info_item}) + else: + container = BECServiceInfoContainer( + service_name=service_name, status=status, info=info, metrics=metrics + ) + self.status_container[service_name] = container + self.service_update.emit(container) - @Slot(dict, dict) + @SafeSlot(dict, dict) def update_service_status( self, services_info: dict[str, StatusMessage], services_metric: dict[str, ServiceMetricMessage], ) -> None: - """Callback function services_metric from BECServiceStatusMixin. - It updates the status of all services. + """Callback for BECServiceStatusMixin.services_update. Updates the status of all services. Args: - services_info (dict): A dictionary containing the service status for all running BEC services. - services_metric (dict): A dictionary containing the service metrics for all running BEC services. + services_info (dict): The service status for all running BEC services. + services_metric (dict): The service metrics for all running BEC services. """ - checked = [self.box_name] + services_info = dict(services_info) + checked = list(self.CORE_SERVICES) # FIXME: We simply replace the pydantic message with dict for now until we refactor the widget for val in services_info.values(): val.info = val.info.model_dump() if isinstance(val.info, BaseModel) else val.info - services_info = self.update_core_services(services_info, services_metric) - checked.extend(self.CORE_SERVICES) + services_info = self.update_core_services(services_info, services_metric, refresh=False) for service_name, msg in sorted(services_info.items()): checked.append(service_name) metric_msg = services_metric.get(service_name, None) metrics = metric_msg.metrics if metric_msg else None - if service_name in self.status_container: - if not msg: - self.add_tree_item(service_name, "NOTCONNECTED", {}, metrics) - continue - self._update_status_container(service_name, msg.status, msg.info, metrics) - self.service_update.emit(self.status_container[service_name]["info"]) + if not msg: + self._update_status_container(service_name, "NOTCONNECTED", {}, metrics) continue + self._update_status_container(service_name, msg.status, msg.info, metrics) + self.remove_stale_services(checked) + self._refresh_model() - self.add_tree_item(service_name, msg.status, msg.info, metrics) - self.check_redundant_tree_items(checked) - - def update_core_services(self, services_info: dict, services_metric: dict) -> dict: - """Update the core services of BEC, and emit the updated status to the BECStatusBox. + def update_core_services( + self, services_info: dict, services_metric: dict, refresh: bool = True + ) -> dict: + """Update the core services of BEC and emit their combined state. Args: - services_info (dict): A dictionary containing the service status of different services. - services_metric (dict): A dictionary containing the service metrics of different services. + services_info (dict): The service status of different services. + services_metric (dict): The service metrics of different services. + refresh (bool): Whether to refresh the view right away. Returns: - dict: The services_info dictionary after removing the info updates related to the CORE_SERVICES + dict: The services_info dictionary without the CORE_SERVICES entries. """ core_state = BECStatus.RUNNING - for service_name in sorted(self.CORE_SERVICES): + for service_name in self.CORE_SERVICES: metric_msg = services_metric.get(service_name, None) metrics = metric_msg.metrics if metric_msg else None msg = services_info.pop(service_name, None) - if service_name not in self.status_container: - if not msg: - self.add_tree_item(service_name, "NOTCONNECTED", {}, metrics) - continue - self.add_tree_item(service_name, msg.status, msg.info, metrics) - continue if not msg: - self.status_container[service_name]["info"].status = "NOTCONNECTED" + self._update_status_container(service_name, "NOTCONNECTED", {}, metrics) core_state = None - else: - self._update_status_container(service_name, msg.status, msg.info, metrics) - if core_state: - core_state = msg.status if msg.status.value < core_state.value else core_state + continue + info = msg.info.model_dump() if isinstance(msg.info, BaseModel) else msg.info + self._update_status_container(service_name, msg.status, info, metrics) + if core_state and msg.status.value < core_state.value: + core_state = msg.status - self.service_update.emit(self.status_container[service_name]["info"]) - - self.bec_core_state.emit(core_state.name if core_state else "NOTCONNECTED") + self.core_state = core_state.name if core_state else "NOTCONNECTED" + self.bec_core_state.emit(self.core_state) + if refresh: + self._refresh_model() return services_info - def check_redundant_tree_items(self, checked: list) -> None: - """Utility method to check and remove redundant objects from the BECStatusBox. + def remove_stale_services(self, checked: list) -> None: + """Remove services that are no longer reported by BEC. Args: - checked (list): A list of services that are currently running. + checked (list): The services that are currently known. """ - to_be_deleted = [key for key in self.status_container if key not in checked] + for key in [key for key in self.status_container if key not in checked]: + self.status_container.pop(key) - for key in to_be_deleted: - obj = self.status_container.pop(key) - item = obj["item"] - self.status_container[self.box_name]["item"].removeChild(item) + def _refresh_model(self) -> None: + rows = build_rows(self.status_container.values(), self.CORE_SERVICES) + self.model.set_rows(rows, self.core_state) - def add_tree_item( - self, service_name: str, status: BECStatus, info: dict = None, metrics: dict = None - ) -> None: - """Method to add a new QTreeWidgetItem together with a StatusItem to the tree widget. - - Args: - service_name (str): The name of the service. - status (BECStatus): The status of the service. - info (dict): The information about the service. - metrics (dict): The metrics of the service. - """ - item_widget = self._create_status_widget(service_name, status, info, metrics) - item = QTreeWidgetItem() - self.service_update.connect(item_widget.update_config) - self.status_container[self.box_name]["item"].addChild(item) - self.tree.setItemWidget(item, 0, item_widget) - self.status_container[service_name].update({"item": item, "widget": item_widget}) - - @Slot(QTreeWidgetItem, int) - def on_tree_item_double_clicked(self, item: QTreeWidgetItem, column: int) -> None: - """Callback function for double clicks on individual QTreeWidgetItems in the collapsed section. - - Args: - item (QTreeWidgetItem): The item that was double clicked. - column (int): The column that was double clicked. - """ - for _, objects in self.status_container.items(): - if objects["item"] == item: - objects["widget"].show_popup() + def _update_compact_state(self) -> None: + tone = self.model.tone + self.set_global_state( + {"success": "success", "warning": "warning", "emergency": "emergency"}.get( + tone, "default" + ) + ) + self.tooltip = f"{self.model.headline}\n{self.model.detail}".strip() def cleanup(self): """Cleanup the BECStatusBox widget.""" self.bec_service_status.cleanup() + if isinstance(self.view, StatusBoxWidgetView): + self.view.cleanup() return super().cleanup() diff --git a/bec_widgets/widgets/services/bec_status_box/qml/StatusBox.qml b/bec_widgets/widgets/services/bec_status_box/qml/StatusBox.qml new file mode 100644 index 00000000..2c99419d --- /dev/null +++ b/bec_widgets/widgets/services/bec_status_box/qml/StatusBox.qml @@ -0,0 +1,286 @@ +import QtQuick +import QtQuick.Layouts +import QtQuick.Controls.Basic + +// BEC status box: overall state header plus grouped, expandable service rows. +// Context properties: theme (QmlTheme), statusModel (ServiceStatusModel). +Rectangle { + id: root + color: theme.background + implicitWidth: 360 + implicitHeight: header.height + list.contentHeight + 16 + + property string expandedService: "" + + function toneColor(tone) { + if (tone === "success") return theme.success + if (tone === "warning") return theme.warning + if (tone === "emergency") return theme.emergency + return theme.muted + } + function iconUrl(name, color, filled) { + return "image://material/" + name + "?color=" + encodeURIComponent(color.toString()) + + (filled ? "&filled=1" : "") + } + + // ---- header: overall state ---------------------------------------------------------- + Rectangle { + id: header + anchors { left: parent.left; right: parent.right; top: parent.top; margins: 8 } + height: 56 + radius: 8 + color: Qt.rgba(root.toneColor(statusModel.tone).r, root.toneColor(statusModel.tone).g, + root.toneColor(statusModel.tone).b, theme.dark ? 0.16 : 0.12) + border.color: Qt.rgba(root.toneColor(statusModel.tone).r, root.toneColor(statusModel.tone).g, + root.toneColor(statusModel.tone).b, 0.45) + + RowLayout { + anchors { fill: parent; leftMargin: 12; rightMargin: 12 } + spacing: 10 + Rectangle { + id: dot + Layout.alignment: Qt.AlignVCenter + width: 12; height: 12; radius: 6 + color: root.toneColor(statusModel.tone) + SequentialAnimation on opacity { + running: statusModel.tone === "emergency" + loops: Animation.Infinite + onRunningChanged: if (!running) dot.opacity = 1 + NumberAnimation { to: 0.35; duration: 700; easing.type: Easing.InOutQuad } + NumberAnimation { to: 1.0; duration: 700; easing.type: Easing.InOutQuad } + } + } + ColumnLayout { + Layout.fillWidth: true + spacing: 2 + Text { + objectName: "headline" + Layout.fillWidth: true + text: statusModel.headline + color: theme.text + font.pixelSize: 14; font.bold: true + elide: Text.ElideRight + } + Text { + Layout.fillWidth: true + text: statusModel.detail + color: theme.muted + font.pixelSize: 11 + elide: Text.ElideRight + } + } + } + } + + // ---- service list ------------------------------------------------------------------- + ListView { + id: list + objectName: "serviceList" + anchors { left: parent.left; right: parent.right; top: header.bottom; bottom: parent.bottom + topMargin: 4; leftMargin: 8; rightMargin: 8; bottomMargin: 4 } + clip: true + model: statusModel + spacing: 2 + boundsBehavior: Flickable.StopAtBounds + ScrollBar.vertical: ScrollBar { policy: ScrollBar.AsNeeded } + + section.property: "group" + section.delegate: Item { + required property string section + width: ListView.view.width + height: 28 + Text { + anchors { left: parent.left; leftMargin: 4; bottom: parent.bottom; bottomMargin: 4 } + text: section.toUpperCase() + color: theme.muted + font.pixelSize: 10; font.bold: true; font.letterSpacing: 0.8 + } + } + + delegate: Rectangle { + id: rowItem + required property int index + required property string serviceName + required property string displayName + required property string subtitle + required property string status + required property string statusLabel + required property string tone + required property string icon + required property string version + required property bool versionMismatch + required property string uptime + required property var details + + readonly property bool expanded: root.expandedService === serviceName + readonly property color accent: root.toneColor(tone) + + width: ListView.view.width + height: rowContent.height + radius: 6 + color: hover.hovered || expanded ? theme.card : "transparent" + border.color: expanded ? theme.border : "transparent" + Behavior on height { NumberAnimation { duration: 120; easing.type: Easing.OutCubic } } + + HoverHandler { id: hover } + + Column { + id: rowContent + width: parent.width + + // main line + MouseArea { + width: parent.width + height: 44 + cursorShape: Qt.PointingHandCursor + onClicked: root.expandedService = rowItem.expanded ? "" : rowItem.serviceName + + RowLayout { + anchors { fill: parent; leftMargin: 8; rightMargin: 8 } + spacing: 10 + + Image { + id: statusIcon + Layout.preferredWidth: 20; Layout.preferredHeight: 20 + sourceSize: Qt.size(20, 20) + source: root.iconUrl(rowItem.icon, rowItem.accent, true) + RotationAnimation on rotation { + running: rowItem.status === "BUSY" + loops: Animation.Infinite + from: 0; to: 360; duration: 1200 + onRunningChanged: if (!running) statusIcon.rotation = 0 + } + } + ColumnLayout { + Layout.fillWidth: true + spacing: 1 + Text { + Layout.fillWidth: true + text: rowItem.displayName + color: theme.text + font.pixelSize: 13 + elide: Text.ElideRight + } + Text { + Layout.fillWidth: true + visible: text.length > 0 + text: rowItem.subtitle + color: theme.muted + font.pixelSize: 11 + elide: Text.ElideRight + } + } + Rectangle { + visible: rowItem.versionMismatch + Layout.preferredHeight: 18 + Layout.preferredWidth: mismatchText.implicitWidth + 12 + radius: 9 + color: "transparent" + border.color: theme.warning + ToolTip.visible: mismatchHover.hovered + ToolTip.text: "bec_lib " + rowItem.version + " differs from the other services" + HoverHandler { id: mismatchHover } + Text { + id: mismatchText + anchors.centerIn: parent + text: "v" + rowItem.version + color: theme.warning + font.pixelSize: 10 + } + } + Rectangle { + Layout.preferredHeight: 20 + Layout.preferredWidth: statusText.implicitWidth + 16 + radius: 10 + color: Qt.rgba(rowItem.accent.r, rowItem.accent.g, rowItem.accent.b, 0.15) + Text { + id: statusText + anchors.centerIn: parent + text: rowItem.statusLabel + color: rowItem.accent + font.pixelSize: 11; font.bold: true + } + } + Image { + Layout.preferredWidth: 16; Layout.preferredHeight: 16 + sourceSize: Qt.size(16, 16) + source: root.iconUrl("expand_more", theme.muted, false) + rotation: rowItem.expanded ? 180 : 0 + Behavior on rotation { NumberAnimation { duration: 120 } } + } + } + } + + // details + Item { + visible: rowItem.expanded + width: parent.width + height: visible ? detailsGrid.implicitHeight + 16 : 0 + + GridLayout { + id: detailsGrid + anchors { left: parent.left; right: copyButton.left; top: parent.top + leftMargin: 38; rightMargin: 8 } + columns: 2 + columnSpacing: 12 + rowSpacing: 3 + Repeater { + model: rowItem.expanded ? rowItem.details : [] + delegate: Text { + required property var modelData + required property int index + Layout.row: index + Layout.column: 0 + text: modelData.key + color: theme.muted + font.pixelSize: 11 + } + } + Repeater { + model: rowItem.expanded ? rowItem.details : [] + delegate: Text { + required property var modelData + required property int index + Layout.row: index + Layout.column: 1 + Layout.fillWidth: true + text: modelData.value + color: theme.text + font.pixelSize: 11 + elide: Text.ElideMiddle + } + } + } + Rectangle { + id: copyButton + anchors { right: parent.right; top: parent.top; rightMargin: 8 } + width: 28; height: 28; radius: 6 + color: copyArea.containsMouse ? Qt.rgba(theme.text.r, theme.text.g, theme.text.b, 0.08) + : "transparent" + ToolTip.visible: copyArea.containsMouse + ToolTip.text: copied ? "Copied" : "Copy details" + property bool copied: false + Image { + anchors.centerIn: parent + sourceSize: Qt.size(16, 16) + source: root.iconUrl(copyButton.copied ? "check" : "content_copy", + theme.muted, false) + } + MouseArea { + id: copyArea + anchors.fill: parent + hoverEnabled: true + cursorShape: Qt.PointingHandCursor + onClicked: { + statusModel.copyDetails(rowItem.serviceName) + copyButton.copied = true + copiedTimer.restart() + } + } + Timer { id: copiedTimer; interval: 1500; onTriggered: copyButton.copied = false } + } + } + } + } + } +} diff --git a/bec_widgets/widgets/services/bec_status_box/status_box_widgets.py b/bec_widgets/widgets/services/bec_status_box/status_box_widgets.py new file mode 100644 index 00000000..6be23e99 --- /dev/null +++ b/bec_widgets/widgets/services/bec_status_box/status_box_widgets.py @@ -0,0 +1,515 @@ +"""QWidget view of the BEC status box. + +It shows the same content and interactions as the QML view (``qml/StatusBox.qml``): an overall state +header and grouped, expandable service rows. Rows are kept and updated in place when the model +changes, instead of being rebuilt. +""" + +from __future__ import annotations + +from bec_qthemes import material_icon +from qtpy.QtCore import QSize, Qt, QTimer, Signal +from qtpy.QtGui import QColor, QPainter, QTransform +from qtpy.QtWidgets import ( + QApplication, + QFrame, + QGridLayout, + QHBoxLayout, + QLabel, + QScrollArea, + QSizePolicy, + QToolButton, + QVBoxLayout, + QWidget, +) + +from bec_widgets.utils.colors import get_theme_name, rgba +from bec_widgets.widgets.services.bec_status_box.status_model import ServiceRow, ServiceStatusModel + +_FALLBACK = { + "BG": "#1e1f22", + "CARD_BG": "#2b2c2f", + "FG": "#dddddd", + "BORDER": "#464749", + "DISABLED_FG": "#6f6f71", + "ACCENT_WARNING": "#eda200", + "ACCENT_EMERGENCY": "#e0534b", + "ACCENT_SUCCESS": "#59a869", +} + + +def theme_colors() -> dict[str, QColor]: + """Return the theme colors used by the status box, with dark-theme fallbacks.""" + theme = getattr(QApplication.instance(), "theme", None) + if theme is None: + return {key: QColor(value) for key, value in _FALLBACK.items()} + return {key: theme.color(key, value) for key, value in _FALLBACK.items()} + + +def tone_color(colors: dict[str, QColor], tone: str) -> QColor: + """Map a tone name to a theme color.""" + return { + "success": colors["ACCENT_SUCCESS"], + "warning": colors["ACCENT_WARNING"], + "emergency": colors["ACCENT_EMERGENCY"], + }.get(tone, colors["DISABLED_FG"]) + + +class _Dot(QWidget): + """A small filled circle that can pulse.""" + + def __init__(self, parent=None): + super().__init__(parent) + self.setFixedSize(12, 12) + self._color = QColor("gray") + self._opacity = 1.0 + + def set_color(self, color: QColor) -> None: + self._color = QColor(color) + self.update() + + def set_opacity(self, opacity: float) -> None: + self._opacity = opacity + self.update() + + def paintEvent(self, _event): + painter = QPainter(self) + painter.setRenderHint(QPainter.RenderHint.Antialiasing) + painter.setOpacity(self._opacity) + painter.setPen(Qt.PenStyle.NoPen) + painter.setBrush(self._color) + painter.drawEllipse(self.rect()) + + +class ElidedLabel(QLabel): + """A label that elides its text instead of growing the layout.""" + + def __init__(self, parent=None, mode=Qt.TextElideMode.ElideRight): + super().__init__(parent) + self._mode = mode + self.setSizePolicy(QSizePolicy.Policy.Ignored, QSizePolicy.Policy.Preferred) + self.setMinimumWidth(0) + + def paintEvent(self, _event): + painter = QPainter(self) + painter.setPen(self.palette().color(self.foregroundRole())) + rect = self.contentsRect() + text = self.fontMetrics().elidedText(self.text(), self._mode, rect.width()) + painter.drawText(rect, int(self.alignment()), text) + + +class StatusHeader(QFrame): + """Overall state of BEC: a colored dot, a headline and a detail line.""" + + def __init__(self, parent=None): + super().__init__(parent) + self.setObjectName("StatusHeader") + self.setFixedHeight(56) + layout = QHBoxLayout(self) + layout.setContentsMargins(12, 0, 12, 0) + layout.setSpacing(10) + self.dot = _Dot(self) + text = QVBoxLayout() + text.setSpacing(2) + self.headline = ElidedLabel(self) + self.headline.setObjectName("headline") + self.detail = ElidedLabel(self) + self.detail.setObjectName("detail") + text.addStretch() + text.addWidget(self.headline) + text.addWidget(self.detail) + text.addStretch() + layout.addWidget(self.dot) + layout.addLayout(text, 1) + self._tone = "neutral" + self._pulse_up = False + self._pulse = QTimer(self) + self._pulse.setInterval(50) + self._pulse.timeout.connect(self._step_pulse) + + def set_summary(self, headline: str, detail: str, tone: str) -> None: + """Show the summary of the model.""" + self.headline.setText(headline) + self.detail.setText(detail) + self._tone = tone + if tone == "emergency": + self._pulse.start() + else: + self._pulse.stop() + self.dot.set_opacity(1.0) + self.apply_theme() + + def _step_pulse(self): + step = 0.65 / 14 + opacity = self.dot._opacity + (step if self._pulse_up else -step) + if opacity <= 0.35 or opacity >= 1.0: + self._pulse_up = not self._pulse_up + self.dot.set_opacity(min(1.0, max(0.35, opacity))) + + def apply_theme(self) -> None: + """Restyle with the current theme colors.""" + colors = theme_colors() + accent = tone_color(colors, self._tone) + alpha = 41 if get_theme_name() != "light" else 31 + self.dot.set_color(accent) + self.setStyleSheet( + f"#StatusHeader {{ background: {rgba(accent, alpha)}; border: 1px solid " + f"{rgba(accent, 115)}; border-radius: 8px; }}" + f"#headline {{ color: {colors['FG'].name()}; font-size: 14px; font-weight: bold; }}" + f"#detail {{ color: {colors['DISABLED_FG'].name()}; font-size: 11px; }}" + ) + + def stop(self) -> None: + """Stop the pulse timer.""" + self._pulse.stop() + + +class ServiceRowWidget(QFrame): + """One service: status icon, name, subtitle, status chip and expandable details.""" + + toggled = Signal(str) + copy_requested = Signal(str) + + def __init__(self, row: ServiceRow, parent=None): + super().__init__(parent) + self.setObjectName("ServiceRow") + self.setCursor(Qt.CursorShape.PointingHandCursor) + self.setAttribute(Qt.WidgetAttribute.WA_Hover) + self.row = None + self.expanded = False + self._hovered = False + self._angle = 0 + + outer = QVBoxLayout(self) + outer.setContentsMargins(0, 0, 0, 0) + outer.setSpacing(0) + + main = QWidget(self) + main.setFixedHeight(44) + line = QHBoxLayout(main) + line.setContentsMargins(8, 0, 8, 0) + line.setSpacing(10) + self.icon = QLabel(main) + self.icon.setFixedSize(20, 20) + names = QVBoxLayout() + names.setSpacing(1) + self.name = ElidedLabel(main) + self.name.setObjectName("name") + self.subtitle = ElidedLabel(main) + self.subtitle.setObjectName("subtitle") + names.addStretch() + names.addWidget(self.name) + names.addWidget(self.subtitle) + names.addStretch() + self.mismatch = QLabel(main) + self.mismatch.setObjectName("mismatch") + self.chip = QLabel(main) + self.chip.setObjectName("chip") + self.chip.setAlignment(Qt.AlignmentFlag.AlignCenter) + self.chip.setFixedHeight(20) + self.mismatch.setAlignment(Qt.AlignmentFlag.AlignCenter) + self.mismatch.setFixedHeight(18) + self.chevron = QLabel(main) + self.chevron.setFixedSize(16, 16) + line.addWidget(self.icon) + line.addLayout(names, 1) + line.addWidget(self.mismatch) + line.addWidget(self.chip) + line.addWidget(self.chevron) + outer.addWidget(main) + + self.details = QWidget(self) + details_layout = QHBoxLayout(self.details) + details_layout.setContentsMargins(38, 0, 8, 8) + self.details_grid = QGridLayout() + self.details_grid.setHorizontalSpacing(12) + self.details_grid.setVerticalSpacing(3) + self.copy_button = QToolButton(self.details) + self.copy_button.setToolTip("Copy details") + self.copy_button.setAutoRaise(True) + self.copy_button.setIconSize(QSize(16, 16)) + self.copy_button.clicked.connect(lambda: self.copy_requested.emit(self.row.service_name)) + self.copy_button.clicked.connect(self._show_copied) + details_layout.addLayout(self.details_grid, 1) + details_layout.addWidget(self.copy_button, 0, Qt.AlignmentFlag.AlignTop) + self.details.setVisible(False) + outer.addWidget(self.details) + self._detail_labels: list[tuple[QLabel, QLabel]] = [] + self._copied = False + self.set_row(row) + + def set_row(self, row: ServiceRow) -> None: + """Update the widget for a new version of its row, restyling only when the state changed.""" + previous = getattr(self, "row", None) + restyle = previous is None or (previous.tone, previous.icon) != (row.tone, row.icon) + self.row = row + self.name.setText(row.display_name) + self.subtitle.setText(row.subtitle) + self.subtitle.setVisible(bool(row.subtitle)) + self.chip.setText(row.status_label) + self.mismatch.setText(f"v{row.version}") + self.mismatch.setToolTip(f"bec_lib {row.version} differs from the other services") + self.mismatch.setVisible(row.version_mismatch) + if self.expanded: + self._fill_details() + if restyle: + self.apply_theme() + + @property + def busy(self) -> bool: + """Whether the status icon spins.""" + return self.row.status == "BUSY" + + def set_expanded(self, expanded: bool) -> None: + """Show or hide the details.""" + self.expanded = expanded + if expanded: + self._fill_details() + self.details.setVisible(expanded) + self.apply_theme() + + def _fill_details(self): + details = self.row.details + while len(self._detail_labels) < len(details): + key, value = QLabel(self.details), QLabel(self.details) + key.setObjectName("detailKey") + value.setObjectName("detailValue") + value.setSizePolicy(QSizePolicy.Policy.Ignored, QSizePolicy.Policy.Preferred) + value.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse) + idx = len(self._detail_labels) + self.details_grid.addWidget(key, idx, 0) + self.details_grid.addWidget(value, idx, 1) + self._detail_labels.append((key, value)) + for idx, (key, value) in enumerate(self._detail_labels): + visible = idx < len(details) + key.setVisible(visible) + value.setVisible(visible) + if visible: + key.setText(details[idx][0]) + value.setText(details[idx][1]) + self.details_grid.setColumnStretch(1, 1) + + def _show_copied(self): + self._copied = True + self.copy_button.setToolTip("Copied") + self._paint_copy_icon() + QTimer.singleShot(1500, self._reset_copied) + + def _reset_copied(self): + self._copied = False + self.copy_button.setToolTip("Copy details") + self._paint_copy_icon() + + def _paint_copy_icon(self): + colors = theme_colors() + self.copy_button.setIcon( + material_icon( + "check" if self._copied else "content_copy", + size=(16, 16), + color=colors["DISABLED_FG"], + convert_to_pixmap=False, + ) + ) + + def rotate_icon(self) -> None: + """Advance the busy spinner by one step.""" + self._angle = (self._angle + 30) % 360 + self._paint_status_icon() + + def _paint_status_icon(self): + accent = tone_color(theme_colors(), self.row.tone) + pixmap = material_icon(self.row.icon, size=(20, 20), color=accent, filled=True) + if self._angle and self.busy: + rotated = pixmap.transformed( + QTransform().rotate(self._angle), Qt.TransformationMode.SmoothTransformation + ) + # crop back to the original size around the center + x = (rotated.width() - pixmap.width()) // 2 + y = (rotated.height() - pixmap.height()) // 2 + pixmap = rotated.copy(x, y, pixmap.width(), pixmap.height()) + self.icon.setPixmap(pixmap) + + def apply_theme(self) -> None: + """Restyle with the current theme colors.""" + colors = theme_colors() + accent = tone_color(colors, self.row.tone) + if not self.busy: + self._angle = 0 + self._paint_status_icon() + self._paint_copy_icon() + self.chevron.setPixmap( + material_icon( + "expand_less" if self.expanded else "expand_more", + size=(16, 16), + color=colors["DISABLED_FG"], + ) + ) + highlighted = self.expanded or self._hovered + background = colors["CARD_BG"].name() if highlighted else "transparent" + border = colors["BORDER"].name() if self.expanded else "transparent" + self.setStyleSheet( + f"#ServiceRow {{ background: {background}; border: 1px solid {border};" + f" border-radius: 6px; }}" + f"#name {{ color: {colors['FG'].name()}; font-size: 13px; }}" + f"#subtitle, #detailKey {{ color: {colors['DISABLED_FG'].name()}; font-size: 11px; }}" + f"#detailValue {{ color: {colors['FG'].name()}; font-size: 11px; }}" + f"#chip {{ color: {accent.name()}; background: {rgba(accent, 38)};" + f" border-radius: 10px; padding: 0px 8px; font-size: 11px; font-weight: bold; }}" + f"#mismatch {{ color: {colors['ACCENT_WARNING'].name()}; border: 1px solid" + f" {colors['ACCENT_WARNING'].name()}; border-radius: 9px; padding: 0px 6px;" + f" font-size: 10px; }}" + ) + + def event(self, event): + if event.type() == event.Type.HoverEnter: + self._hovered = True + self.apply_theme() + elif event.type() == event.Type.HoverLeave: + self._hovered = False + self.apply_theme() + return super().event(event) + + def mouseReleaseEvent(self, event): + if event.button() == Qt.MouseButton.LeftButton and event.position().y() <= 44: + self.toggled.emit(self.row.service_name) + super().mouseReleaseEvent(event) + + +class StatusBoxWidgetView(QWidget): + """QWidget view bound to a ServiceStatusModel.""" + + def __init__(self, model: ServiceStatusModel, parent=None): + super().__init__(parent) + self.model = model + self.expanded_service = "" + self.rows: dict[str, ServiceRowWidget] = {} + self.section_labels: dict[str, QLabel] = {} + + layout = QVBoxLayout(self) + layout.setContentsMargins(8, 8, 8, 4) + layout.setSpacing(4) + self.header = StatusHeader(self) + layout.addWidget(self.header) + + self.scroll = QScrollArea(self) + self.scroll.setWidgetResizable(True) + self.scroll.setFrameShape(QFrame.Shape.NoFrame) + self.scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff) + self.list_widget = QWidget(self.scroll) + self.list_widget.setObjectName("serviceList") + self.list_layout = QVBoxLayout(self.list_widget) + self.list_layout.setContentsMargins(0, 0, 0, 0) + self.list_layout.setSpacing(2) + self.list_layout.addStretch() + self.scroll.setWidget(self.list_widget) + layout.addWidget(self.scroll, 1) + self.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Expanding) + + self._spinner = QTimer(self) + self._spinner.setInterval(100) + self._spinner.timeout.connect(self._spin) + + model.rowsChanged.connect(self.sync) + model.summaryChanged.connect(self._update_summary) + self._update_summary() + self.sync() + self.apply_theme() + + def _update_summary(self): + self.header.set_summary(self.model.headline, self.model.detail, self.model.tone) + + def sync(self) -> None: + """Bring the row widgets in line with the model, reusing existing widgets.""" + rows = self.model.rows + names = {row.service_name for row in rows} + for name in list(self.rows): + if name not in names: + widget = self.rows.pop(name) + self.list_layout.removeWidget(widget) + widget.deleteLater() + if self.expanded_service not in names: + self.expanded_service = "" + + position = 0 + current_group = None + for row in rows: + if row.group != current_group: + current_group = row.group + label = self._section_label(row.group) + self._place(label, position) + position += 1 + widget = self.rows.get(row.service_name) + if widget is None: + widget = ServiceRowWidget(row, self.list_widget) + widget.toggled.connect(self._toggle) + widget.copy_requested.connect(self.model.copyDetails) + self.rows[row.service_name] = widget + elif widget.row != row: + widget.set_row(row) + self._place(widget, position) + position += 1 + groups = {row.group for row in rows} + for group, label in self.section_labels.items(): + label.setVisible(group in groups) + + if any(widget.busy for widget in self.rows.values()): + if not self._spinner.isActive(): + self._spinner.start() + else: + self._spinner.stop() + + def _place(self, widget: QWidget, position: int): + if self.list_layout.indexOf(widget) != position: + self.list_layout.removeWidget(widget) + self.list_layout.insertWidget(position, widget) + widget.setVisible(True) + + def _section_label(self, group: str) -> QLabel: + label = self.section_labels.get(group) + if label is None: + label = QLabel(group.upper(), self.list_widget) + label.setObjectName("section") + label.setContentsMargins(4, 10, 0, 4) + self.section_labels[group] = label + self._style_section(label) + return label + + def _style_section(self, label: QLabel): + colors = theme_colors() + label.setStyleSheet( + f"#section {{ color: {colors['DISABLED_FG'].name()}; font-size: 10px;" + f" font-weight: bold; letter-spacing: 0.8px; }}" + ) + + def _toggle(self, service_name: str): + previous = self.rows.get(self.expanded_service) + if previous is not None: + previous.set_expanded(False) + if self.expanded_service == service_name: + self.expanded_service = "" + return + self.expanded_service = service_name + self.rows[service_name].set_expanded(True) + + def _spin(self): + for widget in self.rows.values(): + if widget.busy: + widget.rotate_icon() + + def apply_theme(self) -> None: + """Restyle all parts with the current theme colors.""" + colors = theme_colors() + self.setStyleSheet(f"StatusBoxWidgetView {{ background: {colors['BG'].name()}; }}") + self.scroll.setStyleSheet( + "QScrollArea { background: transparent; } #serviceList { background: transparent; }" + ) + self.header.apply_theme() + for label in self.section_labels.values(): + self._style_section(label) + for widget in self.rows.values(): + widget.apply_theme() + + def cleanup(self) -> None: + """Stop the timers.""" + self._spinner.stop() + self.header.stop() diff --git a/bec_widgets/widgets/services/bec_status_box/status_model.py b/bec_widgets/widgets/services/bec_status_box/status_model.py new file mode 100644 index 00000000..d7ed3107 --- /dev/null +++ b/bec_widgets/widgets/services/bec_status_box/status_model.py @@ -0,0 +1,419 @@ +"""View-independent data layer for the BEC status box. + +``ServiceStatusModel`` turns the raw service info of the BEC client into display rows. The QML view +binds to it as a list model, and the QWidget view reads the same rows, so both views show exactly +the same content. +""" + +from __future__ import annotations + +import re +import time +from collections import Counter +from dataclasses import dataclass, field +from datetime import datetime + +from qtpy.QtCore import ( + Property, + QAbstractListModel, + QByteArray, + QModelIndex, + QObject, + Qt, + Signal, + Slot, +) +from qtpy.QtWidgets import QApplication + +# status -> (label, tone, material icon) +STATUS_STYLE = { + "RUNNING": ("Running", "success", "check_circle"), + "BUSY": ("Busy", "success", "progress_activity"), + "IDLE": ("Idle", "warning", "pause_circle"), + "ERROR": ("Error", "emergency", "error"), + "NOTCONNECTED": ("Not connected", "emergency", "cloud_off"), +} + +FRIENDLY_NAMES = { + "DeviceServer": "Device server", + "ScanServer": "Scan server", + "ScanBundler": "Scan bundler", + "FileWriterManager": "File writer", + "SciHub": "SciHub", + "DAPServer": "Data processing", + "CLIBECClient": "IPython client", + "BECIPythonClient": "IPython client", + "BECClient": "Client", + "BECGuiClient": "GUI client", +} + + +def friendly_name(service_name: str) -> tuple[str, str]: + """Return a readable name and a short instance id for a BEC service name. + + Args: + service_name (str): The raw service name, e.g. "DeviceServer" or "CLIBECClient/". + + Returns: + tuple[str, str]: The display name and the short instance id ("" for unique services). + """ + base, _, instance = service_name.partition("/") + name = FRIENDLY_NAMES.get(base) + if name is None: + name = re.sub(r"(?<=[a-z])(?=[A-Z])", " ", base) + name = name[:1].upper() + name[1:].lower() if name else service_name + return name, instance[:6] + + +def format_duration(seconds: float) -> str: + """Format a duration in seconds as a short human readable string. + + Args: + seconds (float): The duration. + + Returns: + str: For example "45 s", "12 min", "3 h 5 min" or "2 d 4 h". + """ + seconds = max(0, int(seconds)) + if seconds < 60: + return f"{seconds} s" + minutes = seconds // 60 + if minutes < 60: + return f"{minutes} min" + hours, minutes = divmod(minutes, 60) + if hours < 24: + return f"{hours} h {minutes} min" + days, hours = divmod(hours, 24) + return f"{days} d {hours} h" + + +@dataclass +class ServiceRow: + """One display row of the status box.""" + + service_name: str + display_name: str + instance: str + group: str + status: str + status_label: str + tone: str + icon: str + host: str = "" + user: str = "" + version: str = "" + version_mismatch: bool = False + uptime: str = "" + cpu: str = "" + memory: str = "" + details: list[tuple[str, str]] = field(default_factory=list) + + @property + def subtitle(self) -> str: + """The secondary line: who runs the service and where.""" + parts = [] + if self.user and self.host: + parts.append(f"{self.user}@{self.host}") + elif self.host: + parts.append(self.host) + if self.instance: + parts.append(f"#{self.instance}") + if not parts and self.status == "NOTCONNECTED": + parts.append("No heartbeat from this service") + return " · ".join(parts) + + def details_text(self) -> str: + """Plain text version of the details, used for the copy button.""" + lines = [f"{self.service_name}: {self.status_label}"] + lines += [f"{key}: {value}" for key, value in self.details] + return "\n".join(lines) + + +CORE_GROUP = "Core services" +OTHER_GROUP = "Clients and other services" + + +def build_rows(services, core_services: list[str], now: float | None = None) -> list[ServiceRow]: + """Build display rows from the status box's service info containers. + + Args: + services: Iterable of BECServiceInfoContainer-like objects (service_name, status, info, + metrics). + core_services (list[str]): Names of the core services, shown first in their own group. + now (float | None): Current time, for uptime. Defaults to time.time(). + + Returns: + list[ServiceRow]: Core services in the given order, then all other services by name. + """ + now = time.time() if now is None else now + services = list(services) + rows = [] + for svc in services: + info = svc.info or {} + metrics = svc.metrics or {} + status = svc.status if svc.status in STATUS_STYLE else "NOTCONNECTED" + label, tone, icon = STATUS_STYLE[status] + name, instance = friendly_name(svc.service_name) + versions = info.get("versions") or {} + version = versions.get("bec_lib", "") or info.get("version", "") + row = ServiceRow( + service_name=svc.service_name, + display_name=name, + instance=instance, + group=CORE_GROUP if svc.service_name in core_services else OTHER_GROUP, + status=status, + status_label=label, + tone=tone, + icon=icon, + host=str(info.get("hostname") or metrics.get("hostname") or ""), + user=str(info.get("user") or metrics.get("username") or ""), + version=str(version), + ) + if "create_time" in metrics: + row.uptime = format_duration(now - metrics["create_time"]) + if "cpu_percent" in metrics: + row.cpu = f"{metrics['cpu_percent']:.0f} %" + if "memory_in_mb" in metrics: + row.memory = f"{metrics['memory_in_mb']:.0f} MB" + details = [("Service", svc.service_name)] + for key, value in ( + ("Host", row.host), + ("User", row.user), + ("PID", metrics.get("pid", "")), + ("Uptime", row.uptime), + ("CPU", row.cpu), + ("Memory", row.memory), + ("Threads", metrics.get("num_threads", "")), + ): + if value not in ("", None): + details.append((key, str(value))) + if "create_time" in metrics: + started = datetime.fromtimestamp(metrics["create_time"]).strftime("%Y-%m-%d %H:%M:%S") + details.append(("Started", started)) + for package, pkg_version in versions.items(): + details.append((package, str(pkg_version))) + row.details = details + rows.append(row) + + # flag services whose bec_lib version differs from the one most services run + versions = Counter(row.version for row in rows if row.version) + if len(versions) > 1: + common = versions.most_common(1)[0][0] + for row in rows: + row.version_mismatch = bool(row.version) and row.version != common + + core_order = {name: idx for idx, name in enumerate(core_services)} + rows.sort( + key=lambda r: ( + r.group != CORE_GROUP, + core_order.get(r.service_name, 0), + r.display_name, + r.instance, + ) + ) + return rows + + +def summarize(rows: list[ServiceRow], core_state: str) -> tuple[str, str, str]: + """Summarize the overall state for the header. + + Args: + rows (list[ServiceRow]): The display rows. + core_state (str): The combined state of the core services. + + Returns: + tuple[str, str, str]: (headline, detail line, tone). + """ + core = [row for row in rows if row.group == CORE_GROUP] + others = [row for row in rows if row.group != CORE_GROUP] + down = [row for row in core if row.status in ("NOTCONNECTED", "ERROR")] + idle = [row for row in core if row.status == "IDLE"] + if down: + if len(down) == 1: + headline = f"{down[0].display_name}: {down[0].status_label.lower()}" + else: + headline = f"{len(down)} core services down" + tone = "emergency" + elif idle: + headline = ( + f"{idle[0].display_name} is idle" if len(idle) == 1 else f"{len(idle)} services idle" + ) + tone = "warning" + elif core and core_state in ("RUNNING", "BUSY"): + headline = "All core services running" + tone = "success" + else: + headline = "Waiting for BEC services" + tone = "neutral" + ok = sum(1 for row in core if row.tone == "success") + detail = f"{ok}/{len(core)} core services up" + if others: + detail += f" · {len(others)} other" if len(others) == 1 else f" · {len(others)} others" + mismatched = [row for row in rows if row.version_mismatch] + if mismatched: + detail += f" · {len(mismatched)} version mismatch" + return headline, detail, tone + + +class ServiceStatusModel(QAbstractListModel): + """List model of BEC services with a summary, shared by the QML and QWidget views.""" + + summaryChanged = Signal() + rowsChanged = Signal() + + ROLE_NAMES = ( + "serviceName", + "displayName", + "subtitle", + "group", + "status", + "statusLabel", + "tone", + "icon", + "version", + "versionMismatch", + "uptime", + "cpu", + "memory", + "details", + ) + + def __init__(self, parent: QObject | None = None): + super().__init__(parent) + self._rows: list[ServiceRow] = [] + self._headline = "Waiting for BEC services" + self._detail = "" + self._tone = "neutral" + self._roles = {Qt.ItemDataRole.UserRole + i: n for i, n in enumerate(self.ROLE_NAMES)} + self._role_ids = {name: role for role, name in self._roles.items()} + + # ---- QAbstractListModel ------------------------------------------------------------------- + + def rowCount(self, parent: QModelIndex = QModelIndex()) -> int: # noqa: B008 + if parent.isValid(): + return 0 + return len(self._rows) + + def roleNames(self) -> dict[int, QByteArray]: + return {role: QByteArray(name.encode()) for role, name in self._roles.items()} + + def data(self, index: QModelIndex, role: int = Qt.ItemDataRole.DisplayRole): + if not index.isValid() or not 0 <= index.row() < len(self._rows): + return None + row = self._rows[index.row()] + if role == Qt.ItemDataRole.DisplayRole: + return row.display_name + name = self._roles.get(role) + if name is None: + return None + return self._value(row, name) + + @staticmethod + def _value(row: ServiceRow, name: str): + match name: + case "serviceName": + return row.service_name + case "displayName": + return row.display_name + case "subtitle": + return row.subtitle + case "group": + return row.group + case "status": + return row.status + case "statusLabel": + return row.status_label + case "tone": + return row.tone + case "icon": + return row.icon + case "version": + return row.version + case "versionMismatch": + return row.version_mismatch + case "uptime": + return row.uptime + case "cpu": + return row.cpu + case "memory": + return row.memory + case "details": + return [{"key": key, "value": value} for key, value in row.details] + return None + + # ---- public API --------------------------------------------------------------------------- + + @property + def rows(self) -> list[ServiceRow]: + """The current display rows.""" + return list(self._rows) + + def set_rows(self, rows: list[ServiceRow], core_state: str) -> None: + """Replace the rows, emitting fine-grained model signals instead of a reset. + + Args: + rows (list[ServiceRow]): The new rows, already sorted. + core_state (str): The combined state of the core services. + """ + new_keys = [row.service_name for row in rows] + # remove rows that disappeared + for idx in reversed(range(len(self._rows))): + if self._rows[idx].service_name not in new_keys: + self.beginRemoveRows(QModelIndex(), idx, idx) + del self._rows[idx] + self.endRemoveRows() + # insert or move rows into place, update changed ones + for target, row in enumerate(rows): + current = [r.service_name for r in self._rows] + if row.service_name not in current: + self.beginInsertRows(QModelIndex(), target, target) + self._rows.insert(target, row) + self.endInsertRows() + continue + source = current.index(row.service_name) + if source != target: + self.beginMoveRows(QModelIndex(), source, source, QModelIndex(), target) + self._rows.insert(target, self._rows.pop(source)) + self.endMoveRows() + if self._rows[target] != row: + changed = [ + self._role_ids[name] + for name in self.ROLE_NAMES + if self._value(self._rows[target], name) != self._value(row, name) + ] + self._rows[target] = row + index = self.index(target) + self.dataChanged.emit(index, index, changed) + + headline, detail, tone = summarize(self._rows, core_state) + if (headline, detail, tone) != (self._headline, self._detail, self._tone): + self._headline, self._detail, self._tone = headline, detail, tone + self.summaryChanged.emit() + self.rowsChanged.emit() + + @Property(str, notify=summaryChanged) + def headline(self) -> str: + """The one-line overall state, e.g. "All core services running".""" + return self._headline + + @Property(str, notify=summaryChanged) + def detail(self) -> str: + """The secondary summary line with counts.""" + return self._detail + + @Property(str, notify=summaryChanged) + def tone(self) -> str: + """The overall tone: success, warning, emergency or neutral.""" + return self._tone + + @Slot(str) + def copyDetails(self, service_name: str) -> None: + """Copy the details of a service to the clipboard. + + Args: + service_name (str): The raw service name. + """ + for row in self._rows: + if row.service_name == service_name: + QApplication.clipboard().setText(row.details_text()) + return diff --git a/tests/unit_tests/test_bec_status_box.py b/tests/unit_tests/test_bec_status_box.py index b7d93c5d..e6e8cd9e 100644 --- a/tests/unit_tests/test_bec_status_box.py +++ b/tests/unit_tests/test_bec_status_box.py @@ -1,47 +1,86 @@ # pylint: skip-file +import time from unittest import mock import pytest from bec_lib.messages import BECStatus, ServiceInfo, ServiceMetricMessage, StatusMessage +from qtpy.QtQuickWidgets import QQuickWidget +from qtpy.QtWidgets import QApplication from bec_widgets.tests.client_mocks import mocked_client +from bec_widgets.tests.utils import create_widget from bec_widgets.widgets.services.bec_status_box.bec_status_box import ( BECServiceInfoContainer, BECStatusBox, ) +from bec_widgets.widgets.services.bec_status_box.status_box_widgets import StatusBoxWidgetView +from bec_widgets.widgets.services.bec_status_box.status_model import ( + CORE_GROUP, + OTHER_GROUP, + build_rows, + format_duration, + friendly_name, +) -@pytest.fixture -def service_status_fixture(): - yield mock.MagicMock() +def _versions(lib="3.39.5"): + return { + "bec_lib": lib, + "bec_server": lib, + "bec_ipython_client": lib, + "bec_widgets": "3.39.5", + "ophyd_devices": "1.51.0", + } -@pytest.fixture -def status_box(qtbot, mocked_client, service_status_fixture): - widget = BECStatusBox(client=mocked_client, bec_service_status_mixin=service_status_fixture) - qtbot.addWidget(widget) - qtbot.waitExposed(widget) +def _status(name, status=BECStatus.RUNNING, lib="3.39.5"): + info = ServiceInfo(user="e12345", hostname="console", versions=_versions(lib)) + return StatusMessage(name=name, status=status, info=info) + + +def _metrics(name, **extra): + metrics = {"pid": 42, "cpu_percent": 4.2, "memory_in_mb": 120.0, "create_time": time.time()} + metrics.update(extra) + return ServiceMetricMessage(name=name, metrics=metrics) + + +def _all_services(extra=(), skip=(), core_status=BECStatus.RUNNING): + names = [n for n in BECStatusBox.CORE_SERVICES if n not in skip] + list(extra) + info = { + n: _status(n, core_status if n in BECStatusBox.CORE_SERVICES else BECStatus.RUNNING) + for n in names + } + metrics = {n: _metrics(n) for n in names} + return info, metrics + + +@pytest.fixture(params=["qml", "widgets"]) +def status_box(request, qtbot, mocked_client): + widget = create_widget( + qtbot, + BECStatusBox, + client=mocked_client, + bec_service_status_mixin=mock.MagicMock(), + view=request.param, + ) + assert widget.view_kind == request.param yield widget +def test_initial_state(status_box): + assert status_box.get_server_state() == "IDLE" + assert status_box.model.rowCount() == 0 + assert status_box.model.headline == "Waiting for BEC services" + + +def test_view_type(status_box): + expected = QQuickWidget if status_box.view_kind == "qml" else StatusBoxWidgetView + assert isinstance(status_box.view, expected) + + def test_update_top_item(status_box): - assert status_box.tree.children()[0].children()[0].config.status == "IDLE" - name = status_box.box_name status_box.update_top_item_status(status="RUNNING") - assert status_box.status_container[name]["info"].status == "RUNNING" - assert status_box.tree.children()[0].children()[0].config.status == "RUNNING" - - -def test_create_status_widget(status_box): - name = "test_service" - status = BECStatus.IDLE - info = {"test": "test"} - metrics = {"metric": "test_metric"} - item = status_box._create_status_widget(name, status, info, metrics) - assert item.config.service_name == name - assert item.config.status == status.name - assert item.config.info == info - assert item.config.metrics == metrics + assert status_box.get_server_state() == "RUNNING" def test_bec_service_container(status_box): @@ -52,78 +91,169 @@ def test_bec_service_container(status_box): expected_return = BECServiceInfoContainer( service_name=name, status=status.name, info=info, metrics=metrics ) - assert status_box.box_name in status_box.status_container - assert len(status_box.status_container) == 1 - status_box._update_status_container(name, status, info, metrics) - assert len(status_box.status_container) == 2 - assert status_box.status_container[name]["info"] == expected_return + with mock.patch.object(status_box, "service_update") as service_update: + status_box._update_status_container(name, status, info, metrics) + service_update.emit.assert_called_once_with(expected_return) + assert status_box.status_container[name] == expected_return -def test_add_tree_item(status_box): - name = "test_service" - status = BECStatus.IDLE - info = {"test": "test"} - metrics = {"metric": "test_metric"} - assert len(status_box.tree.children()[0].children()) == 1 - status_box.add_tree_item(name, status, info, metrics) - assert len(status_box.tree.children()[0].children()) == 2 - assert name in status_box.status_container +def test_all_core_running(status_box): + status_box.update_service_status(*_all_services(extra=["CLIBECClient/abcdef123"])) + model = status_box.model + assert status_box.get_server_state() == "RUNNING" + assert model.headline == "All core services running" + assert model.tone == "success" + assert model.detail == "5/5 core services up · 1 other" + rows = model.rows + assert [r.service_name for r in rows[:5]] == BECStatusBox.CORE_SERVICES + assert rows[5].display_name == "IPython client" + assert rows[5].group == OTHER_GROUP + assert rows[5].subtitle == "e12345@console · #abcdef" -def test_update_service_status(status_box): - """Also checks check redundant tree items""" - name = "test_service" - status = BECStatus.IDLE - info = StatusMessage(name=name, status=status, info=ServiceInfo(user="test", hostname="host")) - metrics = {"metric": "test_metric"} - status_box.add_tree_item(name, status, info, {}) - not_connected_name = "invalid_service" - status_box.add_tree_item(not_connected_name, status, info, metrics) +def test_core_service_missing(status_box): + status_box.update_service_status(*_all_services(skip=["ScanBundler"])) + model = status_box.model + assert status_box.get_server_state() == "NOTCONNECTED" + assert model.headline == "Scan bundler: not connected" + assert model.tone == "emergency" + row = next(r for r in model.rows if r.service_name == "ScanBundler") + assert row.status_label == "Not connected" + assert row.subtitle == "No heartbeat from this service" - services_status = { - name: StatusMessage( - name=name, status=status, info=ServiceInfo(user="test", hostname="host") + +def test_compact_led_follows_summary(status_box): + with mock.patch.object(status_box, "set_global_state") as set_state: + status_box.update_service_status(*_all_services(skip=["SciHub"])) + set_state.assert_called_with("emergency") + status_box.update_service_status(*_all_services()) + set_state.assert_called_with("success") + + +def test_stale_services_removed(status_box): + status_box.update_service_status(*_all_services(extra=["CLIBECClient/1", "CLIBECClient/2"])) + assert status_box.model.rowCount() == 7 + status_box.update_service_status(*_all_services(extra=["CLIBECClient/2"])) + assert status_box.model.rowCount() == 6 + assert "CLIBECClient/1" not in status_box.status_container + + +def test_rows_updated_in_place(status_box): + """A status change emits dataChanged for one row instead of resetting the model.""" + status_box.update_service_status(*_all_services()) + model = status_box.model + resets, changes = [], [] + model.modelReset.connect(lambda: resets.append(True)) + model.dataChanged.connect(lambda first, last, roles: changes.append(first.row())) + info, metrics = _all_services() + info["ScanServer"] = _status("ScanServer", BECStatus.BUSY) + metrics = {k: v for k, v in metrics.items()} + status_box.update_service_status(info, metrics) + assert not resets + idx = BECStatusBox.CORE_SERVICES.index("ScanServer") + assert idx in changes + assert model.rows[idx].status_label == "Busy" + + +def test_version_mismatch(status_box): + info, metrics = _all_services(extra=["CLIBECClient/old"]) + info["CLIBECClient/old"] = _status("CLIBECClient/old", lib="3.0.0") + status_box.update_service_status(info, metrics) + mismatched = [r.service_name for r in status_box.model.rows if r.version_mismatch] + assert mismatched == ["CLIBECClient/old"] + assert "1 version mismatch" in status_box.model.detail + + +def test_copy_details(status_box): + status_box.update_service_status(*_all_services()) + status_box.model.copyDetails("DeviceServer") + text = QApplication.clipboard().text() + assert text.startswith("DeviceServer: Running") + assert "Host: console" in text + assert "PID: 42" in text + + +def test_widget_view_expand_and_reuse(qtbot, mocked_client): + box = create_widget( + qtbot, + BECStatusBox, + client=mocked_client, + bec_service_status_mixin=mock.MagicMock(), + view="widgets", + ) + box.update_service_status(*_all_services()) + view = box.view + row_widget = view.rows["DeviceServer"] + view._toggle("DeviceServer") + assert row_widget.expanded and not row_widget.details.isHidden() + view._toggle("ScanServer") + assert not row_widget.expanded + assert view.rows["ScanServer"].expanded + box.update_service_status(*_all_services()) + assert view.rows["DeviceServer"] is row_widget + assert view.rows["ScanServer"].expanded + + +def test_qml_view_loads(qtbot, mocked_client): + box = create_widget( + qtbot, + BECStatusBox, + client=mocked_client, + bec_service_status_mixin=mock.MagicMock(), + view="qml", + ) + box.update_service_status(*_all_services()) + root = box.view.rootObject() + assert root is not None + assert not box.view.errors() + service_list = root.findChild(object, "serviceList") + assert service_list.property("count") == 5 + root.setProperty("expandedService", "DeviceServer") + assert root.property("expandedService") == "DeviceServer" + + +def test_qml_falls_back_to_widgets(qtbot, mocked_client): + with mock.patch( + "bec_widgets.utils.qml_host.create_quick_widget", side_effect=RuntimeError("no GL") + ): + box = create_widget( + qtbot, + BECStatusBox, + client=mocked_client, + bec_service_status_mixin=mock.MagicMock(), + view="qml", ) - } - services_metrics = {name: ServiceMetricMessage(name=name, metrics=metrics)} - - with mock.patch.object(status_box, "update_core_services", return_value=services_status): - assert not_connected_name in status_box.status_container - status_box.update_service_status(services_status, services_metrics) - assert status_box.status_container[name]["widget"].config.metrics == metrics - assert not_connected_name not in status_box.status_container + assert box.view_kind == "widgets" + assert isinstance(box.view, StatusBoxWidgetView) -def test_update_core_services(status_box): - status_box.CORE_SERVICES = ["test_service"] - name = "test_service" - status = BECStatus.RUNNING - info = ServiceInfo(user="test", hostname="host") - metrics = {"metric": "test_metric"} - services_status = {name: StatusMessage(name=name, status=status, info=info)} - services_metrics = {name: ServiceMetricMessage(name=name, metrics=metrics)} - - status_box.update_core_services(services_status, services_metrics) - assert status_box.tree.children()[0].children()[0].config.status == "RUNNING" - assert status_box.status_container[name]["widget"].config.metrics == metrics - - status = BECStatus.IDLE - services_status = {name: StatusMessage(name=name, status=status, info=info)} - services_metrics = {name: ServiceMetricMessage(name=name, metrics=metrics)} - status_box.update_core_services(services_status, services_metrics) - assert status_box.tree.children()[0].children()[0].config.status == status.name - assert status_box.status_container[name]["widget"].config.metrics == metrics +def test_friendly_name(): + assert friendly_name("DeviceServer") == ("Device server", "") + assert friendly_name("CLIBECClient/0123456789") == ("IPython client", "012345") + assert friendly_name("MyNewService") == ("My new service", "") -def test_double_click_item(status_box): - name = "test_service" - status = BECStatus.IDLE - info = ServiceInfo(user="test", hostname="host") - metrics = {"MyData": "This should be shown nicely"} - status_box.add_tree_item(name, status, info, metrics) - container = status_box.status_container[name] - item = container["item"] - status_item = container["widget"] - with mock.patch.object(status_item, "show_popup") as mock_show_popup: - status_box.tree.itemDoubleClicked.emit(item, 0) - assert mock_show_popup.call_count == 1 +def test_format_duration(): + assert format_duration(5) == "5 s" + assert format_duration(125) == "2 min" + assert format_duration(3 * 3600 + 300) == "3 h 5 min" + assert format_duration(2 * 86400 + 4 * 3600) == "2 d 4 h" + + +def test_build_rows_order_and_groups(): + services = [ + BECServiceInfoContainer("CLIBECClient/b", "RUNNING", {}, None), + BECServiceInfoContainer("ScanServer", "ERROR", {}, None), + BECServiceInfoContainer("DeviceServer", "IDLE", {}, None), + BECServiceInfoContainer("Weird", "SOMETHING", {}, None), + ] + rows = build_rows(services, ["DeviceServer", "ScanServer"]) + assert [r.service_name for r in rows] == [ + "DeviceServer", + "ScanServer", + "CLIBECClient/b", + "Weird", + ] + assert [r.group for r in rows] == [CORE_GROUP, CORE_GROUP, OTHER_GROUP, OTHER_GROUP] + assert rows[1].tone == "emergency" + assert rows[3].status == "NOTCONNECTED" From e9223c6f0912a1b59ca1d0f4332516f453a494e3 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 9 Oct 2026 21:11:48 +0000 Subject: [PATCH 2/2] fix(bec_status_box): keep QWidget rows at natural height when the list is short Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_015TeMnCZ9FjLpwiDkPu6K9j --- .../widgets/services/bec_status_box/status_box_widgets.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/bec_widgets/widgets/services/bec_status_box/status_box_widgets.py b/bec_widgets/widgets/services/bec_status_box/status_box_widgets.py index 6be23e99..88a290f4 100644 --- a/bec_widgets/widgets/services/bec_status_box/status_box_widgets.py +++ b/bec_widgets/widgets/services/bec_status_box/status_box_widgets.py @@ -175,6 +175,8 @@ class ServiceRowWidget(QFrame): self.setObjectName("ServiceRow") self.setCursor(Qt.CursorShape.PointingHandCursor) self.setAttribute(Qt.WidgetAttribute.WA_Hover) + # keep rows at their natural height so short lists do not spread out + self.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Maximum) self.row = None self.expanded = False self._hovered = False