feat(widgets): added simple bec queue widget

2026-05-07 23:32:12 +02:00 · 2024-06-26 16:19:19 +02:00
parent ca02132c8d
commit 3faee98ec8
7 changed files with 311 additions and 0 deletions
@@ -0,0 +1,111 @@
+from bec_lib.endpoints import MessageEndpoints
+from qtpy.QtCore import Qt, Slot
+from qtpy.QtWidgets import QHeaderView, QTableWidget, QTableWidgetItem, QWidget
+
+from bec_widgets.utils.bec_connector import BECConnector, ConnectionConfig
+
+
+class BECQueue(BECConnector, QTableWidget):
+    """
+    Widget to display the BEC queue.
+    """
+
+    def __init__(
+        self,
+        parent: QWidget | None = None,
+        client=None,
+        config: ConnectionConfig = None,
+        gui_id: str = None,
+    ):
+        super().__init__(client, config, gui_id)
+        QTableWidget.__init__(self, parent=parent)
+        self.setColumnCount(3)
+        self.setHorizontalHeaderLabels(["Scan Number", "Type", "Status"])
+        header = self.horizontalHeader()
+        header.setSectionResizeMode(QHeaderView.Stretch)
+        self.bec_dispatcher.connect_slot(self.update_queue, MessageEndpoints.scan_queue_status())
+        self.reset_content()
+
+    @Slot(dict, dict)
+    def update_queue(self, content, _metadata):
+        """
+        Update the queue table with the latest queue information.
+
+        Args:
+            content (dict): The queue content.
+            _metadata (dict): The metadata.
+        """
+        # only show the primary queue for now
+        queue_info = content.get("queue", {}).get("primary", {}).get("info", [])
+        self.setRowCount(len(queue_info))
+        self.clearContents()
+
+        if not queue_info:
+            self.reset_content()
+            return
+
+        for index, item in enumerate(queue_info):
+            blocks = item.get("request_blocks", [])
+            scan_types = []
+            scan_numbers = []
+            status = item.get("status", "")
+            for request_block in blocks:
+                scan_type = request_block.get("content", {}).get("scan_type", "")
+                if scan_type:
+                    scan_types.append(scan_type)
+                scan_number = request_block.get("scan_number", "")
+                if scan_number:
+                    scan_numbers.append(str(scan_number))
+            if scan_types:
+                scan_types = ", ".join(scan_types)
+            if scan_numbers:
+                scan_numbers = ", ".join(scan_numbers)
+            self.set_row(index, scan_numbers, scan_types, status)
+
+    def format_item(self, content: str) -> QTableWidgetItem:
+        """
+        Format the content of the table item.
+
+        Args:
+            content (str): The content to be formatted.
+
+        Returns:
+            QTableWidgetItem: The formatted item.
+        """
+        item = QTableWidgetItem(content)
+        item.setTextAlignment(Qt.AlignHCenter | Qt.AlignVCenter)
+        return item
+
+    def set_row(self, index: int, scan_number: str, scan_type: str, status: str):
+        """
+        Set the row of the table.
+
+        Args:
+            index (int): The index of the row.
+            scan_number (str): The scan number.
+            scan_type (str): The scan type.
+            status (str): The status.
+        """
+
+        self.setItem(index, 0, self.format_item(scan_number))
+        self.setItem(index, 1, self.format_item(scan_type))
+        self.setItem(index, 2, self.format_item(status))
+
+    def reset_content(self):
+        """
+        Reset the content of the table.
+        """
+
+        self.setRowCount(1)
+        self.set_row(0, "", "", "")
+
+
+if __name__ == "__main__":  # pragma: no cover
+    import sys
+
+    from qtpy.QtWidgets import QApplication
+
+    app = QApplication(sys.argv)
+    widget = BECQueue()
+    widget.show()
+    sys.exit(app.exec_())