acquisitor.hpp

Go to the documentation of this file.

 
#ifndef RTCTK_METADATACOLLECTOR_ACQUISITOR_HPP
#define RTCTK_METADATACOLLECTOR_ACQUISITOR_HPP
 
#include <Metadaqif.hpp>
#include <Rtctkif.hpp>
#include <rtctk/componentFramework/events.rad.hpp>
#include <rtctk/componentFramework/exceptions.hpp>
#include <rtctk/componentFramework/rtcComponent.hpp>
#include <rtctk/componentFramework/stopToken.hpp>
#include <rtctk/metadataCollector/acquisitor.rad.hpp>
 
namespace rtctk::metadataCollector {
 
using namespace rtctk::componentFramework;
 
template <typename Super>
struct Runnable;
 
template <typename Super>
struct Loopaware;

class DaqRequestRejected : public metadaqif::DaqException {
public:
    DaqRequestRejected(const std::string& request_id,
                       const std::string& session_id,
                       const std::string& state_id);
};

class DaqReply : public metadaqif::DaqReply {
public:
    DaqReply() = default;
 
    DaqReply(const std::string& id);
 
    std::string getId() const override;
 
    void setId(const std::string& id) override;
 
    bool hasKey() const override;
 
    bool keyEquals(const metadaqif::DaqReply& other) const override;
 
    std::unique_ptr<metadaqif::DaqReply> clone() const override;
 
    std::unique_ptr<metadaqif::DaqReply> cloneKey() const override;
 
private:
    std::string m_id;
};

class DaqStopReply : public metadaqif::DaqStopReply {
public:
    DaqStopReply() = default;
 
    DaqStopReply(const std::string& id,
                 const std::vector<std::string>& files,
                 const std::string& kw);
 
    std::string getId() const override;
 
    void setId(const std::string& id) override;
 
    std::string getKeywords() const override;
 
    void setKeywords(const std::string& kw) override;
 
    std::vector<std::string> getFiles() const override;
 
    void setFiles(const std::vector<std::string>& fs) override;
 
    bool hasKey() const override;
 
    bool keyEquals(const metadaqif::DaqStopReply& other) const override;
 
    std::unique_ptr<metadaqif::DaqStopReply> clone() const override;
 
    std::unique_ptr<metadaqif::DaqStopReply> cloneKey() const override;
 
private:
    std::string m_id;
    std::vector<std::string> m_files;
    std::string m_kw;
};

class DaqStatus : public metadaqif::DaqStatus {
public:
    DaqStatus() = default;
 
    std::string getId() const override;
 
    void setId(const std::string& id) override;
 
    std::string getKeywords() const override;
 
    void setKeywords(const std::string& kw) override;
 
    std::string getMessage() const override;
 
    void setMessage(const std::string& msg) override;
 
    double getTimestamp() const override;
 
    void setTimestamp(double ts) override;
 
    metadaqif::DaqState getState() const override;
 
    void setState(metadaqif::DaqState st) override;
 
    std::vector<std::string> getFiles() const override;
 
    void setFiles(const std::vector<std::string>& fs) override;
 
    bool hasKey() const override;
 
    bool keyEquals(const metadaqif::DaqStatus& other) const override;
 
    std::unique_ptr<metadaqif::DaqStatus> clone() const override;
 
    std::unique_ptr<metadaqif::DaqStatus> cloneKey() const override;
 
private:
    std::string m_id;
    std::string m_msg;
    double m_ts;
    metadaqif::DaqState m_st;
    std::vector<std::string> m_files;
    std::string m_kw;
};

class AcquisitorCmdsImpl : public metadaqif::AsyncMetaDaq {
public:
    static void Register(CommandReplier& replier, StateMachineEngine& engine);
 
    AcquisitorCmdsImpl(StateMachineEngine& engine);
 
    ::elt::mal::future<std::shared_ptr<metadaqif::DaqReply>>
    StartDaq(const std::string& id) override;
 
    ::elt::mal::future<std::shared_ptr<metadaqif::DaqStopReply>>
    StopDaq(const std::string& id) override;
 
    ::elt::mal::future<std::shared_ptr<metadaqif::DaqReply>>
    AbortDaq(const std::string& id) override;
 
    ::elt::mal::future<std::shared_ptr<metadaqif::DaqStatus>>
    GetDaqStatus(const std::string& id) override;
 
private:
    log4cplus::Logger& m_logger;
    StateMachineEngine& m_engine;
};

template <typename Super>
struct Acquisitor : Super {
    static_assert(std::is_base_of_v<RtcComponent, Super>, "'Acquisitor' requires 'RtcComponent'");
    static_assert(not is_base_of_template_v<Runnable, Super>, "'Acquisitor' excludes 'Runnable'");
    static_assert(not is_base_of_template_v<Loopaware, Super>, "'Acquisitor' excludes 'Loopaware'");
 
    class BizLogicIf : public Super::BizLogicIf {
    public:
        virtual void ActivityPreparing(StopToken st, const std::string& id) {};
        virtual void ActivityAcquiring(StopToken st) {};
        virtual DaqStopReply ActivityFinalising(StopToken st) {
            return {};
        };
        virtual void ActivityAborting(StopToken st) {};
        virtual void ActivityRecovering(StopToken st) {};
    };
 
    class InputStage : public Super::InputStage {
    public:
        using Super::InputStage::InputStage;
 
        void Start() override {
            Super::InputStage::Start();
 
            AcquisitorCmdsImpl::Register(this->m_replier, this->m_engine);
        }
    };
 
    class OutputStage : public Super::OutputStage {
    public:
        OutputStage(StateMachineEngine& engine, BizLogicIf& bl) : Super::OutputStage(engine, bl) {
            // Handlers ###################################################################
 
            engine.RegisterRejectHandler(
                events::StartDaq::ID, [](const rad::AnyEvent& event, const std::string& state) {
                    if (auto req = rad::GetPayloadNothrow<events::StartDaq>(event); req) {
                        const auto& id = req->GetRequestPayload();
                        req->SetException(DaqRequestRejected{events::StartDaq::ID, id, state});
                    }
                });
 
            engine.RegisterRejectHandler(
                events::StopDaq::ID, [](const rad::AnyEvent& event, const std::string& state) {
                    if (auto req = rad::GetPayloadNothrow<events::StopDaq>(event); req) {
                        const auto& id = req->GetRequestPayload();
                        req->SetException(DaqRequestRejected{events::StopDaq::ID, id, state});
                    }
                });
 
            engine.RegisterRejectHandler(
                events::AbortDaq::ID, [](const rad::AnyEvent& event, const std::string& state) {
                    if (auto req = rad::GetPayloadNothrow<events::AbortDaq>(event); req) {
                        const auto& id = req->GetRequestPayload();
                        req->SetException(DaqRequestRejected{events::AbortDaq::ID, id, state});
                    }
                });
 
            engine.RegisterRejectHandler(
                events::GetDaqStatus::ID, [](const rad::AnyEvent& event, const std::string& state) {
                    if (auto req = rad::GetPayloadNothrow<events::GetDaqStatus>(event); req) {
                        const auto& id = req->GetRequestPayload();
                        req->SetException(DaqRequestRejected{events::GetDaqStatus::ID, id, state});
                    }
                });
 
            // No Disable in states ###############################################################
 
            this->m_no_disable_in_states.push_back("On::Operational::Preparing");
            this->m_no_disable_in_states.push_back("On::Operational::Acquiring");
            this->m_no_disable_in_states.push_back("On::Operational::Finalising");
            this->m_no_disable_in_states.push_back("On::Operational::Aborting");
            this->m_no_disable_in_states.push_back("On::Operational::Recovering");
            this->m_no_disable_in_states.push_back("On::Operational::Error");
 
            // No Update in states ################################################################
 
            this->m_no_update_in_states.push_back("On::Operational::Preparing");
            this->m_no_update_in_states.push_back("On::Operational::Acquiring");
            this->m_no_update_in_states.push_back("On::Operational::Finalising");
            this->m_no_update_in_states.push_back("On::Operational::Aborting");
            this->m_no_update_in_states.push_back("On::Operational::Recovering");
            this->m_no_update_in_states.push_back("On::Operational::Error");
 
            // ClearHistory #####################################################################
 
            engine.RegisterAction("ActionClearHistory",
                                  [this](auto c) { m_acquisition_history.clear(); });
 
            // StartDaq #####################################################################
 
            engine.RegisterGuard("GuardPreparingAllowed", [this](auto c) {
                auto req = GetPayloadNothrow<events::StartDaq>(c);
                auto id = req->GetRequestPayload();
// #define FEATURE_REQUIRE_UNIQUE_ID_FOR_NEW_SESSION
#ifdef FEATURE_REQUIRE_UNIQUE_ID_FOR_NEW_SESSION
                return m_acquisition_history.count(id) == 0;
#else
                    return true;
#endif
            });
 
            engine.RegisterAction("ActionPreparingEntry", [this](auto c) {
                m_tmp_start_request = GetPayloadNothrow<events::StartDaq>(c);
                m_active_id = m_tmp_start_request->GetRequestPayload();
                std::string id = m_active_id.value();
 
                DaqStatus status;
                status.setId(id);
                status.setState(metadaqif::Acquiring);
                status.setTimestamp(GetCurrentTime());
                m_acquisition_history[id] = status;
            });
 
            engine.RegisterAction("ActionPreparingDone", [this](auto c) {
                if (m_tmp_start_request) {
                    std::string id = m_active_id.value();
                    m_tmp_start_request->SetReplyValue(std::make_shared<DaqReply>(id));
                    m_tmp_start_request = nullptr;
                }
            });
 
            engine.RegisterAction("ActionPreparingFailed", [this](auto c) {
                if (m_tmp_start_request) {
                    std::string id = m_active_id.value();
                    auto eptr = GetPayloadNothrow<events::Error>(c);
                    std::string msg = NestedExceptionPrinter(std::move(*eptr)).Str();
                    m_acquisition_history.at(id).setState(metadaqif::Failed);
                    m_acquisition_history.at(id).setTimestamp(GetCurrentTime());
                    m_acquisition_history.at(id).setMessage(msg);
                    m_tmp_start_request->SetException(metadaqif::DaqException(id, msg));
                    m_tmp_start_request = nullptr;
                }
            });
 
            // StopDaq #####################################################################
 
            engine.RegisterGuard("GuardFinalisingAllowed", [this](auto c) {
                auto req = GetPayloadNothrow<events::StopDaq>(c);
                auto id = req->GetRequestPayload();
                return m_active_id.value() == id;
            });
 
            engine.RegisterAction("ActionFinalisingEntry", [this](auto c) {
                m_tmp_stop_request = GetPayloadNothrow<events::StopDaq>(c);
            });
 
            engine.RegisterAction("ActionFinalisingDone", [this](auto c) {
                if (m_tmp_stop_request) {
                    std::string id = m_active_id.value();
                    m_acquisition_history.at(id).setState(metadaqif::Succeeded);
                    m_acquisition_history.at(id).setTimestamp(GetCurrentTime());
 
                    m_tmp_stop_request->SetReplyValue(
                        std::make_shared<DaqStopReply>(m_acquisition_history.at(id).getId(),
                                                       m_acquisition_history.at(id).getFiles(),
                                                       m_acquisition_history.at(id).getKeywords()));
                    m_tmp_stop_request = nullptr;
                }
            });
 
            engine.RegisterAction("ActionFinalisingFailed", [this](auto c) {
                if (m_tmp_stop_request) {
                    std::string id = m_active_id.value();
                    auto eptr = GetPayloadNothrow<events::Error>(c);
                    std::string msg = NestedExceptionPrinter(std::move(*eptr)).Str();
                    m_acquisition_history.at(id).setState(metadaqif::Failed);
                    m_acquisition_history.at(id).setTimestamp(GetCurrentTime());
                    m_acquisition_history.at(id).setMessage(msg);
                    m_tmp_stop_request->SetException(metadaqif::DaqException(id, msg));
                    m_tmp_stop_request = nullptr;
                }
            });
 
            // AbortDaq #####################################################################
 
            engine.RegisterGuard("GuardAbortingAllowed", [this](auto c) {
                auto req = GetPayloadNothrow<events::AbortDaq>(c);
                auto id = req->GetRequestPayload();
                return m_active_id.value() == id;
            });
 
            engine.RegisterAction("ActionAbortingEntry", [this](auto c) {
                m_tmp_abort_request = GetPayloadNothrow<events::AbortDaq>(c);
            });
 
            engine.RegisterAction("ActionAbortingDone", [this](auto c) {
                if (m_tmp_abort_request) {
                    std::string id = this->m_active_id.value();
                    m_acquisition_history.at(id).setState(metadaqif::Aborted);
                    m_acquisition_history.at(id).setTimestamp(GetCurrentTime());
                    m_tmp_abort_request->SetReplyValue(std::make_shared<DaqReply>(id));
                    m_tmp_abort_request = nullptr;
                }
            });
 
            // GetDaqStatus #####################################################################
 
            engine.RegisterAction("ActionGetDaqStatus", [this](auto c) {
                auto request = GetPayloadNothrow<events::GetDaqStatus>(c);
                auto id = request->GetRequestPayload();
 
                if (m_acquisition_history.count(id)) {
                    request->SetReplyValue(std::move(m_acquisition_history.at(id).clone()));
                } else {
                    auto status = std::make_shared<DaqStatus>();
                    status->setId(id);
                    status->setState(metadaqif::NotStarted);
                    status->setTimestamp(0);
                    request->SetReplyValue(status);
                }
            });
 
            // Recover #####################################################################
 
            engine.RegisterAction("ActionRecoveringEntry", [this](auto c) {
                std::string id = m_active_id.value();
                auto eptr = GetPayloadNothrow<events::Error>(c);
                std::string msg = NestedExceptionPrinter(std::move(*eptr)).Str();
                m_acquisition_history.at(id).setState(metadaqif::Failed);
                m_acquisition_history.at(id).setTimestamp(GetCurrentTime());
                m_acquisition_history.at(id).setMessage(msg);
            });
 
            // Activities #####################################################################
 
            engine.RegisterActivity(
                "ActivityPreparing",
                [this](StopToken stop_token) {
                    std::string id = m_active_id.value();
                    static_cast<BizLogicIf&>(this->m_logic).ActivityPreparing(stop_token, id);
                },
                this->m_success_handler,
                this->m_error_handler);
 
            engine.RegisterActivity(
                "ActivityFinalising",
                [this](StopToken stop_token) {
                    std::string id = m_active_id.value();
                    auto result =
                        static_cast<BizLogicIf&>(this->m_logic).ActivityFinalising(stop_token);
                    m_acquisition_history.at(id).setId(result.getId());
                    m_acquisition_history.at(id).setFiles(result.getFiles());
                    m_acquisition_history.at(id).setKeywords(result.getKeywords());
                },
                this->m_success_handler,
                this->m_error_handler);
 
            engine.RegisterActivity(
                "ActivityAcquiring",
                [this](StopToken stop_token) {
                    static_cast<BizLogicIf&>(this->m_logic).ActivityAcquiring(stop_token);
                },
                nullptr,
                this->m_error_handler);
 
            engine.RegisterActivity(
                "ActivityAborting",
                [this](StopToken stop_token) {
                    static_cast<BizLogicIf&>(this->m_logic).ActivityAborting(stop_token);
                },
                this->m_success_handler,
                this->m_error_handler);
 
            engine.RegisterActivity(
                "ActivityRecovering",
                [this](StopToken stop_token) {
                    static_cast<BizLogicIf&>(this->m_logic).ActivityRecovering(stop_token);
                },
                this->m_success_handler,
                this->m_error_handler);
        }
 
    private:
        double GetCurrentTime() {
            using namespace std::chrono;
            auto t = high_resolution_clock::now();
            return duration_cast<nanoseconds>(t.time_since_epoch()).count();
        }
 
        std::shared_ptr<rad::cii::Request<std::shared_ptr<metadaqif::DaqReply>, std::string>>
            m_tmp_start_request;
        std::shared_ptr<rad::cii::Request<std::shared_ptr<metadaqif::DaqStopReply>, std::string>>
            m_tmp_stop_request;
        std::shared_ptr<rad::cii::Request<std::shared_ptr<metadaqif::DaqReply>, std::string>>
            m_tmp_abort_request;
        std::map<std::string, DaqStatus> m_acquisition_history;
 
        std::optional<std::string> m_active_id;
    };
 
    struct ModelBuilder : public Super::ModelBuilder {
    public:
        ModelBuilder(StateMachineEngine& engine) : Super::ModelBuilder(engine) {
            // clang-format off
            this->mm.ModStateType("On::Operational", Parallel);
 
            const std::string parent_region = "On::Operational:";
 
            this->mm.AddState(Composite, parent_region,                  "On::Operational");
            this->mm.AddState(Initial,   "On::Operational::Initial",     parent_region);
            this->mm.AddState(Simple,    "On::Operational::Idle",        parent_region);
            this->mm.AddState(Simple,    "On::Operational::Acquiring",   parent_region,   "ActivityAcquiring");
            this->mm.AddState(Simple,    "On::Operational::Recovering",  parent_region,   "ActivityRecovering",  "ActionRecoveringEntry");
            this->mm.AddState(Simple,    "On::Operational::Preparing",   parent_region,   "ActivityPreparing",   "ActionPreparingEntry");
            this->mm.AddState(Simple,    "On::Operational::Finalising",  parent_region,   "ActivityFinalising",  "ActionFinalisingEntry");
            this->mm.AddState(Simple,    "On::Operational::Aborting",    parent_region,   "ActivityAborting",    "ActionAbortingEntry");
 
            this->mm.AddTrans("On::Operational::Initial",    "On::Operational::Idle",       "",                    "",  "ActionClearHistory");
            this->mm.AddTrans(parent_region,                 "",                            "events.GetDaqStatus", "",  "ActionGetDaqStatus");
            this->mm.AddTrans("On::Operational::Idle",       "On::Operational::Preparing",  "events.StartDaq",     "GuardPreparingAllowed");
            this->mm.AddTrans("On::Operational::Preparing",  "On::Operational::Acquiring",  "events.Done",         "",   "ActionPreparingDone");
            this->mm.AddTrans("On::Operational::Preparing",  "On::Operational::Idle",       "events.Error",        "",   "ActionPreparingFailed");
            this->mm.AddTrans("On::Operational::Acquiring",  "On::Operational::Finalising", "events.StopDaq",      "GuardFinalisingAllowed");
            this->mm.AddTrans("On::Operational::Finalising", "On::Operational::Idle",       "events.Done",         "",   "ActionFinalisingDone");
            this->mm.AddTrans("On::Operational::Finalising", "On::Operational::Idle",       "events.Error",        "",   "ActionFinalisingFailed");
            this->mm.AddTrans("On::Operational::Acquiring",  "On::Operational::Aborting",   "events.AbortDaq",     "GuardAbortingAllowed");
            this->mm.AddTrans("On::Operational::Aborting",   "On::Operational::Idle",       "events.Done",         "",   "ActionAbortingDone");
            this->mm.AddTrans("On::Operational::Acquiring",  "On::Operational::Recovering", "events.Error");
            this->mm.AddTrans("On::Operational::Recovering", "On::Operational::Idle",       "events.Done");
            // clang-format on
        }
    };
};
 
}  // namespace rtctk::metadataCollector
 
#endif  // RTCTK_COMPONENTFRAMEWORK_ACQUISITOR_HPP