Creating a Custom Telemetry Recorder

This example shows how to implement a Telemetry Recorder. A Telemetry Recorder allows the recording of shared memory queues, events and datapoints, which are recorded by recording units. These recording units have to be defined to create a Telemetry Recorder component.

Provided Example

The provided example implements a Telemetry Recorder for one of the example topics as well as recording events. It is not intended to be used in an actual implementation, but rather as a template for an actual Telemetry Recorder.

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/exampleTelRec

Modules

The provided example is composed into the following waf modules:

app - The application including default configuration
scripts - Helper scripts for deployment and control

Dependencies

The provided example component depends on the following modules:

rtctk.componentFramework.rtcComponent - basic RTC Component functionality
rtctk.componentFramework.services.dataRecording - common recording infrastructure
rtctk.reusableComponents.telRec.lib - business logic class with implementation

In addition, the following applications are used:

rtctkClient - to send basic commands to the application
ExampleShmPub - an example SHM publisher from the DataTask example that provides example data

Running the Example Telemetry Recorder

The exampleTelRec can be run in isolation or as part of an SRTC system. To record telemetry data, a shared memory queue needs to be created. This can be achieved using a Shared Memory Publisher (shmPub) or a Telemetry Subscriber (telSub). The queue name and the destination file name need to be provided through the Runtime Configuration Repository.

To make the example as simple as possible, a script rtctkExampleTelRec.sh is provided to start an example shared memory publisher and an example recorder.

rtctkExampleTelRec.sh

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

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

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

# All commands until this point can also be run using
# rtctkExampleTelRec.sh run


rtctkExampleTelRec.sh send Idle
rtctkExampleTelRec.sh send Disable
rtctkExampleTelRec.sh send Reset

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

The recorded data will be stored in the $DATAROOT directory. The sub-directory can be customised by changing the session_id setting in the Runtime Configuration Repository.

Development Guide

This section explains how to create a simple Telemetry Recorder from scratch. For more detailed description see the Telemetry Recorder section.

Customisation

This section explains how to customise a Telemetry Recorder on the source code level.

Recording Units

A Telemetry Recorder hosts one or many recording units. The RTC Toolkit provides recording units for IPCQ, Event, and DataPoint recording.

Each recording unit has it’s own state and might wait for a condition before starting. E.g. the IPCQ recording unit supports waiting for a specific sample ID before actually writing the data. It is also possible to follow another recording unit (leader), so that the recording is done while the other recording unit is recording.

It is up to the instrument RTC developer to decide which recording units a specific recorder shall have.

Business Logic Factory

The example below shows how to create a new Telemetry Recorder with two recording units:

IpcqRecordingUnit for shared memory recording with settings from IpcqRecordingInfo
EventRecordingUnit to record any events on a configurable topic

// 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 ru_factory = [](const std::string& name, ServiceContainer& services) {

            RecUnitListType rec_units;

            AddRecUnit<IpcqRecordingUnit<IpcqRecordingInfo>>(
                    rec_units, name, "unit_1", services);

            AddRecUnit<EventRecordingUnit<>>(
                    rec_units, name, "unit_2", services);

            return rec_units;
        };

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

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

To create a custom Telemetry Recorder, the toolkit-provided BusinessLogic class in rtctk::telemetryRecorder::BusinessLogic shall be used. The class manages multiple user-defined Recording Units that are provided via a nested factory method.

The user-provided factory method bl_factory with inner factory ru_factory is used to define the recording units. The factory is then passed to the component runner for deferred invocation.

The method AddRecUnit is used to instantiate new recording units and add them to the rec_units list. The method takes the recording unit type as a template argument and accepts different arguments depending on the selected recording unit type.

Every unit type accepts common arguments component_name, unit_id and a reference to the Service Container. Component name and unit id (e.g. ipcq_unit) are used to retrieve the unit configuration from the Runtime Configuration Repository and to publish unit state and additional monitoring information to the Online Database. Additionally, generated files are named according to the units that created them. The service handle is required so that units can access services such as databases, event channel, etc.

RecordingInfo

Compile-time configuration of the IpcqRecordingUnit requires provision of a RecordingInfo trait class. This class defines the FITS binary table layout and enables the recording unit to extract and write data accordingly. In the example, it is implemented as follows:

struct IpcqRecordingInfo {
    // mandatory member to specify shm topic type
    using Topic = ScaoLoopTopic;

    // typedefs are for convenience, so that we have short type names
    using Slopes = gsl::span<const float>; // 2nd tparam defaults to gsl::dynamic_extent
    using Intensities = gsl::span<const float, N_SUBAPS>;
    using Commands = gsl::span<const float, N_COMMANDS>;

    // mandatory member to specify the Writer type
    using Recorder = FitsRecorder<uint32_t, Slopes, Intensities, Commands>;

    // mandatory member to specify column titles and units
    static constexpr Recorder::ColumnDescription COLUMNS = Recorder::ColumnDescription {
        { {"sample_id"sv, ""sv},
          {"slopes"sv, ""sv},
          {"intensities"sv, ""sv},
          {"commands"sv, ""sv} }
    };

    // mandatory method to extract data from shm into tuple
    static Recorder::TupleType AsTuple(const Topic& data) {
        // TupleType is defined automatically as sd::tuple<"Recorder template arguments">
        // in this case specifically: std::tuple<uint32_t, Slopes, Intensities, Commands>

        // static asserts for sanity checking (can be omitted)
        static_assert(IS_DYNAMIC_SPAN_TYPE<decltype(Slopes(data.wfs.slopes.data(), 10))>);
        static_assert(IS_STATIC_SPAN_TYPE<decltype(Intensities(data.wfs.intensities))>);
        static_assert(IS_STATIC_SPAN_TYPE<decltype(Commands(data.commands))>);

        // the actual call to extract data tuple containing sample_id and spans
        return std::make_tuple(data.sample_id,
                               Slopes(data.wfs.slopes.data(), 10), // only take first 10
                               Intensities(data.wfs.intensities),
                               Commands(data.commands));
    }
};

Mandatory parts that must be defined are:

Topic: the user-defined shared memory topic type (usually a struct) that is being recorded. The Topic is required to have a member sample_id of type uint32_t.
Recorder: the recorder type (usually a templated FitsRecorder class). The template arguments describe the input data type of each FITS column.
COLUMNS: a list of column-name and column-unit pairs of the FITS binary table. (names are provided as string_views, thus the preceding string_view_literal sv)
AsTuple: A function that extracts data from the shared memory struct and returns a std::tuple with the same template arguments as the Recorder.

In this example, the Slopes, Intensities, and Commands aliases are used for convenience to make the other type definitions simpler.

In the function AsTuple, the larger data members of the SHM sample are converted to span types; this prevents unnecessary data copies and enables faster writing. When using static span types the compiler can do more optimisation but the size of the spans must be known at compile-time. When using dynamic span types, setting the span size can be deferred to runtime; this is useful if the shared memory topic contains data arrays with the notion of maximum size and currently used size. In this case, the dynamic span can be set to the currently used size to record only valid data ranges.

Obviously, the provided column information (types and order) must be consistent between Recorder, COLUMNS and when creating the tuple in method AsTuple.

Configuration

After component customisation, the persistent configuration needs to be prepared as follows:

static:
  rec_units:
    unit_1:
      shm_queue_name: !cfg.type:string 'exampleTelRecQueue'
      cpu_affinity: !cfg.type:int32 '0'
    unit_2:
      leader_list: !cfg.type:vector_string [unit_1]
      topic_name: !cfg.type:string 'configuration_topic'
dynamic:
  session_id: !cfg.type:string 'session_1'
  rec_units:
    unit_1:
      subsample_factor: !cfg.type:int64 '10'
      telemetry_subset: !cfg.type:vector_string [intensities, slopes]
      start_at_sample_id: !cfg.type:int64 '0'
      stop_after_num_samples: !cfg.type:int64 '10'

Note that each recording unit type may have different configuration parameters. Also note that the recording session_id is considered a dynamic configuration parameter.

More information on individual configuration parameters can be found in the Telemetry Recorder section.