server.ua_fsm.UaFiniteStateMachine

class documentation

class UaFiniteStateMachine:

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

Base class for all finite state machine OPC-UA representations based on transitions library.

Method	`__init__`	Undocumented
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.
Class Variable	`ua_last_transition_id`	Undocumented
Instance Variable	`before_state`	Undocumented
Instance Variable	`exclude_states`	Undocumented
Instance Variable	`logger`	Undocumented
Instance Variable	`name`	Undocumented
Instance Variable	`represent_states`	Undocumented
Instance Variable	`represent_transitions`	Undocumented
Instance Variable	`state_number`	Undocumented
Instance Variable	`transition_number`	Undocumented
Instance Variable	`ua_available_state_node_ids`	Maps the state name to an OPC UA node ID representing the state in the OPC UA address space.
Instance Variable	`ua_available_transitions_node_ids`	Maps the transition name to an OPC UA node ID representing the transition in the OPC UA address space.
Instance Variable	`ua_node`	OPC UA node 'StateMachine' representing this instance.
Instance Variable	`ua_node_available_states`	OPC UA node 'AvailableStates'.
Instance Variable	`ua_node_available_transitions`	OPC UA node 'AvailableTransitions'.
Instance Variable	`ua_node_type_definition`	Undocumented
Instance Variable	`ua_state_id`	OPC UA node 'Id' property of the 'CurrentState' node.
Instance Variable	`ua_state_variable`	OPC UA node 'CurrentState' of the state machine.
Property	`current_state`	Undocumented
Async Method	`_add_available_states_transitions`	Undocumented
Async Method	`_after_state_change`	Handle callback from state machine after a state changes. Used to update the OPC-UA representation.
Async Method	`_init_state`	Add ua_node_id attribute to given state.
Async Method	`_init_transition`	Undocumented
Async Method	`_init_ua_states`	Add all states to OPC-UA state machine representation and sets the initial state of the FSM.
Async Method	`_init_ua_transitions`	Add all transitions to OPC-UA state machine representation.
Instance Variable	`_internal_machine`	Undocumented
Instance Variable	`_write_func`	Undocumented

def __init__(self, *, name: str, states: Sequence[Enum], initial: AsyncState | Enum, write_func: Callable[[Enum], Awaitable[None]], represent_transitions: bool = False, represent_states: bool = False, exclude_states: Sequence[Enum] | None = None): ¶

Undocumented

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, state_machine_type: Node | None = None): ¶

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).
state_machine_type:`Node \| None`	#TODO OPC UA node of the state machine type definition; normally not needed since node has method 'read_type_definition()', parameter used as workaround if reason is expected, None is used as default cause lock is no opc ua state machine

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.

ua_last_transition_id: Node = ¶

Undocumented

before_state: AsyncState | None = ¶

Undocumented

exclude_states = ¶

Undocumented

logger = ¶

Undocumented

name = ¶

Undocumented

represent_states = ¶

Undocumented

represent_transitions = ¶

Undocumented

state_number: int = ¶

Undocumented

transition_number: int = ¶

Undocumented

ua_available_state_node_ids: dict[str, NodeId] = ¶

Maps the state name to an OPC UA node ID representing the state in the OPC UA address space.

ua_available_transitions_node_ids: dict[str, NodeId] = ¶

Maps the transition name to an OPC UA node ID representing the transition in the OPC UA address space.

ua_node: Node = ¶

OPC UA node 'StateMachine' representing this instance.

ua_node_available_states: Node | None = ¶

OPC UA node 'AvailableStates'.

ua_node_available_transitions: Node | None = ¶

OPC UA node 'AvailableTransitions'.

ua_node_type_definition: Node | None = ¶

Undocumented

ua_state_id: Node | None = ¶

OPC UA node 'Id' property of the 'CurrentState' node.

ua_state_variable: Node = ¶

OPC UA node 'CurrentState' of the state machine.

@property

current_state: AsyncState = ¶

Undocumented

async def _add_available_states_transitions(self): ¶

Undocumented

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

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

async def _init_state(self, state: State, parent_node: Node | None = None): ¶

Add ua_node_id attribute to given state.

This is achieved by adding a OPC-UA node representing the given state to the OPC-UA representation of the state machine.

async def _init_transition(self, transition: AsyncTransition): ¶

Undocumented

async def _init_ua_states(self): ¶

Add all states to OPC-UA state machine representation and sets the initial state of the FSM.

async def _init_ua_transitions(self): ¶

Add all transitions to OPC-UA state machine representation.

_internal_machine = ¶

Undocumented

_write_func = ¶

Undocumented