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'}

Parameters:

config (StateMachineConfig | None)

__init__(config=None)

Initialize unified state machine.

Parameters:

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

get_all_states()

Get current state of all layers.

Return type:

Dict[str, str]

get_lifecycle_state()

Get lifecycle state.

Return type:

LifecycleState

get_operational_state()

Get operational state.

Return type:

OperationalState

get_health_state()

Get health state.

Return type:

HealthState

is_operational()

Check if module can execute tasks.

Return type:

bool

is_healthy()

Check if module health is OK.

Return type:

bool

is_initializing()

Check if module is initializing.

Return type:

bool

is_active()

Check if module lifecycle is Active.

Return type:

bool

is_suspended()

Check if module is suspended.

Return type:

bool

is_recovering()

Check if module is in recovery.

Return type:

bool

is_shutting_down()

Check if module is shutting down.

Return type:

bool

is_offline()

Check if module is offline.

Return type:

bool

start(metadata=None)

Start module initialization.

Return type:

LifecycleState

Parameters:

metadata (Dict[str, Any] | None)

complete_initialization(result=None)

Complete initialization successfully.

Return type:

LifecycleState

Parameters:

result (Dict[str, Any] | None)

fail_initialization(error=None)

Mark initialization as failed.

Return type:

LifecycleState

Parameters:

error (str | None)

shutdown(reason=None)

Begin controlled shutdown.

Return type:

LifecycleState

Parameters:

reason (str | None)

complete_shutdown()

Complete shutdown process.

Return type:

LifecycleState

suspend(reason=None)

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

Transitions: Active → Suspended

Parameters:

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

Return type:

LifecycleState

Returns:

New lifecycle state

resume_from_suspend(info=None)

Resume module from Suspended state.

Transitions: Suspended → Active

Parameters:

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

Return type:

LifecycleState

Returns:

New lifecycle state

enter_recovery(fault_info=None)

Enter recovery mode due to fault.

Transitions: Active → Recovering

Parameters:

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

Return type:

LifecycleState

Returns:

New lifecycle state

complete_recovery(recovery_info=None)

Complete recovery successfully.

Transitions: Recovering → Active

Parameters:

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

Return type:

LifecycleState

Returns:

New lifecycle state

fail_recovery(error=None)

Mark recovery as failed.

Transitions: Recovering → ShuttingDown

Parameters:

error (Optional[str]) – Recovery failure reason

Return type:

LifecycleState

Returns:

New lifecycle state

set_ready(metadata=None)

Signal readiness for tasks.

Return type:

OperationalState

Parameters:

metadata (Dict[str, Any] | None)

start_task(task_info=None)

Start task execution.

Return type:

OperationalState

Parameters:

task_info (Dict[str, Any] | None)

pause(reason=None)

Pause current task.

Return type:

OperationalState

Parameters:

reason (str | None)

resume()

Resume paused task.

Return type:

OperationalState

stop(result=None)

Mark task as completed.

Return type:

OperationalState

Parameters:

result (Dict[str, Any] | None)

reset()

Reset operational state to Idle.

Return type:

OperationalState

report_warning(warning_info=None)

Report non-critical warning.

Return type:

HealthState

Parameters:

warning_info (Dict[str, Any] | None)

report_fault(fault_info=None)

Report critical fault.

Return type:

HealthState

Parameters:

fault_info (Dict[str, Any] | None)

recover(recovery_info=None)

Attempt recovery from fault.

Return type:

HealthState

Parameters:

recovery_info (Dict[str, Any] | None)

emergency_stop(reason)

Trigger emergency stop.

Return type:

Dict[str, str]

Parameters:

reason (str)

on_lifecycle_change(callback, priority=0)

Subscribe to lifecycle state changes.

Parameters:

• callback (Callable)

• priority (int)

on_operational_change(callback, priority=0)

Subscribe to operational state changes.

Parameters:

• callback (Callable)

• priority (int)

on_health_change(callback, priority=0)

Subscribe to health state changes.

Parameters:

• callback (Callable)

• priority (int)

on_any_change(callback, priority=0)

Subscribe to any state change across all layers.

Parameters:

• callback (Callable)

• priority (int)

get_diagnostic_info()

Get comprehensive diagnostic information.

Return type:

Dict[str, Any]

get_history(limit=None)

Get state transition history.

Parameters:

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).

Parameters:

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

Return type:

bool

Returns:

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

Return type:

bool

Returns:

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

Parameters:

state_machine (StateMachine)

__init__(state_machine)

Initialize the operational state machine.

Parameters:

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

get_operational_state()

Get current operational state.

Return type:

OperationalState

get_all_states()

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

Return type:

Dict[str, str]

is_idle()

Check if in IDLE state.

Return type:

bool

is_ready()

Check if in READY state.

Return type:

bool

is_running()

Check if in RUNNING state.

Return type:

bool

is_paused()

Check if in PAUSED state.

Return type:

bool

is_stopped()

Check if in STOPPED state.

Return type:

bool

is_error()

Check if in ERROR state.

Return type:

bool

get_operation_counter()

Get the current operation reference counter value.

Return type:

int

Returns:

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.

Parameters:

• 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.

Return type:

Dict[str, Any]

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

Parameters:

• event_type (EventType)

• timestamp (datetime)

• origin_layer (str | None)

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

• event_id (str | None)

Return type:

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'}

Parameters:

config (StateMachineConfig | None)

__init__(config=None)

Initialize unified state machine.

Parameters:

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

get_all_states()

Get current state of all layers.

Return type:

Dict[str, str]

get_lifecycle_state()

Get lifecycle state.

Return type:

LifecycleState

get_operational_state()

Get operational state.

Return type:

OperationalState

get_health_state()

Get health state.

Return type:

HealthState

is_operational()

Check if module can execute tasks.

Return type:

bool

is_healthy()

Check if module health is OK.

Return type:

bool

is_initializing()

Check if module is initializing.

Return type:

bool

is_active()

Check if module lifecycle is Active.

Return type:

bool

is_suspended()

Check if module is suspended.

Return type:

bool

is_recovering()

Check if module is in recovery.

Return type:

bool

is_shutting_down()

Check if module is shutting down.

Return type:

bool

is_offline()

Check if module is offline.

Return type:

bool

start(metadata=None)

Start module initialization.

Return type:

LifecycleState

Parameters:

metadata (Dict[str, Any] | None)

complete_initialization(result=None)

Complete initialization successfully.

Return type:

LifecycleState

Parameters:

result (Dict[str, Any] | None)

fail_initialization(error=None)

Mark initialization as failed.

Return type:

LifecycleState

Parameters:

error (str | None)

shutdown(reason=None)

Begin controlled shutdown.

Return type:

LifecycleState

Parameters:

reason (str | None)

complete_shutdown()

Complete shutdown process.

Return type:

LifecycleState

suspend(reason=None)

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

Transitions: Active → Suspended

Parameters:

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

Return type:

LifecycleState

Returns:

New lifecycle state

resume_from_suspend(info=None)

Resume module from Suspended state.

Transitions: Suspended → Active

Parameters:

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

Return type:

LifecycleState

Returns:

New lifecycle state

enter_recovery(fault_info=None)

Enter recovery mode due to fault.

Transitions: Active → Recovering

Parameters:

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

Return type:

LifecycleState

Returns:

New lifecycle state

complete_recovery(recovery_info=None)

Complete recovery successfully.

Transitions: Recovering → Active

Parameters:

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

Return type:

LifecycleState

Returns:

New lifecycle state

fail_recovery(error=None)

Mark recovery as failed.

Transitions: Recovering → ShuttingDown

Parameters:

error (Optional[str]) – Recovery failure reason

Return type:

LifecycleState

Returns:

New lifecycle state

set_ready(metadata=None)

Signal readiness for tasks.

Return type:

OperationalState

Parameters:

metadata (Dict[str, Any] | None)

start_task(task_info=None)

Start task execution.

Return type:

OperationalState

Parameters:

task_info (Dict[str, Any] | None)

pause(reason=None)

Pause current task.

Return type:

OperationalState

Parameters:

reason (str | None)

resume()

Resume paused task.

Return type:

OperationalState

stop(result=None)

Mark task as completed.

Return type:

OperationalState

Parameters:

result (Dict[str, Any] | None)

reset()

Reset operational state to Idle.

Return type:

OperationalState

report_warning(warning_info=None)

Report non-critical warning.

Return type:

HealthState

Parameters:

warning_info (Dict[str, Any] | None)

report_fault(fault_info=None)

Report critical fault.

Return type:

HealthState

Parameters:

fault_info (Dict[str, Any] | None)

recover(recovery_info=None)

Attempt recovery from fault.

Return type:

HealthState

Parameters:

recovery_info (Dict[str, Any] | None)

emergency_stop(reason)

Trigger emergency stop.

Return type:

Dict[str, str]

Parameters:

reason (str)

on_lifecycle_change(callback, priority=0)

Subscribe to lifecycle state changes.

Parameters:

• callback (Callable)

• priority (int)

on_operational_change(callback, priority=0)

Subscribe to operational state changes.

Parameters:

• callback (Callable)

• priority (int)

on_health_change(callback, priority=0)

Subscribe to health state changes.

Parameters:

• callback (Callable)

• priority (int)

on_any_change(callback, priority=0)

Subscribe to any state change across all layers.

Parameters:

• callback (Callable)

• priority (int)

get_diagnostic_info()

Get comprehensive diagnostic information.

Return type:

Dict[str, Any]

get_history(limit=None)

Get state transition history.

Parameters:

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).

Parameters:

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

Return type:

bool

Returns:

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

Return type:

bool

Returns:

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'

Parameters:

fsm (StateMachine | None)

__init__(fsm=None)

Initialize lifecycle layer.

Parameters:

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

get_state()

Get current lifecycle state.

Return type:

LifecycleState

get_state_name()

Get current lifecycle state as string.

Return type:

str

is_initializing()

Check if module is initializing.

Return type:

bool

is_active()

Check if module is active.

Return type:

bool

is_recovering()

Check if module is recovering.

Return type:

bool

is_shutting_down()

Check if module is shutting down.

Return type:

bool

is_offline()

Check if module is offline.

Return type:

bool

is_suspended()

Check if module is suspended.

Return type:

bool

can_accept_tasks()

Check if module can accept operational tasks.

Return type:

bool

start(metadata=None)

Start module initialization.

Transitions: Offline → Initializing

Parameters:

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

Return type:

LifecycleState

Returns:

New lifecycle state

Raises:

InvalidTransitionError – If not in Offline state

complete_initialization(result=None)

Mark initialization as successful.

Transitions: Initializing → Active

Parameters:

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

Return type:

LifecycleState

Returns:

New lifecycle state

fail_initialization(error=None)

Mark initialization as failed.

Transitions: Initializing → Recovering

Parameters:

error (Optional[str]) – Error description

Return type:

LifecycleState

Returns:

New lifecycle state

shutdown(reason=None)

Begin controlled shutdown.

Transitions: Active → ShuttingDown

Parameters:

reason (Optional[str]) – Shutdown reason

Return type:

LifecycleState

Returns:

New lifecycle state

complete_shutdown()

Complete shutdown process.

Transitions: ShuttingDown → Offline

Return type:

LifecycleState

Returns:

New lifecycle state

enter_suspend(reason=None)

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

Transitions: Active → Suspended

Parameters:

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

Return type:

LifecycleState

Returns:

New lifecycle state

Raises:

InvalidTransitionError – If not in Active state

resume_from_suspend(info=None)

Resume module from Suspended state.

Transitions: Suspended → Active

Parameters:

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

Return type:

LifecycleState

Returns:

New lifecycle state

Raises:

InvalidTransitionError – If not in Suspended state

enter_recovery(fault_info=None)

Enter recovery mode due to fault.

Transitions: Active → Recovering

Parameters:

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

Return type:

LifecycleState

Returns:

New lifecycle state

complete_recovery(recovery_info=None)

Complete recovery successfully.

Transitions: Recovering → Active

Parameters:

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

Return type:

LifecycleState

Returns:

New lifecycle state

fail_recovery(error=None)

Mark recovery as failed.

Transitions: Recovering → ShuttingDown

Parameters:

error (Optional[str]) – Recovery failure reason

Return type:

LifecycleState

Returns:

New lifecycle state

on_state_change(callback, priority=0)

Register callback for lifecycle state changes.

Parameters:

• callback – Function(layer, old_state, new_state)

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

get_info()

Get lifecycle layer information.

Return type:

Dict[str, Any]

Returns:

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'

Parameters:

fsm (StateMachine)

__init__(fsm)

Initialize operational layer.

Parameters:

fsm (StateMachine) – StateMachine instance to control

get_state()

Get current operational state.

Return type:

OperationalState

get_state_name()

Get current operational state as string.

Return type:

str

is_idle()

Check if operational is idle.

Return type:

bool

is_ready()

Check if operational is ready for tasks.

Return type:

bool

is_running()

Check if task is running.

Return type:

bool

is_paused()

Check if task is paused.

Return type:

bool

is_stopped()

Check if task is stopped.

Return type:

bool

can_start_task()

Check if new task can be started.

Return type:

bool

is_busy()

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

Return type:

bool

set_ready(metadata=None)

Signal readiness for task execution.

Transitions: Idle → Ready

Parameters:

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

Return type:

OperationalState

Returns:

New operational state

start_task(task_info=None)

Start task execution.

Transitions: Ready → Running

Parameters:

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

Return type:

OperationalState

Returns:

New operational state

pause(reason=None)

Pause current task.

Transitions: Running → Paused

Parameters:

reason (Optional[str]) – Pause reason

Return type:

OperationalState

Returns:

New operational state

resume()

Resume paused task.

Transitions: Paused → Running

Return type:

OperationalState

Returns:

New operational state

stop(result=None)

Mark task as completed.

Transitions: Running → Completed

Parameters:

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

Return type:

OperationalState

Returns:

New operational state

reset()

Reset from Completed to Ready state.

Transitions: Completed → Ready

Return type:

OperationalState

Returns:

New operational state

on_state_change(callback, priority=0)

Register callback for operational state changes.

Parameters:

• callback – Function(layer, old_state, new_state)

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

get_info()

Get operational layer information.

Return type:

Dict[str, Any]

Returns:

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()

Parameters:

fsm (StateMachine)

__init__(fsm)

Initialize health layer.

Parameters:

fsm (StateMachine) – StateMachine instance to control

get_state()

Get current health state.

Return type:

HealthState

get_state_name()

Get current health state as string.

Return type:

str

is_healthy()

Check if health is OK.

Return type:

bool

is_warning()

Check if there are warnings.

Return type:

bool

is_critical()

Check if health is critical.

Return type:

bool

is_degraded()

Check if health is degraded (warning or worse).

Return type:

bool

is_operational_safe()

Check if safe for operational tasks.

Return type:

bool

report_warning(warning_info=None)

Report non-critical warning.

Transitions: OK → Warning

Parameters:

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

Return type:

HealthState

Returns:

New health state

clear_warning(clearance_info=None)

Clear active warnings.

Transitions: Warning → OK

Parameters:

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

Return type:

HealthState

Returns:

New health state

report_fault(fault_info=None)

Report critical fault.

Transitions: OK/Warning → Critical

Parameters:

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

Return type:

HealthState

Returns:

New health state

recover(recovery_info=None)

Attempt recovery from fault.

Transitions: Critical → OK/Warning

Parameters:

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

Return type:

HealthState

Returns:

New health state

check_and_report(metrics, warning_threshold=None)

Check metrics and automatically report appropriate health state.

Parameters:

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

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

Return type:

HealthState

Returns:

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

Parameters:

reason (str) – Emergency stop reason

Return type:

Dict[str, str]

Returns:

New state of all layers

on_state_change(callback, priority=0)

Register callback for health state changes.

Parameters:

• callback – Function(layer, old_state, new_state)

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

get_info()

Get health layer information.

Return type:

Dict[str, Any]

Returns:

Dictionary with state and status info

get_severity_level()

Get health severity as numeric level.

Return type:

int

Returns:

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.

Parameters:

• 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.

Return type:

Dict[str, Any]

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

Parameters:

• from_state (str)

• to_state (str)

• layer (str)

• event (StateEvent)

• timestamp (datetime)

• success (bool)

• error_message (str | None)

Return type:

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.

Parameters:

• 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)

Return type:

OperationalState

Parameters:

current_operational (OperationalState)

operational_on_shutdown(current_operational=OperationalState.IDLE)

Return type:

OperationalState

Parameters:

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)

Parameters:

• initial_lifecycle (LifecycleState)

• initial_operational (OperationalState)

• initial_health (HealthState)

• enable_transition_log (bool)

• max_history_size (int)

• strict_mode (bool)

Return type:

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'}

Parameters:

• config (StateMachineConfig | None)

• loop (AbstractEventLoop)

__init__(config=None, loop=None)

Initialize the state machine.

Parameters:

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

• loop (AbstractEventLoop)

get_current_state()

Get current state of all layers.

Returns:

‘lifecycle’, ‘operational’, ‘health’

Return type:

Dict[str, str]

get_lifecycle_state()

Get current lifecycle state.

Return type:

LifecycleState

get_operational_state()

Get current operational state.

Return type:

OperationalState

get_health_state()

Get current health state.

Return type:

HealthState

is_active()

Check if module is in Active lifecycle state.

Return type:

bool

is_operational()

Check if module can accept operational tasks.

Return type:

bool

is_healthy()

Check if module health is OK.

Return type:

bool

send_event(event)

Send event to state machine and trigger transitions.

Parameters:

event (StateEvent) – StateEvent object

Return type:

Dict[str, str]

Returns:

New state after processing event

Raises:

• 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.

Parameters:

• 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.

Parameters:

• layer (StateType | str)

• callback (Callable)

get_history(limit=None)

Get state transition history.

Parameters:

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

Return type:

List[StateTransition]

Returns:

List of StateTransition objects

clear_history()

Clear transition history.

get_diagnostic_info()

Get comprehensive diagnostic information.

Return type:

Dict[str, Any]

Returns:

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.

Parameters:

• 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.

Return type:

Dict[str, Any]

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

Parameters:

• event_type (EventType)

• timestamp (datetime)

• origin_layer (str | None)

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

• event_id (str | None)

Return type:

None

vyra_base.state.state_events.get_event_target_layer(event_type)

Get the primary layer affected by an event.

Return type:

str

Parameters:

event_type (EventType)

vyra_base.state.state_events.is_interrupt_event(event_type)

Check if event is a system interrupt.

Return type:

bool

Parameters:

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.

Return type:

bool

Parameters:

• 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.

Return type:

bool

Parameters:

• 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.

Return type:

bool

Parameters:

• 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.

Return type:

bool

Parameters:

• 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.

Parameters:

• 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.

Parameters:

• 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

Parameters:

• 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

Return type:

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

Return type:

Callable

Returns:

Decorated function with state management

Parameters:

• func (Callable | None)

• required_states (Set[OperationalState] | None)

• pre_transition (OperationalState | None)

• success_transition (OperationalState | None)

• failure_transition (OperationalState | None)

• use_reference_counting (bool)