mirror of
https://github.com/bec-project/bec_widgets.git
synced 2026-04-09 10:10:55 +02:00
Compare commits
5 Commits
feature/sc
...
fix/refact
| Author | SHA1 | Date | |
|---|---|---|---|
| e2f074b1aa | |||
| 011103fde3 | |||
| f90bc00c18 | |||
| 63a0056388 | |||
| 5d435bd5ee |
@@ -5,15 +5,16 @@ The widget automatically updates the status of all running BEC services, and dis
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from collections import defaultdict
|
||||
from dataclasses import dataclass
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import qdarktheme
|
||||
from bec_lib.utils.import_utils import lazy_import_from
|
||||
from pydantic import BaseModel, Field, field_validator
|
||||
from qtpy.QtCore import QObject, QTimer, Signal, Slot
|
||||
from qtpy.QtWidgets import QTreeWidget, QTreeWidgetItem
|
||||
from qtpy.QtCore import Signal, Slot
|
||||
from qtpy.QtWidgets import QTreeWidget, QTreeWidgetItem, QVBoxLayout, QWidget
|
||||
|
||||
from bec_widgets.utils.bec_connector import BECConnector, ConnectionConfig
|
||||
from bec_widgets.utils.bec_connector import BECConnector
|
||||
from bec_widgets.widgets.bec_status_box.status_item import StatusItem
|
||||
|
||||
if TYPE_CHECKING:
|
||||
@@ -21,77 +22,25 @@ if TYPE_CHECKING:
|
||||
|
||||
# TODO : Put normal imports back when Pydantic gets faster
|
||||
BECStatus = lazy_import_from("bec_lib.messages", ("BECStatus",))
|
||||
StatusMessage = lazy_import_from("bec_lib.messages", ("StatusMessage",))
|
||||
|
||||
|
||||
class BECStatusBoxConfig(ConnectionConfig):
|
||||
pass
|
||||
|
||||
|
||||
class BECServiceInfoContainer(BaseModel):
|
||||
@dataclass
|
||||
class BECServiceInfoContainer:
|
||||
"""Container to store information about the BEC services."""
|
||||
|
||||
service_name: str
|
||||
status: BECStatus | str = Field(
|
||||
default="NOTCONNECTED",
|
||||
description="The status of the service. Can be any of the BECStatus names, or NOTCONNECTED.",
|
||||
)
|
||||
status: str
|
||||
info: dict
|
||||
metrics: dict | None
|
||||
model_config: dict = {"validate_assignment": True}
|
||||
|
||||
@field_validator("status")
|
||||
@classmethod
|
||||
def validate_status(cls, v):
|
||||
"""Validate input for status. Accept BECStatus and NOTCONNECTED.
|
||||
|
||||
Args:
|
||||
v (BECStatus | str): The input value.
|
||||
|
||||
Returns:
|
||||
str: The validated status.
|
||||
"""
|
||||
if v in list(BECStatus.__members__.values()):
|
||||
return v.name
|
||||
if v in list(BECStatus.__members__.keys()) or v == "NOTCONNECTED":
|
||||
return v
|
||||
raise ValueError(
|
||||
f"Status must be one of {BECStatus.__members__.values()} or 'NOTCONNECTED'. Input {v}"
|
||||
)
|
||||
|
||||
|
||||
class BECServiceStatusMixin(QObject):
|
||||
"""A mixin class to update the service status, and metrics.
|
||||
It emits a signal 'services_update' when the service status is updated.
|
||||
|
||||
Args:
|
||||
client (BECClient): The client object to connect to the BEC server.
|
||||
"""
|
||||
|
||||
services_update = Signal(dict, dict)
|
||||
|
||||
def __init__(self, client: BECClient):
|
||||
super().__init__()
|
||||
self.client = client
|
||||
self._service_update_timer = QTimer()
|
||||
self._service_update_timer.timeout.connect(self._get_service_status)
|
||||
self._service_update_timer.start(1000)
|
||||
|
||||
def _get_service_status(self):
|
||||
"""Pull latest service and metrics updates from REDIS for all services, and emit both via 'services_update' signal."""
|
||||
# pylint: disable=protected-access
|
||||
self.client._update_existing_services()
|
||||
self.services_update.emit(self.client._services_info, self.client._services_metric)
|
||||
|
||||
|
||||
class BECStatusBox(BECConnector, QTreeWidget):
|
||||
"""A widget to display the status of different BEC services.
|
||||
This widget automatically updates the status of all running BEC services, and displays their status.
|
||||
Information about the individual services is collapsible, and double clicking on
|
||||
the individual service will display the metrics about the service.
|
||||
class BECStatusBox(QWidget):
|
||||
"""An autonomous widget to display the status of BEC services.
|
||||
|
||||
Args:
|
||||
parent Optional : The parent widget for the BECStatusBox. Defaults to None.
|
||||
service_name Optional(str): The name of the top service label. Defaults to "BEC Server".
|
||||
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.
|
||||
gui_id Optional(str): The unique id for the widget. Defaults to None.
|
||||
@@ -99,58 +48,68 @@ class BECStatusBox(BECConnector, QTreeWidget):
|
||||
|
||||
CORE_SERVICES = ["DeviceServer", "ScanServer", "SciHub", "ScanBundler", "FileWriterManager"]
|
||||
|
||||
service_update = Signal(dict)
|
||||
service_update = Signal(BECServiceInfoContainer)
|
||||
bec_core_state = Signal(str)
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
parent=None,
|
||||
service_name: str = "BEC Server",
|
||||
box_name: str = "BEC Server",
|
||||
client: BECClient = None,
|
||||
config: BECStatusBoxConfig | dict = None,
|
||||
bec_service_status_mixin: BECServiceStatusMixin = None,
|
||||
gui_id: str = None,
|
||||
):
|
||||
if config is None:
|
||||
config = BECStatusBoxConfig(widget_class=self.__class__.__name__)
|
||||
else:
|
||||
if isinstance(config, dict):
|
||||
config = BECStatusBoxConfig(**config)
|
||||
super().__init__(client=client, config=config, gui_id=gui_id)
|
||||
QTreeWidget.__init__(self, parent=parent)
|
||||
QWidget.__init__(self, parent=parent)
|
||||
self.setLayout(QVBoxLayout(self))
|
||||
self.tree = QTreeWidget(self)
|
||||
self.layout().addWidget(self.tree)
|
||||
self.tree.setHeaderHidden(True)
|
||||
self.tree.setStyleSheet(
|
||||
"QTreeWidget::item:!selected "
|
||||
"{ "
|
||||
"border: 1px solid gainsboro; "
|
||||
"border-left: none; "
|
||||
"border-top: none; "
|
||||
"}"
|
||||
"QTreeWidget::item:selected {}"
|
||||
)
|
||||
self.box_name = box_name
|
||||
self.status_container = defaultdict(lambda: {"info": None, "item": None, "widget": None})
|
||||
|
||||
self.service_name = service_name
|
||||
self.config = config
|
||||
|
||||
self.bec_service_info_container = {}
|
||||
self.tree_items = {}
|
||||
self.tree_top_item = None
|
||||
|
||||
if not bec_service_status_mixin:
|
||||
bec_service_status_mixin = BECServiceStatusMixin(client=self.client)
|
||||
self.bec_service_status = bec_service_status_mixin
|
||||
self.connector = BECConnector(client=client, gui_id=gui_id)
|
||||
|
||||
self.init_ui()
|
||||
self.bec_service_status.services_update.connect(self.update_service_status)
|
||||
|
||||
self.bec_core_state.connect(self.update_top_item_status)
|
||||
self.itemDoubleClicked.connect(self.on_tree_item_double_clicked)
|
||||
self.tree.itemDoubleClicked.connect(self.on_tree_item_double_clicked)
|
||||
self.startTimer(
|
||||
1000
|
||||
) # use qobject's own timer instead of creating one, which may be stopped from another thread(?)
|
||||
|
||||
def timerEvent(self, event):
|
||||
"""Get the latest service status from the BEC server."""
|
||||
# pylint: disable=protected-access
|
||||
self.connector.client._update_existing_services()
|
||||
self.update_service_status(
|
||||
self.connector.client._services_info, self.connector.client._services_metric
|
||||
)
|
||||
|
||||
def init_ui(self) -> None:
|
||||
"""Initialize the UI for the status box, and add QTreeWidget as the basis for the status box."""
|
||||
self.init_ui_tree_widget()
|
||||
top_label = self._create_status_widget(self.service_name, status=BECStatus.IDLE)
|
||||
self.tree_top_item = QTreeWidgetItem()
|
||||
self.tree_top_item.setExpanded(True)
|
||||
self.tree_top_item.setDisabled(True)
|
||||
self.addTopLevelItem(self.tree_top_item)
|
||||
self.setItemWidget(self.tree_top_item, 0, top_label)
|
||||
"""Init the UI for the BECStatusBox widget"""
|
||||
top_label = self._create_status_widget(self.box_name, status=BECStatus.IDLE)
|
||||
tree_item = QTreeWidgetItem(self.tree)
|
||||
tree_item.setExpanded(True)
|
||||
tree_item.setDisabled(True)
|
||||
self.status_container[self.box_name].update({"item": tree_item, "widget": top_label})
|
||||
self.tree.setItemWidget(tree_item, 0, top_label)
|
||||
self.tree.addTopLevelItem(tree_item)
|
||||
self.service_update.connect(top_label.update_config)
|
||||
self._initialized = True
|
||||
|
||||
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 bec_service_info_container.
|
||||
information about the service in the status_container.
|
||||
|
||||
Args:
|
||||
service_name (str): The name of the service.
|
||||
@@ -163,16 +122,8 @@ class BECStatusBox(BECConnector, QTreeWidget):
|
||||
"""
|
||||
if info is None:
|
||||
info = {}
|
||||
self._update_bec_service_container(service_name, status, info, metrics)
|
||||
item = StatusItem(
|
||||
parent=self,
|
||||
config={
|
||||
"service_name": service_name,
|
||||
"status": status.name,
|
||||
"info": info,
|
||||
"metrics": metrics,
|
||||
},
|
||||
)
|
||||
self._update_status_container(service_name, status, info, metrics)
|
||||
item = StatusItem(parent=self.tree, config=self.status_container[service_name]["info"])
|
||||
return item
|
||||
|
||||
@Slot(str)
|
||||
@@ -183,30 +134,32 @@ class BECStatusBox(BECConnector, QTreeWidget):
|
||||
Args:
|
||||
status (BECStatus): The state of the core services.
|
||||
"""
|
||||
self.bec_service_info_container[self.service_name].status = status
|
||||
self.service_update.emit(self.bec_service_info_container[self.service_name].model_dump())
|
||||
self.status_container[self.box_name]["info"].status = status
|
||||
self.service_update.emit(self.status_container[self.box_name]["info"])
|
||||
|
||||
def _update_bec_service_container(
|
||||
def _update_status_container(
|
||||
self, service_name: str, status: BECStatus, info: dict, metrics: dict = None
|
||||
) -> None:
|
||||
"""Update the bec_service_info_container with the newest status and metrics for the BEC service.
|
||||
"""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.
|
||||
|
||||
Args:
|
||||
service_name (str): The name of the service.
|
||||
service_info (StatusMessage): A class containing the service status.
|
||||
service_metric (ServiceMetricMessage): A class containing the service metrics.
|
||||
status (BECStatus): The status of the service.
|
||||
info (dict): The information about the service.
|
||||
metrics (dict): The metrics of the service.
|
||||
"""
|
||||
container = self.bec_service_info_container.get(service_name, None)
|
||||
container = self.status_container[service_name].get("info", None)
|
||||
|
||||
if container:
|
||||
container.status = status
|
||||
container.status = status.name
|
||||
container.info = info
|
||||
container.metrics = metrics
|
||||
return
|
||||
service_info_item = BECServiceInfoContainer(
|
||||
service_name=service_name, status=status, info=info, metrics=metrics
|
||||
service_name=service_name, status=status.name, info=info, metrics=metrics
|
||||
)
|
||||
self.bec_service_info_container.update({service_name: service_info_item})
|
||||
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:
|
||||
@@ -217,7 +170,7 @@ class BECStatusBox(BECConnector, QTreeWidget):
|
||||
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.
|
||||
"""
|
||||
checked = []
|
||||
checked = [self.box_name]
|
||||
services_info = self.update_core_services(services_info, services_metric)
|
||||
checked.extend(self.CORE_SERVICES)
|
||||
|
||||
@@ -225,28 +178,13 @@ class BECStatusBox(BECConnector, QTreeWidget):
|
||||
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.tree_items:
|
||||
self._update_bec_service_container(
|
||||
service_name=service_name, status=msg.status, info=msg.info, metrics=metrics
|
||||
)
|
||||
self.service_update.emit(self.bec_service_info_container[service_name].model_dump())
|
||||
continue
|
||||
|
||||
item_widget = self._create_status_widget(
|
||||
service_name=service_name, status=msg.status, info=msg.info, metrics=metrics
|
||||
)
|
||||
item = QTreeWidgetItem()
|
||||
item.setDisabled(True)
|
||||
self.service_update.connect(item_widget.update_config)
|
||||
self.tree_top_item.addChild(item)
|
||||
self.setItemWidget(item, 0, item_widget)
|
||||
self.tree_items.update({service_name: (item, item_widget)})
|
||||
|
||||
self.check_redundant_tree_items(checked)
|
||||
if service_name not in self.status_container:
|
||||
self.add_tree_item(service_name, msg.status, msg.info, metrics)
|
||||
self._update_status_container(service_name, msg.status, msg.info, metrics)
|
||||
self.service_update.emit(self.status_container[service_name]["info"])
|
||||
|
||||
def update_core_services(self, services_info: dict, services_metric: dict) -> dict:
|
||||
"""Method to process status and metrics updates of core services (stored in CORE_SERVICES).
|
||||
If a core services is not connected, it should not be removed from the status widget
|
||||
"""Update the core services of BEC, and emit the updated status to the BECStatusBox.
|
||||
|
||||
Args:
|
||||
services_info (dict): A dictionary containing the service status of different services.
|
||||
@@ -255,42 +193,26 @@ class BECStatusBox(BECConnector, QTreeWidget):
|
||||
Returns:
|
||||
dict: The services_info dictionary after removing the info updates related to the CORE_SERVICES
|
||||
"""
|
||||
bec_core_state = "RUNNING"
|
||||
core_state = BECStatus.RUNNING
|
||||
for service_name in sorted(self.CORE_SERVICES):
|
||||
metric_msg = services_metric.get(service_name, None)
|
||||
metrics = metric_msg.metrics if metric_msg else None
|
||||
if service_name not in services_info:
|
||||
self.bec_service_info_container[service_name].status = "NOTCONNECTED"
|
||||
bec_core_state = "ERROR"
|
||||
else:
|
||||
msg = services_info.pop(service_name)
|
||||
self._update_bec_service_container(
|
||||
service_name=service_name, status=msg.status, info=msg.info, metrics=metrics
|
||||
)
|
||||
bec_core_state = (
|
||||
"RUNNING" if (msg.status.value > 1 and bec_core_state == "RUNNING") else "ERROR"
|
||||
)
|
||||
msg = services_info.pop(service_name, None)
|
||||
if msg is None:
|
||||
msg = StatusMessage(name=service_name, status=BECStatus.ERROR, info={})
|
||||
if service_name not in self.status_container:
|
||||
self.add_tree_item(service_name, msg.status, msg.info, metrics)
|
||||
|
||||
if service_name in self.tree_items:
|
||||
self.service_update.emit(self.bec_service_info_container[service_name].model_dump())
|
||||
continue
|
||||
self.add_tree_item(service_name, msg.status, msg.info, metrics)
|
||||
self._update_status_container(service_name, msg.status, msg.info, metrics)
|
||||
core_state = msg.status if msg.status.value < core_state.value else core_state
|
||||
|
||||
self.bec_core_state.emit(bec_core_state)
|
||||
self.service_update.emit(self.status_container[service_name]["info"])
|
||||
|
||||
# self.add_tree_item(service_name, msg.status, msg.info, metrics)
|
||||
|
||||
self.bec_core_state.emit(core_state.name if core_state else "NOTCONNECTED")
|
||||
return services_info
|
||||
|
||||
def check_redundant_tree_items(self, checked: list) -> None:
|
||||
"""Utility method to check and remove redundant objects from the BECStatusBox.
|
||||
|
||||
Args:
|
||||
checked (list): A list of services that are currently running.
|
||||
"""
|
||||
to_be_deleted = [key for key in self.tree_items if key not in checked]
|
||||
|
||||
for key in to_be_deleted:
|
||||
item, _ = self.tree_items.pop(key)
|
||||
self.tree_top_item.removeChild(item)
|
||||
|
||||
def add_tree_item(
|
||||
self, service_name: str, status: BECStatus, info: dict = None, metrics: dict = None
|
||||
) -> None:
|
||||
@@ -298,30 +220,18 @@ class BECStatusBox(BECConnector, QTreeWidget):
|
||||
|
||||
Args:
|
||||
service_name (str): The name of the service.
|
||||
service_status_msg (StatusMessage): The status 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=service_name, status=status, info=info, metrics=metrics
|
||||
)
|
||||
item = QTreeWidgetItem()
|
||||
item_widget = self._create_status_widget(service_name, status, info, metrics)
|
||||
toplevel_item = self.status_container[self.box_name]["item"]
|
||||
item = QTreeWidgetItem(toplevel_item) # setDisabled=True
|
||||
toplevel_item.addChild(item)
|
||||
self.tree.setItemWidget(item, 0, item_widget)
|
||||
self.service_update.connect(item_widget.update_config)
|
||||
self.tree_top_item.addChild(item)
|
||||
self.setItemWidget(item, 0, item_widget)
|
||||
self.tree_items.update({service_name: (item, item_widget)})
|
||||
|
||||
def init_ui_tree_widget(self) -> None:
|
||||
"""Initialise the tree widget for the status box."""
|
||||
self.setHeaderHidden(True)
|
||||
self.setStyleSheet(
|
||||
"QTreeWidget::item:!selected "
|
||||
"{ "
|
||||
"border: 1px solid gainsboro; "
|
||||
"border-left: none; "
|
||||
"border-top: none; "
|
||||
"}"
|
||||
"QTreeWidget::item:selected {}"
|
||||
)
|
||||
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:
|
||||
@@ -331,24 +241,47 @@ class BECStatusBox(BECConnector, QTreeWidget):
|
||||
item (QTreeWidgetItem): The item that was double clicked.
|
||||
column (int): The column that was double clicked.
|
||||
"""
|
||||
for _, (tree_item, status_widget) in self.tree_items.items():
|
||||
if tree_item == item:
|
||||
status_widget.show_popup()
|
||||
for _, objects in self.status_container.items():
|
||||
if objects["item"] == item:
|
||||
objects["widget"].show_popup()
|
||||
|
||||
def closeEvent(self, event):
|
||||
super().cleanup()
|
||||
QTreeWidget().closeEvent(event)
|
||||
self.connector.cleanup()
|
||||
|
||||
|
||||
def main():
|
||||
"""Main method to run the BECStatusBox widget."""
|
||||
|
||||
# pylint: disable=import-outside-toplevel
|
||||
from bec_lib.client import BECClient
|
||||
from bec_lib.logger import bec_logger
|
||||
from bec_lib.service_config import ServiceConfig
|
||||
from qtpy.QtWidgets import QApplication
|
||||
|
||||
from bec_widgets.utils.bec_dispatcher import QtRedisConnector
|
||||
|
||||
# logging has to be configured before create QApplication,
|
||||
# otherwise it ends badly with segfault...
|
||||
# (seems to be a threading issue with loguru and probably Redis connector,
|
||||
# which has to be a QtRedisConnector for Qt apps... Otherwise it is not
|
||||
# thread-safe somehow ; didn't want to debug all this now)
|
||||
logger = bec_logger.logger
|
||||
|
||||
service_config = ServiceConfig()
|
||||
bec_logger.configure(
|
||||
service_config.redis,
|
||||
QtRedisConnector,
|
||||
service_name="test_status_box",
|
||||
service_config=service_config.service_config,
|
||||
)
|
||||
|
||||
app = QApplication(sys.argv)
|
||||
|
||||
qdarktheme.setup_theme("auto")
|
||||
main_window = BECStatusBox()
|
||||
main_window.show()
|
||||
|
||||
client = BECClient()
|
||||
status = BECStatusBox(parent=None, client=client, gui_id="test")
|
||||
status.show()
|
||||
sys.exit(app.exec())
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
{'files': ['bec_status_box.py']}
|
||||
54
bec_widgets/widgets/bec_status_box/bec_status_box_plugin.py
Normal file
54
bec_widgets/widgets/bec_status_box/bec_status_box_plugin.py
Normal file
@@ -0,0 +1,54 @@
|
||||
# Copyright (C) 2022 The Qt Company Ltd.
|
||||
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
|
||||
|
||||
from qtpy.QtDesigner import QDesignerCustomWidgetInterface
|
||||
from qtpy.QtGui import QIcon
|
||||
|
||||
from bec_widgets.widgets.bec_status_box.bec_status_box import BECStatusBox
|
||||
|
||||
DOM_XML = """
|
||||
<ui language='c++'>
|
||||
<widget class='BECStatusBox' name='bec_status_box'>
|
||||
</widget>
|
||||
</ui>
|
||||
"""
|
||||
|
||||
|
||||
class BECStatusBoxPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self._form_editor = None
|
||||
|
||||
def createWidget(self, parent):
|
||||
t = BECStatusBox(parent)
|
||||
return t
|
||||
|
||||
def domXml(self):
|
||||
return DOM_XML
|
||||
|
||||
def group(self):
|
||||
return ""
|
||||
|
||||
def icon(self):
|
||||
return QIcon()
|
||||
|
||||
def includeFile(self):
|
||||
return "bec_status_box"
|
||||
|
||||
def initialize(self, form_editor):
|
||||
self._form_editor = form_editor
|
||||
|
||||
def isContainer(self):
|
||||
return False
|
||||
|
||||
def isInitialized(self):
|
||||
return self._form_editor is not None
|
||||
|
||||
def name(self):
|
||||
return "BECStatusBox"
|
||||
|
||||
def toolTip(self):
|
||||
return "Widget to display the BECStatus from all active services."
|
||||
|
||||
def whatsThis(self):
|
||||
return self.toolTip()
|
||||
@@ -0,0 +1,15 @@
|
||||
def main(): # pragma: no cover
|
||||
from qtpy import PYSIDE6
|
||||
|
||||
if not PYSIDE6:
|
||||
print("PYSIDE6 is not available in the environment. Cannot patch designer.")
|
||||
return
|
||||
from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection
|
||||
|
||||
from bec_widgets.widgets.bec_status_box.bec_status_box_plugin import BECStatusBoxPlugin
|
||||
|
||||
QPyDesignerCustomWidgetCollection.addCustomWidget(BECStatusBoxPlugin())
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
main()
|
||||
@@ -2,17 +2,12 @@
|
||||
The widget is bound to be used with the BECStatusBox widget."""
|
||||
|
||||
import enum
|
||||
import sys
|
||||
from datetime import datetime
|
||||
|
||||
import qdarktheme
|
||||
from bec_lib.utils.import_utils import lazy_import_from
|
||||
from pydantic import Field
|
||||
from qtpy.QtCore import Qt, Slot
|
||||
from qtpy.QtWidgets import QDialog, QHBoxLayout, QLabel, QStyle, QVBoxLayout, QWidget
|
||||
|
||||
from bec_widgets.utils.bec_connector import ConnectionConfig
|
||||
|
||||
# TODO : Put normal imports back when Pydantic gets faster
|
||||
BECStatus = lazy_import_from("bec_lib.messages", ("BECStatus",))
|
||||
|
||||
@@ -27,39 +22,29 @@ class IconsEnum(enum.Enum):
|
||||
NOTCONNECTED = "SP_TitleBarContextHelpButton"
|
||||
|
||||
|
||||
class StatusWidgetConfig(ConnectionConfig):
|
||||
"""Configuration class for the status item widget."""
|
||||
|
||||
service_name: str
|
||||
status: str
|
||||
info: dict
|
||||
metrics: dict | None
|
||||
icon_size: tuple = Field(default=(24, 24), description="The size of the icon in the widget.")
|
||||
font_size: int = Field(16, description="The font size of the text in the widget.")
|
||||
|
||||
|
||||
class StatusItem(QWidget):
|
||||
"""A widget to display the status of a service.
|
||||
|
||||
Args:
|
||||
parent: The parent widget.
|
||||
config (dict): The configuration for the service.
|
||||
config (dict): The configuration for the service, must be a BECServiceInfoContainer.
|
||||
"""
|
||||
|
||||
def __init__(self, parent=None, config: dict = None):
|
||||
if config is None:
|
||||
config = StatusWidgetConfig(widget_class=self.__class__.__name__)
|
||||
else:
|
||||
if isinstance(config, dict):
|
||||
config = StatusWidgetConfig(**config)
|
||||
self.config = config
|
||||
def __init__(self, parent: QWidget = None, config=None):
|
||||
QWidget.__init__(self, parent=parent)
|
||||
if config is None:
|
||||
# needed because we need parent to be the first argument for QT Designer
|
||||
raise ValueError(
|
||||
"Please initialize the StatusItem with a BECServiceInfoContainer for config, received None."
|
||||
)
|
||||
self.config = config
|
||||
self.parent = parent
|
||||
self.layout = None
|
||||
self.config = config
|
||||
self._popup_label_ref = {}
|
||||
self._label = None
|
||||
self._icon = None
|
||||
self.icon_size = (24, 24)
|
||||
|
||||
self._popup_label_ref = {}
|
||||
self.init_ui()
|
||||
|
||||
def init_ui(self) -> None:
|
||||
@@ -74,23 +59,21 @@ class StatusItem(QWidget):
|
||||
self.update_ui()
|
||||
|
||||
@Slot(dict)
|
||||
def update_config(self, config: dict) -> None:
|
||||
"""Update the configuration of the status item widget.
|
||||
This method is invoked from the parent widget.
|
||||
The UI values are later updated based on the new configuration.
|
||||
def update_config(self, config) -> None:
|
||||
"""Update the config of the status item widget.
|
||||
|
||||
Args:
|
||||
config (dict): Config updates from parent widget.
|
||||
config (dict): Config updates from parent widget, must be a BECServiceInfoContainer.
|
||||
"""
|
||||
if config["service_name"] != self.config.service_name:
|
||||
if self.config is None or config.service_name != self.config.service_name:
|
||||
return
|
||||
self.config.status = config["status"]
|
||||
self.config.info = config["info"]
|
||||
self.config.metrics = config["metrics"]
|
||||
self.config = config
|
||||
self.update_ui()
|
||||
|
||||
def update_ui(self) -> None:
|
||||
"""Update the UI of the labels, and popup dialog."""
|
||||
if self.config is None:
|
||||
return
|
||||
self.set_text()
|
||||
self.set_status()
|
||||
self._set_popup_text()
|
||||
@@ -99,8 +82,8 @@ class StatusItem(QWidget):
|
||||
"""Set the text of the QLabel basae on the config."""
|
||||
service = self.config.service_name
|
||||
status = self.config.status
|
||||
if "BECClient" in service.split("/"):
|
||||
service = service.split("/")[0] + "/..." + service.split("/")[1][-4:]
|
||||
if len(service.split("/")) > 1 and service.split("/")[0].startswith("BEC"):
|
||||
service = service.split("/", maxsplit=1)[0] + "/..." + service.split("/")[1][-4:]
|
||||
if status == "NOTCONNECTED":
|
||||
status = "NOT CONNECTED"
|
||||
text = f"{service} is {status}"
|
||||
@@ -110,7 +93,7 @@ class StatusItem(QWidget):
|
||||
"""Set the status icon for the status item widget."""
|
||||
icon_name = IconsEnum[self.config.status].value
|
||||
icon = self.style().standardIcon(getattr(QStyle.StandardPixmap, icon_name))
|
||||
self._icon.setPixmap(icon.pixmap(*self.config.icon_size))
|
||||
self._icon.setPixmap(icon.pixmap(*self.icon_size))
|
||||
self._icon.setAlignment(Qt.AlignmentFlag.AlignRight)
|
||||
|
||||
def show_popup(self) -> None:
|
||||
@@ -153,19 +136,3 @@ class StatusItem(QWidget):
|
||||
def _cleanup_popup_label(self) -> None:
|
||||
"""Cleanup the popup label."""
|
||||
self._popup_label_ref.clear()
|
||||
|
||||
|
||||
def main():
|
||||
"""Run the status item widget."""
|
||||
# pylint: disable=import-outside-toplevel
|
||||
from qtpy.QtWidgets import QApplication
|
||||
|
||||
app = QApplication(sys.argv)
|
||||
qdarktheme.setup_theme("auto")
|
||||
main_window = StatusItem()
|
||||
main_window.show()
|
||||
sys.exit(app.exec())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
# pylint: skip-file
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
@@ -15,9 +16,7 @@ def service_status_fixture():
|
||||
|
||||
@pytest.fixture
|
||||
def status_box(qtbot, mocked_client, service_status_fixture):
|
||||
widget = BECStatusBox(
|
||||
client=mocked_client, service_name="test", bec_service_status_mixin=service_status_fixture
|
||||
)
|
||||
widget = BECStatusBox(client=mocked_client, bec_service_status_mixin=service_status_fixture)
|
||||
qtbot.addWidget(widget)
|
||||
qtbot.waitExposed(widget)
|
||||
yield widget
|
||||
@@ -25,8 +24,9 @@ def status_box(qtbot, mocked_client, service_status_fixture):
|
||||
|
||||
def test_update_top_item(status_box):
|
||||
assert status_box.children()[0].children()[0].config.status == "IDLE"
|
||||
name = status_box.box_name
|
||||
status_box.update_top_item_status(status="RUNNING")
|
||||
assert status_box.bec_service_info_container["test"].status == "RUNNING"
|
||||
assert status_box.status_container[name]["info"].status == "RUNNING"
|
||||
assert status_box.children()[0].children()[0].config.status == "RUNNING"
|
||||
|
||||
|
||||
@@ -48,13 +48,13 @@ def test_bec_service_container(status_box):
|
||||
info = {"test": "test"}
|
||||
metrics = {"metric": "test_metric"}
|
||||
expected_return = BECServiceInfoContainer(
|
||||
service_name=name, status=status, info=info, metrics=metrics
|
||||
service_name=name, status=status.name, info=info, metrics=metrics
|
||||
)
|
||||
assert status_box.service_name in status_box.bec_service_info_container
|
||||
assert len(status_box.bec_service_info_container) == 1
|
||||
status_box._update_bec_service_container(name, status, info, metrics)
|
||||
assert len(status_box.bec_service_info_container) == 2
|
||||
assert status_box.bec_service_info_container[name] == expected_return
|
||||
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
|
||||
|
||||
|
||||
def test_add_tree_item(status_box):
|
||||
@@ -65,7 +65,7 @@ def test_add_tree_item(status_box):
|
||||
assert len(status_box.children()[0].children()) == 1
|
||||
status_box.add_tree_item(name, status, info, metrics)
|
||||
assert len(status_box.children()[0].children()) == 2
|
||||
assert name in status_box.tree_items
|
||||
assert name in status_box.status_container
|
||||
|
||||
|
||||
def test_update_service_status(status_box):
|
||||
@@ -82,10 +82,10 @@ def test_update_service_status(status_box):
|
||||
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.tree_items
|
||||
assert not_connected_name in status_box.status_container
|
||||
status_box.update_service_status(services_status, services_metrics)
|
||||
assert status_box.tree_items[name][1].config.metrics == metrics
|
||||
assert not_connected_name not in status_box.tree_items
|
||||
assert status_box.status_container[name]["widget"].config.metrics == metrics
|
||||
assert not_connected_name not in status_box.status_container
|
||||
|
||||
|
||||
def test_update_core_services(status_box):
|
||||
@@ -99,14 +99,14 @@ def test_update_core_services(status_box):
|
||||
|
||||
status_box.update_core_services(services_status, services_metrics)
|
||||
assert status_box.children()[0].children()[0].config.status == "RUNNING"
|
||||
assert status_box.tree_items[name][1].config.metrics == metrics
|
||||
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.children()[0].children()[0].config.status == "ERROR"
|
||||
assert status_box.tree_items[name][1].config.metrics == metrics
|
||||
assert status_box.children()[0].children()[0].config.status == status.name
|
||||
assert status_box.status_container[name]["widget"].config.metrics == metrics
|
||||
|
||||
|
||||
def test_double_click_item(status_box):
|
||||
@@ -115,7 +115,9 @@ def test_double_click_item(status_box):
|
||||
info = {"test": "test"}
|
||||
metrics = {"MyData": "This should be shown nicely"}
|
||||
status_box.add_tree_item(name, status, info, metrics)
|
||||
item, status_item = status_box.tree_items[name]
|
||||
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.itemDoubleClicked.emit(item, 0)
|
||||
assert mock_show_popup.call_count == 1
|
||||
|
||||
Reference in New Issue
Block a user