bteng 0.2.7

Modular Behavior Tree execution engine for Python

BTEng — Behavior Tree Engine for Python

BTEng is a modular, production-grade Behavior Tree execution engine for Python. It is designed for robotics, automation, simulation, and any tick-driven control system that needs robust, reactive decision-making.

The library provides a compact core runtime, a rich node library, a thread-safe blackboard, XML tree loading, plugin-based custom nodes, execution tracing, structured logging, runtime introspection, and optional ZMQ event streaming — without imposing a large framework on your application.

---

Table of Contents

• Features
• Installation
• Quick Start
• Core Concepts
• Custom Nodes
• XML Trees
• Introspection & Logging
• Execution Tracing & Replay
• ZMQ Event Streaming
• Unit Testing
• CLI
• Documentation
• License

---

Features

Category	Capabilities
Execution	Tick-based engine, event-loop executor, pause/resume, max-tick budget
Control nodes	Sequence, Fallback/Selector, Parallel, ReactiveSequence, ReactiveFallback
Decorators	Inverter, Retry, Timeout, RateController, ForceSuccess, ForceFailure
Leaf nodes	ActionNode, ConditionNode, StatefulActionNode, AsyncActionNode
Blackboard	Scoped, observable, provenance history, schema validation, change subscriptions
Builder API	Fluent TreeBuilder for programmatic tree construction
XML loader	Full XML parser with port-default resolution and subtree support
Factory	@register_node decorator, NodeManifest, dynamic plugin loading
Introspection	Inspector — per-node stats, active-path tracking, explain log
Logging	Logger — console and JSON-lines sinks, auto-wired to Inspector
Tracing	ExecutionTracer — per-tick frame recording, JSON export, replay
Concurrency	ThreadPool (auto-injected), CancellationToken for async actions
Testing	MockActionNode, MockConditionNode, BehaviorTreeTest
Streaming	ZmqPublisher — streams tick events via ZMQ PUB (optional dep)
CLI	bteng run for command-line tree execution

---

Installation

From PyPI

pip install bteng

Optional extras

pip install "bteng[zmq]"   # ZMQ publisher — stream tick events to dashboards
pip install "bteng[dev]"   # Development tools: pytest, build, twine

Requirements: Python 3.9 or newer.

---

Quick Start

Programmatic tree

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 NavigateTo(ActionNode):
    def tick(self) -> NodeStatus:
        goal = self.blackboard.get("goal", "home")
        print(f"Navigating to {goal}")
        return NodeStatus.SUCCESS


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

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

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

Fluent builder + executor

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 inspector 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)

---

Core Concepts

Status model

Every node returns a NodeStatus each tick:

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	Ticks children left-to-right; stops at first FAILURE or RUNNING
Fallback / Selector	Ticks children left-to-right; stops at first SUCCESS or RUNNING
Parallel	Ticks all children simultaneously with 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 preempt

Decorators

Decorator	Purpose
Inverter	Swaps SUCCESS ↔ FAILURE; passes RUNNING unchanged
Retry(max_attempts)	Retries child on FAILURE up to N times
Timeout(duration)	Returns FAILURE if child exceeds duration (seconds)
RateController(hz)	Rate-limits child ticking; returns cached status between ticks
ForceSuccess	Always returns SUCCESS unless child is RUNNING
ForceFailure	Always returns FAILURE unless child is RUNNING

Blackboard

A thread-safe, scoped key-value store shared across the tree:

from bteng import Blackboard

bb = Blackboard.create("robot")
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 None, not a missing-key default)
bb.has("stopped")             # True

# Change subscriptions
sub = bb.subscribe(lambda key, val: print(f"{key} changed to {val}"))
bb.set("pose", (2.0, 3.0, 0.0))  # fires 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")   # transparently writes bb["goal"]
child.get("local_goal")           # transparently reads bb["goal"]

---

Custom Nodes

Define new node types by subclassing and declaring ports:

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")   # resolves XML attribute or default
        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 configuration via self.config (NodeConfig with blackboard, port mappings, and static params). self.blackboard is a convenience shortcut.

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

---

XML Trees

Load and execute trees defined in XML — useful for decoupling behavior from code:

<?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>

Port syntax:

• goal="{target_goal}" — blackboard reference; reads/writes key target_goal
• mode="inspection" — static literal parameter
• InputPort(default=...) is applied when the XML attribute is absent

from bteng import XMLTreeParser, Blackboard

bb   = Blackboard.create("demo")
bb.set("target_goal", "dock_station")

parser = XMLTreeParser()
root   = parser.parse_file("mission.xml", blackboard=bb)

See docs/xml_spec.md for the full specification.

---

Introspection & Logging

Attach an Inspector and Logger to any TreeExecutor — no changes to node code required:

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 setup needed
executor.tick_until_result()

# Per-node statistics
for uid, stats in inspector.all_stats().items():
    print(f"{uid:40s}  ticks={stats.tick_count:4d}  "
          f"total={stats.total_duration*1000:.1f}ms")

---

Execution Tracing & Replay

Record every tick for offline analysis, regression testing, or replay:

from bteng import ExecutionTracer, TreeExecutor

tracer   = ExecutionTracer()
executor = TreeExecutor()
executor.set_tree(tree)
executor.set_tracer(tracer)
executor.tick_until_result()

# Export full trace
with open("trace.json", "w") as f:
    f.write(tracer.export_json())

# Compact replay format (smaller file)
with open("replay.json", "w") as f:
    f.write(tracer.export_replay())

# Load and inspect a previously recorded run
tracer2 = ExecutionTracer()
tracer2.load_replay(open("replay.json").read())
frame = tracer2.replay_frame(0)
print(frame.tick_index, frame.blackboard_snapshot)

---

ZMQ Event Streaming

Stream tick events to an external dashboard or visualiser with zero coupling between the engine and the consumer:

pip install "bteng[zmq]"

from bteng.introspection import ZmqPublisher

pub = ZmqPublisher(port=1667)   # default port — compatible with BehaviorTree.CPP convention
pub.attach(inspector)
pub.start()

# ... run tree as normal ...

pub.stop()

Each published message is a JSON object on ZMQ topic bteng:

{
  "ts":     1234.567,
  "uid":    "a1b2c3d4",
  "name":   "NavigateTo",
  "type":   "action",
  "status": "SUCCESS",
  "dur_ms": 12.3,
  "reason": ""
}

---

Unit Testing

BehaviorTreeTest and MockActionNode / MockConditionNode provide a purpose-built test harness:

from bteng import (
    MockActionNode, MockConditionNode, BehaviorTreeTest,
    NodeStatus, SequenceNode, Tree, TreeMetadata,
)

condition = MockConditionNode("IsReady")
condition.set_result(True)

action = MockActionNode("Navigate")
action.set_ticks_to_complete(3)    # returns RUNNING for 2 ticks, then SUCCESS

root = SequenceNode("root", children=[condition, action])
tree = Tree(TreeMetadata(id="test"), root)

result = (
    BehaviorTreeTest(tree)
    .expect_final_status(NodeStatus.SUCCESS)
    .set_max_ticks(10)
    .run()
)
assert result, result.error_message

---

CLI

Execute and visualize trees from the command line:

bteng run mission.xml --plugin my_robot_nodes.py --tree main --hz 10 --log run.json -v

---

Documentation

Document	Contents
API usage	Executor, Inspector, Logger, Tracer, EventBus usage patterns
Architecture	Component relationships and data-flow diagram
XML specification	Full XML format reference with port syntax
Node system	Node lifecycle, port model, custom node authoring

---

License

BTEng is distributed under a proprietary license. See LICENSE for terms.