cameraserver

plantimager.commons.examples.cameraserver

DummyCamera

DummyCamera(context, url)

Bases: Camera, RPCServer

Dummy Camera serving images from a shared dataset for testing purposes.

Parameters:

Name	Type	Description	Default
context	Context	ZeroMQ context used to create sockets for RPC communication.	required
url	str	Endpoint address that the RPC server will bind to.	required

Source code in plantimager/commons/examples/cameraserver.py

def __init__(self, context: zmq.Context, url: str):
    RPCServer.__init__(self, context, url)
    self._mode = CameraMode.STILL
    self._video_url = "tcp://test_url:1234"
    self._rotation = 0
    self._encoding = "jpeg"
    self._config = {}
    self._image_provider = image_provider()

config

config(value)

    Sets camera controls/parameters from Picamera2 Appendix C.

    Parameters

    value : dict
        Control names and values (e.g., {'Brightness': 0.5, 'Contrast': 1.2}).

    See Also

    Picamera2 Manual Appendix C:

https://pip-assets.raspberrypi.com/categories/652-raspberry-pi-camera-module-2/documents/RP-008156-DS-2-picamera2-manual.pdf?disposition=inline#%5B%7B%22num%22%3A10333%2C%22gen%22%3A0%7D%2C%7B%22name%22%3A%22XYZ%22%7D%2C70.86614%2C781.0236%2C0%5D

Source code in plantimager/commons/examples/cameraserver.py

    @config.setter
    def config(self, value: dict):
        """
        Sets camera controls/parameters from Picamera2 Appendix C.

        Parameters
        ----------
        value : dict
            Control names and values (e.g., {'Brightness': 0.5, 'Contrast': 1.2}).

        See Also
        --------
        Picamera2 Manual Appendix C:
https://pip-assets.raspberrypi.com/categories/652-raspberry-pi-camera-module-2/documents/RP-008156-DS-2-picamera2-manual.pdf?disposition=inline#%5B%7B%22num%22%3A10333%2C%22gen%22%3A0%7D%2C%7B%22name%22%3A%22XYZ%22%7D%2C70.86614%2C781.0236%2C0%5D
        """
        self._config.update(value)
        # Apply controls to the running camera
        self.configChanged.emit(self._config)

get_image

get_image(lores=False)

Return a JPEG‑encoded image (as a memoryview) and a metadata dict.

The lores flag is ignored for the dummy implementation.

Source code in plantimager/commons/examples/cameraserver.py

@RPCServer.register_method_buffer(timeout=10000)
def get_image(self, lores=False) -> (memoryview, dict):
    """Return a JPEG‑encoded image (as a memoryview) and a metadata dict.

    The *lores* flag is ignored for the dummy implementation.
    """
    if self.mode != CameraMode.STILL:
        self.mode = CameraMode.STILL
    image = next(self._image_provider)
    # image = block_mean(image, 5)
    # image = np.clip(image.astype(int) + np.round(np.random.normal(0.0, 0.5, image.shape)), 0, 255)
    if self._encoding == "jpeg":
        buffer = encode_jpeg(image.astype(np.uint8), quality=95, colorsubsampling="420", fastdct=True)
    elif self._encoding == "png":
        rgb_image = image[..., ::-1]
        pil_img = Image.fromarray(rgb_image)
        buf = io.BytesIO()
        pil_img.save(buf, format="PNG")
        buffer = buf.getvalue()
    else:
        raise ValueError(f"Unknown encoding: {self._encoding}")
    time.sleep(CAMERASERVER_LAG/1000)
    return memoryview(buffer), {"format": self._encoding, "rotation": self._rotation, "size": image.shape}

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

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")

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")

image_provider

image_provider()

Generates an infinite sequence of images from a directory.

Yields:

Type	Description
ndarray	Image data loaded by matplotlib.pyplot.imread from the current file. The generator loops indefinitely, restarting after the last image.

Source code in plantimager/commons/examples/cameraserver.py

def image_provider():
    """Generates an infinite sequence of images from a directory.

    Yields
    ------
    numpy.ndarray
        Image data loaded by ``matplotlib.pyplot.imread`` from the current file.
        The generator loops indefinitely, restarting after the last image.
    """
    image_path = os.getenv("PI3_CAMERASERVER_IMAGE")
    if image_path is None:
        dataset_path = get_test_dataset('real_plant')
        image_path = str(dataset_path / "images")

    images = sorted(os.listdir(image_path))
    n = len(images)
    i = 0
    while True:
        image = plt.imread(os.path.join(image_path, images[i % n]))
        # SimpleJPEG only supports RGB, so drop the alpha channel if it exists.
        if image.ndim == 3 and image.shape[2] == 4:
            # keep only the first three channels (R, G, B)
            image = image[:, :, :3]
        yield image
        i += 1