Optional Modules ================ ``pysm`` keeps the default import small: .. code-block:: python 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 ------------------------ .. list-table:: :header-rows: 1 * - 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: .. code-block:: python 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: .. code-block:: python 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: .. code-block:: python 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: .. code-block:: python 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: .. code-block:: python 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. .. code-block:: python 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: .. code-block:: python 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. .. code-block:: python 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: .. code-block:: python 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: .. code-block:: python 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.