robokudo.utils.tree_execution

Tree execution utilities for RoboKudo.

This module provides utilities for executing behavior trees in ROS environments. It supports:

  • One-shot tree execution with GUI option

  • Rate-limited tree ticking

  • Status monitoring and completion detection

  • ROS time handling

Dependencies:

  • py_trees for behavior tree functionality

  • rospy for ROS integration

Attributes

_T

Functions

_tick_tree_until(→ typing_extensions.Optional[_T])

Tick an already setup tree until stop_condition returns a result.

run_tree_once(...)

Execute a behavior tree once with monitoring.

run_tree_until_successful_cas_count(→ list)

Execute a tree until a requested number of distinct successful CASes exist.

Module Contents

robokudo.utils.tree_execution._T
robokudo.utils.tree_execution._tick_tree_until(ae_root: py_trees_ros.trees.BehaviourTree, node: rclpy.node.Node, stop_condition: Callable[[int], typing_extensions.Optional[_T]], tick_rate: int) → typing_extensions.Optional[_T]

Tick an already setup tree until stop_condition returns a result.

robokudo.utils.tree_execution.run_tree_once(tree: py_trees.behaviour.Behaviour, node: rclpy.node.Node, include_gui: bool = False, max_iterations: int = 500, tick_rate: int = 5) → typing_extensions.Optional[py_trees.common.Status]

Execute a behavior tree once with monitoring.

This function: * Grows the tree with optional GUI * Sets up all descendants * Ticks the tree at specified rate * Monitors for completion via OneShot decorator * Handles ROS time exceptions

Parameters:

  • tree – Root behavior of tree to execute

  • node – a ros node

  • include_gui – Whether to show GUI visualization, defaults to False

  • max_iterations – Maximum number of ticks before timeout, defaults to 500

  • tick_rate – Rate to tick tree in Hz, defaults to 5

Returns:

Final tree status (SUCCESS/FAILURE) or None if timed out

robokudo.utils.tree_execution.run_tree_until_successful_cas_count(tree: py_trees.behaviour.Behaviour, node: rclpy.node.Node, expected_cas_count: int, include_gui: bool = False, max_iterations: int = 500, tick_rate: int = 20, on_successful_cas: typing_extensions.Optional[Callable[[robokudo.cas.CAS], None]] = None) → list

Execute a tree until a requested number of distinct successful CASes exist.

The tree is setup once without a OneShot wrapper so continuously running pipelines can process multiple percepts using the same ROS node. on_successful_cas is called immediately after each new successful CAS is observed.