`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`¶