startDaqV2.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_libjson
 * @copyright
 * (c) Copyright ESO 2022
 * All Rights Reserved
 * ESO (eso.org) is an Intergovernmental Organisation, and therefore special legal conditions apply.
 */
#include <daq/json/startDaqV2.hpp>
 
#include <fmt/format.h>
 
#include "internalParseUtils.hpp"
 
using Json = nlohmann::json;
using JsonPointer = nlohmann::json_pointer<Json>;
 
namespace daq::json {
 
using ErrorType = StartDaqV2SpecError;
 
/**
 * Parse the common parts of PrimaryDataSource and MetadataSource.
 *
 *     {
 *       "sourceName": <str>,
 *       "rrUri": <str>,
 *       "keywordRules": [<keywordrules]
 *     }
 */
void ParseDataSource(StartDaqV2Spec::DataSource& out,
                     Json const& json,
                     JsonPointer const& breadcrumb) {
    AssertIsObject<ErrorType>(json, breadcrumb);
    out.source_name = JsonObject<ErrorType, std::string>::Get(json, "sourceName", breadcrumb);
    out.rr_uri = JsonObject<ErrorType, std::string>::Get(json, "rrUri", breadcrumb);
    if (json.contains("initialKeywords")) {
        out.initial_keywords =
            ParseInitialKeywords(json["initialKeywords"], breadcrumb / "initialKeywords");
    }
    if (json.contains("keywordRules")) {
        out.keyword_rules = ParseKeywordRules(json["keywordRules"], breadcrumb / "keywordRules");
    }
}
 
StartDaqV2Spec ParseStartDaqV2Spec(nlohmann::json const& json) {
    try {
        StartDaqV2Spec result = {};
 
        JsonPointer breadcrumb("");
        AssertIsObject<StartDaqV2SpecError>(json, breadcrumb);
 
        if (json.contains("id")) {
            result.id = JsonObject<ErrorType, std::string>::Get(json, "id", breadcrumb, true);
        }
 
        if (json.contains("filePrefix")) {
            result.file_prefix =
                JsonObject<ErrorType, std::string>::Get(json, "filePrefix", breadcrumb, true);
        }
        if (json.contains("awaitCompletionInterval")) {
            auto value =
                JsonObject<ErrorType, double>::Get(json, "awaitCompletionInterval", breadcrumb);
            if (value < 0.0) {
                throw MakeParseException<ErrorType>(breadcrumb / "awaitCompletionInterval",
                                                    "interval must be > 0.0s");
            }
            result.await_completion_interval =
                std::chrono::duration_cast<std::chrono::milliseconds>(
                    std::chrono::duration<double>(value));
        }
 
        if (json.contains("mergeTarget")) {
            StartDaqV2Spec::MergeTarget m;
            m.source_name = JsonObject<SchemaError, std::string>::Get(
                json["mergeTarget"], "sourceName", breadcrumb / "mergeTarget");
            result.merge_target = m;
        }
        {
            // Sources
            auto [json_sources, breadcrumb_sources] = GetMember<ErrorType>(
                json, "sources", breadcrumb, [](Json const& member, JsonPointer const& breadcrumb) {
                    AssertIsArray<ErrorType>(member, breadcrumb);
                });
            auto index = 0u;
            auto found_merge_target = 0u;
            for (auto const& json_source : json_sources) {
                auto breadcrumb_source = breadcrumb_sources / index;
                auto source_type =
                    JsonObject<ErrorType, std::string>::Get(json_source, "type", breadcrumb_source);
                if (source_type == "primarySource") {
                    StartDaqV2Spec::PrimaryDataSource s;
                    ParseDataSource(s, json_source, breadcrumb_source);
                    if (result.merge_target && result.merge_target->source_name == s.source_name) {
                        found_merge_target++;
                    }
                    result.sources.emplace_back(std::move(s));
                } else if (source_type == "metadataSource") {
                    StartDaqV2Spec::MetadataSource s;
                    ParseDataSource(s, json_source, breadcrumb_source);
                    if (result.merge_target && result.merge_target->source_name == s.source_name) {
                        found_merge_target++;
                    }
                    result.sources.emplace_back(std::move(s));
                } else if (source_type == "fitsKeywords") {
                    auto kws = ParseFitsKeywordsSource(json_source, breadcrumb_source);
                    if (result.merge_target &&
                        result.merge_target->source_name == kws.source_name) {
                        // Merge target cannot be keywords
                        throw MakeParseException<ErrorType>(
                            breadcrumb / "mergeTarget" / "sourceName",
                            "Merge target `{}` refers to fitsKeywords which cannot be a merge "
                            "target",
                            kws.source_name);
                    }
                    result.sources.emplace_back(std::move(kws));
                } else if (source_type == "fitsFile") {
                    auto file = ParseFitsFileSource(json_source, breadcrumb_source);
                    if (result.merge_target &&
                        result.merge_target->source_name == file.source_name) {
                        found_merge_target++;
                    }
                    result.sources.emplace_back(std::move(file));
                } else {
                    throw MakeUnknownVariantException<ErrorType>(
                        breadcrumb_source / "type",
                        "'primarySource', 'metadataSource', 'fitsFile' or 'fitsKeywords'",
                        source_type.c_str());
                }
                index++;
            }
            if (result.merge_target) {
                if (found_merge_target == 0) {
                        throw MakeParseException<ErrorType>(
                            breadcrumb / "mergeTarget" / "sourceName",
                            "Merge target `{}` does not match any sources",
                            result.merge_target->source_name);
                }
                else if (found_merge_target > 1) {
                        throw MakeParseException<ErrorType>(
                            breadcrumb / "mergeTarget" / "sourceName",
                            "Merge target `{}` matches more than one source",
                            result.merge_target->source_name);
                }
            }
        }
        if (json.contains("receivers")) {
            result.receivers = ParseReceiverList(json["receivers"], breadcrumb / "receivers");
        }
        return result;
    } catch (std::exception const& e) {
        std::throw_with_nested(ErrorType("Schema error while parsing StartDaqV2 specification"));
    }
}
 
bool operator==(StartDaqV2Spec::PrimaryDataSource const& lhs,
                StartDaqV2Spec::PrimaryDataSource const& rhs) noexcept {
    return lhs.source_name == rhs.source_name && lhs.rr_uri == rhs.rr_uri &&
           lhs.keyword_rules == rhs.keyword_rules;
}
 
bool operator==(StartDaqV2Spec::MetadataSource const& lhs,
                StartDaqV2Spec::MetadataSource const& rhs) noexcept {
    return lhs.source_name == rhs.source_name && lhs.rr_uri == rhs.rr_uri &&
           lhs.keyword_rules == rhs.keyword_rules;
}
 
bool operator==(StartDaqV2Spec::MergeTarget const& lhs,
                StartDaqV2Spec::MergeTarget const& rhs) noexcept {
    return lhs.source_name == rhs.source_name;
}
 
bool operator==(StartDaqV2Spec const& lhs, StartDaqV2Spec const& rhs) noexcept {
    return lhs.id == rhs.id && lhs.file_prefix == rhs.file_prefix &&
           lhs.await_completion_interval == rhs.await_completion_interval &&
           lhs.merge_target == rhs.merge_target && lhs.sources == rhs.sources &&
           lhs.receivers == rhs.receivers;
}
 
// NOLINTNEXTLINE
void to_json(nlohmann::json& out, StartDaqV2Spec::DataSource const& s) {
    out = Json{{"sourceName", s.source_name}, {"rrUri", s.rr_uri}};
    if (s.initial_keywords.has_value()) {
        out["initialKeywords"] = *s.initial_keywords;
    }
    if (!s.keyword_rules.empty()) {
        out["keywordRules"] = s.keyword_rules;
    }
}
 
// NOLINTNEXTLINE
void to_json(nlohmann::json& out, StartDaqV2Spec::PrimaryDataSource const& s) {
    to_json(out, static_cast<StartDaqV2Spec::DataSource const&>(s));
    out["type"] = "primarySource";
}
 
// NOLINTNEXTLINE
void to_json(nlohmann::json& out, StartDaqV2Spec::MetadataSource const& s) {
    to_json(out, static_cast<StartDaqV2Spec::DataSource const&>(s));
    out["type"] = "metadataSource";
}
 
// NOLINTNEXTLINE
void to_json(nlohmann::json& out, StartDaqV2Spec::DataSourceTypes const& s) {
    std::visit([&](auto const& ds) { to_json(out, ds); }, s);
}
 
// NOLINTNEXTLINE
void to_json(nlohmann::json& out, StartDaqV2Spec::MergeTarget const& t) {
    out = Json{{"sourceName", t.source_name}};
}
 
// NOLINTNEXTLINE
void to_json(nlohmann::json& json, StartDaqV2Spec const& spec) {
    json = nlohmann::json{
        {"id", spec.id}, {"filePrefix", spec.file_prefix}, {"sources", spec.sources}};
    if (spec.await_completion_interval.has_value()) {
        json["awaitCompletionInterval"] = std::chrono::duration_cast<std::chrono::duration<double>>(
                                              *spec.await_completion_interval)
                                              .count();
    }
    if (spec.merge_target.has_value()) {
        json["mergeTarget"] = *spec.merge_target;
    }
    json["receivers"] = spec.receivers;
}
 
}  // namespace daq::json