# Copyright (c) DataLab Platform Developers, BSD 3-Clause license, see LICENSE file."""DataLab remote control----------------------This module provides utilities to control DataLab from a Python script (e.g. withSpyder) or from a Jupyter notebook.The :class:`RemoteClient` class provides the main interface to DataLab XML-RPC server."""from__future__importannotationsimportfunctoolsimportimportlibimportsysimportthreadingimporttimeimportwarningsfromcollections.abcimportCallablefromioimportBytesIOfromtypingimportTYPE_CHECKINGfromxmlrpc.clientimportBinary,ServerProxyfromxmlrpc.serverimportSimpleXMLRPCServerimportguidata.datasetasgdsimportnumpyasnpfromguidata.ioimportJSONReader,JSONWriterfromqtpyimportQtCoreasQCimportcdlfromcdl.configimportConf,initializefromcdl.core.baseproxyimportAbstractCDLControl,BaseProxyfromcdl.core.model.baseimportitems_to_json,json_to_itemsfromcdl.core.model.imageimportImageObj,create_imagefromcdl.core.model.signalimportSignalObj,create_signalfromcdl.envimportexecenvfromcdl.utils.miscimportis_version_at_leastifTYPE_CHECKING:fromcdl.core.gui.mainimportCDLMainWindow# pylint: disable=invalid-name # Allows short reference names like x, y, ...# pylint: disable=duplicate-codedefarray_to_rpcbinary(data:np.ndarray)->Binary:"""Convert NumPy array to XML-RPC Binary object, with shape and dtype. The array is converted to a binary string using NumPy's native binary format. Args: data: NumPy array to convert Returns: XML-RPC Binary object """dbytes=BytesIO()np.save(dbytes,data,allow_pickle=False)returnBinary(dbytes.getvalue())defrpcbinary_to_array(binary:Binary)->np.ndarray:"""Convert XML-RPC binary to NumPy array. Args: binary: XML-RPC Binary object Returns: NumPy array """dbytes=BytesIO(binary.data)returnnp.load(dbytes,allow_pickle=False)defdataset_to_json(param:gds.DataSet)->list[str]:"""Convert guidata DataSet to JSON data. The JSON data is a list of three elements: - The first element is the module name of the DataSet class - The second element is the class name of the DataSet class - The third element is the JSON data of the DataSet instance Args: param: guidata DataSet to convert Returns: JSON data """writer=JSONWriter()param.serialize(writer)param_json=writer.get_json()klass=param.__class__return[klass.__module__,klass.__name__,param_json]defjson_to_dataset(param_data:list[str])->gds.DataSet:"""Convert JSON data to guidata DataSet. Args: param_data: JSON data Returns: guidata DataSet """param_module,param_clsname,param_json=param_datamod=importlib.__import__(param_module,fromlist=[param_clsname])klass=getattr(mod,param_clsname)param=klass()reader=JSONReader(param_json)param.deserialize(reader)returnparamdefremote_call(func:Callable)->object:"""Decorator for method calling DataLab main window remotely"""@functools.wraps(func)defmethod_wrapper(*args,**kwargs):"""Decorator wrapper function"""self=args[0]# extracting 'self' from method argumentsself.is_ready=Falseoutput=func(*args,**kwargs)whilenotself.is_ready:QC.QCoreApplication.processEvents()time.sleep(0.05)returnoutputreturnmethod_wrapper# Note: RemoteServer can't inherit from AbstractCDLControl because it is a QThread# and most of the methods are not returning expected data typesclassRemoteServer(QC.QThread):"""XML-RPC server QThread"""SIG_SERVER_PORT=QC.Signal(int)SIG_CLOSE_APP=QC.Signal()SIG_RAISE_WINDOW=QC.Signal()SIG_ADD_OBJECT=QC.Signal(object)SIG_LOAD_FROM_FILES=QC.Signal(list)SIG_SELECT_OBJECTS=QC.Signal(list,str)SIG_SELECT_GROUPS=QC.Signal(list,str)SIG_SELECT_ALL_GROUPS=QC.Signal(str)SIG_DELETE_METADATA=QC.Signal(bool,bool)SIG_SWITCH_TO_PANEL=QC.Signal(str)SIG_TOGGLE_AUTO_REFRESH=QC.Signal(bool)SIG_TOGGLE_SHOW_TITLES=QC.Signal(bool)SIG_RESET_ALL=QC.Signal()SIG_SAVE_TO_H5=QC.Signal(str)SIG_OPEN_H5=QC.Signal(list,bool,bool)SIG_IMPORT_H5=QC.Signal(str,bool)SIG_CALC=QC.Signal(str,object)SIG_RUN_MACRO=QC.Signal(str)SIG_STOP_MACRO=QC.Signal(str)SIG_IMPORT_MACRO_FROM_FILE=QC.Signal(str)def__init__(self,win:CDLMainWindow)->None:QC.QThread.__init__(self)self.port:int=Noneself.is_ready=Trueself.server:SimpleXMLRPCServer|None=Noneself.win=winwin.SIG_READY.connect(self.cdl_is_ready)win.SIG_CLOSING.connect(self.shutdown_server)self.SIG_CLOSE_APP.connect(win.close)self.SIG_RAISE_WINDOW.connect(win.raise_window)self.SIG_ADD_OBJECT.connect(win.add_object)self.SIG_LOAD_FROM_FILES.connect(win.load_from_files)self.SIG_SELECT_OBJECTS.connect(win.select_objects)self.SIG_SELECT_GROUPS.connect(win.select_groups)self.SIG_SELECT_ALL_GROUPS.connect(lambdapanel:win.select_groups(None,panel))self.SIG_DELETE_METADATA.connect(win.delete_metadata)self.SIG_SWITCH_TO_PANEL.connect(win.set_current_panel)self.SIG_TOGGLE_AUTO_REFRESH.connect(win.toggle_auto_refresh)self.SIG_TOGGLE_SHOW_TITLES.connect(win.toggle_show_titles)self.SIG_RESET_ALL.connect(win.reset_all)self.SIG_SAVE_TO_H5.connect(win.save_to_h5_file)self.SIG_OPEN_H5.connect(win.open_h5_files)self.SIG_IMPORT_H5.connect(win.import_h5_file)self.SIG_CALC.connect(win.calc)self.SIG_RUN_MACRO.connect(win.run_macro)self.SIG_STOP_MACRO.connect(win.stop_macro)self.SIG_IMPORT_MACRO_FROM_FILE.connect(win.import_macro_from_file)defserve(self)->None:"""Start server and serve forever"""withSimpleXMLRPCServer(("127.0.0.1",0),logRequests=False,allow_none=True)asserver:self.server=serverserver.register_introspection_functions()self.register_functions(server)self.port=server.server_address[1]self.notify_port(self.port)withexecenv.context(xmlrpcport=self.port):server.serve_forever()defshutdown_server(self)->None:"""Shutdown server"""ifself.serverisnotNone:self.server.shutdown()self.server=Nonedefnotify_port(self,port:int)->None:"""Notify automatically attributed port. This method is called after the server port has been automatically attributed. It notifies the port number to the main window. Args: port: Server port number """self.SIG_SERVER_PORT.emit(port)@classmethoddefcheck_remote_functions(cls)->None:"""Check if all AbstractCDLControl methods are implemented in RemoteServer"""mlist=[]formethodinAbstractCDLControl.get_public_methods():ifnothasattr(cls,method):mlist.append(method)ifmlist:raiseRuntimeError(f"{cls} is missing some methods: {','.join(mlist)}")defregister_functions(self,server:SimpleXMLRPCServer)->None:"""Register functions"""fornameinAbstractCDLControl.get_public_methods():server.register_function(getattr(self,name))defrun(self)->None:"""Thread execution method"""if"coverage"insys.modules:# The following is required to make coverage work with threading# pylint: disable=protected-accesssys.settrace(threading._trace_hook)self.serve()defcdl_is_ready(self)->None:"""Called when DataLab is ready to process new requests"""self.is_ready=True@staticmethoddefget_version()->str:"""Return DataLab public version"""returncdl.__version__defclose_application(self)->None:"""Close DataLab application"""self.SIG_CLOSE_APP.emit()defraise_window(self)->None:"""Raise DataLab window"""self.SIG_RAISE_WINDOW.emit()@remote_calldefget_current_panel(self)->str:"""Return current panel name. Returns: str: Panel name (valid values: 'signal', 'image', 'macro') """returnself.win.get_current_panel()@remote_calldefset_current_panel(self,panel:str)->None:"""Switch to panel. Args: panel (str): Panel name (valid values: 'signal', 'image', 'macro') """self.SIG_SWITCH_TO_PANEL.emit(panel)@remote_calldeftoggle_auto_refresh(self,state:bool)->None:"""Toggle auto refresh state. Args: state (bool): True to enable auto refresh, False to disable it """self.SIG_TOGGLE_AUTO_REFRESH.emit(state)@remote_calldeftoggle_show_titles(self,state:bool)->None:"""Toggle show titles state. Args: state (bool): True to enable show titles, False to disable it """self.SIG_TOGGLE_SHOW_TITLES.emit(state)@remote_calldefreset_all(self)->None:"""Reset all application data"""self.SIG_RESET_ALL.emit()@remote_calldefsave_to_h5_file(self,filename:str)->None:"""Save to a DataLab HDF5 file. Args: filename (str): HDF5 file name (with extension .h5) """self.SIG_SAVE_TO_H5.emit(filename)@remote_calldefopen_h5_files(self,h5files:list[str]|None=None,import_all:bool|None=None,reset_all:bool|None=None,)->None:"""Open a DataLab HDF5 file or import from any other HDF5 file. Args: h5files (list[str] | None): HDF5 file names. Defaults to None. import_all (bool | None): Import all objects from HDF5 file. Defaults to None. reset_all (bool | None): Reset all application data. Defaults to None. """self.SIG_OPEN_H5.emit(h5files,import_all,reset_all)@remote_calldefimport_h5_file(self,filename:str,reset_all:bool|None=None)->None:"""Open DataLab HDF5 browser to Import HDF5 file. Args: filename (str): HDF5 file name reset_all (bool | None): Reset all application data. Defaults to None. """self.SIG_IMPORT_H5.emit(filename,reset_all)@remote_calldefload_from_files(self,filenames:list[str])->None:"""Open objects from files in current panel (signals/images). Args: filenames: list of file names """self.SIG_LOAD_FROM_FILES.emit(filenames)@remote_calldefadd_signal(self,title:str,xbinary:Binary,ybinary:Binary,xunit:str|None=None,yunit:str|None=None,xlabel:str|None=None,ylabel:str|None=None,)->bool:# pylint: disable=too-many-arguments"""Add signal data to DataLab. Args: title (str): Signal title xbinary (Binary): X data ybinary (Binary): Y data xunit (str | None): X unit. Defaults to None. yunit (str | None): Y unit. Defaults to None. xlabel (str | None): X label. Defaults to None. ylabel (str | None): Y label. Defaults to None. Returns: bool: True if successful """xdata=rpcbinary_to_array(xbinary)ydata=rpcbinary_to_array(ybinary)signal=create_signal(title,xdata,ydata)signal.xunit=xunitsignal.yunit=yunitsignal.xlabel=xlabelsignal.ylabel=ylabelself.SIG_ADD_OBJECT.emit(signal)returnTrue@remote_calldefadd_image(self,title:str,zbinary:Binary,xunit:str|None=None,yunit:str|None=None,zunit:str|None=None,xlabel:str|None=None,ylabel:str|None=None,zlabel:str|None=None,)->bool:# pylint: disable=too-many-arguments"""Add image data to DataLab. Args: title (str): Image title zbinary (Binary): Z data xunit (str | None): X unit. Defaults to None. yunit (str | None): Y unit. Defaults to None. zunit (str | None): Z unit. Defaults to None. xlabel (str | None): X label. Defaults to None. ylabel (str | None): Y label. Defaults to None. zlabel (str | None): Z label. Defaults to None. Returns: bool: True if successful """data=rpcbinary_to_array(zbinary)image=create_image(title,data)image.xunit=xunitimage.yunit=yunitimage.zunit=zunitimage.xlabel=xlabelimage.ylabel=ylabelimage.zlabel=zlabelself.SIG_ADD_OBJECT.emit(image)returnTrue@remote_calldefget_sel_object_uuids(self,include_groups:bool=False)->list[str]:"""Return selected objects uuids. Args: include_groups: If True, also return objects from selected groups. Returns: List of selected objects uuids. """returnself.win.get_sel_object_uuids(include_groups)@remote_calldefselect_objects(self,selection:list[int|str],panel:str|None=None,)->None:"""Select objects in current panel. Args: selection: List of object numbers (1 to N) or uuids to select panel: panel name (valid values: "signal", "image"). If None, current panel is used. Defaults to None. """self.SIG_SELECT_OBJECTS.emit(selection,panel)@remote_calldefselect_groups(self,selection:list[int|str]|None=None,panel:str|None=None)->None:"""Select groups in current panel. Args: selection: List of group numbers (1 to N), or list of group uuids, or None to select all groups. Defaults to None. panel (str | None): panel name (valid values: "signal", "image"). If None, current panel is used. Defaults to None. """ifselectionisNone:self.SIG_SELECT_ALL_GROUPS.emit(panel)else:self.SIG_SELECT_GROUPS.emit(selection,panel)@remote_calldefdelete_metadata(self,refresh_plot:bool=True,keep_roi:bool=False)->None:"""Delete metadata of selected objects Args: refresh_plot: Refresh plot. Defaults to True. keep_roi: Keep ROI. Defaults to False. """self.SIG_DELETE_METADATA.emit(refresh_plot,keep_roi)@remote_calldefcalc(self,name:str,param_data:list[str]|None=None)->bool:"""Call compute function ``name`` in current panel's processor. Args: name: Compute function name param_data: Compute function parameters. Defaults to None. Returns: True if successful, False otherwise """ifparam_dataisNone:param=Noneelse:param=json_to_dataset(param_data)self.SIG_CALC.emit(name,param)returnTrue@remote_calldefget_group_titles_with_object_infos(self,)->tuple[list[str],list[list[str]],list[list[str]]]:"""Return groups titles and lists of inner objects uuids and titles. Returns: Tuple: groups titles, lists of inner objects uuids and titles """returnself.win.get_group_titles_with_object_infos()@remote_calldefget_object_titles(self,panel:str|None=None)->list[str]:"""Get object (signal/image) list for current panel. Objects are sorted by group number and object index in group. Args: panel: panel name (valid values: "signal", "image", "macro"). If None, current data panel is used (i.e. signal or image panel). Returns: List of object titles """returnself.win.get_object_titles(panel)@remote_calldefget_object(self,nb_id_title:int|str|None=None,panel:str|None=None,)->list[str]:"""Get object (signal/image) from index. Args: nb_id_title: Object number, or object id, or object title. Defaults to None (current object). panel: Panel name. Defaults to None (current panel). Returns: Object Raises: KeyError: if object not found """obj=self.win.get_object(nb_id_title,panel)ifobjisNone:returnNonereturndataset_to_json(obj)@remote_calldefget_object_uuids(self,panel:str|None=None)->list[str]:"""Get object (signal/image) list for current panel. Objects are sorted by group number and object index in group. Args: panel (str | None): Panel name. Defaults to None. Returns: list[str]: Object uuids """returnself.win.get_object_uuids(panel)@remote_calldefget_object_shapes(self,nb_id_title:int|str|None=None,panel:str|None=None,)->list:"""Get plot item shapes associated to object (signal/image). Args: nb_id_title: Object number, or object id, or object title. Defaults to None (current object). panel: Panel name. Defaults to None (current panel). Returns: List of plot item shapes """items=self.win.get_object_shapes(nb_id_title,panel)returnitems_to_json(items)@remote_calldefadd_annotations_from_items(self,items_json:str,refresh_plot:bool=True,panel:str|None=None)->None:"""Add object annotations (annotation plot items). Args: items_json (str): JSON string of annotation items refresh_plot (bool | None): refresh plot. Defaults to True. panel (str | None): panel name (valid values: "signal", "image"). If None, current panel is used. """items=json_to_items(items_json)ifitems:self.win.add_annotations_from_items(items,refresh_plot,panel)@remote_calldefadd_label_with_title(self,title:str|None=None,panel:str|None=None)->None:"""Add a label with object title on the associated plot Args: title (str | None): Label title. Defaults to None. If None, the title is the object title. panel (str | None): panel name (valid values: "signal", "image"). If None, current panel is used. """self.win.add_label_with_title(title,panel)@remote_calldefrun_macro(self,number_or_title:int|str|None=None)->None:"""Run macro. Args: number: Number of the macro (starting at 1). Defaults to None (run current macro, or does nothing if there is no macro). """self.SIG_RUN_MACRO.emit(number_or_title)@remote_calldefstop_macro(self,number_or_title:int|str|None=None)->None:"""Stop macro. Args: number: Number of the macro (starting at 1). Defaults to None (stop current macro, or does nothing if there is no macro). """self.SIG_STOP_MACRO.emit(number_or_title)@remote_calldefimport_macro_from_file(self,filename:str)->None:"""Import macro from file Args: filename: Filename. """self.SIG_IMPORT_MACRO_FROM_FILE.emit(filename)RemoteServer.check_remote_functions()# === Python 2.7 client side:## # See doc/remotecontrol_py27.py for an almost complete Python 2.7# # implementation of RemoteClient class## import io# from xmlrpclib import ServerProxy, Binary# import numpy as np# def array_to_binary(data):# """Convert NumPy array to XML-RPC Binary object, with shape and dtype"""# dbytes = io.BytesIO()# np.save(dbytes, data, allow_pickle=False)# return Binary(dbytes.getvalue())# s = ServerProxy("http://127.0.0.1:8000")# data = np.array([[3, 4, 5], [7, 8, 0]], dtype=np.uint16)# s.add_image("toto", array_to_binary(data))defget_cdl_xmlrpc_port()->str:"""Return DataLab current XML-RPC port Returns: XML-RPC port Raises: ConnectionRefusedError: DataLab has not yet been executed """# The following is valid only when using Python 3.8+ with DataLab# installed on the client side. In any other situation, please use the# ``get_cdl_xmlrpc_port`` function from doc/remotecontrol_py27.py.initialize()try:returnConf.main.rpc_server_port.get()exceptRuntimeErrorasexc:raiseConnectionRefusedError("DataLab has not yet been executed")fromexc
[docs]classRemoteClient(BaseProxy):"""Object representing a proxy/client to DataLab XML-RPC server. This object is used to call DataLab functions from a Python script. Examples: Here is a simple example of how to use RemoteClient in a Python script or in a Jupyter notebook: >>> from cdl.core.remote import RemoteClient >>> proxy = RemoteClient() >>> proxy.connect() Connecting to DataLab XML-RPC server...OK (port: 28867) >>> proxy.get_version() '1.0.0' >>> proxy.add_signal("toto", np.array([1., 2., 3.]), np.array([4., 5., -1.])) True >>> proxy.get_object_titles() ['toto'] >>> proxy["toto"] <cdl.core.model.signal.SignalObj at 0x7f7f1c0b4a90> >>> proxy[1] <cdl.core.model.signal.SignalObj at 0x7f7f1c0b4a90> >>> proxy[1].data array([1., 2., 3.]) """def__init__(self)->None:super().__init__()self.port:str=Noneself._cdl:ServerProxydef__connect_to_server(self,port:str|None=None)->None:"""Connect to DataLab XML-RPC server. Args: port (str | None): XML-RPC port to connect to. If not specified, the port is automatically retrieved from DataLab configuration. Raises: ConnectionRefusedError: DataLab is currently not running """ifportisNone:port=execenv.xmlrpcportifportisNone:port=get_cdl_xmlrpc_port()self.port=portself._cdl=ServerProxy(f"http://127.0.0.1:{port}",allow_none=True)try:version=self.get_version()exceptConnectionRefusedErrorasexc:raiseConnectionRefusedError("DataLab is currently not running")fromexc# If DataLab version is not compatible with this client, show a warning using# standard `warnings` module:minor_version=".".join(cdl.__version__.split(".")[:2])ifnotis_version_at_least(version,minor_version):warnings.warn(f"DataLab server version ({version}) may not be fully compatible with "f"this DataLab client version ({cdl.__version__}).\n"f"Please upgrade the server to {minor_version} or higher.")
[docs]defconnect(self,port:str|None=None,timeout:float|None=None,retries:int|None=None,)->None:"""Try to connect to DataLab XML-RPC server. Args: port (str | None): XML-RPC port to connect to. If not specified, the port is automatically retrieved from DataLab configuration. timeout (float | None): Timeout in seconds. Defaults to 5.0. retries (int | None): Number of retries. Defaults to 10. Raises: ConnectionRefusedError: Unable to connect to DataLab ValueError: Invalid timeout (must be >= 0.0) ValueError: Invalid number of retries (must be >= 1) """timeout=5.0iftimeoutisNoneelsetimeoutretries=10ifretriesisNoneelseretriesiftimeout<0.0:raiseValueError("timeout must be >= 0.0")ifretries<1:raiseValueError("retries must be >= 1")execenv.print("Connecting to DataLab XML-RPC server...",end="")for_indexinrange(retries):try:self.__connect_to_server(port=port)breakexceptConnectionRefusedError:time.sleep(timeout/retries)else:execenv.print("KO")raiseConnectionRefusedError("Unable to connect to DataLab")execenv.print(f"OK (port: {self.port})")
[docs]defdisconnect(self)->None:"""Disconnect from DataLab XML-RPC server."""# This is not mandatory with XML-RPC, but if we change protocol in the# future, it may be useful to have a disconnect method.self._cdl=None
[docs]defis_connected(self)->bool:"""Return True if connected to DataLab XML-RPC server."""ifself._cdlisnotNone:try:self.get_version()returnTrueexceptConnectionRefusedError:self._cdl=NonereturnFalse
[docs]defget_method_list(self)->list[str]:"""Return list of available methods."""returnself._cdl.system.listMethods()
# === Following methods should match the register functions in XML-RPC server
[docs]defadd_signal(self,title:str,xdata:np.ndarray,ydata:np.ndarray,xunit:str|None=None,yunit:str|None=None,xlabel:str|None=None,ylabel:str|None=None,)->bool:# pylint: disable=too-many-arguments"""Add signal data to DataLab. Args: title (str): Signal title xdata (numpy.ndarray): X data ydata (numpy.ndarray): Y data xunit (str | None): X unit. Defaults to None. yunit (str | None): Y unit. Defaults to None. xlabel (str | None): X label. Defaults to None. ylabel (str | None): Y label. Defaults to None. Returns: bool: True if signal was added successfully, False otherwise Raises: ValueError: Invalid xdata dtype ValueError: Invalid ydata dtype """obj=SignalObj()obj.set_xydata(xdata,ydata)obj.check_data()xbinary=array_to_rpcbinary(xdata)ybinary=array_to_rpcbinary(ydata)returnself._cdl.add_signal(title,xbinary,ybinary,xunit,yunit,xlabel,ylabel)
[docs]defadd_image(self,title:str,data:np.ndarray,xunit:str|None=None,yunit:str|None=None,zunit:str|None=None,xlabel:str|None=None,ylabel:str|None=None,zlabel:str|None=None,)->bool:# pylint: disable=too-many-arguments"""Add image data to DataLab. Args: title (str): Image title data (numpy.ndarray): Image data xunit (str | None): X unit. Defaults to None. yunit (str | None): Y unit. Defaults to None. zunit (str | None): Z unit. Defaults to None. xlabel (str | None): X label. Defaults to None. ylabel (str | None): Y label. Defaults to None. zlabel (str | None): Z label. Defaults to None. Returns: bool: True if image was added successfully, False otherwise Raises: ValueError: Invalid data dtype """obj=ImageObj()obj.data=dataobj.check_data()zbinary=array_to_rpcbinary(data)returnself._cdl.add_image(title,zbinary,xunit,yunit,zunit,xlabel,ylabel,zlabel)
[docs]defcalc(self,name:str,param:gds.DataSet|None=None)->None:"""Call compute function ``name`` in current panel's processor. Args: name: Compute function name param: Compute function parameter. Defaults to None. Raises: ValueError: unknown function """ifparamisNone:returnself._cdl.calc(name)returnself._cdl.calc(name,dataset_to_json(param))
[docs]defget_object(self,nb_id_title:int|str|None=None,panel:str|None=None,)->SignalObj|ImageObj:"""Get object (signal/image) from index. Args: nb_id_title: Object number, or object id, or object title. Defaults to None (current object). panel: Panel name. Defaults to None (current panel). Returns: Object Raises: KeyError: if object not found """param_data=self._cdl.get_object(nb_id_title,panel)ifparam_dataisNone:returnNonereturnjson_to_dataset(param_data)
[docs]defget_object_shapes(self,nb_id_title:int|str|None=None,panel:str|None=None,)->list:"""Get plot item shapes associated to object (signal/image). Args: nb_id_title: Object number, or object id, or object title. Defaults to None (current object). panel: Panel name. Defaults to None (current panel). Returns: List of plot item shapes """items_json=self._cdl.get_object_shapes(nb_id_title,panel)returnjson_to_items(items_json)
[docs]defadd_annotations_from_items(self,items:list,refresh_plot:bool=True,panel:str|None=None)->None:"""Add object annotations (annotation plot items). Args: items (list): annotation plot items refresh_plot (bool | None): refresh plot. Defaults to True. panel (str | None): panel name (valid values: "signal", "image"). If None, current panel is used. """items_json=items_to_json(items)ifitems_jsonisnotNone:self._cdl.add_annotations_from_items(items_json,refresh_plot,panel)
# ----- Proxy specific methods ------------------------------------------------# (not available symetrically in AbstractCDLControl)
[docs]defadd_object(self,obj:SignalObj|ImageObj)->None:"""Add object to DataLab. Args: obj (SignalObj | ImageObj): Signal or image object """# TODO [P1]: Would it be better to use directly a remote "add_object" method?# This would require to implement the add_object method in the# XML-RPC server. And first of all, to check if performance is# really better or not. This is equivalent to comparing the performance# between JSON transfer (using "json_to_dataset") and binary transfer# (using "array_to_rpcbinary") through XML-RPC.## If it is better, then here is what should be done:# - Implement add_object method in AbstractCDLProcessor instead of in# BaseProxy# - Implement add_object method in XML-RPC server# - Remove add_object method from BaseProxy# - Rewrite add_object method in LocalProxy and RemoteClient to use# the remote methodifisinstance(obj,SignalObj):self.add_signal(obj.title,obj.x,obj.y,obj.xunit,obj.yunit,obj.xlabel,obj.ylabel)elifisinstance(obj,ImageObj):self.add_image(obj.title,obj.data,obj.xunit,obj.yunit,obj.zunit,obj.xlabel,obj.ylabel,obj.zlabel,)