CSV Exporter

Description

Example processor which shows how to implement a adtfdat_processing::SingleStreamProcessor to process a DDL based stream and export the content to *.csv file.

Usage

adtfdat_tool --plugin csv_exporter.adtffileplugin --export <adtfdat> --stream <name> --processorid csv --output <csv> [--property decimal_places=<value> --property separator=<value> --property timestamp_base_ns=<value> --property timestamp_decimals=<value> --property columns=<value>]

Source

 
#pragma once
 
#include <adtfdat_processing/processor.h>
#include <memory>
#include <fstream>
#include <ddl/codec/codec_factory.h>
 
namespace {
    // Helper for enum to string decoding of a specific type from DDL.
    // You will need one instance for each unique CodecTypeInfo with enum representation.
    class EnumDecoder
    {
    public:
        // Safe base type for everything legally enum-alike.
        using index_t = uint64_t;
        // Container type for strings, chosen based on what we need to interface with.
        using value_t = std::string;
 
        EnumDecoder(const ::ddl::codec::CodecTypeInfo* type_info) : _invalid(), _table(generate(type_info))
        {
        }
 
        EnumDecoder(const EnumDecoder&) = delete;
        EnumDecoder(EnumDecoder&&) noexcept = delete;
 
        template<class N>
        const value_t& lookup(const N& value) const noexcept
        {
            if (auto it = _table.find(static_cast<index_t>(value)); it != _table.end())
            {
                return it->second;
            }
            return _invalid;
        }
 
    private:
        using lookup_t = std::unordered_map<index_t, value_t>;
        static lookup_t generate(const ::ddl::codec::CodecTypeInfo* type_info)
        {
            lookup_t tmp;
            for (const auto& name : type_info->getEnumElementNames())
            {
                auto value = type_info->getEnumElementValue(name);
                tmp.try_emplace(value.asUInt64(), name);
            }
            return tmp;
        }
        const value_t _invalid;
        const lookup_t _table;
    };
}
 
class CsvExporter : public adtfdat_processing::SingleStreamProcessor
{
public:
    CsvExporter();
    ~CsvExporter();
 
    std::string getProcessorIdentifier() const override
    {
        return "csv";
    }
 
    bool isCompatible(const adtf_file::Stream& stream) const override;
    void open(const adtf_file::Stream& stream, const std::string& destination_url) override;
    void process(const adtf_file::FileItem& item) override;
 
private:
    void process(const std::shared_ptr<const adtf_file::WriteSample>& sample);
    void rebuildIndex(const ddl::codec::CodecFactory& codec_factory, bool data_is_serialized);
 
    std::ofstream _csv_file;
    std::string _separator;
    uint64_t _value_precision = 4;
    uint64_t _timestamp_precision = 0;
    int64_t _timestamp_divisor = 1;
 
    std::vector<std::string> _columns;
 
    // Enum decoding is possible either with a full Codec and CodecIndex with the a_util::variant::Variant API (not recommended), or manually via type reflection.
    // With LeafCodecIndex, we need an additional helper which we can construct from CodecTypeInfo.
    // In return, LeafCodecIndex is also independet of CodecFactory once created.
    struct ElementContext {
        ddl::codec::LeafCodecIndex value_decoder;
        std::shared_ptr<EnumDecoder> enum_decoder;
    };
    // List of factory indepdent indizes for the currently active stream type.
    std::vector<ElementContext> _indices;
};

 
#include "csv_exporter.h"
#include <adtf_file/stream_type.h>
#include <adtf_file/sample.h>
#include <iomanip>
#include <string>
#include <string_view>
#include <variant>
#include <adtfdat_processing/ddl_helpers.h>
 
static adtf_file::PluginInitializer initializer([] {
    adtf_file::getObjects().push_back(
        std::make_shared<adtfdat_processing::ProcessorFactoryImplementation<CsvExporter>>());
});
 
static std::vector<std::string> explode(std::string_view text, std::string_view separator)
{
    std::vector<std::string> result;
    while (!text.empty())
    {
        auto pos = text.find(separator);
        if (pos != text.npos)
        {
            auto prefix = text.substr(0, pos);
            if (!prefix.empty())
            {
                result.emplace_back(prefix);
            }
            text = text.substr(pos + separator.size());
        }
        else
        {
            result.emplace_back(text);
            text = {};
        }
    }
    return result;
}
 
CsvExporter::CsvExporter()
{
    setConfiguration({
                         {"decimal_places", {"4", "uint32"}},
                         {"separator", {";", "string"}},
                         {"timestamp_base_ns", {"1", "uint64"}},
                         {"timestamp_decimals", {"0", "uint32"}},
                         {"columns", {"", "string"}},
                     });
}
 
CsvExporter::~CsvExporter()
{
}
 
bool CsvExporter::isCompatible(const adtf_file::Stream& stream) const
{
    try
    {
        // The configuration isn't set yet a this point, so we can at best check for DDL based streamtype.
        // Validating columns can't be done until later.
        adtfdat_processing::createDDLCodecFactoryFromStreamType(stream.initial_type);
        return true;
    }
    catch (...)
    {
        return false;
    }
}
 
void CsvExporter::open(const adtf_file::Stream& stream, const std::string& destination_url)
{
    _csv_file.exceptions(std::ofstream::failbit | std::ofstream::badbit);
 
    try
    {
        _csv_file.open(destination_url);
    }
    catch (...)
    {
        std::throw_with_nested(std::runtime_error("unable to open output csv file '" +
                                                  destination_url + "' for stream " + stream.name));
    }
 
    const auto config = getConfiguration();
    _separator = adtf_file::getPropertyValue<std::string>(config, "separator");
    _value_precision = adtf_file::getPropertyValue<uint64_t>(config, "decimal_places");
    _timestamp_precision = adtf_file::getPropertyValue<uint64_t>(config, "timestamp_decimals");
    _timestamp_divisor = adtf_file::getPropertyValue<int64_t>(config, "timestamp_base_ns");
    auto fields = adtf_file::getPropertyValue<std::string>(config, "columns");
    _columns = explode(fields, _separator);
 
    // Use of LeafCodecIndex means we can safely discard the factory after index creation.
    // If you were to use CodecIndex with Codec instead, you would need to persist the factory as a member.
    ddl::codec::CodecFactory codec_factory;
    bool data_is_serialized = false;
 
    // Initial stream type is (usually) not repeated as a stream item - handle it now.
    std::tie(codec_factory, data_is_serialized) = adtfdat_processing::createDDLCodecFactoryFromStreamType(stream.initial_type);
 
    if (_columns.empty()) {
        // If we got none, then we just take all of them instead
        adtfdat_processing::for_each_leaf_element(
            codec_factory.getElements(),
            [&](const decltype(codec_factory)::Element& element)
            {
                _columns.push_back(element.getFullName());
            }
        );
    }
 
    rebuildIndex(codec_factory, data_is_serialized);
 
    _csv_file << "timestamp";
 
    for (const auto& column : _columns)
    {
        _csv_file << config.at("separator").value << column;
    }
 
    _csv_file << "\n";
}
 
void CsvExporter::process(const adtf_file::FileItem& item)
{
    if (auto sample = std::dynamic_pointer_cast<const adtf_file::WriteSample>(item.stream_item))
    {
        process(sample);
    }
    else if (auto stream_type = std::dynamic_pointer_cast<const adtf_file::StreamType>(item.stream_item))
    {
        // Stream types may change arbitrarily often.
        // Usually there should be at least one stream item other than a StreamType in between, but that is not guaranteed.
        // Don't expect followup stream types to be compatible with the initial stream type.
 
        // New streamtype requires a new codec_factory, and even the representation may have changed.
        ddl::codec::CodecFactory codec_factory;
        bool data_is_serialized = false;
 
        std::tie(codec_factory, data_is_serialized) = adtfdat_processing::createDDLCodecFactoryFromStreamType(stream_type);
        rebuildIndex(codec_factory, data_is_serialized);
    }
    // We don't care about trigger type stream items in this context, CSV has no such thing.
}
 
void CsvExporter::process(const std::shared_ptr<const adtf_file::WriteSample>& sample)
{
    using namespace a_util;
 
    auto buffer = sample->beginBufferRead();
 
    if (_timestamp_precision)
    {
        _csv_file << std::fixed << std::setprecision(_timestamp_precision) << (sample->getTimeStamp().count() / (double)_timestamp_divisor);
    }
    else
    {
        _csv_file << (sample->getTimeStamp().count() / _timestamp_divisor);
    }
 
    _csv_file << std::fixed << std::setprecision(_value_precision);
 
    for (const auto& index : _indices)
    {
 
        _csv_file << _separator;
 
        if (index.enum_decoder) {
            auto value = ddl::codec::LeafValueGetter<EnumDecoder::index_t>::getValue(buffer.first, buffer.second, index.value_decoder.getLayout());
            const auto& text = index.enum_decoder->lookup(value);
            // When dealing with "enums", always expect that only parts of the value range are mapped to literals.
            // Especially when dealing with AUTOSAR modeled signals, usually only a few selected error codes have a text representation.
            if (!text.empty()) {
                _csv_file << text;
                continue;
            }
        }
 
        // Deal with all primitive types which may have been used in DDL.
        using variant_t = std::variant<bool, int8_t, uint8_t, int16_t, uint16_t, int32_t, uint32_t, int64_t, uint64_t, float, double>;
        const auto value = ddl::codec::LeafValueGetter<variant_t>::getValue(buffer.first, buffer.second, index.value_decoder.getLayout());
        std::visit(
            [this](auto&& value) {
                // Ensure that we don't end up with character encodings.
                if constexpr (std::is_integral_v<std::remove_reference_t<decltype(value)>>)
                {
                    if constexpr (std::is_signed_v<std::remove_reference_t<decltype(value)>>)
                    {
                        _csv_file << static_cast<int64_t>(value);
                    }
                    else
                    {
                        _csv_file << static_cast<uint64_t>(value);
                    }
                }
                else
                {
                    _csv_file << static_cast<double>(value);
                }
            },
            value
        );
    }
    _csv_file << "\n";
}
 
void CsvExporter::rebuildIndex(const ddl::codec::CodecFactory& codec_factory, bool data_is_serialized)
{
    _indices.clear();
 
    // Map keys are only valid in the scope of codec_factory, values can be used safely.
    std::unordered_map<const ddl::codec::CodecTypeInfo*, std::shared_ptr<EnumDecoder>> enum_decoders;
 
    for (const auto& column : _columns)
    {
        // CodecIndex can be used either with a Codec created from CodecFactory - or usually preferably converted into a LeafCodecIndex.
        // LeafCodecIndex eliminates the dynamic layout and type reflection support from CodecIndex, and is in return much faster and smaller.
        // We only support static layout DDL variants in this example, so LeafCodecIndex it is.
        ddl::codec::CodecIndex index;
 
        try
        {
            auto element = codec_factory.getElement(column);
            if (element.hasChildren())
            {
                throw std::runtime_error("Expected scalar, got complex type instead");
            }
            index = element.getIndex();
        }
        catch (...)
        {
            std::throw_with_nested(std::runtime_error("column name '" +
                column + "' is not a valid member for the current stream type"));
        }
 
        // Since LeafCodecIndex drops type infos etc., it no longer includes enum-support of its own.
        // To make up for that, we quickly construct a helper from the available type reflection.
        std::shared_ptr<EnumDecoder> enum_decoder;
 
        // Check if enum decoding will be required.
        auto type_info = index.getLayout().type_info;
        if (type_info->isEnum())
        {
            // One EnumDecoder per unique CodecTypeInfo only - expect enum like types to be reused a lot.
            auto it = enum_decoders.find(type_info);
            if (it == enum_decoders.end())
            {
                it = enum_decoders
                    .try_emplace(type_info, std::make_shared<EnumDecoder>(type_info))
                    .first;
            }
            enum_decoder = it->second;
        }
 
        _indices.push_back(ElementContext{
            // Be careful that LeafCodecIndex needs to be constructed from the correct representation.
            // Same applies if you were to use CodecFactory::makeStaticDecoderFor - you have to respect the stream type.
            // Neither CodecFactory nor CodecIndex are bound to any representation.
            ddl::codec::LeafCodecIndex(index, data_is_serialized ? ddl::DataRepresentation::serialized : ddl::DataRepresentation::deserialized),
            std::move(enum_decoder)
        });
    }
}