from xstate_statemachine import create_machine, Interpreter, State  # etc.

from xstate_statemachine import create_machine

config = {
    "id": "toggle",
    "initial": "inactive",
    "states": {
        "inactive": {"on": {"TOGGLE": "active"}},
        "active":   {"on": {"TOGGLE": "inactive"}},
    },
}
machine = create_machine(config)

from xstate_statemachine import create_machine, MachineLogic

def log_toggle(interpreter, context, event, action_def):
    print(f"Toggled! Count: {context['count']}")

logic = MachineLogic(actions={"logToggle": log_toggle})

config = {
    "id": "toggle",
    "initial": "off",
    "states": {
        "off": {"on": {"FLIP": {"target": "on", "actions": "logToggle"}}},
        "on":  {"on": {"FLIP": {"target": "off", "actions": "logToggle"}}},
    },
}
machine = create_machine(config, logic=logic)

# file: my_logic.py
def log_toggle(interpreter, context, event, action_def):
    context["count"] += 1

# file: main.py
import my_logic
from xstate_statemachine import create_machine

machine = create_machine(config, logic_modules=[my_logic])
# or by string path:
machine = create_machine(config, logic_modules=["my_logic"])

from xstate_statemachine import create_machine

class ToggleLogic:
    def log_toggle(self, interpreter, context, event, action_def):
        context["count"] += 1

machine = create_machine(config, logic_providers=[ToggleLogic()])

from xstate_statemachine import State, build_machine, action

idle    = State("idle", initial=True)
running = State("running")
done    = State("done", final=True)

@action
def log_start(interpreter, context, event, action_def):
    print("Machine started!")

machine = build_machine(
    id="workflow",
    states=[idle, running, done],
    transitions=[
        idle.to(running, event="START", actions=["logStart"]),
        running.to(done, event="FINISH"),
    ],
    actions=[log_start],
    context={"step": 0},
)

State(
    name: str = "",
    *,
    initial: bool = False,
    final: bool = False,
    parallel: bool = False,
    history: Optional[str] = None,
    on: Optional[Dict[str, Any]] = None,
    entry: Optional[List[str]] = None,
    exit: Optional[List[str]] = None,
    after: Optional[Dict[Union[int, str], Any]] = None,
    invoke: Optional[Union[Dict, List]] = None,
    on_done: Optional[Union[str, Dict]] = None,
    always: Optional[Union[str, Dict, List]] = None,
    context: Optional[Dict] = None,
    states: Optional[List[State]] = None,
    tags: Optional[List[str]] = None,
    meta: Optional[Dict[str, Any]] = None,
)

state.to(
    target: State,
    *,
    event: Optional[str] = None,    # REQUIRED -- event name
    guard: Optional[str] = None,    # guard function name
    actions: Optional[List[str]] = None,  # action names to run
    reenter: bool = False,          # force exit/re-entry on self-transitions
) -> Transition

idle = State("idle", initial=True)
active = State("active")

t = idle.to(active, event="ACTIVATE", actions=["logActivation"])

state.internal(
    event: str,
    *,
    guard: Optional[str] = None,
    actions: Optional[List[str]] = None,
) -> Transition

counter = State("counter", initial=True)
t = counter.internal("INCREMENT", actions=["addOne"])

class MyMachine(StateMachine):
    idle = State(initial=True)

    @idle.enter
    def on_idle_entered(self, interpreter, context, event, action_def):
        print("Entered idle!")

class MyMachine(StateMachine):
    active = State()

    @active.exit
    def on_active_exited(self, interpreter, context, event, action_def):
        print("Left active!")

class MyMachine(StateMachine):
    class loading(State, initial=True):
        """A compound initial state defined as a class."""
        fetching = State(initial=True)
        parsing  = State()

    class done(State, final=True):
        """A final state."""
        pass

    class regions(State, parallel=True):
        """A parallel state."""
        pass

from xstate_statemachine import StateMachine, State, action

class TrafficLight(StateMachine):
    machine_id = "trafficLight"
    initial_context = {"cycle_count": 0}

    red    = State(initial=True)
    yellow = State()
    green  = State()

    cycle = (
        red.to(green, event="NEXT")
        | green.to(yellow, event="NEXT")
        | yellow.to(red, event="NEXT", actions=["countCycle"])
    )

    @action
    def count_cycle(self, interpreter, context, event, action_def):
        context["cycle_count"] += 1

machine = TrafficLight.create_machine()

# With a context override:
machine = TrafficLight.create_machine(context={"cycle_count": 10})

MachineBuilder(machine_id: str)

.state(
    name: str,
    *,
    initial: bool = False,
    final: bool = False,
    parallel: bool = False,
    on: Optional[Dict] = None,
    entry: Optional[List] = None,
    exit: Optional[List] = None,
    after: Optional[Dict] = None,
    invoke: Optional[Union[Dict, List]] = None,
    on_done: Optional[Union[str, Dict]] = None,
    always: Optional[Union[str, Dict, List]] = None,
    history: Optional[str] = None,
    tags: Optional[List[str]] = None,
    meta: Optional[Dict[str, Any]] = None,
) -> MachineBuilder

.transition(
    source: str,
    event: str,
    target: str,
    *,
    guard: Optional[str] = None,
    actions: Optional[List[str]] = None,
    reenter: bool = False,
    internal: bool = False,
) -> MachineBuilder

.child_states(
    parent: str,
    *,
    initial: Optional[str] = None,
    states: Optional[Dict[str, Dict]] = None,
    parallel: bool = False,
) -> MachineBuilder

.build(context: Optional[Dict] = None) -> MachineNode

from xstate_statemachine import MachineBuilder

def log_it(interpreter, context, event, action_def):
    context["count"] += 1
    print(f"Count: {context['count']}")

machine = (
    MachineBuilder("counter")
    .context({"count": 0})
    .state("idle", initial=True)
    .state("counting")
    .state("done", final=True)
    .transition("idle", "START", "counting")
    .transition("counting", "INCREMENT", "counting",
                actions=["logIt"], reenter=True)
    .transition("counting", "FINISH", "done")
    .action("logIt", log_it)
    .build()
)

idle = State("idle", initial=True)
premium = State("premium")
basic = State("basic")

# Multiple guarded transitions for the same event
signup = (
    idle.to(premium, event="SIGNUP", guard="isPremium")
    | idle.to(basic, event="SIGNUP")
)
# signup is a TransitionGroup with 2 transitions

group_a = idle.to(s1, event="GO", guard="isA") | idle.to(s2, event="GO", guard="isB")
group_b = idle.to(s3, event="GO")

# Combine groups
all_transitions = group_a | group_b
# all_transitions is a TransitionGroup with 3 transitions

transition(
    source: State,
    event: str,
    target: State,
    *,
    guard: Optional[str] = None,
    actions: Optional[List[str]] = None,
    reenter: bool = False,
    internal: bool = False,
) -> Transition

from xstate_statemachine import State, transition

idle   = State("idle", initial=True)
active = State("active")

# These two are equivalent:
t1 = idle.to(active, event="START")
t2 = transition(idle, "START", active)

# With all options:
t3 = transition(
    idle, "START", active,
    guard="isReady",
    actions=["logStart", "initProcess"],
    reenter=False,
    internal=False,
)

Interpreter(
    machine: MachineNode,
    input: Optional[Any] = None,
    clock: Optional[Clock] = None,
    max_queue_size: Optional[int] = None,
    overflow_policy: OverflowPolicy = OverflowPolicy.RAISE,
    strict: Optional[bool] = None,
    service_executor: Optional[concurrent.futures.Executor] = None,
    service_pool_size: int = DEFAULT_SERVICE_POOL_SIZE,
)

import asyncio
from xstate_statemachine import create_machine, Interpreter, MachineLogic

config = {
    "id": "lightSwitch",
    "initial": "off",
    "context": {"flips": 0},
    "states": {
        "off": {"on": {"TOGGLE": {"target": "on", "actions": "countFlip"}}},
        "on":  {"on": {"TOGGLE": {"target": "off", "actions": "countFlip"}}},
    },
}

def count_flip(interpreter, context, event, action_def):
    context["flips"] += 1

logic = MachineLogic(actions={"countFlip": count_flip})
machine = create_machine(config, logic=logic)

async def main():
    service = await Interpreter(machine).start()

    await service.send("TOGGLE")          # off -> on
    await asyncio.sleep(0.05)             # let event loop process
    print(service.current_state_ids)      # {"lightSwitch.on"}
    print(service.context)                # {"flips": 1}

    await service.send("TOGGLE")          # on -> off
    await asyncio.sleep(0.05)
    print(service.context)                # {"flips": 2}

    await service.stop()

asyncio.run(main())

SyncInterpreter(
    machine: MachineNode,
    input: Optional[Any] = None,
    clock: Optional[Clock] = None,
    strict: Optional[bool] = None,
)

from xstate_statemachine import create_machine, SyncInterpreter, MachineLogic

config = {
    "id": "door",
    "initial": "closed",
    "states": {
        "closed": {"on": {"OPEN": "opened"}},
        "opened": {"on": {"CLOSE": "closed"}},
    },
}

machine = create_machine(config)
interpreter = SyncInterpreter(machine).start()

print(interpreter.current_state_ids)   # {"door.closed"}
interpreter.send("OPEN")
print(interpreter.current_state_ids)   # {"door.opened"}
interpreter.send("CLOSE")
print(interpreter.current_state_ids)   # {"door.closed"}

# Snapshot round-trip
snapshot = interpreter.get_snapshot()
restored = SyncInterpreter.from_snapshot(snapshot, machine)
print(restored.current_state_ids)      # {"door.closed"}

interpreter.stop()

from xstate_statemachine import SyncInterpreter, LoggingInspector

interpreter = (
    SyncInterpreter(machine)
    .use(LoggingInspector())
    .start()
)
# All events, transitions, and actions will be logged

@runtime_checkable
class Clock(Protocol):
    def now(self) -> float: ...
    def set_timeout(self, fn, delay_sec: float, *, owner: Any = None, sync: bool | None = None) -> Any: ...
    def clear_timeout(self, handle: Any) -> None: ...
    def pump(self) -> int: ...

from xstate_statemachine import Interpreter, SyncInterpreter, SimulatedClock

# Sync engine
clock = SimulatedClock()
interp = SyncInterpreter(machine, clock=clock).start()
clock.increment(30_000)   # fires a 30s `after` transition synchronously

# Async engine
async def main():
    clock = SimulatedClock()
    service = await Interpreter(machine, clock=clock).start()
    await clock.increment(30_000)   # fires the 30s `after`, settles the loop

@dataclass(frozen=True)
class Event:
    type: str
    payload: Dict[str, Any] = field(default_factory=dict)

e = Event(type="CLICK", payload={"x": 10})
e.type = "OTHER"  # Raises dataclasses.FrozenInstanceError

e.payload["y"] = 20  # This works -- dict mutation, not field reassignment

from xstate_statemachine import Event

# Simple event
click = Event(type="CLICK")

# Event with payload
login = Event(type="LOGIN", payload={"username": "alice", "role": "admin"})

# Access payload via .data alias
print(login.data["username"])  # "alice"

interpreter.send("CLICK")                          # Event(type="CLICK")
interpreter.send("LOGIN", username="alice")         # Event(type="LOGIN", payload={"username": "alice"})
interpreter.send({"type": "CLICK", "x": 10})       # Event(type="CLICK", payload={"x": 10})

class ActionDefinition:
    def __init__(self, config: Union[str, Dict[str, Any]]): ...

ad = ActionDefinition("myAction")
# ad.type == "myAction", ad.params == None

ad = ActionDefinition({"type": "myAction", "params": {"delay": 100}})
# ad.type == "myAction", ad.params == {"delay": 100}

class DoneEvent(NamedTuple):
    type: str
    data: Any
    src: str

def save_result(interpreter, context, event, action_def):
    # event is a DoneEvent when handling onDone
    context["result"] = event.data

class ErrorEvent(NamedTuple):
    type: str
    error: BaseException
    src: str

from xstate_statemachine import ErrorEvent

def store_error(interpreter, context, event, action_def):
    assert isinstance(event, ErrorEvent)
    context["error"] = str(event.error)

class AfterEvent(NamedTuple):
    type: str
    scheduled_for: float = 0.0   # clock time the deadline was set for
    fired_at: float = 0.0        # clock time the timer actually fired

    @property
    def lateness_ms(self) -> float: ...

def is_system_event(event: Any) -> bool: ...
def system_event(event_type: str, **payload: Any) -> Event: ...

ENGINE_EVENT_SHAPES: Tuple[str, ...]   # ("done.invoke.", "done.state.", "error.platform.", "after.", "xstate.", "___xstate")
SYSTEM_EVENT_PREFIXES: Tuple[str, ...]  # coarser legacy prefixes, kept for compatibility

from xstate_statemachine import Event, is_system_event, system_event

is_system_event(Event("done.review"))                            # False
is_system_event(system_event("___xstate_statemachine_init___"))  # True

class Receipt(NamedTuple):
    state_ids: FrozenSet[str]
    changed: bool
    error: Optional[BaseException] = None
    deferred: bool = False
    denied: bool = False

receipt = await interpreter.send("SUBMIT", wait=True)
if receipt.error is not None:
    print(f"SUBMIT did not run cleanly: {receipt.error}")

class OverflowPolicy(str, Enum):
    RAISE = "raise"
    BLOCK = "block"
    DROP_NEWEST = "drop_newest"

from xstate_statemachine import Interpreter, OverflowPolicy

service = await Interpreter(
    machine, max_queue_size=1000, overflow_policy=OverflowPolicy.DROP_NEWEST
).start()

class PendingInvocation(NamedTuple):
    state_id: str
    invoke_id: str
    src: str

@action
def my_action_name(interpreter, context, event, action_def):
    ...

@action("customActionName")
def whatever(interpreter, context, event, action_def):
    ...

(interpreter: BaseInterpreter, context: Dict, event: Event, action_def: ActionDefinition) -> None

from xstate_statemachine import action

# Auto-named: registered as "incrementCounter"
@action
def increment_counter(interpreter, context, event, action_def):
    context["count"] += 1

# Explicitly named: registered as "logIt"
@action("logIt")
def my_logger(interpreter, context, event, action_def):
    print(f"Event: {event.type}, Context: {context}")

@guard
def is_valid(context, event):
    return context.get("valid", False)

@guard("canProceed")
def check_proceed(context, event):
    return context["step"] > 0

(context: Dict, event: Event) -> bool

from xstate_statemachine import guard

# Auto-named: registered as "hasBalance"
@guard
def has_balance(context, event):
    return context["balance"] > 0

# Explicitly named
@guard("isAdmin")
def check_admin(context, event):
    return event.payload.get("role") == "admin"

# INVALID -- raises NotSupportedError immediately:
# @guard
# async def async_guard(context, event):
#     return True

@service
def fetch_user(interpreter, context, event):
    return {"name": "Alice", "id": 42}

@service("loadData")
async def load(interpreter, context, event):
    # async services only work with the async Interpreter
    data = await some_api_call()
    return data

(interpreter: BaseInterpreter, context: Dict, event: Event) -> Any

from xstate_statemachine import service

# Auto-named: registered as "fetchData"
@service
def fetch_data(interpreter, context, event):
    import requests
    return requests.get("https://api.example.com/data").json()

# Explicitly named, async
@service("processOrder")
async def process(interpreter, context, event):
    await asyncio.sleep(1)  # simulate work
    return {"status": "processed", "order_id": context["order_id"]}

class MachineLogic(Generic[TContext]):
    def __init__(
        self,
        actions: Optional[Mapping[str, ActionCallable]] = None,
        guards: Optional[Mapping[str, GuardCallable]] = None,
        services: Optional[Mapping[str, Union[ServiceCallable, MachineNode]]] = None,
        delays: Optional[Mapping[str, Union[int, float, DelayCallable]]] = None,
    ) -> None: ...

# The callable blueprints pin ARITY and the guard's bool return:
ActionCallable  = Callable[[Any, Any, Any, ActionDefinition], Union[None, Awaitable[None]]]
GuardCallable   = Callable[[Any, Any], bool]
ServiceCallable = Callable[[Any, Any, Any], Any]
DelayCallable   = Callable[[Any, Any], Union[int, float]]

from xstate_statemachine import MachineLogic

class MyLogic(MachineLogic):
    def __init__(self):
        super().__init__(
            actions={
                "logEvent":  self._log_event,
                "increment": self._increment,
            },
            guards={
                "isReady": self._is_ready,
            },
        )

    def _log_event(self, interpreter, context, event, action_def):
        print(f"Event: {event.type}")

    def _increment(self, interpreter, context, event, action_def):
        context["count"] += 1

    @staticmethod
    def _is_ready(context, event):
        return context.get("ready", False)

from xstate_statemachine import MachineLogic

logic = MachineLogic(
    actions={
        "increment": lambda i, ctx, e, a: ctx.update({"count": ctx["count"] + 1}),
    },
    guards={
        "isPositive": lambda ctx, e: ctx["count"] > 0,
    },
    services={
        "fetchData": my_fetch_function,
    },
)

LogicLoader.get_instance() -> LogicLoader

loader = LogicLoader.get_instance()
loader.register_logic_module(my_module)

from xstate_statemachine import create_machine, LogicLoader
import my_actions_module

# Option A: Register globally (affects all future create_machine calls)
loader = LogicLoader.get_instance()
loader.register_logic_module(my_actions_module)
machine = create_machine(config)  # auto-discovers from registered modules

# Option B: Pass modules per-call
machine = create_machine(config, logic_modules=[my_actions_module])

# Option C: Pass by import string
machine = create_machine(config, logic_modules=["my_app.actions"])

# Option D: Use provider instances
class MyProvider:
    def increment_counter(self, interpreter, context, event, action_def):
        context["count"] += 1

machine = create_machine(config, logic_providers=[MyProvider()])

machine.get_state_by_id(state_id: str) -> Optional[StateNode]

machine = create_machine(config)
node = machine.get_state_by_id("myMachine.active.loading")
if node:
    print(f"Found: {node.id}, type: {node.type}")

machine.get_next_state(from_state_id: str, event: Event) -> Optional[Set[str]]

from xstate_statemachine import Event

targets = machine.get_next_state("myMachine.idle", Event(type="START"))
assert targets == {"myMachine.running"}

machine.structure_hash -> str

print(machine.structure_hash)  # e.g. "f21b173044383a6d"

puml = machine.to_plantuml()
print(puml)
# @startuml
# hide empty description
# state "off" as toggle_off
# state "on" as toggle_on
# [*] --> toggle_off
# toggle_off --> toggle_on : TOGGLE
# toggle_on --> toggle_off : TOGGLE
# @enduml

mmd = machine.to_mermaid()
print(mmd)
# stateDiagram-v2
# [*] --> off
# off --> on : TOGGLE
# on --> off : TOGGLE

machine.is_known_event(event_type: str) -> bool

machine = create_machine(config)
assert machine.is_known_event("OPEN")
assert not machine.is_known_event("TYPO_EVENT")

from xstate_statemachine.persistence import SNAPSHOT_VERSION, structure_hash

from xstate_statemachine import (
    create_machine,
    SyncInterpreter,
    XStateMachineError,
    InvalidConfigError,
    ImplementationMissingError,
)

try:
    machine = create_machine(config, logic=logic)
    interp = SyncInterpreter(machine).start()
    interp.send("SOME_EVENT")
except InvalidConfigError as e:
    print(f"Config problem: {e}")
except ImplementationMissingError as e:
    print(f"Missing logic: {e}")
except XStateMachineError as e:
    print(f"General machine error: {e}")

class PluginBase(Generic[TInterpreter]):
    ...

from xstate_statemachine import PluginBase, Interpreter

class AsyncOnlyPlugin(PluginBase[Interpreter]):
    def on_event_received(self, interpreter: Interpreter, event):
        # `interpreter` is correctly typed as async Interpreter
        print(f"Event: {event.type}")

from xstate_statemachine import PluginBase, SyncInterpreter

class MetricsPlugin(PluginBase[SyncInterpreter]):
    def __init__(self):
        self.event_count = 0
        self.transition_count = 0

    def on_event_received(self, interpreter, event):
        self.event_count += 1

    def on_transition(self, interpreter, from_states, to_states, transition):
        self.transition_count += 1

    def report(self):
        print(f"Events: {self.event_count}, Transitions: {self.transition_count}")

# Usage
metrics = MetricsPlugin()
interpreter = SyncInterpreter(machine).use(metrics).start()
interpreter.send("GO")
metrics.report()  # Events: 1, Transitions: 1

class LoggingInspector(PluginBase[Any]):
    def __init__(
        self,
        *,
        redact_keys: Tuple[str, ...] = DEFAULT_REDACT_KEYS,
        log_context: bool = True,
    ) -> None: ...

import logging
from xstate_statemachine import SyncInterpreter, LoggingInspector

# Enable logging output
logging.basicConfig(level=logging.INFO)

interpreter = (
    SyncInterpreter(machine)
    .use(LoggingInspector())
    .start()
)

interpreter.send("START")
# Output:
# INFO: 🕵️ [INSPECT] Event Received: START
# INFO: 🕵️ [INSPECT] Executing Action: logStart
# INFO: 🕵️ [INSPECT] Transition: ['myMachine.idle'] -> ['myMachine.running'] on Event 'START'
# INFO: 🕵️ [INSPECT] New Context: {'count': 0}

from xstate_statemachine import enqueue_actions

def build(args):
    enqueue = args["enqueue"]          # ActionEnqueuer
    enqueue.assign({"n": lambda x: x["context"]["n"] + 1})
    enqueue.raise_({"type": "NEXT"})

{"actions": enqueue_actions(build)}

from xstate_statemachine import initial_transition, pure_transition

snapshot, entry_actions = initial_transition(machine)
next_snapshot, actions = pure_transition(machine, snapshot, "GO")

await wait_for(interp, lambda s: s.matches("fetch.success"), timeout=2)
wait_for_sync(interp, lambda s: s.matches("job.done"), timeout=5)
output = await to_promise(interp)