application.hpp

Go to the documentation of this file.

 
#ifndef CCF_CONTROL_APPLICATION_HPP_
#define CCF_CONTROL_APPLICATION_HPP_
 
#include <mal/Mal.hpp>
 
#include <rad/mal/replier.hpp>
#include <rad/dbAdapterRedis.hpp>
 
#include <Recif.hpp>
 
#include <ifw/ccf/common/base.hpp>
#include <ifw/ccf/common/frameQueue.hpp>
#include <ifw/ccf/common/recipeBase.hpp>
#include <ifw/ccf/common/comAdptBase.hpp>
#include <ifw/ccf/common/execAdptBase.hpp>
#include <ifw/ccf/common/monAdptBase.hpp>
#include <ifw/ccf/common/appBase.hpp>
 
#include <ifw/ccf/control/config.hpp>
#include <ifw/ccf/control/dataContext.hpp>
#include <ifw/ccf/control/actionMgr.hpp>
#include <ifw/ccf/control/stdActions.hpp>
#include <ifw/ccf/control/stdCmdsImpl.hpp>
#include <ifw/ccf/control/dcsCmdsImpl.hpp>
#include <ifw/ccf/control/recCmdsImpl.hpp>
#include <ifw/ccf/control/comAdptSim.hpp>
#include <ifw/ccf/control/monitorThread.hpp>
#include <ifw/ccf/control/acqThread.hpp>
#include <ifw/ccf/control/procThread.hpp>
#include <ifw/ccf/control/pubThread.hpp>
#include <ifw/ccf/control/recWaitRequest.hpp>
 
 
namespace ifw::ccf::control {

   // @{
   const std::string MON_THR_TAG    = "MonThr";
   const std::string ACQ_THR_TAG    = "AcqThr";
   const std::string INPUT_Q_TAG    = "InputQueue";
   // Used as a fmt format string; fmt (v11) format strings are consteval under
   // C++20, so this must be usable in a constant expression — a const
   // std::string is not, a constexpr std::string_view is.
   constexpr std::string_view OUTPUT_Q_TAG = "OutputQueue::{}";
   // @}

   // @{
   const std::string THREAD_SIG_TERM           = "SigTerm";
   const std::string THREAD_SIG_SETUP          = "SigSetup";
   const std::string THREAD_SIG_NEW_DATA       = "SigNewData";
   const std::string THREAD_SIG_PUBLISH        = "SigPublish";
   const std::string THREAD_SIG_START          = "SigStart";
   // @}
 
   //const std::string THREAD_SYNC_ACK_OK      = "SigAckOk";
   //const std::string THREAD_SYNC_ACK_ERROR   = "SigAckError";
   //const std::string THREAD_SYNC_UNKNOWN     = "SigUnknown";

   // @{
   const std::string ACTIVE_REC_MUTEX          = "ActiveRecMutex";
   const std::string REC_HISTORY_MUTEX         = "RecHistoryMutex";
   const std::string MUTEX_REC_WAIT_REQUESTS   = "RecWaitRequests";
   const std::string MUTEX_TELEMETRY           = "Telemetry";
   const std::string MUTEX_TELEMETRY_CACHE     = "TelemetryCache";
   // @}

   template <typename CMD_TYPE, typename EX_TYPE>
   void CcfRequestRejectHandler(const rad::AnyEvent& event, const std::string& state) {
     LOG4CPLUS_TRACE_METHOD(Logger(), __PRETTY_FUNCTION__);
     auto req = rad::GetPayloadNothrow<CMD_TYPE>(event);
     if (req == nullptr) {
       LOG4CPLUS_ERROR(Logger(), event.GetId() << " has no associated request!");
       return;
     }
     req->SetException(EX_TYPE("Request rejected in state " + state, 0));
   }
 

class Application: public ifw::ccf::common::AppBase {
 public:

    static Application* s_app_instance;

    static Application& Instance();
 
    virtual ~Application();

    elt::mal::Mal& GetMal();

    int Execute(int argc, char* argv[]);

    void CreateThreads();

    void LoadInitSetup();

    void HandleSetup();

    void Dismantle();
 
    // Returns true if image acquisition is active.
    bool AcquisitionActive();

    void StartAcquistion(bool force = false, bool exec_com_adapter_start = true);

    void StopAcquistion();

    void CheckStopAcquistion();

    void IncrementFrameCounter();

    std::string GetState();

    void SetRequestStatus(const std::string& request,
                          const std::string& status);

    std::string GetRequestStatus(const std::string& request) const;

    std::string ListAdapters();

    template <class COM_ADPT_TYPE, class SIM_COM_ADPT_TYPE>
    void RegisterComAdapters(const COM_ADPT_TYPE& com_adapter_factory_obj,
                             const SIM_COM_ADPT_TYPE& sim_com_adapter_factory_obj) {
      LOG4CPLUS_TRACE_METHOD(Logger(), __PRETTY_FUNCTION__);
      // if (m_com_adapter != nullptr) {
      //    delete m_com_adapter;
      // }
      // if (m_sim_com_adapter != nullptr) {
      //    delete m_sim_com_adapter;
      // }
      m_com_adapter = std::make_unique<COM_ADPT_TYPE>();
      ifw::ccf::AssertPtr(m_com_adapter.get(), "Communication Adapter", IFWLOC);
      m_sim_com_adapter = std::make_unique<SIM_COM_ADPT_TYPE>();
      ifw::ccf::AssertPtr(m_sim_com_adapter.get(), "Simulation Communication Adapter", IFWLOC);
    }

    template <class MON_ADPT_TYPE>
    void RegisterMonAdapter(const MON_ADPT_TYPE& mon_adapter_factory_obj) {
      LOG4CPLUS_TRACE_METHOD(Logger(), __PRETTY_FUNCTION__);
      m_mon_adapter = std::make_unique<MON_ADPT_TYPE>();
      ifw::ccf::AssertPtr(m_com_adapter.get(), "Monitoring Adapter", IFWLOC);
    }

    template <class EXEC_ADPT_TYPE>
    void RegisterExecAdapter(const EXEC_ADPT_TYPE& exec_adapter_factory_obj) {
     LOG4CPLUS_TRACE_METHOD(Logger(), __PRETTY_FUNCTION__);
     m_exec_adapter = std::make_unique<EXEC_ADPT_TYPE>();
     ifw::ccf::AssertPtr(m_exec_adapter.get(), "Execution Adapter", IFWLOC);
    }

    ifw::ccf::common::ComAdptBase& GetComAdapter();

    ifw::ccf::common::MonAdptBase& GetMonAdapter();

    ifw::ccf::common::ExecAdptBase& GetExecAdapter();

    void ConnectDevice();

    void DisconnectDevice();

    const std::vector<std::string>& GetProcThreadNames() const;

    const std::vector<std::string>& GetPubThreadNames(const std::string& proc_thread_name) const;

    const std::list<PubThread*>& GetPubThreads() const;

    ProcThread& GetProcThread(const std::string& proc_thread_name);

    bool IsProcThread(const std::string& thread_name);

    bool IsPubThread(const std::string& thread_name);
 
    Application(const Application&) = delete;              
    Application& operator=(const Application&) = delete;   

   // Recording handling

   void UpdateRecordingStatus();

   void SetActiveRecStatusObj(std::shared_ptr<recif::RecStatus>& active_rec);

   void UpdateActiveRecStatusObj(const ifw::ccf::common::PublisherStatus& sum_rec_pub_status);

   void UpdateRecStatusInDb();

   void GetActiveRecStatus(std::shared_ptr<recif::RecStatus>& active_rec) const;
 
   int GetFramesRemainingActiveRecStatus() const;

   void MoveActiveRecStatusToHistory(const ifw::ccf::PubStatus status,
                                     std::shared_ptr<recif::RecStatus>& active_rec);

   void SetRecordingStatus(const ifw::ccf::PubStatus status);

   void GetRecStatus(const std::string& rec_id,
                     std::shared_ptr<recif::RecStatus>& active_rec) const;

   void GetMostRecentRecStatus(std::shared_ptr<recif::RecStatus>& active_rec) const;

   void AddRecWaitRequest(std::shared_ptr<RecWaitRequest>& req_wait_req);

   void CheckRecWaitRequests(const bool recording_completed=false);

   void CleanUpRecWaitRequests();

   void AcqAutoStartedAtRecStart();

   void UpdateControlSetupPars();

   void ActivateRecording();
 
   void HandleTelemetry(const double time_now);
 
   std::map<std::string, TelemetryCacheEntry>& TelemetryCache();
 
 private:
    Application();
 
    int8_t                               m_simulation{-1};
    std::unique_ptr<rad::cii::Replier>   m_mal_replier;
    std::shared_ptr<elt::mal::Mal>       m_mal{nullptr};
    std::unique_ptr<DataContext>         m_data_ctx;
    ActionMgr                            m_action_mgr;
    //StdActions*                          m_actions_std;
    bool                                 m_dismantle_invoked{false};
 
    std::map<std::string, std::string>   m_request_status_map;
 
    // TODO: Handle this in the ifw::ccf::common::ComAdptBase class.
    // See EICSSW-2138.
    std::unique_ptr<ifw::ccf::common::ComAdptBase>       m_com_adapter{nullptr};
    std::unique_ptr<ifw::ccf::common::ComAdptBase>       m_sim_com_adapter{nullptr};
 
    std::unique_ptr<ifw::ccf::common::MonAdptBase>       m_mon_adapter;
 
    std::unique_ptr<ifw::ccf::common::ExecAdptBase>      m_exec_adapter;
 
    std::map<std::string, std::shared_ptr<ifw::ccf::common::FrameQueue>> m_frame_queues;
 
    std::vector<std::string>                        m_proc_thread_names;
    std::map<std::string, std::vector<std::string>> m_proc_to_pub_thread_names;
 
    std::map<std::string, bool>                     m_pub_thread_names;
 
    // For easy access to the Processing Threads. The threads are handled by the MPTK.
    std::list<ProcThread*>                          m_proc_threads;
 
    // For easy access to the Publisher Thread. The threads are handled by the MPTK.
    std::list<PubThread*>                           m_pub_threads;
 
    // Acquisition control.
    ifw::ccf::ExpoMode                                   m_acq_setup_expo_mode{ifw::ccf::ExpoMode::INACTIVE};
    double                                          m_acq_setup_expo_time{0.0};
    int64_t                                         m_acq_setup_nb_of_frames{0};
    int64_t                                         m_acq_status_frame_count{0};
    double                                          m_acq_end_time{UINT_MAX};
 
    // Recording session control.
    std::shared_ptr<recif::RecStatus>               m_active_recording;
    int64_t                                         m_exp_nb_of_frames;
    int32_t                                         m_max_sz_rec_history{1000};
    int32_t                                         m_rec_history_exp{INT_MAX};
    bool                                            m_cfg_auto_stop_acq_after_rec{false};
    bool                                            m_acq_auto_started_at_rec_start{false};
    // Maps "<rec id>" into the set of associated Recording Sessions.
    std::map<std::string, std::shared_ptr<recif::RecStatus>>  m_rec_history;
    std::vector<std::string>                                  m_rec_id_history;
 
    std::map<std::string, std::string>                        m_file_stored_to_pub_id;
 
    // Pending ReqWait Requests.
    std::map<std::string, std::shared_ptr<RecWaitRequest>>    m_rec_wait_requests;
 
    // Telemetry.
    void InitTelemetry();
    bool                                                m_tel_enabled{false};
    double                                              m_tel_period{0};
    bool                                                m_disable_during_acq{true};
    std::map<std::string, ifw::ccf::control::TelemetryNode>  m_tel_nodes;
    std::vector<std::string>                            m_tel_names;
    double                                              m_time_last_tel_sampling_exec{0};
    std::map<std::string, TelemetryCacheEntry>          m_tel_cache;
    // Per-node last-emit timestamp for throttling repeated "no usable value"
    // WARN messages emitted by HandleTelemetry. See TEL_WARN_THROTTLE_PERIOD.
    std::map<std::string, double>                       m_tel_warn_last_emit;
};
 
 
inline Application& App() {
   return Application::Instance();
}
 
} // namespace ifw::ccf::control
 
#endif // CCFCONTROL_APPLICATION_HPP_