Extensions and Specialized Machines

class MachineFactory:
    @staticmethod
    def get_predefined(graph=False, nested=False, locked=False, asyncio=False):
        """
        Retrieve machine classes by required functionality.
        
        Parameters:
        - graph: Include diagram generation support
        - nested: Include hierarchical state support
        - locked: Include thread safety
        - asyncio: Include async/await support (Python 3+ only)
        
        Returns:
        Machine class with requested capabilities
        """

class GraphMachine(Machine):
    """
    Machine extension with graph support for diagram generation.
    Inherits all Machine functionality plus diagram capabilities.
    """
    
    def get_graph(self, **kwargs):
        """
        Generate a graph representation of the state machine.
        
        Returns:
        Graph object that can be rendered to various formats
        """

class HierarchicalMachine(Machine):
    """
    Machine extension for hierarchical/nested state machines.
    Supports nested states with entry/exit semantics and state inheritance.
    """

class NestedState(State):
    """State extension for hierarchical machines with nesting support."""

class NestedTransition(Transition):
    """Transition extension for hierarchical machines."""

class NestedEvent(Event):
    """Event extension for nested states (NOT compatible with simple Machine instances)."""

class NestedEventData(EventData):
    """EventData extension for nested state machines."""

def resolve_order(state_tree):
    """
    Convert a model state tree into a list of state paths for correct processing order.
    
    Parameters:
    - state_tree: Nested state structure
    
    Returns:
    list: Ordered list of state paths
    """

class LockedMachine(Machine):
    """
    Machine class which manages contexts for thread safety.
    Uses threading.RLock to ensure atomic state transitions.
    """

class PicklableLock:
    """A wrapper for threading.Lock which discards state during pickling."""

class IdentManager:
    """Manages thread identity to detect if current thread already has lock."""

class LockedEvent(Event):
    """Event type which uses parent's machine context map when triggered."""

class AsyncMachine(Machine):
    """
    Machine extension with async/await support.
    Enables asynchronous callback execution and event processing.
    """
    
    async def dispatch(self, trigger, *args, **kwargs):
        """
        Asynchronously trigger an event on all models.
        
        Parameters:
        - trigger: Name of the trigger method
        - args: Positional arguments
        - kwargs: Keyword arguments
        
        Returns:
        bool: True if at least one transition was successful
        """

class AsyncState(State):
    """State extension for asynchronous state machines."""

class AsyncTransition(Transition):
    """Transition extension for async callback support."""

class AsyncEvent(Event):
    """Event extension for async callback support."""

class AsyncEventData(EventData):
    """EventData extension for async machines."""

class AsyncCondition:
    """Condition extension for async callback support."""

class AsyncTimeout(AsyncState):
    """AsyncState extension with timeout functionality."""

class HierarchicalAsyncMachine(HierarchicalMachine, AsyncMachine):
    """Combination of HierarchicalMachine and AsyncMachine."""

class NestedAsyncState(NestedState, AsyncState):
    """Combination of NestedState and AsyncState."""

class NestedAsyncTransition(AsyncTransition, NestedTransition):
    """Combination of AsyncTransition and NestedTransition."""

class NestedAsyncEvent(NestedEvent):
    """Combination of NestedEvent for async processing."""

class MarkupMachine(Machine):
    """Machine extension that can be configured with dictionaries/markup."""

class HierarchicalMarkupMachine(MarkupMachine, HierarchicalMachine):
    """Combination of MarkupMachine and HierarchicalMachine."""

class Tags(State):
    """State extension that allows tags to be assigned to states."""

class Error(State):
    """State extension for error handling with automatic error clearing."""

class Timeout(State):
    """State extension with timeout functionality."""

class Volatile(State):
    """State extension that is automatically cleared after execution."""

class Retry(State):
    """State extension with retry functionality."""

class VolatileObject:
    """Utility class for volatile state management."""

def add_state_features(*args):
    """
    Convenience function to combine multiple state features.
    
    Parameters:
    - args: State feature classes to combine
    
    Returns:
    Combined state class
    """

class LockedHierarchicalMachine(LockedMachine, HierarchicalMachine):
    """A threadsafe hierarchical machine."""

class LockedGraphMachine(LockedMachine, GraphMachine):
    """A threadsafe machine with graph support."""

class LockedHierarchicalGraphMachine(LockedHierarchicalMachine, GraphMachine):
    """A threadsafe hierarchical machine with graph support."""

class AsyncGraphMachine(AsyncMachine, GraphMachine):
    """A machine that supports asynchronous event/callback processing with Graphviz support."""

class HierarchicalAsyncGraphMachine(HierarchicalAsyncMachine, GraphMachine):
    """A hierarchical machine that supports asynchronous event/callback processing with Graphviz support."""

from transitions.extensions import MachineFactory

# Get a machine with multiple capabilities
MachineClass = MachineFactory.get_predefined(graph=True, nested=True, locked=True)
machine = MachineClass(model=my_model, states=states, transitions=transitions)

from transitions.extensions import GraphMachine

class Robot:
    pass

states = ['idle', 'working', 'maintenance']
transitions = [
    {'trigger': 'start', 'source': 'idle', 'dest': 'working'},
    {'trigger': 'break_down', 'source': 'working', 'dest': 'maintenance'},
    {'trigger': 'repair', 'source': 'maintenance', 'dest': 'idle'}
]

robot = Robot()
machine = GraphMachine(model=robot, states=states, transitions=transitions, initial='idle')

# Generate a diagram
graph = machine.get_graph()
# graph.draw('robot_states.png', prog='dot')  # Requires graphviz

from transitions.extensions import HierarchicalMachine

class Player:
    pass

states = [
    {'name': 'alive', 'children': ['healthy', 'injured']},
    'dead'
]

transitions = [
    {'trigger': 'take_damage', 'source': 'alive_healthy', 'dest': 'alive_injured'},
    {'trigger': 'heal', 'source': 'alive_injured', 'dest': 'alive_healthy'},
    {'trigger': 'die', 'source': 'alive', 'dest': 'dead'},
    {'trigger': 'resurrect', 'source': 'dead', 'dest': 'alive_healthy'}
]

player = Player()
machine = HierarchicalMachine(
    model=player, 
    states=states, 
    transitions=transitions, 
    initial='alive_healthy'
)

print(player.state)  # 'alive_healthy'
player.take_damage()
print(player.state)  # 'alive_injured'

import asyncio
from transitions.extensions import AsyncMachine

class AsyncRobot:
    async def on_enter_working(self):
        print("Starting work...")
        await asyncio.sleep(1)  # Simulate work
        print("Work completed!")

states = ['idle', 'working']
transitions = [
    {'trigger': 'start_work', 'source': 'idle', 'dest': 'working'},
    {'trigger': 'finish_work', 'source': 'working', 'dest': 'idle'}
]

async def main():
    robot = AsyncRobot()
    machine = AsyncMachine(model=robot, states=states, transitions=transitions, initial='idle')
    
    await robot.start_work()  # Asynchronously transitions to working state

# asyncio.run(main())

import threading
from transitions.extensions import LockedMachine

class SharedResource:
    def __init__(self):
        self.data = 0

states = ['idle', 'processing', 'complete']
transitions = [
    {'trigger': 'start', 'source': 'idle', 'dest': 'processing'},
    {'trigger': 'finish', 'source': 'processing', 'dest': 'complete'},
    {'trigger': 'reset', 'source': 'complete', 'dest': 'idle'}
]

resource = SharedResource()
machine = LockedMachine(model=resource, states=states, transitions=transitions, initial='idle')

# Safe to use from multiple threads
def worker():
    resource.start()
    # Process data...
    resource.finish()

# Multiple threads can safely interact with the state machine
threads = [threading.Thread(target=worker) for _ in range(5)]
for t in threads:
    t.start()