entrypoint.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_dpm_merge
 * @copyright ESO - European Southern Observatory
 */
#include "entrypoint.hpp"
 
#include <iostream>
 
#include <fmt/format.h>
#include <log4cplus/loggingmacros.h>
 
#include <daq/dpm/keywordEx.hpp>
#include <daq/fits/cfitsio.hpp>
 
#include "merge.hpp"
 
namespace fs = std::filesystem;
namespace {
template <class T>
constexpr bool AlwaysFalse() {
    return false;
}
}  // namespace
 
namespace daq::dpm::merge {
 
std::optional<KeywordRuleProcessor::DefaultRule>
Convert(std::optional<::daq::json::InitialKeywords> const& rhs) {
    if (!rhs.has_value()) {
        return std::nullopt;
    }
    switch (*rhs) {
    case daq::json::InitialKeywords::All:
        return KeywordRuleProcessor::DefaultRule::All;
    case daq::json::InitialKeywords::None:
        return KeywordRuleProcessor::DefaultRule::None;
    case daq::json::InitialKeywords::User:
        return KeywordRuleProcessor::DefaultRule::User;
    default:
        throw std::invalid_argument(
            fmt::format("Unknown value of daq::json::InitialKeywords: {}",
                        static_cast<std::underlying_type_t<daq::json::InitialKeywords>>(*rhs)));
    }
}
 
/**
 * Reports in JSON format
 */
class JsonReporter : public StatusReporter {
public:
    JsonReporter(log4cplus::Logger logger) : m_logger(std::move(logger)) {
    }
    virtual void PostAlert(std::string const& id, std::string const& message) override {
        LOG4CPLUS_WARN(m_logger, "Alert: " << message);
        Report("alert", nlohmann::json({{"id", id}, {"message", message}}));
    }
 
private:
    void Report(char const* type, nlohmann::json const& content) {
        auto ts =
            std::chrono::nanoseconds(std::chrono::system_clock::now().time_since_epoch()).count();
        nlohmann::json j{{"type", type}, {"timestamp", ts}, {"content", content}};
        std::cout << j;
    }
    log4cplus::Logger m_logger;
};
 
/**
 * Reports in unspecified human-readable format
 */
class SimpleReporter : public StatusReporter {
public:
    virtual void PostAlert(std::string const& id, std::string const& message) override {
        std::cout << "[event] alert: " << message << '\n';
    }
};
 
class Formatter : public KeywordFormatter {
public:
    virtual fits::LiteralKeyword FormatKeyword(fits::KeywordVariant const& keyword) override {
        // @todo: Use dictionary to validate and format
        return fits::Format(keyword);
    }
};
 
class Sorter : public KeywordSorter {
public:
    virtual void SortKeywords(std::vector<fits::LiteralKeyword>& keywords) override {
        fits::v1::StandardSort(keywords);
    }
};
 
fs::path MakeSourcePath(fs::path source, fs::path const& root) {
    if (!source.is_absolute()) {
        return root / source;
    }
    return source;
}
 
std::unique_ptr<KeywordRuleProcessor> MakeKeywordRuleProcessor(json::KeywordRules const& rules) {
    // If rules is an empty list it means "do not copy anything"
    // whereas a completly undefined list is treated as "copy with provided rules"
    auto processor = std::make_unique<StandardKeywordRuleProcessor>();
    for (json::KeywordRuleTypes const& rule : rules) {
        std::visit(
            [&](auto const& rule) {
                using T = std::decay_t<decltype(rule)>;
                if constexpr (std::is_same_v<T, json::KeywordFilter>) {
                    LOG4CPLUS_DEBUG("daq.dpmmerge", fmt::format("Adding KeywordRule 'filter'"));
                    processor->AddRule([filter = KeywordEx(std::begin(rule.selection_patterns),
                                                           std::end(rule.selection_patterns))](
                                           fits::KeywordVector const& kws) -> fits::KeywordVector {
                        fits::KeywordVector result;
                        result.reserve(kws.size());
                        Filter(std::begin(kws), std::end(kws), std::back_inserter(result), filter);
                        return result;
                    });
                } else if constexpr (std::is_same_v<T, json::KeywordTransform>) {
                    LOG4CPLUS_DEBUG("daq.dpmmerge", fmt::format("Adding KeywordRule 'transform'"));
                    processor->AddRule([filter = KeywordEx(std::begin(rule.selection_patterns),
                                                           std::end(rule.selection_patterns)),
                                        regex = std::regex(rule.regex),
                                        format = rule.format](
                                           fits::KeywordVector const& kws) -> fits::KeywordVector {
                        fits::KeywordVector result;
                        result.reserve(kws.size());
                        Transform(std::begin(kws),
                                  std::end(kws),
                                  std::back_inserter(result),
                                  filter,
                                  regex,
                                  format.c_str());
                        return result;
                    });
                } else {
                    static_assert(AlwaysFalse<T>(), "non exhaustive visitor");
                }
            },
            rule);
    }
    return processor;
}
 
int Entrypoint(fs::path const& root,
               std::optional<fs::path> const& opt_out_path,
               json::DpSpec const& spec,
               SourceResolver const& resolver,
               bool dry_run,
               bool use_json) {
    fits::MemoryFitsFile dry_run_file =
        dry_run ? fits::MemoryFitsFile(4096u) : fits::MemoryFitsFile();
 
    auto out_path = opt_out_path.value_or(
        root / fs::path(fmt::format("{}{}.fits", spec.target.file_prefix, spec.target.file_id)));
    TargetSource target_source = [&] {
        if (spec.target.source) {
            /* in-place merge */
            LOG4CPLUS_INFO("daq.dpmmerge",
                           fmt::format("*in-place* target specified act as merge target: '{}'",
                                       spec.target.source->source_name));
 
            auto kw_rules = MakeKeywordRuleProcessor(spec.target.source->keyword_rules);
            auto path = MakeSourcePath(
                resolver.Resolve({spec.target.source->source_name, spec.target.source->location}),
                root);
            if (dry_run) {
                // Open existing in read-only mode
                return TargetSource(spec.target.source->source_name,
                                    path,
                                    Convert(spec.target.source->initial_keywords),
                                    std::move(kw_rules),
                                    fits::Open(path.c_str(), fits::OpenMode::ReadOnly));
            } else {
                return TargetSource(spec.target.source->source_name,
                                    path,
                                    Convert(spec.target.source->initial_keywords),
                                    std::move(kw_rules));
            }
        } else {
            /* new empty, but valid, file to act as the merge target */
            LOG4CPLUS_INFO("daq.dpmmerge",
                           "no *in-place* target specified -> creating minimal FITS file to "
                           "act as merge target");
            // Create new target file, initialized without primary data array
            auto path =
                MakeSourcePath(fmt::format("{}-in-progress.fits", spec.target.file_id), root);
            std::optional<KeywordRuleProcessor::DefaultRule> initial_keywords = std::nullopt;
            if (dry_run) {
                // Use in-memory file
                return TargetSource("" /* no name */,
                                    path,
                                    initial_keywords,
                                    std::make_unique<StandardKeywordRuleProcessor>(),
                                    std::move(dry_run_file).GetOwnedFile());
            } else {
                auto ptr = fits::CreateEmpty(path.c_str());
                fits::InitPrimaryHduNoImage(ptr.get());
                return TargetSource("" /* no name */,
                                    path,
                                    initial_keywords,
                                    std::make_unique<StandardKeywordRuleProcessor>());
            }
        }
    }();
 
    std::vector<SourceTypes> sources;
    for (json::DpSpec::SourceTypes const& dp_source : spec.sources) {
        std::visit(
            [&](auto const& source) {
                using T = std::decay_t<decltype(source)>;
 
                auto kw_rules = MakeKeywordRuleProcessor(source.keyword_rules);
                if constexpr (std::is_same_v<T, json::FitsKeywordsSource>) {
                    sources.emplace_back(std::in_place_type<FitsKeywordsSource>,
                                         source.source_name,
                                         source.keywords,
                                         Convert(source.initial_keywords),
                                         std::move(kw_rules));
                } else if constexpr (std::is_same_v<T, json::FitsFileSource>) {
                    auto path = MakeSourcePath(
                        resolver.Resolve({source.source_name, source.location}), root);
                    sources.emplace_back(std::in_place_type<FitsFileSource>,
                                         source.source_name,
                                         std::move(path),
                                         source.location,
                                         Convert(source.initial_keywords),
                                         std::move(kw_rules));
                } else {
                    static_assert(AlwaysFalse<T>(), "non exhaustive visitor");
                }
            },
            dp_source);
    }
 
    // Execute merge
    auto logger = log4cplus::Logger::getInstance("daq.dpmmerge");
    JsonReporter json_reporter(logger);
    SimpleReporter simple_reporter;
    Formatter formatter;
    Sorter sorter;
    Operations ops{use_json ? static_cast<StatusReporter&>(json_reporter)
                            : static_cast<StatusReporter&>(simple_reporter),
                   formatter,
                   sorter,
                   logger};
 
    daq::dpm::merge::Params params;
    params.arcfile = spec.target.file_id + ".fits";
    params.origfile = out_path.filename().native();
 
    try {
        Merge(ops, params, target_source, sources, dry_run);
    } catch (...) {
        std::throw_with_nested(std::runtime_error("Merge failed"));
    }
 
    // Finally move file into output and change permission
    target_source.Close();
 
    if (!dry_run) {
        fs::permissions(target_source.GetFilePath(),
                        fs::perms::owner_read | fs::perms::group_read | fs::perms::others_read,
                        fs::perm_options::replace);
        fs::rename(target_source.GetFilePath(), out_path);
    }
    return 0;
}
 
}  // namespace daq::dpm::merge