Metadata-Version: 2.1
Name: vention-state-machine
Version: 0.1.0
Summary: Declarative state machine framework for machine apps
License: Proprietary
Author: VentionCo
Requires-Python: >=3.9,<3.11
Classifier: License :: Other/Proprietary License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Requires-Dist: asyncio (>=3.4.3,<4.0.0)
Requires-Dist: coverage (>=7.10.1,<8.0.0)
Requires-Dist: fastapi (>=0.116.1,<0.117.0)
Requires-Dist: graphviz (>=0.21,<0.22)
Requires-Dist: httpx (>=0.28.1,<0.29.0)
Requires-Dist: transitions (>=0.9.3,<0.10.0)
Requires-Dist: uvicorn (>=0.35.0,<0.36.0)
Description-Content-Type: text/markdown

# vention-state-machine

A lightweight wrapper around `transitions` for building async-safe, recoverable hierarchical state machines with minimal boilerplate.

## ✨ Features

- Built-in `ready` / `fault` states
- Global transitions: `to_fault`, `reset`
- Optional state recovery (`recover__state`)
- Async task spawning and cancellation
- Timeouts and auto-fault handling
- Transition history recording with timestamps + durations
- Guard conditions for blocking transitions
- Global state change callbacks for logging/MQTT

## 🧠 Domain-Specific Language

This library uses a **declarative domain-specific language (DSL)** to define state machines in a readable and structured way. The key building blocks are:

- **`State`**: Represents a single leaf node in the state machine. Declared as a class attribute.
    
- **`StateGroup`**: A container for related states. It generates a hierarchical namespace for its child states (e.g. `MyGroup.my_state` becomes `"MyGroup_my_state"`).
    
- **`Trigger`**: Defines a named event that can cause state transitions. It is both callable (returns its name as a string) and composable (can generate transition dictionaries via `.transition(...)`).
    
This structure allows you to define states and transitions with strong typing and full IDE support — without strings scattered across your codebase.

For Example: 
```python
class MyStates(StateGroup):
    idle: State = State()
    working: State = State()

class Triggers:
    begin = Trigger("begin")
    finish = Trigger("finish")
```
You can then define transitions declaratively:
```python
TRANSITIONS = [
    Triggers.finish.transition(MyStates.working, MyStates.idle),
]
```

## 🧱 Base States and Triggers

Every machine comes with built-in:

- **States**:
  - `ready`: initial state
  - `fault`: global error state
- **Triggers**:
  - `start`: transition into the first defined state
  - `to_fault`: jump to fault from any state
  - `reset`: recover from fault back to ready

You can reference these via:

```python
from state_machine.core import BaseStates, BaseTriggers

state_machine.trigger(BaseTriggers.RESET.value)
assert state_machine.state == BaseStates.READY.value
```

## 🚀 Quick Start
### 1. Define Your States and Triggers

```python
from state_machine.defs import StateGroup, State, Trigger

class Running(StateGroup):
	picking: State = State()
	placing: State = State()
	homing: State = State()

class States:
	running = Running()

class Triggers:
	start = Trigger("start")
	finished_picking = Trigger("finished_picking")
	finished_placing = Trigger("finished_placing")
	finished_homing = Trigger("finished_homing")
	to_fault = Trigger("to_fault")
	reset = Trigger("reset")
```

### 2. Define Transitions
```python
TRANSITIONS  = [
	Triggers.start.transition("ready", States.running.picking),
	Triggers.finished_picking.transition(States.running.picking, States.running.placing),
	Triggers.finished_placing.transition(States.running.placing, States.running.homing),
	Triggers.finished_homing.transition(States.running.homing, States.running.picking)
]
```

### 3. Implement Your State Machine

```python
from state_machine.core import StateMachine

from state_machine.decorators import on_enter_state, auto_timeout, guard, on_state_change

class CustomMachine(StateMachine):

def __init__(self):
	super().__init__(states=States, transitions=TRANSITIONS)

	# Automatically trigger to_fault after 5s if no progress
	@on_enter_state(States.running.picking)
	@auto_timeout(5.0, Triggers.to_fault)
	def enter_picking(self, _):
		print("🔹 Entering picking")

	@on_enter_state(States.running.placing)
	def enter_placing(self, _):
		print("🔸 Entering placing")
		
	@on_enter_state(States.running.homing)
	def enter_homing(self, _):
		print("🔺 Entering homing")

	# Guard condition - only allow reset when safety conditions are met
	@guard(Triggers.reset)
	def check_safety_conditions(self) -> bool:
		"""Only allow reset when estop is not pressed."""
		return not self.estop_pressed

	# Global state change callback for MQTT publishing
	@on_state_change
	def publish_state_to_mqtt(self, old_state: str, new_state: str, trigger: str):
		"""Publish state changes to MQTT."""
		mqtt_client.publish("machine/state", {
		    "old_state": old_state,
		    "new_state": new_state,
		    "trigger": trigger
		})
```

### 4. Start It
```python
state_machine = StateMachine()
state_machine.start() # Enters last recorded state (if recovery enabled), else first state
```

## 🌐 Optional FastAPI Router
This library provides a FastAPI-compatible router that automatically exposes your state machine over HTTP. This is useful for:

- Triggering transitions via HTTP POST
- Inspecting current state and state history
 #### Example
 ```python
from fastapi import FastAPI
from state_machine.router import build_router
from state_machine.core import StateMachine

state_machine = StateMachine(...)
state_machine.start()

app = FastAPI()
app.include_router(build_router(state_machine))
```

### Available Routes
* `GET /state`: Returns current and last known state
* `GET /history`: Returns list of recent state transitions
* `POST /<trigger_name>`: Triggers a transition by name

You can expose only a subset of triggers by passing them explicitly:
```python
from state_machine.defs import Trigger
# Only create endpoints for 'start' and 'reset'
router = build_router(state_machine, triggers=[Trigger("start"), Trigger("reset")])
```

## Diagram Visualization

- `GET /diagram.svg`: Returns a Graphviz-generated SVG of the state machine.
- Current state is highlighted in red.
- Previous state and the transition taken are highlighted in blue.
- Requires Graphviz installed on the system. If Graphviz is missing, the endpoint returns 503 Service Unavailable.

Example usage:
```bash
curl http://localhost:8000/diagram.svg > machine.svg
open machine.svg
```

## 🧪 Testing & History
-  `state_machine.history`: List of all transitions with timestamps and durations
-  `state_machine.last(n)`: Last `n` transitions
-  `state_machine.record_last_state()`: Manually record current state for later recovery
-  `state_machine.get_last_state()`: Retrieve recorded state
  
## ⏲ Timeout Example
Any `on_enter_state` method can be wrapped with `@auto_timeout(seconds, trigger_fn)`, e.g.:

```python
@auto_timeout(5.0, Triggers.to_fault)
```
This automatically triggers `to_fault()` if the state remains active after 5 seconds.

## 🔁 Recovery Example
Enable `enable_last_state_recovery=True` and use:
```python
state_machine.start()
```
If a last state was recorded, it will trigger `recover__{last_state}` instead of `start`.

## 🧩 How Decorators Work

Decorators attach metadata to your methods:

- `@on_enter_state(state)` binds to the state's entry callback
- `@on_exit_state(state)` binds to the state's exit callback
- `@auto_timeout(seconds, trigger)` schedules a timeout once the state is entered
- `@guard(trigger)` adds a condition that must be true for the transition to proceed
- `@on_state_change` registers a global callback that fires on every state transition

The library automatically discovers and wires these up when your machine is initialized.

## 🛡️ Guard Conditions

Guard conditions allow you to block transitions based on runtime conditions. They can be applied to single or multiple triggers:

```python
# Single trigger
@guard(Triggers.reset)
def check_safety_conditions(self) -> bool:
    """Only allow reset when estop is not pressed."""
    return not self.estop_pressed

# Multiple triggers - same guard applies to both
@guard(Triggers.reset, Triggers.start)
def check_safety_conditions(self) -> bool:
    """Check safety conditions for both reset and start."""
    return not self.estop_pressed and self.safety_system_ok

# Multiple guard functions for the same trigger - ALL must pass
@guard(Triggers.reset)
def check_estop(self) -> bool:
    return not self.estop_pressed

@guard(Triggers.reset)
def check_safety_system(self) -> bool:
    return self.safety_system_ok
```

If any guard function returns `False`, the transition is blocked and the state machine remains in its current state. When multiple guard functions are applied to the same trigger, **ALL conditions must pass** for the transition to be allowed.

## 📡 State Change Callbacks

Global state change callbacks are perfect for logging, MQTT publishing, or other side effects. They fire after every successful state transition:

```python
@on_state_change
def publish_to_mqtt(self, old_state: str, new_state: str, trigger: str) -> None:
    """Publish state changes to MQTT."""
    mqtt_client.publish("machine/state", {
        "old_state": old_state,
        "new_state": new_state,
        "trigger": trigger,
        "timestamp": datetime.now().isoformat()
    })

@on_state_change
def log_transitions(self, old_state: str, new_state: str, trigger: str) -> None:
    """Log all state transitions."""
    print(f"State change: {old_state} -> {new_state} (trigger: {trigger})")
```

Multiple state change callbacks can be registered and they will all be called in the order they were defined. Callbacks only fire on **successful transitions** - blocked transitions (due to guard conditions) do not trigger callbacks.
```
