wip refactor 2

Author: wyzula-jan

bec_widgets/widgets/progress/progress_backend.py

@@ -6,7 +6,6 @@
 from typing import Literal
 
 import numpy as np
-from bec_lib import messages
 from bec_lib.endpoints import MessageEndpoints
 from qtpy.QtCore import QObject, QTimer, Signal
 
@@ -140,7 +139,6 @@ def __init__(self, bec_dispatcher, parent: QObject | None = None):
         self.bec_dispatcher = bec_dispatcher
         self._progress_source: ProgressSource | None = None
         self._progress_device: str | None = None
-        self._queue_connected = False
         self.task: ProgressTask | None = None
         self.scan_number: int | None = None
         self._active_scan_id: str | None = None
@@ -167,30 +165,9 @@ def start(
         *,
         source: ProgressSource | None = ProgressSource.SCAN_PROGRESS,
         device: str | None = None,
-        connect_queue: bool = False,
-        refresh_queue: bool = False,
     ) -> None:
         if source is not None:
             self.set_progress_source(source, device=device)
-        if connect_queue:
-            self.connect_to_queue()
-        if refresh_queue:
-            self.refresh_queue()
-
-    def connect_to_queue(self) -> None:
-        if self._queue_connected:
-            return
-        self.bec_dispatcher.connect_slot(self.on_queue_update, MessageEndpoints.scan_queue_status())
-        self._queue_connected = True
-
-    def refresh_queue(self) -> None:
-        connector = getattr(self.bec_dispatcher.client, "connector", None)
-        if connector is None:
-            return
-        msg = connector.get(MessageEndpoints.scan_queue_status())
-        if msg is None:
-            return
-        self.on_queue_update(msg.content, msg.metadata)
 
     def set_progress_source(self, source: ProgressSource, device: str | None = None) -> None:
         if source == ProgressSource.DEVICE_PROGRESS and not device:
@@ -264,42 +241,6 @@ def clear_task(self, *, emit_finished: bool = True) -> None:
         if emit_finished:
             self.progress_finished.emit(self.current_snapshot(value=0, max_value=100, done=True))
 
-    def on_queue_update(self, msg_content: dict, metadata: dict):
-        queue_info = self._extract_active_queue_info(msg_content)
-        if queue_info is None:
-            self.clear_task()
-            self.set_progress_source(ProgressSource.SCAN_PROGRESS)
-            return
-
-        active_request_block = queue_info.active_request_block
-        scan_id = active_request_block.scan_id or str(active_request_block.scan_number)
-        rid = getattr(active_request_block, "RID", None) or metadata.get("RID")
-        if self.task is None or self._active_scan_id != scan_id:
-            self._start_task(scan_id, rid=rid)
-        else:
-            self._active_rid = rid or self._active_rid
-
-        self.scan_number = active_request_block.scan_number
-        self.source_changed.emit(self.current_snapshot(value=0, max_value=100, done=False))
-
-    def _extract_active_queue_info(self, msg_content: dict):
-        if "queue" not in msg_content:
-            return None
-        if "primary" not in msg_content["queue"]:
-            return None
-        primary_queue = msg_content.get("queue").get("primary")
-        if primary_queue is None or not isinstance(primary_queue, messages.ScanQueueStatus):
-            return None
-        primary_queue_info = primary_queue.info
-        if len(primary_queue_info) == 0:
-            return None
-        scan_info = primary_queue_info[0]
-        if scan_info is None or scan_info.active_request_block is None:
-            return None
-        if scan_info.status.lower() != "running":
-            return None
-        return scan_info
-
     def on_progress_update(self, msg_content: dict, metadata: dict):
         if self._progress_source is None:
             return
@@ -323,6 +264,9 @@ def process_progress_message(
         )
         scan_id = metadata.get("scan_id") or metadata.get("RID")
         rid = metadata.get("RID")
+        scan_number = metadata.get("scan_number")
+        if scan_number is not None:
+            self.scan_number = scan_number
         is_new_scan = False
         previous_scan_id = self._active_scan_id
         previous_rid = self._active_rid
@@ -367,8 +311,3 @@ def process_progress_message(
     def cleanup(self) -> None:
         self.clear_task(emit_finished=False)
         self._disconnect_progress_source()
-        if self._queue_connected:
-            self.bec_dispatcher.disconnect_slot(
-                self.on_queue_update, MessageEndpoints.scan_queue_status()
-            )
-            self._queue_connected = False

bec_widgets/widgets/progress/scan_progressbar/scan_progressbar.py

@@ -65,9 +65,7 @@ def __init__(
         self.progress_tracker.progress_updated.connect(self._on_progress_snapshot)
         self.progress_tracker.progress_finished.connect(self._on_progress_finished)
         self.progress_tracker.source_changed.connect(self._on_progress_source_changed)
-        self.progress_tracker.start(
-            source=ProgressSource.SCAN_PROGRESS, connect_queue=True, refresh_queue=True
-        )
+        self.progress_tracker.start(source=ProgressSource.SCAN_PROGRESS)
 
     @property
     def task(self):
@@ -109,12 +107,6 @@ def _progress_device(self):
     def _progress_device(self, value):
         self.progress_tracker._progress_device = value
 
-    def connect_to_queue(self):
-        """
-        Connect to the queue status signal.
-        """
-        self.progress_tracker.connect_to_queue()
-
     def set_progress_source(self, source: ProgressSource, device=None):
         """
         Set the source of the progress.
@@ -155,6 +147,7 @@ def _on_progress_source_changed(self, snapshot: ProgressSnapshot):
 
     def _on_progress_snapshot(self, snapshot: ProgressSnapshot):
         self.update_labels()
+        self.update_source_label(snapshot.source, device=snapshot.device)
         self.progressbar.set_maximum(snapshot.max_value)
         self.progressbar.state = ProgressState.from_bec_status(snapshot.status)
         self.progressbar.set_value(snapshot.value)
@@ -200,13 +193,6 @@ def update_labels(self):
         self.ui.elapsed_time_label.setText(self.task.time_elapsed)
         self.ui.remaining_time_label.setText(self.task.time_remaining)
 
-    @SafeSlot(dict, dict, verify_sender=True)
-    def on_queue_update(self, msg_content, metadata):
-        """
-        Update the progress bar based on the queue status.
-        """
-        self.progress_tracker.on_queue_update(msg_content, metadata)
-
     def cleanup(self):
         self.progress_tracker.cleanup()
         self.progressbar.close()