pyfcstm.simulate.runtime

Simulation runtime for executing hierarchical finite state machines.

This module implements a cycle-based execution engine for state machines parsed from the pyfcstm DSL. The runtime maintains an execution stack of active states, variable mappings, and internal frame modes that control lifecycle progression.

Each cycle advances the state machine until reaching a stable boundary: either a stoppable state (non-pseudo leaf state), termination (empty stack), or validation failure (changes rolled back). The runtime uses speculative validation to ensure transitions can reach valid stable states before executing them.

The execution model uses internal frame modes (active, after_entry, init_wait, post_child_exit) to track execution phases. Transitions are selected from the current stack-top state’s transitions in declaration order. Parent-level transitions are only considered after a child explicitly exits to parent via [*] transitions.

Lifecycle actions execute in a specific order: enter when entering states, during while in leaf states each cycle, and exit when leaving states. For leaf states, the during chain includes ancestor aspect actions (>> during before/after) that execute before and after the leaf’s own during actions. Pseudo states skip ancestor aspect actions entirely.

Composite states have special during before/after actions that execute only at composite boundaries: during before runs when entering from parent ([*] -> Child), and during after runs when exiting to parent (Child -> [*]). These do NOT execute during child-to-child transitions.

Validation works by cloning the execution context and speculatively executing the transition until reaching a stable boundary. If validation succeeds, the real transition executes. If validation fails or exceeds safety limits (1000 steps or 64 stack depth), the transition is rejected and variables roll back.

Abstract actions can be implemented by registering Python handlers that receive read-only execution context. Handlers can be registered individually or organized in classes using the @abstract_handler decorator.

Hot Start Feature:

The runtime supports “hot start” mode, which allows starting execution from an arbitrary state without executing enter actions. This is useful for:

  • Debugging: Jump directly to a specific state to test behavior

  • State Recovery: Resume execution from a known state and variable configuration

  • Testing: Verify state-specific logic without executing full initialization

Hot start is enabled by providing initial_state and initial_vars parameters to the constructor. The runtime constructs the execution stack directly to the target state, bypassing all enter actions. For composite states, the runtime automatically performs initial transitions during the first cycle to find a stoppable leaf state.

Basic usage:

from pyfcstm.simulate import SimulationRuntime

runtime = SimulationRuntime(state_machine)
runtime.cycle()  # Execute first cycle
runtime.cycle(['EventName'])  # Execute with events

# Access state and variables
current_state = runtime.current_state
variables = runtime.vars

Hot start usage:

# Start from a specific state with custom variable values
runtime = SimulationRuntime(
    state_machine,
    initial_state="System.Active",
    initial_vars={"counter": 10, "flag": 1}
)
# First cycle starts from Active state without executing enter actions
runtime.cycle()

Abstract handler registration:

def my_handler(ctx):
    print(f"State: {ctx.get_full_state_path()}")

runtime.register_abstract_handler('System.Active.Init', my_handler)

# Or use decorator-based registration
from pyfcstm.simulate import abstract_handler

class MyHandlers:
    @abstract_handler('System.Active.Init')
    def handle_init(self, ctx):
        print(f"Counter: {ctx.get_var('counter')}")

runtime.register_handlers_from_object(MyHandlers())

SimulationRuntimeDfsError

class pyfcstm.simulate.runtime.SimulationRuntimeDfsError[source]

Raised when speculative validation exceeds safety limits without converging.

This exception indicates that the state machine contains an invalid unbounded execution chain that prevents the runtime from reaching a stable stoppable state. The runtime enforces two safety limits during speculative validation:

  1. Step Limit: Maximum 1000 speculative steps per validation attempt

  2. Depth Limit: Maximum 64 structural stack frames

When either limit is exceeded, validation aborts and raises this exception to prevent infinite loops or stack overflow.

Why This Happens:

This error typically occurs when a state machine has:

  • Composite states with no valid initial transitions

  • Chains of pseudo states that never reach a stoppable state

  • Circular dependencies between exit transitions and parent continuations

  • Guard conditions that create infinite validation loops

Error Prevention:

To avoid this error, ensure your state machine:

  • Every composite state has at least one unconditional initial transition

  • Pseudo state chains eventually reach a stoppable leaf state

  • Exit transitions lead to valid parent continuations

  • Guard conditions don’t create circular validation dependencies

Example of problematic state machine:

>>> dsl_code = '''
... state Root {
...     state A {
...         [*] -> B : if [false];  # Always blocked
...         pseudo state B;
...     }
...     [*] -> A;
... }
... '''
>>> ast = parse_with_grammar_entry(dsl_code, 'state_machine_dsl')
>>> sm = parse_dsl_node_to_state_machine(ast)
>>> runtime = SimulationRuntime(sm)
>>> runtime.cycle()  # Raises SimulationRuntimeDfsError

Note

This exception is raised during validation, not during normal execution. If you encounter this error, review your state machine definition for unbounded execution chains or missing stoppable states.

SimulationRuntime

class pyfcstm.simulate.runtime.SimulationRuntime(state_machine: StateMachine, abstract_error_mode: Literal['raise', 'log'] = 'raise', history_size: int | None = None, initial_state: str | Tuple[str, ...] | State | None = None, initial_vars: Dict[str, int | float] | None = None)[source]

Runtime environment for executing hierarchical finite state machines.

This class provides the stateful execution engine for state machines parsed from the pyfcstm DSL. It maintains an active state stack, variable mappings, and lifecycle flags that control cycle-based execution semantics.

Core Execution Model:

The runtime implements cycle-based execution where each cycle() call advances the state machine until reaching a stable boundary. Cycles can end in three ways:

  1. Stoppable State: Reached a leaf state (non-pseudo) where execution can stabilize

  2. Termination: The state machine has ended (empty stack)

  3. Validation Failure: Cannot reach a stoppable state, changes rolled back

Transition Selection:

Transitions are always selected from the current stack-top state’s transitions_from list in declaration order. This is critical for understanding cross-level transition behavior:

  • A leaf state can only fire its own transitions

  • Parent-level transitions are only considered after the child exits to parent

  • This explains why some transitions require explicit exit transitions first

For example, if state System1.A wants to trigger a parent-level transition System1 -> System2, it must first exit to parent via A -> [*], which puts System1 in post_child_exit mode where parent-level transitions can be considered.

Validation and Rollback:

When a stoppable state has an enabled transition, the runtime performs speculative validation to ensure the transition can eventually reach another stoppable state or terminate. This prevents cycles from getting stuck in non-stoppable configurations.

If validation fails or a cycle cannot reach a stoppable state, all variable changes are rolled back and the runtime remains at the previous stable boundary. For the initial cycle, rollback pins the runtime at the root boundary in init_wait mode.

DFS Safety Limits:

To prevent infinite loops in pathological state machines, validation enforces two safety limits:

  • Maximum 1000 speculative steps per validation attempt

  • Maximum 64 structural stack depth

  • Repeated execution states are pruned automatically

When these limits are exceeded, SimulationRuntimeDfsError is raised.

Parameters:

state_machine (StateMachine) – The state machine model to simulate.

Variables:

  • state_machine (StateMachine) – The state machine being simulated.

  • stack (List[_Frame]) – Active frames ordered from root to current execution point.

  • vars (Dict[str, Union[int, float]]) – Mutable variable values visible to guards, effects, and actions.

  • _initialized (bool) – Whether the runtime has performed root entry.

  • _ended (bool) – Whether execution has terminated (empty stack).

Example:

>>> from pyfcstm.dsl import parse_with_grammar_entry
>>> from pyfcstm.model import parse_dsl_node_to_state_machine
>>> from pyfcstm.simulate import SimulationRuntime
>>> dsl_code = '''
... def int counter = 0;
... state System {
...     state Idle {
...         during { counter = counter + 1; }
...     }
...     state Active {
...         during { counter = counter + 10; }
...     }
...     [*] -> Idle;
...     Idle -> Active :: Start;
... }
... '''
>>> ast = parse_with_grammar_entry(dsl_code, 'state_machine_dsl')
>>> sm = parse_dsl_node_to_state_machine(ast)
>>> runtime = SimulationRuntime(sm)
>>> runtime.cycle()
>>> runtime.current_state.path
('System', 'Idle')
>>> runtime.vars['counter']
1
>>> runtime.cycle(['System.Idle.Start'])
>>> runtime.current_state.path
('System', 'Active')
>>> runtime.vars['counter']
11
__init__(state_machine: StateMachine, abstract_error_mode: Literal['raise', 'log'] = 'raise', history_size: int | None = None, initial_state: str | Tuple[str, ...] | State | None = None, initial_vars: Dict[str, int | float] | None = None)[source]

Initialize the simulation runtime with a state machine model.

This constructor prepares the runtime for execution by initializing variable storage from the state machine’s variable definitions and setting up the initial execution stack with the root state. Variables are initialized in declaration order, allowing later initializers to reference earlier variables.

The runtime stack is initialized with the root state in init_wait mode, allowing current_state to be accessed immediately. Full initialization (entering the root state and executing lifecycle actions) is deferred until the first cycle() call.

Hot Start Mode:

If initial_state is provided, the runtime performs a “hot start” by directly constructing the execution stack to the specified state without executing enter actions. This simulates having already entered and stabilized at that state. The initial_vars parameter allows overriding variable values for the hot start.

Parameters:

  • state_machine (StateMachine) – The state machine model to simulate.

  • abstract_error_mode (Literal['raise', 'log']) – Error handling mode for abstract handler exceptions. 'raise' (default) stops execution and raises the exception. 'log' logs the exception and continues execution.

  • history_size (Optional[int]) – Maximum number of history entries to keep. None (default) means unlimited history.

  • initial_state (Optional[Union[str, Tuple[str, ...], State]]) – Optional initial state for hot start. If provided, the runtime will start from this state without executing enter actions. Supports string path ("System.Active"), tuple path (('System', 'Active')), or State object. Defaults to None (start from root state).

  • initial_vars (Optional[Dict[str, Union[int, float]]]) – Optional variable overrides for hot start. Only variables defined in the state machine can be overridden. Defaults to None.

Returns:

None.

Return type:

None

Example - Default initialization:

>>> from pyfcstm.dsl import parse_with_grammar_entry
>>> from pyfcstm.model import parse_dsl_node_to_state_machine
>>> from pyfcstm.simulate import SimulationRuntime
>>> dsl_code = '''
... def int x = 10;
... def int y = x + 5;  # Can reference earlier variables
... state Root {
...     state A;
...     [*] -> A;
... }
... '''
>>> ast = parse_with_grammar_entry(dsl_code, 'state_machine_dsl')
>>> sm = parse_dsl_node_to_state_machine(ast)
>>> runtime = SimulationRuntime(sm)
>>> runtime.vars
{'x': 10, 'y': 15}
>>> runtime.is_ended
False
>>> runtime.current_state.name
'Root'
>>> runtime.current_state.path
('Root',)

Example - Hot start from specific state:

>>> dsl_code = '''
... def int counter = 0;
... state System {
...     state Idle {
...         during { counter = counter + 1; }
...     }
...     state Active {
...         during { counter = counter + 10; }
...     }
...     [*] -> Idle;
...     Idle -> Active :: Start;
... }
... '''
>>> ast = parse_with_grammar_entry(dsl_code, 'state_machine_dsl')
>>> sm = parse_dsl_node_to_state_machine(ast)
>>> runtime = SimulationRuntime(
...     sm,
...     initial_state="System.Active",
...     initial_vars={"counter": 10}
... )
>>> runtime.current_state.path
('System', 'Active')
>>> runtime.vars['counter']
10
>>> runtime.cycle()  # First cycle starts from Active state
>>> runtime.vars['counter']
20

Note

Variable initialization and stack setup happen during construction, but state entry actions are deferred until the first cycle() call. This allows inspection of initial variable values and the root state before execution begins.

Note

Hot start mode constructs the stack without executing enter actions, simulating having already entered the target state. For composite states, the runtime will automatically attempt initial transitions to find a stoppable leaf state during the first cycle.

property abstract_handler_errors: List[Tuple[str, Exception]]

Get list of abstract handler errors (only in ‘log’ mode).

When abstract_error_mode is 'log', exceptions from abstract handlers are logged and collected here instead of stopping execution.

Returns:

List of (action_path, exception) tuples

Return type:

List[Tuple[str, Exception]]

Example:

>>> runtime = SimulationRuntime(sm, abstract_error_mode='log')
>>> runtime.cycle()
>>> for action_path, exception in runtime.abstract_handler_errors:
...     print(f"Error in {action_path}: {exception}")
property brief_stack: List[Tuple[Tuple[str, ...], str]]

Return a compact representation of the active execution stack.

Each tuple contains the state’s full path and the frame’s internal mode. This representation is useful for debugging, testing, and understanding the runtime’s current execution phase without inspecting internal frame objects directly.

Frame Modes:

  • active: Normal execution state

  • after_entry: Just entered a leaf state, during already executed

  • init_wait: Composite state waiting for initial transition

  • post_child_exit: Parent state after child exited via [*]

Returns:

List of (state_path, mode) tuples from root to current state.

Return type:

List[Tuple[Tuple[str, …], str]]

Example:

>>> from pyfcstm.dsl import parse_with_grammar_entry
>>> from pyfcstm.model import parse_dsl_node_to_state_machine
>>> from pyfcstm.simulate import SimulationRuntime
>>> dsl_code = '''
... state System {
...     state SubSystem {
...         state Active;
...         [*] -> Active;
...     }
...     [*] -> SubSystem;
... }
... '''
>>> ast = parse_with_grammar_entry(dsl_code, 'state_machine_dsl')
>>> sm = parse_dsl_node_to_state_machine(ast)
>>> runtime = SimulationRuntime(sm)
>>> runtime.cycle()
>>> runtime.brief_stack
[(('System',), 'active'), (('System', 'SubSystem'), 'active'), (('System', 'SubSystem', 'Active'), 'active')]

Note

This property is primarily useful for testing and debugging. It provides insight into the runtime’s internal execution state without exposing the internal _Frame objects.

clear_all_abstract_handlers() → int[source]

Clear all registered abstract handlers.

Returns:

Total number of handlers removed

Return type:

int

Example:

>>> count = runtime.clear_all_abstract_handlers()
>>> print(f"Cleared {count} handlers")
property current_state: State

Get the current active state at the top of the execution stack.

This property returns the state currently being executed by the runtime. For leaf states, this is the state where during actions execute. For composite states in init_wait mode, this is the parent waiting for an initial transition.

Returns:

The current active state.

Return type:

State

Raises:

IndexError – If the runtime has ended and the stack is empty.

Example:

>>> from pyfcstm.dsl import parse_with_grammar_entry
>>> from pyfcstm.model import parse_dsl_node_to_state_machine
>>> from pyfcstm.simulate import SimulationRuntime
>>> dsl_code = '''
... state System {
...     state Idle;
...     state Active;
...     [*] -> Idle;
...     Idle -> Active :: Start;
... }
... '''
>>> ast = parse_with_grammar_entry(dsl_code, 'state_machine_dsl')
>>> sm = parse_dsl_node_to_state_machine(ast)
>>> runtime = SimulationRuntime(sm)
>>> runtime.current_state.name
'System'
>>> runtime.current_state.path
('System',)
>>> runtime.cycle()
>>> runtime.current_state.name
'Idle'
>>> runtime.cycle(['System.Idle.Start'])
>>> runtime.current_state.name
'Active'

Note

Before the first cycle() call, this returns the root state. After the runtime has ended, accessing this property will raise an IndexError.

cycle(events: List[str | Event] | None = None)[source]

Execute a full runtime cycle until reaching a stable boundary.

This method advances the state machine through transitions and lifecycle actions until one of three conditions is met:

  1. Stoppable State: Reached a leaf state (non-pseudo) where execution can stabilize

  2. Termination: The state machine has ended (empty stack)

  3. Validation Failure: Cannot reach a stoppable state, changes rolled back

Cycle Execution Flow:

The cycle operates in two phases:

  1. Speculative Execution: Clone the current context and simulate the cycle to validate it can reach a stable boundary

  2. Commit or Rollback: If validation succeeds, commit the changes; otherwise, rollback to the previous stable state

Event Handling:

Events can be provided as either event objects or dot-separated path strings. Multiple events can be active simultaneously, allowing complex transition chains to execute in a single cycle.

Flexible Event Path Formats:

The runtime supports multiple event path formats for maximum convenience:

  • Full paths: "Root.System.Active.Start" - Complete path from root

  • Relative paths: "Start" - Resolved from current state

  • Parent-relative: ".error" or "..system.error" - Navigate up hierarchy

  • Absolute: "/global.shutdown" - Resolved from root state

The runtime intelligently determines which resolution method to use based on the path syntax and current runtime state. This allows you to use the most convenient format for your use case without worrying about the details.

Validation and Safety:

When a stoppable state has an enabled transition, the runtime validates that taking the transition can eventually reach another stoppable state or terminate. This prevents cycles from getting stuck in non-stoppable configurations.

Validation enforces safety limits: - Maximum 1000 speculative steps per validation - Maximum 64 structural stack depth - Repeated execution states are pruned automatically

Rollback Behavior:

If a cycle cannot reach a stoppable state, all variable changes are rolled back: - For normal cycles: Restore previous stack and variables - For initial cycle: Pin runtime at root boundary in init_wait mode - All side effects from failed validation are discarded

Parameters:

events (List[Union[str, Event]], optional) – Events available for the current cycle. Can be event objects or dot-separated path strings.

Returns:

None.

Return type:

None

Example - Basic cycle execution:

>>> from pyfcstm.dsl import parse_with_grammar_entry
>>> from pyfcstm.model import parse_dsl_node_to_state_machine
>>> from pyfcstm.simulate import SimulationRuntime
>>> dsl_code = '''
... def int counter = 0;
... state System {
...     state Idle {
...         during { counter = counter + 1; }
...     }
...     state Active {
...         during { counter = counter + 10; }
...     }
...     [*] -> Idle;
...     Idle -> Active :: Start;
... }
... '''
>>> ast = parse_with_grammar_entry(dsl_code, 'state_machine_dsl')
>>> sm = parse_dsl_node_to_state_machine(ast)
>>> runtime = SimulationRuntime(sm)
>>> runtime.cycle()  # Initialize and reach Idle
>>> runtime.current_state.path
('System', 'Idle')
>>> runtime.vars['counter']
1
>>> runtime.cycle(['System.Idle.Start'])  # Transition to Active
>>> runtime.current_state.path
('System', 'Active')

Example - Flexible event path formats:

>>> dsl_code = '''
... state Root {
...     event global_stop;
...     state System {
...         event system_error;
...         state Idle {
...             event start;
...         }
...         state Active {
...             event pause;
...         }
...         [*] -> Idle;
...         Idle -> Active :: start;
...         Active -> Idle :: pause;
...     }
...     [*] -> System;
... }
... '''
>>> ast = parse_with_grammar_entry(dsl_code, 'state_machine_dsl')
>>> sm = parse_dsl_node_to_state_machine(ast)
>>> runtime = SimulationRuntime(sm)
>>> runtime.cycle()
>>> runtime.current_state.path
('Root', 'System', 'Idle')
>>> # Use relative path from current state - most convenient!
>>> runtime.cycle(['start'])
>>> runtime.current_state.path
('Root', 'System', 'Active')
>>> # Use parent-relative path to access parent's event
>>> runtime.cycle(['.system_error'])  # Access System.system_error
>>> # Use absolute path to access root event
>>> runtime.cycle(['/global_stop'])  # Access Root.global_stop
>>> # Full path still works for backward compatibility
>>> runtime.cycle(['Root.System.Active.pause'])
>>> runtime.current_state.path
('Root', 'System', 'Idle')

Example - Exit transitions and parent continuation:

>>> dsl_code = '''
... def int x = 0;
... state System1 {
...     state A {
...         during { x = x + 1; }
...     }
...     [*] -> A;
...     A -> [*] :: Exit;
... }
... state System2 {
...     state B {
...         during { x = x + 10; }
...     }
...     [*] -> B;
... }
... [*] -> System1;
... System1 -> System2 :: Switch;
... '''
>>> ast = parse_with_grammar_entry(dsl_code, 'state_machine_dsl')
>>> sm = parse_dsl_node_to_state_machine(ast)
>>> runtime = SimulationRuntime(sm)
>>> runtime.cycle()
>>> runtime.current_state.path
('System1', 'A')
>>> # Provide both Exit and Switch events
>>> runtime.cycle(['System1.A.Exit', 'Switch'])
>>> runtime.current_state.path
('System2', 'B')

Example - Validation preventing invalid transitions:

>>> dsl_code = '''
... state Root {
...     state A;
...     state B {
...         [*] -> C : if [false];  # Blocked initial transition
...         state C;
...     }
...     [*] -> A;
...     A -> B :: Go;  # Rejected by validation
... }
... '''
>>> ast = parse_with_grammar_entry(dsl_code, 'state_machine_dsl')
>>> sm = parse_dsl_node_to_state_machine(ast)
>>> runtime = SimulationRuntime(sm)
>>> runtime.cycle()
>>> runtime.cycle(['Root.A.Go'])  # Rejected - cannot reach stoppable
>>> runtime.current_state.path
('Root', 'A')  # Remains in A due to validation failure

Example - Multiple cycles with state changes:

>>> dsl_code = '''
... def int counter = 0;
... state Root {
...     state A {
...         during { counter = counter + 1; }
...     }
...     [*] -> A;
...     A -> A : if [counter >= 3];  # Self-transition after 3 cycles
... }
... '''
>>> ast = parse_with_grammar_entry(dsl_code, 'state_machine_dsl')
>>> sm = parse_dsl_node_to_state_machine(ast)
>>> runtime = SimulationRuntime(sm)
>>> runtime.cycle()
>>> runtime.vars['counter']
1
>>> runtime.cycle()
>>> runtime.vars['counter']
2
>>> runtime.cycle()
>>> runtime.vars['counter']
3
>>> runtime.cycle()  # Self-transition fires
>>> runtime.vars['counter']
4  # Exit + re-enter + during

Note

Once the runtime has ended (is_ended is True), subsequent cycle() calls are no-ops. Create a new SimulationRuntime instance to restart execution.

Warning

If validation exceeds safety limits (1000 steps or 64 stack depth), SimulationRuntimeDfsError is raised. This indicates an invalid state machine with unbounded execution chains.

property error_info: Tuple[str, Exception] | None

Get error information if runtime is in error state.

Returns:

Tuple of (action_path, exception) if in error state, None otherwise.

Return type:

Optional[Tuple[str, Exception]]

Example:

>>> if runtime.is_error_state:
...     action_path, exception = runtime.error_info
...     print(f"Error in {action_path}: {exception}")
get_abstract_handlers(action_path: str) → List[Callable[[ReadOnlyExecutionContext], None]][source]

Get all registered handlers for an abstract action.

Parameters:

action_path (str) – Full path to the abstract action

Returns:

List of registered handlers (may be empty)

Return type:

List[Callable[[ReadOnlyExecutionContext], None]]

Example:

>>> handlers = runtime.get_abstract_handlers("System.Active.InitHardware")
>>> print(f"Found {len(handlers)} handlers")
has_abstract_handlers(action_path: str) → bool[source]

Check if any handlers are registered for an abstract action.

Parameters:

action_path (str) – Full path to the abstract action

Returns:

True if at least one handler is registered

Return type:

bool

Example:

>>> if runtime.has_abstract_handlers("System.Active.InitHardware"):
...     print("Handlers registered")
property is_ended: bool

Indicate whether the runtime has finished execution.

Once True, the state machine has terminated and the execution stack is empty. Subsequent cycle() calls become no-ops. To restart execution, create a new SimulationRuntime instance.

Termination Conditions:

The runtime ends when: - The root state exits to [*] - An exit transition from a root-level state completes - The execution stack becomes empty for any reason

Returns:

True if execution has ended, False otherwise.

Return type:

bool

Example:

>>> from pyfcstm.dsl import parse_with_grammar_entry
>>> from pyfcstm.model import parse_dsl_node_to_state_machine
>>> from pyfcstm.simulate import SimulationRuntime
>>> dsl_code = '''
... def int counter = 0;
... state System {
...     state Active {
...         during { counter = counter + 1; }
...     }
...     [*] -> Active;
...     Active -> [*] : if [counter >= 3];
... }
... '''
>>> ast = parse_with_grammar_entry(dsl_code, 'state_machine_dsl')
>>> sm = parse_dsl_node_to_state_machine(ast)
>>> runtime = SimulationRuntime(sm)
>>> runtime.is_ended
False
>>> runtime.cycle()
>>> runtime.is_ended
False
>>> runtime.cycle()
>>> runtime.is_ended
False
>>> runtime.cycle()
>>> runtime.is_ended
False
>>> runtime.cycle()  # counter >= 3, exit transition fires
>>> runtime.is_ended
True
>>> runtime.cycle()  # No-op, runtime already ended
>>> runtime.is_ended
True

Note

Once the runtime has ended, the only way to restart execution is to create a new SimulationRuntime instance with the same or a different state machine model.

property is_error_state: bool

Indicate whether the runtime is in an error state.

When abstract_error_mode is 'raise' and an abstract handler raises an exception, the runtime enters an error state. In this state, the runtime preserves the error information and prevents further execution.

Returns:

True if runtime is in error state, False otherwise.

Return type:

bool

Example:

>>> runtime = SimulationRuntime(sm, abstract_error_mode='raise')
>>> runtime.is_error_state
False
>>> # After handler raises exception
>>> runtime.is_error_state
True
register_abstract_handler(action_path: str, handler: Callable[[ReadOnlyExecutionContext], None]) → None[source]

Register a Python function to handle an abstract action.

Multiple handlers can be registered for the same abstract action. They will be executed in registration order.

Parameters:

  • action_path (str) – Full path to the abstract action (e.g., “System.Active.InitHardware”)

  • handler (Callable[[ReadOnlyExecutionContext], None]) – Python function that receives read-only context

Raises:

ValueError – If action_path is empty or invalid

Example:

>>> def my_init(ctx: ReadOnlyExecutionContext):
...     print(f"Initializing in state {ctx.get_state_name()}")
...     print(f"Counter value: {ctx.get_var('counter')}")
>>>
>>> runtime.register_abstract_handler("System.Active.InitHardware", my_init)
>>>
>>> # Register another handler for the same action
>>> def my_init2(ctx: ReadOnlyExecutionContext):
...     print(f"Second handler for {ctx.action_name}")
>>>
>>> runtime.register_abstract_handler("System.Active.InitHardware", my_init2)
register_handlers_from_object(obj: object) → int[source]

Register all decorated methods from an object as abstract handlers.

This method scans the object for methods decorated with abstract_handler() and automatically registers them with their specified action paths.

This provides a convenient way to organize multiple related handlers in a single class, maintaining state and shared logic between handlers.

Parameters:

obj (object) – Object instance containing decorated handler methods

Returns:

Number of handlers registered

Return type:

int

Example:

>>> from pyfcstm.simulate import abstract_handler
>>> class MyHandlers:
...     def __init__(self):
...         self.init_count = 0
...         self.monitor_count = 0
...
...     @abstract_handler('System.Active.Init')
...     def handle_init(self, ctx: ReadOnlyExecutionContext):
...         self.init_count += 1
...         print(f"Init called {self.init_count} times")
...
...     @abstract_handler('System.Active.Monitor')
...     def handle_monitor(self, ctx: ReadOnlyExecutionContext):
...         self.monitor_count += 1
...         counter = ctx.get_var('counter')
...         print(f"Monitor: counter={counter}, called {self.monitor_count} times")
...
...     def helper_method(self):
...         # Not decorated, won't be registered
...         pass
>>>
>>> handlers = MyHandlers()
>>> count = runtime.register_handlers_from_object(handlers)
>>> print(f"Registered {count} handlers")
Registered 2 handlers

Note

Only methods decorated with abstract_handler() will be registered. Other methods are ignored.

unregister_abstract_handler(action_path: str, handler: Callable[[ReadOnlyExecutionContext], None] | None = None) → int[source]

Unregister abstract handler(s) for an action.

If handler is None, removes all handlers for the action. If handler is provided, removes only that specific handler.

Parameters:

  • action_path (str) – Full path to the abstract action

  • handler (Optional[Callable[[ReadOnlyExecutionContext], None]]) – Specific handler to remove, or None to remove all

Returns:

Number of handlers removed

Return type:

int

Example:

>>> # Remove all handlers for an action
>>> count = runtime.unregister_abstract_handler("System.Active.InitHardware")
>>> print(f"Removed {count} handlers")
>>>
>>> # Remove specific handler
>>> runtime.unregister_abstract_handler("System.Active.InitHardware", my_init)