ubii.framework.protocol module

class ubii.framework.protocol.AbstractProtocol

Bases: Generic[T_EnumFlag], ABC

ABC to implement a state machine with async callbacks. A AbstractProtocol can be used as an async context manager, to start and stop the protocol automatically, or be “manually” started and stopped using the protocols methods.

state_changes: Mapping[Tuple[T_EnumFlag | None, ...], Callback]

Assign to this mapping in your concrete implementation to define callbacks for state changes.

starting_state: T_EnumFlag

Assign a state to this attribute as the protocols starting state when implementing a concrete protocol

end_state: T_EnumFlag

Assign a state to this attribute as the protocols end state when implementing a concrete protocol

property finished: bool

Convenience property that simply returns whether the protocols state is its end state.

property was_started: bool

Convenience property that returns whether a protocol has already been started

property context: SimpleNamespace

It’s encouraged to overwrite this and return a dataclass or something that is better type-able in your concrete implementation

change_context

Condition to notify / wait for changed context

task_nursery

This manages the tasks created by the protocol as well as teardown behaviour if the protocol is stopped

get_state_change_callback

This callable uses get_matching_value() to return the appropriate state change from state_changes for a tuple of states

property state: T_EnumFlag

Makes shared access to the protocol state easy

start() → AbstractProtocol[T_EnumFlag]

Starts the protocol task.

It is encouraged to wait for the protocol if the task is cancelled instead of waiting for the task. In either case, canceling the task will trigger the sentinel task to handle the protocol teardown, which might - depending on the protocol implementation - mean that closing the event loop shortly after the cancellation might raise exceptions from the scheduled cleanup operations, so it’s mandatory to sleep for a short time (e.g. asyncio.sleep(0)) before closing the loop if your protocol schedules async tasks during its teardown.

Returns:

reference to self – the started protocol

See also

RunProtocol – specialized coroutine that is used to create the task

task_nursery – the object that handles cancellation / stopping of the task when necessary

async stop() → None

Gracefully shut down the protocol by setting the protocol state to the end state. This will call the appropriate state change callback (for the \(current \rightarror end\) state change) and finish the task created by start()

Returns control back to the caller after the task created by start() stopped and all teardown callbacks have been scheduled

class ubii.framework.protocol.RunProtocol(protocol: AbstractProtocol)

Bases: CoroutineWrapper

A specialized coroutine to wait for changing protocol state, get the appropriate callbacks from the protocols state_changes and execute them.

Starting a protocol using the start() method, starts a RunProtocol coroutine as a task using the protocol’s AbstractProtocol.task_nursery.

__init__(protocol: AbstractProtocol)

Parameters:

protocol – a protocol to run

protocol

Reference to the protocol that is running in this coroutine