API Reference

Complete API reference for the VYRA State Machine system.

UnifiedStateMachine

The main class that manages all three state layers.

class vyra_base.state.unified.UnifiedStateMachine(config=None)

Bases: object

Unified interface for 3-layer state machine.

Combines Lifecycle, Operational, and Health layers into a single easy-to-use API. Recommended for most use cases.

Example

>>> usm = UnifiedStateMachine()
>>> usm.start()  # Lifecycle
>>> usm.complete_initialization()
>>> usm.set_ready()  # Operational
>>> usm.start_task({'task_id': '123'})
>>> usm.report_warning({'cpu': '85%'})  # Health
>>> usm.get_all_states()
{'lifecycle': 'Active', 'operational': 'Running', 'health': 'Warning'}

Parameter:

config (StateMachineConfig | None)

__init__(config=None)

Initialize unified state machine.

Parameter:

config (Optional[StateMachineConfig]) – Configuration for state machine behavior

get_all_states()

Get current state of all layers.

Rückgabetyp:

Dict[str, str]

get_lifecycle_state()

Get lifecycle state.

Rückgabetyp:

LifecycleState

get_operational_state()

Get operational state.

Rückgabetyp:

OperationalState

get_health_state()

Get health state.

Rückgabetyp:

HealthState

is_operational()

Check if module can execute tasks.

Rückgabetyp:

bool

is_healthy()

Check if module health is OK.

Rückgabetyp:

bool

is_initializing()

Check if module is initializing.

Rückgabetyp:

bool

is_active()

Check if module lifecycle is Active.

Rückgabetyp:

bool

is_suspended()

Check if module is suspended.

Rückgabetyp:

bool

is_recovering()

Check if module is in recovery.

Rückgabetyp:

bool

is_shutting_down()

Check if module is shutting down.

Rückgabetyp:

bool

is_offline()

Check if module is offline.

Rückgabetyp:

bool

start(metadata=None)

Start module initialization.

Rückgabetyp:

LifecycleState

Parameter:

metadata (Dict[str, Any] | None)

complete_initialization(result=None)

Complete initialization successfully.

Rückgabetyp:

LifecycleState

Parameter:

result (Dict[str, Any] | None)

fail_initialization(error=None)

Mark initialization as failed.

Rückgabetyp:

LifecycleState

Parameter:

error (str | None)

shutdown(reason=None)

Begin controlled shutdown.

Rückgabetyp:

LifecycleState

Parameter:

reason (str | None)

complete_shutdown()

Complete shutdown process.

Rückgabetyp:

LifecycleState

suspend(reason=None)

Suspend the module temporarily (e.g. for maintenance or updates).

Transitions: Active → Suspended

Parameter:

reason (Optional[str]) – Optional reason for suspension

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

resume_from_suspend(info=None)

Resume module from Suspended state.

Transitions: Suspended → Active

Parameter:

info (Optional[Dict[str, Any]]) – Optional context about the resume

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

enter_recovery(fault_info=None)

Enter recovery mode due to fault.

Transitions: Active → Recovering

Parameter:

fault_info (Optional[Dict[str, Any]]) – Fault description and context

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

complete_recovery(recovery_info=None)

Complete recovery successfully.

Transitions: Recovering → Active

Parameter:

recovery_info (Optional[Dict[str, Any]]) – Recovery details

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

fail_recovery(error=None)

Mark recovery as failed.

Transitions: Recovering → ShuttingDown

Parameter:

error (Optional[str]) – Recovery failure reason

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

set_ready(metadata=None)

Signal readiness for tasks.

Rückgabetyp:

OperationalState

Parameter:

metadata (Dict[str, Any] | None)

start_task(task_info=None)

Start task execution.

Rückgabetyp:

OperationalState

Parameter:

task_info (Dict[str, Any] | None)

pause(reason=None)

Pause current task.

Rückgabetyp:

OperationalState

Parameter:

reason (str | None)

resume()

Resume paused task.

Rückgabetyp:

OperationalState

stop(result=None)

Mark task as completed.

Rückgabetyp:

OperationalState

Parameter:

result (Dict[str, Any] | None)

reset()

Reset operational state to Idle.

Rückgabetyp:

OperationalState

report_warning(warning_info=None)

Report non-critical warning.

Rückgabetyp:

HealthState

Parameter:

warning_info (Dict[str, Any] | None)

report_fault(fault_info=None)

Report critical fault.

Rückgabetyp:

HealthState

Parameter:

fault_info (Dict[str, Any] | None)

recover(recovery_info=None)

Attempt recovery from fault.

Rückgabetyp:

HealthState

Parameter:

recovery_info (Dict[str, Any] | None)

emergency_stop(reason)

Trigger emergency stop.

Rückgabetyp:

Dict[str, str]

Parameter:

reason (str)

on_lifecycle_change(callback, priority=0)

Subscribe to lifecycle state changes.

Parameter:

• callback (Callable)

• priority (int)

on_operational_change(callback, priority=0)

Subscribe to operational state changes.

Parameter:

• callback (Callable)

• priority (int)

on_health_change(callback, priority=0)

Subscribe to health state changes.

Parameter:

• callback (Callable)

• priority (int)

on_any_change(callback, priority=0)

Subscribe to any state change across all layers.

Parameter:

• callback (Callable)

• priority (int)

get_diagnostic_info()

Get comprehensive diagnostic information.

Rückgabetyp:

Dict[str, Any]

get_history(limit=None)

Get state transition history.

Parameter:

limit (int | None)

clear_history()

Clear transition history.

send_event(event)

Send an event to the state machine.

This method is required for compatibility with OperationalStateMachine which calls _state_machine.send_event(event).

Parameter:

event – StateEvent to send to the FSM

standard_startup()

Execute standard module startup sequence.

Sequence: 1. Start initialization 2. Complete initialization 3. Set operational ready

Rückgabetyp:

bool

Rückgabe:

True if startup successful, False otherwise

standard_shutdown()

Execute standard module shutdown sequence.

Sequence: 1. Begin shutdown 2. Complete any pending tasks (module responsibility) 3. Complete shutdown

Rückgabetyp:

bool

Rückgabe:

True if shutdown successful, False otherwise

OperationalStateMachine

Base class for components that use operational state management.

class vyra_base.state.operational_state_machine.OperationalStateMachine(state_machine)

Bases: object

Base class for modules that need automatic operational state management.

This class integrates with the 3-layer StateMachine and provides a clean interface for defining lifecycle methods that automatically handle state transitions.

Subclasses should implement one or more of the following methods:

• initialize(): Setup and initialization logic

• pause(): Pause current operation

• resume(): Resume paused operation

• stop(): Stop current operation

• reset(): Reset to initial state

For dynamic operations, use the @operation decorator.

All methods should return True on success, False on failure. Exceptions are caught and treated as failures.

Example

>>> class MyModule(OperationalStateMachine):
...     def __init__(self, state_machine):
...         super().__init__(state_machine)
...         self.data = None
...
...     def initialize(self):
...         # This runs when initialize() is called
...         # Pre-condition: IDLE state
...         # On success: IDLE -> READY (operation counter reset)
...         # On failure: IDLE -> ERROR
...         self.data = []
...         print("Initialized!")
...         return True
>>>
>>> # Usage
>>> fsm = StateMachine()
>>> module = MyModule(fsm)
>>> module.initialize()  # Automatic state management
>>> # Now in READY state, ready for operations

Parameter:

state_machine (StateMachine)

__init__(state_machine)

Initialize the operational state machine.

Parameter:

state_machine (StateMachine) – The underlying 3-layer StateMachine instance

get_operational_state()

Get current operational state.

Rückgabetyp:

OperationalState

get_all_states()

Get all current states (lifecycle, operational, health).

Rückgabetyp:

Dict[str, str]

is_idle()

Check if in IDLE state.

Rückgabetyp:

bool

is_ready()

Check if in READY state.

Rückgabetyp:

bool

is_running()

Check if in RUNNING state.

Rückgabetyp:

bool

is_paused()

Check if in PAUSED state.

Rückgabetyp:

bool

is_stopped()

Check if in STOPPED state.

Rückgabetyp:

bool

is_error()

Check if in ERROR state.

Rückgabetyp:

bool

get_operation_counter()

Get the current operation reference counter value.

Rückgabetyp:

int

Rückgabe:

Number of currently active operations

State Types

LifecycleState

class vyra_base.state.state_types.LifecycleState(*values)

Bases: Enum

Lifecycle states define the existence and high-level life of a module.

Similar to ROS2 Lifecycle nodes and IEC 61131-3 PLC states.

INITIALIZING = 'Initializing'

ACTIVE = 'Active'

RECOVERING = 'Recovering'

SUSPENDED = 'Suspended'

SHUTTING_DOWN = 'ShuttingDown'

OFFLINE = 'Offline'

OperationalState

class vyra_base.state.state_types.OperationalState(*values)

Bases: Enum

Operational states define the runtime activity of a module.

Represents what the module is currently doing during normal operation.

IDLE = 'Idle'

READY = 'Ready'

RUNNING = 'Running'

PAUSED = 'Paused'

STOPPED = 'Stopped'

ERROR = 'Error'

HealthState

class vyra_base.state.state_types.HealthState(*values)

Bases: Enum

Health states describe the integrity and error conditions of a module.

Follows safety integrity levels (SIL) from IEC 61508.

HEALTHY = 'Healthy'

WARNING = 'Warning'

CRITICAL = 'Critical'

Event System

StateEvent

class vyra_base.state.state_events.StateEvent(event_type, timestamp=<factory>, origin_layer=None, payload=None, event_id=None)

Bases: object

Immutable event for state machine transitions.

Contains all information about a state transition event including timestamp, origin, and optional payload for tracing and debugging.

Parameter:

• event_type (EventType)

• timestamp (datetime)

• origin_layer (str | None)

• payload (Dict[str, Any] | None)

• event_id (str | None)

event_type: EventType

timestamp: datetime

origin_layer: str | None = None

payload: Dict[str, Any] | None = None

event_id: str | None = None

to_dict()

Convert event to dictionary for serialization.

Rückgabetyp:

Dict[str, Any]

__init__(event_type, timestamp=<factory>, origin_layer=None, payload=None, event_id=None)

Parameter:

• event_type (EventType)

• timestamp (datetime)

• origin_layer (str | None)

• payload (Dict[str, Any] | None)

• event_id (str | None)

Rückgabetyp:

None

EventType

class vyra_base.state.state_events.EventType(*values)

Bases: Enum

Event types for state transitions.

START = 'start'

INIT_SUCCESS = 'init_success'

INIT_FAILURE = 'init_failure'

SET_SUSPENDED = 'set_suspended'

RESUME_SUSPENDED = 'resume_suspended'

SHUTDOWN = 'shutdown'

FINISHED = 'finished'

FAULT_DETECTED = 'fault_detected'

RECOVERY_SUCCESS = 'recovery_success'

RECOVERY_FAILED = 'recovery_failed'

SET_READY = 'set_ready'

TASK_START = 'task_start'

TASK_PAUSE = 'task_pause'

TASK_RESUME = 'task_resume'

TASK_COMPLETE = 'task_complete'

TASK_STOP = 'task_stop'

TASK_RESET = 'task_reset'

TASK_ERROR = 'task_error'

WARN = 'warn'

CLEAR_WARNING = 'clear_warning'

FAULT = 'fault'

RECOVER = 'recover'

RESET = 'reset'

INTERRUPT = 'interrupt'

EMERGENCY_STOP = 'emergency_stop'

PRIORITY_OVERRIDE = 'priority_override'

State Machines

The state machine implementation consists of three layer classes that work together within the UnifiedStateMachine. These are internal implementation details.

For usage, refer to the UnifiedStateMachine class above.

Decorators

The @remote_service decorator is used to mark methods that should be exposed as ROS2 services. It automatically handles state transitions and validation.

Usage:

from vyra_base.com import remote_service

class MyComponent(OperationalStateMachine):
    @remote_service()
    def initialize(self, request=None, response=None):
        return True

Exceptions

InvalidTransitionError

Raised when an invalid state transition is attempted.

from vyra_base.state.state_types import InvalidTransitionError

try:
    state_machine.handle_event(EventType.TASK_START)
except InvalidTransitionError as e:
    print(f"Invalid transition: {e}")

OperationalStateError

Raised when an operational state error occurs.

from vyra_base.state.state_types import OperationalStateError

try:
    component.start()
except OperationalStateError as e:
    print(f"Operational error: {e}")

Complete Module Reference

state.unified

Integration module for easy unified state machine access.

Provides a simple API that combines all three layers into a single unified interface for module state management.

class vyra_base.state.unified.UnifiedStateMachine(config=None)

Bases: object

Unified interface for 3-layer state machine.

Combines Lifecycle, Operational, and Health layers into a single easy-to-use API. Recommended for most use cases.

Example

>>> usm = UnifiedStateMachine()
>>> usm.start()  # Lifecycle
>>> usm.complete_initialization()
>>> usm.set_ready()  # Operational
>>> usm.start_task({'task_id': '123'})
>>> usm.report_warning({'cpu': '85%'})  # Health
>>> usm.get_all_states()
{'lifecycle': 'Active', 'operational': 'Running', 'health': 'Warning'}

Parameter:

config (StateMachineConfig | None)

__init__(config=None)

Initialize unified state machine.

Parameter:

config (Optional[StateMachineConfig]) – Configuration for state machine behavior

get_all_states()

Get current state of all layers.

Rückgabetyp:

Dict[str, str]

get_lifecycle_state()

Get lifecycle state.

Rückgabetyp:

LifecycleState

get_operational_state()

Get operational state.

Rückgabetyp:

OperationalState

get_health_state()

Get health state.

Rückgabetyp:

HealthState

is_operational()

Check if module can execute tasks.

Rückgabetyp:

bool

is_healthy()

Check if module health is OK.

Rückgabetyp:

bool

is_initializing()

Check if module is initializing.

Rückgabetyp:

bool

is_active()

Check if module lifecycle is Active.

Rückgabetyp:

bool

is_suspended()

Check if module is suspended.

Rückgabetyp:

bool

is_recovering()

Check if module is in recovery.

Rückgabetyp:

bool

is_shutting_down()

Check if module is shutting down.

Rückgabetyp:

bool

is_offline()

Check if module is offline.

Rückgabetyp:

bool

start(metadata=None)

Start module initialization.

Rückgabetyp:

LifecycleState

Parameter:

metadata (Dict[str, Any] | None)

complete_initialization(result=None)

Complete initialization successfully.

Rückgabetyp:

LifecycleState

Parameter:

result (Dict[str, Any] | None)

fail_initialization(error=None)

Mark initialization as failed.

Rückgabetyp:

LifecycleState

Parameter:

error (str | None)

shutdown(reason=None)

Begin controlled shutdown.

Rückgabetyp:

LifecycleState

Parameter:

reason (str | None)

complete_shutdown()

Complete shutdown process.

Rückgabetyp:

LifecycleState

suspend(reason=None)

Suspend the module temporarily (e.g. for maintenance or updates).

Transitions: Active → Suspended

Parameter:

reason (Optional[str]) – Optional reason for suspension

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

resume_from_suspend(info=None)

Resume module from Suspended state.

Transitions: Suspended → Active

Parameter:

info (Optional[Dict[str, Any]]) – Optional context about the resume

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

enter_recovery(fault_info=None)

Enter recovery mode due to fault.

Transitions: Active → Recovering

Parameter:

fault_info (Optional[Dict[str, Any]]) – Fault description and context

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

complete_recovery(recovery_info=None)

Complete recovery successfully.

Transitions: Recovering → Active

Parameter:

recovery_info (Optional[Dict[str, Any]]) – Recovery details

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

fail_recovery(error=None)

Mark recovery as failed.

Transitions: Recovering → ShuttingDown

Parameter:

error (Optional[str]) – Recovery failure reason

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

set_ready(metadata=None)

Signal readiness for tasks.

Rückgabetyp:

OperationalState

Parameter:

metadata (Dict[str, Any] | None)

start_task(task_info=None)

Start task execution.

Rückgabetyp:

OperationalState

Parameter:

task_info (Dict[str, Any] | None)

pause(reason=None)

Pause current task.

Rückgabetyp:

OperationalState

Parameter:

reason (str | None)

resume()

Resume paused task.

Rückgabetyp:

OperationalState

stop(result=None)

Mark task as completed.

Rückgabetyp:

OperationalState

Parameter:

result (Dict[str, Any] | None)

reset()

Reset operational state to Idle.

Rückgabetyp:

OperationalState

report_warning(warning_info=None)

Report non-critical warning.

Rückgabetyp:

HealthState

Parameter:

warning_info (Dict[str, Any] | None)

report_fault(fault_info=None)

Report critical fault.

Rückgabetyp:

HealthState

Parameter:

fault_info (Dict[str, Any] | None)

recover(recovery_info=None)

Attempt recovery from fault.

Rückgabetyp:

HealthState

Parameter:

recovery_info (Dict[str, Any] | None)

emergency_stop(reason)

Trigger emergency stop.

Rückgabetyp:

Dict[str, str]

Parameter:

reason (str)

on_lifecycle_change(callback, priority=0)

Subscribe to lifecycle state changes.

Parameter:

• callback (Callable)

• priority (int)

on_operational_change(callback, priority=0)

Subscribe to operational state changes.

Parameter:

• callback (Callable)

• priority (int)

on_health_change(callback, priority=0)

Subscribe to health state changes.

Parameter:

• callback (Callable)

• priority (int)

on_any_change(callback, priority=0)

Subscribe to any state change across all layers.

Parameter:

• callback (Callable)

• priority (int)

get_diagnostic_info()

Get comprehensive diagnostic information.

Rückgabetyp:

Dict[str, Any]

get_history(limit=None)

Get state transition history.

Parameter:

limit (int | None)

clear_history()

Clear transition history.

send_event(event)

Send an event to the state machine.

This method is required for compatibility with OperationalStateMachine which calls _state_machine.send_event(event).

Parameter:

event – StateEvent to send to the FSM

standard_startup()

Execute standard module startup sequence.

Sequence: 1. Start initialization 2. Complete initialization 3. Set operational ready

Rückgabetyp:

bool

Rückgabe:

True if startup successful, False otherwise

standard_shutdown()

Execute standard module shutdown sequence.

Sequence: 1. Begin shutdown 2. Complete any pending tasks (module responsibility) 3. Complete shutdown

Rückgabetyp:

bool

Rückgabe:

True if shutdown successful, False otherwise

state.lifecycle_layer

Lifecycle Layer - High-level API for module lifecycle management.

This layer controls module existence states: - Startup and initialization - Activation and deactivation - Recovery from failures - Controlled shutdown

Thread-safe wrapper around StateMachine for lifecycle operations.

class vyra_base.state.lifecycle_layer.LifecycleLayer(fsm=None)

Bases: object

High-level API for lifecycle state management.

Provides intuitive methods for module lifecycle control: - start() - Begin initialization - activate() - Enter active state - shutdown() - Controlled deactivation - recover() - Recover from faults

Example

>>> lifecycle = LifecycleLayer()
>>> lifecycle.start()
>>> lifecycle.complete_initialization()
>>> lifecycle.get_state()
'Active'

Parameter:

fsm (StateMachine | None)

__init__(fsm=None)

Initialize lifecycle layer.

Parameter:

fsm (Optional[StateMachine]) – Existing StateMachine instance. Creates new if None.

get_state()

Get current lifecycle state.

Rückgabetyp:

LifecycleState

get_state_name()

Get current lifecycle state as string.

Rückgabetyp:

str

is_initializing()

Check if module is initializing.

Rückgabetyp:

bool

is_active()

Check if module is active.

Rückgabetyp:

bool

is_recovering()

Check if module is recovering.

Rückgabetyp:

bool

is_shutting_down()

Check if module is shutting down.

Rückgabetyp:

bool

is_offline()

Check if module is offline.

Rückgabetyp:

bool

is_suspended()

Check if module is suspended.

Rückgabetyp:

bool

can_accept_tasks()

Check if module can accept operational tasks.

Rückgabetyp:

bool

start(metadata=None)

Start module initialization.

Transitions: Offline → Initializing

Parameter:

metadata (Optional[Dict[str, Any]]) – Optional initialization parameters

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

Verursacht:

InvalidTransitionError – If not in Offline state

complete_initialization(result=None)

Mark initialization as successful.

Transitions: Initializing → Active

Parameter:

result (Optional[Dict[str, Any]]) – Optional initialization results

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

fail_initialization(error=None)

Mark initialization as failed.

Transitions: Initializing → Recovering

Parameter:

error (Optional[str]) – Error description

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

shutdown(reason=None)

Begin controlled shutdown.

Transitions: Active → ShuttingDown

Parameter:

reason (Optional[str]) – Shutdown reason

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

complete_shutdown()

Complete shutdown process.

Transitions: ShuttingDown → Offline

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

enter_suspend(reason=None)

Suspend the module temporarily (e.g. for maintenance or updates).

Transitions: Active → Suspended

Parameter:

reason (Optional[str]) – Optional reason for suspension

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

Verursacht:

InvalidTransitionError – If not in Active state

resume_from_suspend(info=None)

Resume module from Suspended state.

Transitions: Suspended → Active

Parameter:

info (Optional[dict]) – Optional context about the resume

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

Verursacht:

InvalidTransitionError – If not in Suspended state

enter_recovery(fault_info=None)

Enter recovery mode due to fault.

Transitions: Active → Recovering

Parameter:

fault_info (Optional[Dict[str, Any]]) – Fault description and context

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

complete_recovery(recovery_info=None)

Complete recovery successfully.

Transitions: Recovering → Active

Parameter:

recovery_info (Optional[Dict[str, Any]]) – Recovery details

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

fail_recovery(error=None)

Mark recovery as failed.

Transitions: Recovering → ShuttingDown

Parameter:

error (Optional[str]) – Recovery failure reason

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

on_state_change(callback, priority=0)

Register callback for lifecycle state changes.

Parameter:

• callback – Function(layer, old_state, new_state)

• priority (int) – Callback priority (higher = earlier execution)

get_info()

Get lifecycle layer information.

Rückgabetyp:

Dict[str, Any]

Rückgabe:

Dictionary with state and capability info

state.operational_layer

Operational Layer - High-level API for task and activity management.

This layer controls operational states: - Task lifecycle (ready, running, paused, completed) - Processing and delegation - Blocking and unblocking - Auto-reset mechanisms

Thread-safe wrapper around StateMachine for operational control.

class vyra_base.state.operational_layer.OperationalLayer(fsm)

Bases: object

High-level API for operational state management.

Provides intuitive methods for task control: - ready() - Signal readiness for tasks - start_task() - Begin task execution - pause() / resume() - Pause and resume tasks - complete() - Mark task completion

Example

>>> operational = OperationalLayer(fsm)
>>> operational.ready()
>>> operational.start_task({'task_id': '123'})
>>> operational.get_state()
'Running'

Parameter:

fsm (StateMachine)

__init__(fsm)

Initialize operational layer.

Parameter:

fsm (StateMachine) – StateMachine instance to control

get_state()

Get current operational state.

Rückgabetyp:

OperationalState

get_state_name()

Get current operational state as string.

Rückgabetyp:

str

is_idle()

Check if operational is idle.

Rückgabetyp:

bool

is_ready()

Check if operational is ready for tasks.

Rückgabetyp:

bool

is_running()

Check if task is running.

Rückgabetyp:

bool

is_paused()

Check if task is paused.

Rückgabetyp:

bool

is_stopped()

Check if task is stopped.

Rückgabetyp:

bool

can_start_task()

Check if new task can be started.

Rückgabetyp:

bool

is_busy()

Check if actively working (running, processing, delegating).

Rückgabetyp:

bool

set_ready(metadata=None)

Signal readiness for task execution.

Transitions: Idle → Ready

Parameter:

metadata (Optional[Dict[str, Any]]) – Optional readiness metadata

Rückgabetyp:

OperationalState

Rückgabe:

New operational state

start_task(task_info=None)

Start task execution.

Transitions: Ready → Running

Parameter:

task_info (Optional[Dict[str, Any]]) – Task description and parameters

Rückgabetyp:

OperationalState

Rückgabe:

New operational state

pause(reason=None)

Pause current task.

Transitions: Running → Paused

Parameter:

reason (Optional[str]) – Pause reason

Rückgabetyp:

OperationalState

Rückgabe:

New operational state

resume()

Resume paused task.

Transitions: Paused → Running

Rückgabetyp:

OperationalState

Rückgabe:

New operational state

stop(result=None)

Mark task as completed.

Transitions: Running → Completed

Parameter:

result (Optional[Dict[str, Any]]) – Task results and metrics

Rückgabetyp:

OperationalState

Rückgabe:

New operational state

reset()

Reset from Completed to Ready state.

Transitions: Completed → Ready

Rückgabetyp:

OperationalState

Rückgabe:

New operational state

on_state_change(callback, priority=0)

Register callback for operational state changes.

Parameter:

• callback – Function(layer, old_state, new_state)

• priority (int) – Callback priority (higher = earlier execution)

get_info()

Get operational layer information.

Rückgabetyp:

Dict[str, Any]

Rückgabe:

Dictionary with state and capability info

state.health_layer

Health Layer - High-level API for system health monitoring and management.

This layer monitors system integrity: - Health status tracking (OK, Warning, Overloaded, Faulted, Critical) - Warning and fault reporting - Load management - Recovery coordination - Escalation to lifecycle layer

Thread-safe wrapper around StateMachine for health management.

class vyra_base.state.health_layer.HealthLayer(fsm)

Bases: object

High-level API for health state management.

Provides intuitive methods for health monitoring: - report_warning() - Report non-critical issues - recover() - Attempt recovery - escalate() - Escalate to critical state

Example

>>> health = HealthLayer(fsm)
>>> health.report_warning({'cpu': '85%'})
>>> health.is_degraded()
True
>>> health.clear_warning()

Parameter:

fsm (StateMachine)

__init__(fsm)

Initialize health layer.

Parameter:

fsm (StateMachine) – StateMachine instance to control

get_state()

Get current health state.

Rückgabetyp:

HealthState

get_state_name()

Get current health state as string.

Rückgabetyp:

str

is_healthy()

Check if health is OK.

Rückgabetyp:

bool

is_warning()

Check if there are warnings.

Rückgabetyp:

bool

is_critical()

Check if health is critical.

Rückgabetyp:

bool

is_degraded()

Check if health is degraded (warning or worse).

Rückgabetyp:

bool

is_operational_safe()

Check if safe for operational tasks.

Rückgabetyp:

bool

report_warning(warning_info=None)

Report non-critical warning.

Transitions: OK → Warning

Parameter:

warning_info (Optional[Dict[str, Any]]) – Warning details (metrics, thresholds, etc.)

Rückgabetyp:

HealthState

Rückgabe:

New health state

clear_warning(clearance_info=None)

Clear active warnings.

Transitions: Warning → OK

Parameter:

clearance_info (Optional[Dict[str, Any]]) – Clearance details

Rückgabetyp:

HealthState

Rückgabe:

New health state

report_fault(fault_info=None)

Report critical fault.

Transitions: OK/Warning → Critical

Parameter:

fault_info (Optional[Dict[str, Any]]) – Fault details

Rückgabetyp:

HealthState

Rückgabe:

New health state

recover(recovery_info=None)

Attempt recovery from fault.

Transitions: Critical → OK/Warning

Parameter:

recovery_info (Optional[Dict[str, Any]]) – Recovery details

Rückgabetyp:

HealthState

Rückgabe:

New health state

check_and_report(metrics, warning_threshold=None)

Check metrics and automatically report appropriate health state.

Parameter:

• metrics (Dict[str, Any]) – System metrics to evaluate

• warning_threshold (Optional[float]) – Threshold for warning state

Rückgabetyp:

HealthState

Rückgabe:

New health state after evaluation

Example

>>> health.check_and_report(
...     {'cpu_usage': 0.85},
...     warning_threshold=0.7,
...     overload_threshold=0.9
... )

emergency_stop(reason)

Trigger emergency stop (affects all layers).

This is an interrupt event that immediately: - Sets lifecycle to Deactivated - Sets operational to Idle - Sets health to Warning

Parameter:

reason (str) – Emergency stop reason

Rückgabetyp:

Dict[str, str]

Rückgabe:

New state of all layers

on_state_change(callback, priority=0)

Register callback for health state changes.

Parameter:

• callback – Function(layer, old_state, new_state)

• priority (int) – Callback priority (higher = earlier execution)

get_info()

Get health layer information.

Rückgabetyp:

Dict[str, Any]

Rückgabe:

Dictionary with state and status info

get_severity_level()

Get health severity as numeric level.

Rückgabetyp:

int

Rückgabe:

0 = HEALTHY, 1 = Warning, 2 = Critical

state.state_machine

Core 3-layer state machine engine for industrial automation.

This implementation follows industrial standards: - IEC 61508 (Functional Safety) - IEC 61131-3 (PLC Programming) - ISO 13849 (Safety of Machinery)

Features: - Thread-safe operation with RLock - Event-driven architecture - Comprehensive logging and tracing - Rule-based state transitions - Layer interaction validation

exception vyra_base.state.state_machine.StateMachineError

Bases: Exception

Base exception for state machine errors.

exception vyra_base.state.state_machine.InvalidTransitionError

Bases: StateMachineError

Raised when an invalid state transition is attempted.

exception vyra_base.state.state_machine.LayerViolationError

Bases: StateMachineError

Raised when layer interaction rules are violated.

class vyra_base.state.state_machine.StateTransition(from_state, to_state, layer, event, timestamp=<factory>, success=True, error_message=None)

Bases: object

Record of a state transition for auditing and debugging.

Parameter:

• from_state (str)

• to_state (str)

• layer (str)

• event (StateEvent)

• timestamp (datetime)

• success (bool)

• error_message (str | None)

from_state: str

to_state: str

layer: str

event: StateEvent

timestamp: datetime

success: bool = True

error_message: str | None = None

to_dict()

Convert transition to dictionary.

Rückgabetyp:

Dict[str, Any]

__init__(from_state, to_state, layer, event, timestamp=<factory>, success=True, error_message=None)

Parameter:

• from_state (str)

• to_state (str)

• layer (str)

• event (StateEvent)

• timestamp (datetime)

• success (bool)

• error_message (str | None)

Rückgabetyp:

None

class vyra_base.state.state_machine.StateMachineConfig(initial_lifecycle=LifecycleState.OFFLINE, initial_operational=OperationalState.IDLE, initial_health=HealthState.HEALTHY, enable_transition_log=True, max_history_size=1000, strict_mode=True)

Bases: object

Configuration for state machine behavior.

Parameter:

• initial_lifecycle (LifecycleState)

• initial_operational (OperationalState)

• initial_health (HealthState)

• enable_transition_log (bool)

• max_history_size (int)

• strict_mode (bool)

initial_lifecycle: LifecycleState = 'Offline'

initial_operational: OperationalState = 'Idle'

initial_health: HealthState = 'Healthy'

operational_on_recovery(current_operational=OperationalState.IDLE)

Rückgabetyp:

OperationalState

Parameter:

current_operational (OperationalState)

operational_on_shutdown(current_operational=OperationalState.IDLE)

Rückgabetyp:

OperationalState

Parameter:

current_operational (OperationalState)

enable_transition_log: bool = True

max_history_size: int = 1000

strict_mode: bool = True

__init__(initial_lifecycle=LifecycleState.OFFLINE, initial_operational=OperationalState.IDLE, initial_health=HealthState.HEALTHY, enable_transition_log=True, max_history_size=1000, strict_mode=True)

Parameter:

• initial_lifecycle (LifecycleState)

• initial_operational (OperationalState)

• initial_health (HealthState)

• enable_transition_log (bool)

• max_history_size (int)

• strict_mode (bool)

Rückgabetyp:

None

class vyra_base.state.state_machine.StateMachine(config=None, loop=None)

Bases: object

Professional 3-layer state machine for industrial automation.

Architecture: - Lifecycle Layer: Controls module existence (startup/lifetime/shutdown) - Operational Layer: Manages runtime activity (tasks/processing) - Health Layer: Monitors system integrity (health/warnings/errors)

Layer Interaction Rules: 1. Lifecycle → Operational: Controls allowed states 2. Health → Operational: Regulates behavior 3. Health → Lifecycle: Escalates critical issues 4. Operational cannot directly affect Lifecycle or Health

Thread Safety: All public methods are thread-safe using reentrant locks.

Example

>>> config = StateMachineConfig()
>>> fsm = StateMachine(config)
>>> fsm.send_event(StateEvent(EventType.START))
>>> fsm.get_current_state()
{'lifecycle': 'Initializing', 'operational': 'Idle', 'health': 'OK'}

Parameter:

• config (StateMachineConfig | None)

• loop (AbstractEventLoop)

__init__(config=None, loop=None)

Initialize the state machine.

Parameter:

• config (Optional[StateMachineConfig]) – Configuration object. Uses defaults if None.

• loop (AbstractEventLoop)

get_current_state()

Get current state of all layers.

Rückgabe:

‚lifecycle‘, ‚operational‘, ‚health‘

Rückgabetyp:

Dict[str, str]

get_lifecycle_state()

Get current lifecycle state.

Rückgabetyp:

LifecycleState

get_operational_state()

Get current operational state.

Rückgabetyp:

OperationalState

get_health_state()

Get current health state.

Rückgabetyp:

HealthState

is_active()

Check if module is in Active lifecycle state.

Rückgabetyp:

bool

is_operational()

Check if module can accept operational tasks.

Rückgabetyp:

bool

is_healthy()

Check if module health is OK.

Rückgabetyp:

bool

send_event(event)

Send event to state machine and trigger transitions.

Parameter:

event (StateEvent) – StateEvent object

Rückgabetyp:

Dict[str, str]

Rückgabe:

New state after processing event

Verursacht:

• InvalidTransitionError – In strict mode, if transition is invalid

• LayerViolationError – If layer rules are violated

subscribe(layer, callback, priority=0)

Subscribe to state changes on a specific layer.

Parameter:

• layer (Union[StateType, str]) – Layer name (StateType or str: lifecycle/operational/health/any)

• callback (Callable[[str, str, str], None]) – Function(layer, old_state, new_state)

• priority (int) – Higher priority callbacks are called first

Example

>>> def on_state_change(layer, old, new):
...     print(f"{layer}: {old} -> {new}")
>>> fsm.subscribe(StateType.LIFECYCLE, on_state_change)

unsubscribe(layer, callback)

Remove a callback subscription.

Parameter:

• layer (StateType | str)

• callback (Callable)

get_history(limit=None)

Get state transition history.

Parameter:

limit (Optional[int]) – Maximum number of entries to return (newest first)

Rückgabetyp:

List[StateTransition]

Rückgabe:

List of StateTransition objects

clear_history()

Clear transition history.

get_diagnostic_info()

Get comprehensive diagnostic information.

Rückgabetyp:

Dict[str, Any]

Rückgabe:

Dictionary with current states, history stats, and configuration

state.state_events

Event definitions for the 3-layer state machine.

Events trigger state transitions according to the industrial automation event-driven architecture pattern.

class vyra_base.state.state_events.EventType(*values)

Bases: Enum

Event types for state transitions.

START = 'start'

INIT_SUCCESS = 'init_success'

INIT_FAILURE = 'init_failure'

SET_SUSPENDED = 'set_suspended'

RESUME_SUSPENDED = 'resume_suspended'

SHUTDOWN = 'shutdown'

FINISHED = 'finished'

FAULT_DETECTED = 'fault_detected'

RECOVERY_SUCCESS = 'recovery_success'

RECOVERY_FAILED = 'recovery_failed'

SET_READY = 'set_ready'

TASK_START = 'task_start'

TASK_PAUSE = 'task_pause'

TASK_RESUME = 'task_resume'

TASK_COMPLETE = 'task_complete'

TASK_STOP = 'task_stop'

TASK_RESET = 'task_reset'

TASK_ERROR = 'task_error'

WARN = 'warn'

CLEAR_WARNING = 'clear_warning'

FAULT = 'fault'

RECOVER = 'recover'

RESET = 'reset'

INTERRUPT = 'interrupt'

EMERGENCY_STOP = 'emergency_stop'

PRIORITY_OVERRIDE = 'priority_override'

class vyra_base.state.state_events.StateEvent(event_type, timestamp=<factory>, origin_layer=None, payload=None, event_id=None)

Bases: object

Immutable event for state machine transitions.

Contains all information about a state transition event including timestamp, origin, and optional payload for tracing and debugging.

Parameter:

• event_type (EventType)

• timestamp (datetime)

• origin_layer (str | None)

• payload (Dict[str, Any] | None)

• event_id (str | None)

event_type: EventType

timestamp: datetime

origin_layer: str | None = None

payload: Dict[str, Any] | None = None

event_id: str | None = None

to_dict()

Convert event to dictionary for serialization.

Rückgabetyp:

Dict[str, Any]

__init__(event_type, timestamp=<factory>, origin_layer=None, payload=None, event_id=None)

Parameter:

• event_type (EventType)

• timestamp (datetime)

• origin_layer (str | None)

• payload (Dict[str, Any] | None)

• event_id (str | None)

Rückgabetyp:

None

vyra_base.state.state_events.get_event_target_layer(event_type)

Get the primary layer affected by an event.

Rückgabetyp:

str

Parameter:

event_type (EventType)

vyra_base.state.state_events.is_interrupt_event(event_type)

Check if event is a system interrupt.

Rückgabetyp:

bool

Parameter:

event_type (EventType)

state.state_types

State type definitions for the 3-layer state machine.

This module defines the valid states for each layer according to industrial automation standards (ISO 13849, IEC 61508).

class vyra_base.state.state_types.StateType(*values)

Bases: Enum

Enum for state types.

LIFECYCLE = 'Lifecycle'

OPERATIONAL = 'Operational'

HEALTH = 'Health'

ANY = 'Any'

class vyra_base.state.state_types.LifecycleState(*values)

Bases: Enum

Lifecycle states define the existence and high-level life of a module.

Similar to ROS2 Lifecycle nodes and IEC 61131-3 PLC states.

INITIALIZING = 'Initializing'

ACTIVE = 'Active'

RECOVERING = 'Recovering'

SUSPENDED = 'Suspended'

SHUTTING_DOWN = 'ShuttingDown'

OFFLINE = 'Offline'

class vyra_base.state.state_types.OperationalState(*values)

Bases: Enum

Operational states define the runtime activity of a module.

Represents what the module is currently doing during normal operation.

IDLE = 'Idle'

READY = 'Ready'

RUNNING = 'Running'

PAUSED = 'Paused'

STOPPED = 'Stopped'

ERROR = 'Error'

class vyra_base.state.state_types.HealthState(*values)

Bases: Enum

Health states describe the integrity and error conditions of a module.

Follows safety integrity levels (SIL) from IEC 61508.

HEALTHY = 'Healthy'

WARNING = 'Warning'

CRITICAL = 'Critical'

vyra_base.state.state_types.is_valid_lifecycle_transition(from_state, to_state)

Check if lifecycle transition is valid.

Rückgabetyp:

bool

Parameter:

• from_state (LifecycleState)

• to_state (LifecycleState)

vyra_base.state.state_types.is_valid_operational_transition(from_state, to_state)

Check if operational transition is valid.

Rückgabetyp:

bool

Parameter:

• from_state (OperationalState)

• to_state (OperationalState)

vyra_base.state.state_types.is_valid_health_transition(from_state, to_state)

Check if health transition is valid.

Rückgabetyp:

bool

Parameter:

• from_state (HealthState)

• to_state (HealthState)

vyra_base.state.state_types.is_operational_allowed_in_lifecycle(lifecycle, operational)

Check if operational state is allowed in current lifecycle state.

Rückgabetyp:

bool

Parameter:

• lifecycle (LifecycleState)

• operational (OperationalState)

state.operational_metaclass

Metaclass for automatic operational state management.

This module provides a metaclass that automatically wraps lifecycle methods with state validation and transition logic, following industrial automation best practices.

exception vyra_base.state.operational_metaclass.OperationalStateError

Bases: Exception

Raised when an operation is called in an invalid operational state.

class vyra_base.state.operational_metaclass.MetaOperationalState(name, bases, attrs)

Bases: type

Metaclass that automatically manages operational state transitions.

This metaclass wraps user-defined lifecycle methods and automatically handles:

• Pre-condition state validation

• State transitions before method execution

• Success/failure state transitions after method execution

• Error handling and logging

Supported lifecycle methods and their state transitions:

initialize():

Pre-condition: IDLE

On success: IDLE -> READY (resets operation counter)

On failure: IDLE -> ERROR

pause():

Pre-condition: RUNNING

On success: RUNNING -> PAUSED

On failure: No state change

resume():

Pre-condition: PAUSED

On success: PAUSED -> READY (resets operation counter)

On failure: PAUSED -> ERROR

stop():

Pre-condition: RUNNING, PAUSED

On success: (current) -> STOPPED

On failure: (current) -> ERROR

reset():

Pre-condition: STOPPED, ERROR

On success: (current) -> IDLE

On failure: No state change

Example

>>> class MyModule(OperationalStateMachine):
...     def initialize(self):
...         # Setup hardware, load config, etc.
...         print("Initializing...")
...         return True
...
>>>
>>> module = MyModule(state_machine)
>>> module.initialize()  # Automatically: IDLE -> READY on success

STATE_RULES = {'initialize': {'failure_transition': OperationalState.ERROR, 'pre_transition': None, 'required_states': {OperationalState.IDLE}, 'reset_counter': True, 'success_transition': OperationalState.READY}, 'pause': {'failure_transition': OperationalState.ERROR, 'pre_transition': None, 'required_states': {OperationalState.RUNNING}, 'reset_counter': False, 'success_transition': OperationalState.PAUSED}, 'reset': {'failure_transition': None, 'pre_transition': None, 'required_states': {OperationalState.ERROR, OperationalState.STOPPED}, 'reset_counter': False, 'success_transition': OperationalState.IDLE}, 'resume': {'failure_transition': OperationalState.ERROR, 'pre_transition': None, 'required_states': {OperationalState.PAUSED}, 'reset_counter': True, 'success_transition': OperationalState.READY}, 'stop': {'failure_transition': OperationalState.ERROR, 'pre_transition': None, 'required_states': {OperationalState.PAUSED, OperationalState.RUNNING}, 'reset_counter': False, 'success_transition': OperationalState.STOPPED}}

state.operation_decorator

Operation decorator for automatic state management with reference counting.

This module provides a decorator that can be applied to any method to automatically manage operational state transitions based on active operation count.

class vyra_base.state.operation_decorator.OperationConfig(required_states=None, pre_transition=None, success_transition=None, failure_transition=None, use_reference_counting=True)

Bases: object

Configuration for operation decorator behavior.

Parameter:

• required_states (Set[OperationalState] | None)

• pre_transition (OperationalState | None)

• success_transition (OperationalState | None)

• failure_transition (OperationalState | None)

• use_reference_counting (bool)

__init__(required_states=None, pre_transition=None, success_transition=None, failure_transition=None, use_reference_counting=True)

Initialize operation configuration.

Parameter:

• required_states (Optional[Set[OperationalState]]) – Set of states required before operation can execute

• pre_transition (Optional[OperationalState]) – State to transition to before execution (if not using ref counting)

• success_transition (Optional[OperationalState]) – State to transition to on success (if not using ref counting)

• failure_transition (Optional[OperationalState]) – State to transition to on failure (if not using ref counting)

• use_reference_counting (bool) – If True, use reference counting for READY<->RUNNING transitions

vyra_base.state.operation_decorator.operation(func=None, *, required_states=None, pre_transition=None, success_transition=None, failure_transition=None, use_reference_counting=True)

Decorator for methods that need automatic operational state management.

This decorator wraps a method with state validation and transition logic. When reference counting is enabled (default), it automatically manages READY <-> RUNNING transitions based on the number of active operations.

Default behavior with reference counting: - Pre-condition: READY or RUNNING state - Before execution: If in READY, transition to RUNNING and increment counter - After execution: Decrement counter; if counter reaches 0, transition to READY

Custom behavior (use_reference_counting=False): - Use custom required_states, pre_transition, success/failure transitions

Parameter:

• func (Optional[Callable]) – Function to decorate (provided automatically when used without parentheses)

• required_states (Optional[Set[OperationalState]]) – States that must be active before operation can run

• pre_transition (Optional[OperationalState]) – State to transition to before execution

• success_transition (Optional[OperationalState]) – State to transition to on success

• failure_transition (Optional[OperationalState]) – State to transition to on failure

• use_reference_counting (bool) – Enable automatic reference counting for READY<->RUNNING

Rückgabetyp:

Callable

Example

>>> from vyra_base import state
>>>
>>> class MyModule(state.OperationalStateMachine):
...     @state.operation
...     def process_data(self, data):
...         # This method is automatically wrapped with state management
...         # State will be RUNNING while executing
...         result = self._do_processing(data)
...         return result
...
...     @state.operation(required_states={OperationalState.RUNNING})
...     def critical_task(self):
...         # This can only run when already in RUNNING state
...         pass

Rückgabetyp:

Callable

Rückgabe:

Decorated function with state management

Parameter:

• func (Callable | None)

• required_states (Set[OperationalState] | None)

• pre_transition (OperationalState | None)

• success_transition (OperationalState | None)

• failure_transition (OperationalState | None)

• use_reference_counting (bool)