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 <ocmif/conversion.hpp>
#include <ocmif/parsing.hpp>
 
#include "requestor.hpp"
 
using ocmif::ToString;
 
std::ostream& operator<<(std::ostream& os, std::shared_ptr<ocmif::DaqStatus> const& s) {
    os << std::setprecision(std::numeric_limits<long double>::digits10 + 1);
    os << "DaqStatus:\n"
       << " id: " << s->getId() << "\n"
       << " state: \"" << ToString(s->getState()) << "\"\n"
       << " substate: \"" << ToString(s->getSubState()) << "\"\n"
       << " error: \"" << (s->getError() ? "true" : "false") << "\n"
       << " timestamp: " << s->getTimestamp();
    os << "\n";
 
    return os;
}
 
std::ostream& operator<<(std::ostream& os, std::shared_ptr<ocmif::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<ocmif::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<ocmif::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<ocmif::DaqStatus>> const& s) {
    os << std::setprecision(std::numeric_limits<long double>::digits10 + 1);
    os << "{\n"
          "  \"id\": \""
       << s.t->getId()
       << "\",\n"
          "  \"state\": \""
       << ToString(s.t->getState())
       << "\",\n"
          "  \"substate\": \""
       << ToString(s.t->getSubState())
       << "\",\n"
          "  \"error\": "
       << (s.t->getError() ? "true" : "false")
       << ",\n"
          "  \"timestamp\": "
       << s.t->getTimestamp()
       << "\n"
       << "}";
    return os;
}
 
std::ostream&
operator<<(std::ostream& os, JsonPrint<std::vector<std::shared_ptr<ocmif::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<ocmif::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<ocmif::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) {
}
 
std::string Requestor::Synopsis() {
    return std::string();
}
 
void Requestor::AddOptions(boost::program_options::options_description& opt) {
}
std::string SetLogLevelRequestor::Synopsis() {
    return "[<options>] <logger> <level>";
}
 
void SetLogLevelRequestor::Handle(::stdif::StdCmdsSync& std_cmds,
                                  ::ocmif::OcmDaqSync& daq_cmds,
                                  boost::program_options::variables_map& vm,
                                  std::vector<std::string>& opts) {
    // Note that the command daq.start is also counted.
    if (opts.size() != 3) {
        std::cerr << "setloglevel requires `<logger>` and <level>` arguments\n";
        throw std::invalid_argument("Missing required arguments");
    }
    auto log_info = m_args.mal->createDataEntity<stdif::LogInfo>();
    log_info->setLogger(opts[1]);
    log_info->setLevel(opts[2]);
    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(boost::program_options::options_description& opt) {
    namespace po = boost::program_options;
    opt.add_options()("id", po::value<std::string>(&m_id), "data acquisition identifer")(
        "prefix", po::value<std::string>(&m_prefix), "data product prefix")(
        "properties", po::value<std::string>(&m_properties), "JSON properties");
 
}
 
std::string StartDaqRequestor::Synopsis() {
    return "[<options>] <primary-sources> <metadata-sources>";
}
 
void StartDaqRequestor::Handle(::stdif::StdCmdsSync& std_cmds,
                               ::ocmif::OcmDaqSync& daq_cmds,
                               boost::program_options::variables_map& vm,
                               std::vector<std::string>& opts) {
    // Note that the command daq.start is also counted.
    if (opts.size() != 3) {
        std::cerr << "start daq requires the `<primary-sources>` and <metadata-sources>` arguments\n";
        throw std::invalid_argument("Missing required arguments");
    }
    std::cerr << "Properties: " << m_properties << "\n";
    auto reply = daq_cmds.StartDaq(m_id, m_prefix, opts[1], opts[2], m_properties);
    if (m_args.json) {
        std::cout << JsonPrint(reply);
    } else {
        std::cout << "Reply: " << reply << std::endl;
    }
}
 
void UpdateKeywordsRequestor::AddOptions(boost::program_options::options_description& opt) {
    namespace po = boost::program_options;
}
 
std::string UpdateKeywordsRequestor::Synopsis() {
    return "[<options>] <id> <keywords>";
}
 
void UpdateKeywordsRequestor::Handle(::stdif::StdCmdsSync& std_cmds,
                                     ::ocmif::OcmDaqSync& daq_cmds,
                                     boost::program_options::variables_map& vm,
                                     std::vector<std::string>& opts) {
    // Note that the command daq.updatekeywords is also counted.
    if (opts.size() != 3) {
        std::cerr << "UpdateKeywords requires the `<id>` and <keywords>` arguments\n";
        throw std::invalid_argument("Missing required arguments");
    }
    auto reply = daq_cmds.UpdateKeywords(opts[1], opts[2]);
    if (m_args.json) {
        std::cout << JsonPrint(reply);
    } else {
        std::cout << "Reply: " << reply << std::endl;
    }
}
 
 
void AwaitStateRequestor::AddOptions(boost::program_options::options_description& opt) {
    namespace po = boost::program_options;
}
 
std::string AwaitStateRequestor::Synopsis() {
    return "[<options>] <id> <state> <substate> <timeout[s]>";
}
 
void AwaitStateRequestor::Handle(::stdif::StdCmdsSync& std_cmds,
                                     ::ocmif::OcmDaqSync& daq_cmds,
                                     boost::program_options::variables_map& vm,
                                     std::vector<std::string>& opts) {
    // Note that the command is also counted.
    if (opts.size() != 5) {
        std::cerr << "AwaitDaqState requires the `<id>`, `<state>`, `<substate>` and <timeout>` "
            "arguments\n";
        throw std::invalid_argument("Missing required arguments");
    }
    auto state = ocmif::FromString<ocmif::DaqState>(opts[2]);
    auto substate = ocmif::FromString<ocmif::DaqSubState>(opts[3]);
    double timeout = std::stod(opts[4]);
    auto reply = daq_cmds.AwaitDaqState(opts[1], state, substate, timeout);
    if (m_args.json) {
        std::cout << JsonPrint(reply);
    } else {
        std::cout << "Reply: " << reply << std::endl;
    }
}