A lightweight, object-oriented Python state machine implementation with many extensions
Individual components for states, transitions, events, and conditions that can be used to build custom state machine configurations. These components provide fine-grained control over state machine behavior and enable advanced customization.
Represents a persistent state managed by a Machine with entry/exit callbacks and trigger handling.
class State:
def __init__(
self,
name,
on_enter=None,
on_exit=None,
ignore_invalid_triggers=None,
final=False
):
"""
Initialize a state.
Parameters:
- name: The name of the state (str or Enum)
- on_enter: Callable(s) triggered when state is entered
- on_exit: Callable(s) triggered when state is exited
- ignore_invalid_triggers: Override machine's ignore_invalid_triggers setting
- final: Whether this is a final state
"""
def enter(self, event_data):
"""
Triggered when a state is entered.
Parameters:
- event_data: EventData object containing transition context
"""
def exit(self, event_data):
"""
Triggered when a state is exited.
Parameters:
- event_data: EventData object containing transition context
"""
def add_callback(self, trigger, func):
"""
Add a new enter or exit callback.
Parameters:
- trigger: 'enter' or 'exit'
- func: Callback function to add
"""
@property
def name(self):
"""The name of the state."""
@property
def value(self):
"""The state's value (equivalent to name for string states)."""Represents a transition between states with conditions, callbacks, and execution logic.
class Transition:
def __init__(
self,
source,
dest,
conditions=None,
unless=None,
before=None,
after=None,
prepare=None
):
"""
Initialize a transition.
Parameters:
- source: Source state name
- dest: Destination state name
- conditions: Callbacks that must return True for transition to execute
- unless: Callbacks that must return False for transition to execute
- before: Callbacks executed before the transition
- after: Callbacks executed after the transition
- prepare: Callbacks executed before condition checks
"""
def execute(self, event_data):
"""
Execute the transition if conditions are met.
Parameters:
- event_data: EventData object containing transition context
Returns:
bool: True if transition was executed, False otherwise
"""
def add_callback(self, trigger, func):
"""
Add a new before, after, or prepare callback.
Parameters:
- trigger: 'before', 'after', or 'prepare'
- func: Callback function to add
"""Manages a collection of transitions assigned to the same trigger and handles event execution.
class Event:
def __init__(self, name, machine):
"""
Initialize an event.
Parameters:
- name: The name of the event/trigger
- machine: The Machine instance this event belongs to
"""
def add_transition(self, transition):
"""
Add a transition to the list of potential transitions.
Parameters:
- transition: Transition object to add
"""
def trigger(self, model, *args, **kwargs):
"""
Execute all transitions that match the current state.
Parameters:
- model: The model object to trigger on
- args: Positional arguments passed to callbacks
- kwargs: Keyword arguments passed to callbacks
Returns:
bool: True if at least one transition was successful
"""
def add_callback(self, trigger, func):
"""
Add a new before or after callback to all available transitions.
Parameters:
- trigger: 'before' or 'after'
- func: Callback function to add
"""Container for relevant data related to an ongoing transition attempt, passed to all callbacks.
class EventData:
def __init__(self, state, event, machine, model, args, kwargs):
"""
Initialize event data.
Parameters:
- state: The State from which the Event was triggered
- event: The triggering Event
- machine: The current Machine instance
- model: The model/object the machine is bound to
- args: Optional positional arguments from trigger method
- kwargs: Optional keyword arguments from trigger method
"""
def update(self, state):
"""
Update the EventData object with a new state.
Parameters:
- state: New State object
"""
# Attributes available on EventData instances:
state: State # The State from which the Event was triggered
event: Event # The triggering Event
machine: Machine # The current Machine instance
model: object # The model/object the machine is bound to
args: list # Optional positional arguments from trigger method
kwargs: dict # Optional keyword arguments from trigger method
transition: Transition # Currently active transition
error: Exception # In case a triggered event causes an Error
result: bool # True if transition successful, False otherwiseHelper class for condition checks with target validation support.
class Condition:
def __init__(self, func, target=True):
"""
Initialize a condition.
Parameters:
- func: The function to call for the condition check (str or callable)
- target: Indicates the target state for condition validation
"""
def check(self, event_data):
"""
Check whether the condition passes.
Parameters:
- event_data: EventData object containing transition context
Returns:
bool: True if condition passes, False otherwise
"""from transitions import State, Machine
# Create states with callbacks
def on_enter_working():
print("Started working!")
def on_exit_working():
print("Stopped working!")
working_state = State(
name='working',
on_enter=on_enter_working,
on_exit=on_exit_working
)
# Use in machine
states = ['idle', working_state, 'done']
machine = Machine(model=MyModel(), states=states, initial='idle')from transitions import Machine, Transition
class Robot:
def __init__(self):
self.battery = 100
self.task_queue = []
def has_battery(self):
return self.battery > 20
def has_tasks(self):
return len(self.task_queue) > 0
def start_task(self):
if self.task_queue:
task = self.task_queue.pop(0)
print(f"Starting task: {task}")
self.battery -= 10
# Create custom transitions with multiple conditions
states = ['idle', 'working', 'charging']
transitions = [
{
'trigger': 'start_work',
'source': 'idle',
'dest': 'working',
'conditions': ['has_battery', 'has_tasks'],
'before': 'start_task'
},
{
'trigger': 'finish_work',
'source': 'working',
'dest': 'idle'
},
{
'trigger': 'charge',
'source': ['idle', 'working'],
'dest': 'charging',
'unless': lambda: self.battery > 90 # Don't charge if battery is high
}
]
robot = Robot()
robot.task_queue = ['task1', 'task2', 'task3']
machine = Machine(model=robot, states=states, transitions=transitions, initial='idle')
# Test the conditions
robot.start_work() # Will work if battery > 20 and tasks availablefrom transitions import Machine
class LoggingRobot:
def __init__(self):
self.activity_log = []
def log_transition(self, event_data):
"""Log all transition details"""
log_entry = {
'from_state': event_data.state.name,
'to_state': event_data.transition.dest,
'trigger': event_data.event.name,
'timestamp': time.time(),
'args': event_data.args,
'kwargs': event_data.kwargs
}
self.activity_log.append(log_entry)
print(f"Transition: {log_entry['from_state']} -> {log_entry['to_state']} via {log_entry['trigger']}")
states = ['idle', 'working', 'done']
transitions = [
{
'trigger': 'start',
'source': 'idle',
'dest': 'working',
'after': 'log_transition'
},
{
'trigger': 'finish',
'source': 'working',
'dest': 'done',
'after': 'log_transition'
}
]
robot = LoggingRobot()
machine = Machine(
model=robot,
states=states,
transitions=transitions,
initial='idle',
send_event=True # This ensures EventData is passed to callbacks
)
robot.start(task_id=123, priority='high') # Arguments will be in event_data
robot.finish(result='success')from transitions import Machine, Event
class CustomMachine(Machine):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# Override event creation to add custom behavior
def _create_event(self, trigger, machine):
"""Create custom event with additional logging"""
event = Event(trigger, machine)
# Add custom behavior to all events
original_trigger = event.trigger
def logged_trigger(model, *args, **kwargs):
print(f"Event '{trigger}' triggered on {model}")
result = original_trigger(model, *args, **kwargs)
print(f"Event result: {result}")
return result
event.trigger = logged_trigger
return event
# This machine will log all event triggers
machine = CustomMachine(model=MyModel(), states=['A', 'B'], transitions=[
{'trigger': 'go', 'source': 'A', 'dest': 'B'}
], initial='A')from transitions import Machine, Condition
class SmartThermostat:
def __init__(self):
self.temperature = 20
self.target_temp = 22
self.hvac_enabled = True
def too_cold(self):
return self.temperature < self.target_temp - 1
def too_hot(self):
return self.temperature > self.target_temp + 1
def hvac_available(self):
return self.hvac_enabled
def start_heating(self):
print(f"Starting heating. Current: {self.temperature}°C, Target: {self.target_temp}°C")
def start_cooling(self):
print(f"Starting cooling. Current: {self.temperature}°C, Target: {self.target_temp}°C")
states = ['idle', 'heating', 'cooling', 'error']
transitions = [
{
'trigger': 'check_temperature',
'source': 'idle',
'dest': 'heating',
'conditions': ['too_cold', 'hvac_available'],
'before': 'start_heating'
},
{
'trigger': 'check_temperature',
'source': 'idle',
'dest': 'cooling',
'conditions': ['too_hot', 'hvac_available'],
'before': 'start_cooling'
},
{
'trigger': 'check_temperature',
'source': ['heating', 'cooling'],
'dest': 'idle',
'unless': ['too_cold', 'too_hot'] # Return to idle if temperature is right
},
{
'trigger': 'hvac_failure',
'source': ['heating', 'cooling'],
'dest': 'error'
}
]
thermostat = SmartThermostat()
machine = Machine(model=thermostat, states=states, transitions=transitions, initial='idle')
# Simulate temperature changes
thermostat.temperature = 18 # Too cold
thermostat.check_temperature() # Should start heating
thermostat.temperature = 22 # Right temperature
thermostat.check_temperature() # Should return to idleInstall with Tessl CLI
npx tessl i tessl/pypi-transitions