Metadata-Version: 2.4
Name: bteng
Version: 0.2.5
Summary: Modular Behavior Tree execution engine for Python
Author: BTEng contributors
License-Expression: LicenseRef-BTEng-Proprietary
Project-URL: Homepage, https://github.com/mdirzpr/BTEng
Project-URL: Documentation, https://github.com/mdirzpr/BTEng/tree/main/docs
Project-URL: Bug Tracker, https://github.com/mdirzpr/BTEng/issues
Keywords: behavior-tree,robotics,automation,planning,control
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Science/Research
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Scientific/Engineering
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE
Provides-Extra: zmq
Requires-Dist: pyzmq>=24; extra == "zmq"
Provides-Extra: viz
Requires-Dist: graphviz>=0.20; extra == "viz"
Provides-Extra: dev
Requires-Dist: build; extra == "dev"
Requires-Dist: pytest; extra == "dev"
Requires-Dist: twine; extra == "dev"
Requires-Dist: graphviz>=0.20; extra == "dev"
Dynamic: license-file

# BTEng

BTEng is a lightweight Behavior Tree execution engine for Python, built for
robotics, automation, simulation, and other tick-driven control systems.

It gives you a small core runtime, reusable control/decorator nodes, a
thread-safe blackboard, XML tree loading, plugin-based custom nodes, execution
tracing, structured logging, runtime introspection, and ZMQ event streaming —
without forcing a large framework around your application.

> Status: Python implementation is feature-complete. C++ support is planned as
> a future direction.

## Why BTEng?

Behavior Trees are useful when a system needs to make repeated decisions while
reacting to changing world state. In robotics and automation, that usually means
checking preconditions, running long-lived actions, interrupting work when safety
or environment conditions change, and falling back to recovery behavior.

BTEng is designed around those needs:

- Tick-based execution with `SUCCESS`, `FAILURE`, `RUNNING`, and `IDLE` states
- Standard control nodes: `Sequence`, `Fallback` / `Selector`, `Parallel`
- Reactive control nodes for interruptible behavior
- Decorators for retry, timeout, rate control, inversion, and forced result
- Synchronous, stateful, and async (background-thread) action nodes
- Thread-safe blackboard with scoped subtree remapping and change subscriptions
- XML tree definitions inspired by BehaviorTree.CPP style workflows
- Fluent Python `TreeBuilder` API for programmatic tree construction
- Runtime node registration and plugin loading for application-specific actions
- `Inspector` for per-node statistics, active-path tracking, and explainability
- Structured `Logger` with console and JSON-lines sinks, auto-wired to Inspector
- `ZmqPublisher` for streaming tick events to external dashboards (optional)
- `ExecutionTracer` for full tick replay and regression testing
- `ThreadPool` auto-injected into async nodes by the executor
- `BehaviorTreeTest` for unit-testing trees with mock nodes

## Installation

```bash
git clone <repo-url>
cd BTEng
pip install -e .
```

Optional extras:

```bash
pip install -e ".[zmq]"   # ZMQ publisher — stream tick events externally
pip install -e ".[viz]"   # Graphviz export / PNG rendering
pip install -e ".[dev]"   # development tools (pytest, build, twine)
```

Python 3.9 or newer required.

## Quick Start

### Programmatic tree

```python
from bteng import (
    ActionNode, BehaviorTreeEngine, Blackboard,
    ConditionNode, NodeConfig, NodeStatus, SequenceNode,
)


class BatteryOK(ConditionNode):
    def tick(self) -> NodeStatus:
        level = self.blackboard.get("battery_level", 0)
        return NodeStatus.SUCCESS if level > 20 else NodeStatus.FAILURE


class Navigate(ActionNode):
    def tick(self) -> NodeStatus:
        goal = self.blackboard.get("goal", "home")
        print(f"Navigating to {goal}")
        return NodeStatus.SUCCESS


bb = Blackboard.create("robot_state")
bb.set("battery_level", 87)
bb.set("goal", "charging_station")

cfg  = NodeConfig(blackboard=bb)
root = SequenceNode("mission", children=[
    BatteryOK("battery_ok", cfg),
    Navigate("navigate", cfg),
], config=cfg)

engine = BehaviorTreeEngine(root, blackboard=bb)
status = engine.run_until_complete()
print(status)  # NodeStatus.SUCCESS
```

### Fluent builder + executor

```python
from bteng import (
    TreeBuilder, Blackboard, NodeStatus,
    TreeExecutor, ExecutorConfig,
    Inspector, Logger, LogLevel,
)

bb   = Blackboard.create("demo")
tree = (
    TreeBuilder(blackboard=bb)
    .tree_id("Mission")
    .sequence("root")
        .condition("BatteryOK", lambda: bb.get("battery_level", 0) > 20)
        .action("Navigate",     lambda: NodeStatus.SUCCESS)
    .end()
    .build()
)

inspector = Inspector.create()
logger    = Logger.create()
logger.add_console_sink(colored=True)
logger.set_min_level(LogLevel.DEBUG)

executor = TreeExecutor(ExecutorConfig(tick_interval=0.02))
executor.set_tree(tree)
executor.set_inspector(inspector)   # auto-injects into all nodes
executor.set_logger(logger)         # auto-wired to inspector events

bb.set("battery_level", 87)
status = executor.tick_until_result(max_ticks=100)
print(status)
```

Run the included examples:

```bash
python examples/01_simple_sequence.py
python examples/02_reactive_behavior.py
python examples/03_async_action.py
python examples/04_subtree_usage.py
```

## Core Concepts

### Status model

Every node returns a `NodeStatus`:

| Status | Meaning |
|--------|---------|
| `SUCCESS` | node completed successfully |
| `FAILURE` | node failed |
| `RUNNING` | node is active and must be ticked again |
| `IDLE` | node is inactive or has been halted |

### Control nodes

| Node | Behavior |
|------|----------|
| `Sequence` | Left-to-right; stops at first FAILURE or RUNNING |
| `Fallback` / `Selector` | Left-to-right; stops at first SUCCESS or RUNNING |
| `Parallel` | Ticks all children; configurable success/failure thresholds |
| `ReactiveSequence` | Restarts from child[0] every tick; earlier conditions interrupt later actions |
| `ReactiveFallback` | Restarts from child[0] every tick; higher-priority child can interrupt |

### Decorators

| Decorator | Purpose |
|-----------|---------|
| `Inverter` | Swap SUCCESS ↔ FAILURE, pass RUNNING |
| `Retry(max_attempts)` | Retry on FAILURE up to N times |
| `Timeout(duration)` | FAILURE if child exceeds duration (seconds) |
| `RateController(hz)` | Rate-limit child ticking; return cached status |
| `ForceSuccess` | Always SUCCESS (unless RUNNING) |
| `ForceFailure` | Always FAILURE (unless RUNNING) |

### Blackboard

Shared, thread-safe key-value store with scoped subtree isolation:

```python
from bteng import Blackboard

bb = Blackboard.create("robot_state")
bb.set("pose", (1.0, 2.0, 0.0))
bb.set("stopped", None)       # None is a valid stored value
bb.get("pose")                # (1.0, 2.0, 0.0)
bb.get("stopped")             # None  (the stored value, not a default)
bb.has("stopped")             # True

# Change subscriptions
sub = bb.subscribe(lambda key, val: print(f"{key} = {val}"))
bb.set("pose", (2.0, 3.0, 0.0))  # triggers callback
bb.unsubscribe(sub)

# Scoped child blackboard for subtree port isolation
child = Blackboard.create_child(bb, remapping={"local_goal": "goal"})
child.set("local_goal", "dock")   # writes bb["goal"]
child.get("local_goal")           # reads bb["goal"]
```

## Custom Nodes

```python
from bteng import ActionNode, InputPort, OutputPort, NodeStatus, register_node


@register_node()
class DetectObject(ActionNode):
    @classmethod
    def provided_ports(cls):
        return [
            InputPort("camera", default="rgb"),
            OutputPort("object_pose"),
        ]

    def tick(self) -> NodeStatus:
        camera = self.get_input("camera")   # "rgb" if not mapped in XML
        pose   = run_detector(camera)
        if pose is None:
            self.set_failure_reason("detector returned no result")
            return NodeStatus.FAILURE
        self.set_output("object_pose", pose)
        return NodeStatus.SUCCESS
```

Access the full node config via `self.config` (a `NodeConfig` with blackboard,
port mappings, and static params). `self.blackboard` is a shortcut.

For long-running work use `StatefulActionNode` (three-phase: `on_start` /
`on_running` / `on_halted`) or `AsyncActionNode` (runs `execute_async(token)` in
a background thread; the executor auto-injects a shared `ThreadPool`).

## XML Trees

```xml
<?xml version="1.0" encoding="UTF-8"?>
<BTEng format_version="1.0" main_tree_to_execute="main">
  <Tree ID="main">
    <ReactiveFallback name="root">
      <ReactiveSequence name="navigate_if_safe">
        <Condition ID="PathClear"/>
        <Timeout duration="10.0">
          <Action ID="NavigateTo" goal="{target_goal}"/>
        </Timeout>
      </ReactiveSequence>
      <Action ID="StopRobot"/>
    </ReactiveFallback>
  </Tree>

  <TreeNodesModel>
    <Condition ID="PathClear"/>
    <Action ID="NavigateTo">
      <input_port name="goal"/>
    </Action>
    <Action ID="StopRobot"/>
  </TreeNodesModel>
</BTEng>
```

- `goal="{target_goal}"` — blackboard reference
- `mode="inspection"` — static parameter
- `InputPort(default=...)` is applied when the XML attribute is absent

See [docs/xml_spec.md](docs/xml_spec.md) for the full specification.

## Introspection & Logging

```python
from bteng import Inspector, Logger, LogLevel, TreeExecutor

inspector = Inspector.create()
logger    = Logger.create()
logger.add_console_sink(colored=True)
logger.add_json_file_sink("/tmp/bt_run.jsonl")

executor = TreeExecutor()
executor.set_tree(tree)
executor.set_inspector(inspector)
executor.set_logger(logger)      # auto-wired; no extra subscription needed
executor.tick_until_result()

for uid, stats in inspector.all_stats().items():
    print(uid, stats.tick_count, f"{stats.total_duration*1000:.1f}ms")
```

## ZMQ Event Streaming

Stream tick events to an external dashboard or visualiser with zero coupling:

```bash
pip install bteng[zmq]
```

```python
from bteng.introspection import ZmqPublisher

pub = ZmqPublisher(port=1667)   # default — matches BehaviorTree.CPP convention
pub.attach(inspector)
pub.start()
# ... run tree ...
pub.stop()
```

Subscriber receives JSON messages on topic `bteng`:

```json
{"ts": 1234.5, "uid": "a1b2c3", "name": "Navigate",
 "type": "action", "status": "SUCCESS", "dur_ms": 12.3, "reason": ""}
```

## CLI

```bash
bteng run mission.xml --plugin my_robot_nodes.py --tree main --hz 10 --log run.json -v
bteng viz mission.xml --output tree.dot
bteng viz mission.xml --output tree.png --format png
```

## Project Layout

```text
bteng/
  core/           TreeNode, NodeStatus, NodeConfig, Tree, TreeExecutor, TreeBuilder
  nodes/          Control, decorator, and leaf nodes
  blackboard/     Thread-safe shared state with scoped remapping
  factory/        Node registry and @register_node decorator
  xml_parser/     XML-to-tree parser with port-default resolution
  concurrency/    ThreadPool, CancellationToken
  introspection/  Inspector, Logger, ZmqPublisher
  logging/        ExecutionTracer (frame-based tick recording)
  testing/        MockActionNode, BehaviorTreeTest
  plugins/        Dynamic plugin loading
  visualization/  Graphviz export and rendering helpers
docs/             Architecture, XML spec, API, and node-system reference
examples/         Working examples (01–04)
```

## Documentation

- [API usage](docs/api_usage.md)
- [Architecture](docs/architecture.md)
- [XML specification](docs/xml_spec.md)
- [Node system](docs/node_system.md)

## License

BTEng is distributed under a proprietary license. See [LICENSE](LICENSE).
