rpc_controller

plantimager.controller.scanner.rpc_controller Link

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 Link

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.

camera_names Link

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 Link

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 Link

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 Link

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 Link

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` Link

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` Link

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:

Name	Type	Description
`decorator`	`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
    -------
    decorator: 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 Link

register_to_registry(type_, name, registry_url)

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

Returns:

Name	Type	Description
`accepted_name`	`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) -> 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

    Returns
    -------
    accepted_name: str
        Name of this device as accepted by the registry.

    """
    logger.debug(f"Register device {name} of type {type_} to {registry_url}")
    self.name, self.uuid = register_device(
        self.context, type_,
        f"{self.url}:{self.port}",
        name, registry_url,
        overwrite=True,
    )
    self.registry_addr = registry_url if self.name else ""
    return self.name

run_scan Link

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 Link

serve_forever()

Starts serving requests for this RPCServer.

Source code in plantimager/commons/RPC.py

def serve_forever(self):
    """
    Starts serving requests for this RPCServer.

    Returns
    -------

    """
    self._stop = False
    while not self._stop:
        notify_watchdog()
        if self._socket.poll(100, zmq.POLLIN) == 0:
            continue
        request = self._socket.recv_json()
        match request["event"]:
            case RPCEvents.FIND_PEER_ADDRESS:
                logger.info("Finding peer address")
                s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                port = random.randint(49152, 65535)
                s.bind(("", port))  # may crash if unlucky
                s.listen(1)
                self._socket.send_json({"success": True, "port": port})
                c, addr_info = s.accept()
                self.peer_addr = addr_info[0]
                c.shutdown(socket.SHUT_RDWR)
                c.close()
                s.shutdown(socket.SHUT_RDWR)
                s.close()
                del s
                logger.info("Connected to peer at address: {}".format(self.peer_addr))
            case RPCEvents.GET_INVENTORY:
                logger.info("Sending inventory")
                response = {
                    "json_methods": {name: func._timeout for name, func in self._json_methods.items()},
                    "buffer_methods": {name: func._timeout for name, func in self._buffer_methods.items()},
                    "signals": list(self._signals.keys()),
                    "properties": list(self._rpc_properties.keys()),
                    "name": self.name,
                }
                logger.info(response)
                self._socket.send_json(response)
            case RPCEvents.INIT_SIGNALS_HANDLING:
                logger.info("Initializing signal handling")
                address, port = request["address"], request["port"]
                self._signal_socket = self.context.socket(zmq.REQ)
                self._signal_socket.connect(f"tcp://{address}:{port}")
                for sig_name, sig in self._signals.items():
                    sig.connect(partial(self._send_signal, sig_name))
                self._socket.send_json({"success": True})
                logger.info("Successfully initialized signal handling")
            case RPCEvents.METHOD_CALL:
                method: str = request["method"]
                params: dict[str, bytes] = request["params"]
                logger.info(f"Executing {method} with params {params}")
                if method in self._json_methods:
                    self._exec_json(self._json_methods[method], params)
                elif method in self._buffer_methods:
                    self._exec_buffer(self._buffer_methods[method], params)
                else:
                    logger.error(f"Method {method} not implemented")
                    self._socket.send_json({"success": False, "error": f"Method {method} not implemented"})
            case RPCEvents.PROPERTY_GET:
                prop_name: str = request["property"]
                logger.debug(f"Getting property {prop_name}")
                try:
                    val = getattr(self, prop_name)
                except Exception as e:
                    logger.error(f"Failed to execute 'get property' for property {prop_name}")
                    traceback_str = traceback.format_exc(limit=10)
                    print(traceback_str, file=sys.stderr)
                    self._socket.send_json({"success": False, "error": str(e), "traceback": traceback_str})
                else:
                    self._socket.send_json({
                        "success": True, "value": val,
                    })
            case RPCEvents.PROPERTY_SET:
                prop_name: str = request["property"]
                val = request["value"]
                logger.debug(f"Setting property {prop_name} to {val}")
                try:
                    setattr(self, prop_name, val)
                except Exception as e:
                    logger.error(f"Failed to execute 'set property' for property {prop_name}")
                    traceback_str = traceback.format_exc(limit=10)
                    print(traceback_str, file=sys.stderr)
                    self._socket.send_json({"success": False, "error": str(e), "traceback": traceback_str})
                else:
                    self._socket.send_json({
                        "success": True,
                    })
            case RPCEvents.STOP_SERVER:
                self._socket.send_json({"success": True})
                break
    logger.info("Server stopped")

set_config Link

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 Link

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 Link

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)

rpc_controller

plantimager.controller.scanner.rpc_controller Link

RPCControllerServer Link

camera_names Link

handle_scanner_changed Link

max_progress Link

progress Link

ready_to_scan Link

register_method_buffer staticmethod Link

register_method_json staticmethod Link

register_to_registry Link

run_scan Link

serve_forever Link

set_config Link

set_dataset_name Link

set_db_url Link

register_method_buffer `staticmethod` Link

register_method_json `staticmethod` Link