dataPointRecordingUnit.hpp

Go to the documentation of this file.

 
#ifndef RTCTK_COMPONENTFRAMEWORK_DATAPOINTRECORDINGUNIT_HPP
#define RTCTK_COMPONENTFRAMEWORK_DATAPOINTRECORDINGUNIT_HPP
 
#include "ecs/taiClock.hpp"
#include "rtctk/componentFramework/dataRecorder.hpp"
#include "rtctk/componentFramework/exceptions.hpp"
#include "rtctk/componentFramework/fitsDataRecorder.hpp"
#include "rtctk/componentFramework/recordingInfo.hpp"
#include "rtctk/componentFramework/recordingUnit.hpp"
 
#include <numapp/numapolicies.hpp>
#include <numapp/thread.hpp>
 
#include <cstdint>
#include <exception>
#include <mutex>
#include <string>
#include <string_view>
#include <thread>
 
namespace rtctk::componentFramework {
 
using namespace std::string_view_literals;
 
template <typename DpType,
          typename OutputStage =
              FitsRecorder<ecs::TaiClock::time_point::rep,
                           std::conditional_t<IsSpanConvertibleV<DpType>, AsSpan<DpType>, DpType>>>
class DataPointRecordingUnit : public RecordingUnit {
public:
    // we are using a bit-mask to configure when to capture data
    using CaptureMask = uint32_t;
    static constexpr uint32_t CAPTURE_ON_START = 1;
    static constexpr uint32_t CAPTURE_ON_STOP = 2;
    static constexpr uint32_t CAPTURE_ON_CHANGE = 4;
 
    using OutputDpType = std::conditional_t<IsSpanConvertibleV<DpType>, AsSpan<DpType>, DpType>;
 
    DataPointRecordingUnit(std::string const& comp_id,
                           std::string const& unit_id,
                           RuntimeRepoIf& rtr,
                           OldbIf& oldb,
                           DataPointPath const& dp_path)
        : RecordingUnit(comp_id, unit_id, rtr, oldb)
        , m_rtr(rtr)
        , m_dp_path(dp_path)
        , m_dp_buffer()
        , m_output(COLUMNS) {
        DataPointPath path = DataPointPath(comp_id) / DataPointPath("static") /
                             DataPointPath("rec_units") / DataPointPath(unit_id) /
                             DataPointPath("capture_mask");
        if (rtr.DataPointExists(path)) {
            m_capture_mask = rtr.GetDataPoint<int32_t>(path);
        } else {
            m_capture_mask = CAPTURE_ON_START + CAPTURE_ON_CHANGE + CAPTURE_ON_STOP;
        }
 
        Update();
    }
 
    virtual ~DataPointRecordingUnit() {
        m_stop = true;
        if (m_process_thread.joinable()) {
            m_process_thread.join();
        }
        SetStopped();
    }
    void Prepare(const std::filesystem::path& file_path) override {
        SetState(State::PREPARING,
                 State::STOPPED,
                 "Error in transition to PREPARING, DataPointRecordingUnit not in STOPPED State");
 
        m_file_path = file_path / (GetId() + ".fits");
 
        if (not m_rtr.DataPointExists(m_dp_path)) {
            SetFailed(nullptr);
            CII_THROW(InvalidSetting, "DataPoint does not exist");
        }
 
        if (m_capture_mask == 0) {
            SetFailed(nullptr);
            CII_THROW(InvalidSetting, "Invalid capture mask 0");
        }
 
        auto policies = numapp::NumaPolicies();
        m_process_thread = numapp::MakeThread(m_unit_id, policies, [&]() { return Process(); });
    }
    void Start() override {
        m_start = true;
    }
    std::optional<std::filesystem::path> Stop() override {
        m_stop = true;
        if (m_process_thread.joinable()) {
            m_process_thread.join();
        }
        auto file = m_file_path;
        m_file_path.reset();
        SetStopped();
        return file;
    }
 
private:
    void Process() {
        using namespace std::chrono_literals;
 
        m_output.Open(*m_file_path);
 
        if (m_capture_mask & CAPTURE_ON_CHANGE) {
            m_rtr.Subscribe(m_dp_path, m_dp_buffer, [this](auto& path, auto& value) {
                if (GetState() == State::RUNNING) {
                    try {
                        m_output.Write(AsTuple(value));
                    } catch (...) {
                        SetFailed(std::current_exception());
                        return;
                    }
                }
            });
        }
 
        SetState(State::IDLE,
                 State::PREPARING,
                 "Error in transition to IDLE, DataPointRecordingUnit not in PREPARING State");
 
        while (m_stop == false) {
            switch (GetState()) {
            case State::IDLE: {
                if (m_start) {
                    SetState(
                        State::WAITING,
                        State::IDLE,
                        "Error in transition to WAITING, DataPointRecordingUnit not in IDLE State");
                    break;
                }
                std::this_thread::sleep_for(1ms);
                break;
            }
            case State::WAITING: {
                if (not HasLeaders() or (HasLeaders() and HasFirstLeaderStarted())) {
                    if (m_capture_mask & CAPTURE_ON_START) {
                        m_rtr.ReadDataPoint(m_dp_path, m_dp_buffer);
                        m_output.Write(AsTuple(m_dp_buffer));
                    }
 
                    SetState(State::RUNNING,
                             State::WAITING,
                             "Error in transition to RUNNING, DataPointRecordingUnit not in "
                             "WAITING State");
                    break;
                }
                std::this_thread::sleep_for(1ms);
                break;
            }
            case State::RUNNING: {
                if (HasLeaders() and HasLastLeaderFinished()) {
                    SetState(State::FINISHED,
                             State::RUNNING,
                             "Error in transition to FINISHED, DataPointRecordingUnit not in "
                             "RUNNING State");
                    break;
                }
                std::this_thread::sleep_for(1ms);
            }
            default: {
                std::this_thread::sleep_for(1ms);
            }
            }
        }
 
        if (m_capture_mask & CAPTURE_ON_CHANGE) {
            m_rtr.Unsubscribe(m_dp_path);
        }
 
        if (m_capture_mask & CAPTURE_ON_STOP) {
            m_rtr.ReadDataPoint(m_dp_path, m_dp_buffer);
            m_output.Write(AsTuple(m_dp_buffer));
        }
 
        m_output.Close();
        // State is set to stopped after join in Stop function
        m_start = false;
        m_stop = false;
        ResetLeaderStates();
    }
 
    static std::tuple<ecs::TaiClock::time_point::rep, OutputDpType> AsTuple(DpType const& data) {
        auto ts = ecs::TaiClock::now().time_since_epoch().count();
        if constexpr (IsSpanConvertibleV<DpType>) {
            return std::make_tuple(ts, ToSpan(data));
        } else {
            return std::make_tuple(ts, data);
        }
    }
 
    static constexpr typename OutputStage::ColumnDescription COLUMNS =
        typename OutputStage::ColumnDescription{{{"timestamp"sv, ""sv}, {"payload"sv, ""sv}}};
 
    RuntimeRepoIf& m_rtr;
 
    std::optional<std::filesystem::path> m_file_path;
 
    DataPointPath m_dp_path;
 
    DpType m_dp_buffer;
 
    CaptureMask m_capture_mask;
 
    OutputStage m_output;
 
    std::atomic<bool> m_start = false;
    std::atomic<bool> m_stop = false;
    std::thread m_process_thread;
};
 
}  // namespace rtctk::componentFramework
 
#endif  // RTCTK_COMPONENTFRAMEWORK_DATAPOINTRECORDINGUNIT_HPP