Py Trees for ROS

https://py-trees-ros.readthedocs.io/en/release-2.4.x
Admin
Py Trees for ROS is a Python library that implements behavior trees for Robot Operating System...
Tokens:86,106
Snippets:431
Trust Score:8.3
Update:9 months ago
Show doc for...
Tokens:
Raw
### PyTrees ROS Action Client - Setup and Initialization

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/modules

The `setup` method initializes the action client services and subscribers. It takes keyword arguments, requiring a ROS 2 node to be passed. The `initialise` method resets internal variables and starts a new goal request, often involving derived class methods to obtain the goal.

```python
class FromCallback(FromBlackboard, ABC):
    # ... other methods ...
    def setup(self, **kwargs):
        """
        Setup the action client services and subscribers.

        Parameters
        ----------
        **kwargs : dict
            distribute arguments to this behaviour and in turn, all of it’s children 

        Raises
        ------
        KeyError
            if a ros2 node isn’t passed under the key ‘node’ in kwargs
        TimedOutError
            if the action server could not be found
        """
        # Implementation to setup services and subscribers
        pass

    def initialise(self):
        """
        Call derived class get_goal method and write the returned action goal to the blackboard.
        """
        # Implementation to get goal and write to blackboard
        pass

class FromConstant(FromBlackboard):
    # ... other methods ...
    def setup(self, **kwargs):
        """
        Setup the action client services and subscribers.

        Parameters
        ----------
        **kwargs : dict
            distribute arguments to this behaviour and in turn, all of it’s children 

        Raises
        ------
        KeyError
            if a ros2 node isn’t passed under the key ‘node’ in kwargs
        TimedOutError
            if the action server could not be found
        """
        # Implementation to setup services and subscribers
        pass

    def initialise(self):
        """
        Reset the internal variables and kick off a new goal request.
        """
        # Implementation to reset variables and kick off goal
        pass
```

--------------------------------

### FromBlackboard setup Method

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/_modules/py_trees_ros/action_clients

The `setup` method for the `FromBlackboard` class is responsible for initializing the action client services and subscribers. It takes keyword arguments, which are expected to include a ROS 2 node under the key 'node'.

```python
    def setup(self, **kwargs):
        """
        Setup the action client services and subscribers.

        Args:
            **kwargs (:obj:`dict`): distribute arguments to this
               behaviour and in turn, all of it's children

        Raises:
            :class:`KeyError`: if a ros2 node isn't passed under the key 'node' in kwargs
        """

```

--------------------------------

### Example Usage of FromBlackboard with WaitForBlackboardVariable

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/_modules/py_trees_ros/action_clients

This example demonstrates how to use the FromBlackboard behaviour in conjunction with WaitForBlackboardVariable to ensure a goal is available on the blackboard before attempting to send it to the action server. It forms part of a sequence behaviour.

```python
sequence = py_trees.composites.Sequence(name="Sequence", memory=True)
wait_for_goal = py_trees.behaviours.WaitForBlackboardVariable(
    name="WaitForGoal",
    variable_name="/my_goal"
)
action_client = py_trees_ros.action_clients.FromBlackboard(
    action_type=py_trees_actions.Dock,
    action_name="dock",
    name="ActionClient"
)
sequence.add_children([wait_for_goal, action_client])
```

--------------------------------

### Initialize and Setup Transform Listener in Python

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/_modules/py_trees_ros/transforms

This snippet demonstrates the initialization and setup of a transform listener using `py_trees_ros.transforms.ToBlackboard`. It configures the listener with target and source frames, QoS profiles, and registers a variable name on the blackboard. The `setup` method ensures a ROS2 node is provided, and the `listener` is created with the specified buffer, node, and QoS settings.

```python
def __init__(
    self,
    name: str,
    variable_name,
    target_frame: str,
    source_frame: str,
    qos_profile: rclpy.qos.QoSProfile,
    static_qos_profile: rclpy.qos.QoSProfile=None,
    clearing_policy: py_trees.common.ClearingPolicy=py_trees.common.ClearingPolicy.ON_INITIALISE,
):
    super().__init__(name=name)
    self.variable_name = variable_name
    self.blackboard = self.attach_blackboard_client(name)
    self.blackboard.register_key(
        key=self.variable_name,
        access=py_trees.common.Access.WRITE
    )
    self.target_frame = target_frame
    self.source_frame = source_frame
    self.qos_profile = qos_profile
    self.static_qos_profile = static_qos_profile
    self.clearing_policy = clearing_policy
    if self.clearing_policy == py_trees.common.ClearingPolicy.ON_SUCCESS:
        raise TypeError("ON_SUCCESS is not a valid policy for transforms.ToBlackboard")
    self.buffer = tf2_ros.Buffer()
    # initialise the blackboard
    self.blackboard.set(self.variable_name, None)

def setup(self, **kwargs):
    """
    Initialises the transform listener.

    Args:
        **kwargs (:obj:`dict`): distribute arguments to this
           behaviour and in turn, all of it's children

    Raises:
        KeyError: if a ros2 node isn't passed under the key 'node' in kwargs
    """
    try:
        self.node = kwargs['node']
    except KeyError as e:
        error_message = "didn't find 'node' in setup's kwargs [{}][{}]".format(self.name, self.__class__.__name__)
        raise KeyError(error_message) from e  # 'direct cause' traceability
    self.listener = tf2_ros.TransformListener(
        buffer=self.buffer,
        node=self.node,
        # spin_thread=False,
        qos=self.qos_profile,
        static_qos=self.static_qos_profile
    )
```

--------------------------------

### Setup Blackboard Watcher Services in Python

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/_modules/py_trees_ros/blackboard

The `setup` method creates a ROS node and searches for required services, such as listing, opening, and closing connections to the blackboard. It takes a timeout value, raising exceptions if services are not found or if multiple instances are found.  This method is crucial for establishing communication with the blackboard.

```python
def setup(self, timeout_sec: float):
    """
    Creates the node and checks that all of the server-side services are available
    for calling on to open a connection.

    Args:
        timeout_sec: time (s) to wait (use common.Duration.INFINITE to block indefinitely)

    Raises:
        :class:`~py_trees_ros.exceptions.NotFoundError`: if no services were found
        :class:`~py_trees_ros.exceptions.MultipleFoundError`: if multiple services were found
    """
    self.node = rclpy.create_node(
        node_name=utilities.create_anonymous_node_name(node_name='watcher'),
        start_parameter_services=False
    )
    for service_name in self.service_names.keys():
        # can raise NotFoundError and MultipleFoundError
        self.service_names[service_name] = utilities.find_service(
            node=self.node,
            service_type=self.service_type_strings[service_name],
            namespace=self.namespace_hint,
            timeout=timeout_sec
        )
```

--------------------------------

### ROS Action Client Setup and Server Connection

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/_modules/py_trees_ros/action_clients

Demonstrates the setup of a ROS 2 action client, including waiting for the action server to become available. It handles different waiting strategies, including continuous waiting with periodic logging and immediate success without waiting. Raises TimedOutError if the server is not found within the specified timeout.

```python
self.logger.debug("{}.setup()".format(self.qualified_name))
try:
    self.node = kwargs['node']
except KeyError as e:
    error_message = "didn't find 'node' in setup's kwargs [{}][{}]".format(self.qualified_name)
    raise KeyError(error_message) from e  # 'direct cause' traceability

self.action_client = rclpy.action.ActionClient(
    node=self.node,
    action_type=self.action_type,
    action_name=self.action_name,
    callback_group=self.callback_group,
)
result = None
if self.wait_for_server_timeout_sec > 0.0:
    result = self.action_client.wait_for_server(timeout_sec=self.wait_for_server_timeout_sec)
elif self.wait_for_server_timeout_sec == 0.0:
    result = True # don't wait and don't check if the server is ready
else:
    iterations = 0
    period_sec = -1.0*self.wait_for_server_timeout_sec
    while not result:
        iterations += 1
        result = self.action_client.wait_for_server(timeout_sec=period_sec)
        if not result:
            self.node.get_logger().warning(
                "waiting for action server ... [{}s][{}][{}]".format(
                    iterations * period_sec,
                    self.action_name,
                    self.qualified_name
                )
            )
if not result:
    self.feedback_message = "timed out waiting for the server [{}]".format(self.action_name)
    self.node.get_logger().error("{}[{}]".format(self.feedback_message, self.qualified_name))
    raise exceptions.TimedOutError(self.feedback_message)
else:
    self.feedback_message = "... connected to action server [{}]".format(self.action_name)
    self.node.get_logger().info("{}[{}]".format(self.feedback_message, self.qualified_name))
```

--------------------------------

### Handle ROS Tree Setup with Timeouts in PyTrees

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/_modules/py_trees_ros/trees

Illustrates the setup process for a ROS behavior tree, including handling potential timeouts during initialization. It retrieves the setup timeout parameter and manages `RuntimeError` exceptions, re-raising them as `TimedOutError` if necessary.

```python
########################################
        # Behaviours
        ########################################
        try:
            super().setup(
                timeout=setup_timeout,
                visitor=visitor,
                node=self.node,
                **kwargs
            )
        except RuntimeError as e:
            if str(e) == "tree setup interrupted or timed out":
                raise exceptions.TimedOutError(str(e))
            else:
                raise
```

--------------------------------

### SetupLogger Visitor for py_trees_ros Behaviour Tree Setup

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/_modules/py_trees_ros/visitors

The SetupLogger visitor logs the name and setup timings of each behaviour in a py_trees_ros behaviour tree to the ROS debug channel. It requires an rclpy node for logging and is typically used with py_trees_ros.trees.TreeManager.setup. It measures and reports the total tree setup time.

```python
class SetupLogger(py_trees.visitors.VisitorBase):
    """
    Use as a visitor to :meth:`py_trees_ros.trees.TreeManager.setup`
    to log the name and timings of each behaviours' setup
    to the ROS debug channel.

    Args:
        node: an rclpy node that will provide debug logger
    """
    def __init__(self, node: rclpy.node.Node):
        super().__init__(full=True)
        self.node = node


    def initialise(self):
        """
        Initialise the timestamping chain.
        """
        self.start_time = time.monotonic()
        self.last_time = self.start_time


    def run(self, behaviour):
        current_time = time.monotonic()
        self.node.get_logger().debug(
            "'{}'.setup: {:.4f}s".format(behaviour.name, current_time - self.last_time)
        )
        self.last_time = current_time


    def finalise(self):
        current_time = time.monotonic()
        self.node.get_logger().debug(
            "Total tree setup time: {:.4f}s".format(current_time - self.start_time)
        )
```

--------------------------------

### Blackboard Watcher Script Entry Point (Python)

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/_modules/py_trees_ros/programs/blackboard_watcher

The main function serves as the entry point for the blackboard watcher script. It handles argument parsing, ROS2 initialization, BlackboardWatcher setup, and the execution of commands like listing or streaming blackboard variables. It includes error handling for setup and execution phases.

```python
import sys
import rclpy
import py_trees_ros.blackboard
import py_trees_ros.exceptions
import py_trees_ros.utilities
import argparse
from std_msgs.msg import String
from py_trees.console import console

def command_line_argument_parser(formatted_for_sphinx=False):
    parser = argparse.ArgumentParser(description='Watch a ROS2 blackboard.')
    parser.add_argument('-l', '--list', action='store_true', help='list blackboard variables')
    parser.add_argument('-v', '--variables', nargs='+', help='list of variables to watch')
    parser.add_argument('--visited', action='store_true', help='filter on visited path')
    parser.add_argument('--activity', action='store_true', help='with activity stream')
    parser.add_argument('--namespace', help='namespace of the blackboard')
    return parser

def pretty_print_variables(variables):
    for variable in variables:
        print(f'  - {variable.name}: {variable.value}\n')

def main(command_line_args=sys.argv[1:]):
    """
    Entry point for the blackboard watcher script.
    """
    parser = command_line_argument_parser(formatted_for_sphinx=False)
    args = parser.parse_args(command_line_args)

    rclpy.init(args=None)
    blackboard_watcher = py_trees_ros.blackboard.BlackboardWatcher(
        namespace_hint=args.namespace
    )
    subscription = None
    ####################
    # Setup
    ####################
    try:
        blackboard_watcher.setup(timeout_sec=2.0)
    except py_trees_ros.exceptions.NotFoundError as e:
        print(console.red + "\nERROR: {}\n".format(str(e)) + console.reset)
        sys.exit(1)
    except py_trees_ros.exceptions.MultipleFoundError as e:
        print(console.red + "\nERROR: {}\n".format(str(e)) + console.reset)
        if args.namespace is None:
            print(console.red + "\nERROR: select one with the --namespace argument\n" + console.reset)
        else:
            print(console.red + "\nERROR: but none matching the requested '{}'\n".format(args.namespace) + console.reset)
        sys.exit(1)

    ####################
    # Execute
    ####################
    result = 0
    try:
        if args.list:
            request, client = blackboard_watcher.create_service_client('list')
            future = client.call_async(request)
            rclpy.spin_until_future_complete(blackboard_watcher.node, future)
            if future.result() is None:
                raise py_trees_ros.exceptions.ServiceError(
                    "service call failed [{}]".format(future.exception())
                )
            pretty_print_variables(future.result().variables)
        else:
            request, client = blackboard_watcher.create_service_client('open')
            request.variables = [variable.strip(',[]') for variable in args.variables]
            request.filter_on_visited_path = args.visited
            request.with_activity_stream = args.activity
            future = client.call_async(request)
            rclpy.spin_until_future_complete(blackboard_watcher.node, future)
            response = future.result()
            blackboard_watcher.node.destroy_client(client)
            watcher_topic_name = response.topic
            blackboard_watcher.node.get_logger().info(
                "creating subscription [{}]".format(watcher_topic_name)
            )
            subscription = blackboard_watcher.node.create_subscription(
                msg_type=String,
                topic=watcher_topic_name,
                callback=blackboard_watcher.echo_blackboard_contents,
                qos_profile=py_trees_ros.utilities.qos_profile_unlatched()
            )
            try:
                rclpy.spin(blackboard_watcher.node)
            except (KeyboardInterrupt, rclpy.executors.ExternalShutdownException):
                pass
            finally:
                pass

    except (py_trees_ros.exceptions.NotReadyError,
            py_trees_ros.exceptions.ServiceError,
            py_trees_ros.exceptions.TimedOutError) as e:
        print(console.red + "\nERROR: {}".format(str(e)) + console.reset)
        result = 1
    finally:
        if subscription is not None:
            blackboard_watcher.node.destroy_subscription(subscription)
        blackboard_watcher.shutdown()
        rclpy.try_shutdown()
        sys.exit(result)

```

--------------------------------

### py-trees-tree-watcher Command Examples

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/features

Illustrates different uses of the `py-trees-tree-watcher` utility for visualizing and monitoring behaviour trees. Examples include rendering a static dot graph, connecting to live streams, and including runtime information like blackboard variable access and timing statistics.

```bash
# render the tree as a dot graph (does not include runtime information)
$ py-trees-tree-watcher --dot-graph
# connect to an existing snapshot stream (e.g. the default, if it is enabled)
$ py-trees-tree-watcher /tree/snapshots
# open and connect to a snapshot stream, visualise the tree graph and it's changes only
$ py-trees-tree-watcher 
# open a snapshot stream and include visited blackboard variables
$ py-trees-tree-watcher -b
# open a snapshot stream and include blackboard access details (activity)
$ py-trees-tree-watcher -a
# open a snapshot stream and include timing statistics
$ py-trees-tree-watcher -s
```

--------------------------------

### SetupLogger Visitor for py-trees-ros in Python

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/modules

A visitor class for `py_trees_ros.trees.TreeManager.setup()` that logs the name and timings of each behavior's setup phase to the ROS debug channel. It requires an rclpy node to access the logger.

```python
import rclpy
import py_trees_ros

# Assuming 'node' is an rclpy.node.Node instance
# setup_logger = py_trees_ros.visitors.SetupLogger(node)
# tree_manager.setup(visitors=[setup_logger])

```

--------------------------------

### Exchange Setup

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/_modules/py_trees_ros/blackboard

Sets up the ROS communication layer for the Exchange by initializing publishers and services on a given ROS node.

```APIDOC
## POST /websites/py-trees-ros_readthedocs_io_en_release-2_4_x/setup

### Description
This method initializes the ROS communication layer for the Exchange. It creates ROS services for 'get_variables', 'open', and 'close' stream operations, and prepares for publishing blackboard watcher information.

### Method
```
setup(node: rclpy.node.Node)
```

### Endpoint
```
~/blackboard_streams/<service_name>
```

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
- **node** (rclpy.node.Node) - Required - The ROS node to hook the communication services onto.

### Request Example
```python
import rclpy
from py_trees_ros.blackboard import Exchange

rclpy.init()
my_node = rclpy.create_node('my_ros_node')
exchange = Exchange()
exchange.setup(node=my_node)
```

### Response
#### Success Response (200)
- **None**: This method configures ROS services and does not return a value.

#### Response Example
```json
{
  "example": "ROS services (get_variables, open, close) created on the provided ROS node."
}
```
```

--------------------------------

### Setup and Execute Tree Watcher Node

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/_modules/py_trees_ros/programs/tree_watcher

Initializes the ROS environment, sets up the tree watcher, and enters an execution loop. It handles potential errors during setup, such as discovery failures or timeouts, and gracefully shuts down the watcher and ROS when interrupted or completed.

```python
    ####################
    # Setup
    ####################
    rclpy.init(args=None)
    try:
        tree_watcher.setup(timeout_sec=5.0)
    # setup discovery fails
    except py_trees_ros.exceptions.NotFoundError as e:
        print(console.red + "\nERROR: {}\n".format(str(e)) + console.reset)
        sys.exit(1)
    # setup discovery finds duplicates
    except py_trees_ros.exceptions.MultipleFoundError as e:
        print(console.red + "\nERROR: {}\n".format(str(e)) + console.reset)
        if args.namespace is None:
            print(console.red + "\nERROR: select one with the --namespace argument\n" + console.reset)
        else:
            print(console.red + "\nERROR: but none matching the requested '{}'\n".format(args.namespace) + console.reset)
        sys.exit(1)
    except py_trees_ros.exceptions.TimedOutError as e:
        print(console.red + "\nERROR: {}\n".format(str(e)) + console.reset)
        sys.exit(1)

    ####################
    # Execute
    ####################
    executor = rclpy.executors.SingleThreadedExecutor()
    executor.add_node(node=tree_watcher.node)
    try:
        while True:
            if not rclpy.ok():
                break
            if tree_watcher.done:
                if tree_watcher.xdot_process is None:
                    # no xdot found on the system, just break out and finish
                    break
                elif tree_watcher.xdot_process.poll() is not None:
                    # xdot running, wait for it to terminate
                    break
            executor.spin_once(timeout_sec=0.1)
    except (KeyboardInterrupt, rclpy.executors.ExternalShutdownException):
        pass
    finally:
        if tree_watcher.xdot_process is not None:
            if tree_watcher.xdot_process.poll() is not None:
                tree_watcher.xdot_process.terminate()
        tree_watcher.shutdown()
        rclpy.try_shutdown()
```

--------------------------------

### FromBlackboard Behaviour Setup in Python

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/_modules/py_trees_ros/service_clients

Sets up the FromBlackboard behaviour by creating a ROS service client and waiting for the service server to become available. It requires a ROS node to be passed in the kwargs. Raises exceptions if the node is not provided or if the server times out.

```python
def setup(self, **kwargs):
        """
        Setup the service client and ensure it is available.

        Args:
            **kwargs (:obj:`dict`): distribute arguments to this
               behaviour and in turn, all of it's children

        Raises:
            :class:`KeyError`: if a ros2 node isn't passed under the key 'node' in kwargs
            :class:`~py_trees_ros.exceptions.TimedOutError`: if the service server could not be found
        """
        self.logger.debug("{}.setup()".format(self.qualified_name))
        try:
            self.node = kwargs["node"]
        except KeyError as e:
            error_message = "didn't find 'node' in setup's kwargs [{}][{}]".format(self.qualified_name)
            raise KeyError(error_message) from e  # 'direct cause' traceability

        self.service_client = self.node.create_client(
            srv_type=self.service_type,
            srv_name=self.service_name,
            callback_group=self.callback_group,
        )

        result = None
        if self.wait_for_server_timeout_sec > 0.0:
            result = self.service_client.wait_for_service(timeout_sec=self.wait_for_server_timeout_sec)
        elif self.wait_for_server_timeout_sec == 0.0:
            result = True # don't wait and don't check if the server is ready
        else:
            iterations = 0
            period_sec = -1.0 * self.wait_for_server_timeout_sec
            while not result:
                iterations += 1
                result = self.service_client.wait_for_service(timeout_sec=period_sec)
                if not result:
                    self.node.get_logger().warning(
                        "waiting for service server ... [{}s][{}][{}]".format(

```

--------------------------------

### Watcher Setup Method

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/_modules/py_trees_ros/trees

Sets up the ROS 2 node and establishes connections to the necessary services for opening and closing snapshot streams. It handles service discovery and client creation. Raises exceptions for missing services, multiple found services, or timeouts.

```python
def setup(self, timeout_sec: float):
        """
        Creates the node and checks that all of the server-side services are available
        for calling on to open a connection.

        Args:
            timeout_sec: time (s) to wait (use common.Duration.INFINITE to block indefinitely)

        Raises:
            :class:`~py_trees_ros.exceptions.NotFoundError`: if services/topics were not found
            :class:`~py_trees_ros.exceptions.MultipleFoundError`: if multiple services were found
            :class:`~py_trees_ros.exceptions.TimedOutError`: if service clients time out waiting for the server
        """
        self.node = rclpy.create_node(
            node_name=utilities.create_anonymous_node_name(node_name='tree_watcher'),
            start_parameter_services=False
        )
        # open a snapshot stream
        if self.topic_name is None:
            # discover actual service names
            for service_name in self.service_names.keys():
                # can raise NotFoundError and MultipleFoundError
                self.service_names[service_name] = utilities.find_service(
                    node=self.node,
                    service_type=self.service_type_strings[service_name],
                    namespace=self.namespace_hint,
                    timeout=timeout_sec
                )
            # create service clients
            self.services["open"] = self.create_service_client(key="open")
            self.services["close"] = self.create_service_client(key="close")
            # request a stream
            request = self.service_types["open"].Request()
            request.parameters.blackboard_data = self.parameters.blackboard_data
            request.parameters.blackboard_activity = self.parameters.blackboard_activity
            request.parameters.snapshot_period = self.parameters.snapshot_period
            future = self.services["open"].call_async(request)
            rclpy.spin_until_future_complete(self.node, future)
            response = future.result()
            self.topic_name = response.topic_name
        # connect to a snapshot stream
        start_time = time.monotonic()
        while True:
            elapsed_time = time.monotonic() - start_time
            if elapsed_time > timeout_sec:
                raise exceptions.TimedOutError("timed out waiting for a snapshot stream publisher [{}]".format(self.topic_name))
            if self.node.count_publishers(self.topic_name) > 0:
                break
            time.sleep(0.1)
        self.subscriber = self.node.create_subscription(
            msg_type=py_trees_msgs.BehaviourTree,
            topic=self.topic_name,
            callback=self.callback_snapshot,

```

--------------------------------

### Setup ROS Tree

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/_modules/py_trees_ros/trees

Sets up publishers, exchange, and ROS relevant pre/post tick handlers to the tree. Relays this call down to all the behaviours in the tree.

```APIDOC
## POST /setup_ros_tree

### Description
Sets up publishers, exchange, and ROS relevant pre/post tick handlers to the tree. Relays this call down to all the behaviours in the tree.

### Method
POST

### Endpoint
/setup_ros_tree

### Parameters
#### Path Parameters
None

#### Query Parameters
* **node** (rclpy.node.Node) - Optional - ROS Node object. If None (default), creates its own node.
* **node_name** (string) - Optional - Name of ROS node created. Only takes effect if `node` is None.
* **timeout** (float) - Optional - Time (s) to wait (use common.Duration.INFINITE to block indefinitely).
* **visitor** (object) - Optional - Runnable entities on each node after it's setup.

#### Request Body
None

### Request Example
```json
{
  "node": null,
  "node_name": "my_ros_node",
  "timeout": 5.0,
  "visitor": null
}
```

### Response
#### Success Response (200)
* **status** (string) - Indicates success or failure of the setup.

#### Response Example
```json
{
  "status": "success"
}
```

### Error Handling
* **rclpy.exceptions.NotInitializedException**: rclpy not yet initialized.
* **TypeError**: node argument is not a valid rclpy.node.Node object.
* **Exception**: Any exception raised by child behaviors.
```

--------------------------------

### Customizing Feedback Message in FromBlackboard

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/_modules/py_trees_ros/action_clients

This example shows how to customize the feedback message generated by the FromBlackboard behaviour. A lambda function is provided to the `generate_feedback_message` argument, which formats the feedback data into a human-readable string.

```python
action_client = py_trees_ros.action_clients.FromBlackboard(
    action_type=py_trees_actions.Dock,
    action_name="dock",
    name="ActionClient",
    generate_feedback_message=lambda msg: "{:.2f}%%".format(msg.feedback.percentage_completed)
)
```

--------------------------------

### py-trees-blackboard-watcher Command Examples

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/features

Demonstrates various ways to use the `py-trees-blackboard-watcher` command-line utility to inspect and monitor the ROS blackboard. It covers listing keys, streaming all variables, filtering by visited variables or activity, and watching specific variables or nested fields.

```bash
# list all keys on the blackboard
$ py-trees-blackboard-watcher --list
# stream all variables
$ py-trees-blackboard-watcher 
# stream only visited variables and access details
$ py-trees-blackboard-watcher --visited --activity
# stream a single variable
$ py-trees-blackboard-watcher odometry
# stream only a single field within a variable
$ py-trees-blackboard-watcher odometry.pose.pose.position
```

--------------------------------

### FromBlackboard Class Definition and Initialization

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/_modules/py_trees_ros/action_clients

This snippet shows the definition of the `FromBlackboard` class, inheriting from `py_trees.behaviour.Behaviour`. It includes the constructor (`__init__`) which sets up the blackboard client, registers the 'goal' key, and initializes action client-related attributes. It also defines a dictionary for mapping action goal statuses to strings.

```python
class FromBlackboard(py_trees.behaviour.Behaviour):
    """
    An action client interface that draws goals from the blackboard. The
    lifecycle of this behaviour works as follows:

    * :meth:`initialise`: check blackboard for a goal and send
    * :meth:`update`: if a goal was sent, monitor progress
    * :meth:`terminate`: if interrupted while running, send a cancel request

    As a consequence, the status of this behaviour can be interpreted as follows:

    * :data:`~py_trees.common.Status.FAILURE`: no goal was found to send,
      it was rejected or it failed while executing
    * :data:`~py_trees.common.Status.RUNNING`: a goal was sent and is still
      executing on the server
    * :data:`~py_trees.common.Status.SUCCESS`: sent goal has completed with success
    """
    def __init__(self, 
                 name: str,
                 action_type: typing.Any,
                 action_name: str,
                 key: str,
                 generate_feedback_message: typing.Callable[[typing.Any], str] = None,
                 wait_for_server_timeout_sec: float = -3.0,
                 callback_group: typing.Optional[rclpy.callback_groups.CallbackGroup] = None,
                 ):
        super().__init__(name)
        self.action_type = action_type
        self.action_name = action_name
        self.wait_for_server_timeout_sec = wait_for_server_timeout_sec
        self.callback_group = callback_group

        self.blackboard = self.attach_blackboard_client(name=self.name)
        self.blackboard.register_key(
            key="goal",
            access=py_trees.common.Access.READ,
            # make sure to namespace it if not already
            remap_to=py_trees.blackboard.Blackboard.absolute_name("/", key)
        )
        self.generate_feedback_message = generate_feedback_message

        self.node = None
        self.action_client = None

        self.status_strings = {
                action_msgs.GoalStatus.STATUS_UNKNOWN : "STATUS_UNKNOWN",  # noqa
                action_msgs.GoalStatus.STATUS_ACCEPTED : "STATUS_ACCEPTED",  # noqa
                action_msgs.GoalStatus.STATUS_EXECUTING: "STATUS_EXECUTING",  # noqa
                action_msgs.GoalStatus.STATUS_CANCELING: "STATUS_CANCELING",  # noqa
                action_msgs.GoalStatus.STATUS_SUCCEEDED: "STATUS_SUCCEEDED",  # noqa
                action_msgs.GoalStatus.STATUS_CANCELED : "STATUS_CANCELED",  # noqa
                action_msgs.GoalStatus.STATUS_ABORTED  : "STATUS_ABORTED"  # noqa
            }

```

--------------------------------

### ROS Initialization

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/modules

Handles the ROS initialization of publishers and services, allowing for offline construction of behaviors and trees for visualization purposes.

```APIDOC
## ROS Initialization

### Description
This method handles the ROS initialisation of publishers and services. It is kept outside of the constructor to enable construction of behaviours and trees offline, so that dot graphs and other visualisations of the tree can be created.

### Parameters
#### Path Parameters
- None

#### Query Parameters
- None

#### Request Body
- **node** (Node) - Required - The node to hook ROS communications on.

### Request Example
```json
{
  "node": "<rclpy.node.Node object>"
}
```

### Response
#### Success Response (200)
- None (This is a method for initialization)

#### Response Example
```json
{}
```
```

--------------------------------

### ROS Service: Get Blackboard Variables

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/_modules/py_trees_ros/blackboard

Provides a list of all variable names currently stored in the py-trees blackboard.

```APIDOC
## ROS Service: Get Blackboard Variables

### Description
This ROS service allows clients to query and retrieve a list of all available variable names stored within the py-trees blackboard. It does not return the values of these variables, only their names.

### Method
```
call_service('~/get_variables')
```

### Endpoint
```
~/get_variables
```

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None (This is a service call, not a request body)

### Request Example
```bash
rosservice call ~/get_variables
```

### Response
#### Success Response (200)
- **variables** (list[str]) - A list containing the names of all variables in the blackboard.

#### Response Example
```json
{
  "variables": [
    "my_var_1",
    "group.my_var_2",
    "another_variable"
  ]
}
```
```

--------------------------------

### Initialize Action Goal from Blackboard - Python

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/_modules/py_trees_ros/action_clients

This Python code illustrates the `initialise` method for a behaviour that retrieves goal attributes from the blackboard, constructs an action goal object, and then writes this goal back to the blackboard. It utilizes a dictionary comprehension for efficient attribute retrieval.

```python
def initialise(self):
        """
        Read from the blackboard the attributes to create a goal object; if succeeded, write it to the blackboard.
        """
        self.logger.debug("%s.initialise()" % self.__class__.__name__)
        goal_attributes = {goal_attr: self.blackboard.get(bb_key) for goal_attr, bb_key in self.goal_fields.items()}
        goal = self.action_type.Goal(**goal_attributes)
        self.blackboard.set(name=self.key, value=goal)

        super().initialise()
```

--------------------------------

### Programs API

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/modules

Entry points to the programs (tools and utilities), such as the blackboard watcher.

```APIDOC
## CLI /py-trees-blackboard-watcher

### Description
Opens up a window onto the blackboard. Introspect on the entire blackboard or a part thereof and receive a stream of updates whenever values change.

### Usage
```bash
# list all keys on the blackboard
$ py-trees-blackboard-watcher --list

# stream all variables
$ py-trees-blackboard-watcher 

# stream only visited variables and access details
$ py-trees-blackboard-watcher --visited --activity

# stream a single variable
$ py-trees-blackboard-watcher odometry
```

### Parameters
#### Command Line Arguments
- **--list** - Lists all keys on the blackboard.
- **--visited** - Streams only visited variables.
- **--activity** - Shows access details.

### Response
Streams changes to blackboard variables in real-time.

```

--------------------------------

### Visitors API

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/modules

Provides visitor classes for traversing and processing behaviour trees, such as logging setup or converting trees to messages.

```APIDOC
## POST /visitors/setup_logger

### Description
Use as a visitor to `py_trees_ros.trees.TreeManager.setup()` to log the name and timings of each behaviours’ setup to the ROS debug channel.

### Method
POST

### Endpoint
/visitors/setup_logger

### Parameters
#### Request Body
- **node** (rclpy.node.Node) - Required - An rclpy node that will provide a debug logger.

### Response
#### Success Response (200)
- **SetupLogger** (object) - An instance of the SetupLogger visitor.

#### Response Example
{
  "example": "SetupLogger instance"
}

## POST /visitors/tree_to_msg_visitor

### Description
Visits the entire tree and gathers all behaviours as messages for the tree logging publishers.

### Method
POST

### Endpoint
/visitors/tree_to_msg_visitor

### Response
#### Success Response (200)
- **TreeToMsgVisitor** (object) - An instance of the TreeToMsgVisitor.

#### Response Example
{
  "example": "TreeToMsgVisitor instance"
}

```

--------------------------------

### Get py_trees ROS Home Directory (Python)

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/_modules/py_trees_ros/utilities

Determines the default directory used by py-trees for logging, bagging, and other operations. This is a ROS 2 specific path.

```python
import os
import pathlib

def get_py_trees_home():
    """
    Find the default home directory used for logging, bagging and other
    esoterica.
    """
    # TODO: update with replacement for rospkg.get_ros_home() when it arrives
    home = os.path.join(str(pathlib.Path.home()), ".ros2", "py_trees")
    return home
```

--------------------------------

### FromBlackboard Behaviour Initialization in Python

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/_modules/py_trees_ros/service_clients

Initializes the FromBlackboard behaviour, which acts as a service client interface drawing requests from the blackboard. It configures blackboard keys for requests and responses, and prepares for service client creation. Dependencies include py_trees, typing, and rclpy.

```python
class FromBlackboard(py_trees.behaviour.Behaviour):
    """
    A service client interface that draws requests from the blackboard. The
    lifecycle of this behaviour works as follows:

    * :meth:`initialise`: check blackboard for a request and send
    * :meth:`update`: if a request was sent, monitor progress
    * :meth:`terminate`: if interrupted while running, send a cancel request

    As a consequence, the status of this behaviour can be interpreted as follows:

    * :data:`~py_trees.common.Status.FAILURE`: no request was found to send,
      the server was not ready, or it failed while executing
    * :data:`~py_trees.common.Status.RUNNING`: a request was sent and is still
      executing
    * :data:`~py_trees.common.Status.SUCCESS`: sent request has completed with success

    To block on the arrival of a request on the blackboard, use with the
    :class:`py_trees.behaviours.WaitForBlackboardVariable` behaviour. e.g.

    Args:
        name: name of the behaviour
        service_type: spec type for the service
        service_name: where you can find the service
        key_request: name of the key for the request on the blackboard
        key_response: optional name of the key for the response on the blackboard (default: None)
        wait_for_server_timeout_sec: use negative values for a blocking but periodic check (default: -3.0)
        callback_group: callback group for the service client

    .. note::
       The default setting for timeouts (a negative value) will suit
       most use cases. With this setting the behaviour will periodically check and
       issue a warning if the server can't be found. Actually aborting the setup can
       usually be left up to the behaviour tree manager.
    """
    def __init__(self,
                 name: str,
                 service_type: typing.Any,
                 service_name: str,
                 key_request: str,
                 key_response: str | None = None,
                 wait_for_server_timeout_sec: float = -3.0,
                 callback_group: typing.Optional[rclpy.callback_groups.CallbackGroup] = None,
                 ):
        super().__init__(name)
        self.service_type = service_type
        self.service_name = service_name
        self.wait_for_server_timeout_sec = wait_for_server_timeout_sec
        self.callback_group = callback_group

        self.blackboard = self.attach_blackboard_client(name=self.name)
        self.blackboard.register_key(
            key="request",
            access=py_trees.common.Access.READ,
            # make sure to namespace it if not already
            remap_to=py_trees.blackboard.Blackboard.absolute_name("/", key_request)
        )
        self.write_response_to_blackboard = key_response is not None
        if self.write_response_to_blackboard:
            self.blackboard.register_key(
                key="response",
                access=py_trees.common.Access.WRITE,
                # make sure to namespace it if not already
                remap_to=py_trees.blackboard.Blackboard.absolute_name("/", key_response)
            )

        self.node = None
        self.service_client = None
```

--------------------------------

### Define NotReadyError for py_trees_ros Exceptions

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/_modules/py_trees_ros/exceptions

The NotReadyError is raised when methods are called that expect ROS2-specific setup to be completed but it has not occurred. This helps in ensuring proper initialization of py_trees_ros components.

```python
[docs]class NotReadyError(Exception):
    """
    Typically used when methods have been called that expect, but have not
    pre-engaged in the ROS2 specific setup typical of py_trees_ros classes
    and behaviours.
    """
    pass

```

--------------------------------

### Resolve ROS Name in Python

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/modules

A convenience function to get the resolved name of a topic or service, similar to ROS1's `resolved_name`. It requires the ROS node and the name to resolve.

```python
import rclpy
import py_trees_ros

# Assuming 'node' is an rclpy.node.Node instance
# resolved = py_trees_ros.utilities.resolve_name(node, 'my_topic_or_service')

```

--------------------------------

### Get py_trees_ros Home Directory in Python

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/modules

Retrieves the default home directory used by py-trees-ros for logging, bagging, and other operations. This function helps locate configuration or data directories.

```python
import py_trees_ros

home_dir = py_trees_ros.utilities.get_py_trees_home()
print(f'py_trees_ros home directory: {home_dir}')
```

--------------------------------

### Initialize Action Client with Blackboard Key - Python

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/_modules/py_trees_ros/action_clients

This Python constructor for a `FromCallback` action client sets up a unique blackboard key for the goal and initializes the parent class, passing along necessary parameters for ROS action communication and blackboard integration.

```python
def __init__(self,
                 name: str,
                 action_type: typing.Any,
                 action_name: str,
                 generate_feedback_message: typing.Callable[[typing.Any], str] = None,
                 wait_for_server_timeout_sec: float = -3.0,
                 callback_group: typing.Optional[rclpy.callback_groups.CallbackGroup] = None,
                 ):
        unique_id = uuid.uuid4()
        self.key = "/goal_" + str(unique_id)
        super().__init__(
            action_type=action_type,
            action_name=action_name,
            key=self.key,
            name=name,
            generate_feedback_message=generate_feedback_message,
            wait_for_server_timeout_sec=wait_for_server_timeout_sec,
            callback_group=callback_group,
        )
        # The parent constructor already instantiated a blackboard client
        self.blackboard.register_key(
            key=self.key,
            access=py_trees.common.Access.WRITE,
        )
```

--------------------------------

### Get Latched QoS Profile in Python

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/modules

Provides a convenience retrieval for a latched Quality of Service (QoS) profile, commonly used for publishers and subscribers in ROS where the last message should be retained.

```python
import py_trees_ros

latched_qos = py_trees_ros.utilities.qos_profile_latched()

```

--------------------------------

### Initialize Goal from Callback and Write to Blackboard - Python

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/_modules/py_trees_ros/action_clients

This Python `initialise` method for `FromCallback` calls a derived class's `get_goal` method to obtain an action goal and then writes this goal to the blackboard. This is crucial for behaviours that dynamically generate their goals.

```python
def initialise(self):
        """
        Call derived class `get_goal` method and write the returned action goal to the blackboard.
        """
        self.logger.debug("%s.initialise()" % self.__class__.__name__)
        self.blackboard.set(name=self.key, value=self.get_goal())

        super().initialise()
```

--------------------------------

### Register Blackboard Key

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/_modules/py_trees_ros/service_clients

Registers a key on the blackboard for writing access. This is typically done during the setup phase of a behavior tree node to ensure a specific data point is available for modification.

```python
self.blackboard.register_key(
            key=self.key_request,
            access=py_trees.common.Access.WRITE,
        )
```

--------------------------------

### Get Unlatched QoS Profile in Python

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/modules

Returns the default Quality of Service (QoS) profile for an unlatched topic within py-trees-ros. This is typically used for standard message streaming where past messages are not retained.

```python
import py_trees_ros

unlatched_qos = py_trees_ros.utilities.qos_profile_unlatched()

```

--------------------------------

### Handle Get Variables Service Request (Python)

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/_modules/py_trees_ros/blackboard

Python service callback to retrieve the available blackboard variables. It populates the response with a list of nested keys found in the blackboard, allowing clients to query the structure of the blackboard.

```python
def _get_variables_service(self, unused_request, response):
        response.variables = self._get_nested_keys()
        return response
```

--------------------------------

### Handle Pre-Tick Statistics in py-trees-ros

Source: https://py-trees-ros.readthedocs.io/en/release-2.4.x/_modules/py_trees_ros/trees

The pre-tick handler for statistics collection. It manages time-series data for tick intervals, resets statistics, and records the start time of a tick. This function is essential for performance analysis of the behaviour tree.

```python
def _statistics_pre_tick_handler(self, tree: py_trees.trees.BehaviourTree):
    """
    Pre-tick handler that resets the statistics and starts the clock.

    Args:
        tree (:class:`~py_trees.trees.BehaviourTree`): the behaviour tree that has just been ticked
    """
    if len(self.time_series) == 10:
        self.time_series.popleft()
        self.tick_interval_series.popleft()

    rclpy_start_time = rclpy.clock.Clock().now()
    self.time_series.append(conversions.rclpy_time_to_float(rclpy_start_time))
    if len(self.time_series) == 1:
        self.tick_interval_series.append(0.0)
    else:
        self.tick_interval_series.append(self.time_series[-1] - self.time_series[-2])

    self.statistics = py_trees_msgs.Statistics()
    self.statistics.count = self.count
    self.statistics.stamp = rclpy_start_time.to_msg()
    self.statistics.tick_interval = self.tick_interval_series[-1]
    self.statistics.tick_interval_average = sum(self.tick_interval_series) / len(self.tick_interval_series)
    if len(self.tick_interval_series) > 1:
        self.statistics.tick_interval_variance = statistics.variance(
            self.tick_interval_series,
            self.statistics.tick_interval_average
        )
    else:
        self.statistics.tick_interval_variance = 0.0
```
Tokens:
Raw