diff --git a/bec_widgets/cli/__init__.py b/bec_widgets/cli/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/bec_widgets/cli/client.py b/bec_widgets/cli/client.py new file mode 100644 index 00000000..927263fc --- /dev/null +++ b/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. + """ diff --git a/bec_widgets/cli/client_utils.py b/bec_widgets/cli/client_utils.py new file mode 100644 index 00000000..c769c629 --- /dev/null +++ b/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 diff --git a/bec_widgets/cli/generate_cli.py b/bec_widgets/cli/generate_cli.py new file mode 100644 index 00000000..bce160da --- /dev/null +++ b/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") diff --git a/bec_widgets/cli/server.py b/bec_widgets/cli/server.py new file mode 100644 index 00000000..c4445f12 --- /dev/null +++ b/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) diff --git a/bec_widgets/widgets/figure/figure.py b/bec_widgets/widgets/figure/figure.py index 70061caf..1f18715c 100644 --- a/bec_widgets/widgets/figure/figure.py +++ b/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): diff --git a/bec_widgets/widgets/plots/plot_base.py b/bec_widgets/widgets/plots/plot_base.py index 7511e96c..322d730e 100644 --- a/bec_widgets/widgets/plots/plot_base.py +++ b/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.