State Machine

The StateMachine class provides a wrapper around the transitions library, adding timeout support and integration with pydoover's async patterns. Use state machines to model complex device behaviors with clear state definitions and transitions.

Import

from pydoover.state import StateMachine

Constructor

StateMachine(
    states: list[dict | str],
    transitions: list[dict],
    model: object = None,
    initial: str = None,
    queued: bool = False,
    **kwargs
)

Parameters

Parameter	Type	Default	Description
`states`	`list`	Required	State definitions (strings or dicts with metadata)
`transitions`	`list`	Required	Transition definitions between states
`model`	`object`	`None`	Object to attach state methods to (usually `self`)
`initial`	`str`	`None`	Starting state name
`queued`	`bool`	`False`	Queue transitions for thread-safe processing

State Definitions

States can be simple strings or dictionaries with additional configuration:

# Simple states
states = ["off", "on", "error"]

# States with metadata
states = [
    {"name": "off", "timeout": 5, "on_timeout": "set_on"},
    {"name": "on", "timeout": 10, "on_timeout": "set_off"},
    {"name": "error"}
]

State Properties

Property	Type	Description
`name`	`str`	State identifier (required)
`timeout`	`int`	Seconds before automatic transition
`on_timeout`	`str`	Trigger to fire when timeout expires

Transition Definitions

Transitions define how the state machine moves between states:

transitions = [
    {"trigger": "set_on", "source": "off", "dest": "on"},
    {"trigger": "set_off", "source": "on", "dest": "off"},
    {"trigger": "set_error", "source": "*", "dest": "error"},
    {"trigger": "clear_error", "source": "error", "dest": "off"},
]

Transition Properties

Property	Type	Description
`trigger`	`str`	Method name that triggers this transition
`source`	`str`	State(s) this transition can start from (`"*"` = any)
`dest`	`str`	Target state
`conditions`	`str` or `list`	Method(s) that must return True for transition
`unless`	`str` or `list`	Method(s) that must return False for transition
`before`	`str` or `list`	Method(s) to call before transition
`after`	`str` or `list`	Method(s) to call after transition

State Callbacks

Define methods on your model that are called when entering or exiting states:

class SampleState:
    states = [
        {"name": "off", "timeout": 5, "on_timeout": "set_on"},
        {"name": "on"}
    ]

    transitions = [
        {"trigger": "set_on", "source": "off", "dest": "on"},
        {"trigger": "set_off", "source": "on", "dest": "off"},
    ]

    def __init__(self):
        self.state_machine = StateMachine(
            states=self.states,
            transitions=self.transitions,
            model=self,
            initial="off",
            queued=True,
        )

    # Called when entering the "off" state
    async def on_enter_off(self):
        print("Entered off state")

    # Called when entering the "on" state
    async def on_enter_on(self):
        print("Entered on state")

    # Called when exiting the "on" state
    async def on_exit_on(self):
        print("Exiting on state")

Callback Naming Convention

Pattern	When Called
`on_enter_{state}`	When entering the state
`on_exit_{state}`	When exiting the state

Callbacks can be synchronous or asynchronous functions.

Triggering Transitions

After creating the state machine with a model, trigger methods are added to the model:

state = SampleState()

# Trigger transitions directly
state.set_on()   # Transitions from "off" to "on"
state.set_off()  # Transitions from "on" to "off"

# Check current state
print(state.state)  # "on" or "off"

# Check if in a specific state
if state.is_on():
    print("Currently in on state")

Timeouts

States with timeout and on_timeout properties automatically trigger transitions after the specified duration:

states = [
    {"name": "warming_up", "timeout": 30, "on_timeout": "ready"},
    {"name": "ready"},
    {"name": "cooling_down", "timeout": 60, "on_timeout": "idle"},
    {"name": "idle"}
]

transitions = [
    {"trigger": "ready", "source": "warming_up", "dest": "ready"},
    {"trigger": "start_cooldown", "source": "ready", "dest": "cooling_down"},
    {"trigger": "idle", "source": "cooling_down", "dest": "idle"},
]

In this example:

After entering warming_up, the machine automatically transitions to ready after 30 seconds
After entering cooling_down, the machine automatically transitions to idle after 60 seconds

Queued Transitions

Enable queued=True for thread-safe transition processing. This is recommended when transitions may be triggered from multiple async tasks:

self.state_machine = StateMachine(
    states=self.states,
    transitions=self.transitions,
    model=self,
    initial="off",
    queued=True,  # Safe for concurrent triggers
)

Real-World Example: Warning State Machine

This example from a Doover application template shows a state machine managing warning states with automatic timeouts:

from pydoover.state import StateMachine

class SampleState:
    states = [
        {"name": "warning_disabled", "timeout": 5, "on_timeout": "set_warning_enabled"},
        {"name": "warning_enabled", "timeout": 5, "on_timeout": "set_warning_disabled"},
    ]

    transitions = [
        {"trigger": "set_warning_enabled", "source": "warning_disabled", "dest": "warning_enabled"},
        {"trigger": "set_warning_disabled", "source": "warning_enabled", "dest": "warning_disabled"},
    ]

    def __init__(self):
        self.state_machine = StateMachine(
            states=self.states,
            transitions=self.transitions,
            model=self,
            initial="warning_disabled",
            queued=True,
        )

    async def on_enter_warning_disabled(self):
        print("Warning disabled - will enable in 5 seconds")

    async def on_enter_warning_enabled(self):
        print("Warning enabled - will disable in 5 seconds")

Source: app-template

Integration with Applications

State machines are typically initialized in the application's setup() method:

from pydoover.docker import Application
from .app_state import SampleState

class MyApplication(Application):
    async def setup(self):
        self.state = SampleState()

    async def main_loop(self):
        # Check state and respond accordingly
        if self.state.is_warning_enabled():
            self.ui.warning_indicator.hidden = False
        else:
            self.ui.warning_indicator.hidden = True

Conditional Transitions

Add conditions to transitions that must be met before the transition can occur:

class PumpState:
    states = ["stopped", "running", "error"]

    transitions = [
        {
            "trigger": "start",
            "source": "stopped",
            "dest": "running",
            "conditions": "is_safe_to_start"  # Must return True
        },
        {
            "trigger": "stop",
            "source": "running",
            "dest": "stopped"
        },
    ]

    def is_safe_to_start(self):
        """Check if conditions are safe for starting."""
        return self.temperature < 80 and not self.emergency_stop