1
0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-03-04 07:52:55 +01:00

fix(bec_status): adjust bec status widget to info and version signature

This commit is contained in:
2026-01-23 16:18:52 +01:00
committed by Klaus Wakonig
parent d577fac02f
commit 709ffd6927
3 changed files with 27 additions and 10 deletions

View File

@@ -9,6 +9,7 @@ from dataclasses import dataclass
from typing import TYPE_CHECKING
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
@@ -18,9 +19,10 @@ from bec_widgets.widgets.services.bec_status_box.status_item import StatusItem
if TYPE_CHECKING: # pragma: no cover
from bec_lib.client import BECClient
# TODO : Put normal imports back when Pydantic gets faster
BECStatus = lazy_import_from("bec_lib.messages", ("BECStatus",))
from bec_lib.messages import BECStatus, ServiceMetricMessage, StatusMessage
else:
# TODO : Put normal imports back when Pydantic gets faster
BECStatus = lazy_import_from("bec_lib.messages", ("BECStatus",))
@dataclass
@@ -200,7 +202,11 @@ class BECStatusBox(BECWidget, CompactPopupWidget):
self.status_container[service_name].update({"info": service_info_item})
@Slot(dict, dict)
def update_service_status(self, services_info: dict, services_metric: dict) -> None:
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.
@@ -209,6 +215,9 @@ class BECStatusBox(BECWidget, CompactPopupWidget):
services_metric (dict): A dictionary containing the service metrics for all running BEC services.
"""
checked = [self.box_name]
# 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)

View File

@@ -141,7 +141,11 @@ class StatusItem(QWidget):
metrics_text = (
f"<b>SERVICE:</b> {self.config.service_name}<br><b>STATUS:</b> {self.config.status}<br>"
)
metrics_text += f"<b>BEC_LIB VERSION:</b> {self.config.info['version']}<br>"
if "version" in self.config.info:
metrics_text += f"<b>BEC_LIB VERSION:</b> {self.config.info['version']}<br>"
if "versions" in self.config.info:
for component, version in self.config.info["versions"].items():
metrics_text += f"<b>{component.upper()} VERSION:</b> {version}<br>"
if self.config.metrics:
for key, value in self.config.metrics.items():
if key == "create_time":

View File

@@ -2,7 +2,7 @@
from unittest import mock
import pytest
from bec_lib.messages import BECStatus, ServiceMetricMessage, StatusMessage
from bec_lib.messages import BECStatus, ServiceInfo, ServiceMetricMessage, StatusMessage
from bec_widgets.widgets.services.bec_status_box.bec_status_box import (
BECServiceInfoContainer,
@@ -75,13 +75,17 @@ def test_update_service_status(status_box):
"""Also checks check redundant tree items"""
name = "test_service"
status = BECStatus.IDLE
info = {"test": "test"}
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)
services_status = {name: StatusMessage(name=name, status=status, info=info)}
services_status = {
name: StatusMessage(
name=name, status=status, info=ServiceInfo(user="test", hostname="host")
)
}
services_metrics = {name: ServiceMetricMessage(name=name, metrics=metrics)}
with mock.patch.object(status_box, "update_core_services", return_value=services_status):
@@ -95,7 +99,7 @@ def test_update_core_services(status_box):
status_box.CORE_SERVICES = ["test_service"]
name = "test_service"
status = BECStatus.RUNNING
info = {"test": "test"}
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)}
@@ -115,7 +119,7 @@ def test_update_core_services(status_box):
def test_double_click_item(status_box):
name = "test_service"
status = BECStatus.IDLE
info = {"test": "test"}
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]