makeDpSpec.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_libdaq
 * @copyright
 * (c) Copyright ESO 2022
 * All Rights Reserved
 * ESO (eso.org) is an Intergovernmental Organisation, and therefore special legal conditions apply.
 */
#include <daq/makeDpSpec.hpp>
 
#include <boost/exception/enable_current_exception.hpp>
#include <fmt/format.h>
#include <log4cplus/loggingmacros.h>
 
#include <daq/fits/json.hpp>
#include <daq/json/dpSpec.hpp>
 
namespace daq {
 
std::string MakeOcmName(DaqContext const& ctx) {
    return fmt::format("@{}", ctx.process_name);
}
 
/**
 * Make OCM keywords source.
 */
json::FitsKeywordsSource MakeOcmKeywords(DaqContext const& ctx, log4cplus::Logger& logger) {
    json::FitsKeywordsSource kws;
    kws.keywords = ctx.keywords;
    kws.source_name = MakeOcmName(ctx);
    return kws;
}
 
namespace v1 {
 
json::DpSpec MakeDataProductSpecification(DaqContext const& ctx, log4cplus::Logger& logger) {
    json::DpSpec dp_spec;
    dp_spec.id = ctx.id;
    dp_spec.target.file_id = ctx.file_id;
    dp_spec.target.file_prefix = ctx.dp_name_prefix;
 
    // Add OCM keywords
    if (!ctx.keywords.empty()) {
        dp_spec.sources.emplace_back(MakeOcmKeywords(ctx, logger));
    }
 
    // V1 heuristics for when DaqContext has no specification provided by user.
    // If
    // - number of primary sources == 1
    // - number of files from that primary  == 1
    // then we automatically designate it the *in-place* target.
    std::optional<std::string> target_source_name;
 
    if (1 == ctx.prim_sources.size() &&
        1 == std::count_if(ctx.results.begin(),
                           ctx.results.end(),
                           [source = ctx.prim_sources[0].name](DpPart const& part) {
                               return part.SourceName() == source;
                           })) {
        auto it = std::find_if(ctx.results.begin(),
                               ctx.results.end(),
                               [source = ctx.prim_sources[0].name](DpPart const& part) {
                                   return part.SourceName() == source;
                               });
        assert(it != ctx.results.end());
        if (std::holds_alternative<std::string>(it->Part())) {
            // At this point we have:
            // 1 primary source
            // 1 output fits file
            auto const& path = std::get<std::string>(it->Part());
            target_source_name = ctx.prim_sources[0].name;
            LOG4CPLUS_DEBUG(logger,
                            fmt::format("{}: Heuristics resulted in using the file "
                                        "{} from {} as *in-place* merge target.",
                                        ctx.id,
                                        path,
                                        *target_source_name));
            auto& source = dp_spec.target.source.emplace();
            source.source_name = it->SourceName();
            source.location = path;
        }
    }
 
    if (ctx.results.empty()) {
        throw boost::enable_current_exception(
            std::invalid_argument("Cannot create data product specification with no results"));
    }
 
    for (DpPart const& r : ctx.results) {
        if (target_source_name && *target_source_name == r.SourceName()) {
            // Skip target file
            continue;
        }
        if (std::holds_alternative<fits::KeywordVector>(r.Part())) {
            // keywords
            json::FitsKeywordsSource s{};
            s.source_name = r.SourceName();
            s.keywords = std::get<fits::KeywordVector>(r.Part());
            dp_spec.sources.push_back(s);
        } else if (std::holds_alternative<std::string>(r.Part())) {
            json::FitsFileSource s{};
            s.source_name = r.SourceName();
            s.location = std::get<std::string>(r.Part());
            dp_spec.sources.push_back(s);
        }
    }
 
    return dp_spec;
}
 
}  // namespace v1
 
namespace v2 {
 
/**
 * Per data source common specification that is only used for more efficient lookup.
 */
struct CommonSourceSpecifications {
    /**
     * Position index in original specification, used to order sources.
     */
    std::size_t index = std::numeric_limits<std::size_t>::max();
    std::optional<json::InitialKeywords> initial_keywords = std::nullopt;
    json::KeywordRules keyword_rules = {};
};
 
std::unordered_map<std::string, CommonSourceSpecifications>
MakeCommonSpecifications(json::StartDaqV2Spec const& spec) {
    // NOTE: index 0 is reserved for internal OCM keywords
    std::size_t index = 1;
    std::unordered_map<std::string, CommonSourceSpecifications> lookup_map;
 
    for (auto const& source : spec.sources) {
        CommonSourceSpecifications common;
        common.index = index++;
        std::string source_name;
        std::visit(
            [&](auto const& v) {
                source_name = v.source_name;
                common.initial_keywords = v.initial_keywords;
                common.keyword_rules = v.keyword_rules;
            },
            source);
 
        lookup_map.emplace(std::move(source_name), std::move(common));
    }
    return lookup_map;
}
 
/**
 * Creates and returns the `/sources` and `/target` structures using DaqContext::specification.
 *
 * Assumptions:
 * - source names are unique (DAQ result `DpPart` is uniqely identifying where it comes from.
 */
json::DpSpec MakeDataProductSpecification(DaqContext const& ctx, log4cplus::Logger& logger) {
    json::DpSpec dp_spec;
    dp_spec.id = ctx.id;
    dp_spec.target.file_id = ctx.file_id;
    dp_spec.target.file_prefix = ctx.dp_name_prefix;
 
    // Receivers are simply copied from StartDaqV2Spec
    if (ctx.specification) {
        dp_spec.receivers = ctx.specification->receivers;
        LOG4CPLUS_DEBUG(logger,
                        fmt::format("{}: Number of receivers added from StartDaqV2Spec: {}",
                                    ctx.id,
                                    dp_spec.receivers.size()));
    } else {
        LOG4CPLUS_DEBUG(logger,
                        fmt::format("{}: No receivers added from StartDaqV2Spec: {}",
                                    ctx.id,
                                    dp_spec.receivers.size()));
    }
 
    // Add OCM keywords
    if (!ctx.keywords.empty()) {
        dp_spec.sources.emplace_back(MakeOcmKeywords(ctx, logger));
    }
 
    auto lookup = [lookup_map = MakeCommonSpecifications(*ctx.specification)](
                      std::string const& source_name) -> CommonSourceSpecifications const* {
        auto it = lookup_map.find(source_name);
        if (it != std::end(lookup_map)) {
            return &it->second;
        }
        return nullptr;
    };
 
    // Objectives:
    // - Use order from specification.sources
    // - Use for each result use the specified keywordRules from specification.sources.
    // - Use specified target (use lookup by sourceName)
 
    for (DpPart const& r : ctx.results) {
        LOG4CPLUS_DEBUG(logger, "Adding DpPart " << r);
        auto* common = lookup(r.SourceName());
 
        if (ctx.specification->merge_target.has_value() &&
            ctx.specification->merge_target->source_name == r.SourceName() &&
            std::holds_alternative<std::string>(r.Part())) {
            LOG4CPLUS_DEBUG(logger, "Considering merge target " << r);
            // Source is a file (as identified by std::string) that matches requested merge_target
            // name -> Add as merge target!
 
            if (!dp_spec.target.source.has_value()) {
                json::FitsFileSource s{};
                s.source_name = r.SourceName();
                s.location = std::get<std::string>(r.Part());
                dp_spec.target.source = s;
                // NOTE keyword rules are not used for target yet.
                LOG4CPLUS_DEBUG(logger,
                                fmt::format("Added merge target source from {} with file {}",
                                            s.source_name,
                                            s.location));
                continue;
            } else {
                LOG4CPLUS_WARN(logger,
                               fmt::format("Multiple source files matched as merge-target! First "
                                           "one has been chosen: {}",
                                           dp_spec.target.source->location));
            }
        }
 
        // Non merge-target sources which are either JSON keywords or FITS files
        if (std::holds_alternative<fits::KeywordVector>(r.Part())) {
            json::FitsKeywordsSource s{};
            s.source_name = r.SourceName();
            s.keywords = std::get<fits::KeywordVector>(r.Part());
            if (common != nullptr) {
                LOG4CPLUS_INFO(logger,
                               "Has common, and " << (!common->keyword_rules.empty()
                                                          ? "has keyword rules"
                                                          : "does NOT have keyword rules"));
                s.initial_keywords = common->initial_keywords;
                s.keyword_rules = common->keyword_rules;
            }
            dp_spec.sources.push_back(s);
        } else if (std::holds_alternative<std::string>(r.Part())) {
            json::FitsFileSource s{};
            s.source_name = r.SourceName();
            s.location = std::get<std::string>(r.Part());
            if (common != nullptr) {
                LOG4CPLUS_INFO(logger,
                               "Has common, and " << (!common->keyword_rules.empty()
                                                          ? "has keyword rules"
                                                          : "does NOT have keyword rules"));
                s.initial_keywords = common->initial_keywords;
                s.keyword_rules = common->keyword_rules;
            }
            dp_spec.sources.push_back(s);
        }
    }
 
    // Lookup index of provided source.
    auto index_of =
        [&, ocm_name = MakeOcmName(ctx)](json::DpSpec::SourceTypes const& s) -> std::size_t {
        auto const& name = std::visit([](auto const& t) { return t.source_name; }, s);
        if (name == ocm_name) {
            // Special case for OCM which needs to come first.
            return 0;
        }
        auto* res = lookup(name);
        if (res != nullptr) {
            return res->index;
        }
        return std::numeric_limits<std::size_t>::max();
    };
 
    // Stable sort sources according to the index order.
    std::stable_sort(std::begin(dp_spec.sources),
                     std::end(dp_spec.sources),
                     [&](json::DpSpec::SourceTypes const& a, json::DpSpec::SourceTypes const& b) {
                         return index_of(a) < index_of(b);
                     });
 
    return dp_spec;
}
 
}  // namespace v2
 
json::DpSpec MakeDataProductSpecification(DaqContext const& ctx, log4cplus::Logger& logger) {
    LOG4CPLUS_DEBUG(logger,
                    "MakeDataProductSpecification: DaqContext has a specification: "
                        << (ctx.specification.has_value() ? "yes" : "no"));
    if (ctx.specification.has_value()) {
        return v2::MakeDataProductSpecification(ctx, logger);
    } else {
        return v1::MakeDataProductSpecification(ctx, logger);
    }
}
 
}  // namespace daq