core

class cyclonedds.core.Qos(*policies, base=None)[source]

Bases: object

This class represents a collections of policies. It allows for easy inspection of this set. When you retrieve a Qos object from an entity modifying that object would actually does not change the Qos of the entity. To reflect this Qos objects are immutable.

x == y

Checks if two Qos objects contain the same policies. This is a full comparison, not a match.

x != y

Checks if two Qos objects do not contain the same policies.

p in qos

Check if a Policy p is contained in Qos object qos. You can use all levels of generalization, for example: Policy.History in qos, Policy.History.KeepLast in qos and Policy.History.KeepLast(1) in qos.

qos[p]

Obtain the Policy matched with p from the Qos object, for example: qos[Policy.History] -> Policy.History.KeepAll

iter(x)

The Qos object supports iteration over it’s contents.

len(x)

Return the number of Policies in the Qos object.

str(x)

Human-readable description of the contained Qos policies.

Parameters

base (Optional[Qos]) –

policies

A sorted tuple of the Policies contained in this Qos object

Type

Tuple[BasePolicy]

__init__(*policies, base=None)[source]

Initialize a Qos object

Parameters
  • *policies (BasePolicy) – Pass in any number of constructed Policies.

  • base (Qos, optional) – Optionally inherit policies from another Qos object. Inherited policies are overwritten by those newly set.

Raises
  • TypeError – If you pass something that is not a Policy or use a base that is not a Qos object this will be treated as a TypeError.

  • ValueError – If you pass two overlapping Policies, for example Policy.History.KeepLast(10) and Policy.History.KeepAll this will be treated as a ValueError.

asdict()[source]

Convert a Qos object to a python dictionary.

Returns

Fully describe the Qos object using a python dictionary with only built-in types (dict, list, string, int, boolean). This format is not guaranteed to stay consistent between cyclonedds versions but can be useful for debugging or use within an application.

Return type

dict

classmethod fromdict(data)[source]

Convert a python dictionary as generated by asdict() to a Qos object.

Returns

Note that the format of the python dictionary is not guaranteed between cyclonedds versions thus storing these dictionaries to disk and loading them again is not recommended.

Return type

Qos

Parameters

data (dict) –

class cyclonedds.core.Policy(**kwargs)[source]

The Policy class is fully static and should never need to be instantiated.

See also

qoshowto

How to work with Qos and Policy, TODO.

class Reliability(**kwargs)[source]

The Reliability Qos Policy

Examples

>>> Policy.Reliability.BestEffort
>>> Policy.Reliability.Reliable(max_blocking_time=duration(seconds=1))
Reliable[source]

alias of cyclonedds.qos.Policy.Reliability.Reliable

class Durability(**kwargs)[source]

The Durability Qos Policy

Examples

>>> Policy.Durability.Volatile
>>> Policy.Durability.TransientLocal
>>> Policy.Durability.Transient
>>> Policy.Durability.Persistent
class History(**kwargs)[source]

The History Qos Policy

Examples

>>> Policy.History.KeepAll
>>> Policy.History.KeepLast(depth=10)
KeepAll

The type of this entity is not publicly specified.

Type

Tuple[PolicyType, Any]

KeepLast[source]

alias of cyclonedds.qos.Policy.History.KeepLast

class ResourceLimits(max_samples=- 1, max_instances=- 1, max_samples_per_instance=- 1)[source]

The ResourceLimits Qos Policy

Examples

>>> Policy.ResourceLimits(
>>>     max_samples=10,
>>>     max_instances=10,
>>>     max_samples_per_instance=2
>>> )
Parameters
  • max_samples (int) –

  • max_instances (int) –

  • max_samples_per_instance (int) –

Return type

None

max_samples

Max number of samples total.

Type

int

max_instances

Max number of instances total.

Type

int

max_samples_per_instance

Max number of samples per instance.

Type

int

class PresentationAccessScope(**kwargs)[source]

The Presentation Access Scope Qos Policy

Examples

>>> Policy.PresentationAccessScope.Instance(coherent_access=True, ordered_access=False)
>>> Policy.PresentationAccessScope.Topic(coherent_access=True, ordered_access=False)
>>> Policy.PresentationAccessScope.Group(coherent_access=True, ordered_access=False)
Instance[source]

alias of cyclonedds.qos.Policy.PresentationAccessScope.Instance

Topic[source]

alias of cyclonedds.qos.Policy.PresentationAccessScope.Topic

Group[source]

alias of cyclonedds.qos.Policy.PresentationAccessScope.Group

class Lifespan(lifespan)[source]

The Lifespan Qos Policy

Examples

>>> Policy.Lifespan(duration(seconds=2))
Parameters

lifespan (int) –

Return type

None

lifespan

Expiration time relative to the source timestamp of a sample in nanoseconds.

Type

int

class Deadline(deadline)[source]

The Deadline Qos Policy

Examples

>>> Policy.Deadline(deadline=duration(seconds=2))
Parameters

deadline (int) –

Return type

None

deadline

Deadline of a sample in nanoseconds.

Type

int

class LatencyBudget(budget)[source]

The Latency Budget Qos Policy

Examples

>>> Policy.LatencyBudget(duration(seconds=2))
Parameters

budget (int) – Latency budget in nanoseconds.

Return type

None

class Ownership(**kwargs)[source]

The Ownership Qos Policy

Examples

>>> Policy.Ownership.Shared
>>> Policy.Ownership.Exclusive
Shared
Type

Policy.Ownership.Shared

Exclusive
Type

Policy.Ownership.Exclusive

class OwnershipStrength(strength)[source]

The Ownership Strength Qos Policy

Examples

>>> Policy.OwnershipStrength(strength=2)
Parameters

strength (int) – Ownership strength as integer.

Return type

None

class Liveliness(**kwargs)[source]

The Liveliness Qos Policy

Examples

>>> Policy.Liveliness.Automatic(lease_duration=duration(seconds=10))
>>> Policy.Liveliness.ManualByParticipant(lease_duration=duration(seconds=10))
>>> Policy.Liveliness.ManualByTopic(lease_duration=duration(seconds=10))
Automatic[source]

alias of cyclonedds.qos.Policy.Liveliness.Automatic

ManualByParticipant[source]

alias of cyclonedds.qos.Policy.Liveliness.ManualByParticipant

ManualByTopic[source]

alias of cyclonedds.qos.Policy.Liveliness.ManualByTopic

class TimeBasedFilter(filter_time)[source]

The TimeBasedFilter Qos Policy

Examples

>>> Policy.TimeBasedFilter(filter_fn=duration(seconds=2))
Parameters

filter_time (int) –

Return type

None

filter_time

Minimum time between samples in nanoseconds. Use the helper function duration to write the duration in a human readable format.

Type

int

class Partition(partitions)[source]

The Partition Qos Policy

Examples

>>> Policy.Partition(partitions=["partition_a", "partition_b", "partition_c"])
>>> Policy.Partition(partitions=[f"partition_{i}" for i in range(100)])
Parameters

partitions (Sequence[str]) –

Return type

None

partitions
Type

Sequence[str]

class TransportPriority(priority)[source]

The TransportPriority Qos Policy

Examples

>>> Policy.TransportPriority(priority=10)
Parameters

priority (int) –

Return type

None

priority
Type

int

class DestinationOrder[source]

The DestinationOrder Qos Policy

Examples

>>> Policy.DestinationOrder.ByReceptionTimestamp
>>> Policy.DestinationOrder.BySourceTimestamp
class WriterDataLifecycle(autodispose)[source]

The WriterDataLifecycle Qos Policy

Examples

>>> Policy.WriterDataLifecycle(autodispose=False)
Parameters

autodispose (bool) –

Return type

None

autodispose
Type

bool

class ReaderDataLifecycle(autopurge_nowriter_samples_delay, autopurge_disposed_samples_delay)[source]

The ReaderDataLifecycle Qos Policy

Examples

>>> Policy.ReaderDataLifecycle(
>>>     autopurge_nowriter_samples_delay=duration(minutes=2),
>>>     autopurge_disposed_samples_delay=duration(minutes=5)
>>> )
Parameters
  • autopurge_nowriter_samples_delay (int) –

  • autopurge_disposed_samples_delay (int) –

Return type

None

autopurge_nowriter_samples_delay
Type

bool

autopurge_disposed_samples_delay
Type

bool

class DurabilityService(cleanup_delay, history, max_samples, max_instances, max_samples_per_instance)[source]

The DurabilityService Qos Policy

Examples

>>> Policy.DurabilityService(
>>>     cleanup_delay=duration(minutes=2.5),
>>>     history=Policy.History.KeepLast(20),
>>>     max_samples=2000,
>>>     max_instances=200,
>>>     max_samples_per_instance=25
>>> )
Parameters
  • cleanup_delay (int) –

  • history (Union[Policy.History.KeepAll, Policy.History.KeepLast]) –

  • max_samples (int) –

  • max_instances (int) –

  • max_samples_per_instance (int) –

Return type

None

cleanup_delay
Type

int

history
Type

Policy.History.KeepAll, Policy.History.KeepLast

max_samples
Type

int

max_instances
Type

int

max_samples_per_instance
Type

int

class IgnoreLocal(**kwargs)[source]

The IgnoreLocal Qos Policy

Examples

>>> Policy.IgnoreLocal.Nothing
>>> Policy.IgnoreLocal.Participant
>>> Policy.IgnoreLocal.Process
class Userdata(data)[source]

The Userdata Qos Policy

Examples

>>> Policy.Userdata(data=b"Hello, World!")
Parameters

data (bytes) –

Return type

None

class Topicdata(data)[source]

The Topicdata Qos Policy

Examples

>>> Policy.Topicdata(data=b"Hello, World!")
Parameters

data (bytes) –

Return type

None

class Groupdata(data)[source]

The Groupdata Qos Policy

Examples

>>> Policy.Groupdata(data=b"Hello, World!")
Parameters

data (bytes) –

Return type

None

class Property(key, value)[source]

The Property Qos Policy

Examples

>>> Policy.Property(key="host", value="central")
Parameters
  • key (str) –

  • value (str) –

Return type

None

class BinaryProperty(key, value)[source]

The BinaryProperty Qos Policy

Examples

>>> Policy.BinaryProperty(key="host", value=b"central")
Parameters
Return type

None

class TypeConsistency(**kwargs)[source]

The TypeConsistency Qos Policy

Examples

>>> Policy.TypeConsistency.DisallowTypeCoercion
>>> Policy.TypeConsistency.AllowTypeCoercion
DisallowTypeCoercion[source]

alias of cyclonedds.qos.Policy.TypeConsistency.DisallowTypeCoercion

AllowTypeCoercion[source]

alias of cyclonedds.qos.Policy.TypeConsistency.AllowTypeCoercion

class DataRepresentation(use_cdrv0_representation=False, use_xcdrv2_representation=False)[source]

The DataRepresentation Qos Policy

Parameters
  • use_cdrv0_representation (bool) –

  • use_xcdrv2_representation (bool) –

Return type

None

class EntityName(name)[source]

The EntityName Qos Policy

Parameters

name (str) –

Return type

None