Control

This module provides classes for a simple remote procedure call (RPC) framework.

ATTRIBUTE	DESCRIPTION
`BYTE_SIZE`	size of the byte. TYPE: `int`

CLASS	DESCRIPTION
`ClassMethods`	class to store methods of a class.
`TwoTierQueue`	a queue that can handle two types of items: normal and high-priority.
`Proxy`	a proxy class to handle remote method calls.
`Controller`	a class to control the flow of data and commands between models and views.

Documentation last updated: 2025-06-11

ClassMethods `dataclass`

ClassMethods(name: str, methods: dict[str, dict[str, str]])

Class to store methods of a class.

ATTRIBUTE	DESCRIPTION
`name`	name of the class. TYPE: `str`
`methods`	dictionary of methods and their parameters. TYPE: `dict[str, dict[str, str]]`

Controller

Controller(
    role: str,
    interpreter: Interpreter,
    *,
    relay_delay: int = 1,
)

A class to control the flow of data and commands between models and views.

ATTRIBUTE	DESCRIPTION
`role`	the role of the controller TYPE: `str`
`interpreter`	the interpreter to use TYPE: `Interpreter`
`address`	the address of the controller TYPE: `str \| None`
`relay_delay`	delay for relaying data TYPE: `int`
`relays`	list of relays TYPE: `list`
`callbacks`	dictionary of callbacks TYPE: `dict[str, dict[str, Callable]]`
`events`	dictionary of events TYPE: `dict[str, Event]`
`command_queue`	command queue TYPE: `TwoTierQueue`
`data_buffer`	data buffer TYPE: `dict`
`objects`	dictionary of objects TYPE: `dict`
`object_methods`	dictionary of object methods TYPE: `dict[str, ClassMethods]`
`object_attributes`	dictionary of object attributes TYPE: `dict[str, tuple[str]]`
`execution_event`	event for execution loop TYPE: `Event`
`registry`	object registry TYPE: `dict[str, list[str]]`

METHOD	DESCRIPTION
`receiveRequest`	receive a request
`transmitData`	transmit data
`broadcastRegistry`	broadcast the registry
`register`	register an object
`unregister`	unregister an object
`extractMetadata`	extract metadata from a command
`extractMethods`	extract methods from an object
`exposeAttributes`	expose attributes of registered objects
`exposeMethods`	expose methods of registered objects
`start`	start the execution loop
`stop`	stop the execution loop
`executeCommand`	execute a command
`transmitRequest`	transmit a request
`receiveData`	receive data
`retrieveData`	retrieve data
`getAttributes`	get attributes of the controller
`getMethods`	get methods of the controller
`relay`	relay a request or data
`relayRequest`	relay a request
`relayData`	relay data
`subscribe`	subscribe to a relay
`unsubscribe`	unsubscribe from a relay
`setAddress`	set the address of the controller

Initialize the Controller class.

PARAMETER	DESCRIPTION
`role`	the role of the controller TYPE: `str`
`interpreter`	the interpreter to use TYPE: `Interpreter`
`relay_delay`	delay for relaying data. Defaults to 1. TYPE: `int` DEFAULT: `1`

registry `property` `writable`

registry: dict[str, str]

Object registry

broadcastRegistry

broadcastRegistry(target: Iterable[str] | None = None)

Broadcast the registry

PARAMETER	DESCRIPTION
`target`	the target addresses. Defaults to None. TYPE: `Iterable[str] \| None` DEFAULT: `None`

executeCommand

executeCommand(
    command: Mapping[str, Any],
) -> tuple[Any, dict[str, Any]]

Execute a command

PARAMETER	DESCRIPTION
`command`	the command to execute TYPE: `Mapping[str, Any]`

RETURNS	DESCRIPTION
`tuple[Any, dict[str, Any]]`	tuple[Any, dict[str, Any]]: the result of the command and the status

exposeAttributes

exposeAttributes()

Expose attributes of registered objects

exposeMethods

exposeMethods()

Expose methods of registered objects

extractMetadata

extractMetadata(
    command: Mapping[str, Any],
) -> dict[str, Any]

Extract metadata from a command

PARAMETER	DESCRIPTION
`command`	the command to extract metadata from TYPE: `Mapping[str, Any]`

RETURNS	DESCRIPTION
`dict[str, Any]`	dict[str, Any]: the extracted metadata

extractMethods `staticmethod`

extractMethods(new_object: Callable) -> ClassMethods

Extract methods from an object

PARAMETER	DESCRIPTION
`new_object`	the object to extract methods from TYPE: `Callable`

RETURNS	DESCRIPTION
`ClassMethods`	the extracted methods TYPE: `ClassMethods`

getAttributes

getAttributes(
    target: Iterable[int] | None = None,
    *,
    private: bool = True,
) -> dict

Get attributes

PARAMETER	DESCRIPTION
`target`	the target addresses. Defaults to None. TYPE: `Iterable[int] \| None` DEFAULT: `None`
`private`	flag to indicate private transmission. Defaults to True. TYPE: `bool` DEFAULT: `True`

RETURNS	DESCRIPTION
`dict`	the attributes TYPE: `dict`

getMethods

getMethods(
    target: Iterable[int] | None = None,
    *,
    private: bool = True,
) -> dict

Get methods

PARAMETER	DESCRIPTION
`target`	the target addresses. Defaults to None. TYPE: `Iterable[int] \| None` DEFAULT: `None`
`private`	flag to indicate private transmission. Defaults to True. TYPE: `bool` DEFAULT: `True`

RETURNS	DESCRIPTION
`dict`	the methods TYPE: `dict`

receiveData

receiveData(
    packet: str | bytes | None = None,
    *,
    sender: str | None = None,
    **kwargs,
)

Receive data

PARAMETER	DESCRIPTION
`packet`	the packet to receive. Defaults to None. TYPE: `str \| bytes` DEFAULT: `None`
`sender`	the sender of the packet. Defaults to None. TYPE: `str \| None` DEFAULT: `None`

receiveRequest

receiveRequest(
    packet: str | bytes | None = None,
    *,
    sender: str | None = None,
    **kwargs,
)

Receive a request

PARAMETER	DESCRIPTION
`packet`	the request to receive. Defaults to None. TYPE: `str \| bytes` DEFAULT: `None`
`sender`	the sender of the request. Defaults to None. TYPE: `str \| None` DEFAULT: `None`

register

register(
    new_object: Callable, object_id: str | None = None
)

Register an object

PARAMETER	DESCRIPTION
`new_object`	the object to register TYPE: `Callable`
`object_id`	the ID of the object. Defaults to None. TYPE: `str \| None` DEFAULT: `None`

relay

relay(
    packet: str | bytes | None,
    callback_type: str,
    addresses: Iterable[int] | None = None,
)

Relay a message

PARAMETER	DESCRIPTION
`packet`	the message to relay TYPE: `str \| bytes \| None`
`callback_type`	the callback type TYPE: `str`
`addresses`	the target addresses. Defaults to None. TYPE: `Iterable[int] \| None` DEFAULT: `None`

relayData

relayData(packet: str | bytes | None = None, **kwargs)

Relay data

PARAMETER	DESCRIPTION
`packet`	the packet to relay. Defaults to None. TYPE: `str \| bytes` DEFAULT: `None`

relayRequest

relayRequest(packet: str | bytes | None = None, **kwargs)

Relay a request

PARAMETER	DESCRIPTION
`packet`	the request to relay. Defaults to None. TYPE: `str \| bytes` DEFAULT: `None`

retrieveData

retrieveData(
    request_id: str,
    timeout: int | float = 5,
    *,
    sender: str | None = None,
    min_count: int | None = 1,
    max_count: int | None = 1,
    default: Any | None = None,
    data_only: bool = True,
    close_request: bool = True,
) -> Any | dict[tuple[str, str], Any]

Retrieve data

PARAMETER	DESCRIPTION
`request_id`	the request ID TYPE: `str`
`timeout`	the timeout. Defaults to 5. TYPE: `int \| float` DEFAULT: `5`
`min_count`	the minimum count. Defaults to 1. TYPE: `int \| None` DEFAULT: `1`
`max_count`	the maximum count. Defaults to 1. TYPE: `int \| None` DEFAULT: `1`
`default`	the default value. Defaults to None. TYPE: `Any \| None` DEFAULT: `None`
`data_only`	flag to indicate data only. Defaults to True. TYPE: `bool` DEFAULT: `True`
`close_request`	flag to indicate close request. Defaults to True. TYPE: `bool` DEFAULT: `True`

RETURNS	DESCRIPTION
`Any \| dict[tuple[str, str], Any]`	Any \| dict[tuple[str,str], Any]: the retrieved data

setAddress

setAddress(address: int | str)

Set the address

PARAMETER	DESCRIPTION
`address`	the address TYPE: `int \| str`

start

start()

Start the execution loop

stop

stop()

Stop the execution loop

subscribe

subscribe(
    callback: Callable,
    callback_type: str,
    address: int | str | None = None,
    *,
    relay: bool = False,
)

Subscribe to a callback

PARAMETER	DESCRIPTION
`callback`	the callback to subscribe TYPE: `Callable`
`callback_type`	the callback type TYPE: `str`
`address`	the address. Defaults to None. TYPE: `int \| str \| None` DEFAULT: `None`
`relay`	flag to indicate relay. Defaults to False. TYPE: `bool` DEFAULT: `False`

transmitData

transmitData(
    data: Any,
    *,
    metadata: Mapping[str, Any] | None = None,
    status: Mapping[str, Any] | None = None,
)

Transmit data

PARAMETER	DESCRIPTION
`data`	the data to transmit TYPE: `Any`
`metadata`	the metadata to include. Defaults to None. TYPE: `Mapping[str, Any] \| None` DEFAULT: `None`
`status`	the status to include. Defaults to None. TYPE: `Mapping[str, Any] \| None` DEFAULT: `None`

transmitRequest

transmitRequest(
    command: Mapping[str, Any],
    target: Iterable[int | str] | None = None,
    *,
    private: bool = True,
    priority: bool = False,
    rank: int | None = None,
) -> str

Transmit a request

PARAMETER	DESCRIPTION
`command`	the command to transmit TYPE: `Mapping[str, Any]`
`target`	the target addresses. Defaults to None. TYPE: `Iterable[int \| str] \| None` DEFAULT: `None`
`private`	flag to indicate private transmission. Defaults to True. TYPE: `bool` DEFAULT: `True`
`priority`	flag to indicate high-priority transmission. Defaults to False. TYPE: `bool` DEFAULT: `False`
`rank`	rank of the high-priority transmission. Defaults to None. TYPE: `int` DEFAULT: `None`

RETURNS	DESCRIPTION
`str`	the request ID TYPE: `str`

unregister

unregister(
    object_id: str | None = None,
    old_object: Callable | None = None,
) -> bool

Unregister an object

PARAMETER	DESCRIPTION
`object_id`	the ID of the object. Defaults to None. TYPE: `str \| None` DEFAULT: `None`
`old_object`	the object to unregister. Defaults to None. TYPE: `Callable \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`bool`	flag indicating success TYPE: `bool`

unsubscribe

unsubscribe(
    callback_type: str, address: int | str
) -> Callable | None

Unsubscribe from a callback

PARAMETER	DESCRIPTION
`callback_type`	the callback type TYPE: `str`
`address`	the address TYPE: `int \| str`

RETURNS	DESCRIPTION
`Callable \| None`	Callable\|None: the unsubscribed callback

Proxy

Proxy(prime: Callable, object_id: str | None = None)

A proxy class to handle remote method calls.

ATTRIBUTE	DESCRIPTION
`prime`	the object to create a proxy for TYPE: `Callable`
`object_id`	the ID of the object TYPE: `str`
`controller`	the controller bound to the proxy TYPE: `Controller`
`remote`	flag to indicate remote method calls TYPE: `bool`

METHOD	DESCRIPTION
`factory`	factory method to create a new class with methods of the prime object
`createMethodEmitter`	create a method emitter for the proxy class
`createPropertyEmitter`	create a property emitter for the proxy class
`bindController`	bind a controller to the proxy
`releaseController`	release the controller from the proxy

Initialize the Proxy class.

PARAMETER	DESCRIPTION
`prime`	the object to create a proxy for TYPE: `Callable`
`object_id`	the ID of the object. Defaults to None. TYPE: `str \| None` DEFAULT: `None`

bindController

bindController(controller: Controller)

Bind a controller to the proxy.

PARAMETER	DESCRIPTION
`controller`	the controller to bind to the proxy TYPE: `Controller`

createMethodEmitter `staticmethod`

createMethodEmitter(method: Callable) -> Callable

Create a method emitter for the proxy class.

PARAMETER	DESCRIPTION
`method`	the method to create an emitter for TYPE: `Callable`

RETURNS	DESCRIPTION
`Callable`	the method emitter TYPE: `Callable`

createPropertyEmitter `staticmethod`

createPropertyEmitter(attr_name: str) -> property

Create a property emitter for the proxy class.

PARAMETER	DESCRIPTION
`attr_name`	the name of the property to create an emitter for TYPE: `str`

RETURNS	DESCRIPTION
`property`	the property emitter TYPE: `property`

factory `classmethod`

factory(
    prime: Callable, object_id: str | None = None
) -> Type[Proxy]

Factory method to create a new class with methods and properties of the prime object.

PARAMETER	DESCRIPTION
`prime`	the object to create a proxy for TYPE: `Callable`
`object_id`	the ID of the object. Defaults to None. TYPE: `str \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`Type[Proxy]`	Type[Proxy]: the new class with methods and properties of the prime object

releaseController

releaseController() -> Controller

Release the controller from the proxy.

RETURNS	DESCRIPTION
`Controller`	the controller that was bound to the proxy TYPE: `Controller`

TwoTierQueue

TwoTierQueue()

A queue that can handle two types of items: normal and high-priority.

ATTRIBUTE	DESCRIPTION
`normal_queue`	queue for normal items. TYPE: `Queue`
`high_priority_queue`	queue for high-priority items. TYPE: `PriorityQueue`
`last_used_queue_normal`	flag to indicate the last used queue. TYPE: `bool`
`priority_counter`	counter for high-priority items. TYPE: `int`

METHOD	DESCRIPTION
`qsize`	return the size of the queue
`empty`	check if the queue is empty
`full`	check if the queue is full
`put`	put an item in the queue
`put_nowait`	put an item in the queue without waiting
`get`	get an item from the queue
`get_nowait`	get an item from the queue without waiting
`task_done`	mark a task as done
`join`	wait for all tasks to be done
`put_first`	put an item at the front of the priority queue
`put_priority`	put a high-priority item in the queue
`put_queue`	put an item in the queue
`reset`	reset the queue

empty

empty()

Check if the queue is empty.

full

full()

Check if the queue is full.

get

get(
    block: bool = True, timeout: float | None = None
) -> Any

Get an item from the queue.

PARAMETER	DESCRIPTION
`block`	flag to block the queue. Defaults to True. TYPE: `bool` DEFAULT: `True`
`timeout`	time to wait for the queue. Defaults to None. TYPE: `float` DEFAULT: `None`

RETURNS	DESCRIPTION
`Any`	item from the queue. TYPE: `Any`

get_nowait

get_nowait() -> Any

Get an item from the queue without waiting.

RETURNS	DESCRIPTION
`Any`	item from the queue. TYPE: `Any`

join

join()

Wait for all tasks to be done.

put

put(
    item: Any,
    block: bool = True,
    timeout: float | None = None,
    *,
    priority: bool = False,
    rank: int | None = None,
)

Put an item in the queue.

PARAMETER	DESCRIPTION
`item`	item to put in the queue. TYPE: `Any`
`block`	flag to block the queue. Defaults to True. TYPE: `bool` DEFAULT: `True`
`timeout`	time to wait for the queue. Defaults to None. TYPE: `float` DEFAULT: `None`
`priority`	flag to indicate high-priority item. Defaults to False. TYPE: `bool` DEFAULT: `False`
`rank`	rank of the high-priority item. Defaults to None. TYPE: `int` DEFAULT: `None`

put_first

put_first(item: Any)

Put an item at the front of the priority queue.

PARAMETER	DESCRIPTION
`item`	item to put in the queue. TYPE: `Any`

put_nowait

put_nowait(
    item: Any,
    *,
    priority: bool = False,
    rank: int | None = None,
)

Put an item in the queue without waiting.

PARAMETER	DESCRIPTION
`item`	item to put in the queue. TYPE: `Any`
`priority`	flag to indicate high-priority item. Defaults to False. TYPE: `bool` DEFAULT: `False`
`rank`	rank of the high-priority item. Defaults to None. TYPE: `int \| None` DEFAULT: `None`

put_priority

put_priority(
    item: Any,
    rank: int,
    block: bool = True,
    timeout: float | None = None,
)

Put a high-priority item in the queue.

PARAMETER	DESCRIPTION
`item`	item to put in the queue. TYPE: `Any`
`rank`	rank of the high-priority item TYPE: `int`
`block`	flag to block the queue. Defaults to True. TYPE: `bool` DEFAULT: `True`
`timeout`	time to wait for the queue. Defaults to None. TYPE: `float` DEFAULT: `None`

put_queue

put_queue(
    item: Any,
    block: bool = True,
    timeout: float | None = None,
)