requestor.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_ocm_serverctl
 * @copyright ESO - European Southern Observatory
 *
 * @brief
 */
#include <chrono>
#include <string>
 
#include <fmt/format.h>
 
#include <daq/conversion.hpp>
#include <daqif/parsing.hpp>
 
#include "requestor.hpp"
 
using daq::ToString;
 
std::ostream& operator<<(std::ostream& os, std::shared_ptr<daqif::DaqStatus> const& s) {
    os << std::setprecision(std::numeric_limits<long double>::digits10 + 1);
    os << "DaqStatus:\n"
       << " id: " << s->getId() << "\n"
       << " file_id: " << s->getFileId() << "\n"
       << " state: \"" << ToString(s->getState()) << "\"\n"
       << " substate: \"" << ToString(s->getSubState()) << "\"\n"
       << " error: \"" << (s->getError() ? "true" : "false") << "\n"
       << " message: " << s->getMessage() << "\n"
       << " result: " << s->getResult() << "\n"
       << " timestamp: " << s->getTimestamp();
    os << "\n";
 
    return os;
}
 
std::ostream& operator<<(std::ostream& os, std::shared_ptr<daqif::DaqReply> const& s) {
    os << "DaqReply:\n"
       << " id: " << s->getId() << "\n"
       << " error: " << (s->getError() ? "true" : "false") << "\n";
    return os;
}
 
std::ostream& operator<<(std::ostream& os, std::shared_ptr<daqif::AwaitDaqReply> const& s) {
    os << "AwaitDaqReply:\n"
       << " timeout: " << (s->getTimeout() ? "true" : "false") << "\n"
       << " status: " << s->getStatus() << "\n";
    return os;
}
 
std::ostream&
operator<<(std::ostream& os, std::vector<std::shared_ptr<daqif::DaqStatus>> const& vec) {
    for (auto const& s : vec) {
        os << JsonPrint(s);
        os << "\n";
    }
    return os;
}
 
std::ostream& operator<<(std::ostream& os, JsonPrint<std::shared_ptr<daqif::DaqStatus>> const& s) {
    os << std::setprecision(std::numeric_limits<long double>::digits10 + 1);
    os << "{\n"
          "  \"id\": \""
       << s.t->getId()
       << "\",\n"
          "  \"file_id\": \""
       << s.t->getFileId()
       << "\",\n"
          "  \"state\": \""
       << ToString(s.t->getState())
       << "\",\n"
          "  \"substate\": \""
       << ToString(s.t->getSubState())
       << "\",\n"
          "  \"error\": "
       << (s.t->getError() ? "true" : "false")
       << ",\n"
          "  \"result\": \""
       << s.t->getResult()
       << "\",\n"
          "  \"timestamp\": "
       << s.t->getTimestamp() << "\n"
       << "}";
    return os;
}
 
std::ostream&
operator<<(std::ostream& os, JsonPrint<std::vector<std::shared_ptr<daqif::DaqStatus>>> const& vec) {
    os << "[\n";
    size_t num = 0;
    for (auto const& s : vec.t) {
        num++;
        os << JsonPrint(s);
        if (num < vec.t.size()) {
            os << ",\n";
        }
    }
    os << "\n]";
    return os;
}
 
std::ostream& operator<<(std::ostream& os, JsonPrint<std::shared_ptr<daqif::DaqReply>> const& s) {
    os << "{\n"
          "  \"id\": \""
       << s.t->getId()
       << "\",\n"
          "  \"error\": "
       << (s.t->getError() ? "true" : "false") << "\n";
    os << "}";
 
    return os;
}
 
std::ostream&
operator<<(std::ostream& os, JsonPrint<std::shared_ptr<daqif::AwaitDaqReply>> const& s) {
    os << "{\n"
          "  \"timeout\": "
       << (s.t->getTimeout() ? "true" : "false")
       << ",\n"
          "  \"status\": "
       << JsonPrint(s.t->getStatus()) << "\n";
    os << "}";
 
    return os;
}
 
std::ostream& operator<<(std::ostream& os, JsonPrint<std::string> const& s) {
    os << "\"" << s.t << "\"";
    return os;
}
 
Requestor::Requestor(CommonArgs& args) : m_args(args) {
}
 
void Requestor::AddOptions(CLI::App* app) {
}
 
void SetLogLevelRequestor::AddOptions(CLI::App* sub) {
    sub->add_option("logger", m_logger, "Logger name");
    sub->add_option("level", m_level, "Log level");
}
 
void SetLogLevelRequestor::Handle(::stdif::StdCmdsSync& std_cmds,
                                  ::daqif::OcmDaqControlSync& daq_cmds,
                                  CLI::App* app) {
    auto log_info = m_args.mal->createDataEntity<stdif::LogInfo>();
    log_info->setLogger(m_logger);
    log_info->setLevel(m_level);
    auto reply = std_cmds.SetLogLevel(log_info);
    if (m_args.json) {
        std::cout << JsonPrint(reply);
    } else {
        std::cout << "Reply: " << reply << std::endl;
    }
}
 
void StartDaqRequestor::AddOptions(CLI::App* sub) {
    sub->add_option("--id", m_id, "DAQ ID");
    sub->add_option("--prefix", m_prefix, "File prefix");
    sub->add_option("--properties", m_properties, "JSON properties");
    sub->add_option("primary", m_primary_sources, "Primary data sources");
    sub->add_option("metadata", m_metadata_sources, "Metadata sources");
}
 
void StartDaqRequestor::Handle(::stdif::StdCmdsSync& std_cmds,
                               ::daqif::OcmDaqControlSync& daq_cmds,
                               CLI::App* app) {
    auto reply =
        daq_cmds.StartDaq(m_id, m_prefix, m_primary_sources, m_metadata_sources, m_properties);
    if (m_args.json) {
        std::cout << JsonPrint(reply);
    } else {
        std::cout << "Reply: " << reply << std::endl;
    }
}
 
void StartDaqV2Requestor::AddOptions(CLI::App* sub) {
    auto file_or_json = [](std::string const& arg) -> std::string {
        if (arg.size() > 1 && arg.front() == '@') {
            auto path = arg.substr(1);
            if (path == "-") {
                std::string file_body((std::istreambuf_iterator<char>(std::cin)),
                                      (std::istreambuf_iterator<char>()));
                return file_body;
            } else {
                std::ifstream f(path);
                if (!f.is_open()) {
                    throw std::invalid_argument(
                        fmt::format("File not found or readable: {}", arg.substr(1)));
                }
                std::string file_body((std::istreambuf_iterator<char>(f)),
                                      (std::istreambuf_iterator<char>()));
                return file_body;
            }
        } else {
            return arg;
        }
    };
    sub->add_option("specification",
                    m_spec,
                    "JSON specification. Load from file by prefixing with '@', e.g. @file.json\n"
                    "or from stdin '-' with @-")
        ->transform(file_or_json)
        ->required();
}
 
void StartDaqV2Requestor::Handle(::stdif::StdCmdsSync& std_cmds,
                                 ::daqif::OcmDaqControlSync& daq_cmds,
                                 CLI::App* app) {
    auto reply = daq_cmds.StartDaqV2(m_spec);
    if (m_args.json) {
        std::cout << JsonPrint(reply);
    } else {
        std::cout << "Reply: " << reply << std::endl;
    }
}
 
void UpdateKeywordsRequestor::AddOptions(CLI::App* sub) {
    sub->add_option("id", m_id, "DAQ ID");
    sub->add_option("keywords", m_keywords, "JSON encoded list of keywords");
}
 
void UpdateKeywordsRequestor::Handle(::stdif::StdCmdsSync& std_cmds,
                                     ::daqif::OcmDaqControlSync& daq_cmds,
                                     CLI::App* app) {
    auto reply = daq_cmds.UpdateKeywords(m_id, m_keywords);
    if (m_args.json) {
        std::cout << JsonPrint(reply);
    } else {
        std::cout << "Reply: " << reply << std::endl;
    }
}
 
void AwaitStateRequestor::AddOptions(CLI::App* sub) {
    sub->add_option("id", m_id, "DAQ ID");
    sub->add_option("state", m_state, "State to await");
    sub->add_option("substate", m_substate, "Substate to await");
    sub->add_option("timeout", m_timeout, "timeout in [fractional] seconds");
    sub->callback([&] {
        auto min_timeout = m_timeout + 2;
        if (m_args.timeout < min_timeout) {
            std::cerr << "Note: Setting request timeout to " << min_timeout
                      << "s due to await timeout exceeding request timeout\n";
            m_args.timeout = min_timeout;
        }
    });
}
 
void AwaitStateRequestor::Handle(::stdif::StdCmdsSync& std_cmds,
                                 ::daqif::OcmDaqControlSync& daq_cmds,
                                 CLI::App* app) {
    auto state = daqif::FromString<daqif::DaqState>(m_state);
    auto substate = daqif::FromString<daqif::DaqSubState>(m_substate);
    auto reply = daq_cmds.AwaitDaqState(m_id, state, substate, m_timeout);
    if (m_args.json) {
        std::cout << JsonPrint(reply);
    } else {
        std::cout << "Reply: " << reply << std::endl;
    }
}