Metadata-Version: 2.4
Name: petrus-engine
Version: 0.1.0
Summary: Petrus: an embedded Petri net workflow engine with colored tokens, timed transitions, and time travel.
Project-URL: Homepage, https://github.com/henriquebastos/petrus-engine
Project-URL: Documentation, https://github.com/henriquebastos/petrus-engine
Project-URL: Repository, https://github.com/henriquebastos/petrus-engine
Project-URL: Issues, https://github.com/henriquebastos/petrus-engine/issues
Author-email: Henrique Bastos <henrique@bastos.net>
License-Expression: MIT
License-File: LICENSE
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Programming Language :: Python :: Implementation :: CPython
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.12
Requires-Dist: pydantic>=2.7
Provides-Extra: cli
Requires-Dist: black>=24; extra == 'cli'
Description-Content-Type: text/markdown

# Petrus: an embedded Petri net workflow engine for Python.

Petrus executes workflows defined as [Petri nets](https://en.wikipedia.org/wiki/Petri_net).
You describe the *structure* of a workflow — places, transitions, and arcs — and provide
the *behavior* separately as plain Python callables. The engine takes care of token flow,
enabling rules, guards, timed transitions, and a full state history with time travel.

## What you can do with Petrus?

1. Define nets in JSON files (editable by tooling or a visual editor) or with a concise Python DSL.
2. Keep structure and behavior separate: transitions reference guards and handlers by name; you bind the callables at runtime.
3. Flow typed payloads through the net with colored tokens — guards and handlers declare typed parameters and the engine extracts the right payload automatically.
4. Compose nets from reusable subnets via file references, with input/output/awaiting interface places marking the boundaries.
5. Travel in time: every fire is recorded, so you can undo, redo, and branch from any past state.
6. Drive long-running workflows with split-phase firing, designed to await async handlers — with zero runtime dependency on any workflow platform.

## How to use it?

Install from PyPI:

```bash
uv add petrus-engine        # or: pip install petrus-engine
```

Build a net with the DSL, bind behavior, and run it:

```python
from petrus.dsl import Net
from petrus.engine import PetriNetRuntime

net = Net("approval")
p, t = net.p, net.t
p.pending(initial=1) > t.validate > p.validated
p.validated > t.approve(guard="is_approved") > p.approved
p.validated > t.reject(guard="is_rejected") > p.rejected

guards = {
    "is_approved": lambda d: d.value == "approve",
    "is_rejected": lambda d: d.value == "reject",
}

rt = PetriNetRuntime.from_schema(net.build(), guards=guards)

rt.fire(abs(net.t.validate))
```

A transition is enabled when its input places hold enough tokens, its inhibitor and
read arcs are satisfied, and its guard (if any) returns `True` for the payloads peeked
from the input tokens. Firing consumes the input tokens and deposits a merged output
token in every output place.

### Loading nets from JSON files

Nets can live in JSON files (`fileVersion: "2"` flat format) and reference other net
files as subnets. Resolve a root file and hand the schema to the engine:

```python
from petrus.schema import FlatNet

net = FlatNet.from_filename("nets/net_approval.json")
rt = PetriNetRuntime.from_schema(net.to_schema(), guards=guards, handlers=handlers)
```

### CLI tools

```bash
# Generate a .pyi stub from a module exposing SCHEMA (for IDE autocomplete)
petrus-cli gen-stub myapp.nets -o nets/__init__.pyi

# Validate and canonically format net JSON files in-place (pre-commit friendly)
python -m petrus.cli.format_json --fail-on-change nets/net_*.json
```

Stub generation formats its output with `black`; install the `cli` extra to use it:

```bash
uv add "petrus-engine[cli]"
```

### Timed transitions

`petrus.timers.TimerManager` syncs timer tasks with the net's enabled timed
transitions. It is clock- and sleep-agnostic: by default it uses the system UTC
clock and `asyncio.sleep`, and you can inject your own primitives — to control
time in tests, or to integrate with a scheduling runtime:

```python
from petrus.timers import TimerManager

tm = TimerManager()  # system UTC clock + asyncio.sleep
tm = TimerManager(now=my_clock, sleep=my_sleep)  # custom primitives
```

## Thank you to Routable

[Routable](https://routable.com) sponsored the development of this library.
Working at [Routable](https://routable.com) is an awesome experience, with a developer-first culture that fosters innovation and growth.
If you're interested in joining a dynamic team, [check out our job opportunities here](https://routable.com/careers/)!

## Authors

- Henrique Bastos <henrique@bastos.net>
