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)

Bases: object

High-level API for health state management.

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

Example

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

Parameter:

fsm (StateMachine)

__init__(fsm)

Initialize health layer.

Parameter:

fsm (StateMachine) – StateMachine instance to control

get_state()

Get current health state.

Rückgabetyp:

HealthState

get_state_name()

Get current health state as string.

Rückgabetyp:

str

is_healthy()

Check if health is OK.

Rückgabetyp:

bool

is_warning()

Check if there are warnings.

Rückgabetyp:

bool

is_critical()

Check if health is critical.

Rückgabetyp:

bool

is_degraded()

Check if health is degraded (warning or worse).

Rückgabetyp:

bool

is_operational_safe()

Check if safe for operational tasks.

Rückgabetyp:

bool

report_warning(warning_info=None)

Report non-critical warning.

Transitions: OK → Warning

Parameter:

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

Rückgabetyp:

HealthState

Rückgabe:

New health state

clear_warning(clearance_info=None)

Clear active warnings.

Transitions: Warning → OK

Parameter:

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

Rückgabetyp:

HealthState

Rückgabe:

New health state

report_fault(fault_info=None)

Report critical fault.

Transitions: OK/Warning → Critical

Parameter:

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

Rückgabetyp:

HealthState

Rückgabe:

New health state

recover(recovery_info=None)

Attempt recovery from fault.

Transitions: Critical → OK/Warning

Parameter:

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

Rückgabetyp:

HealthState

Rückgabe:

New health state

check_and_report(metrics, warning_threshold=None)

Check metrics and automatically report appropriate health state.

Parameter:

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

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

Rückgabetyp:

HealthState

Rückgabe:

New health state after evaluation

Example

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

emergency_stop(reason)

Trigger emergency stop (affects all layers).

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

Parameter:

reason (str) – Emergency stop reason

Rückgabetyp:

Dict[str, str]

Rückgabe:

New state of all layers

on_state_change(callback, priority=0)

Register callback for health state changes.

Parameter:

• callback – Function(layer, old_state, new_state)

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

get_info()

Get health layer information.

Rückgabetyp:

Dict[str, Any]

Rückgabe:

Dictionary with state and status info

get_severity_level()

Get health severity as numeric level.

Rückgabetyp:

int

Rückgabe:

0 = HEALTHY, 1 = Warning, 2 = Critical

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)

Bases: object

High-level API for lifecycle state management.

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

Example

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

Parameter:

fsm (StateMachine | None)

__init__(fsm=None)

Initialize lifecycle layer.

Parameter:

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

get_state()

Get current lifecycle state.

Rückgabetyp:

LifecycleState

get_state_name()

Get current lifecycle state as string.

Rückgabetyp:

str

is_initializing()

Check if module is initializing.

Rückgabetyp:

bool

is_active()

Check if module is active.

Rückgabetyp:

bool

is_recovering()

Check if module is recovering.

Rückgabetyp:

bool

is_shutting_down()

Check if module is shutting down.

Rückgabetyp:

bool

is_offline()

Check if module is offline.

Rückgabetyp:

bool

is_suspended()

Check if module is suspended.

Rückgabetyp:

bool

can_accept_tasks()

Check if module can accept operational tasks.

Rückgabetyp:

bool

start(metadata=None)

Start module initialization.

Transitions: Offline → Initializing

Parameter:

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

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

Verursacht:

InvalidTransitionError – If not in Offline state

complete_initialization(result=None)

Mark initialization as successful.

Transitions: Initializing → Active

Parameter:

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

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

fail_initialization(error=None)

Mark initialization as failed.

Transitions: Initializing → Recovering

Parameter:

error (Optional[str]) – Error description

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

shutdown(reason=None)

Begin controlled shutdown.

Transitions: Active → ShuttingDown

Parameter:

reason (Optional[str]) – Shutdown reason

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

complete_shutdown()

Complete shutdown process.

Transitions: ShuttingDown → Offline

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

enter_suspend(reason=None)

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

Transitions: Active → Suspended

Parameter:

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

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

Verursacht:

InvalidTransitionError – If not in Active state

resume_from_suspend(info=None)

Resume module from Suspended state.

Transitions: Suspended → Active

Parameter:

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

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

Verursacht:

InvalidTransitionError – If not in Suspended state

enter_recovery(fault_info=None)

Enter recovery mode due to fault.

Transitions: Active → Recovering

Parameter:

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

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

complete_recovery(recovery_info=None)

Complete recovery successfully.

Transitions: Recovering → Active

Parameter:

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

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

fail_recovery(error=None)

Mark recovery as failed.

Transitions: Recovering → ShuttingDown

Parameter:

error (Optional[str]) – Recovery failure reason

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

on_state_change(callback, priority=0)

Register callback for lifecycle state changes.

Parameter:

• callback – Function(layer, old_state, new_state)

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

get_info()

Get lifecycle layer information.

Rückgabetyp:

Dict[str, Any]

Rückgabe:

Dictionary with state and capability info

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)

Bases: object

High-level API for operational state management.

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

Example

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

Parameter:

fsm (StateMachine)

__init__(fsm)

Initialize operational layer.

Parameter:

fsm (StateMachine) – StateMachine instance to control

get_state()

Get current operational state.

Rückgabetyp:

OperationalState

get_state_name()

Get current operational state as string.

Rückgabetyp:

str

is_idle()

Check if operational is idle.

Rückgabetyp:

bool

is_ready()

Check if operational is ready for tasks.

Rückgabetyp:

bool

is_running()

Check if task is running.

Rückgabetyp:

bool

is_paused()

Check if task is paused.

Rückgabetyp:

bool

is_stopped()

Check if task is stopped.

Rückgabetyp:

bool

can_start_task()

Check if new task can be started.

Rückgabetyp:

bool

is_busy()

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

Rückgabetyp:

bool

set_ready(metadata=None)

Signal readiness for task execution.

Transitions: Idle → Ready

Parameter:

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

Rückgabetyp:

OperationalState

Rückgabe:

New operational state

start_task(task_info=None)

Start task execution.

Transitions: Ready → Running

Parameter:

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

Rückgabetyp:

OperationalState

Rückgabe:

New operational state

pause(reason=None)

Pause current task.

Transitions: Running → Paused

Parameter:

reason (Optional[str]) – Pause reason

Rückgabetyp:

OperationalState

Rückgabe:

New operational state

resume()

Resume paused task.

Transitions: Paused → Running

Rückgabetyp:

OperationalState

Rückgabe:

New operational state

stop(result=None)

Mark task as completed.

Transitions: Running → Completed

Parameter:

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

Rückgabetyp:

OperationalState

Rückgabe:

New operational state

reset()

Reset from Completed to Ready state.

Transitions: Completed → Ready

Rückgabetyp:

OperationalState

Rückgabe:

New operational state

on_state_change(callback, priority=0)

Register callback for operational state changes.

Parameter:

• callback – Function(layer, old_state, new_state)

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

get_info()

Get operational layer information.

Rückgabetyp:

Dict[str, Any]

Rückgabe:

Dictionary with state and capability info

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)

Bases: Enum

Event types for state transitions.

START = 'start'

INIT_SUCCESS = 'init_success'

INIT_FAILURE = 'init_failure'

SET_SUSPENDED = 'set_suspended'

RESUME_SUSPENDED = 'resume_suspended'

SHUTDOWN = 'shutdown'

FINISHED = 'finished'

FAULT_DETECTED = 'fault_detected'

RECOVERY_SUCCESS = 'recovery_success'

RECOVERY_FAILED = 'recovery_failed'

SET_READY = 'set_ready'

TASK_START = 'task_start'

TASK_PAUSE = 'task_pause'

TASK_RESUME = 'task_resume'

TASK_COMPLETE = 'task_complete'

TASK_STOP = 'task_stop'

TASK_RESET = 'task_reset'

TASK_ERROR = 'task_error'

WARN = 'warn'

CLEAR_WARNING = 'clear_warning'

FAULT = 'fault'

RECOVER = 'recover'

RESET = 'reset'

INTERRUPT = 'interrupt'

EMERGENCY_STOP = 'emergency_stop'

PRIORITY_OVERRIDE = 'priority_override'

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

Bases: object

Immutable event for state machine transitions.

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

Parameter:

• event_type (EventType)

• timestamp (datetime)

• origin_layer (str | None)

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

• event_id (str | None)

event_type: EventType

timestamp: datetime

origin_layer: str | None = None

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

event_id: str | None = None

to_dict()

Convert event to dictionary for serialization.

Rückgabetyp:

Dict[str, Any]

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

Parameter:

• event_type (EventType)

• timestamp (datetime)

• origin_layer (str | None)

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

• event_id (str | None)

Rückgabetyp:

None

vyra_base.state.state_events.get_event_target_layer(event_type)

Get the primary layer affected by an event.

Rückgabetyp:

str

Parameter:

event_type (EventType)

vyra_base.state.state_events.is_interrupt_event(event_type)

Check if event is a system interrupt.

Rückgabetyp:

bool

Parameter:

event_type (EventType)

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

Bases: Exception

Base exception for state machine errors.

exception vyra_base.state.state_machine.InvalidTransitionError

Bases: StateMachineError

Raised when an invalid state transition is attempted.

exception vyra_base.state.state_machine.LayerViolationError

Bases: StateMachineError

Raised when layer interaction rules are violated.

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

Bases: object

Record of a state transition for auditing and debugging.

Parameter:

• from_state (str)

• to_state (str)

• layer (str)

• event (StateEvent)

• timestamp (datetime)

• success (bool)

• error_message (str | None)

from_state: str

to_state: str

layer: str

event: StateEvent

timestamp: datetime

success: bool = True

error_message: str | None = None

to_dict()

Convert transition to dictionary.

Rückgabetyp:

Dict[str, Any]

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

Parameter:

• from_state (str)

• to_state (str)

• layer (str)

• event (StateEvent)

• timestamp (datetime)

• success (bool)

• error_message (str | None)

Rückgabetyp:

None

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

Bases: object

Configuration for state machine behavior.

Parameter:

• initial_lifecycle (LifecycleState)

• initial_operational (OperationalState)

• initial_health (HealthState)

• enable_transition_log (bool)

• max_history_size (int)

• strict_mode (bool)

initial_lifecycle: LifecycleState = 'Offline'

initial_operational: OperationalState = 'Idle'

initial_health: HealthState = 'Healthy'

operational_on_recovery(current_operational=OperationalState.IDLE)

Rückgabetyp:

OperationalState

Parameter:

current_operational (OperationalState)

operational_on_shutdown(current_operational=OperationalState.IDLE)

Rückgabetyp:

OperationalState

Parameter:

current_operational (OperationalState)

enable_transition_log: bool = True

max_history_size: int = 1000

strict_mode: bool = True

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

Parameter:

• initial_lifecycle (LifecycleState)

• initial_operational (OperationalState)

• initial_health (HealthState)

• enable_transition_log (bool)

• max_history_size (int)

• strict_mode (bool)

Rückgabetyp:

None

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

Bases: object

Professional 3-layer state machine for industrial automation.

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

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

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

Example

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

Parameter:

• config (StateMachineConfig | None)

• loop (AbstractEventLoop)

__init__(config=None, loop=None)

Initialize the state machine.

Parameter:

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

• loop (AbstractEventLoop)

get_current_state()

Get current state of all layers.

Rückgabe:

‚lifecycle‘, ‚operational‘, ‚health‘

Rückgabetyp:

Dict[str, str]

get_lifecycle_state()

Get current lifecycle state.

Rückgabetyp:

LifecycleState

get_operational_state()

Get current operational state.

Rückgabetyp:

OperationalState

get_health_state()

Get current health state.

Rückgabetyp:

HealthState

is_active()

Check if module is in Active lifecycle state.

Rückgabetyp:

bool

is_operational()

Check if module can accept operational tasks.

Rückgabetyp:

bool

is_healthy()

Check if module health is OK.

Rückgabetyp:

bool

send_event(event)

Send event to state machine and trigger transitions.

Parameter:

event (StateEvent) – StateEvent object

Rückgabetyp:

Dict[str, str]

Rückgabe:

New state after processing event

Verursacht:

• InvalidTransitionError – In strict mode, if transition is invalid

• LayerViolationError – If layer rules are violated

subscribe(layer, callback, priority=0)

Subscribe to state changes on a specific layer.

Parameter:

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

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

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

Example

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

unsubscribe(layer, callback)

Remove a callback subscription.

Parameter:

• layer (StateType | str)

• callback (Callable)

get_history(limit=None)

Get state transition history.

Parameter:

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

Rückgabetyp:

List[StateTransition]

Rückgabe:

List of StateTransition objects

clear_history()

Clear transition history.

get_diagnostic_info()

Get comprehensive diagnostic information.

Rückgabetyp:

Dict[str, Any]

Rückgabe:

Dictionary with current states, history stats, and configuration

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)

Bases: Enum

Enum for state types.

LIFECYCLE = 'Lifecycle'

OPERATIONAL = 'Operational'

HEALTH = 'Health'

ANY = 'Any'

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

Bases: Enum

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

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

INITIALIZING = 'Initializing'

ACTIVE = 'Active'

RECOVERING = 'Recovering'

SUSPENDED = 'Suspended'

SHUTTING_DOWN = 'ShuttingDown'

OFFLINE = 'Offline'

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

Bases: Enum

Operational states define the runtime activity of a module.

Represents what the module is currently doing during normal operation.

IDLE = 'Idle'

READY = 'Ready'

RUNNING = 'Running'

PAUSED = 'Paused'

STOPPED = 'Stopped'

ERROR = 'Error'

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

Bases: Enum

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

Follows safety integrity levels (SIL) from IEC 61508.

HEALTHY = 'Healthy'

WARNING = 'Warning'

CRITICAL = 'Critical'

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

Check if lifecycle transition is valid.

Rückgabetyp:

bool

Parameter:

• from_state (LifecycleState)

• to_state (LifecycleState)

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

Check if operational transition is valid.

Rückgabetyp:

bool

Parameter:

• from_state (OperationalState)

• to_state (OperationalState)

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

Check if health transition is valid.

Rückgabetyp:

bool

Parameter:

• from_state (HealthState)

• to_state (HealthState)

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

Check if operational state is allowed in current lifecycle state.

Rückgabetyp:

bool

Parameter:

• lifecycle (LifecycleState)

• operational (OperationalState)

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)

Bases: object

Unified interface for 3-layer state machine.

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

Example

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

Parameter:

config (StateMachineConfig | None)

__init__(config=None)

Initialize unified state machine.

Parameter:

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

get_all_states()

Get current state of all layers.

Rückgabetyp:

Dict[str, str]

get_lifecycle_state()

Get lifecycle state.

Rückgabetyp:

LifecycleState

get_operational_state()

Get operational state.

Rückgabetyp:

OperationalState

get_health_state()

Get health state.

Rückgabetyp:

HealthState

is_operational()

Check if module can execute tasks.

Rückgabetyp:

bool

is_healthy()

Check if module health is OK.

Rückgabetyp:

bool

is_initializing()

Check if module is initializing.

Rückgabetyp:

bool

is_active()

Check if module lifecycle is Active.

Rückgabetyp:

bool

is_suspended()

Check if module is suspended.

Rückgabetyp:

bool

is_recovering()

Check if module is in recovery.

Rückgabetyp:

bool

is_shutting_down()

Check if module is shutting down.

Rückgabetyp:

bool

is_offline()

Check if module is offline.

Rückgabetyp:

bool

start(metadata=None)

Start module initialization.

Rückgabetyp:

LifecycleState

Parameter:

metadata (Dict[str, Any] | None)

complete_initialization(result=None)

Complete initialization successfully.

Rückgabetyp:

LifecycleState

Parameter:

result (Dict[str, Any] | None)

fail_initialization(error=None)

Mark initialization as failed.

Rückgabetyp:

LifecycleState

Parameter:

error (str | None)

shutdown(reason=None)

Begin controlled shutdown.

Rückgabetyp:

LifecycleState

Parameter:

reason (str | None)

complete_shutdown()

Complete shutdown process.

Rückgabetyp:

LifecycleState

suspend(reason=None)

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

Transitions: Active → Suspended

Parameter:

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

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

resume_from_suspend(info=None)

Resume module from Suspended state.

Transitions: Suspended → Active

Parameter:

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

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

enter_recovery(fault_info=None)

Enter recovery mode due to fault.

Transitions: Active → Recovering

Parameter:

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

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

complete_recovery(recovery_info=None)

Complete recovery successfully.

Transitions: Recovering → Active

Parameter:

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

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

fail_recovery(error=None)

Mark recovery as failed.

Transitions: Recovering → ShuttingDown

Parameter:

error (Optional[str]) – Recovery failure reason

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

set_ready(metadata=None)

Signal readiness for tasks.

Rückgabetyp:

OperationalState

Parameter:

metadata (Dict[str, Any] | None)

start_task(task_info=None)

Start task execution.

Rückgabetyp:

OperationalState

Parameter:

task_info (Dict[str, Any] | None)

pause(reason=None)

Pause current task.

Rückgabetyp:

OperationalState

Parameter:

reason (str | None)

resume()

Resume paused task.

Rückgabetyp:

OperationalState

stop(result=None)

Mark task as completed.

Rückgabetyp:

OperationalState

Parameter:

result (Dict[str, Any] | None)

reset()

Reset operational state to Idle.

Rückgabetyp:

OperationalState

report_warning(warning_info=None)

Report non-critical warning.

Rückgabetyp:

HealthState

Parameter:

warning_info (Dict[str, Any] | None)

report_fault(fault_info=None)

Report critical fault.

Rückgabetyp:

HealthState

Parameter:

fault_info (Dict[str, Any] | None)

recover(recovery_info=None)

Attempt recovery from fault.

Rückgabetyp:

HealthState

Parameter:

recovery_info (Dict[str, Any] | None)

emergency_stop(reason)

Trigger emergency stop.

Rückgabetyp:

Dict[str, str]

Parameter:

reason (str)

on_lifecycle_change(callback, priority=0)

Subscribe to lifecycle state changes.

Parameter:

• callback (Callable)

• priority (int)

on_operational_change(callback, priority=0)

Subscribe to operational state changes.

Parameter:

• callback (Callable)

• priority (int)

on_health_change(callback, priority=0)

Subscribe to health state changes.

Parameter:

• callback (Callable)

• priority (int)

on_any_change(callback, priority=0)

Subscribe to any state change across all layers.

Parameter:

• callback (Callable)

• priority (int)

get_diagnostic_info()

Get comprehensive diagnostic information.

Rückgabetyp:

Dict[str, Any]

get_history(limit=None)

Get state transition history.

Parameter:

limit (int | None)

clear_history()

Clear transition history.

send_event(event)

Send an event to the state machine.

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

Parameter:

event – StateEvent to send to the FSM

standard_startup()

Execute standard module startup sequence.

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

Rückgabetyp:

bool

Rückgabe:

True if startup successful, False otherwise

standard_shutdown()

Execute standard module shutdown sequence.

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

Rückgabetyp:

bool

Rückgabe:

True if shutdown successful, False otherwise

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)

Bases: object

Professional 3-layer state machine for industrial automation.

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

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

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

Example

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

Parameter:

• config (StateMachineConfig | None)

• loop (AbstractEventLoop)

__init__(config=None, loop=None)

Initialize the state machine.

Parameter:

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

• loop (AbstractEventLoop)

get_current_state()

Get current state of all layers.

Rückgabe:

‚lifecycle‘, ‚operational‘, ‚health‘

Rückgabetyp:

Dict[str, str]

get_lifecycle_state()

Get current lifecycle state.

Rückgabetyp:

LifecycleState

get_operational_state()

Get current operational state.

Rückgabetyp:

OperationalState

get_health_state()

Get current health state.

Rückgabetyp:

HealthState

is_active()

Check if module is in Active lifecycle state.

Rückgabetyp:

bool

is_operational()

Check if module can accept operational tasks.

Rückgabetyp:

bool

is_healthy()

Check if module health is OK.

Rückgabetyp:

bool

send_event(event)

Send event to state machine and trigger transitions.

Parameter:

event (StateEvent) – StateEvent object

Rückgabetyp:

Dict[str, str]

Rückgabe:

New state after processing event

Verursacht:

• InvalidTransitionError – In strict mode, if transition is invalid

• LayerViolationError – If layer rules are violated

subscribe(layer, callback, priority=0)

Subscribe to state changes on a specific layer.

Parameter:

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

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

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

Example

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

unsubscribe(layer, callback)

Remove a callback subscription.

Parameter:

• layer (StateType | str)

• callback (Callable)

get_history(limit=None)

Get state transition history.

Parameter:

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

Rückgabetyp:

List[StateTransition]

Rückgabe:

List of StateTransition objects

clear_history()

Clear transition history.

get_diagnostic_info()

Get comprehensive diagnostic information.

Rückgabetyp:

Dict[str, Any]

Rückgabe:

Dictionary with current states, history stats, and configuration

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)

Bases: object

Configuration for state machine behavior.

Parameter:

• initial_lifecycle (LifecycleState)

• initial_operational (OperationalState)

• initial_health (HealthState)

• enable_transition_log (bool)

• max_history_size (int)

• strict_mode (bool)

initial_lifecycle: LifecycleState = 'Offline'

initial_operational: OperationalState = 'Idle'

initial_health: HealthState = 'Healthy'

operational_on_recovery(current_operational=OperationalState.IDLE)

Rückgabetyp:

OperationalState

Parameter:

current_operational (OperationalState)

operational_on_shutdown(current_operational=OperationalState.IDLE)

Rückgabetyp:

OperationalState

Parameter:

current_operational (OperationalState)

enable_transition_log: bool = True

max_history_size: int = 1000

strict_mode: bool = True

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

Parameter:

• initial_lifecycle (LifecycleState)

• initial_operational (OperationalState)

• initial_health (HealthState)

• enable_transition_log (bool)

• max_history_size (int)

• strict_mode (bool)

Rückgabetyp:

None

exception vyra_base.state.StateMachineError

Bases: Exception

Base exception for state machine errors.

exception vyra_base.state.InvalidTransitionError

Bases: StateMachineError

Raised when an invalid state transition is attempted.

exception vyra_base.state.LayerViolationError

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)

Bases: object

Record of a state transition for auditing and debugging.

Parameter:

• from_state (str)

• to_state (str)

• layer (str)

• event (StateEvent)

• timestamp (datetime)

• success (bool)

• error_message (str | None)

from_state: str

to_state: str

layer: str

event: StateEvent

timestamp: datetime

success: bool = True

error_message: str | None = None

to_dict()

Convert transition to dictionary.

Rückgabetyp:

Dict[str, Any]

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

Parameter:

• from_state (str)

• to_state (str)

• layer (str)

• event (StateEvent)

• timestamp (datetime)

• success (bool)

• error_message (str | None)

Rückgabetyp:

None

class vyra_base.state.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.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.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.is_valid_lifecycle_transition(from_state, to_state)

Check if lifecycle transition is valid.

Rückgabetyp:

bool

Parameter:

• from_state (LifecycleState)

• to_state (LifecycleState)

vyra_base.state.is_valid_operational_transition(from_state, to_state)

Check if operational transition is valid.

Rückgabetyp:

bool

Parameter:

• from_state (OperationalState)

• to_state (OperationalState)

vyra_base.state.is_valid_health_transition(from_state, to_state)

Check if health transition is valid.

Rückgabetyp:

bool

Parameter:

• from_state (HealthState)

• to_state (HealthState)

vyra_base.state.is_operational_allowed_in_lifecycle(lifecycle, operational)

Check if operational state is allowed in current lifecycle state.

Rückgabetyp:

bool

Parameter:

• lifecycle (LifecycleState)

• operational (OperationalState)

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

Bases: object

Immutable event for state machine transitions.

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

Parameter:

• event_type (EventType)

• timestamp (datetime)

• origin_layer (str | None)

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

• event_id (str | None)

event_type: EventType

timestamp: datetime

origin_layer: str | None = None

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

event_id: str | None = None

to_dict()

Convert event to dictionary for serialization.

Rückgabetyp:

Dict[str, Any]

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

Parameter:

• event_type (EventType)

• timestamp (datetime)

• origin_layer (str | None)

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

• event_id (str | None)

Rückgabetyp:

None

class vyra_base.state.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'

vyra_base.state.get_event_target_layer(event_type)

Get the primary layer affected by an event.

Rückgabetyp:

str

Parameter:

event_type (EventType)

vyra_base.state.is_interrupt_event(event_type)

Check if event is a system interrupt.

Rückgabetyp:

bool

Parameter:

event_type (EventType)

class vyra_base.state.LifecycleLayer(fsm=None)

Bases: object

High-level API for lifecycle state management.

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

Example

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

Parameter:

fsm (StateMachine | None)

__init__(fsm=None)

Initialize lifecycle layer.

Parameter:

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

get_state()

Get current lifecycle state.

Rückgabetyp:

LifecycleState

get_state_name()

Get current lifecycle state as string.

Rückgabetyp:

str

is_initializing()

Check if module is initializing.

Rückgabetyp:

bool

is_active()

Check if module is active.

Rückgabetyp:

bool

is_recovering()

Check if module is recovering.

Rückgabetyp:

bool

is_shutting_down()

Check if module is shutting down.

Rückgabetyp:

bool

is_offline()

Check if module is offline.

Rückgabetyp:

bool

is_suspended()

Check if module is suspended.

Rückgabetyp:

bool

can_accept_tasks()

Check if module can accept operational tasks.

Rückgabetyp:

bool

start(metadata=None)

Start module initialization.

Transitions: Offline → Initializing

Parameter:

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

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

Verursacht:

InvalidTransitionError – If not in Offline state

complete_initialization(result=None)

Mark initialization as successful.

Transitions: Initializing → Active

Parameter:

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

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

fail_initialization(error=None)

Mark initialization as failed.

Transitions: Initializing → Recovering

Parameter:

error (Optional[str]) – Error description

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

shutdown(reason=None)

Begin controlled shutdown.

Transitions: Active → ShuttingDown

Parameter:

reason (Optional[str]) – Shutdown reason

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

complete_shutdown()

Complete shutdown process.

Transitions: ShuttingDown → Offline

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

enter_suspend(reason=None)

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

Transitions: Active → Suspended

Parameter:

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

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

Verursacht:

InvalidTransitionError – If not in Active state

resume_from_suspend(info=None)

Resume module from Suspended state.

Transitions: Suspended → Active

Parameter:

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

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

Verursacht:

InvalidTransitionError – If not in Suspended state

enter_recovery(fault_info=None)

Enter recovery mode due to fault.

Transitions: Active → Recovering

Parameter:

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

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

complete_recovery(recovery_info=None)

Complete recovery successfully.

Transitions: Recovering → Active

Parameter:

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

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

fail_recovery(error=None)

Mark recovery as failed.

Transitions: Recovering → ShuttingDown

Parameter:

error (Optional[str]) – Recovery failure reason

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

on_state_change(callback, priority=0)

Register callback for lifecycle state changes.

Parameter:

• callback – Function(layer, old_state, new_state)

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

get_info()

Get lifecycle layer information.

Rückgabetyp:

Dict[str, Any]

Rückgabe:

Dictionary with state and capability info

class vyra_base.state.OperationalLayer(fsm)

Bases: object

High-level API for operational state management.

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

Example

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

Parameter:

fsm (StateMachine)

__init__(fsm)

Initialize operational layer.

Parameter:

fsm (StateMachine) – StateMachine instance to control

get_state()

Get current operational state.

Rückgabetyp:

OperationalState

get_state_name()

Get current operational state as string.

Rückgabetyp:

str

is_idle()

Check if operational is idle.

Rückgabetyp:

bool

is_ready()

Check if operational is ready for tasks.

Rückgabetyp:

bool

is_running()

Check if task is running.

Rückgabetyp:

bool

is_paused()

Check if task is paused.

Rückgabetyp:

bool

is_stopped()

Check if task is stopped.

Rückgabetyp:

bool

can_start_task()

Check if new task can be started.

Rückgabetyp:

bool

is_busy()

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

Rückgabetyp:

bool

set_ready(metadata=None)

Signal readiness for task execution.

Transitions: Idle → Ready

Parameter:

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

Rückgabetyp:

OperationalState

Rückgabe:

New operational state

start_task(task_info=None)

Start task execution.

Transitions: Ready → Running

Parameter:

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

Rückgabetyp:

OperationalState

Rückgabe:

New operational state

pause(reason=None)

Pause current task.

Transitions: Running → Paused

Parameter:

reason (Optional[str]) – Pause reason

Rückgabetyp:

OperationalState

Rückgabe:

New operational state

resume()

Resume paused task.

Transitions: Paused → Running

Rückgabetyp:

OperationalState

Rückgabe:

New operational state

stop(result=None)

Mark task as completed.

Transitions: Running → Completed

Parameter:

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

Rückgabetyp:

OperationalState

Rückgabe:

New operational state

reset()

Reset from Completed to Ready state.

Transitions: Completed → Ready

Rückgabetyp:

OperationalState

Rückgabe:

New operational state

on_state_change(callback, priority=0)

Register callback for operational state changes.

Parameter:

• callback – Function(layer, old_state, new_state)

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

get_info()

Get operational layer information.

Rückgabetyp:

Dict[str, Any]

Rückgabe:

Dictionary with state and capability info

class vyra_base.state.HealthLayer(fsm)

Bases: object

High-level API for health state management.

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

Example

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

Parameter:

fsm (StateMachine)

__init__(fsm)

Initialize health layer.

Parameter:

fsm (StateMachine) – StateMachine instance to control

get_state()

Get current health state.

Rückgabetyp:

HealthState

get_state_name()

Get current health state as string.

Rückgabetyp:

str

is_healthy()

Check if health is OK.

Rückgabetyp:

bool

is_warning()

Check if there are warnings.

Rückgabetyp:

bool

is_critical()

Check if health is critical.

Rückgabetyp:

bool

is_degraded()

Check if health is degraded (warning or worse).

Rückgabetyp:

bool

is_operational_safe()

Check if safe for operational tasks.

Rückgabetyp:

bool

report_warning(warning_info=None)

Report non-critical warning.

Transitions: OK → Warning

Parameter:

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

Rückgabetyp:

HealthState

Rückgabe:

New health state

clear_warning(clearance_info=None)

Clear active warnings.

Transitions: Warning → OK

Parameter:

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

Rückgabetyp:

HealthState

Rückgabe:

New health state

report_fault(fault_info=None)

Report critical fault.

Transitions: OK/Warning → Critical

Parameter:

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

Rückgabetyp:

HealthState

Rückgabe:

New health state

recover(recovery_info=None)

Attempt recovery from fault.

Transitions: Critical → OK/Warning

Parameter:

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

Rückgabetyp:

HealthState

Rückgabe:

New health state

check_and_report(metrics, warning_threshold=None)

Check metrics and automatically report appropriate health state.

Parameter:

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

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

Rückgabetyp:

HealthState

Rückgabe:

New health state after evaluation

Example

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

emergency_stop(reason)

Trigger emergency stop (affects all layers).

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

Parameter:

reason (str) – Emergency stop reason

Rückgabetyp:

Dict[str, str]

Rückgabe:

New state of all layers

on_state_change(callback, priority=0)

Register callback for health state changes.

Parameter:

• callback – Function(layer, old_state, new_state)

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

get_info()

Get health layer information.

Rückgabetyp:

Dict[str, Any]

Rückgabe:

Dictionary with state and status info

get_severity_level()

Get health severity as numeric level.

Rückgabetyp:

int

Rückgabe:

0 = HEALTHY, 1 = Warning, 2 = Critical

class vyra_base.state.UnifiedStateMachine(config=None)

Bases: object

Unified interface for 3-layer state machine.

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

Example

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

Parameter:

config (StateMachineConfig | None)

__init__(config=None)

Initialize unified state machine.

Parameter:

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

get_all_states()

Get current state of all layers.

Rückgabetyp:

Dict[str, str]

get_lifecycle_state()

Get lifecycle state.

Rückgabetyp:

LifecycleState

get_operational_state()

Get operational state.

Rückgabetyp:

OperationalState

get_health_state()

Get health state.

Rückgabetyp:

HealthState

is_operational()

Check if module can execute tasks.

Rückgabetyp:

bool

is_healthy()

Check if module health is OK.

Rückgabetyp:

bool

is_initializing()

Check if module is initializing.

Rückgabetyp:

bool

is_active()

Check if module lifecycle is Active.

Rückgabetyp:

bool

is_suspended()

Check if module is suspended.

Rückgabetyp:

bool

is_recovering()

Check if module is in recovery.

Rückgabetyp:

bool

is_shutting_down()

Check if module is shutting down.

Rückgabetyp:

bool

is_offline()

Check if module is offline.

Rückgabetyp:

bool

start(metadata=None)

Start module initialization.

Rückgabetyp:

LifecycleState

Parameter:

metadata (Dict[str, Any] | None)

complete_initialization(result=None)

Complete initialization successfully.

Rückgabetyp:

LifecycleState

Parameter:

result (Dict[str, Any] | None)

fail_initialization(error=None)

Mark initialization as failed.

Rückgabetyp:

LifecycleState

Parameter:

error (str | None)

shutdown(reason=None)

Begin controlled shutdown.

Rückgabetyp:

LifecycleState

Parameter:

reason (str | None)

complete_shutdown()

Complete shutdown process.

Rückgabetyp:

LifecycleState

suspend(reason=None)

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

Transitions: Active → Suspended

Parameter:

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

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

resume_from_suspend(info=None)

Resume module from Suspended state.

Transitions: Suspended → Active

Parameter:

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

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

enter_recovery(fault_info=None)

Enter recovery mode due to fault.

Transitions: Active → Recovering

Parameter:

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

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

complete_recovery(recovery_info=None)

Complete recovery successfully.

Transitions: Recovering → Active

Parameter:

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

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

fail_recovery(error=None)

Mark recovery as failed.

Transitions: Recovering → ShuttingDown

Parameter:

error (Optional[str]) – Recovery failure reason

Rückgabetyp:

LifecycleState

Rückgabe:

New lifecycle state

set_ready(metadata=None)

Signal readiness for tasks.

Rückgabetyp:

OperationalState

Parameter:

metadata (Dict[str, Any] | None)

start_task(task_info=None)

Start task execution.

Rückgabetyp:

OperationalState

Parameter:

task_info (Dict[str, Any] | None)

pause(reason=None)

Pause current task.

Rückgabetyp:

OperationalState

Parameter:

reason (str | None)

resume()

Resume paused task.

Rückgabetyp:

OperationalState

stop(result=None)

Mark task as completed.

Rückgabetyp:

OperationalState

Parameter:

result (Dict[str, Any] | None)

reset()

Reset operational state to Idle.

Rückgabetyp:

OperationalState

report_warning(warning_info=None)

Report non-critical warning.

Rückgabetyp:

HealthState

Parameter:

warning_info (Dict[str, Any] | None)

report_fault(fault_info=None)

Report critical fault.

Rückgabetyp:

HealthState

Parameter:

fault_info (Dict[str, Any] | None)

recover(recovery_info=None)

Attempt recovery from fault.

Rückgabetyp:

HealthState

Parameter:

recovery_info (Dict[str, Any] | None)

emergency_stop(reason)

Trigger emergency stop.

Rückgabetyp:

Dict[str, str]

Parameter:

reason (str)

on_lifecycle_change(callback, priority=0)

Subscribe to lifecycle state changes.

Parameter:

• callback (Callable)

• priority (int)

on_operational_change(callback, priority=0)

Subscribe to operational state changes.

Parameter:

• callback (Callable)

• priority (int)

on_health_change(callback, priority=0)

Subscribe to health state changes.

Parameter:

• callback (Callable)

• priority (int)

on_any_change(callback, priority=0)

Subscribe to any state change across all layers.

Parameter:

• callback (Callable)

• priority (int)

get_diagnostic_info()

Get comprehensive diagnostic information.

Rückgabetyp:

Dict[str, Any]

get_history(limit=None)

Get state transition history.

Parameter:

limit (int | None)

clear_history()

Clear transition history.

send_event(event)

Send an event to the state machine.

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

Parameter:

event – StateEvent to send to the FSM

standard_startup()

Execute standard module startup sequence.

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

Rückgabetyp:

bool

Rückgabe:

True if startup successful, False otherwise

standard_shutdown()

Execute standard module shutdown sequence.

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

Rückgabetyp:

bool

Rückgabe:

True if shutdown successful, False otherwise

class vyra_base.state.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}}

exception vyra_base.state.OperationalStateError

Bases: Exception

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

class vyra_base.state.OperationalStateMachine(state_machine)

Bases: object

Base class for modules that need automatic operational state management.

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

Subclasses should implement one or more of the following methods:

• initialize(): Setup and initialization logic

• pause(): Pause current operation

• resume(): Resume paused operation

• stop(): Stop current operation

• reset(): Reset to initial state

For dynamic operations, use the @operation decorator.

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

Example

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

Parameter:

state_machine (StateMachine)

__init__(state_machine)

Initialize the operational state machine.

Parameter:

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

get_operational_state()

Get current operational state.

Rückgabetyp:

OperationalState

get_all_states()

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

Rückgabetyp:

Dict[str, str]

is_idle()

Check if in IDLE state.

Rückgabetyp:

bool

is_ready()

Check if in READY state.

Rückgabetyp:

bool

is_running()

Check if in RUNNING state.

Rückgabetyp:

bool

is_paused()

Check if in PAUSED state.

Rückgabetyp:

bool

is_stopped()

Check if in STOPPED state.

Rückgabetyp:

bool

is_error()

Check if in ERROR state.

Rückgabetyp:

bool

get_operation_counter()

Get the current operation reference counter value.

Rückgabetyp:

int

Rückgabe:

Number of currently active operations

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

Decorator for methods that need automatic operational state management.

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

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

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

Parameter:

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

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

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

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

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

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

Rückgabetyp:

Callable

Example

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

Rückgabetyp:

Callable

Rückgabe:

Decorated function with state management

Parameter:

• func (Callable | None)

• required_states (Set[OperationalState] | None)

• pre_transition (OperationalState | None)

• success_transition (OperationalState | None)

• failure_transition (OperationalState | None)

• use_reference_counting (bool)

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

Bases: object

Configuration for operation decorator behavior.

Parameter:

• required_states (Set[OperationalState] | None)

• pre_transition (OperationalState | None)

• success_transition (OperationalState | None)

• failure_transition (OperationalState | None)

• use_reference_counting (bool)

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

Initialize operation configuration.

Parameter:

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

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

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

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

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