

#include "Local_Server/DSP/DSP_Config.h"
#include "../server/Global.h"
#include "Core/socket/ByteOrder.h"

#include <drogon/HttpAppFramework.h>
#include <drogon/PubSubService.h>
#include <drogon/WebSocketController.h>

#if!USE_PROCESS

// adsb_set_less_30Mhz_ddc_param_t set_less_30Mhz_ddc_param;
// adsb_set_less_30Mhz_fft_param_t set_less_30Mhz_fft_param;


decltype(&adsb_set_less_30Mhz_ddc_param) psc_set_less_30Mhz_ddc_param;
decltype(&adsb_set_less_30Mhz_fft_param) psc_set_less_30Mhz_fft_param;

#endif


using namespace drogon;
// ws://127.0.0.1:8848/chat











BaseLogger* dsp_logger = nullptr;
DSP_Config::~DSP_Config() = default;

void DSP_Config::close_dsp_device() const {

    auto r = Psc::try_load_function(lib, "adsb_close");
    if (!r) {
        Psc::fail_fast();
    }
    auto tmp_adsb_close = (decltype(&adsb_close)) r.value();
    std::ostringstream oss;
    oss << "stop_program ===== " << VAR_STR_1(stop_program) << std::endl;
    oss << "DSP_Config   adsb_close 开始关闭子库！" << std::endl;
    tmp_adsb_close();
    oss << "DSP_Config   adsb_close 关闭子库结束！" << std::endl;
    dsp_logger->c_debug("dsp", {}, oss.str());
}


void DSP_Config::server(Global *g) {

    auto       &svr = g->svr;
    auto       &api = g->api;
    std::string name = "dsp";

    drogon::app().registerWebSocketControllerRegex(R"(^/websocket/.*$)", "DSP_Ws_Mgr", {Get});

    ws_iq = Ws_Mgr::instance()->register_ws("/download_iq");
    ws_fft = Ws_Mgr::instance()->register_ws("/get_fft");


    svr.Post(api + svr.update + name, [this, g](HTTP_Param) {
        CHECK_JSON_PARAM
        init_update(&params);


#if USE_PROCESS
    adsb_set_less_30Mhz_ddc_param(this->center_freq, this->sample_rate, this->band_width, 1);
    adsb_set_less_30Mhz_fft_param(fft_win_type, fft_point_number, fft_period_ms, 1);
#else
    psc_set_less_30Mhz_ddc_param(this->center_freq, this->sample_rate, this->band_width, 1);
    psc_set_less_30Mhz_fft_param(fft_win_type, fft_point_number, fft_period_ms, 1);
#endif
        set_fft_point_number(this->fft_point_number);
        g->save();
        res->setBody(warp(to_json()).to_json_string());
    });
    svr.Post(api + "get_dsp_config", [this](HTTP_Param) { res->setBody(warp(to_json()).to_json_string()); });


    drogon::app().registerHandler(api + "start_record_iq_stream",
                                  [this](const drogon::HttpRequestPtr                          &request,
                                         std::function<void(const drogon::HttpResponsePtr &)> &&callback) {
                                      this->record_iq_stream_ing = true;
                                      auto res = drogon::HttpResponse::newHttpResponse();
                                      res->setBody("{}");
                                      callback(res);
                                  },
                                  {drogon::Post});


    svr.Post(api + "end_record_iq_stream", [this](HTTP_Param) {
        this->record_iq_stream_ing = false;
        res->setBody("{}");
    });

    svr.Post(api + "restart_device", [this](HTTP_Param) {
#if USE_PROCESS
        restart_device();
#endif
    });
}


void DSP_Config::set_fft_point_number(uint64_t new_fft_point_number, bool force) {
    fft_point_number = new_fft_point_number;
#if !USE_PROCESS
    std::lock_guard<std::mutex> lock(fft_mtx);
    if (!force && fft_data.size() == (new_fft_point_number * sizeof(short))) {
        return;
    }
    fft_data.resize(new_fft_point_number * sizeof(short));
#endif

}


#if USE_PROCESS
void DSP_Config::adsb_set_less_30Mhz_ddc_param(double center_freq, double sample_rate, double band_width, int sw) {
    JSON ret = JSON::object();
    ret.append(JSON("method", "adsb_set_less_30Mhz_ddc_param"));
    ret.append(JSON("center_freq", center_freq));
    ret.append(JSON("sample_rate", sample_rate));
    ret.append(JSON("band_width", band_width));
    ret.append(JSON("sw", sw));
    auto d = ret.to_json_string();
    json_func_call.write((uint8_t *) d.data(), d.size());
}

void DSP_Config::adsb_set_less_30Mhz_fft_param(UHD_FFTWin fft_win_type, size_t fft_point_number, size_t fft_period_ms,
                                               int sw) {
    JSON ret = JSON::object();
    ret.append(JSON("method", "adsb_set_less_30Mhz_fft_param"));
    ret.append(JSON("fft_win_type", fft_win_type));
    ret.append(JSON("fft_point_number", fft_point_number));
    ret.append(JSON("fft_period_ms", fft_period_ms));
    ret.append(JSON("sw", sw));
    auto d = ret.to_json_string();
    json_func_call.write((uint8_t *) d.data(), d.size());
}

void DSP_Config::restart_device() {
    JSON ret = JSON::object();
    ret.append(JSON("method", "restart_device"));
    auto d = ret.to_json_string();
    json_func_call.write((uint8_t *) d.data(), d.size());
}
#endif


#define USE_ERROR_CBK

void DSP_Config::init_env() {

    set_fft_point_number(fft_point_number, true);
    auto                   g = Global::instance();
    Detach_Thread_Manager &manager = g->thread_manager;

    auto handle_iq= [this](void* data) {
        uint8_t* pdata = (uint8_t*) data;
        auto wi = Ws_Mgr::instance();
        if (!wi) return;
        auto d = (UHD_Struct_KFFT *) (pdata);
        if (debug) {
            std::ostringstream oss;
            oss << "UHD_Struct_KFFT{" << mem2hex(std::string((char *) pdata, 64), true, " ") << std::endl;
            oss << "DataType{" << (int) d->DataType << "}  接收到数据:["
                << "] == " << mem2hex(std::string((char *) pdata + 64, 20), true, " ") << std::endl;
            server_logger->c_debug({}, {}, oss.str());
        }
        if (d->DataType == zd_ddc || d->DataType == kd_ddc) {
            auto len = d->UsefullLen;
            bool cross_the_boundary = len > 2048;
            if (cross_the_boundary || debug) {
                auto show_len = (size_t) std::min((decltype(len)) 20, len);

                std::ostringstream oss;
                oss << "iq 有效数据长度 【" << len << "】 有效数据： 【"
                          << std::string((char *) pdata + 64, show_len) << "】" << std::endl;
                server_logger->c_debug({}, {}, oss.str());
            }
            if (cross_the_boundary) {
                std::ostringstream oss;
                oss << LOG_POS << "ERROR: ddc iq 回调返回的有效长度越界！" << std::endl;
                server_logger->c_debug({}, {}, oss.str());
                return;
            }
            if (record_iq_stream_ing) {
                ws_iq->push_data((const char *) pdata , 2112);
            }
        } else if (d->DataType == kd_fft) {
#if !USE_PROCESS
            auto                        idx = d->Param.FFT.SubFrame;
            auto max = fft_point_number/1024;
            if (idx >= max) {
                std::ostringstream oss;
                oss << "收到了不合法kd_fft  Param.FFT.SubFrame  " << idx << std::endl;
                server_logger->c_debug({}, {}, oss.str());
                return;
            }


            std::lock_guard<std::mutex> lock(fft_mtx);
            auto                        pos = fft_data.data() + idx * 2048;
            memcpy(pos, pdata + 64, 2048);
#endif
        }
    };

#if USE_PROCESS
    manager.test_and_start_thread("读取iq进程数据线程", [this, handle_iq](std::atomic<bool> &running) {
        while (running) {
            uint8_t buf[3000];
            size_t  length = 3000;

            while (rb.read(buf, length)) {
                Frequency_Limit lm(0.1);
                if (lm.test()) {
                    std::cout << "读取到iq数据" << std::endl;
                }
                handle_iq(buf);
                //auto data = std::string((char *) buf, length);
                // if (length != 0) {
                //     std::cout << "read:【" << length << "】  【" << mem2hex(data.substr(64, 20)) << "】" << std::endl;
                // }
                if (!running) {
                    break;
                }
            }

            std::this_thread::sleep_for(std::chrono::microseconds(1));
        }
    });
#else
    manager.test_and_start_thread("iq回调处理", [this, handle_iq](std::atomic<bool> &running) {
        Frequency_Limit_ST limit(frame_rate);
        while (running) {
            auto list = iq_memory_buffer.get_all();
            if (debug && list.size() > 1000) {
                std::ostringstream oss;
                oss << "警告： 数据过多 当前数据队列长度:" << list.size() << std::endl;
                server_logger->c_debug({}, {}, oss.str());
            }
            if (list.empty()) {
                std::this_thread::sleep_for(std::chrono::microseconds(1));
                continue;
            }
            auto wi = Ws_Mgr::instance();
            if (!wi) continue;
            auto frame_number = list.size();
            for (int i = 0; i < frame_number; i++) {
                auto pdata = list[i].data();
                handle_iq(pdata);
            }
        }
    });
#endif




    auto interval = 1.0 / (double) frame_rate;
    app().getLoop()->runEvery(interval, [this] {

#if USE_PROCESS
        wi->ws_get_fft->push_data((const char *) psm.shm.data(), fft_point_number * 2);
#else
        std::lock_guard<std::mutex> lock(fft_mtx);
        ws_fft->push_data((const char *) fft_data.data(), fft_data.size());
#endif

    });




#if !USE_PROCESS

    {
        auto r = try_load_library(Psc::get_abs_path(library_path));
        if (!r) {
            Psc::fail_fast();
        }
        lib = r.value();
    }
    {
        auto r = Psc::try_load_function(lib, "adsb_set_less_30Mhz_ddc_param");
        if (!r) {
            Psc::fail_fast();
        }
        psc_set_less_30Mhz_ddc_param = (adsb_set_less_30Mhz_ddc_param_t)r.value();
    }
    {
        auto r = Psc::try_load_function(lib, "adsb_set_less_30Mhz_fft_param");
        if (!r) {
            Psc::fail_fast();
        }
        psc_set_less_30Mhz_fft_param = (adsb_set_less_30Mhz_fft_param_t)r.value();
    }
    using set_iq_cbk_t = decltype(&adsb_set_iq_cbk);
    set_iq_cbk_t set_iq_cbk;
    {
        auto r = Psc::try_load_function(lib, "adsb_set_iq_cbk");
        if (!r) {
            Psc::fail_fast();
        }
        set_iq_cbk = (set_iq_cbk_t)r.value();
    }

    using adsb_set_error_cbk_t = decltype(&set_error_cbk);
    adsb_set_error_cbk_t adsb_set_error_cbk;
    {
        auto r = Psc::try_load_function(lib, "set_error_cbk");
        if (!r) {
            Psc::fail_fast();
        }
        adsb_set_error_cbk = (adsb_set_error_cbk_t)r.value();
    }


    psc_set_less_30Mhz_ddc_param(center_freq, sample_rate, band_width, 1);
    psc_set_less_30Mhz_fft_param(fft_win_type, fft_point_number, fft_period_ms, 1);


    set_iq_cbk(
        [](void *phandle, void *pdata, uint64_t length) {
            if (stop_program != 0) {
                static Frequency_Limit fl;
                if (fl.test()) {
                    std::ostringstream oss;
                    oss  << "set_iq_cbk 回调退出：" << signal_to_string(stop_program) << "  " << VAR_STR_1(stop_program) << LOG_POS << std::endl;
                    dsp_logger->c_debug({}, {}, oss.str());
                }
            }
            if (length != 2112) {
                std::ostringstream oss;
                oss << LOG_POS << "  IQ回调传入的长度不对!" << std::endl;
                dsp_logger->c_debug({}, {}, oss.str());
                return;
            }
            auto that = (DSP_Config *) phandle;
            that->iq_memory_buffer.push(pdata, length); // length 永远等于2112
        },
    (void*)&Global::instance()->dsp_config);


#ifdef USE_ERROR_CBK

    adsb_set_error_cbk([](const char *file_name ,const char *func_name,const char *err_msg) {
        if (stop_program != 0) {
            static Frequency_Limit fl;
            if (fl.test()) {
                std::ostringstream oss;
                oss << "adsb_set_error_cbk 回调退出：" << signal_to_string(stop_program) << "  " << VAR_STR_1(stop_program) << LOG_POS << std::endl;
                dsp_logger->c_debug({}, {}, oss.str());
            }
            return;
        }
        Log_Type lt{
            {},
            {
            {"file_name", std::string(file_name)},
            {"func_name", std::string(func_name)},
            }
        };
        dsp_logger->debug({}, lt,  std::string(err_msg));
    });
#endif



#else
    adsb_set_less_30Mhz_ddc_param(center_freq, sample_rate, band_width, 1);
    adsb_set_less_30Mhz_fft_param(fft_win_type, fft_point_number, fft_period_ms, 1);
#endif

}
