Creating a Custom DDT Server

This example shows how to implement a custom DDT Server. A DDT Server allows the reception of shared memory queue topics, and the publication of a subset of the data to one or many DDT datastreams.

Provided Example

The provided example implements a DDT Server for one of the example topics. It is not intended to be used in an actual implementation, but rather as a template for an actual DDT Server.

Note

For simplicity reasons, the example uses the file-based implementation of OLDB, Persistent and Runtime Configuration Repositories, as well as file-based service discovery. This means that the underlying format of configuration and data points is different than the one used when the above mentioned services are used with the standard backends such as CII configuration service, CII OLDB, etc.

Source Code Location

The example source code can be found in the following sub-directory of the rtctk project:

_examples/exampleDdtServer

Modules

The provided example is composed by the following waf modules:

  • app - The application including default configuration

  • pixelsShmPub - An example IPCQ shared memory publisher

  • scripts - Helper scripts for deployment and control

Dependencies

The provided example component depends on the following modules:

  • rtctk._examples.exampleTopics - example shared memory topic definitions

  • rtctk.componentFramework.rtcComponent - basic RTC Component functionality

  • rtctk.reusableComponents.ddtServer.lib - ddt server infrastructure

  • ddt.datatransfer.datatransferlib - DDT transfer library

In addition, the following applications are used:

  • rtctkClient - to send basic commands to the application

  • ExamplePixelsShmPub - an example SHM publisher that publishes images as arrays.

Running the Example DDT Server

The exampleDdtServer can be run in isolation or as part of an SRTC system. It reads data from a shared memory queue and uses the DDT Transfer Library to send pixels to a datastream. This can be achieved using a Shared Memory Publisher (shmPub). The queue name and the subsample factor need to be provided through the Runtime Configuration Repository.

To make the example as simple as possible a script rtctkExampleDdtServer.sh is provided to start an example shared memory publisher and an example DDT Server.

rtctkExampleDdtServer.sh

After installing the RTC Tk, run the example using the following sequence of commands:

# Deploy and start the example applications
rtctkExampleDdtServer.sh provision
rtctkExampleDdtServer.sh deploy

# Open a second terminal and use the client to step through the life-cycle of the
# respective component instance
rtctkExampleDdtServer.sh send Init
rtctkExampleDdtServer.sh send Enable
rtctkExampleDdtServer.sh send Run

# all commands until this point can also be run using
# rtctkExampleDdtServer.sh run

rtctkExampleDdtServer.sh send Idle
rtctkExampleDdtServer.sh send Disable
rtctkExampleDdtServer.sh send Reset

# Gracefully terminate the applcations and clean-up
rtctkExampleDdtServer.sh undeploy
rtctkExampleDdtServer.sh unprovision

The data read from the topic will be sent to four datastreams. While the server is running, the user may visualize the datastream with the DDT Viewer, a tool part of the DDT library.

The three two-dimensional streams can be visualized with the default ddtViewer application:

# Start ddtViewer GUI and select desired datastream
ddtViewer -l "zpb.rr://127.0.0.1:5001"

The one-dimensional stream can be visualized with the pyDdtPlotsViewer application.

Please refer to the DDT User Manual for a detailed description on how to use the DDT Viewer applications.

Development Guide

This section explains how to create a simple DDT Server from scratch. For a more detailed description, see the DDT Server section.

Customisation

This section explains how to customise a DDT Server on the source code level.

Header Files and Namespaces

To create a custom DDT Server component the following include files and namespaces should be used:

#include <rtctk/exampleTopics/topics.hpp>

#include <rtctk/ddtServer/businessLogic.hpp>
#include <rtctk/ddtServer/ddtPublisherImage1d.hpp>
#include <rtctk/ddtServer/ddtPublisherImage2d.hpp>
#include <rtctk/ddtServer/ipcqDdtForwarder.hpp>
#include <rtctk/componentFramework/rtcComponentMain.hpp>

using namespace rtctk::exampleTopic;
using namespace rtctk::ddtServer;
using namespace rtctk::componentFramework;

Please note that the topic definition header in topics.hpp and the corresponding namespace exampleTopic is project specific, meaning these files should be replaced by the shared memory topic definition files specific to the RTC.

The remaining header files and namespaces are required to make use of the RTC TK DDT Server infrastructure.

DDT Forwarders and Publishers

A RTC Tk DDT Server component may host one or many DDT forwarders. It is up to the instrument RTC developer to decide which and how many forwarders a specific DDT server shall have. Note: currently the RTC Toolkit only provides a single forwarder that ingests IPCQ Telemetry.

Similarly, a DDT Forwarder unit hosts one or many DDT publishers. It is up to the instrument RTC developer to decide which DDT publishers a specific DDT Forwarder shall have. Currently the RTC Toolkit provides publishers for 1-dimensional and 2-dimensional arrays of pixels.

The compile-time configuration of the forwarder requires the provision of a trait class which defines specific types, attributes and extraction methods. In the example it is implemented as follows:

// configuration struct for one DDT stream processor
struct IpcqDdtForwarderInfo {

  // identifier of the DDT processor unit
  constexpr static std::string_view ID = "pixel_subsampled";

  // shared memory topic type
  using Topic = PixelSuperTopic;

  // settings for individual DDT stream
  struct RawPixelFrameDdtStreamInfo {
    constexpr static std::string_view ID = "raw_pixel_frame";

    using PixelType = uint32_t;
    constexpr static unsigned WIDTH = 192;
    constexpr static unsigned HEIGHT = 192;

    static auto Extract(const Topic& sample) {
        return std::make_tuple(sample.sample_id, gsl::span(sample.data_1));
    }
  };

  // settings for individual DDT stream
  struct RawPupilQuadrantDtStreamInfo {
    constexpr static std::string_view ID = "raw_pupil_quadrant";

    using PixelType = uint32_t;
    constexpr static unsigned WIDTH = 96;
    constexpr static unsigned HEIGHT = 96;

    static auto Extract(const Topic& sample) {
        return std::make_tuple(sample.sample_id, gsl::span(sample.data_2));
    }
  };

  // settings for individual DDT stream
  struct ExtractedPupilsDdtStreamInfo {
    constexpr static std::string_view ID = "extracted_pupils";

    using PixelType = float;
    constexpr static unsigned WIDTH = 92;
    constexpr static unsigned HEIGHT = 92;

    static auto Extract(const Topic& sample) {
        return std::make_tuple(sample.sample_id, gsl::span(sample.data_3));
    }
  };

  // settings for individual DDT stream
  struct RawPixelFrameDdtStreamInfo1d {
      constexpr static std::string_view ID = "raw_pixel_frame_1d";

      using PixelType = uint32_t;
      constexpr static unsigned WIDTH = 192 * 192;

      static auto Extract(const Topic& sample) {
          return std::make_tuple(sample.sample_id, gsl::span(sample.data_1));
      }
  };

  // aggregated settings for all DDT publishers
  using DdtPublisherTupleType = std::tuple<DdtPublisherImage2d<RawPixelFrameDdtStreamInfo>,
                                           DdtPublisherImage2d<RawPupilQuadrantDtStreamInfo>,
                                           DdtPublisherImage2d<ExtractedPupilsDdtStreamInfo>,
                                           DdtPublisherImage1d<RawPixelFrameDdtStreamInfo1d>
                                          >;
};

The struct IPCQDdtForwarderInfo defines a new DDT Forwarder named pixel_subsampled that reads data from shared memory topic PixelSuperTopic.

The Forwarder hosts multiple DDT streams: three two-dimensional streams RawPixelFrameDdtStreamInfo, RawPupilQuadrantDdtStreamInfo, ExtractedPupilsDdtStreamInfo and a one-dimensional stream RawPixelFrameDdtStreamInfo1d. Each stream has different attributes, i.e. ID (DDT stream name), PixelType, WIDTH and HEIGHT, as well as an Extract method that is used to retrieve data fram shared memory samples and return them as a tuple of sample_id and a span.

During its construction, the forwarder will construct a tuple of three DdtPublisherImage2d and one DdtPublisherImage1d where each is configured with a different stream info.

Business Logic Factory

The example below shows how to create a new DDT Server with a single DDT Forwarder:

// main entry point for user code
void RtcComponentMain(const Args& args) {

    // will be invoked at component startup
    auto bl_factory = [](const std::string& name, ServiceContainer& services) {

        // inner factory will be invoked in ActivityInitialising
        auto fw_factory = [](const std::string& name, ServiceContainer& services) {

            DdtForwarderListType forwarders;

            AddDdtForwarder<IpcqDdtForwarder<IpcqDdtForwarderInfo>>(forwarders, name, services);

            return forwarders;
        };

        // returns the biz logic object
        return std::make_unique<BusinessLogic>(name, services, std::move(fw_factory));
    };

    // invoke component runner function
    RunAsRtcComponent<BusinessLogic>(args, std::move(bl_factory));
}

To create a custom DDT Server, the toolkit-provided class rtctk::ddtServer::BusinessLogic is used. The class manages multiple user-defined forwarders that are provided via a nested factory method.

The user-provided factory methods bl_factory with inner factory fw_factory are used to define the DDT Forwarder units. The factories are then passed to component runner for deferred invocation.

The method AddDdtForwarder is used to instantiate new IpcqDdtForwarders and add them to a list of forwarders. The method takes the configured forwarder type as a template argument and also accepts common arguments component_name and a reference to the Service Container.

Configuration

After the component customisation, the run-time configuration needs to be prepared as follows:

static:
  forwarders:
    pixel_subsampled:
      shm_queue_name: !cfg.type:string examplePixelsQueue
      thread_policies:
        cpu_affinity: !cfg.type:string 0
dynamic:
  forwarders:
    pixel_subsampled:
      subsample_factor: !cfg.type:int64 1

Note that each DDT forwarder or DDT publisher may have different configuration parameters.

More information on individual configuration parameters can be found in the DDT Server section.