#include "Data_Source.h"

#include <codecvt>

#include "../server/Global.h"
#include "Database.h"

Psc::serial::Serial* create_serial(const std::string& serial_name, Baud_Rate_Type baud_rate) {
    auto serial = new Psc::serial::Serial;
    serial->set_serial_name(serial_name);
    serial->set_baud_rate(baud_rate);
    serial->set_parity(Psc::serial::Parity::NoParity);
    serial->set_data_bits(Psc::serial::DataBits::Data8);
    serial->set_stop_bits(Psc::serial::StopBits::OneStop);
    serial->set_flow_control(Psc::serial::FlowControl::HardwareControl);
    serial->set_buffer_byte_size(10 * 1024);
    if (!serial->open())
    {
        std::cerr << "createSerial " + serial_name + ":" + std::to_string(baud_rate) + " 打开串口失败！\n";
        Psc::fail_fast();
        return nullptr;
    } else
    {
        // std::cerr << "createSerial " + serial_name + ":" + std::to_string(baud_rate) + " 打开串口成功！\n";
    }
    return serial;
}

std::shared_ptr<Data_Source> create_from_json(const JSON* that_json) {
    std::string type = that_json->get_string("type");
    std::shared_ptr<Data_Source> ret = create_data_source_from_type(type);
    ret->from_json(that_json);
    // if (ret->enable)
    // {
    //     ret->open();
    // }
    return ret;
}


std::shared_ptr<Data_Source> Data_Source::that() {
    return shared_from_this();
}
Psc::JSON Data_Source::get_state() {
    Psc::JSON ret = Psc::JSON::object();
    ret.append_list(get_custom_state_json().children);
    ret.append({"ds: read_speed（byte）", read_speed});
    ret.append({"ds: value_statistics（byte）", value_statistics});
    return ret;
}

std::string Data_Source::thread_key() const {
    //return "Data_Source_Handle_Thread:[" + key + "]";
    return key + "_DS_HT";
}

void Data_Source::test_and_attach_thread() {
    //static int t = Global::instance()->mode_acs.read_milliseconds;
    Global::instance()->thread_manager.test_and_start_thread(thread_key(), [this](std::atomic<bool>& running) {
        if (enable) {
            open();
        }
        while (running.load(std::memory_order_acquire) == true)
        {
            handle_in_loop();
            handle_mode_acs_source();
        }
    });
}


std::shared_ptr<SSR::Msg> Data_Source::create_msg(const std::shared_ptr<SSR::Data_Source_Interface> &source,
                        const std::string &packet) {
    if (packet[0] != 0x1a)
    {
        std::cout << "first:" << mem2hex(packet) << std::endl;
        return nullptr;
    }
    std::string error_len = key + "  " + LOG_POS_SIMPLE + std::string(" ") + "mode_s_error_length";
    if (packet.size() < 2) return nullptr;
    auto mt = packet[1];
    if (mt == SSR::Msg::AC)
    {
        if (packet.size() != SSR::Msg::AC_len)
        {
            SSR::mode_s_logger->c_debug(error_len, {},
                                   std::to_string(mt) + " size:" + std::to_string(packet.size()) +
                                       " hex:" + mem2hex(packet) + "  should:" + std::to_string(SSR::Msg::AC_len));
            mode_ac_statistic.add_length_error();
            return nullptr;
        }
        mode_ac_statistic.add();
        return std::make_shared<SSR::Mode_AC_Msg>(that(), packet);
    }
    if (mt == SSR::Msg::S7)
    {
        if (packet.size() != SSR::Mode_S_Msg::S7_len)
        {
            SSR::mode_s_logger->c_debug(error_len, {},
                                   std::to_string(mt) + " size:" + std::to_string(packet.size()) +
                                       " hex:" + mem2hex(packet) + "  should:" + std::to_string(SSR::Msg::S7_len));
            mode_s_statistic.add_length_error("[length error] Msg::S7 handle_mode_s_source");
            return nullptr;
        }
        return std::make_shared<SSR::Mode_S_Msg>(that(), packet);
    }

    if (mt == SSR::Msg::S14)
    {
        if (packet.size() != SSR::Msg::S14_len)
        {
            SSR::mode_s_logger->c_debug(error_len, {},
                                   std::to_string(mt) + " size:" + std::to_string(packet.size()) +
                                       " hex:" + mem2hex(packet) + "  should:" + std::to_string(SSR::Msg::S14_len));
            mode_s_statistic.add_length_error("[length error] Msg::S14 handle_mode_s_source");
            return nullptr;
        }
        return std::make_shared<SSR::Mode_S_Msg>(that(), packet);
    }

    if (mt == SSR::Msg::Radarcape_status) {
        return nullptr; // 不知道如何解析跳过
        if (packet.size() != SSR::Msg::Radarcape_status_len)
        {
            SSR::mode_s_logger->c_debug(error_len, {},
                                   std::to_string(mt) + " size:" + std::to_string(packet.size()) +
                                       " hex:" + mem2hex(packet) + "  should:" + std::to_string(SSR::Msg::Radarcape_status_len));
            mode_s_statistic.add_length_error("[length error]  Msg::Radarcape_status_len handle_mode_s_source");
            return nullptr;
        }
        return std::make_shared<SSR::Msg>(that(), packet);
    }

    if (mt == SSR::Msg::HULC_Status)
    {
        if (packet.size() != SSR::Msg::HULC_len)
        {

            // 找不到协议 size:23 hex:1A34195F0000001500B502FF06F423CE50000090000000  should:5
            SSR::mode_s_logger->c_debug(error_len, {},
                                   Psc::to_string(type) + "  HULC size:" +
                                       std::to_string(packet.size()) + " hex:" + mem2hex(packet));
            return nullptr;
        }
        return std::make_shared<SSR::Msg>(that(), packet);
    }
    // std::cout << "unknown_type:" << memory2hex(packet);
    rotating_log("unknown_type", mem2hex(packet));
    return nullptr;
}


void Data_Source::update_msg_day_time(std::shared_ptr<SSR::Msg> msg) {
    auto cur = std::dynamic_pointer_cast<SSR::Mode_Msg>(msg);
    if (cur != nullptr) {
        auto& pre = last_prase_msg;
        if (!pre) {
            pre = cur;
            return;
        }
        cur->day_num = pre->day_num;
        if (cur->time() < pre->time()) {
            ++cur->day_num;
        }
    }
}

void Data_Source::handle_mode_acs_source() {
    std::string mode_data = read();

    if (mode_data.empty()) {
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
       //std::cout << key + "数据为空，没有消息积压" << std::endl;
        return;
    }


    auto size = mode_data.size();
    read_speed.update(size);
    value_statistics.update(size);

    pre_handle(mode_data);

    SSR::Binary_Format_handle_buffer(buffer, mode_data, [this](std::string& packet) {
        auto msg = create_msg(that(), packet);
        if (!msg) return;
        update_msg_day_time(msg);
        push_to_feed(msg);
        auto mt = msg->type;
        bool mode_s = mt == SSR::Msg::S7 || mt == SSR::Msg::S14;
        if (mt == SSR::Msg::HULC_Status) {
            handle_HULC(packet);
        }
        else if (mt == SSR::Msg::Radarcape_status) {
            auto radarcape_msg = SSR::create_Radarcape_STATUS_Message(packet);
            std::cout << radarcape_msg.toJson().to_json_string() << std::endl;
            SSR::mode_s_logger->debug("Radarcape_status/radarcape", {}, radarcape_msg.toJson().to_json_string());
        }
        else if (mode_s)
        {
            // 拓展点
            handle_mode_s(std::dynamic_pointer_cast<SSR::Mode_S_Msg>(msg));
        }
    });
}


void Data_Source::test_and_stop_thread() const {
    Global::instance()->thread_manager.test_and_stop_thread(thread_key());
}

Psc::JSON Data_Source::get_all_connect_feed_status() {
    Psc::JSON ret = Psc::JSON::object();
    std::vector<Cached_Source_Info> list;
    {
        std::lock_guard g(cdf_mtx);
        list = cached_data_feed_key_list;
    }
    for (auto& i : list) {
        Psc::JSON cur = Psc::JSON::object();
        auto odf = Global::instance()->mode_acs.data_feed_config.map.get(i.key);
        if (odf.has_value()) {
            const auto& df = odf.value();
            i.mode_s_cache_num.update(df->msg_buffer.mode_s_msg_num);
            i.mode_other_cache_num.update(df->msg_buffer.mode_other_msg_num);
            cur.append({"mode_s_msg_num",  i.mode_s_cache_num.to_json()});
            cur.append({"mode_other_msg_num",  i.mode_other_cache_num.to_json()});
        }
        ret.append({i.key, cur});
    }
    return ret;
}
Psc::JSON Data_Source::get_all_connect_feed()  {
    Psc::JSON ret = Psc::JSON::array();
    std::vector<Cached_Source_Info> list;
    {
        std::lock_guard g(cdf_mtx);
        list = cached_data_feed_key_list;
    }
    for (auto& i : list) {
        Psc::JSON cur = Psc::JSON::object();
        cur.append({"key", i.key});
        ret.append(cur);
    }
    return ret;
}


std::vector<std::string> readLines(const std::string& path) {
    std::vector<std::string> lines;

    std::filesystem::path p = std::filesystem::path((const char8_t*)path.c_str());

    //std::filesystem::path p = std::filesystem::u8path(path);
    std::ifstream istream(p);
    if (!istream) {
        std::cerr << "读取模拟数据源 readLines 无法打开文件：" << path << std::endl;
        Psc::fail_fast();
    }

    // 将整个文件读入一个字符串
    std::stringstream stream;
    stream << istream.rdbuf();
    // 使用 stringstream 按行分割内容
    std::string line;
    char ch;
    while (stream.get(ch)) {
        if (ch == ' ') {
            continue;
        }
        if (ch == '\r' || ch == '\n') {
            if (!line.empty()) {
                lines.push_back(line);
                line.clear();
            }

            // if (ch == '\r') {
            //     stream.get(); // 吃掉 \n
            // }
            // if (ch == '\n') {
            //     stream.get(); // 吃掉 \n
            // }
            // 处理 \r\n 组合：如果当前是 \r，下一个是 \n，跳过它
            // if (ch == '\r' && stream.peek() == '\n') {
            //     stream.get(); // 吃掉 \n
            // }

        } else {
            line += ch;
        }
        if (line.size() == 1000) {
            lines.push_back(line);
            line.clear();
        }
    }
    // 最后一行如果没有换行符也处理一下
    if (!line.empty()) {
        lines.push_back(line);
    }
    std::ostringstream oss;
    oss << path + " 总行数：" + std::to_string(lines.size()) + "  有效行数：" + std::to_string(lines.size()) + "\n" << std::flush;
    std::cout << oss.str() << std::flush;
    return lines;
}



std::string extractID(const std::string& logLine) {
    size_t lastSpacePos = logLine.rfind(" ");  // 查找最后一个空格
    if (lastSpacePos != std::string::npos) {
        return logLine.substr(lastSpacePos + 1);  // 从最后一个空格之后提取字符串
    }
    return logLine;
}

time_t convert_to_timestamp(const std::string& str) {
    // 创建一个结构体 tm 来存储解析后的时间
    std::tm timeStruct = {};

    std::istringstream ss(str);
    ss >> std::get_time(&timeStruct, "%Y-%m-%d %H:%M:%S");
    if (ss.fail()) {
        std::cerr << "Failed to parse time" << std::endl;
        return -1; // 如果解析失败，返回 -1
    }

    // 将 tm 转换为 time_t（时间戳）
    time_t timestamp = std::mktime(&timeStruct);

    if (timestamp == -1) {
        std::cerr << "Failed to convert to time_t" << std::endl;
        return -1; // 如果转换失败，返回 -1
    }
    return timestamp;
}

std::tuple<std::string, time_t> extractID2(const std::string& logLine) {
    size_t lastSpacePos = logLine.rfind(" ");  // 查找最后一个空格
    time_t t =  convert_to_timestamp(logLine.substr(0, 19));
    if (lastSpacePos != std::string::npos) {
        return {logLine.substr(lastSpacePos + 1), t};  // 从最后一个空格之后提取字符串
    }
    return {logLine, t};
}


std::string bin_format(const std::string& hex, time_t t){
    std::tm* currentTime = std::localtime(&t);
    auto sec = currentTime->tm_hour * 3600 + currentTime->tm_min * 60 + currentTime->tm_sec;
    SSR::MLAT_timestamp a(sec, 0);
    return create_Binary_Format_memory(hex, 0, &a);
}

std::optional<std::string> File_Data_Source::get_raw_line(int& ret_index) {
    if (index == 0 && part_infos.empty())
    {
        ret_index = -1;
        return std::nullopt;
    }
    std::optional<std::string> log_info = std::nullopt;
    if (index == part_infos.size())
    {
        if (play_mode == Play_Mode::loop)
        {
            index = 0;
        } else
        {
            if (!have_report_play_back_all_success)
            {
                std::ostringstream oss;
                oss << SSR::get_current_date() << " " <<  SSR::get_current_time() << "  [进程:" << std::this_thread::get_id() << "]  " << file_path + "   全部加载成功！\n" << std::flush;
                std::cout << oss.str() << std::flush;
                have_report_play_back_all_success = true;
            }
            ret_index = -1;
            return std::nullopt;
        }
    }
    ret_index = index + 1;
    return part_infos[index++];
}


// 1A 33 1A 1A F1 FB 87 73 7E 7F a8001d81a87543b0a80000

std::string get_true_from_raw_line(Data_Source* ds, File_Data_Type data_type, int index, const std::string& log_info) {
    std::string ret;

    // 2025-01-06 07:20:17 [info] dc2541e65001401a3119b2dd1c7af67132201
    if (data_type == File_Data_Type::BIN) {
        return log_info;
    }
    else if (data_type == File_Data_Type::BIN_Text)
    {
        auto info = extractID(log_info);
        ret = hex2mem(info);
    }
    else if (data_type == File_Data_Type::AVR)
    {
        auto [info, time] = extractID2(log_info);
        ret = bin_format(info, time);
    }
    // 1A 33 10 01 A5 31 FB 9A 5D 8D 78 0E 47 99 08 8E 32 B0 08 BE E9 8A E6 1A 32 10 01 AA 14 61 7E 5E 02 E6 0F 38 7D 87 08
    // 1A 32 10 01 AD B5 64 F7 63 5D 78 0F 9D BA 93 BA
    else if (data_type == File_Data_Type::BIN_Blank_Text)
    {
        std::string info = log_info;
        std::string result = ds->last_char;
        ds->last_char = "";



        for (char c : info)
        {
            if (std::isxdigit(c)) {  // 会判断字符 c 是否是十六进制数字字符
                result += c;
            } else if (std::isspace(c)) {
                //
            } else {
                std::ostringstream oss;
                oss << "存在非法字符[" << c << "]: value:" << (int)c << "   in：" << log_info << LOG_POS;
                server_logger->c_debug("非法字符：", {}, oss.str());
                return "";
            }
        }
        auto size = result.size();
        if (size % 2 != 0) {
            // 保存最后一个字符
            ds->last_char = result.empty() ? "" : std::string(1, result.back());
            result.pop_back();
        }
        ret = hex2mem(result);

        // if (size % 2 != 0)
        // {
        //     std::ostringstream oss;
        //     size_t prefix_num = 20;
        //     size_t suffix_num = 20;
        //     auto origin_size = log_info.size();
        //     prefix_num = std::min(origin_size, prefix_num);
        //     suffix_num = std::min(origin_size, suffix_num);
        //     std::string prefix_data = log_info.substr(0, prefix_num);
        //     std::string suffix_data = log_info.substr(log_info.size() - suffix_num, suffix_num);
        //     oss << "index:" << index << "  长度不为偶数 不能转换成内存：" <<  "[内容]"  << "result_size" << result.size()  <<  "   origin_size:[" << origin_size << "]" << std::endl <<
        //     "  prefix[" << prefix_num << "]:" << prefix_data << std::endl <<
        //     "  suffix[" << suffix_num << "]:" << suffix_data << std::endl <<
        //     std::endl;
        //     server_logger->c_debug("长度不为偶数", {}, oss.str());
        //     //std::cout << "result:" << log_info << std::endl << std::endl;
        //     return "";
        // }

    }
    else if (data_type == File_Data_Type::BIN_Blank_One_Line_With_Escape) {
        auto size = log_info.size();
        if (size % 2 != 0)
        {
            return hex2mem(log_info.substr(0, size - 1));
        }
        return hex2mem(log_info);
    }
    else if (data_type == File_Data_Type::BIN_Blank_One_Line_No_Escape) {
        auto size = log_info.size();
        std::string data;
        if (size % 2 != 0)
        {
            data = log_info.substr(0, size - 1);
        } else {
            data = log_info;
        }
        return SSR::packet_to_escape_format(hex2mem(data));
    }
    else if (data_type == File_Data_Type::SIMPLE_BIN_Blank)
    {
        std::string info = log_info;
        std::string result;
        for (char c : info)
        {
            if (c != ' ')
            {
                result += c;
            }
        }
        // 1   +  1   +     6         +      1             +  2/7/14
        // 1a  +  type  time_stamp       signal_level          data
        int size = result.size();
        if (size != 2*2 && size != 7*2 && size != 14*2)
        {
            std::cout << "line_offset:[" << index << "]  " << "大小" << size <<  " 不为偶数,或不为2，7，14  不能转换成二进制格式：" << result << std::endl;
            return "";
        }
        std::string head("\x1a");
        std::string type;
        std::string time_stamp = "\x11\x22\x33\x44\x55\x66";
        std::string signal_level("\xFF");
        if (size == 2 * 2)
        {
            type = R"(1)";
        } else if (size == 7 * 2)
        {
            type = R"(2)";
        } else if (size == 14 * 2)
        {
            type = R"(3)";
        }
        ret = SSR::packet_to_escape_format(head + type + time_stamp + signal_level + hex2mem(result));

        //std::cout << ret.size() << std::endl;
    }
    else
    {
        std::cout << "未知的回放类型 " << static_cast<int>(data_type) << std::endl;
        Psc::fail_fast();
    }
    return ret;
}




void File_Data_Source::_close()  {
    std::lock_guard g(mtx);
    index = 0;
    part_infos.clear();
}

bool File_Data_Source::_open()  {
    std::lock_guard g(mtx);
    auto path = get_true_file_path();

    namespace fs = std::filesystem;
    if (!fs::exists(path)) {
        state = "文件不存在";
        return false;
    }

    if (data_type == File_Data_Type::BIN) {
        part_infos = readBinaryFileAsString(path, 1000);
    }
    else {
        part_infos = readLines(path);
    }
    state = "已加载" + to_string(part_infos.size()) + "长度数据!";
    return true;
}


std::vector<std::string> File_Data_Source::readBinaryFileAsString(const std::string& filepath, size_t part_size)  {
    // 打开文件（以二进制模式）
    std::ifstream file(filepath, std::ios::binary);

    if (!file) {
        std::cerr << "无法打开文件: " << filepath << std::endl;
        return {};  // 文件打开失败，返回空vector
    }

    // 获取文件大小
    file.seekg(0, std::ios::end);
    size_t fileSize = file.tellg();
    file.seekg(0, std::ios::beg);

    // 存储读取的部分
    std::vector<std::string> parts;

    // 读取文件的每一部分
    size_t bytesRead = 0;
    while (bytesRead < fileSize) {
        // 计算每个部分的长度（最后一部分可能小于part_size）
        size_t remaining = fileSize - bytesRead;
        size_t currentPartSize = (remaining < part_size) ? remaining : part_size;

        // 创建buffer来存储当前部分
        std::string buffer(currentPartSize, '\0');

        // 读取当前部分
        file.read(&buffer[0], currentPartSize);

        // 将读取的部分添加到vector
        parts.push_back(std::move(buffer));

        // 更新已读取字节数
        bytesRead += currentPartSize;
    }

    // 关闭文件
    file.close();

    return parts;
}


std::string File_Data_Source::read() {
    std::lock_guard g(mtx);
    int ret_index = 0;
    if (play_mode != Play_Mode::analysis)
    {
        std::optional<std::string> log_info = get_raw_line(ret_index);
        return log_info.has_value() ? get_true_from_raw_line(this, data_type, ret_index, log_info.value()) : "";
    } else
    {
        int num = 0;
        std::string ret;
        std::optional<std::string> log_info = get_raw_line(ret_index);
        while (log_info.has_value())
        {
            num++;
            auto data = log_info.value();
            auto cur = get_true_from_raw_line(this, data_type, ret_index, data);
            ret += cur;
            // 一定要在这个位置，否则会导致遗漏报文
            if (num == 1000) {
                break;
            }
            log_info = get_raw_line(ret_index);
        }
        // std::cout << "size == " << t.size() << std::endl;
        return ret;
    }
}


void File_Data_Source::handle_mode_acs_source() {
    std::string mode_data = read();
    if (mode_data.empty()) {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
        return;
    }
    pre_handle(mode_data);

    SSR::Binary_Format_handle_buffer(buffer, mode_data, [this](std::string& packet) {
        auto msg = create_msg(that(), packet);
        if (msg == nullptr) return;
        update_msg_day_time(msg);
        push_to_feed(msg);
        auto mt = msg->type;
        bool mode_s = mt == SSR::Msg::S7 || mt == SSR::Msg::S14;
        // 文件回放只处理mode_s
        if (mode_s)
        {
            // 拓展点
            cache_list.push_back(std::make_shared<Cache_Msg>(std::dynamic_pointer_cast<SSR::Mode_S_Msg>(msg)));
        }
    });


    if (cache_list.empty()) return;
    // 初始化流程
    if (pre_cache == nullptr) {
        pre_cache = cache_list.front();
        cache_list.pop_front();

        // 初始化当前开始时间点
        SSR::Play_Back_Time_Point& start = player_clock.start_time;
        start.day_num = pre_cache->day_num;
        start.day_sec = pre_cache->msg->mlat_timestamp.daysec;
    }


    auto sys_time = player_clock.get_cur_time_point();
    while (true) {
        if (cache_list.empty()) break;
        auto pre = pre_cache;
        auto cur = cache_list.front();
        cur->day_num = pre->day_num;
        if (cur->time() < pre->time()) {
            cur->day_num++;
        }
        cache_list.pop_front();
        pre_cache = cur;
        // 时间到了 回放这个文件
        handle_mode_s(cur->msg);
        if (cur->time() > sys_time) {
            break;
        }
    }
}

std::string Dll_Data_Source::read() {

    auto start = std::chrono::steady_clock::now();

    if (read_func_ptr == nullptr) return "";

    auto buf = reinterpret_cast<char *>(buffer.data());
    auto len = read_func_ptr(buf, buffer_size);
    vs.update(len);

    std::string data(buf, len);

    if (data.empty()) {
        auto end = std::chrono::steady_clock::now();
        auto cost = std::chrono::duration_cast<std::chrono::microseconds>(end - start).count();
        //std::cout << "Dll_Data_Source::read cost: " << cost << " us\n";
        return "";
    }

    auto ret = get_true_from_raw_line(this, data_type, -1, data);

    auto end = std::chrono::steady_clock::now();
    auto cost = std::chrono::duration_cast<std::chrono::microseconds>(end - start).count();

   // std::cout << "Dll_Data_Source::read cost: " << cost << " us\n";

    return ret;
}


// std::string Dll_Data_Source::read() {
//     if (read_func_ptr == nullptr) return "";
//     auto buf = reinterpret_cast<char *>(buffer.data());
//     auto len = read_func_ptr(buf, buffer_size);
//     vs.update(len);
//     std::string data(buf, len);
//     // std::ostringstream oss;
//     // oss << "Dll_Data_Source::read == (" << data.size() << ")\n";
//     // std::cout << oss.str() << std::flush;
//     if (data.empty()) return "";
//     auto ret = get_true_from_raw_line(this, data_type, -1, data);
//     return ret;
// }
bool Shared_Memory_Data_Source::_open() {
    sm = std::make_unique<Psc::SM_RingBuffer>();
    sm->init(shared_memory_name, shared_memory_size);
    return true;
}
void Shared_Memory_Data_Source::_close() {
    sm.reset();
}


std::string Shared_Memory_Data_Source::read() {
    auto size = sm->shm.size();
    size_t  length = size;
    std::string data;
    data.resize(size);

    bool ok = sm->read((uint8_t *)data.data(), length);
    if (!ok) {
        return "";
    }

    if (length != size) {
        data.resize(length);
    }
    if (Global::instance()->console_config.mode_s_console) {
        std::cout << "read:" << data << std::endl;
    }
    auto ret = get_true_from_raw_line(this, data_type, -1, data);
    return ret;
}

void Data_Source_Config::server(Global* g) {
    auto& svr = g->svr;
    auto& api = g->api;
    std::string name = "data_source";
    // 返回所有 JSON 数据的 API
    svr.Post(api + "get_data_source_connect", [this](HTTP_Param) {
        CHECK_JSON_PARAM
        HTTP_REQUIRE_VALUE(key, params.try_get_string("key"))
        auto t = map.get(key);
        JSON ret{nullptr};
        if (t.has_value()) {
            ret = t.value()->get_all_connect_feed();
        }
        res->setBody(warp(ret).to_json_string());
    });



    svr.Post(api + "get_data_source_state", [this](HTTP_Param) {
         CHECK_JSON_PARAM
         HTTP_REQUIRE_VALUE(key, params.try_get_string("key"))
         auto t = map.get(key);
         JSON ret{nullptr};
         if (t.has_value()) {
             ret = t.value()->get_state();
         }
         res->setBody(warp(ret).to_json_string());
     });


    svr.Post(api + "get_mode_s_data_source_config", [this](HTTP_Param) {
        res->setBody(warp(to_json()).to_json_string());
    });

    svr.Post(api + svr.update + name, [this, g](HTTP_Param) {
        CHECK_JSON_PARAM
        HTTP_REQUIRE_VALUE(key, params.try_get_string("key"))
        HTTP_REQUIRE_VALUE(ds, map.get(key))
        // 直接关闭
        ds->close();
        ds->from_json(&params);
        if (ds->enable) {
            ds->open();
            ds->test_and_attach_thread();
        } else
        {
            ds->test_and_stop_thread();
        }
        g->save();
        g->mode_acs.source_feed_relation_config.set_need_refresh();
    });
    svr.Post(api + svr.insert + name, [g, this](HTTP_Param) {
        CHECK_JSON_PARAM

        HTTP_REQUIRE_VALUE(index, params.try_get_number<int>("index"))
        HTTP_REQUIRE_PTR(data, params.get("data"))
        HTTP_REQUIRE_VALUE(t, data->try_get_string("type"))
        std::cout << params.to_json_string() << std::endl;
        std::shared_ptr<Data_Source> ds = create_data_source_from_type(t);
        ds->from_json(data);
        HTTP_REQUIRE_VALUE(enable, data->try_get_bool("enable"))
        if (enable)
        {
            ds->open();
            ds->test_and_attach_thread();
        }

        HTTP_REQUIRE_TRUE(map.insert(index, ds), "index")
        //std::cout << params.to_json_string() << std::endl;
        g->mode_acs.source_feed_relation_config.set_need_refresh();
        Config::save();
        res->setBody(warp(to_json()).to_json_string());
    });

    svr.Post(api + svr.remove + name, [g, this](HTTP_Param) {
        CHECK_JSON_PARAM
        HTTP_REQUIRE_VALUE(index, params.try_get_number<int>("index"))
        HTTP_REQUIRE_VALUE(ds, map.try_get(index))
        ds->close();
        ds->test_and_stop_thread();

        HTTP_REQUIRE_TRUE(map.remove(index), "index")
        g->save();
        res->setBody(warp(to_json()).to_json_string());
        g->mode_acs.source_feed_relation_config.set_need_refresh();
    });

    svr.Post(api + svr.rise + name, [g, this](HTTP_Param) {
        CHECK_JSON_PARAM
        HTTP_REQUIRE_VALUE(index, params.try_get_number<int>("index"))
        HTTP_REQUIRE_TRUE(map.swap(index, index - 1), "index")
        g->save();
        res->setBody(warp(to_json()).to_json_string());
        g->mode_acs.source_feed_relation_config.set_need_refresh();
    });
    svr.Post(api + svr.fall + name, [g, this](HTTP_Param) {
        CHECK_JSON_PARAM
        HTTP_REQUIRE_VALUE(index, params.try_get_number<int>("index"))
        HTTP_REQUIRE_TRUE(map.swap(index, index + 1), "index")
        g->save();
        res->setBody(warp(to_json()).to_json_string());
        g->mode_acs.source_feed_relation_config.set_need_refresh();
    });
}
void Data_Source::refresh_data_feed_key_list() {
    //std::cout << key <<  "   refresh_data_feed_key_list" << std::endl;
    std::vector<std::string> tmp;
    for (auto& relation : Global::instance()->mode_acs.source_feed_relation_config.map.list()) {
        if (!relation->enable) continue;
        if (relation->type == "One_to_One_Relation") {
            auto t = dynamic_cast<One_to_One_Relation*>(relation.get());

            //std::cout << VAR_STR_2(t->source_key, this->key) <<  "   refresh_data_feed_key_list" << std::endl;
            if (t->source_key == this->key) {
                tmp.push_back(t->feed_key);
            }
        } else if (relation->type == "First_Source_To_All_Feed_Relation") {
            auto t = dynamic_cast<First_Source_To_All_Feed_Relation*>(relation.get());
            std::shared_ptr<Data_Source> first = nullptr;
            for (const auto& ds : Global::instance()->mode_acs.data_source_config.map.list()) {
                if (ds->enable) {
                    first = ds;
                    break;
                }
            }
            if (first->key == key) {
                for (const auto& df : Global::instance()->mode_acs.data_feed_config.map.list()) {
                    if (df->enable) {
                        tmp.push_back(df->key);
                    }
                }
            }
        }

        std::sort(tmp.begin(), tmp.end());
        // 去重
        auto last = std::unique(tmp.begin(), tmp.end());
        tmp.erase(last, tmp.end());



        {
            std::vector<Cached_Source_Info> tmp_info;

            // 创建 tmp_info 向量，将 tmp 的内容转化为 Cached_Source_Info 对象
            for (const auto& key : tmp) {
                tmp_info.push_back({key});
            }

            {
                // 进入临界区，锁住 mutex，确保线程安全
                std::lock_guard<std::mutex> lock(cdf_mtx);

                // 1. 删除 tmp 中没有的 cached_data_feed_key_list 元素
                auto it = cached_data_feed_key_list.begin();
                while (it != cached_data_feed_key_list.end()) {
                    if (std::find_if(tmp_info.begin(), tmp_info.end(),
                        [&](const Cached_Source_Info& info) {
                            return info.key == it->key;
                        }) == tmp_info.end()) {
                        // 如果当前元素在 tmp 中找不到，删除它
                        it = cached_data_feed_key_list.erase(it);
                        } else {
                            ++it;
                        }
                }

                // 2. 创建 tmp 中有但 cached_data_feed_key_list 没有的元素
                for (const auto& tmp_item : tmp_info) {
                    auto found = std::find_if(cached_data_feed_key_list.begin(), cached_data_feed_key_list.end(),
                                              [&](const Cached_Source_Info& cached_item) {
                                                  return cached_item.key == tmp_item.key;
                                              });
                    if (found == cached_data_feed_key_list.end()) {
                        // 如果 tmp_item 不在 cached_data_feed_key_list 中，添加它
                        cached_data_feed_key_list.push_back(tmp_item);
                    }
                }
            }
        }
    }
}

void Data_Source::push_to_feed(const std::shared_ptr<SSR::Msg>& msg) {
    if (need_refresh_data_feed_key_list) {
        refresh_data_feed_key_list();
        need_refresh_data_feed_key_list = false;
    }
    auto& other_size = Global::instance()->mode_acs.mode_other_max_num;
    auto& s_size = Global::instance()->mode_acs.mode_other_max_num;
    std::vector<std::string> list;



    for (const auto& item : cached_data_feed_key_list) {
        auto& key = item.key;
        auto opt_feed = Global::instance()->mode_acs.data_feed_config.map.get(key);
        if (!opt_feed.has_value()) {
            std::cout << "wrong data feed: " << key << std::endl;
            break;
        }
        std::shared_ptr<Data_Feed>& feed = opt_feed.value();
        if (!feed->enable) continue;


        if (msg->type == SSR::Mode_Msg::T::S7 || msg->type == SSR::Mode_Msg::T::S14) {
            if (feed->msg_buffer.mode_s_msg_num > s_size) {
                continue;
            } else {
                ++feed->msg_buffer.mode_s_msg_num;
            }
        } else {
            if (feed->msg_buffer.mode_other_msg_num > other_size) {
                continue;
            } else {
                ++feed->msg_buffer.mode_other_msg_num;
            }
        }
        std::optional<std::string> binary = convert_to_send_format(this, feed, msg);
        if (binary.has_value()) {
            feed->msg_buffer.push(binary.value());
        }
    }
}

void Data_Source::handle_mode_s(std::shared_ptr<SSR::Mode_S_Msg> mode_s_msg) {
    std::string t = mode_s_msg->mlat_timestamp.to_memory() + mode_s_msg->signal_level + Psc::hex2mem(mode_s_msg->msg_hex);
    auto p = mode_s_msg->packet.substr(2);
    if (t != p) {
        std::cout << mem2hex(t, true, " ") << std::endl;
        std::cout << mem2hex(mode_s_msg->packet, true, " ") << std::endl;
        std::cout << VAR_STR_2(t, mode_s_msg->packet) << std::endl;
    }
    push_to_feed(mode_s_msg);
    auto& cfg = Global::instance()->mode_acs;
    SSR::parse_mode_s_bin(this, mode_s_msg, base_station.get_pos(),cfg.max_speed_m_s, cfg.air_pos_timeout, cfg.surface_pos_timeout);






    auto base = get_aircraft(mode_s_msg->icao);

    if (base)
    {
        auto derived = std::dynamic_pointer_cast<Aircraft>(base);
        derived->refresh_external_database_info();
    }





    if (Global::instance()->mlat.merge) {
        auto& mh = Global::instance()->mlat_handler;
        mh.push(mode_s_msg);
        auto tt = mh.get_all();
        if (!tt.empty()) {
            pure_log("@/logs/mlat.log", "receive[" + std::to_string(tt.size()) + "]:\n");
            for (Mlat_MSG& t : tt) {
                std::ostringstream oss;
                oss << std::fixed << std::setprecision(10);
                oss << "===================================================\n";
                int n = static_cast<int>(t.size());
                auto icao = t.icao().value_or("null");
                oss << "\tmlat_msg:" << n;
                oss << "\ticao:" << icao;
                oss << "\thex:" << t.msg_hex() << std::endl;

                bool have_pos = false;
                if (icao != "null") {
                    auto air = get_aircraft(icao);
                    if (air != nullptr) {
                        auto callsign = air->bds20_call_sign();
                        oss << "\tcall_sign：" << callsign.value_or("null") << std::endl;
                        auto o_pos = air->pos();
                        auto o_alt = air->altitude_meter();
                        if (o_pos.has_value() && o_alt.has_value()) {
                            have_pos = true;
                            auto pos = o_pos.value();
                            alt = o_alt.value();
                            double x, y, z;
                            SSR::CPR::WGS84_LBH_to_XYZ(pos.lon, pos.lat, alt, x, y, z);
                            oss << "\tpos：[" << pos.lon << "," << pos.lat << "," << alt << "]" << std::endl;
                            oss << "\tXYZ：[" << x << "," << y << "," << z << "]" << std::endl;
                        }
                    }
                }
                for (int i = 0; i < t.size(); i++) {
                    auto& msg = t.list[i];
                    oss << "\t\tds:" << msg->source->get_key();
                    oss << "\ttime:" << msg->mlat_timestamp.to_string() << std::endl;
                }
                // if (have_pos) {
                    //std::cout << oss.str() << std::endl;
                    pure_log("@/logs/mlat.log", oss.str());
                // }
                // if (t.size() > 3 && have_pos && icao == "78139F") {
                //     size_t i = 1;
                //     auto range = hex_mlat_map.equal_range(t.msg_hex());
                //     // 遍历与key_to_find对应的所有元素
                //     for (auto it = range.first; it != range.second; ++it) {
                //         oss << "\t" << i++ << "  " << it->second.msg << std::endl;
                //     }
                //     oss << "===================================================\n";
                //     std::cout << oss.str() << std::endl;
                //     pure_log("@/logs/icao/" + icao + "_mlat.log", oss.str(), true);
                // }
            }
        }
    }

}
void Data_Source::pre_handle(const std::string& mode_data) {
    static bool mode_s_console = Global::instance()->console_config.mode_s_console;
    static bool record_playback = Global::instance()->console_config.record_playback;
    //std::cout << source->key + "   read   " << mode_s_data.size() << std::endl;
    if (mode_data.empty()) return;
    if (record_playback) {
        pure_log(get_exe_dir() + "/playback/" + key + "_playback.dat", mem2hex(mode_data, true, " ") + "\n");
    }
    if (mode_s_console) {
        std::cout << key + " read:[mode_s_serial]:" << mem2hex(mode_data) << std::endl;
    }
}