Sophiar2/src/utility/variable_utility2.cpp

#include "core/basic_obj_types.hpp"
#include "core/global_defs.h"
#include "core/sophiar_pool.h"
#include "core/timestamp_helper.hpp"
#include "core/tristate_obj.h"
#include "utility/config_utility.hpp"
#include "utility/versatile_buffer2.hpp"

#include <spdlog/spdlog.h>

#include <fstream>
#include <vector>

using boost::asio::awaitable;

namespace sophiar {

    class multi_variable_utility_base : public tristate_obj {
    public:
        awaitable<void> on_stop() noexcept override {
            for (auto cb: cb_pool) {
                UNREGISTER_CALLBACK(cb);
            }
            cb_pool.clear();
            co_return;
        }

    protected:
        void register_variables(const nlohmann::json &config) {
            if (config.contains("variable_name_list")) {
                ENSURE_ARRAY("variable_name_list");
                for (auto &item: config["variable_name_list"]) {
                    assert(item.is_string());
                    auto var_name = item.get<std::string>();
                    cb_pool.push_back(create_callback(var_name));
                }
            } else {
                auto var_name = LOAD_STRING_ITEM("variable_name");
                cb_pool.push_back(create_callback(var_name));
            }
        }

        virtual sophiar_pool::callback_token_type create_callback(const std::string &var_name) = 0;

    private:
        std::vector<sophiar_pool::callback_token_type> cb_pool;
    };

    class variable_debug_watcher : public multi_variable_utility_base {
    public:
        DEFAULT_NEW_INSTANCE(variable_debug_watcher)

    protected:
        awaitable<bool> on_start(const nlohmann::json &config) noexcept override {
            min_interval = TRY_LOAD_FLOAT_ITEM("minimum_interval_ms", 0) * 1000; // ms -> us
            register_variables(config);
            co_return true;
        }

    private:
        timestamp_type min_interval = 0;

        struct callback_info {
            string_writer buffer;
            timestamp_type last_print_ts = 0;
        };

        sophiar_pool::callback_token_type create_callback(const std::string &var_name) override {
            auto var_info = global_sophiar_pool->query_variable_information(var_name);
            auto cb_info = new callback_info;
            auto cb_func = [=, this]() {
                auto ts = QUERY_VARIABLE_TS(var_info.index);
                if (ts - cb_info->last_print_ts < min_interval) return;
                cb_info->last_print_ts = ts;
                if (!raw_pointer_is_valid(var_info.placeholder, var_info.type_index)) {
                    SPDLOG_DEBUG("{} is empty.", var_name);
                } else {
                    raw_pointer_write_to(cb_info->buffer, var_info.placeholder, var_info.type_index);
                    SPDLOG_DEBUG("{} = {}", var_name, cb_info->buffer.get_string_and_reset());
                }
            };
            auto exit_func = [=]() { delete cb_info; };
            auto token = REGISTER_CALLBACK2(cb_func, exit_func);
            ATTACH_CALLBACK(var_info.index, token);
            return token;
        }
    };

    class variable_validity_watcher : public multi_variable_utility_base {
    public:
        DEFAULT_NEW_INSTANCE(variable_validity_watcher)

    protected:
        awaitable<bool> on_start(const nlohmann::json &config) noexcept override {
            register_variables(config);
            co_return true;
        }

    private:
        sophiar_pool::callback_token_type create_callback(const std::string &var_name) override {
            auto var_info = global_sophiar_pool->query_variable_information(var_name);
            bool is_valid = raw_pointer_is_valid(var_info.placeholder, var_info.type_index);
            auto cb_func = [=]() mutable {
                if (raw_pointer_is_valid(var_info.placeholder, var_info.type_index)) {
                    if (is_valid)[[likely]] return;
                    SPDLOG_DEBUG("{} becomes valid.", var_name);
                    is_valid = true;
                } else {
                    if (!is_valid)[[likely]] return;
                    SPDLOG_DEBUG("{} becomes invalid.", var_name);
                    is_valid = false;
                }
            };
            auto token = REGISTER_CALLBACK(cb_func);
            ATTACH_CALLBACK(var_info.index, token);
            return token;
        }
    };

    class variable_recorder : public tristate_obj {
    public:
        variable_recorder()
                : buffer(string_writer(",")) {
        }

        DEFAULT_NEW_INSTANCE(variable_recorder)

        awaitable<bool> on_init(const nlohmann::json &config) noexcept override {
            auto var_name = LOAD_STRING_ITEM("variable_name");
            auto var_info = global_sophiar_pool->query_variable_information(var_name);
            var_index = var_info.index;
            auto save_file_path = LOAD_STRING_ITEM("save_file");
            ofs = std::ofstream(save_file_path, std::ofstream::out);
            assert(ofs.is_open());
            auto cb_func = [=, this]() mutable {
                auto ts = QUERY_VARIABLE_TS(var_index);
                buffer << (static_cast<double>(ts) / 1000.0); // us -> ms
                raw_pointer_write_to(buffer, var_info.placeholder, var_info.type_index);
                ofs << buffer.get_string_and_reset() << std::endl;
            };
            assert(cb_token == nullptr);
            cb_token = REGISTER_CALLBACK(cb_func);
            co_return true;
        }

        boost::asio::awaitable<bool> on_start(const nlohmann::json &) noexcept override {
            assert(ath_token == nullptr);
            ath_token = ATTACH_CALLBACK(var_index, cb_token);
            co_return true;
        }

        boost::asio::awaitable<void> on_stop() noexcept override {
            DETACH_CALLBACK(ath_token);
            ath_token = nullptr;
            co_return;
        }

        boost::asio::awaitable<void> on_reset() noexcept override {
            UNREGISTER_CALLBACK(cb_token);
            cb_token = nullptr;
            co_return;
        }

    private:
        string_writer buffer;
        std::ofstream ofs;
        variable_index_type var_index = -1;
        sophiar_pool::callback_token_type cb_token = nullptr;
        sophiar_pool::attach_token_type ath_token = nullptr;
    };

    void register_variable_utility() {
        REGISTER_TYPE2(variable_debug_watcher, "variable_watcher");
        REGISTER_TYPE2(variable_validity_watcher, "variable_validity_watcher");
        REGISTER_TYPE2(variable_recorder, "variable_recorder");
    }

}