main.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_ocm_serverctl
 * @copyright ESO - European Southern Observatory
 *
 * @brief
 */
#include <daq/config.hpp>
 
#include <memory>
#include <thread>
#include <unordered_map>
#include <vector>
 
#include <CLI/CLI.hpp>
#include <boost/format.hpp>
 
#include <mal/Cii.hpp>
#include <mal/Mal.hpp>
#include <mal/rr/qos/ConnectionTime.hpp>
#include <mal/rr/qos/ReplyTime.hpp>
#include <mal/utility/LoadMal.hpp>
#include <rad/logger.hpp>
#include <log4cplus/hierarchy.h>
 
#include <Daqif.hpp>
#include <Stdif.hpp>
 
#include <daq/conversion.hpp>
#include <daqif/subscription.hpp>
#include <daqif/uri.hpp>
 
#include "requestor.hpp"
 
volatile std::sig_atomic_t g_signal_status;
void SignalHandler(int signal) {
    g_signal_status = signal;
}
 
 
int main(int argc, char** argv) {
    using elt::mal::Uri;
 
    log4cplus::initialize();
    // Delete mal configuration that writes to stdout, interfering with JSON output
    log4cplus::Logger::getDefaultHierarchy().resetConfiguration();
 
    // Log to stderr
    log4cplus::BasicConfigurator(log4cplus::Logger::getDefaultHierarchy(), true).configure();
    RAD_LOG_TO_CONSOLE(true);
    RAD_LOG_SETLEVEL("TRACE");
 
    std::signal(SIGINT, SignalHandler);
    std::signal(SIGTERM, SignalHandler);
 
    CommonArgs args;
 
    CLI::App app("daqOcmCtl");
    app.add_option(
           "--rep",
           args.req_addr,
           "request endpoint without service stem. e.g. zpb://127.0.0.1:4020, defaults to env. "
           "$OCM_REQUEST_EP")
        ->envname("OCM_REQUEST_EP");
    auto* pep =
        app.add_option("--pep", args.pub_addr, "publish endpoint, defaults to env $OCM_PUBLISH_EP")
            ->envname("OCM_PUBLISH_EP");
    app.add_flag("--json", args.json, "output in JSON format");
    app.add_flag("--status",
                 args.status,
                 "subscribe to server topics (requires --pep or $OCM_PUBLISH_EP to be set)")
        ->needs(pep);
    app.add_option("--timeout,-t", args.timeout, "request timeout in seconds", true);
    app.add_flag_callback(
        "--version",
        [] {
            std::cerr << "daqOcmCtl " << VERSION
#ifdef DEBUG
                      << " (debug build)"
#endif
                      << std::endl;
            throw CLI::Success(VERSION, 0);
        },
        "Print version and exit");
 
    std::vector<std::string> subargs;
 
    std::map<std::string, std::unique_ptr<Requestor>> requestors;
    requestors.emplace(
        std::make_pair("std.getstatus",
                       std::make_unique<SimpleRequestor<decltype(&stdif::StdCmds::GetStatus)>>(
                           &stdif::StdCmds::GetStatus, args)));
    requestors.emplace(
        std::make_pair("std.getstate",
                       std::make_unique<SimpleRequestor<decltype(&stdif::StdCmds::GetState)>>(
                           &stdif::StdCmds::GetState, args)));
    requestors.emplace(
        std::make_pair("std.getversion",
                       std::make_unique<SimpleRequestor<decltype(&stdif::StdCmds::GetVersion)>>(
                           &stdif::StdCmds::GetVersion, args)));
    requestors.emplace(
        std::make_pair("std.setloglevel", std::make_unique<SetLogLevelRequestor>(args)));
    requestors.emplace(
        std::make_pair("std.init",
                       std::make_unique<SimpleRequestor<decltype(&stdif::StdCmds::Init)>>(
                           &stdif::StdCmds::Init, args)));
    requestors.emplace(
        std::make_pair("std.enable",
                       std::make_unique<SimpleRequestor<decltype(&stdif::StdCmds::Enable)>>(
                           &stdif::StdCmds::Enable, args)));
    requestors.emplace(
        std::make_pair("std.disable",
                       std::make_unique<SimpleRequestor<decltype(&stdif::StdCmds::Disable)>>(
                           &stdif::StdCmds::Disable, args)));
    requestors.emplace(
        std::make_pair("std.exit",
                       std::make_unique<SimpleRequestor<decltype(&stdif::StdCmds::Exit)>>(
                           &stdif::StdCmds::Exit, args)));
 
    requestors.emplace(std::make_pair(
        "daq.stop",
        std::make_unique<SimpleDaqRequestor<decltype(&daqif::OcmDaqControl::StopDaq)>>(
            &daqif::OcmDaqControl::StopDaq, args)));
    requestors.emplace(std::make_pair(
        "daq.forcestop",
        std::make_unique<SimpleDaqRequestor<decltype(&daqif::OcmDaqControl::ForceStopDaq)>>(
            &daqif::OcmDaqControl::ForceStopDaq, args)));
    requestors.emplace(std::make_pair(
        "daq.abort",
        std::make_unique<SimpleDaqRequestor<decltype(&daqif::OcmDaqControl::AbortDaq)>>(
            &daqif::OcmDaqControl::AbortDaq, args)));
    requestors.emplace(std::make_pair(
        "daq.forceabort",
        std::make_unique<SimpleDaqRequestor<decltype(&daqif::OcmDaqControl::ForceAbortDaq)>>(
            &daqif::OcmDaqControl::ForceAbortDaq, args)));
 
    requestors.emplace(std::make_pair(
        "daq.getstatus",
        std::make_unique<SimpleDaqRequestor<decltype(&daqif::OcmDaqControl::GetStatus)>>(
            &daqif::OcmDaqControl::GetStatus, args)));
    requestors.emplace(std::make_pair(
        "daq.getactivelist",
        std::make_unique<NoArgRequestor<decltype(&daqif::OcmDaqControl::GetActiveList)>>(
            &daqif::OcmDaqControl::GetActiveList, args)));
 
    requestors.emplace(std::make_pair("daq.start", std::make_unique<StartDaqRequestor>(args)));
    requestors.emplace(std::make_pair("daq.startv2", std::make_unique<StartDaqV2Requestor>(args)));
    requestors.emplace(
        std::make_pair("daq.updatekeywords", std::make_unique<UpdateKeywordsRequestor>(args)));
    requestors.emplace(
        std::make_pair("daq.awaitstate", std::make_unique<AwaitStateRequestor>(args)));
 
    // Allow each command to add their arguments
    for (auto& r : requestors) {
        auto& command = r.first;
        CLI::App* sub = app.add_subcommand(command);
        r.second->AddOptions(sub);
    }
 
    try {
        CLI11_PARSE(app, argc, argv);
 
        // Load ZPB and register MAL
        auto mal = elt::mal::loadMal("zpb", {});
        args.mal = mal.get();
        elt::mal::CiiFactory::getInstance().registerMal("zpb", mal);
 
        std::optional<daqif::Subscription<stdif::Status>> std_status;
        std::optional<daqif::Subscription<daqif::DaqStatus>> daq_status;
 
        if (args.status) {
            if (args.pub_addr.back() != '/') {
                args.pub_addr.push_back('/');
            }
            std_status.emplace(daqif::MakeSubscription<stdif::Status>(
                *mal,
                daqif::MakeServiceUri(args.pub_addr, "std/status"),
                [](auto& sub, auto const& event) {
                    auto const& sample = event.getData();
                    std::cerr << "status: " << sample->getStatus() << std::endl;
                }));
            daq_status.emplace(daqif::MakeSubscription<daqif::DaqStatus>(
                *mal, Uri(args.pub_addr + "daq/status"), [](auto& sub, auto const& event) {
                    auto const& sample = event.getData();
                    std::cerr << "daq: id=" << sample->getId()
                              << ", file_id=" << sample->getFileId()
                              << ", state=" << daq::ToString(sample->getState())
                              << ", substate=" << daq::ToString(sample->getSubState())
                              << ", error=" << (sample->getError() ? "true" : "false")
                              << ", message=" << sample->getMessage()
                              << std::endl;
                }));
        }
 
        auto command = app.get_subcommands();
        if (command.empty()) {
            if (args.status) {
                std::cerr << "no command provided -> will subscribe indefinitely\n";
                while (!g_signal_status) {
                    std::this_thread::sleep_for(std::chrono::milliseconds(100));
                }
                return 0;
            }
            return 1;
        }
 
        auto std_cmds = elt::mal::CiiFactory::getInstance().getClient<stdif::StdCmdsSync>(
            daqif::MakeServiceUri(args.req_addr, "std"),
            {std::make_shared<elt::mal::rr::qos::ReplyTime>(std::chrono::seconds(args.timeout))},
            {});
        auto daq_cmds = elt::mal::CiiFactory::getInstance().getClient<::daqif::OcmDaqControlSync>(
            daqif::MakeServiceUri(args.req_addr, "daq"),
            {std::make_shared<elt::mal::rr::qos::ReplyTime>(std::chrono::seconds(args.timeout))},
            {});
 
        // Look up the requestor in the known commands map
        auto req_it = requestors.find(command[0]->get_name());
        if (req_it != requestors.end()) {
            req_it->second->Handle(*std_cmds, *daq_cmds, command[0]);
            std::cerr << "Done\n";
            return 0;
        } else {
            std::cerr << "error: Unknown command: '" << command[0]->get_name() << "'\n";
            return 1;
        }
    } catch (daqif::DaqException const& e) {
        std::cerr << "DaqException: id: '" << e.getId() << "', message: " << e.getMessage()
                  << std::endl;
        return 1;
    } catch (CLI::ParseError const& e) {
        std::cerr << e.what() << std::endl;
        return 1;
    } catch (std::exception const& e) {
        std::cerr << "error: " << e.what() << std::endl;
        return 1;
    } catch (...) {
        std::cerr << "unknown error" << std::endl;
        return 1;
    }
}