py_branches

Reusable behavior-tree behaviors and decorators for Python, built on py_trees.

py_branches provides higher-level functionality designed to sit on top of the py_trees library. py_trees supplies the machinery for behavior trees; this package supplies the patterns that otherwise get rewritten on top of it — alternating execution, probabilistic selection, blackboard-driven conditionals, cooldowns, run caps, latching, retries, timeouts, and time-based pausing.

Installation

From PyPI:

pip install py_branches

From source (editable):

git clone https://github.com/snwu1996/py_branches.git
cd py_branches
pip install -e .

Modules

Basic Usage

Alternating — cycle through behaviors

import py_trees
from py_branches.alternating import run_alternating

a = py_trees.behaviours.Success(name="A")
b = py_trees.behaviours.Success(name="B")
c = py_trees.behaviours.Success(name="C")

# Run A for 3 ticks, then B for 2 ticks, then C for 4 ticks, then repeat
root = run_alternating("Alternating", [a, b, c], [3, 2, 4])

Alternating — run a child every N ticks

from py_branches.alternating import RunEveryX, RunEveryRange

child = py_trees.behaviours.Success(name="Child")

# Run child every 5th tick
every_5 = RunEveryX(child, name="Every5", every_x_range=(5, 5))

# Run child only on iterations 4–6 out of every 10
windowed = RunEveryRange(child, name="Window", max_range=10, run_range=(4, 6))

Blackboard — conditional execution and variable management

import py_trees
from py_branches.blackboard import (
    IncrementBlackboardVariable,
    RunIfBlackboardVariableEquals,
    SetBlackboardVariableIfCondition,
)

# Set up blackboard
py_trees.blackboard.Blackboard.enable_activity_stream()
client = py_trees.blackboard.Client(name="setup")
client.register_key("counter", access=py_trees.common.Access.WRITE)
client.counter = 0

# Increment a blackboard counter each tick
increment = IncrementBlackboardVariable(
    name="Increment", variable_name="counter", increment_by=1
)

# Only run a child behavior when counter == 5
child = py_trees.behaviours.Success(name="AtFive")
gate = RunIfBlackboardVariableEquals(
    child, name="RunAt5", variable_name="counter", equals=5
)

Pause — random duration pause

from py_branches.pause import PauseUniform

# Pause for a random duration between 1.0 and 3.0 seconds
pause = PauseUniform(name="RandomPause", low=1.0, high=3.0)

Pause — schedule-based pause

from py_branches.pause import load_schedule_file, PauseSchedule

schedule = load_schedule_file("configs/schedules/example_schedule.yaml")
if schedule is None:
    raise SystemExit("schedule file is empty")

# Pauses until the current scheduled window ends; SUCCESS immediately if
# outside all windows.
pause = PauseSchedule(name="PauseSchedule", schedule=schedule)

# As a Selector child that interrupts the work after it: FAILURE (fall
# through) outside all windows, RUNNING during one.
pause = PauseSchedule(name="PauseSchedule", schedule=schedule, fail_outside_window=True)

Random — probabilistic execution

import py_trees
from py_branches.random import RandomRun, random_selector

child = py_trees.behaviours.Success(name="Child")

# Execute child with 70% probability; return FAILURE otherwise
maybe = RandomRun(child, name="Maybe", probability=0.7)

# Weighted random selector: a=20%, b=30%, c=50%
a = py_trees.behaviours.Success(name="A")
b = py_trees.behaviours.Success(name="B")
c = py_trees.behaviours.Success(name="C")
selector = random_selector("WeightedSel", [a, b, c], [0.2, 0.3, 0.5])

Cooldown, Counter, Latch — limit how often a child runs

import py_trees
from py_branches.cooldown import Cooldown
from py_branches.counter import Counter
from py_branches.latch import Latch

child = py_trees.behaviours.Success(name="Child")

# At most one run every 10 seconds; FAILURE while still cooling down
cooled = Cooldown(child, name="Cooldown", duration=10.0)

# Run at most 3 times, then report SUCCESS without ticking the child again
capped = Counter(child, name="Counter", num_runs=3)

# Once the child succeeds, keep reporting SUCCESS forever
latched = Latch(child, name="Latch")

Retry and Timeout — bound failure and duration

from py_branches.delay import DelayExponentialBackoff
from py_branches.retry import Retry
from py_branches.retry import RunUntilFailed
from py_branches.retry import RunUntilXSuccesses
from py_branches.timeout import Timeout

# Re-run on FAILURE up to 3 attempts, waiting 0.5s between them
retried = Retry(child, name="Retry", max_attempts=3, delay=0.5)

# Or back off: 0.5, 1, 2, 4 ... seconds between attempts, never more than 30
backed_off = Retry(
    child,
    name="RetryWithBackoff",
    max_attempts=8,
    delay=DelayExponentialBackoff(0.5, max_delay=30.0),
)

# Re-run on SUCCESS until the child fails (SUCCESS), or FAILURE after 100 runs
repeated = RunUntilFailed(child, name="RunUntilFailed", max_runs=100)

# Re-run until 3 successes in total (SUCCESS), or FAILURE after 10 runs
collected = RunUntilXSuccesses(child, name="Collect", num_successes=3, max_runs=10)

# FAILURE if the child stays RUNNING for more than 2 seconds
bounded = Timeout(child, name="Timeout", duration=2.0)

Surgery — edit a tree after it is built

from py_branches import surgery
from py_branches.pause import PauseUniform, PauseUntilKey

# Replace every real pause with "press space to continue", including the ones
# hiding under decorators - a plain isinstance(node, Composite) walk misses
# those, because a py_trees Decorator is not a Composite.
swapped = surgery.swap_type(
    tree.root,
    (PauseUniform,),
    lambda old: PauseUntilKey(name=old.name, key="space"),
)

surgery.find_by_name(tree.root, "random_break_pause")
surgery.replace(old_node, new_node)  # re-parents, and fixes decorator.decorated
surgery.reset_subtree(tree.root)  # re-arm every Latch and Counter

Visitors — see what the tree actually did

from py_branches.visitors import StatusTransitionVisitor, TimerVisitor

tree = py_trees.trees.BehaviourTree(root=child)

# Log a line only when a leaf changes status, and time every RUNNING stretch
tree.visitors.append(StatusTransitionVisitor())
tree.visitors.append(TimerVisitor())

Clock — test timing without sleeping

Every behavior that measures elapsed time takes a keyword-only clock. It defaults to the real one, so production trees need not pass it; tests hand it a ManualClock and step time forward instead of waiting.

from py_branches.clock import ManualClock
from py_branches.cooldown import Cooldown

clock = ManualClock()
cooled = Cooldown(child, name="Cooldown", duration=5.0, clock=clock)

cooled.tick_once()  # runs
cooled.tick_once()  # cooling, child not ticked
clock.advance(5.0)  # no real time passes
cooled.tick_once()  # re-armed, exactly on the boundary

Running Tests

pytest tests/

Documentation

Full documentation, including an API reference generated from the source and rendered behavior-tree diagrams, is at py-branches.readthedocs.io.

To build it locally:

poetry install --with docs
poetry run sphinx-build -b html docs docs/_build/html

Rendering the diagrams needs the dot binary (apt install graphviz).

License

BSD License. See LICENSE for details.