bec/bec_widgets
0
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2025-07-14 03:31:50 +02:00
Code Activity

refactor: simplify logic in bec_status_box

Browse Source
This commit is contained in:
appel_c
2024-06-27 13:26:02 +02:00
committed by Mathias Guijarro
parent 0e802d8194
commit 5d435bd5ee
3 changed files with 239 additions and 304 deletions
Download Patch File Download Diff File Expand all files Collapse all files

430
bec_widgets/widgets/bec_status_box/bec_status_box.py
View File

@ -5,15 +5,16 @@ The widget automatically updates the status of all running BEC services, and dis
from __future__ import annotations from __future__ import annotations
import sys import sys
from collections import defaultdict
from dataclasses import dataclass
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
import qdarktheme import qdarktheme
from bec_lib.utils.import_utils import lazy_import_from 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.QtCore import QObject, QTimer, Signal, Slot
from qtpy.QtWidgets import QTreeWidget, QTreeWidgetItem from qtpy.QtWidgets import QTreeWidget, QTreeWidgetItem
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 from bec_widgets.widgets.bec_status_box.status_item import StatusItem
if TYPE_CHECKING: if TYPE_CHECKING:
@ -23,45 +24,18 @@ if TYPE_CHECKING:
BECStatus = lazy_import_from("bec_lib.messages", ("BECStatus",)) BECStatus = lazy_import_from("bec_lib.messages", ("BECStatus",))
class BECStatusBoxConfig(ConnectionConfig): @dataclass
pass class BECServiceInfoContainer:
class BECServiceInfoContainer(BaseModel):
"""Container to store information about the BEC services.""" """Container to store information about the BEC services."""
service_name: str service_name: str
status: BECStatus | str = Field( status: str
default="NOTCONNECTED",
description="The status of the service. Can be any of the BECStatus names, or NOTCONNECTED.",
)
info: dict info: dict
metrics: dict | None 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): class BECServiceStatusMixin(QObject):
"""A mixin class to update the service status, and metrics. """Mixin to receive the latest service status from the BEC server and emit it via services_update signal.
It emits a signal 'services_update' when the service status is updated.
Args: Args:
client (BECClient): The client object to connect to the BEC server. client (BECClient): The client object to connect to the BEC server.
@ -77,21 +51,18 @@ class BECServiceStatusMixin(QObject):
self._service_update_timer.start(1000) self._service_update_timer.start(1000)
def _get_service_status(self): def _get_service_status(self):
"""Pull latest service and metrics updates from REDIS for all services, and emit both via 'services_update' signal.""" """Get the latest service status from the BEC server."""
# pylint: disable=protected-access # pylint: disable=protected-access
self.client._update_existing_services() self.client._update_existing_services()
self.services_update.emit(self.client._services_info, self.client._services_metric) self.services_update.emit(self.client._services_info, self.client._services_metric)
class BECStatusBox(BECConnector, QTreeWidget): class BECStatusBox(BECConnector, QTreeWidget):
"""A widget to display the status of different BEC services. """An autonomous widget to display the status of 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.
Args: Args:
parent Optional : The parent widget for the BECStatusBox. Defaults to None. 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 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. 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. gui_id Optional(str): The unique id for the widget. Defaults to None.
@ -99,216 +70,61 @@ class BECStatusBox(BECConnector, QTreeWidget):
CORE_SERVICES = ["DeviceServer", "ScanServer", "SciHub", "ScanBundler", "FileWriterManager"] CORE_SERVICES = ["DeviceServer", "ScanServer", "SciHub", "ScanBundler", "FileWriterManager"]
service_update = Signal(dict) service_update = Signal(BECServiceInfoContainer)
bec_core_state = Signal(str) bec_core_state = Signal(str)
_initialized = False
_bec_status_box = None
def __init__( def __init__(
self, self,
parent=None, parent=None,
service_name: str = "BEC Server", box_name: str = "BEC Server",
client: BECClient = None, client: BECClient = None,
config: BECStatusBoxConfig | dict = None,
bec_service_status_mixin: BECServiceStatusMixin = None, bec_service_status_mixin: BECServiceStatusMixin = None,
gui_id: str = None, gui_id: str = None,
): ):
if config is None: if self._initialized == True:
config = BECStatusBoxConfig(widget_class=self.__class__.__name__) return
else: super().__init__(client=client, gui_id=gui_id)
if isinstance(config, dict):
config = BECStatusBoxConfig(**config)
super().__init__(client=client, config=config, gui_id=gui_id)
QTreeWidget.__init__(self, parent=parent) QTreeWidget.__init__(self, parent=parent)
self.service_name = service_name self.box_name = box_name
self.config = config self.status_container = defaultdict(lambda: {"info": None, "item": None, "widget": None})
self._initialized = False
self.bec_service_info_container = {}
self.tree_items = {}
self.tree_top_item = None
if not bec_service_status_mixin: if not bec_service_status_mixin:
bec_service_status_mixin = BECServiceStatusMixin(client=self.client) bec_service_status_mixin = BECServiceStatusMixin(client=self.client)
self.bec_service_status = bec_service_status_mixin self.bec_service_status = bec_service_status_mixin
self.init_ui() if not self._initialized:
self.bec_service_status.services_update.connect(self.update_service_status) self.init_ui()
self.bec_core_state.connect(self.update_top_item_status) self.bec_service_status.services_update.connect(self.update_service_status)
self.itemDoubleClicked.connect(self.on_tree_item_double_clicked) self.bec_core_state.connect(self.update_top_item_status)
self.itemDoubleClicked.connect(self.on_tree_item_double_clicked)
def __new__(cls, *args, forced: bool = False, **kwargs):
if forced:
cls._initialized = False
cls._bec_status_box = super(BECStatusBox, cls).__new__(cls)
return cls._bec_status_box
if cls._bec_status_box is not None and cls._initialized is True:
return cls._bec_status_box
cls._bec_status_box = super(BECStatusBox, cls).__new__(cls)
return cls._bec_status_box
def init_ui(self) -> None: def init_ui(self) -> None:
"""Initialize the UI for the status box, and add QTreeWidget as the basis for the status box.""" """Init the UI for the BECStatusBox widget, should only take place once."""
self.init_ui_tree_widget() self.init_ui_tree_widget()
top_label = self._create_status_widget(self.service_name, status=BECStatus.IDLE) top_label = self._create_status_widget(self.box_name, status=BECStatus.IDLE)
self.tree_top_item = QTreeWidgetItem() tree_item = QTreeWidgetItem()
self.tree_top_item.setExpanded(True) tree_item.setExpanded(True)
self.tree_top_item.setDisabled(True) tree_item.setDisabled(True)
self.addTopLevelItem(self.tree_top_item) self.status_container[self.box_name].update({"item": tree_item, "widget": top_label})
self.setItemWidget(self.tree_top_item, 0, top_label) self.addTopLevelItem(tree_item)
self.setItemWidget(tree_item, 0, top_label)
self.service_update.connect(top_label.update_config) 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.
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.
"""
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,
},
)
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.bec_service_info_container[self.service_name].status = status
self.service_update.emit(self.bec_service_info_container[self.service_name].model_dump())
def _update_bec_service_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.
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.
"""
container = self.bec_service_info_container.get(service_name, None)
if container:
container.status = status
container.info = info
container.metrics = metrics
return
service_info_item = BECServiceInfoContainer(
service_name=service_name, status=status, info=info, metrics=metrics
)
self.bec_service_info_container.update({service_name: service_info_item})
@Slot(dict, dict)
def update_service_status(self, services_info: dict, services_metric: dict) -> None:
"""Callback function services_metric from BECServiceStatusMixin.
It 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.
"""
checked = []
services_info = self.update_core_services(services_info, services_metric)
checked.extend(self.CORE_SERVICES)
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.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)
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
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.
Returns:
dict: The services_info dictionary after removing the info updates related to the CORE_SERVICES
"""
bec_core_state = "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"
)
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.bec_core_state.emit(bec_core_state)
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:
"""Method to add a new QTreeWidgetItem together with a StatusItem to the tree widget.
Args:
service_name (str): The name of the service.
service_status_msg (StatusMessage): The status of 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()
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: def init_ui_tree_widget(self) -> None:
"""Initialise the tree widget for the status box.""" """Initialise the tree widget for the status box."""
@ -323,6 +139,151 @@ class BECStatusBox(BECConnector, QTreeWidget):
"QTreeWidget::item:selected {}" "QTreeWidget::item:selected {}"
) )
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.
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.
"""
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.service_update.emit(self.status_container[self.box_name]["info"])
def _update_status_container(
self, service_name: str, status: BECStatus, 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.
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.
"""
container = self.status_container[service_name].get("info", None)
if container:
container.status = status.name
container.info = info
container.metrics = metrics
return
service_info_item = BECServiceInfoContainer(
service_name=service_name, status=status.name, info=info, metrics=metrics
)
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:
"""Callback function services_metric from BECServiceStatusMixin.
It 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.
"""
checked = [self.box_name]
services_info = self.update_core_services(services_info, services_metric)
checked.extend(self.CORE_SERVICES)
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:
self._update_status_container(service_name, msg.status, msg.info, metrics)
self.service_update.emit(self.status_container[service_name]["info"])
continue
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.
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.
Returns:
dict: The services_info dictionary after removing the info updates related to the CORE_SERVICES
"""
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
msg = services_info.pop(service_name, None)
if service_name not in self.status_container:
self.add_tree_item(service_name, msg.status, msg.info, metrics)
continue
if not msg:
self.status_container[service_name]["info"].status = "NOTCONNECTED"
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
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.status_container if key not in checked]
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 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() # setDisabled=True
self.service_update.connect(item_widget.update_config)
self.status_container[self.box_name]["item"].addChild(item)
self.setItemWidget(item, 0, item_widget)
self.status_container[service_name].update({"item": item, "widget": item_widget})
@Slot(QTreeWidgetItem, int) @Slot(QTreeWidgetItem, int)
def on_tree_item_double_clicked(self, item: QTreeWidgetItem, column: int) -> None: def on_tree_item_double_clicked(self, item: QTreeWidgetItem, column: int) -> None:
"""Callback function for double clicks on individual QTreeWidgetItems in the collapsed section. """Callback function for double clicks on individual QTreeWidgetItems in the collapsed section.
@ -331,11 +292,16 @@ class BECStatusBox(BECConnector, QTreeWidget):
item (QTreeWidgetItem): The item that was double clicked. item (QTreeWidgetItem): The item that was double clicked.
column (int): The column that was double clicked. column (int): The column that was double clicked.
""" """
for _, (tree_item, status_widget) in self.tree_items.items(): for _, objects in self.status_container.items():
if tree_item == item: if objects["item"] == item:
status_widget.show_popup() objects["widget"].show_popup()
def closeEvent(self, event): def closeEvent(self, event):
"""Upon closing the widget, clean up the BECStatusBox and the QTreeWidget.
Args:
event: The close event.
"""
super().cleanup() super().cleanup()
QTreeWidget().closeEvent(event) QTreeWidget().closeEvent(event)

75
bec_widgets/widgets/bec_status_box/status_item.py
View File

@ -2,17 +2,12 @@
The widget is bound to be used with the BECStatusBox widget.""" The widget is bound to be used with the BECStatusBox widget."""
import enum import enum
import sys
from datetime import datetime from datetime import datetime
import qdarktheme
from bec_lib.utils.import_utils import lazy_import_from from bec_lib.utils.import_utils import lazy_import_from
from pydantic import Field
from qtpy.QtCore import Qt, Slot from qtpy.QtCore import Qt, Slot
from qtpy.QtWidgets import QDialog, QHBoxLayout, QLabel, QStyle, QVBoxLayout, QWidget 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 # TODO : Put normal imports back when Pydantic gets faster
BECStatus = lazy_import_from("bec_lib.messages", ("BECStatus",)) BECStatus = lazy_import_from("bec_lib.messages", ("BECStatus",))
@ -27,39 +22,29 @@ class IconsEnum(enum.Enum):
NOTCONNECTED = "SP_TitleBarContextHelpButton" 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): class StatusItem(QWidget):
"""A widget to display the status of a service. """A widget to display the status of a service.
Args: Args:
parent: The parent widget. 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): def __init__(self, parent: QWidget = None, config=None):
if config is None:
config = StatusWidgetConfig(widget_class=self.__class__.__name__)
else:
if isinstance(config, dict):
config = StatusWidgetConfig(**config)
self.config = config
QWidget.__init__(self, parent=parent) 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.parent = parent
self.layout = None self.layout = None
self.config = config
self._popup_label_ref = {}
self._label = None self._label = None
self._icon = None self._icon = None
self.icon_size = (24, 24)
self._popup_label_ref = {}
self.init_ui() self.init_ui()
def init_ui(self) -> None: def init_ui(self) -> None:
@ -74,23 +59,21 @@ class StatusItem(QWidget):
self.update_ui() self.update_ui()
@Slot(dict) @Slot(dict)
def update_config(self, config: dict) -> None: def update_config(self, config) -> None:
"""Update the configuration of the status item widget. """Update the config of the status item widget.
This method is invoked from the parent widget.
The UI values are later updated based on the new configuration.
Args: 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 return
self.config.status = config["status"] self.config = config
self.config.info = config["info"]
self.config.metrics = config["metrics"]
self.update_ui() self.update_ui()
def update_ui(self) -> None: def update_ui(self) -> None:
"""Update the UI of the labels, and popup dialog.""" """Update the UI of the labels, and popup dialog."""
if self.config is None:
return
self.set_text() self.set_text()
self.set_status() self.set_status()
self._set_popup_text() self._set_popup_text()
@ -99,8 +82,8 @@ class StatusItem(QWidget):
"""Set the text of the QLabel basae on the config.""" """Set the text of the QLabel basae on the config."""
service = self.config.service_name service = self.config.service_name
status = self.config.status status = self.config.status
if "BECClient" in service.split("/"): if len(service.split("/")) > 1 and service.split("/")[0].startswith("BEC"):
service = service.split("/")[0] + "/..." + service.split("/")[1][-4:] service = service.split("/", maxsplit=1)[0] + "/..." + service.split("/")[1][-4:]
if status == "NOTCONNECTED": if status == "NOTCONNECTED":
status = "NOT CONNECTED" status = "NOT CONNECTED"
text = f"{service} is {status}" text = f"{service} is {status}"
@ -110,7 +93,7 @@ class StatusItem(QWidget):
"""Set the status icon for the status item widget.""" """Set the status icon for the status item widget."""
icon_name = IconsEnum[self.config.status].value icon_name = IconsEnum[self.config.status].value
icon = self.style().standardIcon(getattr(QStyle.StandardPixmap, icon_name)) 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) self._icon.setAlignment(Qt.AlignmentFlag.AlignRight)
def show_popup(self) -> None: def show_popup(self) -> None:
@ -153,19 +136,3 @@ class StatusItem(QWidget):
def _cleanup_popup_label(self) -> None: def _cleanup_popup_label(self) -> None:
"""Cleanup the popup label.""" """Cleanup the popup label."""
self._popup_label_ref.clear() 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()

38
tests/unit_tests/test_bec_status_box.py
View File

@ -1,3 +1,4 @@
# pylint: skip-file
from unittest import mock from unittest import mock
import pytest import pytest
@ -15,9 +16,7 @@ def service_status_fixture():
@pytest.fixture @pytest.fixture
def status_box(qtbot, mocked_client, service_status_fixture): def status_box(qtbot, mocked_client, service_status_fixture):
widget = BECStatusBox( widget = BECStatusBox(client=mocked_client, bec_service_status_mixin=service_status_fixture)
client=mocked_client, service_name="test", bec_service_status_mixin=service_status_fixture
)
qtbot.addWidget(widget) qtbot.addWidget(widget)
qtbot.waitExposed(widget) qtbot.waitExposed(widget)
yield widget yield widget
@ -25,8 +24,9 @@ def status_box(qtbot, mocked_client, service_status_fixture):
def test_update_top_item(status_box): def test_update_top_item(status_box):
assert status_box.children()[0].children()[0].config.status == "IDLE" assert status_box.children()[0].children()[0].config.status == "IDLE"
name = status_box.box_name
status_box.update_top_item_status(status="RUNNING") 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" assert status_box.children()[0].children()[0].config.status == "RUNNING"
@ -48,13 +48,13 @@ def test_bec_service_container(status_box):
info = {"test": "test"} info = {"test": "test"}
metrics = {"metric": "test_metric"} metrics = {"metric": "test_metric"}
expected_return = BECServiceInfoContainer( 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 status_box.box_name in status_box.status_container
assert len(status_box.bec_service_info_container) == 1 assert len(status_box.status_container) == 1
status_box._update_bec_service_container(name, status, info, metrics) status_box._update_status_container(name, status, info, metrics)
assert len(status_box.bec_service_info_container) == 2 assert len(status_box.status_container) == 2
assert status_box.bec_service_info_container[name] == expected_return assert status_box.status_container[name]["info"] == expected_return
def test_add_tree_item(status_box): 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 assert len(status_box.children()[0].children()) == 1
status_box.add_tree_item(name, status, info, metrics) status_box.add_tree_item(name, status, info, metrics)
assert len(status_box.children()[0].children()) == 2 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): 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)} services_metrics = {name: ServiceMetricMessage(name=name, metrics=metrics)}
with mock.patch.object(status_box, "update_core_services", return_value=services_status): 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) status_box.update_service_status(services_status, services_metrics)
assert status_box.tree_items[name][1].config.metrics == metrics assert status_box.status_container[name]["widget"].config.metrics == metrics
assert not_connected_name not in status_box.tree_items assert not_connected_name not in status_box.status_container
def test_update_core_services(status_box): 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) status_box.update_core_services(services_status, services_metrics)
assert status_box.children()[0].children()[0].config.status == "RUNNING" 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 status = BECStatus.IDLE
services_status = {name: StatusMessage(name=name, status=status, info=info)} services_status = {name: StatusMessage(name=name, status=status, info=info)}
services_metrics = {name: ServiceMetricMessage(name=name, metrics=metrics)} services_metrics = {name: ServiceMetricMessage(name=name, metrics=metrics)}
status_box.update_core_services(services_status, services_metrics) status_box.update_core_services(services_status, services_metrics)
assert status_box.children()[0].children()[0].config.status == "ERROR" assert status_box.children()[0].children()[0].config.status == status.name
assert status_box.tree_items[name][1].config.metrics == metrics assert status_box.status_container[name]["widget"].config.metrics == metrics
def test_double_click_item(status_box): def test_double_click_item(status_box):
@ -115,7 +115,9 @@ def test_double_click_item(status_box):
info = {"test": "test"} info = {"test": "test"}
metrics = {"MyData": "This should be shown nicely"} metrics = {"MyData": "This should be shown nicely"}
status_box.add_tree_item(name, status, info, metrics) 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: with mock.patch.object(status_item, "show_popup") as mock_show_popup:
status_box.itemDoubleClicked.emit(item, 0) status_box.itemDoubleClicked.emit(item, 0)
assert mock_show_popup.call_count == 1 assert mock_show_popup.call_count == 1