state_machine – State Machine

State Machine - lightweight task scheduler and state machine framework.

class codeair.state_machine.State(state_machine, handler)

Represents a state in the state machine.

Initializes a State instance. (Used by StateMachine - not usually used directly by end-user code)

Parameters:

• state_machine (StateMachine) – The state machine this state belongs to.

• handler (function) – The function that handles state events.

enter()

Called when entering the state.

exit()

Called when exiting the state.

handle_event(event_type, event_data)

Handles events within the state.

Parameters:

• event_type (str) – The type of event.

• event_data (any) – The data associated with the event.

class codeair.state_machine.StateMachine

A simple state machine framework with task scheduling capabilities.

Initializes a StateMachine instance.

add_state(state_func)

Adds a state to the state machine. state_func must be a callable that takes four arguments:

def func(state_machine, event_type, event_data, st_time):
    pass

• The state_func() will be called on every next_state transition, with the event_type set to “enter” or “exit”.

• It will also be called when event() is called, with event_type set to the event type passed to event().

• The st_time argument is the elapsed time since the state was entered, in seconds.

Parameters:

state_func (function) – The function that handles the state.

add_task(callback, interval, args=None)

Adds a task to be run periodically at a specified interval.

Parameters:

• callback (function) – The function to be called.

• interval (float) – The interval in seconds between task executions.

• args (tuple, optional) – Arguments to pass to the callback function.

current()

Return currently active state function

enable_debug(do_debug, log_func=<built-in function print>)

Enables or disables debug logging.

Parameters:

• do_debug (bool) – If True, enables debug logging.

• log_func (function) – The function to use for logging. Defaults to print.

event(event_type, event_data=None)

Injects an event into the state machine, handled by the current state.

Parameters:

• event_type (str) – The type of event.

• event_data (any, optional) – The data associated with the event.

next_state(next_state_func, sched_delay=None)

Transitions to a new state.

Parameters:

• next_state_func (function) – The function that handles the next state.

• sched_delay (float) – Optional schedule delay (seconds) before transition. Use 0 to schedule ASAP and break recursion.

remove_task(callback)

Removes a task from the task list.

Parameters:

callback (function) – The function to be removed.

run_tasks()

Runs all scheduled tasks that are due.

schedule(callback, delay, args=None)

Schedules a task to be run once after a delay.

Parameters:

• callback (function) – The function to be called.

• delay (float) – The delay in seconds before the task is executed.

• args (tuple, optional) – Arguments to pass to the callback function.

start(first_state=None)

Starts or re-starts the state machine.

Parameters:

first_state (function, optional) – The function that handles the initial state. Defaults to the first state added.

sync_task(callback, ns_sync=None)

Set task periodic interval to align with ns_sync as basis. Default to now (time.monotonic_ns())

Parameters:

• callback (function) – The function to be synced.

• ns_sync (int, optional) – The time in nanoseconds to sync to. Defaults to None, which uses the current time.

Download Source