# stateforward.hsm
A high-performance, hierarchical state machine (HSM) implementation for Python with asyncio support. This library provides a declarative, async-first approach to building complex state machines with efficient O(1) runtime performance.
## Features
* **Hierarchical States**: Support for nested state machines with proper scoping
* **Async/Await**: Full asyncio support for actions, guards, effects, and activities
* **Precomputed Lookups**: O(1) performance through precomputed transition and element maps
* **Event-Driven**: Deterministic event processing with guards and effects
* **Timers**: Built-in support for one-shot and recurring timers
* **Activities**: Long-running async operations that auto-cancel on state exit
* **Error Handling**: Automatic error event dispatching for robust error recovery
* **Type Safety**: Strong typing with comprehensive type hints
## Installation
```bash
pip install stateforward.hsm
```
## Quick Start
```python
import asyncio
import hsm
class Counter(hsm.Instance):
def __init__(self):
super().__init__()
self.value = 0
@staticmethod
async def increment(ctx, self, event):
self.value += 1
print(f"Count: {self.value}")
model = hsm.define('Counter',
hsm.initial(hsm.target('counting')),
hsm.state('counting',
hsm.transition(
hsm.on('inc'),
hsm.target('.'), # Self-transition
hsm.effect(increment)
)
)
)
async def main():
instance = Counter()
ctx = hsm.Context()
# Start the state machine
sm = await hsm.start(ctx, instance, Counter.model)
print(f"Initial state: {sm.state()}")
# Dispatch events
await sm.dispatch(hsm.Event('inc'))
await sm.dispatch(hsm.Event('inc'))
print(f"Final count: {instance.value}")
# Clean shutdown
await hsm.stop(sm)
if __name__ == "__main__":
asyncio.run(main())
```
## Core Concepts
### Declarative Definition
State machines are defined using a declarative API that precomputes all lookups for optimal runtime performance:
```python
model = hsm.define(
"MyMachine",
hsm.initial(hsm.target("ready")),
hsm.state("ready",
hsm.transition(hsm.on("start"), hsm.target("running"))
),
hsm.state("running",
hsm.transition(hsm.on("stop"), hsm.target("ready"))
)
)
```
### States and Hierarchy
* **`hsm.state(name, ...children)`**: Regular states that can contain nested states
* **`hsm.initial(hsm.target(path))`**: Defines the default entry point
* **`hsm.final(name)`**: Terminal states for completion handling
* **`hsm.choice(name, ...transitions)`**: Conditional branching logic
### Transitions
Transitions define how the machine responds to events:
```python
hsm.transition(
hsm.on("event_name"), # Event trigger
hsm.source("current_state"), # Optional explicit source
hsm.target("next_state"), # Destination state
hsm.guard(async_guard_func), # Optional condition
hsm.effect(async_effect_func) # Optional side effect
)
```
### Actions and Behaviors
All behavioral functions are async and receive `(ctx, instance, event)`:
* **`hsm.entry(action)`**: Executed when entering a state
* **`hsm.exit(action)`**: Executed when exiting a state
* **`hsm.effect(action)`**: Executed during transitions
* **`hsm.guard(guard_func)`**: Returns boolean to allow/block transitions
* **`hsm.activity(activity_func)`**: Long-running operations that auto-cancel
### Timers
Built-in timer support for time-based transitions:
```python
from datetime import timedelta
@staticmethod
async def one_second(ctx, self, event):
return timedelta(seconds=1)
hsm.transition(
hsm.after(one_second), # One-shot timer
hsm.target("next_state")
)
hsm.transition(
hsm.every(one_second), # Recurring timer
hsm.target(".")
)
```
### Error Handling
Exceptions in actions automatically dispatch `hsm_error` events:
```python
class RobustMachine(hsm.Instance):
@staticmethod
async def risky_action(ctx, self, event):
if random.random() < 0.1:
raise ValueError("Something went wrong!")
@staticmethod
async def handle_error(ctx, self, event):
print(f"Error occurred: {event.data}")
# Recovery logic here
model = hsm.define('RobustMachine',
hsm.state('working',
hsm.effect(risky_action),
hsm.transition(hsm.on('hsm_error'), hsm.target('error_state'))
),
hsm.state('error_state',
hsm.entry(handle_error)
)
)
```
## Advanced Usage
### Class-Based Pattern
The recommended pattern is to define your state machine as a class:
```python
class TrafficLight(hsm.Instance):
def __init__(self):
super().__init__()
self.cycles = 0
@staticmethod
async def change_to_green(ctx, self, event):
print("🟢 Green light")
self.cycles += 1
model = hsm.define('TrafficLight',
hsm.initial(hsm.target('red')),
hsm.state('red',
hsm.transition(hsm.on('timer'), hsm.target('green'))
),
hsm.state('green',
hsm.entry(change_to_green),
hsm.transition(hsm.on('timer'), hsm.target('yellow'))
),
hsm.state('yellow',
hsm.transition(hsm.on('timer'), hsm.target('red'))
)
)
```
### Path Resolution
State paths are resolved relative to the transition's source state:
* **Child**: `'child_state'`
* **Parent**: `'..'`
* **Sibling**: `'../sibling_state'`
* **Absolute**: `'/MachineName/state'`
* **Self**: `'.'`
### Runtime Lifecycle
```python
# Create instance and context
instance = MyMachine()
ctx = hsm.Context()
# Start the machine
sm = await hsm.start(ctx, instance, MyMachine.model)
# Check current state
current_state = sm.state()
# Dispatch events
await sm.dispatch(hsm.Event('my_event', data={'key': 'value'}))
# Stop the machine
await hsm.stop(sm)
```
## API Reference
### Core Functions
* **`hsm.define(name, ...elements)`**: Create a state machine model
* **`hsm.start(ctx, instance, model)`**: Start a state machine instance
* **`hsm.stop(instance)`**: Stop and cleanup a state machine
### State Elements
* **`hsm.state(name, ...children)`**: Define a state
* **`hsm.initial(target, ...effects)`**: Define initial state
* **`hsm.final(name)`**: Define final state
* **`hsm.choice(name, ...transitions)`**: Define choice pseudostate
### Transitions
* **`hsm.transition(...conditions)`**: Define a transition
* **`hsm.on(event_name)`**: Event condition
* **`hsm.after(duration_func)`**: Timer condition (one-shot)
* **`hsm.every(duration_func)`**: Timer condition (recurring)
* **`hsm.target(path)`**: Target state
* **`hsm.guard(func)`**: Guard condition
* **`hsm.effect(func)`**: Transition effect
### Actions
* **`hsm.entry(func)`**: Entry action
* **`hsm.exit(func)`**: Exit action
* **`hsm.activity(func)`**: Long-running activity
### Runtime
* **`hsm.Instance`**: Base class for state machine instances
* **`hsm.Context`**: Execution context with cancellation support
* **`hsm.Event(name, data=None)`**: Event object
## Performance
This implementation uses precomputed lookup tables for O(1) transition resolution, making it suitable for high-performance applications. All behavioral functions are async-first, enabling efficient concurrent execution.
## License
MIT License - see [LICENSE](LICENSE) file for details.
## Contributing
Contributions are welcome! Please see the main [HSM repository](https://github.com/stateforward/hsm) for contribution guidelines.
Raw data
{
"_id": null,
"home_page": null,
"name": "stateforward.hsm",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.13",
"maintainer_email": null,
"keywords": "async, asyncio, finite-state-machine, hierarchical, hsm, state-machine",
"author": "Stateforward Team",
"author_email": null,
"download_url": "https://files.pythonhosted.org/packages/d1/0b/68ac6cfcda70304d82759f5950faa71105ac9e595a6a54a2cf54b2aa88db/stateforward_hsm-0.1.1.tar.gz",
"platform": null,
"description": "# stateforward.hsm\n\nA high-performance, hierarchical state machine (HSM) implementation for Python with asyncio support. This library provides a declarative, async-first approach to building complex state machines with efficient O(1) runtime performance.\n\n## Features\n\n* **Hierarchical States**: Support for nested state machines with proper scoping\n* **Async/Await**: Full asyncio support for actions, guards, effects, and activities\n* **Precomputed Lookups**: O(1) performance through precomputed transition and element maps\n* **Event-Driven**: Deterministic event processing with guards and effects\n* **Timers**: Built-in support for one-shot and recurring timers\n* **Activities**: Long-running async operations that auto-cancel on state exit\n* **Error Handling**: Automatic error event dispatching for robust error recovery\n* **Type Safety**: Strong typing with comprehensive type hints\n\n## Installation\n\n```bash\npip install stateforward.hsm\n```\n\n## Quick Start\n\n```python\nimport asyncio\nimport hsm\n\nclass Counter(hsm.Instance):\n def __init__(self):\n super().__init__()\n self.value = 0\n\n @staticmethod\n async def increment(ctx, self, event):\n self.value += 1\n print(f\"Count: {self.value}\")\n\n model = hsm.define('Counter',\n hsm.initial(hsm.target('counting')),\n hsm.state('counting',\n hsm.transition(\n hsm.on('inc'),\n hsm.target('.'), # Self-transition\n hsm.effect(increment)\n )\n )\n )\n\nasync def main():\n instance = Counter()\n ctx = hsm.Context()\n\n # Start the state machine\n sm = await hsm.start(ctx, instance, Counter.model)\n print(f\"Initial state: {sm.state()}\")\n\n # Dispatch events\n await sm.dispatch(hsm.Event('inc'))\n await sm.dispatch(hsm.Event('inc'))\n\n print(f\"Final count: {instance.value}\")\n\n # Clean shutdown\n await hsm.stop(sm)\n\nif __name__ == \"__main__\":\n asyncio.run(main())\n```\n\n## Core Concepts\n\n### Declarative Definition\n\nState machines are defined using a declarative API that precomputes all lookups for optimal runtime performance:\n\n```python\nmodel = hsm.define(\n \"MyMachine\",\n hsm.initial(hsm.target(\"ready\")),\n hsm.state(\"ready\",\n hsm.transition(hsm.on(\"start\"), hsm.target(\"running\"))\n ),\n hsm.state(\"running\",\n hsm.transition(hsm.on(\"stop\"), hsm.target(\"ready\"))\n )\n)\n```\n\n### States and Hierarchy\n\n* **`hsm.state(name, ...children)`**: Regular states that can contain nested states\n* **`hsm.initial(hsm.target(path))`**: Defines the default entry point\n* **`hsm.final(name)`**: Terminal states for completion handling\n* **`hsm.choice(name, ...transitions)`**: Conditional branching logic\n\n### Transitions\n\nTransitions define how the machine responds to events:\n\n```python\nhsm.transition(\n hsm.on(\"event_name\"), # Event trigger\n hsm.source(\"current_state\"), # Optional explicit source\n hsm.target(\"next_state\"), # Destination state\n hsm.guard(async_guard_func), # Optional condition\n hsm.effect(async_effect_func) # Optional side effect\n)\n```\n\n### Actions and Behaviors\n\nAll behavioral functions are async and receive `(ctx, instance, event)`:\n\n* **`hsm.entry(action)`**: Executed when entering a state\n* **`hsm.exit(action)`**: Executed when exiting a state\n* **`hsm.effect(action)`**: Executed during transitions\n* **`hsm.guard(guard_func)`**: Returns boolean to allow/block transitions\n* **`hsm.activity(activity_func)`**: Long-running operations that auto-cancel\n\n### Timers\n\nBuilt-in timer support for time-based transitions:\n\n```python\nfrom datetime import timedelta\n\n@staticmethod\nasync def one_second(ctx, self, event):\n return timedelta(seconds=1)\n\nhsm.transition(\n hsm.after(one_second), # One-shot timer\n hsm.target(\"next_state\")\n)\n\nhsm.transition(\n hsm.every(one_second), # Recurring timer\n hsm.target(\".\")\n)\n```\n\n### Error Handling\n\nExceptions in actions automatically dispatch `hsm_error` events:\n\n```python\nclass RobustMachine(hsm.Instance):\n @staticmethod\n async def risky_action(ctx, self, event):\n if random.random() < 0.1:\n raise ValueError(\"Something went wrong!\")\n\n @staticmethod\n async def handle_error(ctx, self, event):\n print(f\"Error occurred: {event.data}\")\n # Recovery logic here\n\n model = hsm.define('RobustMachine',\n hsm.state('working',\n hsm.effect(risky_action),\n hsm.transition(hsm.on('hsm_error'), hsm.target('error_state'))\n ),\n hsm.state('error_state',\n hsm.entry(handle_error)\n )\n )\n```\n\n## Advanced Usage\n\n### Class-Based Pattern\n\nThe recommended pattern is to define your state machine as a class:\n\n```python\nclass TrafficLight(hsm.Instance):\n def __init__(self):\n super().__init__()\n self.cycles = 0\n\n @staticmethod\n async def change_to_green(ctx, self, event):\n print(\"\ud83d\udfe2 Green light\")\n self.cycles += 1\n\n model = hsm.define('TrafficLight',\n hsm.initial(hsm.target('red')),\n hsm.state('red',\n hsm.transition(hsm.on('timer'), hsm.target('green'))\n ),\n hsm.state('green',\n hsm.entry(change_to_green),\n hsm.transition(hsm.on('timer'), hsm.target('yellow'))\n ),\n hsm.state('yellow',\n hsm.transition(hsm.on('timer'), hsm.target('red'))\n )\n )\n```\n\n### Path Resolution\n\nState paths are resolved relative to the transition's source state:\n\n* **Child**: `'child_state'`\n* **Parent**: `'..'`\n* **Sibling**: `'../sibling_state'`\n* **Absolute**: `'/MachineName/state'`\n* **Self**: `'.'`\n\n### Runtime Lifecycle\n\n```python\n# Create instance and context\ninstance = MyMachine()\nctx = hsm.Context()\n\n# Start the machine\nsm = await hsm.start(ctx, instance, MyMachine.model)\n\n# Check current state\ncurrent_state = sm.state()\n\n# Dispatch events\nawait sm.dispatch(hsm.Event('my_event', data={'key': 'value'}))\n\n# Stop the machine\nawait hsm.stop(sm)\n```\n\n## API Reference\n\n### Core Functions\n\n* **`hsm.define(name, ...elements)`**: Create a state machine model\n* **`hsm.start(ctx, instance, model)`**: Start a state machine instance\n* **`hsm.stop(instance)`**: Stop and cleanup a state machine\n\n### State Elements\n\n* **`hsm.state(name, ...children)`**: Define a state\n* **`hsm.initial(target, ...effects)`**: Define initial state\n* **`hsm.final(name)`**: Define final state\n* **`hsm.choice(name, ...transitions)`**: Define choice pseudostate\n\n### Transitions\n\n* **`hsm.transition(...conditions)`**: Define a transition\n* **`hsm.on(event_name)`**: Event condition\n* **`hsm.after(duration_func)`**: Timer condition (one-shot)\n* **`hsm.every(duration_func)`**: Timer condition (recurring)\n* **`hsm.target(path)`**: Target state\n* **`hsm.guard(func)`**: Guard condition\n* **`hsm.effect(func)`**: Transition effect\n\n### Actions\n\n* **`hsm.entry(func)`**: Entry action\n* **`hsm.exit(func)`**: Exit action\n* **`hsm.activity(func)`**: Long-running activity\n\n### Runtime\n\n* **`hsm.Instance`**: Base class for state machine instances\n* **`hsm.Context`**: Execution context with cancellation support\n* **`hsm.Event(name, data=None)`**: Event object\n\n## Performance\n\nThis implementation uses precomputed lookup tables for O(1) transition resolution, making it suitable for high-performance applications. All behavioral functions are async-first, enabling efficient concurrent execution.\n\n## License\n\nMIT License - see [LICENSE](LICENSE) file for details.\n\n## Contributing\n\nContributions are welcome! Please see the main [HSM repository](https://github.com/stateforward/hsm) for contribution guidelines.\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Hierarchical State Machine implementation for Python with asyncio support",
"version": "0.1.1",
"project_urls": {
"Documentation": "https://github.com/stateforward/hsm-python",
"Homepage": "https://github.com/stateforward/hsm",
"Issues": "https://github.com/stateforward/hsm-python/issues",
"Repository": "https://github.com/stateforward/hsm-python"
},
"split_keywords": [
"async",
" asyncio",
" finite-state-machine",
" hierarchical",
" hsm",
" state-machine"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "f04d37543b452818a35b5397b618a1ceed43f667d68b524c4b5c7a11e414d82a",
"md5": "2ec16db25799e723f4142aaa96d1b805",
"sha256": "a52a60d90a368572d8e63524dc5e4df0340238c90669ed00291da262e2715257"
},
"downloads": -1,
"filename": "stateforward_hsm-0.1.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "2ec16db25799e723f4142aaa96d1b805",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.13",
"size": 15990,
"upload_time": "2025-11-03T20:47:41",
"upload_time_iso_8601": "2025-11-03T20:47:41.498906Z",
"url": "https://files.pythonhosted.org/packages/f0/4d/37543b452818a35b5397b618a1ceed43f667d68b524c4b5c7a11e414d82a/stateforward_hsm-0.1.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "d10b68ac6cfcda70304d82759f5950faa71105ac9e595a6a54a2cf54b2aa88db",
"md5": "b49bd6d2c71338706055ba2e8263f15e",
"sha256": "a8a2b1f495aee9b8a45f3fecc4f965951da1d907835b5f16d90d27762c3834a4"
},
"downloads": -1,
"filename": "stateforward_hsm-0.1.1.tar.gz",
"has_sig": false,
"md5_digest": "b49bd6d2c71338706055ba2e8263f15e",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.13",
"size": 575807,
"upload_time": "2025-11-03T20:47:42",
"upload_time_iso_8601": "2025-11-03T20:47:42.415274Z",
"url": "https://files.pythonhosted.org/packages/d1/0b/68ac6cfcda70304d82759f5950faa71105ac9e595a6a54a2cf54b2aa88db/stateforward_hsm-0.1.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-11-03 20:47:42",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "stateforward",
"github_project": "hsm-python",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "stateforward.hsm"
}
JSON
