feat(device-initialization-progress-bar): add progress bar for device initialization

bec_widgets/widgets/progress/device_initialization_progress_bar/device_initialization_progress_bar.py

@@ -0,0 +1,126 @@
+from bec_lib.endpoints import MessageEndpoints
+from bec_lib.messages import DeviceInitializationProgressMessage
+from qtpy.QtCore import Signal
+
+from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
+from bec_widgets.widgets.progress.bec_progressbar.bec_progressbar import BECProgressBar
+
+
+class DeviceInitializationProgressBar(BECProgressBar):
+    """A progress bar that displays the progress of device initialization."""
+
+    # Signal emitted for failed device initializations
+    failed_devices_changed = Signal(list)
+
+    def __init__(self, parent=None, client=None):
+        super().__init__(parent=parent, client=client)
+        self._latest_device_config_msg: dict | None = None
+        self._failed_devices: list[str] = []
+        self.bec_dispatcher.connect_slot(
+            slot=self._update_device_initialization_progress,
+            topics=MessageEndpoints.device_initialization_progress(),
+        )
+        self._reset_progress_bar()
+
+    @SafeProperty(list)
+    def failed_devices(self) -> list[str]:
+        """Get the list of devices that failed to initialize.
+
+        Returns:
+            list[str]: A list of device identifiers that failed during initialization.
+        """
+        return self._failed_devices
+
+    @failed_devices.setter
+    def failed_devices(self, value: list[str]) -> None:
+        self._failed_devices = value
+        self.failed_devices_changed.emit(self.failed_devices)
+
+    @SafeSlot()
+    def reset_failed_devices(self) -> None:
+        """Reset the list of failed devices."""
+        self._failed_devices.clear()
+        self.failed_devices_changed.emit(self.failed_devices)
+
+    @SafeSlot(str)
+    def add_failed_device(self, device: str) -> None:
+        """Add a device to the list of failed devices.
+
+        Args:
+            device (str): The identifier of the device that failed to initialize.
+        """
+        self._failed_devices.append(device)
+        self.failed_devices_changed.emit(self.failed_devices)
+
+    @SafeSlot(dict, dict)
+    def _update_device_initialization_progress(self, msg: dict, metadata: dict) -> None:
+        """Update the progress bar based on device initialization progress messages.
+
+        Args:
+            msg (dict): The device initialization progress message.
+            metadata (dict): Additional metadata about the message.
+        """
+        msg: DeviceInitializationProgressMessage = (
+            DeviceInitializationProgressMessage.model_validate(msg)
+        )
+        if msg.finished is False:
+            self.label_template = "\n".join(
+                [
+                    f"Device initialization for '{msg.device}' is in progress...",
+                    "$value / $maximum - $percentage %",
+                ]
+            )
+        elif msg.finished is True and msg.success is False:
+            self.add_failed_device(msg.device)
+            self.label_template = "\n".join(
+                [
+                    f"Device initialization for '{msg.device}' failed!",
+                    "$value / $maximum - $percentage %",
+                ]
+            )
+        else:
+            self.label_template = "\n".join(
+                [
+                    f"Device initialization for '{msg.device}' succeeded!",
+                    "$value / $maximum - $percentage %",
+                ]
+            )
+        self.set_maximum(msg.total)
+        self.set_value(msg.index)
+        self._update_tool_tip()
+
+    def _reset_progress_bar(self) -> None:
+        """Reset the progress bar to its initial state."""
+        self.label_template = "\n".join(
+            ["Waiting for device initialization...", "$value / $maximum - $percentage %"]
+        )
+        self.set_value(0)
+        self.set_maximum(1)
+        self.reset_failed_devices()
+        self._update_tool_tip()
+
+    def _update_tool_tip(self) -> None:
+        """Update the tooltip to show failed devices if any."""
+        if self._failed_devices:
+            failed_devices_str = ", ".join(sorted(self._failed_devices))
+            self.setToolTip(f"Failed devices: {failed_devices_str}")
+        else:
+            self.setToolTip("No device initialization failures.")
+
+
+if __name__ == "__main__":  # pragma: no cover
+    import sys
+
+    from qtpy.QtWidgets import QApplication
+
+    app = QApplication(sys.argv)
+
+    progressBar = DeviceInitializationProgressBar()
+
+    def my_cb(devices: list):
+        print("Failed devices:", devices)
+
+    progressBar.failed_devices_changed.connect(my_cb)
+    progressBar.show()
+
+    sys.exit(app.exec())

tests/unit_tests/test_device_initialization_progress_bar.py

@@ -0,0 +1,66 @@
+# pylint skip
+import pytest
+from bec_lib.messages import DeviceInitializationProgressMessage
+
+from bec_widgets.widgets.progress.device_initialization_progress_bar.device_initialization_progress_bar import (
+    DeviceInitializationProgressBar,
+)
+
+
+@pytest.fixture
+def progress_bar(qtbot):
+    widget = DeviceInitializationProgressBar()
+    qtbot.addWidget(widget)
+    qtbot.waitExposed(widget)
+    yield widget
+
+
+def test_progress_bar_initialization(progress_bar):
+    """Test the initial state of the DeviceInitializationProgressBar."""
+    assert progress_bar.failed_devices == []
+    assert progress_bar._user_value == 0
+    assert progress_bar._user_maximum == 1
+    assert progress_bar.toolTip() == "No device initialization failures."
+
+
+def test_update_device_initialization_progress(progress_bar, qtbot):
+    """Test updating the progress bar with different device initialization messages."""
+
+    # I. Update with message of running DeviceInitializationProgressMessage, finished=False, success=False
+    msg = DeviceInitializationProgressMessage(
+        device="DeviceA", index=1, total=3, finished=False, success=False
+    )
+
+    progress_bar._update_device_initialization_progress(msg.model_dump(), {})
+    assert progress_bar._user_value == 1
+    assert progress_bar._user_maximum == 3
+    assert (
+        f"Device initialization for '{msg.device}' is in progress..."
+        in progress_bar.center_label.text()
+    )
+
+    # II. Update with message of finished DeviceInitializationProgressMessage, finished=True, success=True
+    msg.finished = True
+    msg.success = True
+    progress_bar._update_device_initialization_progress(msg.model_dump(), {})
+    assert progress_bar._user_value == 1
+    assert progress_bar._user_maximum == 3
+    assert (
+        f"Device initialization for '{msg.device}' succeeded!" in progress_bar.center_label.text()
+    )
+
+    # III. Update with message of finished DeviceInitializationProgressMessage, finished=True, success=False
+    msg.finished = True
+    msg.success = False
+    msg.device = "DeviceB"
+    msg.index = 2
+    with qtbot.waitSignal(progress_bar.failed_devices_changed) as signal_blocker:
+        progress_bar._update_device_initialization_progress(msg.model_dump(), {})
+        assert progress_bar._user_value == 2
+        assert progress_bar._user_maximum == 3
+        assert (
+            f"Device initialization for '{msg.device}' failed!" in progress_bar.center_label.text()
+        )
+        assert signal_blocker.args == [[msg.device]]
+
+        assert progress_bar.toolTip() == f"Failed devices: {msg.device}"