rclpy.event_handler module
- class rclpy.event_handler.EventHandler(*, callback_group: CallbackGroup, callback: Callable[[...], None], event_type: rpyutils.import_c_library.rcl_subscription_event_type_t, parent_impl: rpyutils.import_c_library.Subscription.typing.Any)
- class rclpy.event_handler.EventHandler(*, callback_group: CallbackGroup, callback: Callable[[...], None], event_type: rpyutils.import_c_library.rcl_publisher_event_type_t, parent_impl: rpyutils.import_c_library.Publisher.typing.Any)
Bases:
Waitable
[rmw_requested_deadline_missed_status_t
|rmw_liveliness_changed_status_t
|rmw_message_lost_status_t
|rmw_requested_qos_incompatible_event_status_t
|rmw_incompatible_type_status_t
|rmw_matched_status_t
|rmw_offered_deadline_missed_status_t
|rmw_liveliness_lost_status_t
|_rclpy.rmw_offered_qos_incompatible_event_status_t
|rmw_matched_status_t
|None
]Waitable type to handle QoS events.
- add_to_wait_set(wait_set: rpyutils.import_c_library.WaitSet) None
Add entites to wait set.
- destroy() None
- async execute(taken_data: rpyutils.import_c_library.rmw_requested_deadline_missed_status_t | rpyutils.import_c_library.rmw_liveliness_changed_status_t | rpyutils.import_c_library.rmw_message_lost_status_t | rpyutils.import_c_library.rmw_requested_qos_incompatible_event_status_t | rpyutils.import_c_library.rmw_incompatible_type_status_t | rpyutils.import_c_library.rmw_matched_status_t | rpyutils.import_c_library.rmw_offered_deadline_missed_status_t | rpyutils.import_c_library.rmw_liveliness_lost_status_t | rpyutils.import_c_library.rmw_offered_qos_incompatible_event_status_t | rpyutils.import_c_library.rmw_matched_status_t | None) None
Execute work after data has been taken from a ready wait set.
- get_num_entities() NumberOfEntities
Return number of each type of entity used.
- is_ready(wait_set: rpyutils.import_c_library.WaitSet) bool
Return True if entities are ready in the wait set.
- take_data() rpyutils.import_c_library.rmw_requested_deadline_missed_status_t | rpyutils.import_c_library.rmw_liveliness_changed_status_t | rpyutils.import_c_library.rmw_message_lost_status_t | rpyutils.import_c_library.rmw_requested_qos_incompatible_event_status_t | rpyutils.import_c_library.rmw_incompatible_type_status_t | rpyutils.import_c_library.rmw_matched_status_t | rpyutils.import_c_library.rmw_offered_deadline_missed_status_t | rpyutils.import_c_library.rmw_liveliness_lost_status_t | rpyutils.import_c_library.rmw_offered_qos_incompatible_event_status_t | rpyutils.import_c_library.rmw_matched_status_t | None
Take stuff from lower level so the wait set doesn’t immediately wake again.
- rclpy.event_handler.IncompatibleTypeInfo: TypeAlias
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.typing.Any, topic_name: str) List[EventHandler]
- class rclpy.event_handler.QoSEventHandler(**kwargs)
Bases:
EventHandler
- rclpy.event_handler.QoSPublisherMatchedInfo: TypeAlias
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.typing.Any, topic_name: str) List[EventHandler]