bec/bec_widgets
1
0 Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-10-28 16:11:27 +01:00
Code Activity

merge: bring in qml-status-box-8n18jw for the system dock

Browse Source
This commit is contained in:
Claude committed 2026-10-09 21:34:34 +00:00
2 parents 1a65152554 e9223c6f09
commit 42d46a5dfa
7 files changed
+1731 -246
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

bec_widgets/cli/client.py
+1 -1
View File
@@ -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
bec_widgets/utils/qml_host.py
+173
View File
@@ -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/<name>?color=<hex>&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/<name>?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
bec_widgets/widgets/services/bec_status_box/bec_status_box.py
+116 -156
View File
@@ -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()
bec_widgets/widgets/services/bec_status_box/qml/StatusBox.qml
+286
View File
@@ -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 }
}
}
}
}
}
}
bec_widgets/widgets/services/bec_status_box/status_box_widgets.py
+517
View File
@@ -0,0 +1,517 @@
"""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)
# 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
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()
bec_widgets/widgets/services/bec_status_box/status_model.py
+419
View File
@@ -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/<uuid>".
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
tests/unit_tests/test_bec_status_box.py
+219 -89
View File
@@ -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"