Source code for cdl.proxy

# Copyright (c) DataLab Platform Developers, BSD 3-Clause license, see LICENSE file.

"""
Proxy objects (:mod:`cdl.proxy`)
--------------------------------

The :mod:`cdl.proxy` module provides a way to access DataLab features from a proxy
class.

The list of compute methods accessible from the proxy objects is available in the
:ref:`processor_methods` section.

Remote proxy
^^^^^^^^^^^^

The remote proxy is used when DataLab is started from a different process than the
proxy. In this case, the proxy connects to DataLab XML-RPC server.

.. autoclass:: RemoteProxy
    :members:
    :inherited-members:

Local proxy
^^^^^^^^^^^

The local proxy is used when DataLab is started from the same process as the proxy.
In this case, the proxy is directly connected to DataLab main window instance. The
typical use case is high-level scripting.

.. autoclass:: LocalProxy
    :members:
    :inherited-members:

Proxy context manager
^^^^^^^^^^^^^^^^^^^^^

The proxy context manager is a convenient way to handle proxy creation and
destruction. It is used as follows:

.. code-block:: python

        with proxy_context("local") as proxy:
            proxy.add_signal(...)

The proxy type can be "local" or "remote". For remote proxy, the port can be
specified as "remote:port".

.. note:: The proxy context manager allows to use the proxy in various contexts
    (Python script, Jupyter notebook, etc.). It also allows to switch seamlessly
    between local and remote proxy, keeping the same code inside the context.

.. autofunction:: proxy_context

.. _processor_methods:

Calling processor methods using proxy objects
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

All the proxy objects provide access to the DataLab computing methods exposed by
the processor classes:

- :class:`cdl.core.gui.processor.signal.SignalProcessor`
- :class:`cdl.core.gui.processor.image.ImageProcessor`

.. seealso::

    The list of processor methods is available in tables below.

There are two ways to call a processor method:

1. Using the :meth:`calc` method of the proxy object:

.. code-block:: python

    # Call a method without parameter
    proxy.calc("compute_average")

    # This is equivalent to:
    proxy.calc("average")

    # Call a method with parameters
    p = cdl.param.MovingAverageParam.create(n=30)
    proxy.calc("compute_moving_average", p)

2. Directly calling the processor method from the proxy object:

.. code-block:: python

    # Call a method without parameter
    proxy.compute_average()

    # Call a method with parameters
    p = cdl.param.MovingAverageParam.create(n=30)
    proxy.compute_moving_average(p)

.. warning::

    The `compute_{name}` methods are not statically defined in the proxy classes (and
    not even dynamically). They are nevertheless available through the proxy objects
    thanks to the magic method :meth:`__getattr__` which forwards the call to the
    :meth:`calc` method. However, this means that the methods are not listed in the
    proxy classes documentation, and they are not available in the auto-completion
    feature of your IDE.

Number of compute methods
*************************

.. csv-table:: Number of compute methods
   :file: ../doc/processor_methods_nb.csv
   :header: Signal, Image, Total

Signal processing
*****************

The following table lists the signal processor methods - it is automatically
generated from the source code:

.. csv-table:: Signal processor methods
   :file: ../doc/processor_methods_signal.csv
   :header: Compute method, Description
   :widths: 40, 60

Image processing
****************

The following table lists the image processor methods - it is automatically
generated from the source code:

.. csv-table:: Image processor methods
    :file: ../doc/processor_methods_image.csv
    :header: Compute method, Description
    :widths: 40, 60
"""

from __future__ import annotations

from collections.abc import Generator
from contextlib import contextmanager

import guidata.dataset as gds
import numpy as np

from cdl.core.baseproxy import BaseProxy
from cdl.core.remote import RemoteClient
from cdl.obj import ImageObj, SignalObj
from cdl.utils import qthelpers as qth


[docs] class RemoteProxy(RemoteClient): """DataLab remote proxy class. This class provides access to DataLab features from a proxy class. This is the remote version of proxy, which is used when DataLab is started from a different process than the proxy. Args: autoconnect (bool): Automatically connect to DataLab XML-RPC server. Raises: ConnectionRefusedError: Unable to connect to DataLab ValueError: Invalid timeout (must be >= 0.0) ValueError: Invalid number of retries (must be >= 1) .. note:: The proxy object also allows to access DataLab computing methods exposed by the processor classes (see :ref:`processor_methods`). Examples: Here is a simple example of how to use RemoteProxy in a Python script or in a Jupyter notebook: >>> from cdl.proxy import RemoteProxy >>> proxy = RemoteProxy() 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"] # from title <cdl.core.model.signal.SignalObj at 0x7f7f1c0b4a90> >>> proxy[1] # from number <cdl.core.model.signal.SignalObj at 0x7f7f1c0b4a90> >>> proxy[1].data array([1., 2., 3.]) >>> proxy.set_current_panel("image") """ def __init__(self, autoconnect: bool = True) -> None: super().__init__() if autoconnect: self.connect()
[docs] class LocalProxy(BaseProxy): """DataLab local proxy class. This class provides access to DataLab features from a proxy class. This is the local version of proxy, which is used when DataLab is started from the same process as the proxy. Args: cdl (CDLMainWindow): CDLMainWindow instance. .. note:: The proxy object also allows to access DataLab computing methods exposed by the processor classes (see :ref:`processor_methods`). """
[docs] def add_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 """ return self._cdl.add_signal(title, xdata, ydata, xunit, yunit, xlabel, ylabel)
[docs] def add_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 """ return self._cdl.add_image( title, data, xunit, yunit, zunit, xlabel, ylabel, zlabel )
[docs] def calc(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 """ return self._cdl.calc(name, param)
[docs] def get_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 """ return self._cdl.get_object(nb_id_title, panel)
[docs] def get_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 """ return self._cdl.get_object_shapes(nb_id_title, panel)
[docs] def add_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. """ self._cdl.add_annotations_from_items(items, refresh_plot, panel)
# ----- Proxy specific methods ------------------------------------------------ # (not available symetrically in AbstractCDLControl)
[docs] def add_object(self, obj: SignalObj | ImageObj) -> None: """Add object to DataLab. Args: obj (SignalObj | ImageObj): Signal or image object """ self._cdl.add_object(obj)
[docs] @contextmanager def proxy_context(what: str) -> Generator[LocalProxy | RemoteProxy, None, None]: """Context manager handling CDL proxy creation and destruction. Args: what (str): proxy type ("local" or "remote") For remote proxy, the port can be specified as "remote:port" Yields: Generator[LocalProxy | RemoteProxy, None, None]: proxy LocalProxy if what == "local" RemoteProxy if what == "remote" or "remote:port" Example: with proxy_context("local") as proxy: proxy.add_signal(...) """ assert what == "local" or what.startswith("remote"), "Invalid proxy type" port = None if ":" in what: port = int(what.split(":")[1].strip()) if what == "local": # pylint: disable=import-outside-toplevel, cyclic-import from cdl.core.gui.main import CDLMainWindow with qth.cdl_app_context(exec_loop=True): try: win = CDLMainWindow() proxy = LocalProxy(win) win.show() yield proxy finally: pass else: try: proxy = RemoteProxy(autoconnect=False) proxy.connect(port) yield proxy finally: proxy.disconnect()