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 <nlohmann/json.hpp>
 
#include <daq/conversion.hpp>
#include <daqif/parsing.hpp>
 
#include "requestor.hpp"
 
using daq::ToString;
 
namespace daqif {
// NOLINTNEXTLINE
void to_json(nlohmann::json& j, daqif::DaqStatus const& status) {
    j = nlohmann::json{{"id", status.getId()},
                       {"fileId", status.getFileId()},
                       {"state", ToString(status.getState())},
                       {"substate", ToString(status.getSubState())},
                       {"error", status.getError()},
                       {"result", status.getResult()},
                       {"message", status.getMessage()},
                       {"timestamp", status.getTimestamp()}};
}
 
}  // namespace daqif
 
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 << "----\n";
        os << s;
    }
    return os;
}
 
std::ostream& operator<<(std::ostream& os, JsonPrint<std::shared_ptr<daqif::DaqStatus>> const& s) {
    nlohmann::json j_status;
    daqif::to_json(j_status, *s.t);
    os << j_status.dump(2);
    return os;
}
 
std::ostream&
operator<<(std::ostream& os, JsonPrint<std::vector<std::shared_ptr<daqif::DaqStatus>>> const& vec) {
    nlohmann::json j_array = nlohmann::json::array();
    for (auto const& s : vec.t) {
        nlohmann::json j_status;
        daqif::to_json(j_status, *s);
        j_array.push_back(j_status);
    }
    os << j_array.dump(2);
    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) {
    auto const& reply = *s.t;
    auto const& rep = *reply.getStatus();
    nlohmann::json j_status;
    daqif::to_json(j_status, rep);
    nlohmann::json j{{"timeout", reply.getTimeout()}, {"status", j_status}};
    os << j.dump(2);
    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 << 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,
                    "Space separated list of primary data sources, each in the form `name@uri`, "
                    "e.g. \"dcs1@zpb.rr://10.207.214.220:1234/RecCmds\"");
    sub->add_option("metadata",
                    m_metadata_sources,
                    "Space separated list of metadata data sources, each in the form `name@uri`, "
                    "e.g. \"fcf1@zpb.rr://10.207.214.218:1234/MetaDaq\"");
}
 
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 << 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 << 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 << 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 << std::endl;
    }
}