ddspy: initial qos_provider support impl.

  Implementation of `QosProvider` class.

  Support for `qos_provider` functionality in python API + basic test
  implemntation.

Author: Splinter1984

cyclonedds/__init__.py

@@ -10,7 +10,7 @@
  * SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause
 """
 
-from . import internal, util, qos, core, domain, topic, pub, sub, builtin, dynamic, idl
+from . import internal, util, qos, core, domain, topic, pub, sub, builtin, dynamic, idl, qos_provider
 
 __all__ = [
     "internal",
@@ -24,4 +24,5 @@
     "builtin",
     "dynamic",
     "idl",
+    "qos_provider",
 ]

cyclonedds/internal.py

@@ -285,13 +285,23 @@ class stat_kind(IntEnum):
     DDS_STAT_KIND_LENGTHTIME = 2
 
 
+class qos_kind(IntEnum):
+    DDS_PARTICIPANT_QOS = 0
+    DDS_PUBLISHER_QOS = 1
+    DDS_SUBSCRIBER_QOS = 2
+    DDS_TOPIC_QOS = 3
+    DDS_READER_QOS = 4
+    DDS_WRITER_QOS = 5
+
+
 class stat_value(ct.Union):
     _fields_ = [
         ('u32', ct.c_uint32),
         ('u64', ct.c_uint64),
         ('lengthtime', ct.c_uint64)
     ]
 
+
 class stat_keyvalue(ct.Structure):
     _fields_ = [
         ('name', ct.c_char_p),
@@ -319,6 +329,8 @@ class dds_c_t:  # noqa N801
     destination_order = ct.c_int
     data_representation_id = ct.c_int16
     qos_p = ct.c_void_p
+    qos_provider_p = ct.c_void_p
+    qos_kind = ct.c_int32
     attach = ct.c_void_p
     listener_p = ct.c_void_p
     topic_descriptor_p = ct.c_void_p

cyclonedds/qos_provider.py

@@ -0,0 +1,97 @@
+"""
+ * Copyright(c) 2025 ZettaScale Technology and others
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v. 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0, or the Eclipse Distribution License
+ * v. 1.0 which is available at
+ * http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * SPDX-License-Identifier: EPL-2.0 OR BSD-3-Clause
+"""
+
+import ctypes as ct
+from typing import Any, Optional
+
+from .core import DDSException
+from .internal import DDS, c_call, dds_c_t, qos_kind
+from .qos import (
+    DataReaderQos, DataWriterQos, DomainParticipantQos, PublisherQos, Qos,
+    SubscriberQos, TopicQos, _CQos,
+)
+
+
+class QosProvider(DDS):
+    """
+    Representing QosProvider
+    """
+
+    _qos_provider: dds_c_t.qos_provider_p
+
+    def __init__(
+            self,
+            path: Any,
+            scope: Optional[str] = ""):
+        bpath = path
+        if not isinstance(path, str) and not isinstance(path, bytes):
+            bpath = path.__str__().encode('utf-8')
+        elif isinstance(path, str):
+            bpath = path.encode('utf-8')
+        bscope = scope.encode('utf-8')
+        self._qos_provider = dds_c_t.qos_provider_p(0)
+        ret = self._create_qos_provider_scope(
+            bpath, ct.byref(self._qos_provider), bscope
+        )
+        if ret < 0 or not self._qos_provider:
+            raise DDSException(ret, f"Occured when initialize {repr(self)}")
+
+    def _get_qos_kind(self, key: str, kind: qos_kind) -> Qos:
+        bkey = key.encode('utf-8')
+        cqos = dds_c_t.qos_p(0)
+        ret = self._qos_provider_get_qos(
+            self._qos_provider, kind, bkey, ct.byref(cqos)
+        )
+        if ret < 0:
+            raise DDSException(ret, f"Occured trying to get_topic_qos {repr(self)}")
+        return _CQos.cqos_to_qos(cqos)
+
+    def get_datawriter_qos(self, key: str = "") -> 'DataWriterQos':
+        return self._get_qos_kind(key, qos_kind.DDS_WRITER_QOS).datawriter()
+
+    def get_datareader_qos(self, key: str = "") -> 'DataReaderQos':
+        return self._get_qos_kind(key, qos_kind.DDS_READER_QOS).datareader()
+
+    def get_domain_participant_qos(self, key: str = "") -> 'DomainParticipantQos':
+        return self._get_qos_kind(key, qos_kind.DDS_PARTICIPANT_QOS).domain_participant()
+
+    def get_publisher_qos(self, key: str = "") -> 'PublisherQos':
+        return self._get_qos_kind(key, qos_kind.DDS_PUBLISHER_QOS).publisher()
+
+    def get_subscriber_qos(self, key: str = "") -> 'SubscriberQos':
+        return self._get_qos_kind(key, qos_kind.DDS_SUBSCRIBER_QOS).subscriber()
+
+    def get_topic_qos(self, key: str = "") -> 'TopicQos':
+        return self._get_qos_kind(key, qos_kind.DDS_TOPIC_QOS).topic()
+
+    def __del__(self) -> None:
+        self._delete_qos_provider(self._qos_provider)
+
+    @c_call("dds_create_qos_provider_scope")
+    def _create_qos_provider_scope(
+        self, path: ct.c_char_p, provider: ct.POINTER(dds_c_t.qos_provider_p),
+        scope: ct.c_char_p
+    ) -> dds_c_t.returnv:
+        pass
+
+    @c_call("dds_delete_qos_provider")
+    def _delete_qos_provider(
+        self, provider: dds_c_t.qos_provider_p
+    ) -> None:
+        pass
+
+    @c_call("dds_qos_provider_get_qos")
+    def _qos_provider_get_qos(
+        self, provider: dds_c_t.qos_provider_p, qos_type: dds_c_t.qos_kind,
+        key: ct.c_char_p, qos: ct.POINTER(dds_c_t.qos_p)
+    ) -> dds_c_t.returnv:
+        pass

tests/test_qos_provider.py

@@ -0,0 +1,90 @@
+import pytest
+import os
+from pathlib import Path
+from cyclonedds.core import DDSException
+from cyclonedds.qos_provider import QosProvider
+from cyclonedds.qos import (
+    DataReaderQos, DataWriterQos, DomainParticipantQos, PublisherQos, Qos,
+    SubscriberQos, TopicQos,
+)
+
+
+@pytest.fixture(scope="function")
+def sysdef_file(tmp_path):
+    q_str = """<dds>
+      <qos_library name="qoslib01">
+        <qos_profile name="qosprofile01">
+          <datareader_qos>
+            <history><kind>KEEP_LAST_HISTORY_QOS</kind><depth>5</depth></history>
+          </datareader_qos>
+          <datawriter_qos>
+            <history><kind>KEEP_LAST_HISTORY_QOS</kind><depth>5</depth></history>
+          </datawriter_qos>
+          <topic_qos>
+            <resource_limits><max_instances>5</max_instances></resource_limits>
+          </topic_qos>
+          <publisher_qos>
+            <partition><name><element>part02</element></name></partition>
+          </publisher_qos>
+          <subscriber_qos>
+            <partition><name><element>part01</element></name></partition>
+          </subscriber_qos>
+          <domain_participant_qos>
+            <user_data><value>ZCBzdHJpbmcgZXhhbXBsZQ==</value></user_data>
+          </domain_participant_qos>
+        </qos_profile>
+      </qos_library>
+    </dds>"""
+    tmp_file = (tmp_path / "sysdef_common_test.xml")
+    assert tmp_file.write_text(q_str) != 0
+    return tmp_file
+
+
+def test_qos_provider_create(sysdef_file):
+    qp0 = QosProvider(sysdef_file, "*::qosprofile01")
+    wr_q = qp0.get_datawriter_qos("qoslib01::qosprofile01")
+    assert isinstance(wr_q, DataWriterQos)
+    qp1 = QosProvider("<dds></dds>")
+    ret = True
+    try:
+        wr_q = qp1.get_datawriter_qos("qoslib01::qosprofile01")
+    except DDSException:
+        assert ret  # nothing to get from empty definition
+        ret = False
+    assert not ret
+
+
+def test_qos_provider_get_topic(sysdef_file):
+    qp = QosProvider(sysdef_file, "*::*")
+    qs = Qos.fromdict({"ResourceLimits": {"max_instances": 5}}).topic()
+    assert qp.get_topic_qos("qoslib01::qosprofile01") == qs
+
+
+def test_qos_provider_get_subscriber(sysdef_file):
+    qp = QosProvider(str(sysdef_file), "::::")
+    qs = Qos.fromdict({"Partition": {"partitions": ["part01"]}}).subscriber()
+    assert qp.get_subscriber_qos("qoslib01::qosprofile01") == qs
+
+
+def test_qos_provider_get_publisher(sysdef_file):
+    qp = QosProvider(str(sysdef_file), "::")
+    qs = Qos.fromdict({"Partition": {"partitions": ["part02"]}}).publisher()
+    assert qp.get_publisher_qos("qoslib01::qosprofile01") == qs
+
+
+def test_qos_provider_get_datareader(sysdef_file):
+    qp = QosProvider(str(sysdef_file).encode('utf-8'), "")
+    qs = Qos.fromdict({"History": {"kind": "KeepLast", "depth": 5}}).datareader()
+    assert qp.get_datareader_qos("qoslib01::qosprofile01") == qs
+
+
+def test_qos_provider_get_datawriter(sysdef_file):
+    qp = QosProvider(sysdef_file)
+    qs = Qos.fromdict({"History": {"kind": "KeepLast", "depth": 5}}).datawriter()
+    assert qp.get_datawriter_qos("qoslib01::qosprofile01") == qs
+
+
+def test_qos_provider_get_participant(sysdef_file):
+    qp = QosProvider(sysdef_file)
+    qs = Qos.fromdict({"Userdata": {"data": "ZCBzdHJpbmcgZXhhbXBsZQ=="}}).domain_participant()
+    assert qp.get_domain_participant_qos("qoslib01::qosprofile01") == qs