server.ua_fsm.UaFiniteStateMachine

class documentation

class UaFiniteStateMachine(Generic[_StateType]):

Constructor: UaFiniteStateMachine(name, states, initial, write_func)

Finite state machine with OPC-UA representation based on transitions library.

Method	`__init__`	Create a new instance.
Method	`add_transition`	Add the given transition to the internal state machine.
Async Method	`init`	Asynchronous initialization of the instance.
Async Method	`try_trigger`	Try to trigger an event in the internal state machine.
Instance Variable	`logger`	Undocumented
Instance Variable	`name`	Undocumented
Instance Variable	`ua_node`	OPC UA node 'StateMachine' representing this instance.
Instance Variable	`ua_state_variable`	OPC UA node 'CurrentState' of the state machine.
Property	`current_state`	Return the current state of the state machine.
Async Method	`_after_state_change`	Handle callback from state machine after a state changes. Used to update the OPC-UA representation.
Instance Variable	`_internal_machine`	Undocumented
Instance Variable	`_write_func`	Undocumented

def __init__(self, *, name: str, states: Iterable[_StateType], initial: _StateType, write_func: Callable[[_StateType], Awaitable[None]]): ¶

Create a new instance.

Parameters
name:`str`	Name of the state machine.
states:`Iterable[_StateType]`	States of the state machine.
initial:`_StateType`	Initial state of the state machine.
write_func:`Callable[[_StateType], Awaitable[None]]`	Async callback to write the current state.

def add_transition(self, **kwargs): ¶

Add the given transition to the internal state machine.

For documentation of available keyword arguments, please look into transitions.Machine.add_transition().

async def init(self, state_machine_node: Node): ¶

Asynchronous initialization of the instance.

Adds a new OPC-UA object to the given parent node of some OPC-UA StateMachineType. Adds sub OPC-UA nodes representing all states of the internal state machine.

Parameters
state_machine_node:`Node`	OPC-UA node of the state machine (instantiated from respective types).

async def try_trigger(self, user: UserProtocol | None, trigger_func: str) -> ua.StatusCode: ¶

Try to trigger an event in the internal state machine.

Handles expected exceptions due to wrong state and UaStatusCodeError exceptions due to unsatisfied conditions.

Parameters
user:`UserProtocol \| None`	User which triggers the event.
trigger_func:`str`	Name of the trigger function.

Returns
`ua.StatusCode`	Returns either an OPC-UA status code or what the provided success_func returns.

logger = ¶

Undocumented

name = ¶

Undocumented

ua_node: Node = ¶

OPC UA node 'StateMachine' representing this instance.

ua_state_variable: Node = ¶

OPC UA node 'CurrentState' of the state machine.

@property

current_state: _StateType = ¶

Return the current state of the state machine.

async def _after_state_change(self, *args, **kwargs): ¶

Handle callback from state machine after a state changes. Used to update the OPC-UA representation.

_internal_machine = ¶

Undocumented

_write_func = ¶

Undocumented