ros_sugar.core.event#

Event

Module Contents#

Classes#

InternalEvent

Class to transform a Kompass event to ROS launch event using event key name

OnInternalEvent

ROS EventHandler for InternalEvent.

EventBlackboardEntry

A container for timestamped messages stored in the Event Blackboard.

Event

A runtime trigger driven by ROS2 topic data or a callable condition.

API#

class ros_sugar.core.event.InternalEvent(event_name: str, topics_value: Dict)#

Bases: launch.event.Event

Class to transform a Kompass event to ROS launch event using event key name

property event_name#

Getter of internal event name

Returns:

Event name

Return type:

str

property topics_value#

Getter of internal event name

Returns:

Event name

Return type:

str

class ros_sugar.core.event.OnInternalEvent(*, internal_event_name: str, entities: ros_sugar.utils.SomeEntitiesType, handle_once: bool = False)#

Bases: launch.event_handler.EventHandler

ROS EventHandler for InternalEvent.

handle(event: launch.event.Event, context) → Optional[ros_sugar.utils.SomeEntitiesType]#

Overriding handle to inject event data into the entities.

class ros_sugar.core.event.EventBlackboardEntry#

A container for timestamped messages stored in the Event Blackboard.

This class wraps raw ROS messages with metadata to enable:

  1. Time-To-Live (TTL) Checks: Using timestamp to invalidate old data.

  2. Idempotency: Using a unique id to prevent the same message instance from triggering the same event multiple times.

Parameters:

  • msg (Any) – The actual data payload (e.g., a ROS message).

  • timestamp (float) – The standard Unix timestamp (float) when the message was received.

  • id (str) – A unique UUID4 string identifying this specific message reception instance. Defaults to a new UUID if not provided.

validate(timeout: Optional[float] = None, stale_id: Optional[str] = None)#

Validate the data freshness

Parameters:

  • timeout (Optional[float], optional) – Maximum lifetime, defaults to None

  • stale_id (Optional[str], optional) – ID of the latest stale message, defaults to None

classmethod get(entries_dict: Dict[str, ros_sugar.core.event.EventBlackboardEntry], topic_name: str, timeout: Optional[float], stale_id: Optional[str] = None) → Optional[Any]#

Retrieves data from entries_dict:

  • If data is missing: returns None

  • If data is present but expired: Deletes it and returns None (Lazy Expiration)

  • If data is valid: returns the entry

class ros_sugar.core.event.Event(event_condition: Union[ros_sugar.io.topic.Topic, ros_sugar.condition.Condition, Callable], on_change: bool = False, handle_once: bool = False, keep_event_delay: float = 0.0, check_rate: Optional[float] = None)#

A runtime trigger driven by ROS2 topic data or a callable condition.

An Event monitors one of three condition types and fires registered Actions when the condition becomes true:

  1. Condition expression – a declarative predicate on topic message attributes (e.g. topic.msg.speed > 5.0). The condition is evaluated each time new data arrives on the involved topics.

  2. Topic (on-any) – triggers whenever a valid message is available on the given topic(s), regardless of content.

  3. Callable – a user-supplied function polled at check_rate whose boolean return value determines the trigger state.

Trigger behaviour can be further refined with:

  • on_change – fire only on a rising edge (condition transitions from false to true), rather than every evaluation cycle. A condition that is already true at its first evaluation counts as a rising edge.

  • handle_once – fire at most once and then become inert.

  • keep_event_delay – hold the under-processing flag for a fixed duration after actions complete, throttling re-triggers.

Actions are executed asynchronously via a shared ThreadPoolExecutor.

property under_processing: bool#

If event is triggered and associated action is getting executed

Returns:

Event under processing flag

Return type:

bool

property id: str#

Getter of the event unique id

Returns:

Unique ID

Return type:

str

reset()#

Reset event processing

clear() → None#

Clear event trigger

raise_event_trigger() → None#

Raise event trigger

to_dict() → Dict#

Property to parse the event into a dictionary

Returns:

Event description dictionary

Return type:

Dict

classmethod from_dict(dict_obj: Dict)#

Setter of the event using a dictionary

Parameters:

dict_obj (Dict) – Event description dictionary

to_json() → str#

Property to get/set the event using a json

Returns:

Event description dictionary as json

Return type:

str

classmethod from_json(json_obj: Union[str, bytes, bytearray])#

Property to get/set the event using a json

Parameters:

json_obj (Union[str, bytes, bytearray]) – Event description dictionary as json

get_involved_topics() → List[ros_sugar.io.topic.Topic]#

Get all the topics required for monitoring this event

Returns:

Required topics

Return type:

List[Topic]

get_last_processed_id(topic_name: str) → Optional[str]#

Get the unique ID of the last processed message for a given topic

Parameters:

topic_name (str) – Topic name

Returns:

The last unique ID if the topic was processed earlier, otherwise None

Return type:

Optional[str]

verify_required_action_topics(action: ros_sugar.core.base_action.BaseAction) → None#

Verify the action topic parsers (if present) against an event. Raises a ‘ValueError’ if there is a mismatch.

Parameters:

event (Event) – Event to verify against

register_actions(actions: Union[ros_sugar.core.base_action.BaseAction, Callable, List[Union[ros_sugar.core.base_action.BaseAction, Callable]]]) → None#

Register an Action or a set of Actions to execute on trigger

Parameters:

actions (Union[BaseAction, List[BaseAction]]) – Action or a list of Actions

clear_actions() → None#

Clear all registered on trigger Actions

check_condition(global_topic_cache: Dict[str, ros_sugar.core.event.EventBlackboardEntry]) → None#

Replaces existing trigger logic. Evaluates the root Condition tree against the global cache.

Serialized per event: its host no longer holds one lock across every event it owns, so two topics of this event’s own can arrive at once, and deciding whether it fires reads and writes state that only makes sense one evaluation at a time.

check_action_condition(global_topic_cache: Dict[str, ros_sugar.core.event.EventBlackboardEntry]) → None#

Evaluate the action-based condition and execute registered actions if triggered. Called periodically by a component timer; should only be used on action-based events.