controller_proxy

plantimager.webui.controller_proxy

Controller Proxy Module

This module provides a bridge between the local controller device functionality and remote RPC communication, allowing applications to interact with controller hardware through a network connection.

Key Features

• Singleton pattern implementation ensuring only one controller proxy exists
• Transparent proxying of controller device methods to remote systems
• ZeroMQ-based RPC communication for reliable client-server interactions
• Consistent interface matching the local controller device API

Usage Examples

>>> import zmq
>>> from plantimager.webui.controller_proxy import RPCController
>>>
>>> # Initialize the controller proxy
>>> context = zmq.Context()
>>> controller = RPCController(context, "tcp://localhost:14567")
>>>
>>> # Access the singleton instance elsewhere in your code
>>> same_controller = RPCController.instance()
>>>
>>> # Use controller methods as if they were local
>>> controller.some_method()  # This will be executed on the remote system

RPCController

RPCController(context, url)

Bases: ControllerDevice, RPCClient

Proxy of controller and RPC server.

A singleton class that serves as a proxy between a controller device and an RPC server. Only one instance of this class can exist at a time, and it can be accessed through the instance class method.

Parameters:

Name	Type	Description	Default
context	Context	The ZeroMQ context to use for communication.	required
url	str	The URL to connect to for RPC communication.	required

Attributes:

Name	Type	Description
_instance	RPCController or None	Class variable that holds the single instance of the class.

Notes

This class implements the Singleton design pattern. The first time it is instantiated, it creates a new instance and stores it in the _instance class variable. Subsequent instantiations return the existing instance.

The class inherits from both ControllerDevice and RPCClient to provide controller functionality over an RPC connection.

See Also

plantimager.commons.controller_device.ControllerDevice : Base class for controller device functionality. plantimager.commons.RPC.RPCClient : Base class for RPC client functionality.

Examples:

>>> import zmq
>>> from plantimager.webui.controller_proxy import RPCController
>>> context = zmq.Context()
>>> controller = RPCController(context, "tcp://localhost:14567")
>>> # This returns the same instance
>>> controller2 = RPCController(context, "tcp://localhost:14567")
>>> controller is controller2
True
>>> # Get the singleton instance
>>> RPCController.instance()
RuntimeError: Controller proxy not initialized.

Initialize the RPCController.

Parameters:

Name	Type	Description	Default
context	Context	The ZeroMQ context to use for communication.	required
url	str	The URL to connect to for RPC communication.	required

Source code in plantimager/webui/controller_proxy.py

def __init__(self, context: zmq.Context, url: str):
    """Initialize the RPCController.

    Parameters
    ----------
    context : zmq.Context
        The ZeroMQ context to use for communication.
    url : str
        The URL to connect to for RPC communication.
    """
    RPCClient.__init__(self, context, url, timeout=120_000)
    self.__class__._instance = self

__new__

__new__(context, url)

Create a new instance or return the existing instance.

Parameters:

Name	Type	Description	Default
context	Context	The ZeroMQ context to use for communication.	required
url	str	The URL to connect to for RPC communication.	required

Returns:

Type	Description
RPCController	The singleton instance of this class.

Source code in plantimager/webui/controller_proxy.py

def __new__(cls, context: zmq.Context, url: str):
    """Create a new instance or return the existing instance.

    Parameters
    ----------
    context : zmq.Context
        The ZeroMQ context to use for communication.
    url : str
        The URL to connect to for RPC communication.

    Returns
    -------
    plantimager.commons.controller_device.RPCController
        The singleton instance of this class.
    """
    if cls._instance is None:
        instance = super(RPCController, cls).__new__(cls)
        return instance
    return cls._instance

camera_names abstractmethod

camera_names()

Return the list of camera names.

Source code in plantimager/commons/controller_device.py

@RPCProperty(notify=cameraNamesChanged)
@abstractmethod
def camera_names(self) -> list[str]:
    """Return the list of camera names."""
    pass

execute

execute(method_name, params)

Execute a remote method on the RPC server.

Parameters:

Name	Type	Description	Default
method_name	str	The name of the method to execute.	required
params	dict	A dictionary containing the args and kwargs for the method.	required

Returns:

Type	Description
tuple	A 2-tuple (success, result). If success is True, result contains the return value of the method. If False, result is a tuple containing the (error_message, traceback).

Raises:

Type	Description
TimeoutError	If the server does not respond within the configured timeout.

Source code in plantimager/commons/RPC.py

def execute(self, method_name: str, params: dict) -> tuple[bool, object]:
    """
    Execute a remote method on the RPC server.

    Parameters
    ----------
    method_name : str
        The name of the method to execute.
    params : dict
        A dictionary containing the `args` and `kwargs` for the method.

    Returns
    -------
    tuple
        A 2-tuple `(success, result)`. If `success` is True, `result`
        contains the return value of the method. If False, `result`
        is a tuple containing the `(error_message, traceback)`.

    Raises
    ------
    TimeoutError
        If the server does not respond within the configured timeout.
    """
    package = {
        "event": RPCEvents.METHOD_CALL,
        "method": method_name,
        "params": params,
    }
    if self.socket.poll(timeout=1000, flags=zmq.POLLOUT) == 0:
        logger.error(f"Proxy of {self._interface} at {self.url} did not respond")
        raise TimeoutError(f"Proxy of {self._interface} at {self.url} did not respond")
    logger.debug(f"Executing {package}")
    self.socket.send_json(package, flags=zmq.NOBLOCK)

    if method_name in self._json_methods:
        if self.socket.poll(timeout=self._json_methods[method_name], flags=zmq.POLLIN) == 0:
            logger.error(f"Proxy of {self._interface} at {self.url} did not respond")
            raise TimeoutError(f"Proxy of {self._interface} at {self.url} did not respond")
        reply =  self.socket.recv_json()
        if reply["success"]:
            return True, reply["result"]
        else:
            return False, (reply["error"], reply["traceback"])
    elif method_name in self._buffer_methods:
        if self.socket.poll(timeout=self._buffer_methods[method_name], flags=zmq.POLLIN) == 0:
            logger.error(f"Proxy of {self._interface} at {self.url} did not respond")
            raise TimeoutError(f"Proxy of {self._interface} at {self.url} did not respond")
        reply_frames: list[zmq.Frame] = self.socket.recv_multipart(copy=False)
        buffer_info = json.loads(reply_frames[0].bytes)
        if "error" in buffer_info:
            return False, (buffer_info["error"], buffer_info["traceback"])
        else:
            return True, (reply_frames[1].buffer, buffer_info)
    return False, (Warning(f"Unknown method {method_name}"), "")

instance classmethod

instance()

Get the singleton instance of the RPCController.

Returns:

Type	Description
RPCController	The singleton instance of this class.

Raises:

Type	Description
RuntimeError	If the controller proxy has not been initialized yet.

Source code in plantimager/webui/controller_proxy.py

@classmethod
def instance(cls) -> "RPCController":
    """Get the singleton instance of the RPCController.

    Returns
    -------
    plantimager.commons.controller_device.RPCController
        The singleton instance of this class.

    Raises
    ------
    RuntimeError
        If the controller proxy has not been initialized yet.
    """
    if cls._instance is None:
        raise RuntimeError("Controller proxy not initialized.")
    return cls._instance

max_progress abstractmethod

max_progress()

Return the maximum progress of the scan.

Source code in plantimager/commons/controller_device.py

@RPCProperty(notify=maxProgressChanged)
@abstractmethod
def max_progress(self) -> int:
    """Return the maximum progress of the scan."""
    pass

progress abstractmethod

progress()

Return the current progress of the scan.

Source code in plantimager/commons/controller_device.py

@RPCProperty(notify=progressChanged)
@abstractmethod
def progress(self) -> int:
    """Return the current progress of the scan."""
    pass

ready_to_scan abstractmethod

ready_to_scan()

Return whether the controller is ready to start a scan.

Source code in plantimager/commons/controller_device.py

@RPCProperty(notify=readyToScanChanged)
@abstractmethod
def ready_to_scan(self) -> bool:
    """Return whether the controller is ready to start a scan."""
    pass

register_interface classmethod

register_interface(interface)

Class decorator to bind an interface to an RPCClient subclass.

This decorator inspects the provided interface and dynamically proxies its methods and properties so that calls are forwarded over the network to the RPC server.

Parameters:

Name	Type	Description	Default
interface	type	The interface class defining the methods and properties to proxy.	required

Returns:

Type	Description
callable	A class decorator that applies the proxy logic.

Raises:

Type	Description
RuntimeError	If the decorated class does not inherit from both interface and RPCClient.

Source code in plantimager/commons/RPC.py

@classmethod
def register_interface(cls, interface: type):
    """
    Class decorator to bind an interface to an RPCClient subclass.

    This decorator inspects the provided `interface` and dynamically
    proxies its methods and properties so that calls are forwarded
    over the network to the RPC server.

    Parameters
    ----------
    interface : type
        The interface class defining the methods and properties to proxy.

    Returns
    -------
    callable
        A class decorator that applies the proxy logic.

    Raises
    ------
    RuntimeError
        If the decorated class does not inherit from both `interface`
        and `RPCClient`.
    """
    def _decorator(target_cls: type):
        if interface not in target_cls.__bases__ or cls not in target_cls.__bases__:
            raise RuntimeError(f"{target_cls} must inherit from {interface} and {cls}.")
        for key, val in interface.__dict__.items():
            if inspect.isfunction(val) and not (key.startswith("__") and key.endswith("__")):
                logger.debug(f"registering method {key} in {target_cls}")
                func = decorate(val, cls._method_proxy)
                func.__isabstractmethod__ = False  # Counts as en actual implementation
                setattr(target_cls, key, func)
            elif isinstance(val, RPCSignal):
                pass # nothing to do at this stage
            elif isinstance(val, RPCProperty):
                fget = wraps(val.fget)(partial(cls._property_getter_proxy, property_name=key))
                fset = wraps(val.fset)(partial(cls._property_setter_proxy, property_name=key))
                prop = RPCProperty(fget=fget, fset=fset, fdel=val.fdel, doc=val.__doc__, notify=val._notifier)
                setattr(target_cls, key, prop)
        # abstract methods have been implemented
        target_cls.__abstractmethods__ = frozenset()
        target_cls._interface = interface.__name__
        return target_cls
    return _decorator

run_scan abstractmethod

run_scan()

Start the scan.

Source code in plantimager/commons/controller_device.py

@abstractmethod
def run_scan(self):
    """Start the scan."""
    pass

set_api_token abstractmethod

set_api_token(token)

Set the session token to use for authenticated requests.

Source code in plantimager/commons/controller_device.py

@abstractmethod
def set_api_token(self, token: str):
    """Set the session token to use for authenticated requests."""
    pass

set_config abstractmethod

set_config(config)

Send a configuration dictionary to the controller.

Source code in plantimager/commons/controller_device.py

@abstractmethod
def set_config(self, config: dict):
    """Send a configuration dictionary to the controller."""
    pass

set_dataset_name abstractmethod

set_dataset_name(name)

Set the name of the dataset to be created.

Source code in plantimager/commons/controller_device.py

@abstractmethod
def set_dataset_name(self, name: str):
    """Set the name of the dataset to be created."""
    pass

set_db_url abstractmethod

set_db_url(url)

Set the database URL for the controller.

Source code in plantimager/commons/controller_device.py

@abstractmethod
def set_db_url(self, url: str):
    """Set the database URL for the controller."""
    pass

stop_server

stop_server()

Request the connected RPC server to shut down.

Sends a non-blocking STOP_SERVER event to the remote server. If the server responds, it logs the reply.

Source code in plantimager/commons/RPC.py

def stop_server(self):
    """
    Request the connected RPC server to shut down.

    Sends a non-blocking `STOP_SERVER` event to the remote server.
    If the server responds, it logs the reply.
    """
    logger.info(f"Stopping server {self.url}")
    if self.socket.poll(timeout=1000, flags=zmq.POLLOUT) == 0:
        logger.info(f"Server {self.url} could not be joined (might already be dead)")
        return
    self.socket.send_json({
        "event": RPCEvents.STOP_SERVER
    }, flags=zmq.NOBLOCK)
    if self.socket.poll(timeout=1000, flags=zmq.POLLIN) == 0:
        logger.info(f"Server {self.url} did not respond (might already be dead)")
        return
    reply = self.socket.recv_json()
    logger.debug(f"Got stop reply {reply}")