PR #12924 from Eran: Flow controller & misc

src/device.h

@@ -21,10 +21,30 @@
 namespace librealsense {
 
 
+// Device profiles undergo format conversion before becoming available to the user. This is typically done within each
+// sensor's init_stream_profiles(), but the behavior can be overriden by the user through the device JSON settings (see
+// device ctor) to enable querying the device's raw or basic formats only.
+// 
+// Note that streaming may be available only with full (the default) profiles. The others are for inspection only, and
+// can directly be queried in rs-enumerate-devices.
+// 
+// Note also that default profiles (those returned by get_profiles_tags()) must take the conversion method into account
+// or profiles may not get tagged correctly in non-full modes.
+//
 enum class format_conversion
 {
+    // Report raw profiles as provided by the camera, without any manipulation whatsoever by librealsense.
     raw,
+
+    // Take the raw profiles, perform the librealsense mappings, but then remove any "unwanted" mappings:
+    //      - any conversion between formats is thrown away (e.g., YUYV->RGB8)
+    //      - colored infrared is removed
+    //      - interleaved (Y12I, Y8I) are kept (so become Y16 and Y8, respectively)
+    // See formats_converter::drop_non_basic_formats()
     basic,
+
+    // The default conversion mode: all the librealsense mappings are intact; the user will see profiles including
+    // format conversions (e.g., YUYV->RGB8), etc.
     full
 };
 

third-party/realdds/doc/initialization.md

@@ -174,33 +174,48 @@ Information about a specific stream:
           float bias_variances[3];   
       }
       ```
-    e.g.:
-      ```JSON
-      "intrinsics": {
-          "accel": [1.0,0.0,0.0,0.0,0.0,1.0,0.0,0.0,0.0,0.0,1.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0],
-          "gyro": [1.0,0.0,0.0,0.0,0.0,1.0,0.0,0.0,0.0,0.0,1.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0]
-      }
-      ```
-- `options` is an array of option objects, same as `device-options` above
+- `options` is an array of option objects, same as `device-options` above; stream options are shown in the Viewer
+- `recommended-filters` is an array of filter names to be enabled in the Viewer
 
-Stream options are shown in the Viewer.
+E.g.:
+
+```JSON
+{
+    "id": "stream-options",
+    "stream-name": "Motion",
+    "intrinsics": {
+        "accel": [1.0,0.0,0.0,0.0,0.0,1.0,0.0,0.0,0.0,0.0,1.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0],
+        "gyro": [1.0,0.0,0.0,0.0,0.0,1.0,0.0,0.0,0.0,0.0,1.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0]
+    },
+    "options": [],
+    "recommended-filters": []
+}
+```
 
 #### Video Stream Intrinsics
 
 ```JSON
 {
     "id": "stream-options",
+    "stream-name": "Depth",
     "intrinsics": {
         "width": 1280,
         "height": 720,
         "principal-point": [640.2379150390625,357.3431396484375],
         "focal-length": [631.3428955078125,631.3428955078125]
     },
-    "options": [
-        ["Backlight Compensation",0.0,0.0,1.0,1.0,0.0,"Enable / disable backlight compensation"],
-        ["Brightness",0.0,-64.0,64.0,1.0,0.0,"UVC image brightness"],
-    ],
-    "stream-name": "Infrared 1"
+    "options": [],
+    "recommended-filters": [
+        "Decimation Filter",
+        "HDR Merge",
+        "Filter By Sequence id",
+        "Threshold Filter",
+        "Depth to Disparity",
+        "Spatial Filter",
+        "Temporal Filter",
+        "Hole Filling Filter",
+        "Disparity to Depth"
+    ]
 }
 ```
 

third-party/realdds/include/realdds/dds-participant.h

@@ -1,9 +1,11 @@
 // License: Apache 2.0. See LICENSE file in root directory.
-// Copyright(c) 2022 Intel Corporation. All Rights Reserved.
+// Copyright(c) 2022-4 Intel Corporation. All Rights Reserved.
 #pragma once
 
 #include "dds-defines.h"
 
+#include <fastdds/dds/domain/qos/DomainParticipantQos.hpp>
+
 #include <rsutils/json.h>
 #include <memory>
 #include <functional>
@@ -57,12 +59,31 @@ class dds_participant
     dds_participant( const dds_participant & ) = delete;
     ~dds_participant();
 
+public:
+    // Centralizes our default QoS settings, so that the user can then override before calling init().
+    // Note that init() will try to override further with information from the settings it is passed.
+    //
+    class qos : public eprosima::fastdds::dds::DomainParticipantQos
+    {
+        using super = eprosima::fastdds::dds::DomainParticipantQos;
+
+    public:
+        qos( std::string const & participant_name );
+    };
+
     // Creates the underlying DDS participant and sets the QoS.
     // If callbacks are needed, set them before calling init. Note they may be called before init returns!
     // 
     // The domain ID may be -1: in this case the settings "domain" is queried and a default of 0 is used
     //
-    void init( dds_domain_id, std::string const & participant_name, rsutils::json const & settings );
+    void init( dds_domain_id did, std::string const & participant_name, rsutils::json const & settings )
+    {
+        qos pqos( participant_name );
+        init( did, pqos, settings );
+    }
+    // Same, with custom QoS
+    //
+    void init( dds_domain_id, qos &, rsutils::json const & settings );
 
     bool is_valid() const { return ( nullptr != _participant ); }
     bool operator!() const { return ! is_valid(); }

third-party/realdds/include/realdds/dds-serialization.h

@@ -1,5 +1,5 @@
 // License: Apache 2.0. See LICENSE file in root directory.
-// Copyright(c) 2023 Intel Corporation. All Rights Reserved.
+// Copyright(c) 2023-4 Intel Corporation. All Rights Reserved.
 #pragma once
 
 #include <fastdds/dds/core/policy/QosPolicies.hpp>
@@ -15,14 +15,25 @@ std::ostream & operator<<( std::ostream &, ReliabilityQosPolicyKind );
 std::ostream & operator<<( std::ostream &, ReliabilityQosPolicy const & );
 std::ostream & operator<<( std::ostream &, DurabilityQosPolicyKind );
 std::ostream & operator<<( std::ostream &, DurabilityQosPolicy const & );
+std::ostream & operator<<( std::ostream &, HistoryQosPolicyKind );
 std::ostream & operator<<( std::ostream &, HistoryQosPolicy const & );
 std::ostream & operator<<( std::ostream &, LivelinessQosPolicyKind );
 std::ostream & operator<<( std::ostream &, LivelinessQosPolicy const & );
 std::ostream & operator<<( std::ostream &, DataSharingQosPolicy const & );
 std::ostream & operator<<( std::ostream &, RTPSEndpointQos const & );
+std::ostream & operator<<( std::ostream &, PublishModeQosPolicy const & );
 
 class DomainParticipantQos;
+std::ostream & operator<<( std::ostream &, DomainParticipantQos const & );
+
+class DataWriterQos;
+std::ostream & operator<<( std::ostream &, DataWriterQos const & );
+
 }  // namespace dds
+namespace rtps {
+std::ostream & operator<<( std::ostream &, TransportDescriptorInterface const & );
+std::ostream & operator<<( std::ostream &, FlowControllerDescriptor const & );
+}
 }  // namespace fastdds
 }  // namespace eprosima
 
@@ -37,6 +48,7 @@ void from_json( rsutils::json const &, Duration_t & );
 namespace rtps {
 std::ostream & operator<<( std::ostream &, class WriterProxyData const & );
 std::ostream & operator<<( std::ostream &, class ReaderProxyData const & );
+std::ostream & operator<<( std::ostream &, BuiltinAttributes const & );
 }  // namespace rtps
 
 }  // namespace fastrtps
@@ -46,6 +58,9 @@ std::ostream & operator<<( std::ostream &, class ReaderProxyData const & );
 namespace realdds {
 
 
+class dds_participant;
+
+
 eprosima::fastdds::dds::ReliabilityQosPolicyKind reliability_kind_from_string( std::string const & );
 eprosima::fastdds::dds::DurabilityQosPolicyKind durability_kind_from_string( std::string const & );
 eprosima::fastdds::dds::HistoryQosPolicyKind history_kind_from_string( std::string const & );
@@ -101,6 +116,7 @@ void override_liveliness_qos_from_json( eprosima::fastdds::dds::LivelinessQosPol
 //
 void override_data_sharing_qos_from_json( eprosima::fastdds::dds::DataSharingQosPolicy & qos, rsutils::json const & );
 
+
 // Override QoS endpoint from a JSON source.
 // The JSON is an object:
 //      {
@@ -110,6 +126,25 @@ void override_data_sharing_qos_from_json( eprosima::fastdds::dds::DataSharingQos
 void override_endpoint_qos_from_json( eprosima::fastdds::dds::RTPSEndpointQos & qos, rsutils::json const & );
 
 
+// Override QoS publish- from a JSON source.
+// The JSON is an object:
+//      {
+//          "flow-control": "name"
+//      }
+//
+void override_publish_mode_qos_from_json( eprosima::fastdds::dds::PublishModeQosPolicy & qos, rsutils::json const &, dds_participant const & );
+
+
+// Override FlowController settings from a JSON source.
+// The JSON is an object:
+//      {
+//          "max-bytes-per-period": 16384,
+//          "period-ms": 250
+//      }
+//
+void override_flow_controller_from_json( eprosima::fastdds::rtps::FlowControllerDescriptor &, rsutils::json const & );
+
+
 // Override participant QoS from a JSON source.
 // The JSON is an object:
 //      {

third-party/realdds/include/realdds/dds-topic-reader.h

@@ -4,10 +4,12 @@
 
 #include <fastdds/dds/subscriber/DataReaderListener.hpp>
 #include <fastdds/dds/subscriber/qos/DataReaderQos.hpp>
+#include "dds-defines.h"
 
 #include <rsutils/json-fwd.h>
 #include <functional>
 #include <memory>
+#include <atomic>
 
 
 namespace eprosima {
@@ -41,7 +43,7 @@ class dds_topic_reader
 
     eprosima::fastdds::dds::DataReader * _reader = nullptr;
 
-    int _n_writers = 0;
+    std::atomic< int > _n_writers;
 
 public:
     dds_topic_reader( std::shared_ptr< dds_topic > const & topic );
@@ -89,6 +91,9 @@ class dds_topic_reader
     // The callbacks should be set before we actually create the underlying DDS objects, so the reader does not
     virtual void run( qos const & );
 
+    // Waits until writers are detected; return false on timeout
+    bool wait_for_writers( dds_time timeout );
+
     // Go back to a pre-run() state, such that is_running() returns false
     virtual void stop();
 

third-party/realdds/include/realdds/dds-topic-writer.h

@@ -1,6 +1,5 @@
 // License: Apache 2.0. See LICENSE file in root directory.
-// Copyright(c) 2022 Intel Corporation. All Rights Reserved.
-
+// Copyright(c) 2022-4 Intel Corporation. All Rights Reserved.
 #pragma once
 
 #include <fastdds/dds/publisher/DataWriterListener.hpp>
@@ -73,11 +72,11 @@ class dds_topic_writer : protected eprosima::fastdds::dds::DataWriterListener
                = eprosima::fastdds::dds::RELIABLE_RELIABILITY_QOS,  // default
              eprosima::fastdds::dds::DurabilityQosPolicyKind durability
                = eprosima::fastdds::dds::VOLATILE_DURABILITY_QOS );  // default is transient local
-
-        // Override default values with JSON contents
-        void override_from_json( rsutils::json const & );
     };
 
+    // Override default values with JSON contents
+    void override_qos_from_json( qos &, rsutils::json const & );
+
     // The callbacks should be set before we actually create the underlying DDS objects, so the writer does not
     void run( qos const & = qos() );
 

third-party/realdds/py/pyrealdds.cpp

@@ -97,9 +97,17 @@ json load_rs_settings( json const & local_settings )
 
     // Patch any script-specific settings
     // NOTE: this is also accessed by pyrealsense2, where a "context" hierarchy is still used
-    auto script = script_name();
-    if( auto script_settings = config.nested( script, "context", "dds" ) )
-        settings.override( script_settings, "config-file/" + script + "/context" );
+    try
+    {
+        auto script = script_name();
+        if( auto script_settings = config.nested( script, "context", "dds" ) )
+            settings.override( script_settings, "config-file/" + script + "/context" );
+    }
+    catch( std::exception const & e )
+    {
+        // Expected if we're not in a script
+        LOG_DEBUG( "failed to get script name: " << e.what() );
+    }
 
     // We should always have DDS enabled
     if( settings.is_object() )
@@ -257,19 +265,15 @@ PYBIND11_MODULE(NAME, m) {
               []( dds_participant & self, json const & local_settings, realdds::dds_domain_id domain_id )
                     { self.init( domain_id, script_name(), local_settings ); },
               "local-settings"_a = json::object(), "domain-id"_a = -1 )
-        .def( "init", &dds_participant::init, "domain-id"_a, "participant-name"_a, "local-settings"_a = json::object() )
+        .def( "init",
+              py::overload_cast< realdds::dds_domain_id, std::string const &, json const & >( &dds_participant::init ),
+              "domain-id"_a, "participant-name"_a, "local-settings"_a = json::object() )
         .def( "is_valid", &dds_participant::is_valid )
         .def( "guid", &dds_participant::guid )
         .def( "create_guid", &dds_participant::create_guid )
         .def( "__bool__", &dds_participant::is_valid )
-        .def( "name",
-              []( dds_participant const & self ) {
-                  eprosima::fastdds::dds::DomainParticipantQos qos;
-                  if( ReturnCode_t::RETCODE_OK == self.get()->get_qos( qos ) )
-                      return std::string( qos.name() );
-                  return std::string();
-              } )
-        .def( "name_from_guid", []( dds_guid const & guid ) { return dds_participant::name_from_guid( guid ); } )
+        .def( "name", &dds_participant::name )
+        .def_static( "name_from_guid", []( dds_guid const & guid ) { return dds_participant::name_from_guid( guid ); } )
         .def( "names", []( dds_participant const & self ) { return self.get()->get_participant_names(); } )
         .def( "settings", &dds_participant::settings )
         .def( "__repr__",
@@ -282,9 +286,7 @@ PYBIND11_MODULE(NAME, m) {
                   }
                   else
                   {
-                      eprosima::fastdds::dds::DomainParticipantQos qos;
-                      if( ReturnCode_t::RETCODE_OK == self.get()->get_qos( qos ) )
-                          os << " \"" << qos.name() << "\"";
+                      os << " \"" << self.name() << "\"";
                       os << " " << realdds::print_guid( self.guid() );
                   }
                   os << ">";
@@ -406,14 +408,22 @@ PYBIND11_MODULE(NAME, m) {
                       callback( self, status.total_count, status.total_count_change ); ) )
         .def( "topic", &dds_topic_reader::topic )
         .def( "run", &dds_topic_reader::run )
+        .def( "wait_for_writers", &dds_topic_reader::wait_for_writers )
+        .def( "stop", &dds_topic_reader::stop, 
+              // GIL release needed: stop will wait and cause a hang if we're inside a callback
+              py::call_guard< py::gil_scoped_release >() )
         .def( "qos", []() { return reader_qos(); } )
         .def( "qos", []( reliability r, durability d ) { return reader_qos( r, d ); } );
 
     using writer_qos = realdds::dds_topic_writer::qos;
     py::class_< writer_qos >( m, "writer_qos" )  //
+        .def_property_readonly( "flow_controller",
+                                []( writer_qos const & self ) -> std::string
+                                { return self.publish_mode().flow_controller_name; } )
         .def( "__repr__", []( writer_qos const & self ) {
             std::ostringstream os;
             os << "<" SNAME ".writer_qos";
+            os << self;
             os << ">";
             return os.str();
         } );
@@ -436,6 +446,7 @@ PYBIND11_MODULE(NAME, m) {
                       ( eprosima::fastdds::dds::PublicationMatchedStatus const & status ),
                       callback( self, status.current_count_change ); ) )
         .def( "topic", &dds_topic_writer::topic )
+        .def( "override_qos_from_json", &dds_topic_writer::override_qos_from_json )
         .def( "run", &dds_topic_writer::run )
         .def( "has_readers", &dds_topic_writer::has_readers )
         .def( "wait_for_readers", &dds_topic_writer::wait_for_readers )
@@ -578,6 +589,7 @@ PYBIND11_MODULE(NAME, m) {
         .def_readwrite( "width", &image_msg::width )
         .def_readwrite( "height", &image_msg::height )
         .def_readwrite( "timestamp", &image_msg::timestamp )
+        .def( "__bool__", &image_msg::is_valid )
         .def( "__repr__",
               []( image_msg const & self )
               {
@@ -627,6 +639,8 @@ PYBIND11_MODULE(NAME, m) {
                                                              std::string const & ) >( &blob_msg::create_topic ) )
         .def( "data", []( blob_msg const & self ) { return self.data(); } )
         .def( "size", []( blob_msg const & self ) { return self.data().size(); } )
+        .def( "crc", []( blob_msg const & self ) { return rsutils::number::calc_crc32( self.data().data(), self.data().size() ); } )
+        .def( "__bool__", &blob_msg::is_valid )
         .def( "__repr__",
               []( blob_msg const & self )
               {
@@ -671,6 +685,7 @@ PYBIND11_MODULE(NAME, m) {
             "timestamp",
             []( imu_msg const & self ) { return self.timestamp(); },
             []( imu_msg & self, dds_time time ) { self.timestamp( time ); } )
+        .def( "__bool__", &imu_msg::is_valid )
         .def( "__repr__",
               []( imu_msg const & self )
               {