Utilities and Data Classes

from dataclasses import dataclass
from typing import Any

@dataclass
class TriggerData:
    """
    Container for event trigger information passed through the state machine.
    
    Attributes:
    - machine: StateMachine instance processing the event
    - event: Event object that was triggered
    - model: External model object (if provided)
    - args: Positional arguments passed to the event
    - kwargs: Keyword arguments passed to the event
    """
    machine: StateMachine
    event: Event
    model: Any = field(init=False)
    args: tuple = field(default_factory=tuple)
    kwargs: dict = field(default_factory=dict)

@dataclass
class EventData:
    """
    Complete event processing context containing trigger and transition information.
    
    Attributes:
    - trigger_data: TriggerData instance with event information
    - transition: Transition object being executed
    - state: Current state (computed from trigger_data)
    - source: Source state of transition (computed from transition)
    - target: Target state of transition (computed from transition)
    - result: Result value from callback execution
    - executed: Whether the transition has been executed
    """
    trigger_data: TriggerData
    transition: Transition
    state: State = field(init=False)
    source: State = field(init=False)
    target: State = field(init=False)
    result: Any | None = None
    executed: bool = False

class Events:
    """
    Collection class for managing Event objects in a state machine.
    
    Provides methods for adding events and matching event identifiers.
    """
    def __init__(self): ...
    
    def add(self, events):
        """
        Add events to the collection.
        
        Parameters:
        - events: Event object or iterable of Event objects
        """
    
    def match(self, event: str) -> bool:
        """
        Check if any event in collection matches the given event identifier.
        
        Parameters:
        - event: Event identifier string
        
        Returns:
        True if any event matches, False otherwise
        """
    
    def __iter__(self):
        """Iterate over events in the collection."""
    
    def __len__(self) -> int:
        """Get number of events in the collection."""

class Model:
    """
    Simple model class for external state storage.
    
    Provides a basic foundation for state machine model integration
    with automatic state field management.
    """
    def __init__(self):
        """Initialize model with state field."""
        self.state = None

def register(cls):
    """
    Register a StateMachine class for discovery.
    
    Parameters:
    - cls: StateMachine class to register
    
    Returns:
    The registered class (allows use as decorator)
    """

def get_machine_cls(name: str):
    """
    Get registered StateMachine class by name.
    
    Parameters:
    - name: Fully qualified class name
    
    Returns:
    StateMachine class
    
    Raises:
    ImportError: If class is not found or cannot be imported
    """

def init_registry():
    """Initialize the state machine registry."""

def load_modules(modules=None):
    """
    Load modules containing state machine definitions.
    
    Parameters:
    - modules: List of module names to load (optional)
    """

def qualname(cls) -> str:
    """
    Get fully qualified name of a class.
    
    Parameters:
    - cls: Class object
    
    Returns:
    Fully qualified class name string
    """

def ensure_iterable(obj):
    """
    Ensure object is iterable, wrapping single items in a tuple.
    
    Parameters:
    - obj: Object to make iterable
    
    Returns:
    Iterable version of the object
    """

def run_async_from_sync(coroutine):
    """
    Run async coroutine from synchronous context.
    
    Parameters:
    - coroutine: Async coroutine to execute
    
    Returns:
    Result of coroutine execution
    """

from statemachine import StateMachine, State

class ProcessingMachine(StateMachine):
    idle = State(initial=True)
    working = State()
    done = State(final=True)
    
    start = idle.to(working)
    finish = working.to(done)
    
    def on_start(self, event_data, trigger_data, **kwargs):
        """Access event data structures in callbacks."""
        print(f"Event: {trigger_data.event.id}")
        print(f"Machine: {trigger_data.machine.__class__.__name__}")
        print(f"Transition: {event_data.transition.source.id} -> {event_data.transition.target.id}")
        print(f"Args: {trigger_data.args}")
        print(f"Kwargs: {trigger_data.kwargs}")

# Usage
machine = ProcessingMachine()
machine.send("start", priority="high", task_id="T001")

from statemachine.model import Model

class OrderModel(Model):
    def __init__(self, order_id: str):
        super().__init__()
        self.order_id = order_id
        self.state = "pending"  # Initial state value
        self.created_at = datetime.now()

class OrderMachine(StateMachine):
    pending = State("Pending", initial=True, value="pending")
    paid = State("Paid", value="paid")
    shipped = State("Shipped", value="shipped")
    delivered = State("Delivered", final=True, value="delivered")
    
    pay = pending.to(paid)
    ship = paid.to(shipped)
    deliver = shipped.to(delivered)

# Integration
order_model = OrderModel("ORD-001")
machine = OrderMachine(order_model)

# State is automatically synced with model
machine.send("pay")
print(order_model.state)  # "paid"

from statemachine.registry import register, get_machine_cls

@register
class WorkflowMachine(StateMachine):
    start = State(initial=True)
    end = State(final=True)
    
    proceed = start.to(end)

# Later, dynamically load the class
machine_cls = get_machine_cls("__main__.WorkflowMachine")
instance = machine_cls()
print(f"Loaded: {instance.__class__.__name__}")

from statemachine.utils import qualname, ensure_iterable, run_async_from_sync

# Get fully qualified class name
name = qualname(OrderMachine)
print(name)  # "__main__.OrderMachine"

# Ensure iterable
items = ensure_iterable("single_item")
print(items)  # ("single_item",)

items = ensure_iterable(["already", "iterable"])
print(items)  # ["already", "iterable"]

# Run async from sync context
async def async_operation():
    await asyncio.sleep(0.1)
    return "completed"

result = run_async_from_sync(async_operation())
print(result)  # "completed"