bec/bec_widgets
0
0
Fork 0
mirror of https://github.com/bec-project/bec_widgets.git synced 2025-07-13 19:21:50 +02:00
Code Activity

fix(cli): fixed property access, rebased

Browse Source
This commit is contained in:
wakonig_k
2024-02-24 17:01:54 +01:00
committed by wyzula-jan
parent 4630d78fc2
commit f71dc5c5ab
5 changed files with 55 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
bec_widgets/cli/client.py
View File

@ -1,8 +1,9 @@
# This file was automatically generated by generate_cli.py
from bec_widgets.cli.client_utils import rpc_call, RPCBase, BECFigureClientMixin
from typing import Literal, Optional, overload
from bec_widgets.cli.client_utils import BECFigureClientMixin, RPCBase, rpc_call
class BECPlotBase(RPCBase):
@rpc_call
@ -185,14 +186,16 @@ class BECWaveform1D(RPCBase):
scan_index(int, optional): Index of the scan to be updated. Defaults to None.
"""
@property
@rpc_call
def curves(self) -> "list":
def curves(self) -> "list[BECCurve]":
"""
Get the curves of the plot widget as a list
Returns:
list: List of curves.
"""
@property
@rpc_call
def curves_data(self) -> "dict":
"""

5
bec_widgets/cli/client_utils.py
View File

@ -141,6 +141,11 @@ class RPCBase:
msg_result = response.content["message"].get("result")
if not msg_result:
return None
if isinstance(msg_result, list):
return [self._create_widget_from_msg_result(res) for res in msg_result]
return self._create_widget_from_msg_result(msg_result)
def _create_widget_from_msg_result(self, msg_result):
cls = msg_result.pop("widget_class", None)
if not cls:
return msg_result

25
bec_widgets/cli/generate_cli.py
View File

@ -44,18 +44,26 @@ class {class_name}(RPCBase, BECFigureClientMixin):"""
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"""
if isinstance(obj, property):
self.content += """
@property
@rpc_call"""
sig = str(inspect.signature(obj.fget))
doc = inspect.getdoc(obj.fget)
else:
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 += """
@rpc_call"""
self.content += f"""
@rpc_call
def {method}{str(sig)}:
\"\"\"
{doc}
@ -75,8 +83,9 @@ class {class_name}(RPCBase):"""
if __name__ == "__main__":
import os
from bec_widgets.widgets.figure import BECFigure
from bec_widgets.widgets.plots import BECWaveform1D, BECPlotBase # ,BECCurve
from bec_widgets.widgets.plots import BECPlotBase, BECWaveform1D # ,BECCurve
from bec_widgets.widgets.plots.waveform1d import BECCurve
current_path = os.path.dirname(__file__)

35
bec_widgets/cli/server.py
View File

@ -72,19 +72,30 @@ class BECWidgetsCLIServer:
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)
if not callable(method_obj):
res = method_obj
else:
res = method_obj()
if isinstance(res, BECConnector):
res = {
"gui_id": res.gui_id,
"widget_class": res.__class__.__name__,
"config": res.config.model_dump(),
}
sig = inspect.signature(method_obj)
if sig.parameters:
res = method_obj(*args, **kwargs)
else:
res = method_obj()
if isinstance(res, list):
res = [self.serialize_object(obj) for obj in res]
else:
res = self.serialize_object(res)
return res
def serialize_object(self, obj):
if isinstance(obj, BECConnector):
return {
"gui_id": obj.gui_id,
"widget_class": obj.__class__.__name__,
"config": obj.config.model_dump(),
}
return obj
if __name__ == "__main__":
import argparse
@ -94,5 +105,5 @@ if __name__ == "__main__":
args = parser.parse_args()
# server = BECWidgetsCLIServer(gui_id=args.id)
server = BECWidgetsCLIServer(gui_id="test")
server = BECWidgetsCLIServer(gui_id=args.id)
# server = BECWidgetsCLIServer(gui_id="test")

40
bec_widgets/widgets/plots/waveform1d.py
View File

@ -1,7 +1,7 @@
from __future__ import annotations
from collections import defaultdict
from typing import Literal, Optional, Any
from typing import Any, Literal, Optional
import numpy as np
import pyqtgraph as pg
@ -72,7 +72,6 @@ class BECCurve(BECConnector, pg.PlotDataItem):
"set_symbol_size",
"set_pen_width",
"set_pen_style",
"get_data",
]
def __init__(
@ -205,15 +204,6 @@ class BECCurve(BECConnector, pg.PlotDataItem):
self.config.pen_style = pen_style
self.apply_config()
def get_data(self) -> tuple[np.ndarray, np.ndarray]:
"""
Get the data of the curve.
Returns:
tuple[np.ndarray,np.ndarray]: X and Y data of the curve.
"""
x_data, y_data = self.getData()
return x_data, y_data
class BECWaveform1D(BECPlotBase):
USER_ACCESS = [
@ -221,12 +211,11 @@ class BECWaveform1D(BECPlotBase):
"add_curve_custom",
"remove_curve",
"scan_history",
"get_curves",
"get_curves_data",
"curves",
"curves_data",
"get_curve",
"get_curve_config",
"apply_config",
"get_all_data",
]
scan_signal_update = pyqtSignal()
@ -302,7 +291,7 @@ class BECWaveform1D(BECPlotBase):
self.gui_id = new_gui_id
self.config.gui_id = new_gui_id
for curve in self.curves:
for curve_id, curve in self.curves_data.items():
curve.config.parent_id = new_gui_id
def add_curve_by_config(self, curve_config: CurveConfig | dict) -> BECCurve:
@ -335,21 +324,6 @@ class BECWaveform1D(BECPlotBase):
else:
return curves[curve_id].config
def get_curves(self) -> list: # TODO discuss if it should be marked as @property for RPC
"""
Get the curves of the plot widget as a list
Returns:
list: List of curves.
"""
return self.curves
def get_curves_data(self) -> dict: # TODO discuss if it should be marked as @property for RPC
"""
Get the curves data of the plot widget as a dictionary
Returns:
dict: Dictionary of curves data.
"""
return self.curves_data
def get_curve(self, identifier) -> BECCurve:
"""
@ -650,11 +624,7 @@ class BECWaveform1D(BECPlotBase):
curve.setData(data_x, data_y)
def scan_history(
self,
scan_index: int = None,
scanID: str = None,
):
def scan_history(self, scan_index: int = None, scanID: str = None):
"""
Update the scan curves with the data from the scan storage.
Provide only one of scanID or scan_index.