rpc_controller

plantimager.controller.scanner.rpc_controller

RPC Controller for Plant Imaging Systems.

A module that implements a Remote Procedure Call (RPC) controller for handling client-server communication through JSON-RPC protocol. This enables remote execution of scanner control functions with robust error handling and input validation.

Key Features

• Executes RPC calls using dispatcher to route methods to appropriate handlers
• Validates input parameters against method specifications
• Handles and returns standardized error responses
• Supports both regular and notification RPC calls
• Provides properties for monitoring scan progress
• Exposes scanner configuration and control methods remotely
• Preserves call context for security and tracing purposes

Usage Examples

>>> import zmq
>>> from plantimager.controller.scanner.scanner import Scanner
>>> from plantimager.controller.scanner.rpc_controller import RPCControllerServer
>>> # Create a scanner instance
>>> scanner = Scanner()
>>> # Create a ZeroMQ context
>>> context = zmq.Context()
>>> # Create an RPC server for the scanner
>>> server = RPCControllerServer(context, "tcp://*:5555", scanner)
>>> # Start the server
>>> server.start()

RPCControllerServer

RPCControllerServer(context, url, scanner)

Bases: ControllerDevice, RPCServer

An RPC server controlling a scanner device.

This class combines the functionality of ControllerDevice and RPCServer to expose scanner control capabilities over RPC. It registers methods for configuring and running scans and properties for monitoring scan progress.

Parameters:

Name	Type	Description	Default
context	Context	The context object for the RPC server.	required
url	str	The URL where the RPC server will be available.	required
scanner	Scanner	The scanner device to be controlled via RPC.	required

Attributes:

Name	Type	Description
scanner	Scanner	The scanner device being controlled.
progress	int	The current progress value of the scanner.
max_progress	int	The maximum progress value of the scanner.

Notes

This class connects the scanner's progress signals to its own signals to propagate progress updates to connected clients via RPC properties.

See Also

plantimager.commons.controller_device.ControllerDevice : Base class for controller functionality. plantimager.commons.RPC.RPCServer : Base class for RPC server functionality. plantimager.controller.scanner.scanner.Scanner : The scanner device being controlled.

Initialize the RPC controller server.

Parameters:

Name	Type	Description	Default
context	Context	The context object for the RPC server.	required
url	str	The URL where the RPC server will be available.	required
scanner	Scanner	The scanner device to be controlled via RPC.	required

Source code in plantimager/controller/scanner/rpc_controller.py

def __init__(self, context, url, scanner: Scanner):
    """
    Initialize the RPC controller server.

    Parameters
    ----------
    context : zmq.Context
        The context object for the RPC server.
    url : str
        The URL where the RPC server will be available.
    scanner : plantimager.controller.scanner.scanner.Scanner
        The scanner device to be controlled via RPC.
    """
    RPCServer.__init__(self, context, url)
    self.scanner = scanner
    self.scanner.progressChanged.connect(self.progressChanged.emit)
    self.scanner.maxProgressChanged.connect(self.maxProgressChanged.emit)
    self.scanner.readyToScanChanged.connect(self.readyToScanChanged.emit)
    self.scanner.cameraNamesChanged.connect(self.cameraNamesChanged.emit)

camera_names

camera_names()

Return the list of camera names.

Source code in plantimager/controller/scanner/rpc_controller.py

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

handle_scanner_changed

handle_scanner_changed(scanner)

Update the controlled scanner and synchronize progress values.

This method is called when the scanner device changes. It updates the reference to the scanner and emits signals to update progress values.

Parameters:

Name	Type	Description	Default
scanner	Scanner	The new scanner device to be controlled.	required

Source code in plantimager/controller/scanner/rpc_controller.py

def handle_scanner_changed(self, scanner):
    """Update the controlled scanner and synchronize progress values.

    This method is called when the scanner device changes. It updates the
    reference to the scanner and emits signals to update progress values.

    Parameters
    ----------
    scanner : plantimager.controller.scanner.scanner.Scanner
        The new scanner device to be controlled.
    """
    if self.scanner is not scanner:
        self.scanner = scanner
        self.scanner.progressChanged.connect(self.progressChanged.emit)
        self.scanner.maxProgressChanged.connect(self.maxProgressChanged.emit)
        self.scanner.readyToScanChanged.connect(self.readyToScanChanged.emit)
        self.scanner.cameraNamesChanged.connect(self.cameraNamesChanged.emit)
        self.progressChanged.emit(self.scanner.progress)
        self.maxProgressChanged.emit(self.scanner.max_progress)
        self.readyToScanChanged.emit(self.scanner.ready_to_scan)
        self.cameraNamesChanged.emit(self.scanner.camera_names)

max_progress

max_progress()

Get the maximum progress value of the scanner.

Returns:

Type	Description
int	The maximum progress value.

Source code in plantimager/controller/scanner/rpc_controller.py

@RPCProperty(notify=ControllerDevice.maxProgressChanged)
def max_progress(self):
    """Get the maximum progress value of the scanner.

    Returns
    -------
    int
        The maximum progress value.
    """
    return self.scanner.max_progress

progress

progress()

Get the current progress value of the scanner.

Returns:

Type	Description
int	The current progress value.

Source code in plantimager/controller/scanner/rpc_controller.py

@RPCProperty(notify=ControllerDevice.progressChanged)
def progress(self):
    """Get the current progress value of the scanner.

    Returns
    -------
    int
        The current progress value.
    """
    return self.scanner.progress

ready_to_scan

ready_to_scan()

Get whether the scanner is ready to start a scan.

Returns:

Type	Description
bool	True if the scanner is ready to start a scan, False otherwise.

Source code in plantimager/controller/scanner/rpc_controller.py

@RPCProperty(notify=ControllerDevice.readyToScanChanged)
def ready_to_scan(self) -> bool:
    """Get whether the scanner is ready to start a scan.

    Returns
    -------
    bool
        True if the scanner is ready to start a scan, False otherwise.
    """
    return self.scanner.ready_to_scan

register_method_buffer staticmethod

register_method_buffer(timeout=10000)

Registers this method as remote callable procedure which will transmit its output as a buffer-like object as well as a buffer_info dictionary.

method may take any input which will be serialized via json and must output a 2-tuple: (memoryview or bytes, dict).

Parameters:

Name	Type	Description	Default
timeout	int | None	Specifies how much time in ms a client is expected to wait for the method to finish. If None, waits indefinitely.	10000

Returns:

Name	Type	Description
decorator	Callable[Callable[..., tuple[memoryview | bytes, dict]], Callable[..., tuple[memoryview | bytes, dict]]]

Source code in plantimager/commons/RPC.py

@staticmethod
def register_method_buffer(timeout: int | None = 10000):
    """
    Registers this method as remote callable procedure which will transmit its output as a buffer-like object
    as well as a buffer_info dictionary.

    `method` may take any input which will be serialized via json and must output a 2-tuple:
    (memoryview or bytes, dict).


    Parameters
    ----------
    timeout: int | None
        Specifies how much time in ms a client is expected to wait for the method to finish. If None, waits indefinitely.

    Returns
    -------
    decorator: Callable[Callable[..., tuple[memoryview|bytes, dict]], Callable[..., tuple[memoryview|bytes, dict]]]

    """
    if inspect.isfunction(timeout):
        timeout._is_buffer_method = True
        timeout._timeout = 10000
        return timeout

    def _decorator(method: Callable[..., tuple[memoryview | bytes, dict]]):
        method._is_buffer_method = True
        method._timeout = timeout
        return method

    return _decorator

register_method_json staticmethod

register_method_json(timeout=10000)

Registers this method as remote callable procedure which will transmit its output via json.

It is advised to only send basic types and containers as output (int, float, str, bool, list, tuple, dict, ...) Arguments are also serialized via json.

Parameters:

Name	Type	Description	Default
timeout	int | None	Specifies how much time in ms a client is expected to wait for the method to finish. If None, waits indefinitely.	10000

Returns:

Type	Description
Callable[Callable[..., Any], Callable[..., Any]]

Source code in plantimager/commons/RPC.py

@staticmethod
def register_method_json(timeout: int | None = 10000):
    """
    Registers this method as remote callable procedure which will transmit its output via json.

    It is advised to only send basic types and containers as output (int, float, str, bool, list, tuple, dict, ...)
    Arguments are also serialized via json.

    Parameters
    ----------
    timeout: int | None
        Specifies how much time in ms a client is expected to wait for the method to finish. If None, waits indefinitely.

    Returns
    -------
    Callable[Callable[..., Any], Callable[..., Any]]

    """
    if inspect.isfunction(timeout):
        timeout._is_json_method = True
        timeout._timeout = 10000
        return timeout

    def _decorator(method: Callable[..., Any]):
        method._is_json_method = True
        method._timeout = timeout
        return method

    return _decorator

register_to_registry

register_to_registry(type_, name, registry_url, overwrite=True)

Register this RPCServer to the registry at registry_address as a device of type type_ and name name.

Note: The name not be accepted as is and may be modified by the registry to avoid duplicate. This method returns the accepted name of this device.

Parameters:

Name	Type	Description	Default
type_	str	Name of the device type.	required
name	str	Proposed name of the device.	required
registry_url	str	Url of the device registry. Must have the form "tcp://:" if the registry uses tcp	required
overwrite		Wether or not this device should take preference for the use of this name and overwrite other conflicting devices.	True

Returns:

Type	Description
str	Name of this device as accepted by the registry.

Source code in plantimager/commons/RPC.py

def register_to_registry(self, type_: str, name: str, registry_url: str, overwrite=True) -> str:
    """
    Register this RPCServer to the registry at `registry_address` as a device of type `type_` and name `name`.

    Note: The name not be accepted as is and may be modified by the registry to avoid duplicate. This method
    returns the accepted name of this device.

    Parameters
    ----------
    type_: str
        Name of the device type.
    name: str
        Proposed name of the device.
    registry_url: str
        Url of the device registry. Must have the form "tcp://<ip>:<port>" if the registry uses tcp
    overwrite: bool, optional
        Wether or not this device should take preference for the use of this name and overwrite other
        conflicting devices.

    Returns
    -------
    str
        Name of this device as accepted by the registry.
    """
    logger.debug(f"Register device {name} of type {type_} to {registry_url}")
    self._type = type_
    self.name, self.uuid = register_device(
        self.context, type_,
        f"{self.url}:{self.port}",
        name, registry_url,
        overwrite=overwrite,
    )
    if not self.name:
        logger.warning(f"Failed to register device {name} of type {type_} as {registry_url}")
    self.registry_addr = registry_url if self.name else ""

    # Update cleanup state so finalizer knows what to unregister
    self._cleanup_state["uuid"] = self.uuid
    self._cleanup_state["registry_addr"] = self.registry_addr

    logger.info(f"Successfully registered device {name} of type {type_} to {registry_url}")

    return self.name

run_scan

run_scan()

Start a scanning operation with the current configuration.

Source code in plantimager/controller/scanner/rpc_controller.py

@RPCServer.register_method_json(timeout=None)
def run_scan(self):
    """Start a scanning operation with the current configuration."""
    self.scanner.scan()

serve_forever

serve_forever()

Serve the RPC server indefinitely, handling incoming requests until a stop signal is received.

The method enters a loop that repeatedly notifies the watchdog, checks the server's liveness, waits for a request, and dispatches the request based on its event field. Supported events include peer discovery, inventory retrieval, signal handling initialization, method invocation, property access, and server shutdown. Upon receiving a STOP_SERVER event the loop terminates, sockets are closed, and a log entry is written.

If registered to the registry and the check_alive fails, exits the loop

Returns:

Type	Description
None	The server runs until it is stopped; no value is returned.

Notes

• The private attribute _stop controls the loop termination. It is set to True only when a STOP_SERVER request is processed.
• _socket and _signal_socket are closed during cleanup; if either attribute is None the corresponding close call is skipped.
• notify_watchdog and _alive_check are called on every iteration to maintain server health monitoring.
• The method assumes that _wait_for_request returns a mapping with an event key; a falsy return value causes the loop to continue without processing.

See Also

RPCEvents : Enum defining the possible request events. _handle_find_peer_address, _handle_get_inventory, _handle_init_signal_handling, _handle_method_call, _handle_property_get, _handle_property_set : Private helper methods that implement the handling logic for each event type.

Source code in plantimager/commons/RPC.py

def serve_forever(self):
    """
    Serve the RPC server indefinitely, handling incoming requests until a stop signal is
    received.

    The method enters a loop that repeatedly notifies the watchdog, checks the server's
    liveness, waits for a request, and dispatches the request based on its ``event`` field.
    Supported events include peer discovery, inventory retrieval, signal handling
    initialization, method invocation, property access, and server shutdown. Upon receiving
    a ``STOP_SERVER`` event the loop terminates, sockets are closed, and a log entry is
    written.

    If registered to the registry and the check_alive fails, exits the loop

    Returns
    -------
    None
        The server runs until it is stopped; no value is returned.

    Notes
    -----
    * The private attribute ``_stop`` controls the loop termination. It is set to
      ``True`` only when a ``STOP_SERVER`` request is processed.
    * ``_socket`` and ``_signal_socket`` are closed during cleanup; if either attribute is
      ``None`` the corresponding ``close`` call is skipped.
    * ``notify_watchdog`` and ``_alive_check`` are called on every iteration to maintain
      server health monitoring.
    * The method assumes that ``_wait_for_request`` returns a mapping with an ``event``
      key; a falsy return value causes the loop to continue without processing.

    See Also
    --------
    RPCEvents : Enum defining the possible request events.
    _handle_find_peer_address, _handle_get_inventory, _handle_init_signal_handling,
    _handle_method_call, _handle_property_get, _handle_property_set :
        Private helper methods that implement the handling logic for each event type.
    """
    if self._dead:
        logger.error("RPCServer is in dead state. A new instance must be created.")
        raise RuntimeError("RPCServer is in dead state. A new instance must be created.")
    self._stop = False
    while not self._stop:
        notify_watchdog()
        if not self._alive_check():
            break
        request = self._wait_for_request()
        if not request:
            continue
        try:
            logger.debug(f"Received request: {request['event']}")
            t0 = time.monotonic()
            match request["event"]:
                case RPCEvents.GET_INVENTORY:
                    reply = self._handle_get_inventory()
                    self._send_reply(reply)
                case RPCEvents.INIT_SIGNALS_HANDLING:
                    reply = self._handle_init_signal_handling(request)
                    self._send_reply(reply)
                case RPCEvents.METHOD_CALL:
                    reply, use_multipart = self._handle_method_call(request)
                    self._send_reply(reply, use_multipart)
                case RPCEvents.PROPERTY_GET:
                    reply = self._handle_property_get(request)
                    self._send_reply(reply)
                case RPCEvents.PROPERTY_SET:
                    reply = self._handle_property_set(request)
                    self._send_reply(reply)
                case RPCEvents.STOP_SERVER:
                    self._socket.send_json({"success": True})
                    self._stop = True
            logger.debug(f"Event {request['event']} treated in {time.monotonic() - t0}s")
        except Exception as exc:
            logger.error(f"Unexpected error while handling request: {exc}")
            # If we have a request, we can at least try to answer with a generic error.
            if isinstance(request, dict) and "event" in request:
                self._send_reply(self._make_error_reply(exc, "serve_forever"))
            break  # abort the loop - we are in an inconsistent state

    # cleanup
    if self._socket:
        self._socket.close()
    if self._signal_socket:
        self._signal_socket.close()
    self._dead = True
    logger.info("Server stopped")

set_api_token

set_api_token(token)

Set the api token to use for authenticated requests.

Parameters:

Name	Type	Description	Default
token	str	The session token to use for authenticated requests.	required

Source code in plantimager/controller/scanner/rpc_controller.py

@RPCServer.register_method_json
def set_api_token(self, token: str):
    """Set the api token to use for authenticated requests.

    Parameters
    ----------
    token : str
        The session token to use for authenticated requests.
    """
    self.scanner.set_api_token(token)

set_config

set_config(config)

Configure the scanner with the provided configuration.

Parameters:

Name	Type	Description	Default
config	dict	Configuration dictionary with scanner settings.	required

Source code in plantimager/controller/scanner/rpc_controller.py

@RPCServer.register_method_json
def set_config(self, config):
    """Configure the scanner with the provided configuration.

    Parameters
    ----------
    config : dict
        Configuration dictionary with scanner settings.
    """
    self.scanner.configure_scan(config)

set_dataset_name

set_dataset_name(name)

Set the name of the dataset to be created.

Parameters:

Name	Type	Description	Default
name	str	The name of the dataset to be created.	required

Source code in plantimager/controller/scanner/rpc_controller.py

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

    Parameters
    ----------
    name : str
        The name of the dataset to be created.
    """
    self.scanner.set_scan_id(name)

set_db_url

set_db_url(url)

Set the database URL for the scanner.

Parameters:

Name	Type	Description	Default
url	str	The URL, including protocol and port, of the database to connect to.	required

Source code in plantimager/controller/scanner/rpc_controller.py

@RPCServer.register_method_json
def set_db_url(self, url: str):
    """Set the database URL for the scanner.

    Parameters
    ----------
    url : str
        The URL, including protocol and port, of the database to connect to.
    """
    self.scanner.set_db_url(url)

stop_server

stop_server()

Stop the server and unregister the device if it has been registered.

The method sets the internal stop flag, optionally calls :func:unregister_device to remove the device from a remote registry, clears the identifying attributes (name, uuid, registry_addr), and resets the corresponding entries in _cleanup_state to None. This prevents a second unregister attempt during cleanup.

Raises:

Type	Description
Exception	Propagates any exception raised by :func:unregister_device, e.g. network errors or authentication failures.

Notes

• The device is only unregistered when all three attributes self.name, self.registry_addr and self.uuid evaluate to True. If any of them is falsy, the function simply sets the stop flag and returns.
• self._cleanup_state is updated after a successful unregister to avoid duplicate cleanup actions later in the object's lifecycle.

See Also

unregister_device : Function that removes a device from the registry.

Source code in plantimager/commons/RPC.py

def stop_server(self):
    """
    Stop the server and unregister the device if it has been registered.

    The method sets the internal stop flag, optionally calls
    :func:`unregister_device` to remove the device from a remote registry,
    clears the identifying attributes (``name``, ``uuid``, ``registry_addr``),
    and resets the corresponding entries in ``_cleanup_state`` to ``None``.
    This prevents a second unregister attempt during cleanup.

    Raises
    ------
    Exception
        Propagates any exception raised by :func:`unregister_device`, e.g.
        network errors or authentication failures.

    Notes
    -----
    * The device is only unregistered when all three attributes
      ``self.name``, ``self.registry_addr`` and ``self.uuid`` evaluate to
      ``True``. If any of them is falsy, the function simply sets the
      stop flag and returns.
    * ``self._cleanup_state`` is updated after a successful unregister to
      avoid duplicate cleanup actions later in the object's lifecycle.

    See Also
    --------
    unregister_device : Function that removes a device from the registry.
    """
    self._stop = True
    if self.name and self.registry_addr and self.uuid:
        unregister_device(self.context, self.uuid, self.registry_addr)
        self.name = ""
        self.uuid = ""
        self.registry_addr = ""

        # Clear cleanup state to avoid double unregister
        self._cleanup_state["uuid"] = None
        self._cleanup_state["registry_addr"] = None

        logger.info("Device unregistered successfully")