Optional Modules

pysm keeps the default import small:

from pysm import StateMachine, State, Event

The classic core stays dependency-free. Runtime helpers that need queues, threading, asyncio, serialization helpers, or builder ergonomics live in separate modules and must be imported explicitly.

Core vs Optional Modules

Module	Import	Purpose	MicroPython / `upysm` fit
Core	`from pysm import StateMachine, State, Event`	Classic tiny runtime	Yes
Queued runtime	`from pysm.queued import QueuedStateMachine`	Run-to-completion event scheduling	CPython-oriented optional module
Thread-safe queued runtime	`from pysm.queued import ThreadSafeQueuedStateMachine`	Queued dispatch protected by `threading.RLock`	CPython-only
Async queued runtime	`from pysm.aio import AsyncQueuedStateMachine`	Run-to-completion dispatch for `asyncio` callbacks	CPython 3.7+ only
Serialization	`from pysm.serialization import snapshot, restore`	Snapshot active state and history using stable state paths	Optional utility
Builder	`from pysm.builder import StateMachineBuilder`	Reduce setup boilerplate without metaclasses or magic	Optional utility
Type stubs	`*.pyi` files	IDE and type-checker help without runtime annotations	Development-time only

Core Behavior Updates

The core API is still the same explicit building-block API, but recent runtime hardening added a few details worth making visible:

dispatch() now raises StateMachineException if the machine has not been initialized. Call initialize() on the root machine after adding all states and transitions.
The built-in state_stack, leaf_state_stack, and stack on StateMachine instances are bounded by StateMachine.STACK_SIZE. The default is 32. If you need an unbounded standalone stack, create Stack(maxlen=None) yourself.
The core import does not load asyncio, json, threading, or any of the optional pysm.* helper modules.

Initial Entry Events

By default, initialize() preserves the historical behavior and only selects the active initial path. If you want initial states to receive enter events, opt in explicitly:

from pysm import State, StateMachine

calls = []
ready = State('ready')
ready.handlers = {'enter': lambda state, event: calls.append(state.name)}

machine = StateMachine('m')
machine.add_state(ready, initial=True)
machine.initialize(fire_events_on_init=True)

assert calls == ['ready']

For hierarchical machines, enter handlers fire from the root’s initial child down to the active leaf state. The root machine itself is not entered. Queued and async machines keep the same ordering, but they drain any events dispatched by initial enter handlers only after the initial path is ready.

Queued Run-To-Completion Dispatch

The classic StateMachine preserves historical behavior. If handlers dispatch events and you need deterministic scheduling, use the opt-in queued runtime:

from pysm import Event, State
from pysm.queued import QueuedStateMachine

calls = []
machine = QueuedStateMachine('m')
start = State('start')
ready = State('ready')

def on_enter_ready(state, event):
    machine.dispatch(Event('validate'))
    machine.dispatch(Event('metrics'))

ready.handlers = {'enter': on_enter_ready}

machine.add_state(start, initial=True)
machine.add_state(ready)
machine.add_transition(start, ready, events=['go'])
machine.add_transition(
    ready, None, events=['validate'],
    action=lambda state, event: calls.append('validate'))
machine.add_transition(
    ready, None, events=['metrics'],
    action=lambda state, event: calls.append('metrics'))
machine.initialize()

machine.dispatch(Event('go'))
assert calls == ['validate', 'metrics']

QueuedStateMachine uses two FIFO queues. External events go to the external queue. Events raised while the machine is already processing go to the internal queue. Internal events are drained before the next external event is processed. This gives deterministic run-to-completion behavior without changing the classic core class.

Pass max_internal_steps while debugging if you want to fail loudly on an accidental internal event cycle:

machine = QueuedStateMachine('m', max_internal_steps=100)

If a handler, condition, or transition callback raises, the queued runtime clears pending internal and external work before re-raising the original exception. A later dispatch starts from a clean queue.

Thread-Safe Queued Dispatch

For multi-threaded CPython programs, use the locked queued runtime:

from pysm.queued import ThreadSafeQueuedStateMachine

machine = ThreadSafeQueuedStateMachine('m')

It uses threading.RLock and holds the lock for the whole run-to-completion cycle. Long blocking handlers will therefore block other threads from dispatching into the same machine. Async support is intentionally not part of this class.

Async Queued Dispatch

For asyncio applications, use the async queued runtime:

from pysm import Event, State
from pysm.aio import AsyncQueuedStateMachine

machine = AsyncQueuedStateMachine('m')
idle = State('idle')
ready = State('ready')

async def on_enter_ready(state, event):
    await machine.dispatch(Event('validate'))

ready.handlers = {'enter': on_enter_ready}

machine.add_state(idle, initial=True)
machine.add_state(ready)
machine.add_transition(idle, ready, events=['go'])
machine.add_transition(ready, None, events=['validate'])
machine.initialize()

await machine.dispatch(Event('go'))

AsyncQueuedStateMachine supports both synchronous and asynchronous handlers, conditions, and transition callbacks. Idle dispatches are serialized with asyncio.Lock. Events raised from the currently running transition task are queued as internal events. If another task dispatches into a machine that is already processing, that event is queued as external work and the dispatch returns after enqueueing; the active processor drains it after the current internal queue is empty. Use one event loop per machine.

Use await machine.async_initialize(fire_events_on_init=True) if initial enter handlers need to be awaited.

Typed Packages

The distribution ships py.typed and .pyi files for the core and optional modules. This gives static type checkers a public API surface without adding runtime annotations to the MicroPython-friendly core.

Snapshot And Restore

Serialization returns plain Python primitives. It does not import json and does not serialize callback code or domain objects.

Snapshots include machine-owned runtime state: the current local state for each state machine, the root leaf_state, each machine’s state_stack, and the root leaf_state_stack. They intentionally do not include machine.stack. That stack is user PDA/domain storage and may contain arbitrary objects, so persist it separately if your application needs it.

from pysm.serialization import restore, snapshot

data = snapshot(machine)
# Store data with json.dumps(data), a database JSON column, Redis, etc.

restored = build_the_same_machine_graph()
restored.initialize()
restore(restored, data)

Snapshots use full state paths, for example ['oven', 'door_closed', 'heating', 'baking']. Bare state names are not enough in a hierarchical machine because different branches may contain states with the same name. Restore is strict and raises if the saved topology does not match the machine being restored.

Pass metadata when your persistence layer needs application data next to the runtime state:

data = snapshot(machine, metadata={'schema': 3})

The metadata is copied into the returned dictionary, but restore only uses the runtime fields.

Builder

The builder is an optional setup helper. It does not change the core runtime and does not add methods to your domain objects.

from pysm.builder import StateMachineBuilder

machine = (StateMachineBuilder('toggle')
           .state('off', initial=True)
           .state('on')
           .transition('off', 'on', events=['turn_on'])
           .transition('on', 'off', events=['turn_off'])
           .build())

For nested machines, use paths when short names would be ambiguous:

machine = (StateMachineBuilder('oven')
           .machine('door_closed', initial=True)
           .state('off', initial=True, parent_path='door_closed')
           .machine('heating', parent_path='door_closed')
           .state('baking', initial=True, parent_path='heating')
           .transition('off', 'baking', events='bake')
           .build())

String events and input values are treated as one event or input value, not as iterables of characters. Use a list, tuple, or other iterable when a single transition should match many values.

MicroPython / upysm

The MicroPython-oriented target should stay core-only:

from pysm import StateMachine, State, Event

Do not copy optional modules such as pysm.queued, pysm.aio, pysm.serialization, or pysm.builder into a constrained device build unless you explicitly need them and have measured the memory cost.