vyra_base.state package¶

Submodules¶

vyra_base.state.health_layer module¶

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)[source]¶

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)[source]¶

Initialize health layer.

Parameters:: fsm (StateMachine) – StateMachine instance to control

get_state()[source]¶

Get current health state.

Return type:: HealthState

get_state_name()[source]¶

Get current health state as string.

Return type:: str

is_healthy()[source]¶

Check if health is OK.

Return type:: bool

is_warning()[source]¶

Check if there are warnings.

Return type:: bool

is_critical()[source]¶

Check if health is critical.

Return type:: bool

is_degraded()[source]¶

Check if health is degraded (warning or worse).

Return type:: bool

is_operational_safe()[source]¶

Check if safe for operational tasks.

Return type:: bool

report_warning(warning_info=None)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

Parameters:

callback – Function(layer, old_state, new_state)
priority (int) – Callback priority (higher = earlier execution)

get_info()[source]¶

Get health layer information.

Return type:: Dict[str, Any]
Returns:: Dictionary with state and status info

get_severity_level()[source]¶

Get health severity as numeric level.

Return type:: int
Returns:: 0 = HEALTHY, 1 = Warning, 2 = Critical

vyra_base.state.lifecycle_layer module¶

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)[source]¶

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)[source]¶

Initialize lifecycle layer.

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

get_state()[source]¶

Get current lifecycle state.

Return type:: LifecycleState

get_state_name()[source]¶

Get current lifecycle state as string.

Return type:: str

is_initializing()[source]¶

Check if module is initializing.

Return type:: bool

is_active()[source]¶

Check if module is active.

Return type:: bool

is_recovering()[source]¶

Check if module is recovering.

Return type:: bool

is_shutting_down()[source]¶

Check if module is shutting down.

Return type:: bool

is_offline()[source]¶

Check if module is offline.

Return type:: bool

is_suspended()[source]¶

Check if module is suspended.

Return type:: bool

can_accept_tasks()[source]¶

Check if module can accept operational tasks.

Return type:: bool

start(metadata=None)[source]¶

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)[source]¶

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)[source]¶

Mark initialization as failed.

Transitions: Initializing → Recovering

Parameters:: error (Optional[str]) – Error description
Return type:: LifecycleState
Returns:: New lifecycle state

shutdown(reason=None)[source]¶

Begin controlled shutdown.

Transitions: Active → ShuttingDown

Parameters:: reason (Optional[str]) – Shutdown reason
Return type:: LifecycleState
Returns:: New lifecycle state

complete_shutdown()[source]¶

Complete shutdown process.

Transitions: ShuttingDown → Offline

Return type:: LifecycleState
Returns:: New lifecycle state

enter_suspend(reason=None)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

Parameters:

callback – Function(layer, old_state, new_state)
priority (int) – Callback priority (higher = earlier execution)

get_info()[source]¶

Get lifecycle layer information.

Return type:: Dict[str, Any]
Returns:: Dictionary with state and capability info

vyra_base.state.operational_layer module¶

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)[source]¶

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)[source]¶

Initialize operational layer.

Parameters:: fsm (StateMachine) – StateMachine instance to control

get_state()[source]¶

Get current operational state.

Return type:: OperationalState

get_state_name()[source]¶

Get current operational state as string.

Return type:: str

is_idle()[source]¶

Check if operational is idle.

Return type:: bool

is_ready()[source]¶

Check if operational is ready for tasks.

Return type:: bool

is_running()[source]¶

Check if task is running.

Return type:: bool

is_paused()[source]¶

Check if task is paused.

Return type:: bool

is_stopped()[source]¶

Check if task is stopped.

Return type:: bool

can_start_task()[source]¶

Check if new task can be started.

Return type:: bool

is_busy()[source]¶

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

Return type:: bool

set_ready(metadata=None)[source]¶

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)[source]¶

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)[source]¶

Pause current task.

Transitions: Running → Paused

Parameters:: reason (Optional[str]) – Pause reason
Return type:: OperationalState
Returns:: New operational state

resume()[source]¶

Resume paused task.

Transitions: Paused → Running

Return type:: OperationalState
Returns:: New operational state

stop(result=None)[source]¶

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()[source]¶

Reset from Completed to Ready state.

Transitions: Completed → Ready

Return type:: OperationalState
Returns:: New operational state

on_state_change(callback, priority=0)[source]¶

Parameters:

callback – Function(layer, old_state, new_state)
priority (int) – Callback priority (higher = earlier execution)

get_info()[source]¶

Get operational layer information.

Return type:: Dict[str, Any]
Returns:: Dictionary with state and capability info

vyra_base.state.state_events module¶

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)[source]¶

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)[source]¶

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()[source]¶

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)[source]¶

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)[source]¶

Check if event is a system interrupt.

Return type:: bool
Parameters:: event_type (EventType)

vyra_base.state.state_machine module¶

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[source]¶

Bases: Exception

Base exception for state machine errors.

exception vyra_base.state.state_machine.InvalidTransitionError[source]¶

Bases: StateMachineError

Raised when an invalid state transition is attempted.

exception vyra_base.state.state_machine.LayerViolationError[source]¶

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)[source]¶

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()[source]¶

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)[source]¶

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)[source]¶

Return type:: OperationalState
Parameters:: current_operational (OperationalState)

operational_on_shutdown(current_operational=OperationalState.IDLE)[source]¶

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)[source]¶

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)[source]¶

Initialize the state machine.

Parameters:

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

get_current_state()[source]¶

Get current state of all layers.

Returns:: ‘lifecycle’, ‘operational’, ‘health’
Return type:: Dict[str, str]

get_lifecycle_state()[source]¶

Get current lifecycle state.

Return type:: LifecycleState

get_operational_state()[source]¶

Get current operational state.

Return type:: OperationalState

get_health_state()[source]¶

Get current health state.

Return type:: HealthState

is_active()[source]¶

Check if module is in Active lifecycle state.

Return type:: bool

is_operational()[source]¶

Check if module can accept operational tasks.

Return type:: bool

is_healthy()[source]¶

Check if module health is OK.

Return type:: bool

send_event(event)[source]¶

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)[source]¶

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)[source]¶

Remove a callback subscription.

Parameters:

layer (StateType | str)
callback (Callable)

get_history(limit=None)[source]¶

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()[source]¶: Clear transition history.

get_diagnostic_info()[source]¶

Get comprehensive diagnostic information.

Return type:: Dict[str, Any]
Returns:: Dictionary with current states, history stats, and configuration

vyra_base.state.state_types module¶

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)[source]¶

Bases: Enum

Enum for state types.

LIFECYCLE = 'Lifecycle'¶

OPERATIONAL = 'Operational'¶

HEALTH = 'Health'¶

ANY = 'Any'¶

class vyra_base.state.state_types.LifecycleState(*values)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

Check if operational state is allowed in current lifecycle state.

Return type:

bool

Parameters:

lifecycle (LifecycleState)
operational (OperationalState)

vyra_base.state.unified module¶

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)[source]¶

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)[source]¶

Initialize unified state machine.

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

get_all_states()[source]¶

Get current state of all layers.

Return type:: Dict[str, str]

get_lifecycle_state()[source]¶

Get lifecycle state.

Return type:: LifecycleState

get_operational_state()[source]¶

Get operational state.

Return type:: OperationalState

get_health_state()[source]¶

Get health state.

Return type:: HealthState

is_operational()[source]¶

Check if module can execute tasks.

Return type:: bool

is_healthy()[source]¶

Check if module health is OK.

Return type:: bool

is_initializing()[source]¶

Check if module is initializing.

Return type:: bool

is_active()[source]¶

Check if module lifecycle is Active.

Return type:: bool

is_suspended()[source]¶

Check if module is suspended.

Return type:: bool

is_recovering()[source]¶

Check if module is in recovery.

Return type:: bool

is_shutting_down()[source]¶

Check if module is shutting down.

Return type:: bool

is_offline()[source]¶

Check if module is offline.

Return type:: bool

start(metadata=None)[source]¶

Start module initialization.

Return type:: LifecycleState
Parameters:: metadata (Dict[str, Any] | None)

complete_initialization(result=None)[source]¶

Complete initialization successfully.

Return type:: LifecycleState
Parameters:: result (Dict[str, Any] | None)

fail_initialization(error=None)[source]¶

Mark initialization as failed.

Return type:: LifecycleState
Parameters:: error (str | None)

shutdown(reason=None)[source]¶

Begin controlled shutdown.

Return type:: LifecycleState
Parameters:: reason (str | None)

complete_shutdown()[source]¶

Complete shutdown process.

Return type:: LifecycleState

suspend(reason=None)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

Signal readiness for tasks.

Return type:: OperationalState
Parameters:: metadata (Dict[str, Any] | None)

start_task(task_info=None)[source]¶

Start task execution.

Return type:: OperationalState
Parameters:: task_info (Dict[str, Any] | None)

pause(reason=None)[source]¶

Pause current task.

Return type:: OperationalState
Parameters:: reason (str | None)

resume()[source]¶

Resume paused task.

Return type:: OperationalState

stop(result=None)[source]¶

Mark task as completed.

Return type:: OperationalState
Parameters:: result (Dict[str, Any] | None)

reset()[source]¶

Reset operational state to Idle.

Return type:: OperationalState

report_warning(warning_info=None)[source]¶

Report non-critical warning.

Return type:: HealthState
Parameters:: warning_info (Dict[str, Any] | None)

report_fault(fault_info=None)[source]¶

Report critical fault.

Return type:: HealthState
Parameters:: fault_info (Dict[str, Any] | None)

recover(recovery_info=None)[source]¶

Attempt recovery from fault.

Return type:: HealthState
Parameters:: recovery_info (Dict[str, Any] | None)

emergency_stop(reason)[source]¶

Trigger emergency stop.

Return type:: Dict[str, str]
Parameters:: reason (str)

on_lifecycle_change(callback, priority=0)[source]¶

Subscribe to lifecycle state changes.

Parameters:

callback (Callable)
priority (int)

on_operational_change(callback, priority=0)[source]¶

Subscribe to operational state changes.

Parameters:

callback (Callable)
priority (int)

on_health_change(callback, priority=0)[source]¶

Subscribe to health state changes.

Parameters:

callback (Callable)
priority (int)

on_any_change(callback, priority=0)[source]¶

Subscribe to any state change across all layers.

Parameters:

callback (Callable)
priority (int)

get_diagnostic_info()[source]¶

Get comprehensive diagnostic information.

Return type:: Dict[str, Any]

get_history(limit=None)[source]¶

Get state transition history.

Parameters:: limit (int | None)

clear_history()[source]¶: Clear transition history.

send_event(event)[source]¶

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()[source]¶

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()[source]¶

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

Module contents¶

3-Layer State Machine System for VYRA Modules

This module implements a hierarchical state machine with three layers: - Lifecycle Layer: Module existence and initialization states - Operational Layer: Runtime activity states - Health Layer: Diagnostic and error states

Each layer has specific rules about how it can influence other layers: - Lifecycle controls Operational and Health - Health can override Operational and escalate Lifecycle - Operational cannot directly change Lifecycle or Health (only reports events)

Conforms to industrial standards: - IEC 61508 (Functional Safety) - IEC 61131-3 (PLC Programming) - ISO 13849 (Safety of Machinery)

Author: VYRA Development Team License: See LICENSE file

class vyra_base.state.StateMachine(config=None, loop=None)[source]¶

Bases: object

Professional 3-layer state machine for industrial automation.

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)[source]¶

Initialize the state machine.

Parameters:

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

get_current_state()[source]¶

Get current state of all layers.

Returns:: ‘lifecycle’, ‘operational’, ‘health’
Return type:: Dict[str, str]

get_lifecycle_state()[source]¶

Get current lifecycle state.

Return type:: LifecycleState

get_operational_state()[source]¶

Get current operational state.

Return type:: OperationalState

get_health_state()[source]¶

Get current health state.

Return type:: HealthState

is_active()[source]¶

Check if module is in Active lifecycle state.

Return type:: bool

is_operational()[source]¶

Check if module can accept operational tasks.

Return type:: bool

is_healthy()[source]¶

Check if module health is OK.

Return type:: bool

send_event(event)[source]¶

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)[source]¶

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)[source]¶

Remove a callback subscription.

Parameters:

layer (StateType | str)
callback (Callable)

get_history(limit=None)[source]¶

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()[source]¶: Clear transition history.

get_diagnostic_info()[source]¶

Get comprehensive diagnostic information.

Return type:: Dict[str, Any]
Returns:: Dictionary with current states, history stats, and configuration

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

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)[source]¶

Return type:: OperationalState
Parameters:: current_operational (OperationalState)

operational_on_shutdown(current_operational=OperationalState.IDLE)[source]¶

Return type:: OperationalState
Parameters:: current_operational (OperationalState)

enable_transition_log: bool = True¶

max_history_size: int = 1000¶

strict_mode: bool = 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

exception vyra_base.state.StateMachineError[source]¶

Bases: Exception

Base exception for state machine errors.

exception vyra_base.state.InvalidTransitionError[source]¶

Bases: StateMachineError

Raised when an invalid state transition is attempted.

exception vyra_base.state.LayerViolationError[source]¶

Bases: StateMachineError

Raised when layer interaction rules are violated.

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

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()[source]¶

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.LifecycleState(*values)[source]¶

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.OperationalState(*values)[source]¶

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.HealthState(*values)[source]¶

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.is_valid_lifecycle_transition(from_state, to_state)[source]¶

Check if lifecycle transition is valid.

Return type:

bool

Parameters:

from_state (LifecycleState)
to_state (LifecycleState)

vyra_base.state.is_valid_operational_transition(from_state, to_state)[source]¶

Check if operational transition is valid.

Return type:

bool

Parameters:

from_state (OperationalState)
to_state (OperationalState)

vyra_base.state.is_valid_health_transition(from_state, to_state)[source]¶

Check if health transition is valid.

Return type:

bool

Parameters:

from_state (HealthState)
to_state (HealthState)

vyra_base.state.is_operational_allowed_in_lifecycle(lifecycle, operational)[source]¶

Check if operational state is allowed in current lifecycle state.

Return type:

bool

Parameters:

lifecycle (LifecycleState)
operational (OperationalState)

class vyra_base.state.StateEvent(event_type, timestamp=<factory>, origin_layer=None, payload=None, event_id=None)[source]¶

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()[source]¶

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

class vyra_base.state.EventType(*values)[source]¶

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

vyra_base.state.get_event_target_layer(event_type)[source]¶

Get the primary layer affected by an event.

Return type:: str
Parameters:: event_type (EventType)

vyra_base.state.is_interrupt_event(event_type)[source]¶

Check if event is a system interrupt.

Return type:: bool
Parameters:: event_type (EventType)

class vyra_base.state.LifecycleLayer(fsm=None)[source]¶

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)[source]¶

Initialize lifecycle layer.

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

get_state()[source]¶

Get current lifecycle state.

Return type:: LifecycleState

get_state_name()[source]¶

Get current lifecycle state as string.

Return type:: str

is_initializing()[source]¶

Check if module is initializing.

Return type:: bool

is_active()[source]¶

Check if module is active.

Return type:: bool

is_recovering()[source]¶

Check if module is recovering.

Return type:: bool

is_shutting_down()[source]¶

Check if module is shutting down.

Return type:: bool

is_offline()[source]¶

Check if module is offline.

Return type:: bool

is_suspended()[source]¶

Check if module is suspended.

Return type:: bool

can_accept_tasks()[source]¶

Check if module can accept operational tasks.

Return type:: bool

start(metadata=None)[source]¶

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)[source]¶

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)[source]¶

Mark initialization as failed.

Transitions: Initializing → Recovering

Parameters:: error (Optional[str]) – Error description
Return type:: LifecycleState
Returns:: New lifecycle state

shutdown(reason=None)[source]¶

Begin controlled shutdown.

Transitions: Active → ShuttingDown

Parameters:: reason (Optional[str]) – Shutdown reason
Return type:: LifecycleState
Returns:: New lifecycle state

complete_shutdown()[source]¶

Complete shutdown process.

Transitions: ShuttingDown → Offline

Return type:: LifecycleState
Returns:: New lifecycle state

enter_suspend(reason=None)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

Parameters:

callback – Function(layer, old_state, new_state)
priority (int) – Callback priority (higher = earlier execution)

get_info()[source]¶

Get lifecycle layer information.

Return type:: Dict[str, Any]
Returns:: Dictionary with state and capability info

class vyra_base.state.OperationalLayer(fsm)[source]¶

Bases: object

High-level API for operational state management.

Example

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

Parameters:: fsm (StateMachine)

__init__(fsm)[source]¶

Initialize operational layer.

Parameters:: fsm (StateMachine) – StateMachine instance to control

get_state()[source]¶

Get current operational state.

Return type:: OperationalState

get_state_name()[source]¶

Get current operational state as string.

Return type:: str

is_idle()[source]¶

Check if operational is idle.

Return type:: bool

is_ready()[source]¶

Check if operational is ready for tasks.

Return type:: bool

is_running()[source]¶

Check if task is running.

Return type:: bool

is_paused()[source]¶

Check if task is paused.

Return type:: bool

is_stopped()[source]¶

Check if task is stopped.

Return type:: bool

can_start_task()[source]¶

Check if new task can be started.

Return type:: bool

is_busy()[source]¶

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

Return type:: bool

set_ready(metadata=None)[source]¶

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)[source]¶

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)[source]¶

Pause current task.

Transitions: Running → Paused

Parameters:: reason (Optional[str]) – Pause reason
Return type:: OperationalState
Returns:: New operational state

resume()[source]¶

Resume paused task.

Transitions: Paused → Running

Return type:: OperationalState
Returns:: New operational state

stop(result=None)[source]¶

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()[source]¶

Reset from Completed to Ready state.

Transitions: Completed → Ready

Return type:: OperationalState
Returns:: New operational state

on_state_change(callback, priority=0)[source]¶

Parameters:

callback – Function(layer, old_state, new_state)
priority (int) – Callback priority (higher = earlier execution)

get_info()[source]¶

Get operational layer information.

Return type:: Dict[str, Any]
Returns:: Dictionary with state and capability info

class vyra_base.state.HealthLayer(fsm)[source]¶

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)[source]¶

Initialize health layer.

Parameters:: fsm (StateMachine) – StateMachine instance to control

get_state()[source]¶

Get current health state.

Return type:: HealthState

get_state_name()[source]¶

Get current health state as string.

Return type:: str

is_healthy()[source]¶

Check if health is OK.

Return type:: bool

is_warning()[source]¶

Check if there are warnings.

Return type:: bool

is_critical()[source]¶

Check if health is critical.

Return type:: bool

is_degraded()[source]¶

Check if health is degraded (warning or worse).

Return type:: bool

is_operational_safe()[source]¶

Check if safe for operational tasks.

Return type:: bool

report_warning(warning_info=None)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

Parameters:

callback – Function(layer, old_state, new_state)
priority (int) – Callback priority (higher = earlier execution)

get_info()[source]¶

Get health layer information.

Return type:: Dict[str, Any]
Returns:: Dictionary with state and status info

get_severity_level()[source]¶

Get health severity as numeric level.

Return type:: int
Returns:: 0 = HEALTHY, 1 = Warning, 2 = Critical

class vyra_base.state.UnifiedStateMachine(config=None)[source]¶

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)[source]¶

Initialize unified state machine.

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

get_all_states()[source]¶

Get current state of all layers.

Return type:: Dict[str, str]

get_lifecycle_state()[source]¶

Get lifecycle state.

Return type:: LifecycleState

get_operational_state()[source]¶

Get operational state.

Return type:: OperationalState

get_health_state()[source]¶

Get health state.

Return type:: HealthState

is_operational()[source]¶

Check if module can execute tasks.

Return type:: bool

is_healthy()[source]¶

Check if module health is OK.

Return type:: bool

is_initializing()[source]¶

Check if module is initializing.

Return type:: bool

is_active()[source]¶

Check if module lifecycle is Active.

Return type:: bool

is_suspended()[source]¶

Check if module is suspended.

Return type:: bool

is_recovering()[source]¶

Check if module is in recovery.

Return type:: bool

is_shutting_down()[source]¶

Check if module is shutting down.

Return type:: bool

is_offline()[source]¶

Check if module is offline.

Return type:: bool

start(metadata=None)[source]¶

Start module initialization.

Return type:: LifecycleState
Parameters:: metadata (Dict[str, Any] | None)

complete_initialization(result=None)[source]¶

Complete initialization successfully.

Return type:: LifecycleState
Parameters:: result (Dict[str, Any] | None)

fail_initialization(error=None)[source]¶

Mark initialization as failed.

Return type:: LifecycleState
Parameters:: error (str | None)

shutdown(reason=None)[source]¶

Begin controlled shutdown.

Return type:: LifecycleState
Parameters:: reason (str | None)

complete_shutdown()[source]¶

Complete shutdown process.

Return type:: LifecycleState

suspend(reason=None)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

Signal readiness for tasks.

Return type:: OperationalState
Parameters:: metadata (Dict[str, Any] | None)

start_task(task_info=None)[source]¶

Start task execution.

Return type:: OperationalState
Parameters:: task_info (Dict[str, Any] | None)

pause(reason=None)[source]¶

Pause current task.

Return type:: OperationalState
Parameters:: reason (str | None)

resume()[source]¶

Resume paused task.

Return type:: OperationalState

stop(result=None)[source]¶

Mark task as completed.

Return type:: OperationalState
Parameters:: result (Dict[str, Any] | None)

reset()[source]¶

Reset operational state to Idle.

Return type:: OperationalState

report_warning(warning_info=None)[source]¶

Report non-critical warning.

Return type:: HealthState
Parameters:: warning_info (Dict[str, Any] | None)

report_fault(fault_info=None)[source]¶

Report critical fault.

Return type:: HealthState
Parameters:: fault_info (Dict[str, Any] | None)

recover(recovery_info=None)[source]¶

Attempt recovery from fault.

Return type:: HealthState
Parameters:: recovery_info (Dict[str, Any] | None)

emergency_stop(reason)[source]¶

Trigger emergency stop.

Return type:: Dict[str, str]
Parameters:: reason (str)

on_lifecycle_change(callback, priority=0)[source]¶

Subscribe to lifecycle state changes.

Parameters:

callback (Callable)
priority (int)

on_operational_change(callback, priority=0)[source]¶

Subscribe to operational state changes.

Parameters:

callback (Callable)
priority (int)

on_health_change(callback, priority=0)[source]¶

Subscribe to health state changes.

Parameters:

callback (Callable)
priority (int)

on_any_change(callback, priority=0)[source]¶

Subscribe to any state change across all layers.

Parameters:

callback (Callable)
priority (int)

get_diagnostic_info()[source]¶

Get comprehensive diagnostic information.

Return type:: Dict[str, Any]

get_history(limit=None)[source]¶

Get state transition history.

Parameters:: limit (int | None)

clear_history()[source]¶: Clear transition history.

send_event(event)[source]¶

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()[source]¶

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()[source]¶

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

class vyra_base.state.MetaOperationalState(name, bases, attrs)[source]¶

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

exception vyra_base.state.OperationalStateError[source]¶

Bases: Exception

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

class vyra_base.state.OperationalStateMachine(state_machine)[source]¶

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)[source]¶

Initialize the operational state machine.

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

get_operational_state()[source]¶

Get current operational state.

Return type:: OperationalState

get_all_states()[source]¶

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

Return type:: Dict[str, str]

is_idle()[source]¶

Check if in IDLE state.

Return type:: bool

is_ready()[source]¶

Check if in READY state.

Return type:: bool

is_running()[source]¶

Check if in RUNNING state.

Return type:: bool

is_paused()[source]¶

Check if in PAUSED state.

Return type:: bool

is_stopped()[source]¶

Check if in STOPPED state.

Return type:: bool

is_error()[source]¶

Check if in ERROR state.

Return type:: bool

get_operation_counter()[source]¶

Get the current operation reference counter value.

Return type:: int
Returns:: Number of currently active operations

vyra_base.state.operation(func=None, *, required_states=None, pre_transition=None, success_transition=None, failure_transition=None, use_reference_counting=True)[source]¶

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)

class vyra_base.state.OperationConfig(required_states=None, pre_transition=None, success_transition=None, failure_transition=None, use_reference_counting=True)[source]¶

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)[source]¶

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