procaaso_state_machine 0.0.3

No project description provided

procaaso-state-machine

The ProCaaSo based state machine is an ENS centric implementaion of a finite state machine. It's goal is to provide users a state machine that works well within the ProCaaSo App loop, its handlers, classes, methods and other such details will be outlined below.

StateMachine Class

A deterministic finite acceptor state machine designed for use within the ProCaaSo framework.

• This machine works in conjunction with the 'State' object from procaaso_state.
• State objects are not modifiable after they are added to the state machine instance.
• The state object should be defined and complete before addition to the StateMachine.
• Recommended practice is to place StateMachine logic with try-except statements to catch any errors.

Initialization

state_machine_instance = StateMachine()

Attributes

• instruments (dict): Dictionary containing each instrument object added to a machine.
• __states (dict): A dictionary of state class objects that the state machine will keep track of.
• __currentStateId (int): Tracks identification for the current state.
• __currentTransitions (list): A dictionary of lists, where the key is state, each sublist containing the state's possible transitions.
• __isConfigured (bool): Tracks whether the machine is able to enter its initial state.
• __stateStartTime (float): Timestamp indicating the start time of the current state.
• __currentEvent (Event): The current event associated with the state machine.

Methods

Method: run_routine

run_routine(self, routineName: str = None, **kwargs)

Run a routine or all routines associated with the current state. Amend the 'stateMachine' object into the kwargs of the routines for easy access.

If no routine name is supplied, all routines housed in a state will be executed in the order they are provided. If a routine name is specified, it will run that routine only.

If kwargs are provided and a routine name is supplied, the specified routine will receive the kwargs. If kwargs are provided and no routine name is supplied, all routines will receive the same kwargs.

If a routine supplies a return value while running multiple routines, it will be added to a return value list and returned at end of function If one routine is ran and supploes a return value, that return value is simply returned

Args

• routineName (str, optional): The name of the routine to run. Defaults to None.
• **kwargs: Additional keyword arguments to pass to the routines.

Raises

• Exception: If there is an issue running the routine(s).

Returns

Appends the routine name and the return value for the routine to the return dictionary

• Dict ['routineName'Return': routineReturnValue]
• Or, if a single routine returns a value, return that value
• Or, if now return values are given from the function, doesnt return

Usage

state_machine_instance.run_routine()
state_machine_instance.run_routine("Main Routine")
state_machine_instance.run_routine("Main Routine", returnKwarg="I'm a Keyword Function", anotherVariable="I'm another KWARG")

Method: set_currentEvent

set_currentEvent(self, event: Event)

Sets the current event object.

Parameters

• event (Event): The event object to set as the current event.

Raises

• Exception: If the provided event is not of type 'Event'.

Usage

obj.set_currentEvent(my_event)

Method: get_currentEvent

get_currentEvent(self)

Retrieve the current event object.

Raises

• Exception: If there is no current event set.

Returns

• Event: The current event object.

Usage

current_event = obj.get_currentEvent()

Method: clear_currentEvent

clear_currentEvent(self) -> None

Clear the current event in the StateMachine object.

Usage

state_machine_instance.clear_currentEvent()

Method: add_state

add_state(self, state: State)

Add a State object to the StateMachine.

Parameters

• state (State): The State object to be added.

Raises

• KeyError: If the state ID is already present in the StateMachine.
• Exception: If an unexpected exception occurs during the process.

Usage

state_machine_instance.add_state(my_state_object)

Method: get_states

get_states(self)

Get the dictionary containing the State objects in the StateMachine.

Returns

• dict: The dictionary of State objects.

Raises

• Exception: If an unexpected exception occurs during the process.

Usage

states_dict = state_machine_instance.get_states()

Method: get_current_state_id

get_current_state_id(self)

Get the ID of the current state.

Returns

• int | str: The ID of the current state.

Raises

• Exception: If an unexpected exception occurs during the process.

Usage

current_state_id = state_machine_instance.get_current_state_id()

Method: get_current_transitions

get_current_transitions(self)

Get the list of current transitions.

Returns

• list: The list of current transitions.

Raises

• Exception: If an unexpected exception occurs during the process.

Usage

current_transitions_list = state_machine_instance.get_current_transitions()

Method: transition_state

transition_state(self, newStateId: int)

Transition to a new state: For the current state, check if onExitEnabled:

• If True, run the exit routine
  • If the routine fails, stay in the current state and throw error.
    • If routine succeeds, transition to new state
  • If False, transition to new state

In new state and check onEnterEnabled:

• If true, run the on enter function

  • If the routine fails, stay in the current state and throw error.

  • If routine succeeds exit the transition_state routine

• If false, exit the transition_state routine

Parameters

• newStateId (int | str): The ID of the new state.
• onEnterParameters (dict[str, any]) OPTIONAL: A structure to contain any variables for the on enter function
• onExitParameters (dict[str, any]) OPTIONAL: A struture to contain any variables for the on exit function

Raises

• Exception: If an unexpected exception occurs during the process.

Returns

• Dict['onExitReturn': onExitReturnValue, 'onEnterReturn': onEnterReturnValue]

• Or, if now return values are given from the function, returns none

Usage

sm.transition_state(2)
sm.transition_state(1, onEnterParameters=onEnterParams)
sm.transition_state(3, onExitParameters=onExitParams)
sm.transition_state(2, onEnterParameters=onEnterParams, onExitParameters=onExitParams)

Method: start_state_machine

start_state_machine(self, initialStateId: int | str)

Start the StateMachine. Transitions to itself to trigger an onEnter functionality, if present

Parameters

• initialStateId (int | str): The ID of the initial state.

• onEnterParameters (dict[str, any]) OPTIONAL: A structure to contain any variables for the on enter function

Raises

• Exception: If the StateMachine is already configured and/or running.

Usage

state_machine_instance.start_state_machine(1)

Method: get_is_configured

get_is_configured(self)

Get the value of the isConfigured attribute.

Returns

• bool: The value of the isConfigured attribute.

Raises

• Exception: If an unexpected exception occurs during the process.

Usage

is_configured_value = state_machine_instance.get_is_configured()

Method: add_instrument

add_instrument(self, instruments: dict[str, Any])

Add instruments to the StateMachine.

Parameters

• instruments (dict): An open-ended function so the user can define the data structure they wish to use for their instruments. If the instrument dictionary is already present in the StateMachine dictionary, it will simply update the first entry.

Raises

• TypeError: If the provided instruments are not of type 'dict'.
• Exception: If an unexpected exception occurs during the process.

Usage

state_machine_instance.add_instrument({"instrument1": data1, "instrument2": data2})

Method: get_instruments

get_instruments(self)

Get the instruments dictionary from the StateMachine.

Returns

• dict: The dictionary containing instruments.

Raises

• Exception: If an unexpected exception occurs during the process.

Usage

instruments_dict = state_machine_instance.get_instruments()

Method: get_state_start_time

get_state_start_time(self)

Get the start time of the state.

Returns

• float: The start time of the state.

Raises

• Exception: If retrieving the start time fails.

Usage

start_time = self.get_state_start_time()

Method: get_time_elapsed_in_state

get_time_elapsed_in_state(self)

Get the elapsed time since the state started.

Returns

• float: The elapsed time since the state started.

Raises

• Exception: If retrieving the elapsed time fails.

Usage

elapsed_time = self.get_time_elapsed_in_state()

State Class

A class designed to be used with the ProCaaSo State Machine

Initialization

Initialize a State object. All routines and transitions must be added manually through the class methods.

Parameters

• id (int | str): The state ID. Each state should only have one ID.
• onEnterEnabled (bool): Indicator for if a state should have an on enter function, default to false
• onExitEnabled (bool): Indicator for if a state should have an on exit function, default to false

Raises

• TypeError: If the provided ID is not of type 'int'.

Usage

state_instance = State(0)
state_instance = State(1, onEnterEnabled=True)
state_instance = State(2, onExitEnabled=True)
state_instance = State(3, onEnterEnabled=True, onExitEnabled=True)

Attributes

• __apartOfStateMachine (bool): A boolean to track if the state is a part of a state machine.
• __routines (dict): A dictionary containing the routines registered to run.
• __transitions (list): A list of all valid states to transition to.
• id (int): The state number. Each state should only have one ID.

Methods

Method: set_routine

set_routine(self, routineName: str, routine: Callable)

Set a routine with a given name in the object's routines dictionary. The order in which the routines are added determines the order in which they will be executed. If a routine name is already present in the states dictionary of routines, the routine will be updated, not added. Thus, routine names must be unique within a state object. It is required that all routines accept kwargs by default, such that on call the routines can be passed information.

Parameters

• routineName (str): The name of the routine.
• routine (Callable): The callable object representing the routine.

Raises

• TypeError: If routineName is not of type 'str' or routine is not of type 'Callable'.
• Exception: If an unexpected exception occurs during the process.
• Exception: If state is already a part of the state machine, raises an exception.

Usage

obj.set_routine("my_routine", my_callable_routine)

Method: get_routines

get_routines(self) -> dict

Get the dictionary containing the routines registered in the state.

Returns

• dict: The dictionary of routines.

Usage

routines_dict = obj.get_routines()

Method: set_transition

set_transition(self, transitionId: int)

Set a transition ID in the object's transitions list.

Parameters

• transitionId (int | str): The ID of the transition.

Raises

• TypeError: If transitionId is not of type 'int'.
• Exception: If an unexpected exception occurs during the process.
• Exception: If state is already a part of the state machine, raises an exception.

Usage

obj.set_transition(123)
obj.set_transition('nextStateId')

Method: get_transitions

get_transitions(self) -> List[int]

Get the list of valid state transitions from the current state.

Returns

• List[int | str]: The list of transition IDs.

Usage

transitions_list = obj.get_transitions()

Method: get_apartOfStateMachine

get_apartOfStateMachine(self) -> bool

Check if the state is a part of a state machine.

Returns

• bool: True if the state is a part of a StateMachine object, False otherwise.

Usage

isApartOfStateMachine = obj.get_apartOfStateMachine()

Method: get_id

get_id(self) -> int

Get the ID of the state.

Returns

• int: The state ID.

Usage

state_id = obj.get_id()

Method: get_on_enter_enabled

get_on_enter_enabled(self) -> bool

Retrieve the current state of the onEnterEnabled attribute.

Returns

• bool: The current value of the __onEnterEnabled attribute, indicating whether the onEnter functionality is enabled (True) or disabled (False).

Usage

on_enter_enabled = obj.get_on_enter_enabled()

Method: get_on_exit_enabled

get_on_exit_enabled(self) -> bool

Retrieve the current state of the onExitEnabled attribute.

Returns

• bool: The current value of the __onExitEnabled attribute, indicating whether the onExit functionality is enabled (True) or disabled (False).

Usage

on_exit_enabled = obj.get_on_exit_enabled()

Method: set_on_enter_routine

set_on_enter_routine(self, routine: Callable)

Set the routine to be executed when the state is entered, if onEnter functionality is enabled.

Parameters

• routine (Callable): The function to be executed on entering the state. This function must accept keyword arguments.

Raises

• Exception: If the state is already part of a state machine, the onEnter functionality is disabled, or the supplied routine does not accept keyword arguments.

Notes

• Ensures that the state is not yet part of a state machine.
• Checks if the onEnter functionality is enabled.
• Verifies that the supplied routine is a callable and accepts keyword arguments.

Usage

obj.set_on_enter_routine(my_callable_routine)

Method: get_on_enter_function

get_on_enter_function(self) -> Callable

Retrieve the routine to be executed when the state is entered.

Returns

• Callable: The function to be executed on entering the state, if onEnter functionality is enabled.

Raises

• Exception: If the onEnter functionality is not enabled, an exception is raised indicating that no function may be returned.

Usage

on_enter_function = obj.get_on_enter_function()

Method: set_on_exit_function

set_on_exit_function(self, routine: Callable)

Set the routine to be executed when the state is exited, if onExit functionality is enabled.

Parameters

• routine (Callable): The function to be executed on exiting the state. This function must accept keyword arguments.

Raises

• Exception: If the state is already part of a state machine, the onExit functionality is disabled, or the supplied routine does not accept keyword arguments.

Notes

• Ensures that the state is not yet part of a state machine.
• Checks if the onExit functionality is enabled.
• Verifies that the supplied routine is a callable and accepts keyword arguments.

Usage

obj.set_on_exit_function(my_callable_routine)

Method: get_on_exit_function

get_on_exit_function(self) -> Callable

Retrieve the routine to be executed when the state is exited.

Returns

• Callable: The function to be executed on exiting the state, if onExit functionality is enabled.

Raises

• Exception: If the onExit functionality is not enabled, an exception is raised indicating that no function may be returned.

Usage

on_exit_function = obj.get_on_exit_function()

Event Class

A class to be passed to the ProCaaSo state machine to help build event driven development

Initialization

__init__(self, eventType: str, actuations: Dict[str, Any] = {}) -> None

Initialize the Event object with the specified event type and optional actuations.

Parameters

• eventType (str): The type of the event.
• actuations (Dict[str, Any]): A dictionary containing actuations associated with the event. Default is an empty dictionary.

Raises

• Exception: If an unexpected exception occurs during initialization.

Usage

event = Event("eventType", {"key": "value"})

event = Event('Example Event', {'Actuation': 5})

Attributes

• set_actuations (dict[str,Any]): Dictionary containing each actuation.
• __set_eventType (str): A string with the type of the event.

Methods

Method: set_actuations

set_actuations(self, actuationsDict: Dict[str, Any])

Set the actuations for the event object, overwriting any previous actuations.

Parameters

• actuationsDict (Dict[str, Any]): A dictionary containing actuations.

Raises

• Exception: If the dictionary structure is invalid.

Usage

event.set_actuations({"key": "value"})

Method: get_actuations

get_actuations(self)

Get the actuations associated with the event.

Raises

• Exception: If failed to retrieve actuations.

Usage

actuations = event.get_actuations()

Method: get_eventType

get_eventType(self)

Get the event type associated with the event.

Raises

• Exception: If failed to retrieve the event type.

Usage

eventType = event.get_eventType()