rclpy.event_handler module

class rclpy.event_handler.EventHandler(*, callback_group: CallbackGroup, callback: Callable, event_type: IntEnum, parent_impl)

Bases: Waitable

Waitable type to handle QoS events.

add_to_wait_set(wait_set)

Add entites to wait set.

destroy()
async execute(taken_data)

Execute work after data has been taken from a ready wait set.

get_num_entities()

Return number of each type of entity used.

is_ready(wait_set)

Return True if entities are ready in the wait set.

take_data()

Take stuff from lower level so the wait set doesn’t immediately wake again.

rclpy.event_handler.IncompatibleTypeInfo

Raised when registering a callback for an event type that is not supported.

class rclpy.event_handler.PublisherEventCallbacks(*, deadline: Callable[[rpyutils.import_c_library.rmw_offered_deadline_missed_status_t], None] | None = None, liveliness: Callable[[rpyutils.import_c_library.rmw_liveliness_lost_status_t], None] | None = None, incompatible_qos: Callable[[rpyutils.import_c_library.rmw_requested_qos_incompatible_event_status_t], None] | None = None, incompatible_type: Callable[[rpyutils.import_c_library.rmw_incompatible_type_status_t], None] | None = None, matched: Callable[[rpyutils.import_c_library.rmw_matched_status_t], None] | None = None, use_default_callbacks: bool = True)

Bases: object

Container to provide middleware event callbacks for a Publisher.

Create and return a PublisherEventCallbacks container.

Parameters:

  • deadline – A user-defined callback that is called when the Publisher misses its offered Deadline.

  • liveliness – A user-defined callback that is called when this Publisher fails to signal its Liveliness and is reported as not-alive.

  • incompatible_qos – A user-defined callback that is called when a Subscription with incompatible QoS policies is discovered on subscribed topic.

  • incompatible_type – A user-defined callback that is called when a topic type doesn’t match.

  • matched – A user-defined callback that is called when a Subscription is connected or disconnected.

  • use_default_callbacks – Whether or not to use default callbacks when the user doesn’t supply one

create_event_handlers(callback_group: CallbackGroup, publisher: rpyutils.import_c_library.Publisher, topic_name: str) → List[EventHandler]
class rclpy.event_handler.QoSEventHandler(*args, **kwargs)

Bases: EventHandler

rclpy.event_handler.QoSPublisherMatchedInfo

Payload type for Publisher Incompatible QoS callback.

Mirrors rmw_offered_incompatible_qos_status_t from rmw/types.h

class rclpy.event_handler.SubscriptionEventCallbacks(*, deadline: Callable[[rpyutils.import_c_library.rmw_requested_deadline_missed_status_t], None] | None = None, incompatible_qos: Callable[[rpyutils.import_c_library.rmw_requested_qos_incompatible_event_status_t], None] | None = None, liveliness: Callable[[rpyutils.import_c_library.rmw_liveliness_changed_status_t], None] | None = None, message_lost: Callable[[rpyutils.import_c_library.rmw_message_lost_status_t], None] | None = None, incompatible_type: Callable[[rpyutils.import_c_library.rmw_incompatible_type_status_t], None] | None = None, matched: Callable[[rpyutils.import_c_library.rmw_matched_status_t], None] | None = None, use_default_callbacks: bool = True)

Bases: object

Container to provide middleware event callbacks for a Subscription.

Create a SubscriptionEventCallbacks container.

Parameters:

  • deadline – A user-defined callback that is called when a topic misses our requested Deadline.

  • incompatible_qos – A user-defined callback that is called when a Publisher with incompatible QoS policies is discovered on subscribed topic.

  • liveliness – A user-defined callback that is called when the Liveliness of a Publisher on subscribed topic changes.

  • message_lost – A user-defined callback that is called when a messages is lost.

  • incompatible_type – A user-defined callback that is called when a topic type doesn’t match.

  • matched – A user-defined callback that is called when a Publisher is connected or disconnected.

  • use_default_callbacks – Whether or not to use default callbacks when the user doesn’t supply one

create_event_handlers(callback_group: CallbackGroup, subscription: rpyutils.import_c_library.Subscription, topic_name: str) → List[EventHandler]