feat(cli): added auto updates plugin support

bec_widgets/cli/__init__.py

@@ -1 +1,2 @@
+from .auto_updates import AutoUpdates, ScanInfo
 from .client import BECFigure

bec_widgets/cli/auto_updates.py

@@ -0,0 +1,118 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from pydantic import BaseModel
+
+if TYPE_CHECKING:
+    from .client import BECFigure
+
+
+class ScanInfo(BaseModel):
+    scan_id: str
+    scan_number: int
+    scan_name: str
+    scan_report_devices: list
+    monitored_devices: list
+    status: str
+
+
+class AutoUpdates:
+    def __init__(self, figure: BECFigure, enabled: bool = True):
+        self.enabled = enabled
+        self.figure = figure
+
+    @staticmethod
+    def get_scan_info(msg) -> ScanInfo:
+        """
+        Update the script with the given data.
+        """
+        info = msg.info
+        status = msg.status
+        scan_id = msg.scan_id
+        scan_number = info.get("scan_number", 0)
+        scan_name = info.get("scan_name", "Unknown")
+        scan_report_devices = info.get("scan_report_devices", [])
+        monitored_devices = info.get("readout_priority", {}).get("monitored", [])
+        monitored_devices = [dev for dev in monitored_devices if dev not in scan_report_devices]
+        return ScanInfo(
+            scan_id=scan_id,
+            scan_number=scan_number,
+            scan_name=scan_name,
+            scan_report_devices=scan_report_devices,
+            monitored_devices=monitored_devices,
+            status=status,
+        )
+
+    def run(self, msg):
+        """
+        Run the update function if enabled.
+        """
+        if not self.enabled:
+            return
+        if msg.status != "open":
+            return
+        info = self.get_scan_info(msg)
+        self.handler(info)
+
+    @staticmethod
+    def get_selected_device(monitored_devices, selected_device):
+        """
+        Get the selected device for the plot. If no device is selected, the first
+        device in the monitored devices list is selected.
+        """
+        if selected_device:
+            return selected_device
+        if len(monitored_devices) > 0:
+            sel_device = monitored_devices[0]
+            return sel_device
+        return None
+
+    def handler(self, info: ScanInfo) -> None:
+        """
+        Default update function.
+        """
+        if info.scan_name == "line_scan" and info.scan_report_devices:
+            self.simple_line_scan(info)
+            return
+        if info.scan_name == "grid_scan" and info.scan_report_devices:
+            self.simple_grid_scan(info)
+            return
+        if info.scan_report_devices:
+            self.best_effort(info)
+            return
+
+    def simple_line_scan(self, info: ScanInfo) -> None:
+        """
+        Simple line scan.
+        """
+        dev_x = info.scan_report_devices[0]
+        dev_y = self.get_selected_device(info.monitored_devices, self.figure.selected_device)
+        if not dev_y:
+            return
+        self.figure.clear_all()
+        plt = self.figure.plot(dev_x, dev_y)
+        plt.set(title=f"Scan {info.scan_number}", x_label=dev_x, y_label=dev_y)
+
+    def simple_grid_scan(self, info: ScanInfo) -> None:
+        """
+        Simple grid scan.
+        """
+        dev_x = info.scan_report_devices[0]
+        dev_y = info.scan_report_devices[1]
+        dev_z = self.get_selected_device(info.monitored_devices, self.figure.selected_device)
+        self.figure.clear_all()
+        plt = self.figure.plot(dev_x, dev_y, dev_z, label=f"Scan {info.scan_number}")
+        plt.set(title=f"Scan {info.scan_number}", x_label=dev_x, y_label=dev_y)
+
+    def best_effort(self, info: ScanInfo) -> None:
+        """
+        Best effort scan.
+        """
+        dev_x = info.scan_report_devices[0]
+        dev_y = self.get_selected_device(info.monitored_devices, self.figure.selected_device)
+        if not dev_y:
+            return
+        self.figure.clear_all()
+        plt = self.figure.plot(dev_x, dev_y, label=f"Scan {info.scan_number}")
+        plt.set(title=f"Scan {info.scan_number}", x_label=dev_x, y_label=dev_y)

bec_widgets/cli/client_utils.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import importlib
+import importlib.metadata as imd
 import os
 import select
 import subprocess
@@ -17,6 +18,7 @@ from bec_lib.device import DeviceBase
 from qtpy.QtCore import QCoreApplication
 
 import bec_widgets.cli.client as client
+from bec_widgets.cli.auto_updates import AutoUpdates
 from bec_widgets.utils.bec_dispatcher import BECDispatcher
 
 if TYPE_CHECKING:
@@ -54,68 +56,22 @@ def rpc_call(func):
     return wrapper
 
 
-def get_selected_device(monitored_devices, selected_device):
-    """
-    Get the selected device for the plot. If no device is selected, the first
-    device in the monitored devices list is selected.
-    """
-    if selected_device:
-        return selected_device
-    if len(monitored_devices) > 0:
-        sel_device = monitored_devices[0]
-        return sel_device
-    return None
-
-
-def update_script(figure: BECFigure, msg):
-    """
-    Update the script with the given data.
-    """
-    info = msg.info
-    status = msg.status
-    scan_id = msg.scan_id
-    scan_number = info.get("scan_number", 0)
-    scan_name = info.get("scan_name", "Unknown")
-    scan_report_devices = info.get("scan_report_devices", [])
-    monitored_devices = info.get("readout_priority", {}).get("monitored", [])
-    monitored_devices = [dev for dev in monitored_devices if dev not in scan_report_devices]
-
-    if scan_name == "line_scan" and scan_report_devices:
-        dev_x = scan_report_devices[0]
-        dev_y = get_selected_device(monitored_devices, figure.selected_device)
-        print(f"Selected device: {dev_y}")
-        if not dev_y:
-            return
-        figure.clear_all()
-        plt = figure.plot(dev_x, dev_y)
-        plt.set(title=f"Scan {scan_number}", x_label=dev_x, y_label=dev_y)
-    elif scan_name == "grid_scan" and scan_report_devices:
-        print(f"Scan {scan_number} is running")
-        dev_x = scan_report_devices[0]
-        dev_y = scan_report_devices[1]
-        dev_z = get_selected_device(monitored_devices, figure.selected_device)
-        figure.clear_all()
-        plt = figure.plot(dev_x, dev_y, dev_z, label=f"Scan {scan_number}")
-        plt.set(title=f"Scan {scan_number}", x_label=dev_x, y_label=dev_y)
-    elif scan_report_devices:
-        dev_x = scan_report_devices[0]
-        dev_y = get_selected_device(monitored_devices, figure.selected_device)
-        if not dev_y:
-            return
-        figure.clear_all()
-        plt = figure.plot(dev_x, dev_y, label=f"Scan {scan_number}")
-        plt.set(title=f"Scan {scan_number}", x_label=dev_x, y_label=dev_y)
-
-
 class BECFigureClientMixin:
     def __init__(self, **kwargs) -> None:
         super().__init__(**kwargs)
         self._process = None
-        self.update_script = update_script
+        self.update_script = self._get_update_script()
         self._target_endpoint = MessageEndpoints.scan_status()
         self._selected_device = None
         self.stderr_output = []
 
+    def _get_update_script(self) -> AutoUpdates:
+        eps = imd.entry_points(group="bec.widgets.auto_updates")
+        for ep in eps:
+            if ep.name == "plugin_widgets_update":
+                return ep.load()(figure=self)
+        return None
+
     @property
     def selected_device(self):
         """
@@ -147,8 +103,7 @@ class BECFigureClientMixin:
         if isinstance(msg, messages.ScanStatusMessage):
             if not self.gui_is_alive():
                 return
-            if msg.status == "open":
-                self.update_script(self, msg)
+            self.update_script.run(msg)
 
     def show(self) -> None:
         """