diff --git a/bec_widgets/widgets/progress/device_initialization_progress_bar/__init__.py b/bec_widgets/widgets/progress/device_initialization_progress_bar/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/bec_widgets/widgets/progress/device_initialization_progress_bar/device_initialization_progress_bar.py b/bec_widgets/widgets/progress/device_initialization_progress_bar/device_initialization_progress_bar.py new file mode 100644 index 00000000..093aad71 --- /dev/null +++ b/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()) diff --git a/tests/unit_tests/test_device_initialization_progress_bar.py b/tests/unit_tests/test_device_initialization_progress_bar.py new file mode 100644 index 00000000..9904d919 --- /dev/null +++ b/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}"