dust_dds

Dust DDS python bindings

1from .dust_dds import *
2
3__doc__ = dust_dds.__doc__
4if hasattr(dust_dds, "__all__"):
5    __all__ = dust_dds.__all__
class DomainParticipantFactory:
def get_instance():
def create_participant(self, /, domain_id, qos=None, a_listener=None, mask=Ellipsis):
def delete_participant(self, /, a_participant):
def lookup_participant(self, /, domain_id):
def set_default_participant_qos(self, /, qos):
def get_default_participant_qos(self, /):
def set_qos(self, /, qos):
def get_qos(self, /):
class DomainParticipant:
def create_publisher(self, /, qos=None, a_listener=None, mask=Ellipsis):
def delete_publisher(self, /, a_publisher):
def create_subscriber(self, /, qos=None, a_listener=None, mask=Ellipsis):
def delete_subscriber(self, /, a_subscriber):
def create_topic(self, /, topic_name, type_, qos=None, a_listener=None, mask=Ellipsis):
def delete_topic(self, /, a_topic):
def lookup_topicdescription(self, /, topic_name):
def get_builtin_subscriber(self, /):
def ignore_participant(self, /, handle):
def ignore_topic(self, /, handle):
def ignore_publication(self, /, handle):
def ignore_subscription(self, /, handle):
def get_domain_id(self, /):
def delete_contained_entities(self, /):
def assert_liveliness(self, /):
def set_default_publisher_qos(self, /, qos):
def get_default_publisher_qos(self, /):
def set_default_subscriber_qos(self, /, qos):
def get_default_subscriber_qos(self, /):
def set_default_topic_qos(self, /, qos):
def get_default_topic_qos(self, /):
def get_discovered_participants(self, /):
def get_discovered_participant_data(self, /, participant_handle):
def get_discovered_topics(self, /):
def get_discovered_topic_data(self, /, topic_handle):
def contains_entity(self, /, a_handle):
def get_current_time(self, /):
def set_qos(self, /, qos):
def get_qos(self, /):
def set_listener(self, /, a_listener=None, mask=Ellipsis):
def get_status_changes(self, /):
def enable(self, /):
def get_instance_handle(self, /):
class Publisher:
def create_datawriter(self, /, a_topic, qos=None, a_listener=None, mask=Ellipsis):
def delete_datawriter(self, /, a_datawriter):
def lookup_datawriter(self, /, topic_name):
def suspend_publications(self, /):
def resume_publications(self, /):
def begin_coherent_changes(self, /):
def end_coherent_changes(self, /):
def wait_for_acknowledgments(self, /, max_wait):
def get_participant(self, /):
def delete_contained_entities(self, /):
def set_default_datawriter_qos(self, /, qos):
def get_default_datawriter_qos(self, /):
def set_qos(self, /, qos):
def get_qos(self, /):
def set_listener(self, /, a_listener=None, mask=Ellipsis):
def get_status_changes(self, /):
def enable(self, /):
def get_instance_handle(self, /):
class DataWriter:
def register_instance(self, /, instance):
def register_instance_w_timestamp(self, /, instance, timestamp):
def unregister_instance(self, /, instance, handle=None):
def unregister_instance_w_timestamp(self, /, instance, handle, timestamp):
def get_key_value(self, /, _key_holder, _handle):
def lookup_instance(self, /, instance):
def write(self, /, data, handle=None):
def write_w_timestamp(self, /, data, handle, timestamp):
def dispose(self, /, data, handle):
def dispose_w_timestamp(self, /, data, handle, timestamp):
def wait_for_acknowledgments(self, /, max_wait):
def get_liveliness_lost_status(self, /):
def get_offered_deadline_missed_status(self, /):
def get_offered_incompatible_qos_status(self, /):
def get_publication_matched_status(self, /):
def get_topic(self, /):
def get_publisher(self, /):
def assert_liveliness(self, /):
def get_matched_subscription_data(self, /, subscription_handle):
def get_matched_subscriptions(self, /):
def set_qos(self, /, qos):
def get_qos(self, /):
def set_listener(self, /, a_listener=None, mask=Ellipsis):
def get_statuscondition(self, /):
def get_status_changes(self, /):
def enable(self, /):
def get_instance_handle(self, /):
class Subscriber:
def create_datareader(self, /, a_topic, qos=None, a_listener=None, mask=Ellipsis):
def delete_datareader(self, /, a_datareader):
def lookup_datareader(self, /, topic_name):
def notify_datareaders(self, /):
def get_participant(self, /):
def get_sample_lost_status(self, /):
def delete_contained_entities(self, /):
def set_default_datareader_qos(self, /, qos):
def get_default_datareader_qos(self, /):
def set_qos(self, /, qos):
def get_qos(self, /):
def set_listener(self, /, a_listener=None, mask=Ellipsis):
def get_statuscondition(self, /):
def get_status_changes(self, /):
def enable(self, /):
def get_instance_handle(self, /):
class DataReader:
def read( self, /, max_samples, sample_states=Ellipsis, view_states=Ellipsis, instance_states=Ellipsis):
def take( self, /, max_samples, sample_states=Ellipsis, view_states=Ellipsis, instance_states=Ellipsis):
def read_next_sample(self, /):
def take_next_sample(self, /):
def read_instance( self, /, max_samples, a_handle, sample_states=Ellipsis, view_states=Ellipsis, instance_states=Ellipsis):
def take_instance( self, /, max_samples, a_handle, sample_states=Ellipsis, view_states=Ellipsis, instance_states=Ellipsis):
def read_next_instance( self, /, max_samples, previous_handle, sample_states=Ellipsis, view_states=Ellipsis, instance_states=Ellipsis):
def take_next_instance( self, /, max_samples, previous_handle, sample_states=Ellipsis, view_states=Ellipsis, instance_states=Ellipsis):
def get_key_value(self, /, _key_holder, _handle):
def lookup_instance(self, /, _instance):
def get_liveliness_changed_status(self, /):
def get_requested_deadline_missed_status(self, /):
def get_requested_incompatible_qos_status(self, /):
def get_sample_lost_status(self, /):
def get_sample_rejected_status(self, /):
def get_subscription_matched_status(self, /):
def get_topicdescription(self, /):
def get_subscriber(self, /):
def wait_for_historical_data(self, /, max_wait):
def get_matched_publication_data(self, /, publication_handle):
def get_matched_publications(self, /):
def set_qos(self, /, qos):
def get_qos(self, /):
def set_listener(self, /, a_listener=None, mask=Ellipsis):
def get_statuscondition(self, /):
def get_status_changes(self, /):
def enable(self, /):
def get_instance_handle(self, /):
class Topic:
def get_inconsistent_topic_status(self, /):
def get_participant(self, /):
def get_type_name(self, /):
def get_name(self, /):
def set_qos(self, /, qos):
def get_qos(self, /):
def set_listener(self, /, a_listener=None, mask=Ellipsis):
def get_statuscondition(self, /):
def get_status_changes(self, /):
def enable(self, /):
def get_instance_handle(self, /):
class TypeKind:
boolean = TypeKind.boolean
char8 = TypeKind.char8
int8 = TypeKind.int8
uint8 = TypeKind.uint8
int16 = TypeKind.int16
uint16 = TypeKind.uint16
int32 = TypeKind.int32
uint32 = TypeKind.uint32
int64 = TypeKind.int64
uint64 = TypeKind.uint64
float32 = TypeKind.float32
float64 = TypeKind.float64
class TopicDescription:
def get_participant(self, /):
def get_type_name(self, /):
def get_name(self, /):
class InstanceHandle:
class InconsistentTopicStatus:
def get_total_count(self, /):
def get_total_count_change(self, /):
class LivelinessChangedStatus:
def get_alive_count(self, /):
def get_not_alive_count(self, /):
def get_alive_count_change(self, /):
def get_not_alive_count_change(self, /):
def get_last_publication_handle(self, /):
class LivelinessLostStatus:
def get_total_count(self, /):
def get_total_count_change(self, /):
class OfferedDeadlineMissedStatus:
def get_total_count(self, /):
def get_total_count_change(self, /):
def get_last_instance_handle(self, /):
class OfferedIncompatibleQosStatus:
def get_total_count(self, /):
def get_total_count_change(self, /):
def get_last_policy_id(self, /):
def get_policies(self, /):
class PublicationMatchedStatus:
def get_total_count(self, /):
def get_total_count_change(self, /):
def get_last_subscription_handle(self, /):
def get_current_count(self, /):
def get_current_count_change(self, /):
class QosPolicyCount:
def get_policy_id(self, /):
def get_count(self, /):
class RequestedDeadlineMissedStatus:
def get_total_count(self, /):
def get_total_count_change(self, /):
def get_last_instance_handle(self, /):
class RequestedIncompatibleQosStatus:
def get_total_count(self, /):
def get_total_count_change(self, /):
def get_last_policy_id(self, /):
def get_policies(self, /):
class SampleLostStatus:
def get_total_count(self, /):
def get_total_count_change(self, /):
class SampleRejectedStatus:
def get_total_count(self, /):
def get_total_count_change(self, /):
def get_last_reason(self, /):
def get_last_instance_handle(self, /):
class StatusKind:
InconsistentTopic = StatusKind.InconsistentTopic
OfferedDeadlineMissed = StatusKind.OfferedDeadlineMissed
RequestedDeadlineMissed = StatusKind.RequestedDeadlineMissed
OfferedIncompatibleQos = StatusKind.OfferedIncompatibleQos
RequestedIncompatibleQos = StatusKind.RequestedIncompatibleQos
SampleLost = StatusKind.SampleLost
SampleRejected = StatusKind.SampleRejected
DataOnReaders = StatusKind.DataOnReaders
DataAvailable = StatusKind.DataAvailable
LivelinessLost = StatusKind.LivelinessLost
LivelinessChanged = StatusKind.LivelinessChanged
PublicationMatched = StatusKind.PublicationMatched
SubscriptionMatched = StatusKind.SubscriptionMatched
class SubscriptionMatchedStatus:
def get_total_count(self, /):
def get_total_count_change(self, /):
def get_last_publication_handle(self, /):
def get_current_count(self, /):
def get_current_count_change(self, /):
class Duration:
def get_sec(self, /):
def get_nanosec(self, /):
class DurationKind:
class DurationKind.Finite(dust_dds.DurationKind):
duration
class DurationKind.Infinite(dust_dds.DurationKind):
class Time:
def get_sec(self, /):
def get_nanosec(self, /):
class Condition:
class Condition.StatusCondition(dust_dds.Condition):
condition
class WaitSet:
def wait(self, /, timeout):
def attach_condition(self, /, cond):
def detach_condition(self, /, cond):
def get_conditions(self, /):
class StatusCondition:
def get_enabled_statuses(self, /):
def set_enabled_statuses(self, /, mask):
def get_trigger_value(self, /):
class BuiltInTopicKey:
def get_value(self, /):
class ParticipantBuiltinTopicData:
def get_key(self, /):
def get_user_data(self, /):
class PublicationBuiltinTopicData:
def get_key(self, /):
def participant_key(self, /):
def topic_name(self, /):
def get_type_name(self, /):
def get_durability(self, /):
def get_deadline(self, /):
def get_latency_budget(self, /):
def get_liveliness(self, /):
def get_reliability(self, /):
def get_lifespan(self, /):
def get_user_data(self, /):
def get_ownership(self, /):
def get_destination_order(self, /):
def get_presentation(self, /):
def get_partition(self, /):
def get_topic_data(self, /):
def get_group_data(self, /):
class SubscriptionBuiltinTopicData:
def get_key(self, /):
def participant_key(self, /):
def topic_name(self, /):
def get_type_name(self, /):
def get_durability(self, /):
def get_deadline(self, /):
def get_latency_budget(self, /):
def get_liveliness(self, /):
def get_reliability(self, /):
def get_ownership(self, /):
def get_destination_order(self, /):
def get_user_data(self, /):
def get_time_based_filter(self, /):
def get_presentation(self, /):
def get_partition(self, /):
def get_topic_data(self, /):
def get_group_data(self, /):
class TopicBuiltinTopicData:
def get_key(self, /):
def get_name(self, /):
def get_type_name(self, /):
def get_durability(self, /):
def get_deadline(self, /):
def get_latency_budget(self, /):
def get_liveliness(self, /):
def get_reliability(self, /):
def get_transport_priority(self, /):
def get_lifespan(self, /):
def get_destination_order(self, /):
def get_history(self, /):
def get_resource_limits(self, /):
def get_ownership(self, /):
def get_topic_data(self, /):
class DeadlineQosPolicy:
def get_period(self, /):
def set_period(self, /, value):
class DestinationOrderQosPolicy:
def get_kind(self, /):
def set_kind(self, /, value):
class DestinationOrderQosPolicyKind:
class DurabilityQosPolicy:
def get_kind(self, /):
def set_kind(self, /, value):
class DurabilityQosPolicyKind:
class EntityFactoryQosPolicy:
def get_autoenable_created_entities(self, /):
def set_autoenable_created_entities(self, /, value):
class GroupDataQosPolicy:
def get_value(self, /):
def set_value(self, /, value):
class HistoryQosPolicy:
def get_kind(self, /):
def set_kind(self, /, value):
class HistoryQosPolicyKind:
class HistoryQosPolicyKind.KeepLast(dust_dds.HistoryQosPolicyKind):
depth
class HistoryQosPolicyKind.KeepAll(dust_dds.HistoryQosPolicyKind):
class LatencyBudgetQosPolicy:
def get_duration(self, /):
def set_duration(self, /, value):
class Length:
class Length.Unlimited(dust_dds.Length):
class Length.Limited(dust_dds.Length):
length
class LifespanQosPolicy:
def get_duration(self, /):
def set_duration(self, /, value):
class LivelinessQosPolicy:
def get_kind(self, /):
def set_kind(self, /, value):
def get_lease_duration(self, /):
def set_lease_duration(self, /, value):
class LivelinessQosPolicyKind:
class OwnershipQosPolicy:
def get_kind(self, /):
def set_kind(self, /, value):
class OwnershipQosPolicyKind:
class PartitionQosPolicy:
def get_name(self, /):
def set_name(self, /, value):
class PresentationQosPolicy:
def get_access_scope(self, /):
def set_access_scope(self, /, value):
def get_coherent_access(self, /):
def set_coherent_access(self, /, value):
def get_ordered_access(self, /):
def set_ordered_access(self, /, value):
class ReaderDataLifecycleQosPolicy:
def get_autopurge_nowriter_samples_delay(self, /):
def set_autopurge_nowriter_samples_delay(self, /, value):
def get_autopurge_disposed_samples_delay(self, /):
def set_autopurge_disposed_samples_delay(self, /, value):
class ReliabilityQosPolicy:
def get_kind(self, /):
def set_kind(self, /, value):
def get_max_blocking_time(self, /):
def set_max_blocking_time(self, /, value):
class ReliabilityQosPolicyKind:
class ResourceLimitsQosPolicy:
def get_max_samples(self, /):
def set_max_samples(self, /, value):
def get_max_instances(self, /):
def set_max_instances(self, /, value):
def get_max_samples_per_instance(self, /):
def set_max_samples_per_instance(self, /, value):
class TimeBasedFilterQosPolicy:
def get_minimum_separation(self, /):
def set_minimum_separation(self, /, value):
class TopicDataQosPolicy:
def get_value(self, /):
def set_value(self, /, value):
class TransportPriorityQosPolicy:
def get_value(self, /):
def set_value(self, /, value):
class UserDataQosPolicy:
def get_value(self, /):
def set_value(self, /, value):
class DataRepresentationQosPolicy:
class OwnershipStrengthQosPolicy:
class WriterDataLifecycleQosPolicy:
def get_autodispose_unregistered_instances(self, /):
def set_autodispose_unregistered_instances(self, /, value):
class DataReaderQos:
def get_durability(self, /):
def get_deadline(self, /):
def get_latency_budget(self, /):
def get_liveliness(self, /):
def get_reliability(self, /):
def get_destination_order(self, /):
def get_history(self, /):
def get_resource_limits(self, /):
def get_user_data(self, /):
def get_ownership(self, /):
def get_time_based_filter(self, /):
def get_reader_data_lifecycle(self, /):
def get_representation(self, /):
def get_type_consistency(self, /):
class DataWriterQos:
def get_durability(self, /):
def get_deadline(self, /):
def get_latency_budget(self, /):
def get_liveliness(self, /):
def get_reliability(self, /):
def get_destination_order(self, /):
def get_history(self, /):
def get_resource_limits(self, /):
def get_transport_priority(self, /):
def get_lifespan(self, /):
def get_user_data(self, /):
def get_ownership(self, /):
def get_writer_data_lifecycle(self, /):
class DomainParticipantFactoryQos:
def get_entity_factory(self, /):
class DomainParticipantQos:
def get_user_data(self, /):
def get_entity_factory(self, /):
class PublisherQos:
def get_presentation(self, /):
def set_presentation(self, /, value):
def get_partition(self, /):
def set_partition(self, /, value):
def get_group_data(self, /):
def set_group_data(self, /, value):
def get_entity_factory(self, /):
def set_entity_factory(self, /, value):
class SubscriberQos:
class TopicQos:
def get_topic_data(self, /):
def get_durability(self, /):
def get_deadline(self, /):
def get_latency_budget(self, /):
def get_liveliness(self, /):
def get_reliability(self, /):
def get_destination_order(self, /):
def get_history(self, /):
def get_resource_limits(self, /):
def get_transport_priority(self, /):
def get_lifespan(self, /):
def get_ownership(self, /):
class Sample:
def get_data(self, /):
def get_sample_info(self, /):
class SampleInfo:
def get_sample_state(self, /):
def get_view_state(self, /):
def get_instance_state(self, /):
def get_disposed_generation_count(self, /):
def get_no_writers_generation_count(self, /):
def get_sample_rank(self, /):
def get_generation_rank(self, /):
def get_absolute_generation_rank(self, /):
def get_source_timestamp(self, /):
def get_instance_handle(self, /):
def get_publication_handle(self, /):
def get_valid_data(self, /):
class SampleStateKind:
ANY_SAMPLE_STATE = [SampleStateKind.Read, SampleStateKind.NotRead]
class ViewStateKind:
ANY_VIEW_STATE = [ViewStateKind.New, ViewStateKind.NotNew]
class InstanceStateKind:
ANY_INSTANCE_STATE = [InstanceStateKind.Alive, InstanceStateKind.NotAliveDisposed, InstanceStateKind.NotAliveNoWriters]
NOT_ALIVE_INSTANCE_STATE = [InstanceStateKind.NotAliveDisposed, InstanceStateKind.NotAliveNoWriters]
DEFAULT_RELIABILITY_QOS_POLICY_DATA_READER_AND_TOPICS = <builtins.ReliabilityQosPolicy object>
DEFAULT_RELIABILITY_QOS_POLICY_DATA_WRITER = <builtins.ReliabilityQosPolicy object>