`ros_sugar.condition`#

Module Contents#

Classes#

`ConditionLogicOp`
`Condition`	Condition class used for defining events on incoming topic data. It captures the logic of an expression like `topic.msg.data > 5` to be evaluated at runtime.
`ConditionOperators`	Registry of supported operators for Conditions. Enables serialization by mapping function names to callables.
`MsgConditionBuilder`	Helper class to build paths for accessing ROS message attributes in topics. Used for parsing topic message attributes for Actions and Event parsers.

API#

class ros_sugar.condition.ConditionLogicOp(*args, **kwds)#

Bases: enum.Enum

name()#

value()#

class ros_sugar.condition.Condition(sub_conditions: Optional[List[ros_sugar.condition.Condition]] = None, logic_operator: ros_sugar.condition.ConditionLogicOp = ConditionLogicOp.NONE, topic_name: Optional[str] = None, topic_msg_type: Optional[str] = None, topic_qos_config: Optional[Dict] = None, attribute_path: Optional[List[str]] = None, operator_func: Optional[Callable] = None, ref_value: Any = None)#

Condition class used for defining events on incoming topic data. It captures the logic of an expression like topic.msg.data > 5 to be evaluated at runtime.

Attribute	Type	Description
topic_source	`Topic`	The source Topic object that provides the data stream.
attribute_path	`List[str \| int]`	The traversal path to access the specific field within the ROS message (e.g., `['header', 'stamp', 'sec']`).
operator_func	`Callable`	The comparison function (from `ConditionOperators`) used to evaluate the message field against the reference value.
ref_value	`Any`	The reference value (or list of values) to compare the message field against.

to_dict() → Dict[str, Any]#: Convert the condition tree to a dictionary.

classmethod from_dict(data: Dict[str, Any]) → ros_sugar.condition.Condition#: Reconstruct the condition tree from a dictionary.

to_json() → str#: Serialize to JSON string.

classmethod from_json(json_str: str) → ros_sugar.condition.Condition#: Deserialize from JSON string.

evaluate(topic_data_cache: Dict[str, Any]) → bool#

Recursively evaluate the condition using all sub-conditions

Parameters:: topic_data_cache (Dict[str, Any]) – A dictionary mapping topic_name -> latest_ros_msg
Returns:: True if the condition is met, else False
Return type:: bool

class ros_sugar.condition.ConditionOperators#

Registry of supported operators for Conditions. Enables serialization by mapping function names to callables.

classmethod get_operator(name: str) → Callable#: Retrieve function by name (for Deserialization).

classmethod get_name(func: Callable) → str#: Retrieve name by function (for Serialization).

static is_in(op_obj, ref_list)#: Checks if Topic Value is IN Reference List

static not_in(op_obj, ref_list)#: Checks if Topic Value is NOT IN Reference List

static contains(op_obj, ref_val)#: Topic (string) contains a reference string

static not_contains(op_obj, ref_val)#: Topic (string) does not contain a reference string

static contains_any(op_obj, ref_list)#: Topic (list) contains ANY of Reference (list)

static contains_all(op_obj, ref_list)#: Topic (list) contains ALL of Reference (list)

static not_contains_any(op_obj, ref_list)#: Topic (list) contains NONE of Reference (list)

static not_contains_all(op_obj, ref_list)#: Topic (list) does MISSING at least one of Reference (list)

class ros_sugar.condition.MsgConditionBuilder(topic, path=None)#

Helper class to build paths for accessing ROS message attributes in topics. Used for parsing topic message attributes for Actions and Event parsers.

is_in(other: Union[List, Tuple, str]) → ros_sugar.condition.Condition#: Check if the topic value is inside the provided list/tuple. Usage: topic.msg.status.is_in([1, 2, 3])

not_in(other: Union[List, Tuple, str]) → ros_sugar.condition.Condition#: Check if the topic value is NOT inside the provided list/tuple. Usage: topic.msg.status.not_in([0, -1])

contains_any(other: Union[List, Tuple]) → ros_sugar.condition.Condition#: True if the topic value contains AT LEAST ONE of the values in ‘other’. Equivalent to: set(topic) & set(other) is not empty

contains_all(other: Union[List, Tuple]) → ros_sugar.condition.Condition#: True if the topic value contains ALL of the values in ‘other’. Equivalent to: set(other).issubset(set(topic))

contains(other: Union[List, Tuple, str]) → ros_sugar.condition.Condition#: If value contains another string If other is a list: works same as contains_all

not_contains_any(other: Union[List, Tuple]) → ros_sugar.condition.Condition#: True if the topic value contains NONE of the values in ‘other’. Equivalent to: set(topic).isdisjoint(set(other))

not_contains_all(other: Union[List, Tuple]) → ros_sugar.condition.Condition#: True if the topic value is MISSING at least one value from ‘other’. Inverse of contains_all.

not_contains(other: Union[List, Tuple, str]) → ros_sugar.condition.Condition#: If value does not contain another string If other is a list: works same as not_contains_all

is_true() → ros_sugar.condition.Condition#: Create a condition checking if the boolean attribute is True. Usage: topic.msg.is_enabled.is_true()

is_false() → ros_sugar.condition.Condition#: Create a condition checking if the boolean attribute is False. Usage: topic.msg.is_enabled.is_false()

ros_sugar.condition#

Module Contents#

Classes#

API#

`ros_sugar.condition`#