API

ros_bt_py.node module

Module defining the Node class and helper functions representing a node in the behavior tree.

class ros_bt_py.node.Decorator(node_id: UUID | None = None, name: str | None = None, new_inputs: dict[str, DataContainer] = {}, ros_node: rclpy.node.Node | None = None, debug_manager: DebugManager | None = None, subtree_manager: SubtreeManager | None = None, logging_manager: LoggingManager | None = None, data_flow_manager: DataFlowManager | None = None) → None

Bases: Node

Base class for Decorator nodes.

Decorators have exactly one child and somehow modify that child’s output. Subclasses can add inputs and outputs, but never change max_children.

class ros_bt_py.node.FlowControl(node_id: UUID | None = None, name: str | None = None, new_inputs: dict[str, DataContainer] = {}, ros_node: rclpy.node.Node | None = None, debug_manager: DebugManager | None = None, subtree_manager: SubtreeManager | None = None, logging_manager: LoggingManager | None = None, data_flow_manager: DataFlowManager | None = None) → None

Bases: Node

Base class for flow control nodes.

Flow control nodes (mostly Sequence, Fallback and their derivatives) can have an unlimited number of children and each have a unique set of rules for when to tick which of their children.

class ros_bt_py.node.Leaf(node_id: UUID | None = None, name: str | None = None, new_inputs: dict[str, DataContainer] = {}, ros_node: rclpy.node.Node | None = None, debug_manager: DebugManager | None = None, subtree_manager: SubtreeManager | None = None, logging_manager: LoggingManager | None = None, data_flow_manager: DataFlowManager | None = None) → None

Bases: Node

Base class for leaf nodes in the tree.

Leaf nodes have no children. Subclasses can define inputs, and outputs, but never change max_children.

class ros_bt_py.node.Node(node_id: UUID | None = None, name: str | None = None, new_inputs: dict[str, DataContainer] = {}, ros_node: rclpy.node.Node | None = None, debug_manager: DebugManager | None = None, subtree_manager: SubtreeManager | None = None, logging_manager: LoggingManager | None = None, data_flow_manager: DataFlowManager | None = None) → None

Bases: ABC

Base class for Behavior Tree nodes.

Each node has a set of inputs, outputs and options. At every tick (usually somewhere between 10 and 30 times a second), tick() is called with the appropriate data.

Nodes in a behavior Tree can be roughly divided into two classes, with two sub-classes each:

Leaf Nodes

These do not have any children and can take one of two forms: Predicates and Behaviors. Predicates check a condition and instantly return SUCCEEDED or FAILED. Behaviors are more involved and may return RUNNING, but should be interruptible (see untick()).

Inner Nodes

These too come in two flavors: Combiners and Decorators. Combiners have multiple children and decide which of those children to run (and in what fashion) based on some criteria. Decorators however have only a single child and work with that child’s result - for instance, a Decorator could invert FAILED into SUCCEEDED.

add_child(child: Node, at_index: int | None = None) → Ok[Node] | Err[BehaviorTreeException | TreeTopologyError]

Add a child to this node at the given index.

Return type:

Union[Ok[Node], Err[BehaviorTreeException | TreeTopologyError]]

add_extra_inputs() → Ok[dict[str, DataContainer]] | Err[NodeConfigError]

Return a dictionary of extra inputs that you want to add to the node config. This can access all original inputs, including static values if they exist.

Return type:

Union[Ok[dict[str, DataContainer]], Err[NodeConfigError]]

add_extra_outputs() → Ok[dict[str, DataContainer]] | Err[NodeConfigError]

Return a dictionary of extra outputs that you want to add to the node config. This can access all original inputs, including static values if they exist.

Return type:

Union[Ok[dict[str, DataContainer]], Err[NodeConfigError]]

calculate_utility() → Ok[UtilityBounds] | Err[BehaviorTreeException]

Calculate the utility bounds for this node.

Unlike the other node functions, there is a default implementation for the corresponding method, Node._do_calculate_utility().

However, in order to get meaningful results, one should take care to use as many nodes as possible that provide their own implementation, since the default reports that there is no cost for execution.

Return type:

Union[Ok[UtilityBounds], Err[BehaviorTreeException]]

check_if_in_invalid_state(allowed_states: list[BTNodeState], action_name: str) → Ok[None] | Err[NodeStateError]

Return type:

Union[Ok[None], Err[NodeStateError]]

data_flow_manager: DataFlowManager | None

debug_manager: DebugManager | None

classmethod from_msg(msg: NodeStructure, ros_node: rclpy.node.Node, debug_manager: DebugManager | None = None, subtree_manager: SubtreeManager | None = None, logging_manager: LoggingManager | None = None, data_flow_manager: DataFlowManager | None = None) → Ok[Node] | Err[BehaviorTreeException]

Construct a Node from the given ROS message.

This will try to import the requested node class, instantiate it and populate its name, options, input and output members from the ROS message.

This also catches exceptions raised during node construction and returns wrapped in as result.Err.

Parameters:

msg (NodeStructure)

A ROS message describing a node class. The node class must be available in the current environment (but does not need to be imported before calling this).

Parameters:

debug_manager (Optional[DebugManager], default: None)

The debug manager to use for the newly instantiated node class.

Returns:

Union[Ok[Node], Err[BehaviorTreeException]] –:

An instance of the class named by msg, populated with the values from msg.

Note that this does not include the node’s state. Any node created by this will be in state UNININITIALIZED.

Returns:

BehaviorTreeException if node cannot be instantiated.

get_child_index(child_id: UUID) → int | None

Get the index in the children array of the child with the given name.

This is useful if you want to replace a child with another node.

Returns:

Optional[int] –:

An integer index if a child with the given name exists, None if there’s no such child

get_children_recursive() → Iterator[Node]

Return all nodes that are below this node in the parent-child hirachy recursively.

Return type:

Iterator[Node]

get_logger() → LoggingManager | None

Return type:

Optional[LoggingManager]

get_subtree_msg() → Ok[Tuple[TreeStructure, list[Wiring], list[Wiring]]] | Err[BehaviorTreeException]

Populate a TreeMsg with the subtree rooted at this node.

This can be used to “shove” a subtree to a different host, by using that host’s load_tree service.

The subtree message will have public node data for every piece of node data that is wired to a node outside the subtree.

Returns:

Union[Ok[Tuple[TreeStructure, list[Wiring], list[Wiring]]], Err[BehaviorTreeException]] –:

A tuple consisting of a ros_bt_py_msgs.msg.Tree message and two lists of ros_bt_py_msgs.msg.NodeDataWiring messages (incoming_connections and outgoing_connections). The latter can be used to determine what parameters need to be forwarded to / from the remote executor if the subtree is to be executed remotely.

Crucially, the resulting subtree will not be tick-able until all the incoming wirings from external_connections have been connected.

However, if the subtree is to be shoved to a different executor, it’s enough for the incoming wirings to be connected in the host tree - this will cause input values to be set and sent to the remote executor.

property has_ros_node: bool

static log_errors(func: Callable[[Node], Ok[RET] | Err[BehaviorTreeException]]) → Callable[[Node], Ok[RET] | Err[BehaviorTreeException]]

Return type:

Callable[[Node], Union[Ok[TypeVar(RET)], Err[BehaviorTreeException]]]

logdebug(message: str, stacklevel=3, internal=False) → None

Wrap call to the associated logging manager.

Adds this node’s name and type to the given message

Return type:

None

logerr(message: str, stacklevel=3, internal=False) → None

Wrap call to the associated logging manager.

Adds this node’s name and type to the given message

Return type:

None

logfatal(message: str, stacklevel=3, internal=False) → None

Wrap call to the associated logging manager.

Adds this node’s name and type to the given message

Return type:

None

logging_manager: LoggingManager | None

loginfo(message: str, stacklevel=3, internal=False) → None

Wrap call to the associated logging manager.

Adds this node’s name and type to the given message

Return type:

None

logwarn(message: str, stacklevel=3, internal=False) → None

Wrap call to the associated logging manager.

Adds this node’s name and type to the given message

Return type:

None

node_classes: dict[str, dict[str, type[Node]]] = {}

remove_child(child_id: UUID) → Ok[Node] | Err[KeyError]

Remove the child with the given name and return it.

Parameters:

child_id (UUID) – The uuid of the child to remove

Return type:

Union[Ok[Node], Err[KeyError]]

reset() → Ok[RET] | Err[BehaviorTreeException]

Return type:

Union[Ok[TypeVar(RET)], Err[BehaviorTreeException]]

property ros_node: rclpy.node.Node

Return the associated ROS node instance.

If no instance is present an

setup() → Ok[RET] | Err[BehaviorTreeException]

Return type:

Union[Ok[TypeVar(RET)], Err[BehaviorTreeException]]

shutdown() → Ok[RET] | Err[BehaviorTreeException]

Return type:

Union[Ok[TypeVar(RET)], Err[BehaviorTreeException]]

property state: BTNodeState

State of the node.

subtree_manager: SubtreeManager | None

tick() → Ok[RET] | Err[BehaviorTreeException]

Return type:

Union[Ok[TypeVar(RET)], Err[BehaviorTreeException]]

to_state_msg()

to_structure_msg() → NodeStructure

Populate a ROS message with the information from this Node.

Round-tripping the result through Node.from_msg() should yield a working node object, with the caveat that state will not be preserved.

Returns:

NodeStructure –:

A ROS message that describes the node.

untick() → Ok[RET] | Err[BehaviorTreeException]

Return type:

Union[Ok[TypeVar(RET)], Err[BehaviorTreeException]]

ros_bt_py.node.define_bt_node(node_config: NodeConfig) → Callable[[type[N]], type[N]]

Provide information about this Node’s interface.

Every class that derives, directly or indirectly, from Node, must be decorated with this!

Parameters:

node_config (NodeConfig)

Return type:

Callable[[type[TypeVar(N, bound= Node)]], type[TypeVar(N, bound= Node)]]

This describes your Node’s interface. All inputs, outputs and options defined here are automatically registered with your class. You should not need to register anything manually!

ros_bt_py.node.increment_name(name: str) → str

If name does not already end in a number, add “_2” to it.

Otherwise, increase the number after the underscore.

Return type:

str

ros_bt_py.node.load_node_module(package_name: str) → ModuleType | None

Import the named module at run-time.

If the module contains any (properly decorated) node classes, they will be registered and available to load via the other commands in this class.

Return type:

Optional[ModuleType]

ros_bt_py.data_types module

class ros_bt_py.data_types.BoolType(allow_dynamic: bool = True, allow_static: bool = True, is_static: bool | None = None, value: ANY | None = None)

Bases: BuiltinContainer[bool]

This type holds a simple boolean value

is_compatible(other: DataContainer) → TypeGuard[BoolType]

Check if the given container is compatible with this one, meaning that its constraints are at least as narrow as the ones in self.

This is used to compare configs given on specific nodes with the baseline given on the class config.

This can also serve as a type guard for other, because the checks performed are more strict than simple type equality.

Subclasses should extend this with additional checks where applicable.

Return type:

TypeGuard[BoolType]

serialize_type() → NodeDataType

Returns a serialized version of this data type as a NodeDataType ROS message. Subclasses should extend this if there are additional parameters to be added.

Return type:

NodeDataType

type_identifier: ClassVar[int] = 1

class ros_bt_py.data_types.BuiltinContainer(allow_dynamic: bool = True, allow_static: bool = True, is_static: bool | None = None, value: ANY | None = None)

Bases: DataContainer[BUILTIN]

Common base class for all data type classes that correspond to primitive data types.

set_value(value: BUILTIN) → Ok[None] | Err[str]

Sets the value of this data type, as well as the updated flag if the value differs from the previous one.

Subclasses should validate and clean incoming values before calling super().set_value to assign them.

Return type:

Union[Ok[None], Err[str]]

class ros_bt_py.data_types.BytesType(max_length: int | None = None, valid_values: list[str] | None = None, *args, **kwargs) → None

Bases: StringContainer[bytes]

This type holds a bytes object. The length restriction is set to 1 by default, since the bytes type is mostly used to fill ‘byte’ fields in ROS messages, which only take one byte. Bytes are serializes as hex strings.

max_length: int = 1

type_identifier: ClassVar[int] = 7

class ros_bt_py.data_types.DataContainer(allow_dynamic: bool = True, allow_static: bool = True, is_static: bool | None = None, value: ANY | None = None)

Bases: Generic[ANY], ABC

The common base class for all node io data types.

Defines all interfaces to generically interact with data types, as well as basic implementations where applicable.

allow_dynamic: bool

allow_static: bool

deserialize_value(ser_value: str) → Ok[None] | Err[str]

First pass the given serialized value to json.loads, then to the self._deserialize_value helper function, and finally to self.set_value.

Return type:

Union[Ok[None], Err[str]]

flag_updated()

Set the updated flag to True

classmethod from_msg(msg: NodeDataType) → Ok[DataContainer] | Err[str]

Factory function that returns an instance of this data type, based on the given NodeDataType ROS message.

This only verifies that the type identifier matches and does the final initialization step, the parameters for constructing the data type are handled by the helper function self._dict_from_msg.

Return type:

Union[Ok[DataContainer], Err[str]]

get_runtime_type() → DataContainer

Returns the own runtime type, which is used to determine whether two types have compatible values at runtime (is used to validate wirings between data types).

The returned runtime type usually does NOT hold the value of the original type

Return type:

DataContainer

get_value() → Ok[ANY] | Err[None]

Returns an Ok holding the value or an empty Err if the value is None.

Return type:

Union[Ok[TypeVar(ANY)], Err[None]]

get_value_as(type_: type[ANY]) → Ok[ANY] | Err[Any]

Coerces the stored value to the given type_ if possible. Returns Ok if the type matches and Err if it doesn’t, but both results wrap the same value (or None in case of Err)

Return type:

Union[Ok[TypeVar(ANY)], Err[Any]]

has_value() → bool

Return type:

bool

abstractmethod is_compatible(other: DataContainer) → TypeGuard[DataContainer]

Check if the given container is compatible with this one, meaning that its constraints are at least as narrow as the ones in self.

This is used to compare configs given on specific nodes with the baseline given on the class config.

This can also serve as a type guard for other, because the checks performed are more strict than simple type equality.

Subclasses should extend this with additional checks where applicable.

Return type:

TypeGuard[DataContainer]

is_static: bool

is_updated() → bool

Return type:

bool

reset_updated()

Set the updated flag to False

reset_value()

abstractmethod serialize_type() → NodeDataType

Returns a serialized version of this data type as a NodeDataType ROS message. Subclasses should extend this if there are additional parameters to be added.

Return type:

NodeDataType

serialize_value() → str

Checks whether this type has a set value. If not, just return an empty string. Calls the self._serialize_value helper function to get a json serializable representation of the value, then pass that to json.dumps

Return type:

str

abstractmethod set_value(value: ANY) → Ok[None] | Err[str]

Sets the value of this data type, as well as the updated flag if the value differs from the previous one.

Subclasses should validate and clean incoming values before calling super().set_value to assign them.

Return type:

Union[Ok[None], Err[str]]

type_identifier: ClassVar[int]

class ros_bt_py.data_types.DictType(element_type: DataContainer | None = None, max_length: int | None = None, strict_length: bool | None = None, *args, **kwargs) → None

Bases: IterableContainer[dict[str, Any]]

This type holds a dict of values. The keys of a dict will always be coerced by using str(...).

set_value(value: dict) → Ok[None] | Err[str]

Sets the value of this data type, as well as the updated flag if the value differs from the previous one.

Subclasses should validate and clean incoming values before calling super().set_value to assign them.

Return type:

Union[Ok[None], Err[str]]

type_identifier: ClassVar[int] = 6

class ros_bt_py.data_types.FloatType(min_value: NUM | None = None, max_value: NUM | None = None, *args, **kwargs) → None

Bases: NumericContainer[float]

This type holds a floating point value.

lower_limit: NUM = -1.7976931348623157e+308

set_value(value: float | int) → Ok[None] | Err[str]

Sets the value of this data type, as well as the updated flag if the value differs from the previous one.

Subclasses should validate and clean incoming values before calling super().set_value to assign them.

Return type:

Union[Ok[None], Err[str]]

type_identifier: ClassVar[int] = 3

upper_limit: NUM = 1.7976931348623157e+308

class ros_bt_py.data_types.GenericType(valid_types: list[type] = [<class 'bool'>, <class 'int'>, <class 'float'>, <class 'str'>, <class 'bytes'>, <class 'list'>, <class 'dict'>, <class 'object'>], *args, **kwargs) → None

Bases: TypeContainerMixin, BuiltinContainer[dict]

This holds a type value from the GENERIC_TYPE_MAP keys, which correspond to data types inheriting from BuiltinContainer.

The list of valid types can optionally be constrained by supplying a list of builtin types.

get_value_field() → Ok[DataContainer] | Err[None]

Return type:

Union[Ok[DataContainer], Err[None]]

is_compatible(other: DataContainer) → TypeGuard[GenericType]

Check if the given container is compatible with this one, meaning that its constraints are at least as narrow as the ones in self.

This is used to compare configs given on specific nodes with the baseline given on the class config.

This can also serve as a type guard for other, because the checks performed are more strict than simple type equality.

Subclasses should extend this with additional checks where applicable.

Return type:

TypeGuard[GenericType]

serialize_type() → NodeDataType

Returns a serialized version of this data type as a NodeDataType ROS message. Subclasses should extend this if there are additional parameters to be added.

Return type:

NodeDataType

set_value(value: dict) → Ok[None] | Err[str]

Sets the value of this data type, as well as the updated flag if the value differs from the previous one.

Subclasses should validate and clean incoming values before calling super().set_value to assign them.

Return type:

Union[Ok[None], Err[str]]

type_identifier: ClassVar[int] = 9

valid_types: list[type]

class ros_bt_py.data_types.IntType(min_value: NUM | None = None, max_value: NUM | None = None, *args, **kwargs) → None

Bases: NumericContainer[int]

This type holds an integer value.

lower_limit: NUM = -9223372036854775808

type_identifier: ClassVar[int] = 2

upper_limit: NUM = 18446744073709551615

class ros_bt_py.data_types.IterableContainer(element_type: DataContainer | None = None, max_length: int | None = None, strict_length: bool | None = None, *args, **kwargs) → None

Bases: BuiltinContainer[ITER]

Note that the static/dynamic attributes on element types are ignored, those have to be specified on the iterable itself.

If the element type is omitted, all kinds of values are accepted, but value types that can’t be serialized as json will silently be replaced with “” in any serialized output.

Iterables can also constrained by a maximum length, with a boolean flag to make that limit ‘strict’, which means the iterable has to match that maximum length exactly.

is_compatible(other: DataContainer) → TypeGuard[IterableContainer]

Check if the given container is compatible with this one, meaning that its constraints are at least as narrow as the ones in self.

This is used to compare configs given on specific nodes with the baseline given on the class config.

This can also serve as a type guard for other, because the checks performed are more strict than simple type equality.

Subclasses should extend this with additional checks where applicable.

Return type:

TypeGuard[IterableContainer]

max_length: int = 18446744073709550000

serialize_type() → NodeDataType

Returns a serialized version of this data type as a NodeDataType ROS message. Subclasses should extend this if there are additional parameters to be added.

Return type:

NodeDataType

abstractmethod set_value(value: ITER) → Ok[None] | Err[str]

Sets the value of this data type, as well as the updated flag if the value differs from the previous one.

Subclasses should validate and clean incoming values before calling super().set_value to assign them.

Return type:

Union[Ok[None], Err[str]]

strict_length: bool = False

class ros_bt_py.data_types.IterableReferenceContainer(reference: str, max_length: int | None = None, strict_length: bool | None = None, allow_dynamic: bool = True, allow_static: bool = False, is_static: bool | None = None) → None

Bases: ReferenceContainer

A container for iterable references.

Works similar to the standard IterableContainer, with the difference that the element type is determined by the referenced type value.

max_length: int = 18446744073709551615

strict_length: bool = False

class ros_bt_py.data_types.ListType(element_type: DataContainer | None = None, max_length: int | None = None, strict_length: bool | None = None, *args, **kwargs) → None

Bases: IterableContainer[list[Any]]

This type holds a list of values.

set_value(value: list | array) → Ok[None] | Err[str]

Sets the value of this data type, as well as the updated flag if the value differs from the previous one.

Subclasses should validate and clean incoming values before calling super().set_value to assign them.

Return type:

Union[Ok[None], Err[str]]

type_identifier: ClassVar[int] = 5

class ros_bt_py.data_types.NumericContainer(min_value: NUM | None = None, max_value: NUM | None = None, *args, **kwargs) → None

Bases: BuiltinContainer[NUM]

Base class for numeric type classes, that are constrained by a minimum and maximum value.

is_compatible(other: DataContainer) → TypeGuard[NumericContainer]

Check if the given container is compatible with this one, meaning that its constraints are at least as narrow as the ones in self.

This is used to compare configs given on specific nodes with the baseline given on the class config.

This can also serve as a type guard for other, because the checks performed are more strict than simple type equality.

Subclasses should extend this with additional checks where applicable.

Return type:

TypeGuard[NumericContainer]

lower_limit: NUM

max_value: NUM

min_value: NUM

serialize_type() → NodeDataType

Returns a serialized version of this data type as a NodeDataType ROS message. Subclasses should extend this if there are additional parameters to be added.

Return type:

NodeDataType

set_value(value: NUM) → Ok[None] | Err[str]

Sets the value of this data type, as well as the updated flag if the value differs from the previous one.

Subclasses should validate and clean incoming values before calling super().set_value to assign them.

Return type:

Union[Ok[None], Err[str]]

upper_limit: NUM

class ros_bt_py.data_types.PathType(max_length: int | None = None, valid_values: list[str] | None = None, *args, **kwargs) → None

Bases: StringContainer[str]

This type holds a path uri, which has to start with ‘file://’ or ‘package://’

set_value(value: str) → Ok[None] | Err[str]

Sets the value of this data type, as well as the updated flag if the value differs from the previous one.

Subclasses should validate and clean incoming values before calling super().set_value to assign them.

Return type:

Union[Ok[None], Err[str]]

type_identifier: ClassVar[int] = 8

class ros_bt_py.data_types.ReferenceContainer(reference: str, allow_dynamic: bool = True, allow_static: bool = True, is_static: bool | None = None) → None

Bases: DataContainer[Any]

Common base class for all reference types. Reference types imitate other data types based on the referenced type value. The data type being imitated is stored as self._inner_type.

Defines additional functions necessary to fully initialize these reference types.

flag_updated() → None

Set the updated flag to True

Return type:

None

get_runtime_type() → DataContainer

Reference types are fully transparent at runtime, simply forwarding the type information from the concrete inner type.

Return type:

DataContainer

get_value() → Ok[Any] | Err[None]

Returns an Ok holding the value or an empty Err if the value is None.

Return type:

Union[Ok[Any], Err[None]]

has_value() → bool

Return type:

bool

is_compatible(other: DataContainer) → TypeGuard[ReferenceContainer]

Compatibility here is only evaluated within exact matching references. See also self.get_runtime_type

Return type:

TypeGuard[ReferenceContainer]

is_updated() → bool

Return type:

bool

reset_updated() → None

Set the updated flag to False

Return type:

None

reset_value()

serialize_type() → NodeDataType

Returns a serialized version of this data type as a NodeDataType ROS message. Subclasses should extend this if there are additional parameters to be added.

Return type:

NodeDataType

set_type_map(new_map: dict[str, DataContainer[Any]]) → Ok[None] | Err[str]

Sets the map of data types that the given reference points to. This is necessary for reference types to work properly.

Return type:

Union[Ok[None], Err[str]]

set_value(value: Any) → Ok[None] | Err[str]

Sets the value of this data type, as well as the updated flag if the value differs from the previous one.

Subclasses should validate and clean incoming values before calling super().set_value to assign them.

Return type:

Union[Ok[None], Err[str]]

class ros_bt_py.data_types.ReferenceDictType(reference: str, allow_dynamic: bool = True, allow_static: bool = True, is_static: bool | None = None) → None

Bases: ReferenceContainer

A reference type for dicts. Works like the basic DictType, with the element type being set by reference

type_identifier: ClassVar[int] = 15

class ros_bt_py.data_types.ReferenceListType(reference: str, allow_dynamic: bool = True, allow_static: bool = True, is_static: bool | None = None) → None

Bases: ReferenceContainer

A reference type for lists. Works like the basic ListType, with the element type being set by reference

type_identifier: ClassVar[int] = 14

class ros_bt_py.data_types.ReferenceType(reference: str, allow_dynamic: bool = True, allow_static: bool = True, is_static: bool | None = None) → None

Bases: ReferenceContainer

The basic reference type.

type_identifier: ClassVar[int] = 13

class ros_bt_py.data_types.RosActionName(interface_id=0, *args, **kwargs) → None

Bases: RosNameContainer

Holds a ROS action name as a string.

interface_kind: ClassVar[int] = 3

class ros_bt_py.data_types.RosActionType(allow_dynamic=False, allow_static=True, *args, **kwargs) → None

Bases: RosTypeContainer

This type holds message types of ROS actions.

interface_kind: ClassVar[int] = 3

class ros_bt_py.data_types.RosComponentType(allow_dynamic=False, allow_static=True, *args, **kwargs) → None

Bases: RosTypeContainer

This type holds component message types.

This behaves similar to RosTopicType, except that the interface type indicates that we also want to allow interface components like Service_Request or Action_Goal.

interface_kind: ClassVar[int] = 4

class ros_bt_py.data_types.RosContainer(interface_id=0, *args, **kwargs) → None

Bases: DataContainer

Common base class for all data types related to ROS components (topics, services, actions).

These types are further identified by the kind of interface they refer to, as well as an interface id to connect for example the name and type fields of the same service.

classmethod from_msg(msg: NodeDataType) → Ok[RosContainer] | Err[str]

Factory function that returns an instance of this data type, based on the given NodeDataType ROS message.

This only verifies that the type identifier matches and does the final initialization step, the parameters for constructing the data type are handled by the helper function self._dict_from_msg.

Return type:

Union[Ok[RosContainer], Err[str]]

interface_id: int

interface_kind: ClassVar[int]

is_compatible(other: DataContainer) → TypeGuard[RosContainer]

Check if the given container is compatible with this one, meaning that its constraints are at least as narrow as the ones in self.

This is used to compare configs given on specific nodes with the baseline given on the class config.

This can also serve as a type guard for other, because the checks performed are more strict than simple type equality.

Subclasses should extend this with additional checks where applicable.

Return type:

TypeGuard[RosContainer]

serialize_type() → NodeDataType

Returns a serialized version of this data type as a NodeDataType ROS message. Subclasses should extend this if there are additional parameters to be added.

Return type:

NodeDataType

class ros_bt_py.data_types.RosMessage(*args, **kwargs)

Bases: Protocol

Structural base class for all ROS messages

classmethod get_fields_and_field_types() → dict[str, str]

Return type:

dict[str, str]

class ros_bt_py.data_types.RosMessageType(message_type: type[RosMessage], *args, **kwargs) → None

Bases: RosContainer

Holds the values for a given ROS message, which has to be specified on init.

get_element_fields() → Ok[dict[str, DataContainer]] | Err[str]

Return type:

Union[Ok[dict[str, DataContainer]], Err[str]]

interface_kind: ClassVar[int] = 1

is_compatible(other: DataContainer) → TypeGuard[RosMessageType]

Check if the given container is compatible with this one, meaning that its constraints are at least as narrow as the ones in self.

This is used to compare configs given on specific nodes with the baseline given on the class config.

This can also serve as a type guard for other, because the checks performed are more strict than simple type equality.

Subclasses should extend this with additional checks where applicable.

Return type:

TypeGuard[RosMessageType]

message_type: type[RosMessage]

serialize_type() → NodeDataType

Returns a serialized version of this data type as a NodeDataType ROS message. Subclasses should extend this if there are additional parameters to be added.

Return type:

NodeDataType

set_value(value: Any) → Ok[None] | Err[str]

Sets the value of this data type, as well as the updated flag if the value differs from the previous one.

Subclasses should validate and clean incoming values before calling super().set_value to assign them.

Return type:

Union[Ok[None], Err[str]]

type_identifier: ClassVar[int] = 12

class ros_bt_py.data_types.RosNameContainer(interface_id=0, *args, **kwargs) → None

Bases: RosContainer, BuiltinContainer[str]

Common base class for ROS interface names. Also inherits from BuiltinContainer to include all necessary functionality.

type_identifier: ClassVar[int] = 10

class ros_bt_py.data_types.RosServiceName(interface_id=0, *args, **kwargs) → None

Bases: RosNameContainer

Holds a ROS service name as a string.

interface_kind: ClassVar[int] = 2

class ros_bt_py.data_types.RosServiceType(allow_dynamic=False, allow_static=True, *args, **kwargs) → None

Bases: RosTypeContainer

This type holds message types of ROS services.

interface_kind: ClassVar[int] = 2

class ros_bt_py.data_types.RosTopicName(interface_id=0, *args, **kwargs) → None

Bases: RosNameContainer

Holds a ROS topic name as a string.

interface_kind: ClassVar[int] = 1

class ros_bt_py.data_types.RosTopicType(allow_dynamic=False, allow_static=True, *args, **kwargs) → None

Bases: RosTypeContainer

This type holds message types of ROS topics.

Note that this validation also accepts component messages like Service_Request or Action_Goal, since they’re fully fledged message classes. The interface type just indicates that we are not looking for those.

interface_kind: ClassVar[int] = 1

class ros_bt_py.data_types.RosTypeContainer(allow_dynamic=False, allow_static=True, *args, **kwargs) → None

Bases: TypeContainerMixin, RosContainer

Common base class for all ROS message types.

Defines an abstract self._validate method which checks if the given type is actually a ROS message.

get_value_field() → Ok[DataContainer] | Err[None]

Return type:

Union[Ok[DataContainer], Err[None]]

set_value(value: type) → Ok[None] | Err[str]

Sets the value of this data type, as well as the updated flag if the value differs from the previous one.

Subclasses should validate and clean incoming values before calling super().set_value to assign them.

Return type:

Union[Ok[None], Err[str]]

type_identifier: ClassVar[int] = 11

class ros_bt_py.data_types.StringContainer(max_length: int | None = None, valid_values: list[str] | None = None, *args, **kwargs) → None

Bases: BuiltinContainer[STRING]

Common base class for all “string-like” data types, which are optionally constrained by a maximum length.

Alternatively they can be constrained by a set of valid value options.

is_compatible(other: DataContainer) → TypeGuard[StringContainer]

Check if the given container is compatible with this one, meaning that its constraints are at least as narrow as the ones in self.

This is used to compare configs given on specific nodes with the baseline given on the class config.

This can also serve as a type guard for other, because the checks performed are more strict than simple type equality.

Subclasses should extend this with additional checks where applicable.

Return type:

TypeGuard[StringContainer]

max_length: int = 18446744073709550000

serialize_type() → NodeDataType

Returns a serialized version of this data type as a NodeDataType ROS message. Subclasses should extend this if there are additional parameters to be added.

Return type:

NodeDataType

set_value(value: STRING) → Ok[None] | Err[str]

Sets the value of this data type, as well as the updated flag if the value differs from the previous one.

Subclasses should validate and clean incoming values before calling super().set_value to assign them.

Return type:

Union[Ok[None], Err[str]]

valid_values: list[str] | None

class ros_bt_py.data_types.StringType(max_length: int | None = None, valid_values: list[str] | None = None, *args, **kwargs) → None

Bases: StringContainer[str]

This type holds a string value.

type_identifier: ClassVar[int] = 4

class ros_bt_py.data_types.TypeContainerMixin(allow_dynamic=False, allow_static=True, *args, **kwargs) → None

Bases: DataContainer

Mixin class to signal that a data type is a valid target for reference types.

This forces the values to always be static and defines the get_value_field interface.

abstractmethod get_value_field() → Ok[DataContainer] | Err[None]

Return type:

Union[Ok[DataContainer], Err[None]]

ros_bt_py.data_types.deserialize_class(ser_cls: str) → Ok[type] | Err[str]

Deserialize builtin types

Return type:

Union[Ok[type], Err[str]]

ros_bt_py.data_types.deserialize_type_map_value(val: dict) → dict

Helper function to deserialize type values.

Return type:

dict

ros_bt_py.data_types.get_iotype_for_dict(value_dict: dict) → Ok[DataContainer] | Err[str]

Constructs the data type from a dictionary of type parameters. See also GENERIC_TYPE_MAP

Return type:

Union[Ok[DataContainer], Err[str]]

ros_bt_py.data_types.get_iotype_for_msg(msg: NodeDataType) → Ok[DataContainer] | Err[str]

Parses a NodeDataType message to a data type class that was registered with register_io_type.

Return type:

Union[Ok[DataContainer], Err[str]]

ros_bt_py.data_types.get_message_field_io_type(field_type: str) → Ok[DataContainer] | Err[str]

Constructs a data type based on the field type of a ROS message field, as given by the get_fields_and_field_types() function.

Return type:

Union[Ok[DataContainer], Err[str]]

ros_bt_py.data_types.match_dict_to_valid_types(value: dict, valid_types: list) → Ok[None] | Err[str]

Return type:

Union[Ok[None], Err[str]]

ros_bt_py.data_types.register_io_type(cls: type[CONTAINER]) → type[CONTAINER]

This decorator is used to register concrete data types, which allows them to be found when parsing a NodeDataType message with get_iotype_for_msg.

Return type:

type[TypeVar(CONTAINER, bound= DataContainer)]

ros_bt_py.data_types.serialize_class(cls: type) → str

Serialize builtin types

Return type:

str

ros_bt_py.data_types.serialize_type_map(keys: list[type]) → list[str]

Serialize the valid value list for builtin types.

Return type:

list[str]

ros_bt_py.data_types.serialize_type_map_value(val: dict) → dict

Helper function to serialize type values.

Return type:

dict

ros_bt_py.node_config module

class ros_bt_py.node_config.NodeConfig(inputs: dict[str, DataContainer], outputs: dict[str, DataContainer], max_children: int | None, tags: list[str] | None = None)

Bases: object

copy() → NodeConfig

Implement a custom copy operation that also updates all IO references.

Return type:

NodeConfig

extend(other: NodeConfig) → Ok[None] | Err[NodeConfigError]

Extend the input and output dicts with values from other.

Returns an error if if the two configs are incompatible, either due to duplicate io keys or mismatch in allowed number of child nodes.

Return type:

Union[Ok[None], Err[NodeConfigError]]

class ros_bt_py.node_config.NodeDataMap(name: str, data: dict[str, DataContainer]) → None

Bases: object

This wrapper around a plain dict[str, DataContainer] for easier access to value and updated status and easier error handling.

class ros_bt_py.node_config.NodeInputMap(name: str, data: dict[str, DataContainer]) → None

Bases: NodeDataMap

any_updated(*keys: str) → Ok[bool] | Err[NodeConfigError]

Return type:

Union[Ok[bool], Err[NodeConfigError]]

get_value(key: str) → Ok[Any] | Err[NodeConfigError]

Return type:

Union[Ok[Any], Err[NodeConfigError]]

get_value_as(key: str, type_: type[T]) → Ok[T] | Err[NodeConfigError]

Return type:

Union[Ok[TypeVar(T)], Err[NodeConfigError]]

class ros_bt_py.node_config.NodeOutputMap(name: str, data: dict[str, DataContainer]) → None

Bases: NodeDataMap

set_multiple_values(**value_dict: Any) → Ok[None] | Err[NodeConfigError]

Return type:

Union[Ok[None], Err[NodeConfigError]]

set_value(key: str, value: Any) → Ok[None] | Err[NodeConfigError]

Return type:

Union[Ok[None], Err[NodeConfigError]]

ros_bt_py.helpers module

class ros_bt_py.helpers.BTNodeState

Bases: ABC

ASSIGNED = 'ASSIGNED'

BROKEN = 'BROKEN'

FAILED = 'FAILED'

IDLE = 'IDLE'

PAUSED = 'PAUSED'

RUNNING = 'RUNNING'

SHUTDOWN = 'SHUTDOWN'

SUCCEEDED = 'SUCCEEDED'

UNASSIGNED = 'UNASSIGNED'

UNINITIALIZED = 'UNINITIALIZED'

ros_bt_py.helpers.float_limits_dict(name: str) → dict[str, float]

Return type:

dict[str, float]

ros_bt_py.helpers.int_limits_dict(name: str) → dict[str, int]

Return type:

dict[str, int]

ros_bt_py.helpers.rgetattr(obj, attr, *args)

ros_bt_py.helpers.rsetattr(obj, attr: str, val)

ros_bt_py.ros_helpers module

ros_bt_py.ros_helpers.get_interface_name(msg_metaclass: type) → str

Extract the interface name from a ROS2 message metaclass.

Parameters:

msg_metaclass (type) – The ROS2 message metaclass.

Returns:

str – The interface name in the format ‘package_name/message_type/message_name’.

ros_bt_py.ros_helpers.get_message_constant_fields(message_class) → Ok[list[str]] | Err[NodeConfigError]

Return all constant fields of a message as a list.

Return type:

Union[Ok[list[str]], Err[NodeConfigError]]

ros_bt_py.ros_helpers.publish_message_channels(node: rclpy.node.Node, publisher: rclpy.node.Publisher)

Return all known topic-, service-, and action-names.

ros_bt_py.ros_helpers.ros_to_uuid(ros_uuid_msg: str) → Ok[UUID] | Err[str]

Return type:

Union[Ok[UUID], Err[str]]

ros_bt_py.ros_helpers.uuid_to_ros(uuid: UUID) → str

Return type:

str

ros_bt_py.ros_helpers.wiring_has_id(wiring: Wiring, node_id: UUID) → bool

Return type:

bool

ros_bt_py.exceptions module

exception ros_bt_py.exceptions.AssignmentException

Bases: Exception

exception ros_bt_py.exceptions.BehaviorTreeException

Bases: Exception

exception ros_bt_py.exceptions.MigrationException

Bases: BehaviorTreeException

exception ros_bt_py.exceptions.MissingParentError

Bases: BehaviorTreeException

exception ros_bt_py.exceptions.NodeConfigError

Bases: BehaviorTreeException

exception ros_bt_py.exceptions.NodeStateError

Bases: BehaviorTreeException

exception ros_bt_py.exceptions.TreeTopologyError

Bases: BehaviorTreeException

ros_bt_py.tree_exec_manager module

class ros_bt_py.tree_exec_manager.TreeExecManager(ros_node: rclpy.node.Node, tree_id: UUID = UUID('00000000-0000-0000-0000-000000000000'), name: str = 'UNKNOWN TREE', module_list: List[str] | None = None, debug_manager: DebugManager | None = None, subtree_manager: SubtreeManager | None = None, logging_manager: LoggingManager | None = None, data_flow_manager: DataFlowManager | None = None, tick_frequency_hz: float = 10.0, publish_tree_structure_callback: Callable[[TreeStructureList], None] | None = None, publish_tree_state_callback: Callable[[TreeStateList], None] | None = None, publish_tree_data_callback: Callable[[TreeDataList], None] | None = None, publish_diagnostic_callback: Callable[[diagnostic_msgs.msg.DiagnosticArray], None] | None = None, publish_tick_frequency_callback: Callable[[std_msgs.msg.Float64], None] | None = None, diagnostics_frequency: float = 1.0) → None

Bases: object

Provide methods to load and run a Behavior Tree.

These methods are suited (intended, even) for use as ROS service handlers.

clear(request: ClearTree_Request | None, response: ClearTree_Response) → ClearTree_Response

Return type:

ClearTree_Response

clear_diagnostics_name() → None

Clear the name for ROS diagnostics.

Return type:

None

control_execution(request: ControlTreeExecution_Request, response: ControlTreeExecution_Response) → ControlTreeExecution_Response

Control tree execution.

Parameters:

request (ControlTreeExecution_Request)

Return type:

ControlTreeExecution_Response

Can request a tick, periodic ticking, periodic ticking until the root node reports a result (SUCCEEDED or FAILED), or to stop or reset the entire tree.

data_flow_manager: DataFlowManager

data_to_msg() → TreeData

Return type:

TreeData

debug_manager: DebugManager

diagnostic_callback() → None

Return type:

None

find_root() → Ok[Node | None] | Err[TreeTopologyError]

Find the root node of the tree.

Raises:

TreeTopologyError

if nodes exist, but either no root or multiple roots are found.

Uses the manager-owned _children adjacency list to determine root nodes (nodes not in any child’s list).

Return type:

Union[Ok[Optional[Node]], Err[TreeTopologyError]]

get_logger() → LoggingManager

Return type:

LoggingManager

instantiate_node_from_msg(node_msg: NodeStructure, ros_node: rclpy.node.Node) → Ok[Node] | Err[BehaviorTreeException]

Return type:

Union[Ok[Node], Err[BehaviorTreeException]]

load_tree(request: LoadTree_Request, response: LoadTree_Response) → LoadTree_Response

Load a tree from the given message (which may point to a file).

Parameters:

request (LoadTree_Request)

Return type:

LoadTree_Response

request.tree describes the tree to be loaded, including nodes, wirings and public node data.

If the Tree message itself isn’t populated, but contains a path to load a tree from, we open the file it points to and load that.

load_tree_from_path(request: LoadTreeFromPath_Request, response: LoadTreeFromPath_Response) → LoadTreeFromPath_Response

Wrap around load_tree for convenience.

Return type:

LoadTreeFromPath_Response

logging_manager: LoggingManager

property name: str

publish_data()

Publish the current tree data using the callback supplied to the constructor.

This also checks if data publishing is enabled, so it’s safe to call either way. It will always trigger a state publish either way.

In most cases, you’ll want that callback to publish to a ROS topic.

publish_state()

Publish the current tree state using the callback supplied to the constructor.

In most cases, you’ll want that callback to publish to a ROS topic.

publish_structure()

Publish the current tree structure using the callback supplied to the constructor.

This also triggers a state publish.

In most cases, you’ll want that callback to publish to a ROS topic.

reload_tree(request: ReloadTree_Request | None, response: ReloadTree_Response) → ReloadTree_Response

Reload the currently loaded tree.

Return type:

ReloadTree_Response

property root_id: Ok[UUID] | Err[str]

set_diagnostics_name() → None

Set the tree name for ROS diagnostics.

If the BT has a name, this name will published in diagnostics. Otherwise, the root name of the tree is used.

Return type:

None

set_publish_data(request: std_srvs.srv.SetBool.Request, response: std_srvs.srv.SetBool.Response)

set_publish_subtrees(request: std_srvs.srv.SetBool.Request, response: std_srvs.srv.SetBool.Response) → std_srvs.srv.SetBool.Response

Set the parameters of our SubtreeManager.

Parameters:

request (Request)

Return type:

Response

property state: str

state_to_msg() → TreeState

Return type:

TreeState

structure_to_msg() → TreeStructure

Return type:

TreeStructure

subtree_manager: SubtreeManager

tick() → Ok[None] | Err[BehaviorTreeException]

Execute a tick, starting from the tree’s root.

This behaves differently based on the current configuration of the TreeManager - it can tick once, continuously, until the tree reports a result (either SUCCEEDED or FAILED).

This method should NOT be called directly, but rather triggered via TreeManager.control_execution()!

Return type:

Union[Ok[None], Err[BehaviorTreeException]]

property tick_frequency_hz: float

tick_report_exceptions() → None

Wrap TreeManager.tick() and catch all errors.

Return type:

None

property tree_id: Ok[UUID] | Err[str]

validate_wiring(wiring_msg: Wiring) → Ok[None] | Err[BehaviorTreeException]

Return type:

Union[Ok[None], Err[BehaviorTreeException]]

property wirings: list[Wiring]

ros_bt_py.tree_exec_manager.is_edit_service(func)

Decorate tree editing service handlers to prohibit them from editing while the active tree.

This allows the common behavior of responding with a response that has success=False and an error_message if the tree is not currently editable, relying on all editing service responses to have at least those two members.

It also ensures that all edits are atomic, i.e. external service calls cannot interweave. The lock used to ensure this is a threading.RLock, which means the service handlers can call each other if need be.

ros_bt_py.tree_exec_manager.load_tree_from_file(request: MigrateTree_Request, response: MigrateTree_Response) → MigrateTree_Response

Load a tree file from disk.

Return type:

MigrateTree_Response

ros_bt_py.tree_exec_manager.parse_tree_yaml(tree_yaml: str) → MigrateTree_Response

Return type:

MigrateTree_Response

ros_bt_py.tree_edit_manager module

class ros_bt_py.tree_edit_manager.TreeEditManager(ros_node: rclpy.node.Node, tree_id: UUID = UUID('00000000-0000-0000-0000-000000000000'), name: str = 'UNKNOWN TREE', module_list: List[str] | None = None, debug_manager: DebugManager | None = None, subtree_manager: SubtreeManager | None = None, logging_manager: LoggingManager | None = None, data_flow_manager: DataFlowManager | None = None, tick_frequency_hz: float = 10.0, publish_tree_structure_callback: Callable[[TreeStructureList], None] | None = None, publish_tree_state_callback: Callable[[TreeStateList], None] | None = None, publish_tree_data_callback: Callable[[TreeDataList], None] | None = None, publish_diagnostic_callback: Callable[[diagnostic_msgs.msg.DiagnosticArray], None] | None = None, publish_tick_frequency_callback: Callable[[std_msgs.msg.Float64], None] | None = None, diagnostics_frequency: float = 1.0) → None

Bases: TreeExecManager

Provide methods to edit a Behavior Tree

in addition to the inherited load and run functions.

These methods are suited (intended, even) for use as ROS service handlers.

add_node(request: AddNode_Request, response: AddNode_Response) → AddNode_Response

Add the node in this request to the tree.

Parameters:

request (AddNode_Request) – A request describing the node to add.

Return type:

AddNode_Response

add_node_at_index(request: AddNodeAtIndex_Request, response: AddNodeAtIndex_Response) → AddNodeAtIndex_Response

Add the node in this request to the tree.

The node_id from the request is discarded and a new one is randomly generated. The actual id that the node is assigned is included in the response.

Parameters:

request (AddNodeAtIndex_Request) – A request describing the node to add.

Return type:

AddNodeAtIndex_Response

change_tree_name(request: ChangeTreeName_Request, response: ChangeTreeName_Response) → ChangeTreeName_Response

Change the name of the currently loaded tree.

Return type:

ChangeTreeName_Response

find_nodes_in_cycles() → list[UUID]

Return a list of all nodes in the tree that are part of cycles.

Return type:

list[UUID]

generate_subtree(request: GenerateSubtree_Request, response: GenerateSubtree_Response) → GenerateSubtree_Response

Generate a subtree generated from the provided list of nodes and the loaded tree.

This also adds all relevant parents to the tree message, resulting in a tree that is executable and does not contain any orpahned nodes.

Return type:

GenerateSubtree_Response

get_subtree(request: GetSubtree_Request, response: GetSubtree_Response) → GetSubtree_Response

Return type:

GetSubtree_Response

morph_node(request: MorphNode_Request, response: MorphNode_Response) → MorphNode_Response

Morphs the flow control node into the new node provided in request.new_node.

Return type:

MorphNode_Response

move_node(request: MoveNode_Request, response: MoveNode_Response) → MoveNode_Response

Move the named node to a different parent and insert it at the given index.

Return type:

MoveNode_Response

remove_node(request: RemoveNode_Request, response: RemoveNode_Response) → RemoveNode_Response

Remove the node identified by request.node_name from the tree.

If the parent of the node removed supports enough children to take on all of the removed node’s children, it will. Otherwise, children will be orphaned.

Return type:

RemoveNode_Response

replace_node(request: ReplaceNode_Request, response: ReplaceNode_Response) → ReplaceNode_Response

Replace the named node with new_node.

Will also move all children of the old node to the new one, but only if new_node supports that number of children. Otherwise, this will return an error and leave the tree unchanged.

Return type:

ReplaceNode_Response

set_options(request: SetOptions_Request, response: SetOptions_Response) → SetOptions_Response

Set the option values of a given node.

This is an “edit service”, i.e. it can only be used when the tree has not yet been initialized or has been shut down.

Return type:

SetOptions_Response

unwire_data(request: WireNodeData_Request, response: WireNodeData_Response) → WireNodeData_Response

Disconnect the given pairs of node data.

Parameters:

request (WireNodeData_Request)

Contains a list of ros_bt_py_msgs.msg.NodeDataWiring objects that model connections

Returns:

WireNodeData_Response – ros_bt_py_msgs.src.WireNodeDataResponse or None

wire_data(request: WireNodeData_Request, response: WireNodeData_Response) → WireNodeData_Response

Connect the given pairs of node data to one another.

Parameters:

request (WireNodeData_Request)

Contains a list of :class: ros_bt_py_msgs.msg.NodeDataWiring objects that model connections

Returns:

WireNodeData_Response – ros_bt_py_msgs.src.WireNodeDataResponse or None

ros_bt_py.tree_edit_manager.get_error_message(response: dict | Any) → str

Return type:

str

ros_bt_py.tree_edit_manager.get_success(response: dict | Any) → bool

Return type:

bool

ros_bt_py.data_flow_manager module

class ros_bt_py.data_flow_manager.Connection(source_key, target_id, target_key)

Bases: NamedTuple

source_key: str

Alias for field number 0

target_id: UUID

Alias for field number 1

target_key: str

Alias for field number 2

class ros_bt_py.data_flow_manager.DataFlowManager(incoming_data: dict[str, DataContainer] = {}, outgoing_data: dict[str, DataContainer] = {}) → None

Bases: object

This manages the dataflow for a set of nodes and wirings (usually a tree instance).

On construction we can pass a reference to dictionaries of incoming and outgoing data. The keys of those dictionaries are expected to be in the format node_id.data_key. Keys that cannot be matched are silently ignored, so passing extra is safe.

connections: dict[UUID, list[Connection]]

get_wiring_data() → list[WiringData]

Return type:

list[WiringData]

incoming_data: dict[str, DataContainer]

initialize(nodes: dict[UUID, Node], wirings: list[Wiring]) → Ok[None] | Err[str]

Return type:

Union[Ok[None], Err[str]]

nodes: dict[UUID, Node]

outgoing_data: dict[str, DataContainer]

push_incoming_data() → Ok[None] | Err[str]

Return type:

Union[Ok[None], Err[str]]

push_outputs(node_id: UUID) → Ok[None] | Err[str]

Return type:

Union[Ok[None], Err[str]]

ros_bt_py.logging_manager module

class ros_bt_py.logging_manager.LoggingManager(ros_node: rclpy.node.Node, publish_log_callback: Callable[[BTLogMessage], None] | None = None)

Bases: object

debug(msg: str, node_id: UUID | None = None, node_name: str = '', stacklevel=2, internal=False)

error(msg: str, node_id: UUID | None = None, node_name: str = '', stacklevel=2, internal=False)

fatal(msg: str, node_id: UUID | None = None, node_name: str = '', stacklevel=2, internal=False)

static get_ros_log_name(uuid: UUID, name: str) → str

Return type:

str

get_ros_logger(node_id: UUID | None = None, node_name: str = '') → rclpy.impl.rcutils_logger.RcutilsLogger

Return type:

RcutilsLogger

info(msg: str, node_id: UUID | None = None, node_name: str = '', stacklevel=2, internal=False)

log(level: int, msg: str, node_id: UUID | None = None, node_name: str = '', stacklevel=1)

classmethod log_level_inactive(log_level: int) → bool

Return type:

bool

classmethod set_min_log_level(set_level_req: SetLogLevel_Request, set_level_res: SetLogLevel_Response)

set_tree_id(tree_id: UUID)

set_tree_name(tree_name: str)

warn(msg: str, node_id: UUID | None = None, node_name: str = '', stacklevel=2, internal=False)

ros_bt_py.debug_manager module

class ros_bt_py.debug_manager.DebugManager(ros_node: rclpy.node.Node, node_diagnostics_publish_callback=None)

Bases: object

Manages the collection and publishing of the node diagnostics.

report_state(node_instance, state)

Collect debug state from Node execution.

It measures the time between the beginning and the end of the setup/shutdown/reset/untick function (which includes that of any children).

Additionally, it publishes the node state and execution time to a node diagnostics topic.

Parameters:

• instance – The node

• state – The state of the node

report_tick(node_instance)

Collect debug data during ticks from Node execution.

It measures the time between the beginning and the end of the tick function (which includes the ticks of any children) and calculates a moving window average of execution times as well as a minimum and maximum value.

Parameters:

instance – The node that’s executing

set_collect_node_diagnostics(request: std_srvs.srv.SetBool.Request, response: std_srvs.srv.SetBool.Response) → std_srvs.srv.SetBool.Response

Return type:

Response