ddtPublisherImage1d.hpp

Go to the documentation of this file.

 
#ifndef DDT_PUBLISHER_IMAGE1D_HPP
#define DDT_PUBLISHER_IMAGE1D_HPP
 
#include <rtctk/componentFramework/exceptions.hpp>
#include <rtctk/componentFramework/logger.hpp>
#include <rtctk/componentFramework/runtimeRepoIf.hpp>
#include <rtctk/componentFramework/serviceContainer.hpp>
 
#include <ddt/ddtDataPublisher.hpp>
#include <ddt/ddtEncDecBinaryxD.hpp>
#include <ddt/ddtLogger.hpp>
 
#include <boost/endian.hpp>
#include <fmt/format.h>
 
namespace rtctk::ddtServer {
 
template <typename SI>
class DdtPublisherImage1d {
public:
    using StreamInfo = SI;
 
    using DataPointPath = componentFramework::DataPointPath;
 
    static_assert(SI::ID.find_first_not_of("abcdefghijklmnopqrstuvwxyz_0123456789") ==
                      std::string_view::npos,
                  "DDT Publisher ID contains illegal characters!");
 
    DdtPublisherImage1d(const std::string& db_prefix, ServiceContainer& services)
        : m_logger(componentFramework::GetLogger("rtctk"))
        , m_rtr(services.Get<RuntimeRepoIf>())
        , m_rtr_prefix_static(GetStaticRtrPrefix(db_prefix))
        , m_rtr_prefix_dynamic(GetDynamicRtrPrefix(db_prefix))
        , m_enabled(true)
        , m_broker_endpoint("zpb.rr://127.0.0.1:5001") {
        LOG4CPLUS_INFO(m_logger, fmt::format("Creating DdtPublisherImage1d '{}'", SI::ID));
 
        // publisher configuration
 
        auto broker_endpoint_path = DataPointPath(m_rtr_prefix_static + "/broker_endpoint");
        if (m_rtr.DataPointExists(broker_endpoint_path)) {
            m_broker_endpoint = m_rtr.GetDataPoint<std::string>(broker_endpoint_path);
        }
 
        auto enabled_path = DataPointPath(m_rtr_prefix_dynamic + "/enabled");
        if (m_rtr.DataPointExists(enabled_path)) {
            m_enabled = m_rtr.GetDataPoint<bool>(enabled_path);
        }
 
        // metadata preparation
 
        m_meta_data.meta_data_base.bytes_per_pixel = sizeof(typename SI::PixelType);
        m_meta_data.meta_data_base.number_dimensions = 1;
        m_meta_data.meta_data_base.complete_flag = true;
        m_meta_data.meta_data_base.last_segment = true;
        /*
         * Important note: Endiannes of the source is assumed to be the same as the one of the
         * host architecture. This has been set by design, that the low-level devices has to
         * translate (if necessary) to the RTCTK's endiannes. RTCTK assumes the data comes in the
         * same endiannes as the machine running this process.
         */
        m_meta_data.meta_data_base.byte_order_little_endian =
            static_cast<bool>(boost::endian::order::native);
 
        m_meta_data.meta_data_base.description = SI::ID;
 
        if (typeid(typename SI::PixelType) == typeid(uint8_t)) {
            m_meta_data.meta_data_base.data_type = ddt::UINT8;
        } else if (typeid(typename SI::PixelType) == typeid(int16_t)) {
            m_meta_data.meta_data_base.data_type = ddt::SINT16;
        } else if (typeid(typename SI::PixelType) == typeid(int32_t)) {
            m_meta_data.meta_data_base.data_type = ddt::SINT32;
        } else if (typeid(typename SI::PixelType) == typeid(float)) {
            m_meta_data.meta_data_base.data_type = ddt::FLOAT32;
        } else if (typeid(typename SI::PixelType) == typeid(double)) {
            m_meta_data.meta_data_base.data_type = ddt::FLOAT64;
        } else if (typeid(typename SI::PixelType) == typeid(uint16_t)) {
            m_meta_data.meta_data_base.data_type = ddt::UINT16;
        } else if (typeid(typename SI::PixelType) == typeid(uint32_t)) {
            m_meta_data.meta_data_base.data_type = ddt::UINT32;
        } else {
            CII_THROW(UnsupportedTypeException, typeid(typename SI::PixelType));
        }
 
        m_meta_data.array_dimensions = std::to_string(SI::WIDTH);
 
        // publisher instantiation and registration
 
        m_ddt_publisher = std::make_unique<ddt::DdtDataPublisher>(m_logger);
        m_ddt_publisher->set_topic_id(m_enc_dec.get_topic_id());
        int bytes_per_sample = SI::WIDTH * sizeof(typename SI::PixelType);
        int samples_in_buffer = 10;
        m_ddt_publisher->SetBufferSize(bytes_per_sample, samples_in_buffer);
 
        auto ret =
            m_ddt_publisher->RegisterPublisher(m_broker_endpoint, std::string(SI::ID), false);
        if (ret < 0) {
            CII_THROW(RtctkException,
                      fmt::format("Failed to register DDT Publisher: {} against DDT Broker: {} "
                                  "Check Broker URI, and ensure that Broker process is running.",
                                  SI::ID,
                                  m_broker_endpoint));
        }
    }
 
    void Update() {
        LOG4CPLUS_INFO(m_logger, fmt::format("Updating DdtPublisherImage1d::{}", SI::ID));
 
        auto enabled_path = DataPointPath(m_rtr_prefix_dynamic + "/enabled");
        if (m_rtr.DataPointExists(enabled_path)) {
            m_enabled = m_rtr.GetDataPoint<bool>(enabled_path);
        }
    }
 
    bool IsEnabled() {
        return m_enabled;
    }
 
    void Publish(uint32_t sample_id, const uint8_t* data_ptr, uint32_t elements) {
        m_meta_data.meta_data_base.utc_timestamp = m_enc_dec.get_utc_timestamp();
        m_enc_dec.Encode(m_meta_data);
        std::vector<uint8_t> metadata = m_enc_dec.get_meta_data();
 
        auto data_bytes = elements * sizeof(typename SI::PixelType);
        m_ddt_publisher->WriteData(
            sample_id, data_ptr, data_bytes, metadata.data(), metadata.size());
        m_ddt_publisher->PublishData();
    }
 
private:
    std::string GetStaticRtrPrefix(const std::string& db_prefix) {
        std::string temp = db_prefix;
        temp.insert(temp.find("/"), "/static");
        return "/" + temp + "/" + std::string(SI::ID);
    }
 
    std::string GetDynamicRtrPrefix(const std::string& db_prefix) {
        std::string temp = db_prefix;
        temp.insert(temp.find("/"), "/dynamic");
        return "/" + temp + "/" + std::string(SI::ID);
    }
 
    log4cplus::Logger& m_logger;
 
    RuntimeRepoIf& m_rtr;
 
    std::string m_rtr_prefix_static;
    std::string m_rtr_prefix_dynamic;
 
    bool m_enabled;
 
    std::string m_broker_endpoint;
 
    MetaDataElementsBinaryxD m_meta_data;
 
    DdtEncDecBinaryxD m_enc_dec;
 
    std::unique_ptr<ddt::DdtDataPublisher> m_ddt_publisher;
};
 
}  // namespace rtctk::ddtServer
 
#endif  // DDT_PUBLISHER_IMAGE1D_HPP