larvaworld.lib.ipc.ipc

Classes

Message	A base class for serializable messages.
Client	A client class for handling IPC (Inter-Process Communication) with a server.
Server	A server class that handles inter-process communication (IPC) using a stream-based server.

Module Contents

class larvaworld.lib.ipc.ipc.Message

A base class for serializable messages.

classmethod deserialize(objects: list[Any]) → list[T]

Deserialize a list of objects into their corresponding class instances.

Args:

cls (type): The base class to use for deserialization. objects (list): A list of serialized objects, where each object is either

an instance of Message or a dictionary containing the keys “class”, “args”, and “kwargs”.

Returns:

list: A list of deserialized objects, where each object is an instance

of the class specified in the serialized data.

Raises:

UnknownMessageClass: If the class specified in the serialized data is not found. InvalidSerialization: If there is an error in the deserialization process.

serialize() → dict[str, Any]

Serializes the current object instance into a dictionary.

The dictionary contains the class name, positional arguments, and keyword arguments required to recreate the object instance.

Returns:

dict: A dictionary with the following keys:

• “class” (str): The name of the class of the current object instance.

• “args” (list): The positional arguments used to initialize the object.

• “kwargs” (dict): The keyword arguments used to initialize the object.

class larvaworld.lib.ipc.ipc.Client(server_address: tuple[str, int] | str | bytes)

A client class for handling IPC (Inter-Process Communication) with a server.

addr: tuple[str, int] | str | bytes

sock

connect() → None

close() → None

send(objects: list[Message]) → list[Message]

class larvaworld.lib.ipc.ipc.Server(server_address: tuple[str, int] | str | bytes, callback: Callable[[list[Message]], list[Message]] | None, bind_and_activate: bool = True)

Bases: socketserver.ThreadingUnixStreamServer

A server class that handles inter-process communication (IPC) using a stream-based server.