| Name | vention-state-machine JSON |
| Version |
0.2.0
JSON |
| download |
| home_page | None |
| Summary | Declarative state machine framework for machine apps |
| upload_time | 2025-08-21 18:33:23 |
| maintainer | None |
| docs_url | None |
| author | VentionCo |
| requires_python | <3.11,>=3.9 |
| license | Proprietary |
| keywords |
|
| VCS |
|
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# 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.
```
Raw data
{
"_id": null,
"home_page": null,
"name": "vention-state-machine",
"maintainer": null,
"docs_url": null,
"requires_python": "<3.11,>=3.9",
"maintainer_email": null,
"keywords": null,
"author": "VentionCo",
"author_email": null,
"download_url": "https://files.pythonhosted.org/packages/f6/fd/f729a78707b454c88cfc95d40392a77d615a8cd25654db3b72c7e5fc4180/vention_state_machine-0.2.0.tar.gz",
"platform": null,
"description": "# vention-state-machine\n\nA lightweight wrapper around `transitions` for building async-safe, recoverable hierarchical state machines with minimal boilerplate.\n\n## \u2728 Features\n\n- Built-in `ready` / `fault` states\n- Global transitions: `to_fault`, `reset`\n- Optional state recovery (`recover__state`)\n- Async task spawning and cancellation\n- Timeouts and auto-fault handling\n- Transition history recording with timestamps + durations\n- Guard conditions for blocking transitions\n- Global state change callbacks for logging/MQTT\n\n## \ud83e\udde0 Domain-Specific Language\n\nThis library uses a **declarative domain-specific language (DSL)** to define state machines in a readable and structured way. The key building blocks are:\n\n- **`State`**: Represents a single leaf node in the state machine. Declared as a class attribute.\n \n- **`StateGroup`**: A container for related states. It generates a hierarchical namespace for its child states (e.g. `MyGroup.my_state` becomes `\"MyGroup_my_state\"`).\n \n- **`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(...)`).\n \nThis structure allows you to define states and transitions with strong typing and full IDE support \u2014 without strings scattered across your codebase.\n\nFor Example: \n```python\nclass MyStates(StateGroup):\n idle: State = State()\n working: State = State()\n\nclass Triggers:\n begin = Trigger(\"begin\")\n finish = Trigger(\"finish\")\n```\nYou can then define transitions declaratively:\n```python\nTRANSITIONS = [\n Triggers.finish.transition(MyStates.working, MyStates.idle),\n]\n```\n\n## \ud83e\uddf1 Base States and Triggers\n\nEvery machine comes with built-in:\n\n- **States**:\n - `ready`: initial state\n - `fault`: global error state\n- **Triggers**:\n - `start`: transition into the first defined state\n - `to_fault`: jump to fault from any state\n - `reset`: recover from fault back to ready\n\nYou can reference these via:\n\n```python\nfrom state_machine.core import BaseStates, BaseTriggers\n\nstate_machine.trigger(BaseTriggers.RESET.value)\nassert state_machine.state == BaseStates.READY.value\n```\n\n## \ud83d\ude80 Quick Start\n### 1. Define Your States and Triggers\n\n```python\nfrom state_machine.defs import StateGroup, State, Trigger\n\nclass Running(StateGroup):\n\tpicking: State = State()\n\tplacing: State = State()\n\thoming: State = State()\n\nclass States:\n\trunning = Running()\n\nclass Triggers:\n\tstart = Trigger(\"start\")\n\tfinished_picking = Trigger(\"finished_picking\")\n\tfinished_placing = Trigger(\"finished_placing\")\n\tfinished_homing = Trigger(\"finished_homing\")\n\tto_fault = Trigger(\"to_fault\")\n\treset = Trigger(\"reset\")\n```\n\n### 2. Define Transitions\n```python\nTRANSITIONS = [\n\tTriggers.start.transition(\"ready\", States.running.picking),\n\tTriggers.finished_picking.transition(States.running.picking, States.running.placing),\n\tTriggers.finished_placing.transition(States.running.placing, States.running.homing),\n\tTriggers.finished_homing.transition(States.running.homing, States.running.picking)\n]\n```\n\n### 3. Implement Your State Machine\n\n```python\nfrom state_machine.core import StateMachine\n\nfrom state_machine.decorators import on_enter_state, auto_timeout, guard, on_state_change\n\nclass CustomMachine(StateMachine):\n\ndef __init__(self):\n\tsuper().__init__(states=States, transitions=TRANSITIONS)\n\n\t# Automatically trigger to_fault after 5s if no progress\n\t@on_enter_state(States.running.picking)\n\t@auto_timeout(5.0, Triggers.to_fault)\n\tdef enter_picking(self, _):\n\t\tprint(\"\ud83d\udd39 Entering picking\")\n\n\t@on_enter_state(States.running.placing)\n\tdef enter_placing(self, _):\n\t\tprint(\"\ud83d\udd38 Entering placing\")\n\t\t\n\t@on_enter_state(States.running.homing)\n\tdef enter_homing(self, _):\n\t\tprint(\"\ud83d\udd3a Entering homing\")\n\n\t# Guard condition - only allow reset when safety conditions are met\n\t@guard(Triggers.reset)\n\tdef check_safety_conditions(self) -> bool:\n\t\t\"\"\"Only allow reset when estop is not pressed.\"\"\"\n\t\treturn not self.estop_pressed\n\n\t# Global state change callback for MQTT publishing\n\t@on_state_change\n\tdef publish_state_to_mqtt(self, old_state: str, new_state: str, trigger: str):\n\t\t\"\"\"Publish state changes to MQTT.\"\"\"\n\t\tmqtt_client.publish(\"machine/state\", {\n\t\t \"old_state\": old_state,\n\t\t \"new_state\": new_state,\n\t\t \"trigger\": trigger\n\t\t})\n```\n\n### 4. Start It\n```python\nstate_machine = StateMachine()\nstate_machine.start() # Enters last recorded state (if recovery enabled), else first state\n```\n\n## \ud83c\udf10 Optional FastAPI Router\nThis library provides a FastAPI-compatible router that automatically exposes your state machine over HTTP. This is useful for:\n\n- Triggering transitions via HTTP POST\n- Inspecting current state and state history\n #### Example\n ```python\nfrom fastapi import FastAPI\nfrom state_machine.router import build_router\nfrom state_machine.core import StateMachine\n\nstate_machine = StateMachine(...)\nstate_machine.start()\n\napp = FastAPI()\napp.include_router(build_router(state_machine))\n```\n\n### Available Routes\n* `GET /state`: Returns current and last known state\n* `GET /history`: Returns list of recent state transitions\n* `POST /<trigger_name>`: Triggers a transition by name\n\nYou can expose only a subset of triggers by passing them explicitly:\n```python\nfrom state_machine.defs import Trigger\n# Only create endpoints for 'start' and 'reset'\nrouter = build_router(state_machine, triggers=[Trigger(\"start\"), Trigger(\"reset\")])\n```\n\n## Diagram Visualization\n\n- `GET /diagram.svg`: Returns a Graphviz-generated SVG of the state machine.\n- Current state is highlighted in red.\n- Previous state and the transition taken are highlighted in blue.\n- Requires Graphviz installed on the system. If Graphviz is missing, the endpoint returns 503 Service Unavailable.\n\nExample usage:\n```bash\ncurl http://localhost:8000/diagram.svg > machine.svg\nopen machine.svg\n```\n\n## \ud83e\uddea Testing & History\n- `state_machine.history`: List of all transitions with timestamps and durations\n- `state_machine.last(n)`: Last `n` transitions\n- `state_machine.record_last_state()`: Manually record current state for later recovery\n- `state_machine.get_last_state()`: Retrieve recorded state\n \n## \u23f2 Timeout Example\nAny `on_enter_state` method can be wrapped with `@auto_timeout(seconds, trigger_fn)`, e.g.:\n\n```python\n@auto_timeout(5.0, Triggers.to_fault)\n```\nThis automatically triggers `to_fault()` if the state remains active after 5 seconds.\n\n## \ud83d\udd01 Recovery Example\nEnable `enable_last_state_recovery=True` and use:\n```python\nstate_machine.start()\n```\nIf a last state was recorded, it will trigger `recover__{last_state}` instead of `start`.\n\n## \ud83e\udde9 How Decorators Work\n\nDecorators attach metadata to your methods:\n\n- `@on_enter_state(state)` binds to the state's entry callback\n- `@on_exit_state(state)` binds to the state's exit callback\n- `@auto_timeout(seconds, trigger)` schedules a timeout once the state is entered\n- `@guard(trigger)` adds a condition that must be true for the transition to proceed\n- `@on_state_change` registers a global callback that fires on every state transition\n\nThe library automatically discovers and wires these up when your machine is initialized.\n\n## \ud83d\udee1\ufe0f Guard Conditions\n\nGuard conditions allow you to block transitions based on runtime conditions. They can be applied to single or multiple triggers:\n\n```python\n# Single trigger\n@guard(Triggers.reset)\ndef check_safety_conditions(self) -> bool:\n \"\"\"Only allow reset when estop is not pressed.\"\"\"\n return not self.estop_pressed\n\n# Multiple triggers - same guard applies to both\n@guard(Triggers.reset, Triggers.start)\ndef check_safety_conditions(self) -> bool:\n \"\"\"Check safety conditions for both reset and start.\"\"\"\n return not self.estop_pressed and self.safety_system_ok\n\n# Multiple guard functions for the same trigger - ALL must pass\n@guard(Triggers.reset)\ndef check_estop(self) -> bool:\n return not self.estop_pressed\n\n@guard(Triggers.reset)\ndef check_safety_system(self) -> bool:\n return self.safety_system_ok\n```\n\nIf 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.\n\n## \ud83d\udce1 State Change Callbacks\n\nGlobal state change callbacks are perfect for logging, MQTT publishing, or other side effects. They fire after every successful state transition:\n\n```python\n@on_state_change\ndef publish_to_mqtt(self, old_state: str, new_state: str, trigger: str) -> None:\n \"\"\"Publish state changes to MQTT.\"\"\"\n mqtt_client.publish(\"machine/state\", {\n \"old_state\": old_state,\n \"new_state\": new_state,\n \"trigger\": trigger,\n \"timestamp\": datetime.now().isoformat()\n })\n\n@on_state_change\ndef log_transitions(self, old_state: str, new_state: str, trigger: str) -> None:\n \"\"\"Log all state transitions.\"\"\"\n print(f\"State change: {old_state} -> {new_state} (trigger: {trigger})\")\n```\n\nMultiple 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.\n```",
"bugtrack_url": null,
"license": "Proprietary",
"summary": "Declarative state machine framework for machine apps",
"version": "0.2.0",
"project_urls": null,
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "35b4508b9a59f7f9cb01be14f4c2381c60c4c7a2275d44957bf8cb762a9911fb",
"md5": "57a988b34e820f5deeb2a1f2b8b62e30",
"sha256": "10738dc0b6ec3a4b3f729587c607aa9f9fbcc1d8c15a5e661ff71bbe1badccbf"
},
"downloads": -1,
"filename": "vention_state_machine-0.2.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "57a988b34e820f5deeb2a1f2b8b62e30",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<3.11,>=3.9",
"size": 15215,
"upload_time": "2025-08-21T18:33:22",
"upload_time_iso_8601": "2025-08-21T18:33:22.648418Z",
"url": "https://files.pythonhosted.org/packages/35/b4/508b9a59f7f9cb01be14f4c2381c60c4c7a2275d44957bf8cb762a9911fb/vention_state_machine-0.2.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "f6fdf729a78707b454c88cfc95d40392a77d615a8cd25654db3b72c7e5fc4180",
"md5": "5bd4d0d3e7ec9ceceba4c07b4684cd76",
"sha256": "084387e4af0cc75ab7f93b29319e3612c49aba748dd64aece4ebca1584021117"
},
"downloads": -1,
"filename": "vention_state_machine-0.2.0.tar.gz",
"has_sig": false,
"md5_digest": "5bd4d0d3e7ec9ceceba4c07b4684cd76",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<3.11,>=3.9",
"size": 14932,
"upload_time": "2025-08-21T18:33:23",
"upload_time_iso_8601": "2025-08-21T18:33:23.465280Z",
"url": "https://files.pythonhosted.org/packages/f6/fd/f729a78707b454c88cfc95d40392a77d615a8cd25654db3b72c7e5fc4180/vention_state_machine-0.2.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-21 18:33:23",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "vention-state-machine"
}
