stateful_node

drunc.controller.stateful_node

Classes

StatefulNode(fsm_configuration, publisher=None, init_state='', session='', name='', top_segment_controller=False)

Bases: ABC

Source code in drunc/controller/stateful_node.py

def __init__(
    self,
    fsm_configuration,
    publisher: Optional[Callable[[Any], None]] = None,
    init_state: str = "",
    session: str = "",
    name: str = "",
    top_segment_controller: bool = False,
):
    self.publisher = publisher
    self.custom_origin = {"top_segment_controller": top_segment_controller}
    self.session = session
    self.name = name
    self._ready_state = False
    self.__fsm = FSM(fsm_configuration)
    self.log = get_logger("controller.core.StatefulNode")
    self.__operational_state = OperationalState(
        stateful_node=self,
        initial_value=init_state,
    )
    self.__operational_sub_state = OperationalState(
        stateful_node=self, initial_value=init_state
    )
    self.__included = InclusionState(stateful_node=self, initial_value=True)
    self.__in_error = ErrorState(stateful_node=self, initial_value=False)

Functions

decode_fsm_arguments(fsm_command)

Decodes the arguments of a FSMCommand according to the transition definition in the FSM configuration.

Parameters:

Name	Type	Description	Default
fsm_command	FSMCommand	The FSMCommand containing the command name and arguments.	required

Returns:

Name	Type	Description
dict		A dictionary containing the decoded arguments for the transition.

Raises:

Type	Description
KeyError	If the command name does not correspond to any transition in the FSM configuration.
ValueError	If the arguments cannot be decoded according to the transition definition.

Source code in drunc/controller/stateful_node.py

def decode_fsm_arguments(self, fsm_command: FSMCommand):
    """
    Decodes the arguments of a FSMCommand according to the transition definition in
    the FSM configuration.

    Args:
        fsm_command (FSMCommand): The FSMCommand containing the command name and
            arguments.

    Returns:
        dict: A dictionary containing the decoded arguments for the transition.

    Raises:
        KeyError: If the command name does not correspond to any transition in the
            FSM configuration.
        ValueError: If the arguments cannot be decoded according to the transition
            definition.
    """
    transition = self.get_fsm_transition(fsm_command.command_name)
    return decode_fsm_arguments(fsm_command.arguments, transition.arguments)

Functions