feat(cli): added cli interface, rebased

bec_widgets/cli/client.py

@@ -0,0 +1,161 @@
+# This file was automatically generated by generate_cli.py
+
+from typing import Literal, Optional, overload
+
+from bec_widgets.cli.client_utils import BECFigureClientMixin, RPCBase, rpc_call
+
+
+class BECWaveform1D(RPCBase):
+    @rpc_call
+    def set_x_label(self, label: "str"):
+        """
+        Set the label of the x-axis.
+        Args:
+            label(str): Label of the x-axis.
+        """
+
+    @rpc_call
+    def set_y_label(self, label: "str"):
+        """
+        Set the label of the y-axis.
+        Args:
+            label(str): Label of the y-axis.
+        """
+
+    @rpc_call
+    def set_x_scale(self, scale: "Literal['linear', 'log']" = "linear"):
+        """
+        Set the scale of the x-axis.
+        Args:
+            scale(Literal["linear", "log"]): Scale of the x-axis.
+        """
+
+    @rpc_call
+    def set_y_scale(self, scale: "Literal['linear', 'log']" = "linear"):
+        """
+        Set the scale of the y-axis.
+        Args:
+            scale(Literal["linear", "log"]): Scale of the y-axis.
+        """
+
+    @rpc_call
+    def set_x_lim(self, x_lim: "tuple") -> "None":
+        """
+        Set the limits of the x-axis.
+        Args:
+            x_lim(tuple): Limits of the x-axis.
+        """
+
+    @rpc_call
+    def set_y_lim(self, y_lim: "tuple") -> "None":
+        """
+        Set the limits of the y-axis.
+        Args:
+            y_lim(tuple): Limits of the y-axis.
+        """
+
+    @rpc_call
+    def set_grid(self, x: "bool" = False, y: "bool" = False):
+        """
+        Set the grid of the plot widget.
+        Args:
+            x(bool): Show grid on the x-axis.
+            y(bool): Show grid on the y-axis.
+        """
+
+    @rpc_call
+    def plot_data(self, data_x: "list | np.ndarray", data_y: "list | np.ndarray", **kwargs):
+        """
+        Plot custom data on the plot widget. These data are not saved in config.
+        Args:
+            data_x(list|np.ndarray): x-axis data
+            data_y(list|np.ndarray): y-axis data
+            **kwargs: Keyword arguments for the plot.
+        """
+
+    @rpc_call
+    def remove(self):
+        """
+        Remove the plot widget from the figure.
+        """
+
+    @rpc_call
+    def add_scan(
+        self,
+        x_name: str,
+        x_entry: str,
+        y_name: str,
+        y_entry: str,
+        color: Optional[str] = None,
+        label: Optional[str] = None,
+        symbol: Optional[str] = None,
+        symbol_size: Optional[int] = None,
+        symbol_color: Optional[str] = None,
+        pen_width: Optional[int] = None,
+        pen_style: Optional[Literal["solid", "dash", "dot", "dashdot"]] = None,
+    ):
+        """
+        None
+        """
+
+
+class BECFigure(RPCBase, BECFigureClientMixin):
+    @overload
+    def add_widget(
+        self,
+        widget_type: "Literal['Waveform1D']" = "Waveform1D",
+        widget_id: "str" = Ellipsis,
+        row: "int" = Ellipsis,
+        col: "int" = Ellipsis,
+        config: "dict" = Ellipsis,
+        **axis_kwargs
+    ) -> "BECWaveform1D": ...
+
+    @overload
+    def add_widget(
+        self,
+        widget_type: "Literal['PlotBase']" = "PlotBase",
+        widget_id: "str" = Ellipsis,
+        row: "int" = Ellipsis,
+        col: "int" = Ellipsis,
+        config: "dict" = Ellipsis,
+        **axis_kwargs
+    ) -> "BECPlotBase": ...
+
+    @rpc_call
+    def add_widget(
+        self,
+        widget_type: "Literal['PlotBase', 'Waveform1D']" = "PlotBase",
+        widget_id: "str" = None,
+        row: "int" = None,
+        col: "int" = None,
+        config: "dict" = None,
+        **axis_kwargs
+    ) -> "BECPlotBase":
+        """
+        Add a widget to the figure at the specified position.
+        Args:
+            widget_type(Literal["PlotBase","Waveform1D"]): The type of the widget to add.
+            widget_id(str): The unique identifier of the widget. If not provided, a unique ID will be generated.
+            row(int): The row coordinate of the widget in the figure. If not provided, the next empty row will be used.
+            col(int): The column coordinate of the widget in the figure. If not provided, the next empty column will be used.
+            config(dict): Additional configuration for the widget.
+            **axis_kwargs(dict): Additional axis properties to set on the widget after creation.
+        """
+
+    @rpc_call
+    def remove(
+        self,
+        row: "int" = None,
+        col: "int" = None,
+        widget_id: "str" = None,
+        coordinates: "tuple[int, int]" = None,
+    ) -> "None":
+        """
+        Remove a widget from the figure. Can be removed by its unique identifier or by its coordinates.
+        Args:
+            row(int): The row coordinate of the widget to remove.
+            col(int): The column coordinate of the widget to remove.
+            widget_id(str): The unique identifier of the widget to remove.
+            coordinates(tuple[int, int], optional): The coordinates of the widget to remove.
+        """

bec_widgets/cli/client_utils.py

@@ -0,0 +1,148 @@
+import importlib
+import select
+import subprocess
+import uuid
+from functools import wraps
+
+from bec_lib import MessageEndpoints, messages
+
+import bec_widgets.cli.client as client
+from bec_widgets.utils.bec_dispatcher import BECDispatcher
+
+
+def rpc_call(func):
+    """
+    A decorator for calling a function on the server.
+
+    Args:
+        func: The function to call.
+
+    Returns:
+        The result of the function call.
+    """
+
+    @wraps(func)
+    def wrapper(self, *args, **kwargs):
+        return self._run_rpc(func.__name__, *args, **kwargs)
+
+    return wrapper
+
+
+class BECFigureClientMixin:
+    def __init__(self, **kwargs) -> None:
+        super().__init__(**kwargs)
+        self._process = None
+
+    def show(self) -> None:
+        """
+        Show the figure.
+        """
+        if self._process is None or self._process.poll() is not None:
+            self._start_plot_process()
+
+    def close(self) -> None:
+        """
+        Close the figure.
+        """
+        if self._process is None:
+            return
+        self._run_rpc("close", (), wait_for_rpc_response=False)
+        self._process.kill()
+        self._process = None
+
+    def _start_plot_process(self) -> None:
+        """
+        Start the plot in a new process.
+        """
+        # pylint: disable=subprocess-run-check
+        monitor_module = importlib.import_module("bec_widgets.cli.server")
+        monitor_path = monitor_module.__file__
+
+        command = f"python {monitor_path} --id {self._gui_id}"
+        self._process = subprocess.Popen(
+            command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE
+        )
+
+    def print_log(self) -> None:
+        """
+        Print the log of the plot process.
+        """
+        if self._process is None:
+            return
+        print(self._get_stderr_output())
+
+    def _get_stderr_output(self) -> str:
+        stderr_output = []
+        while self._process.poll() is not None:
+            readylist, _, _ = select.select([self._process.stderr], [], [], 0.1)
+            if not readylist:
+                break
+            line = self._process.stderr.readline()
+            if not line:
+                break
+            stderr_output.append(line.decode("utf-8"))
+        return "".join(stderr_output)
+
+    def __del__(self) -> None:
+        self.close()
+
+
+class RPCBase:
+
+    def __init__(self, gui_id: str = None, config: dict = None, **kwargs) -> None:
+        self._client = BECDispatcher().client
+        self._config = config if config is not None else {}
+        self._gui_id = gui_id if gui_id is not None else str(uuid.uuid4())
+        super().__init__(**kwargs)
+        print(f"RPCBase: {self._gui_id}")
+
+    def _run_rpc(self, method, *args, wait_for_rpc_response=True, **kwargs):
+        """
+        Run the RPC call.
+
+        Args:
+            method: The method to call.
+            args: The arguments to pass to the method.
+            wait_for_rpc_response: Whether to wait for the RPC response.
+            kwargs: The keyword arguments to pass to the method.
+
+        Returns:
+            The result of the RPC call.
+        """
+        request_id = str(uuid.uuid4())
+        rpc_msg = messages.GUIInstructionMessage(
+            action=method,
+            parameter={"args": args, "kwargs": kwargs, "gui_id": self._gui_id},
+            metadata={"request_id": request_id},
+        )
+        print(f"RPCBase: {rpc_msg}")
+        receiver = self._config.get("parent_figure_id", self._gui_id)
+        self._client.producer.send(MessageEndpoints.gui_instructions(receiver), rpc_msg)
+
+        if not wait_for_rpc_response:
+            return None
+        response = self._wait_for_response(request_id)
+        # get class name
+        if not response.content["accepted"]:
+            raise ValueError(response.content["message"]["error"])
+        msg_result = response.content["message"].get("result")
+        if not msg_result:
+            return None
+        cls = msg_result.pop("widget_class", None)
+        if not cls:
+            return msg_result
+
+        cls = getattr(client, cls)
+        print(msg_result)
+        return cls(**msg_result)
+
+    def _wait_for_response(self, request_id):
+        """
+        Wait for the response from the server.
+        """
+        response = None
+        while response is None:
+            response = self._client.producer.get(
+                MessageEndpoints.gui_instruction_response(request_id)
+            )
+        return response

bec_widgets/cli/generate_cli.py

@@ -0,0 +1,84 @@
+import inspect
+import typing
+
+
+class ClientGenerator:
+
+    def __init__(self):
+        self.header = """# This file was automatically generated by generate_cli.py\n
+from bec_widgets.cli.client_utils import rpc_call, RPCBase, BECFigureClientMixin
+from typing import Literal, Optional, overload"""
+
+        self.content = ""
+
+    def generate_client(self, published_classes: list):
+        """
+        Generate the client for the published classes.
+
+        Args:
+            published_classes(list): The list of published classes (e.g. [BECWaveform1D, BECFigure]).
+        """
+        for cls in published_classes:
+            self.content += "\n\n"
+            self.generate_content_for_class(cls)
+
+    def generate_content_for_class(self, cls):
+        """
+        Generate the content for the class.
+        Args:
+            cls: The class for which to generate the content.
+        """
+
+        class_name = cls.__name__
+        module = cls.__module__
+
+        # Generate the header
+        #         self.header += f"""
+        # from {module} import {class_name}"""
+
+        # Generate the content
+        if cls.__name__ == "BECFigure":
+            self.content += f"""
+class {class_name}(RPCBase, BECFigureClientMixin):"""
+        else:
+            self.content += f"""
+class {class_name}(RPCBase):"""
+        for method in cls.USER_ACCESS:
+            obj = getattr(cls, method)
+            sig = str(inspect.signature(obj))
+            doc = inspect.getdoc(obj)
+            overloads = typing.get_overloads(obj)
+            for overload in overloads:
+                sig_overload = str(inspect.signature(overload))
+                self.content += f"""
+    @overload
+    def {method}{str(sig_overload)}: ...
+    """
+
+            self.content += f"""
+    @rpc_call
+    def {method}{str(sig)}:
+        \"\"\"
+{doc}
+        \"\"\""""
+
+    def write(self, file_name: str):
+        """
+        Write the content to a file.
+
+        Args:
+            file_name(str): The name of the file to write to.
+        """
+        with open(file_name, "w", encoding="utf-8") as file:
+            file.write(self.header)
+            file.write(self.content)
+
+
+if __name__ == "__main__":
+    from bec_widgets.widgets.figure import BECFigure
+    from bec_widgets.widgets.plots import BECWaveform1D
+
+    clss = [BECWaveform1D, BECFigure]
+    generator = ClientGenerator()
+    generator.generate_client(clss)
+    generator.write("bec_widgets/cli/client.py")

bec_widgets/cli/server.py

@@ -0,0 +1,88 @@
+import inspect
+
+from bec_lib import MessageEndpoints, messages
+
+from bec_widgets.utils import BECDispatcher
+from bec_widgets.widgets.figure import BECFigure
+from bec_widgets.widgets.plots import BECPlotBase, BECWaveform1D
+
+
+class BECWidgetsCLIServer:
+    WIDGETS = [BECWaveform1D, BECFigure]
+
+    def __init__(self, gui_id: str = None) -> None:
+
+        self.dispatcher = BECDispatcher()
+        self.client = self.dispatcher.client
+        self.client.start()
+        self.gui_id = gui_id
+        self.fig = BECFigure(gui_id=self.gui_id)
+        print(f"Server started with gui_id {self.gui_id}")
+
+        self._rpc_thread = self.client.connector.consumer(
+            topics=MessageEndpoints.gui_instructions(self.gui_id),
+            cb=self._rpc_update_handler,
+            parent=self,
+        )
+        self._rpc_thread.start()
+        self.fig.start()
+
+    @staticmethod
+    def _rpc_update_handler(msg, parent):
+        parent.on_rpc_update(msg.value)
+
+    def on_rpc_update(self, msg: messages.GUIInstructionMessage):
+        try:
+            method = msg.action
+            args = msg.parameter.get("args", [])
+            kwargs = msg.parameter.get("kwargs", {})
+            request_id = msg.metadata.get("request_id")
+            obj = self.get_object_from_config(msg.parameter)
+            res = self.run_rpc(obj, method, args, kwargs)
+            self.send_response(request_id, True, {"result": res})
+        except Exception as e:
+            print(e)
+            self.send_response(request_id, False, {"error": str(e)})
+
+    def send_response(self, request_id: str, accepted: bool, msg: dict):
+        self.client.producer.set(
+            MessageEndpoints.gui_instruction_response(request_id),
+            messages.RequestResponseMessage(accepted=accepted, message=msg),
+            expire=60,
+        )
+
+    def get_object_from_config(self, config: dict):
+        gui_id = config.get("gui_id")
+        if gui_id == self.fig.gui_id:
+            return self.fig
+        if gui_id in self.fig.widgets:
+            obj = self.fig.widgets[config["gui_id"]]
+            return obj
+        raise ValueError(f"Object with gui_id {gui_id} not found")
+
+    def run_rpc(self, obj, method, args, kwargs):
+        method_obj = getattr(obj, method)
+        # check if the method accepts args and kwargs
+        sig = inspect.signature(method_obj)
+        if sig.parameters:
+            res = method_obj(*args, **kwargs)
+        else:
+            res = method_obj()
+        if isinstance(res, BECPlotBase):
+            res = {
+                "gui_id": res.gui_id,
+                "widget_class": res.__class__.__name__,
+                "config": res.config.model_dump(),
+            }
+        return res
+
+
+if __name__ == "__main__":
+    import argparse
+
+    parser = argparse.ArgumentParser(description="BEC Widgets CLI Server")
+    parser.add_argument("--id", type=str, help="The id of the server")
+
+    args = parser.parse_args()
+
+    server = BECWidgetsCLIServer(gui_id=args.id)

bec_widgets/widgets/figure/figure.py

@@ -1,26 +1,24 @@
+from __future__ import annotations
+
 # pylint: disable = no-name-in-module,missing-module-docstring
 import itertools
 import os
 import sys
+from typing import Literal, Optional, overload
 from collections import defaultdict
 from dataclasses import dataclass
 from typing import Literal, Optional
 
 import numpy as np
 import pyqtgraph as pg
+from bec_lib.utils import user_access
 from qtpy.QtWidgets import QVBoxLayout, QMainWindow
 from pydantic import Field
 from pyqtgraph.Qt import uic
 from qtpy.QtWidgets import QApplication, QWidget
 
-from bec_lib.utils import user_access
-
-from bec_widgets.utils import (
-    BECDispatcher,
-    BECConnector,
-    ConnectionConfig,
-)
-from bec_widgets.widgets.plots import WidgetConfig, BECPlotBase, Waveform1DConfig, BECWaveform1D
+from bec_widgets.utils import BECConnector, BECDispatcher, ConnectionConfig
+from bec_widgets.widgets.plots import BECPlotBase, BECWaveform1D, Waveform1DConfig, WidgetConfig
 
 
 class FigureConfig(ConnectionConfig):
@@ -86,13 +84,15 @@ class WidgetHandler:
 
 
 class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
+    USER_ACCESS = ["add_widget", "remove"]
+
     def __init__(
         self,
         parent: Optional[QWidget] = None,
         config: Optional[FigureConfig] = None,
         client=None,
         gui_id: Optional[str] = None,
-    ):
+    ) -> None:
         if config is None:
             config = FigureConfig(widget_class=self.__class__.__name__)
         else:
@@ -144,7 +144,50 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
             for col_idx, widget in enumerate(row):
                 self.addItem(self.widgets[widget], row=row_idx, col=col_idx)
 
-    @user_access
+    @overload
+    def add_widget(
+        self,
+        widget_type: Literal["Waveform1D"] = "Waveform1D",
+        widget_id: str = ...,
+        row: int = ...,
+        col: int = ...,
+        config: dict = ...,
+        **axis_kwargs,
+    ) -> BECWaveform1D: ...
+
+    @overload
+    def add_widget(
+        self,
+        widget_type: Literal["PlotBase"] = "PlotBase",
+        widget_id: str = ...,
+        row: int = ...,
+        col: int = ...,
+        config: dict = ...,
+        **axis_kwargs,
+    ) -> BECPlotBase: ...
+
+    # @overload
+    # def add_widget(
+    #     self,
+    #     widget_type: Literal["Waveform1D"] = "Waveform1D",
+    #     widget_id: str = None,
+    #     row: int = None,
+    #     col: int = None,
+    #     config: dict = None,
+    #     **axis_kwargs,
+    # ) -> BECWaveform1D: ...
+
+    # @overload
+    # def add_widget(
+    #     self,
+    #     widget_type: Literal["PlotBase"] = "PlotBase",
+    #     widget_id: str = None,
+    #     row: int = None,
+    #     col: int = None,
+    #     config: dict = None,
+    #     **axis_kwargs,
+    # ) -> BECPlotBase: ...
+
     def add_widget(
         self,
         widget_type: Literal["PlotBase", "Waveform1D"] = "PlotBase",
@@ -153,7 +196,7 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
         col: int = None,
         config=None,
         **axis_kwargs,
-    ):
+    ) -> BECPlotBase:
         """
         Add a widget to the figure at the specified position.
         Args:
@@ -212,7 +255,8 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
         # Reflect the grid coordinates
         self.change_grid(widget_id, row, col)
 
-    @user_access
+        return widget
+
     def remove(
         self,
         row: int = None,
@@ -310,7 +354,7 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
         """Generate a unique widget ID."""
         existing_ids = set(self.widgets.keys())
         for i in itertools.count(1):
-            widget_id = f"Widget {i}"
+            widget_id = f"widget_{i}"
             if widget_id not in existing_ids:
                 return widget_id
 
@@ -329,9 +373,9 @@ class BECFigure(BECConnector, pg.GraphicsLayoutWidget):
 ##################################################
 ##################################################
 
-from qtconsole.rich_jupyter_widget import RichJupyterWidget
 from qtconsole.inprocess import QtInProcessKernelManager
 import matplotlib.pyplot as plt
+from qtconsole.rich_jupyter_widget import RichJupyterWidget
 
 
 class JupyterConsoleWidget(RichJupyterWidget):

bec_widgets/widgets/plots/plot_base.py

@@ -1,11 +1,13 @@
-from typing import Optional, Literal
+from __future__ import annotations
+
+from typing import Literal, Optional
 
-import pyqtgraph as pg
 import numpy as np
+import pyqtgraph as pg
+from bec_lib.utils import user_access
 from pydantic import BaseModel, Field
 from qtpy.QtWidgets import QWidget
 
-from bec_lib.utils import user_access
 from bec_widgets.utils import BECConnector, ConnectionConfig
 
 
@@ -52,7 +54,6 @@ class BECPlotBase(BECConnector, pg.PlotItem):
 
         self.add_legend()
 
-    @user_access
     def set(self, **kwargs) -> None:
         """
         Set the properties of the plot widget.
@@ -100,7 +101,6 @@ class BECPlotBase(BECConnector, pg.PlotItem):
 
         self.set(**{k: v for k, v in config_mappings.items() if v is not None})
 
-    @user_access
     def set_title(self, title: str):
         """
         Set the title of the plot widget.