#include "Data_Source_Handler.h"
#include <string_view>
#include <tuple>
#include <utility>
#include "Data_Source.h"
#include "Local_Server/server/Global.h"
#include "Local_Server/server/io_coro.h"
using namespace Psc;
std::shared_ptr<Data_Source> ds(Data_Source_Handler* dsh) {
    return dynamic_cast<Data_Source*>(dsh)->that();
}
std::shared_ptr<SSR::Msg> Data_Source_Handler::create_msg(std::string_view packet) {
    auto source = ds(this);
    if (packet[0] != 0x1a) {
        std::cout << "first:" << mem2hex(std::string(packet)) << std::endl;
        return nullptr;
    }
    std::string error_len = source->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) {
            auto packet_hex = mem2hex(std::string(packet));
            SSR::mode_s_logger->c_debug(error_len, {},
                                        std::to_string(mt) + " size:" + std::to_string(packet.size()) +
                                        " hex:" + packet_hex + "  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>(source, packet);
    }
    if (mt == SSR::Msg::S7) {
        if (packet.size() != SSR::Mode_S_Msg::S7_len) {
            auto packet_hex = mem2hex(std::string(packet));
            SSR::mode_s_logger->c_debug(error_len, {},
                                        std::to_string(mt) + " size:" + std::to_string(packet.size()) +
                                        " hex:" + packet_hex + "  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>(source, packet);
    }
    if (mt == SSR::Msg::S14) {
        if (packet.size() != SSR::Msg::S14_len) {
            auto packet_hex = mem2hex(std::string(packet));
            SSR::mode_s_logger->c_debug(error_len, {},
                                        std::to_string(mt) + " size:" + std::to_string(packet.size()) +
                                        " hex:" + packet_hex + "  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>(source, packet);
    }
    if (mt == SSR::Msg::Radarcape_status) {
        return nullptr; // 不知道如何解析跳过
        if (packet.size() != SSR::Msg::Radarcape_status_len) {
            auto packet_hex = mem2hex(std::string(packet));
            SSR::mode_s_logger->c_debug(error_len, {},
                                        std::to_string(mt) + " size:" + std::to_string(packet.size()) + " hex:" +
                                        packet_hex + "  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>(source, packet);
    }
    if (mt == SSR::Msg::HULC_Status) {
        if (packet.size() != SSR::Msg::HULC_len) {
            auto packet_hex = mem2hex(std::string(packet));
            // 找不到协议 size:23 hex:1A34195F0000001500B502FF06F423CE50000090000000
            // should:5
            SSR::mode_s_logger->c_debug(error_len, {},
                                        Psc::to_string(source->type) + "  HULC size:" + std::to_string(packet.size()) +
                                        " hex:" + packet_hex);
            return nullptr;
        }
        return std::make_shared<SSR::Msg>(source, packet);
    }
    // std::cout << "unknown_type:" << memory2hex(packet);
    rotating_log("unknown_type", mem2hex(std::string(packet)));
    return nullptr;
}
extern bool data_source_debug;
size_t Data_Source_Handler::process_mode_acs_data(std::string_view origin_data) {
    // 先读取原始数据 注意多复制了一遍
    std::string mode_data(origin_data);
    std::string name = type + "_" + key;
    if (data_source_debug) {
        std::cout << "before origin_data_transform_mode_data  " << VAR_STR_2(name, mode_data.size()) << std::endl;
    }
    origin_data_transform_mode_data(mode_data);
    if (data_source_debug) {
        std::cout << " after  origin_data_transform_mode_data   " << VAR_STR_2(name, mode_data.size()) << key
                  << std::endl;
    }
    static Value_Growth_Multi_T mt;
    auto t = key;
    if (data_source_debug) {
        std::cout << name << " 读取到:" << VAR_STR_2(key, mode_data.size()) << " " << std::endl;
    }
    if (mt.test(key, mode_data.size())) {
        std::cout << VAR_STR_2(key, mode_data.size()) << " 数据增长过快，可能内存积压" << std::endl;
    }
    size_t ret = 0;
    if (mode_data.empty()) {
        return ret;
    }
    auto source = ds(this);
    auto size = mode_data.size();
    read_speed.update(size);
    value_statistics.update(size);
    // 预处理数据
    static bool mode_s_console = Global::instance()->console_config.mode_s_console;
    static bool record_playback = Global::instance()->console_config.record_playback;
    if (record_playback) {
        pure_log(get_exe_dir() + "/playback/" + source->key + "_playback.dat", mem2hex(mode_data, true, " ") + "\n");
    }
    if (mode_s_console) {
        std::cout << source->key + " read:[mode_s_serial]:" << mem2hex(mode_data) << std::endl;
    }
    auto handle_packet = [this, source, &ret](std::string& packet) {
        ret++;
        auto msg = create_msg(packet);
        if (!msg)
            return;
        auto mt = msg->type;
        bool mode_s = mt == SSR::Msg::S7 || mt == SSR::Msg::S14;
        if (mt == SSR::Msg::HULC_Status) {
            source->push_to_feed(msg);
            source->handle_HULC(packet);
        }
        else if (mt == SSR::Msg::Radarcape_status) {
            source->push_to_feed(msg);
            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) {
            // 拓展点
            auto mode_s_msg = std::dynamic_pointer_cast<SSR::Mode_S_Msg>(msg);
            source->push_to_feed(mode_s_msg);
            handle_mode_s(mode_s_msg);
        }
        else {
            source->push_to_feed(msg);
        }
    };
    auto f = [this, source, handle_packet](std::string& packet) {
        const auto wait = Global::instance()->mode_acs.settings.member<&Mode_ACS_Config_Data::wait_process_msg>().read([](const auto& value) { return value; });
        if (wait) {
            handle_packet(packet);
        }
        else {
            auto co = Coro::instance();
            auto executor = co->process_data->get_executor();
            asio::post(executor, [handle_packet, packet = std::move(packet)]() mutable { handle_packet(packet); });
        }
    };
    SSR::Binary_Format_handle_buffer(source->buffer, mode_data, f);
    return ret;
}
void Data_Source_Handler::handle_mode_s(std::shared_ptr<SSR::Mode_S_Msg>& mode_s_msg) {
    auto source = ds(this);
    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) {
        auto data = std::format("编解码出问题了(组合的数据，拼接的数据)  hex:[{}] [{}]  bin:[{}] [{}]",
                                mem2hex(t, true, " "), mem2hex(mode_s_msg->packet, true, " "), t, mode_s_msg->packet);
        Psc::fail_fast(data);
    }
    const auto mode_config = Global::instance()->mode_acs.settings.read([](const auto& value) {
        return std::tuple{value.time_space_filter, value.speed_filter, value.aircraft_change_list_adsb_range_filter, value.aircraft_change_list_adsb_range_factor, value.max_speed_m_s, value.air_pos_timeout, value.surface_pos_timeout};
    });
    const auto source_config = source->settings.read([](const auto& value) {
        return std::tuple{value.ignore_msg_time, value.base_station_has_valid_position, value.lat, value.lon, value.alt};
    });
    const auto [time_space_filter, speed_filter, range_filter, range_factor, max_speed_m_s, air_pos_timeout, surface_pos_timeout] = mode_config;
    const auto [use_system_time, base_station_has_valid_position, lat, lon, alt] = source_config;
    auto base_station_pos = source->base_station.get_pos();
    if (!base_station_pos && base_station_has_valid_position)
        base_station_pos = SSR::Position_3D{lat, lon, alt};
    SSR::ADS_B_T::Constraint air_constraint{max_speed_m_s, air_pos_timeout, SSR::cpr_cb, time_space_filter,
                                            speed_filter, use_system_time, base_station_pos, alt, range_filter, range_factor};
    SSR::ADS_B_T::Constraint surface_constraint{max_speed_m_s, surface_pos_timeout, SSR::cpr_cb, time_space_filter,
                                                speed_filter, use_system_time, base_station_pos, alt, range_filter, range_factor};
    SSR::parse_mode_s_bin(source.get(), mode_s_msg, base_station_pos, air_constraint, surface_constraint);
    auto base = source->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.settings.member<&MLAT_Config_Data::merge>().read([](const auto& value) { return value; })) {
        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 = source->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();
                            const auto altitude = o_alt.value();
                            source->settings.member<&Data_Source_Data::alt>().write([altitude](double& value) {
                                value = altitude;
                            });
                            double x, y, z;
                            SSR::CPR::WGS84_LBH_to_XYZ(pos.lon, pos.lat, altitude, x, y, z);
                            oss << "\tpos：[" << pos.lon << "," << pos.lat << "," << altitude << "]" << 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;
                }
                pure_log("@/logs/mlat.log", oss.str());
            }
        }
    }
}
void Data_Source_Handler::handle_HULC(std::string_view packet) {
    std::uint8_t len = SSR::get_len(packet);
    std::uint8_t id = SSR::get_id(packet);
    auto msg = mem2hex(std::string(packet));
    if (len != packet.size() - 4) {
        SSR::mode_s_logger->debug("HULC/error_length", {},
                                  "需要: " + std::to_string(len) + " 当前: " + std::to_string(packet.size()) +
                                  " hex: " + mem2hex(std::string(packet)));
        return;
    }
    if (id == 1) {
        // 状态消息
        auto status_msg = SSR::create_HULC_Status_Message(packet);
        bool gps_ok = status_msg.GPS_device_detected() && status_msg.GPS_valid() && status_msg.GPS_has_valid_fix();
        if (gps_ok) {
        }
        auto g = Global::instance();
        base_station.set_msg(status_msg);
        Log_Type type({}, {{"msg", std::string(msg)}});
        SSR::mode_s_logger->debug("HULC/status", type, status_msg.toJson().to_json_string());
        if (gps_ok) {
        }
    }
    else if (id == 24) {
        SSR::mode_s_logger->debug("HULC/reply", {}, msg);
    }
    else {
        SSR::mode_s_logger->debug("HULC/unknown_id", {}, msg);
    }
}
void Data_Source_Handler::refresh_data_feed_key_list() {
    auto* global = Global::instance();
    std::map<std::string, std::string> assignment;
    const auto sources = global->mode_acs.data_source_config.map.list();
    const auto feeds = global->mode_acs.data_feed_config.map.list();
    const auto relations = global->mode_acs.source_feed_relation_config.map.list();
    for (const auto& relation : relations) {
        relation->settings.read([&](const auto& config) {
            if (!config.enable || relation->type != "One_to_One_Relation")
                return;
            const auto source = global->mode_acs.data_source_config.map.get(config.source_key);
            const auto feed = global->mode_acs.data_feed_config.map.get(config.feed_key);
            if (source.has_value() && feed.has_value() && source.value()->enabled() && feed.value()->enabled() && !assignment.contains(config.feed_key))
                assignment.emplace(config.feed_key, config.source_key);
        });
    }
    for (const auto& relation : relations) {
        relation->settings.read([&](const auto& config) {
            if (!config.enable || relation->type != "Name_One_To_Many_Relation")
                return;
            const auto source = global->mode_acs.data_source_config.map.get(config.source_key);
            if (!source.has_value() || !source.value()->enabled())
                return;
            const std::string& prefix = config.feed_name.empty() ? config.source_key : config.feed_name;
            for (const auto& feed : feeds) {
                if (!feed->enabled() || assignment.contains(feed->key) || !feed->key.starts_with(prefix))
                    continue;
                assignment.emplace(feed->key, config.source_key);
            }
        });
    }
    bool use_default = false;
    for (const auto& relation : relations) {
        if (relation->type == "First_Source_To_All_Feed_Relation" && relation->settings.member<&Source_Feed_Relation_Data::enable>().read([](const auto& value) { return value; })) {
            use_default = true;
            break;
        }
    }
    if (use_default) {
        auto first = std::find_if(sources.begin(), sources.end(), [](const auto& source) {
            return source->enabled();
        });
        if (first != sources.end()) {
            for (const auto& feed : feeds) {
                if (feed->enabled() && !assignment.contains(feed->key))
                    assignment.emplace(feed->key, (*first)->key);
            }
        }
    }
    std::vector<Cached_Source_Info> next;
    for (const auto& [feed_key, source_key] : assignment) {
        if (source_key == key)
            next.push_back({feed_key});
    }
    std::lock_guard<std::mutex> lock(cdf_mtx);
    for (auto& item : next) {
        auto old = std::find_if(cached_data_feed_key_list.begin(), cached_data_feed_key_list.end(), [&](const Cached_Source_Info& value) {
            return value.key == item.key;
        });
        if (old != cached_data_feed_key_list.end()) {
            item.mode_s_cache_num = old->mode_s_cache_num;
            item.mode_other_cache_num = old->mode_other_cache_num;
        }
    }
    cached_data_feed_key_list = std::move(next);
}

std::optional<std::string> convert_to_send_format(Data_Source_Handler* ds, const std::shared_ptr<Data_Feed>& feed,
                                                  const std::shared_ptr<SSR::Msg>& msg) {
    auto& type = msg->type;
    auto fs = dynamic_cast<File_Data_Source*>(msg->source.get());
    if (fs) {
        if (fs->specific.member<&File_Data_Source_Data::play_mode>().read([](const auto& value) { return value; }) == Play_Mode::analysis) {
            // std::cout << fs->key << "  analysis  i ==" << i << std::endl;
        }
    }
    // 出口输出的条件
    const auto feed_format = feed->output_format.read([](const auto& value) { return value; });
    const auto output_format = feed_format.type;
    const auto use_mode_ac = feed_format.use_mode_ac;
    const auto use_status = feed_format.use_status;
    const auto mode_s_output_type = feed_format.mode_s_output_type;
    // 计算出是否输出消息
    std::optional<std::string> send_msg = std::nullopt;
    if (type == SSR::Msg::HULC_Status) {
        if (!use_status)
            return std::nullopt;
        auto data = static_cast<SSR::Msg*>(msg.get());
        if (output_format == Output_Data_Format::BIN) {
            send_msg = SSR::packet_to_escape_format(data->packet);
        }
    }
    else if (type == SSR::Msg::Radarcape_status) {
        auto data = static_cast<SSR::Msg*>(msg.get());
        if (output_format == Output_Data_Format::BIN) {
            send_msg = SSR::packet_to_escape_format(data->packet);
        }
    }
    else if (type == SSR::Msg::AC) {
        if (!use_mode_ac)
            return std::nullopt;
        auto data = static_cast<SSR::Mode_AC_Msg*>(msg.get());
        std::string& msg_hex = data->msg_hex;
        SSR::MLAT_timestamp& mlat_timestamp = data->mlat_timestamp;
        if (output_format == Output_Data_Format::BIN) {
            send_msg = SSR::packet_to_escape_format(msg->packet);
        }
        else if (output_format == Output_Data_Format::AVR) {
            send_msg = SSR::create_AVR_format(msg_hex);
        }
        else if (output_format == Output_Data_Format::AVR_MLAT) {
            send_msg = create_MLAT_AVR_format(msg_hex, &mlat_timestamp);
        }
    }
    else if (type == SSR::Msg::S7 || type == SSR::Msg::S14) {
        auto s_msg = static_cast<SSR::Mode_S_Msg*>(msg.get());
        std::string& icao = s_msg->icao;
        SSR::Downlink_Format& df = s_msg->df;
        std::string& msg_hex = s_msg->msg_hex;
        SSR::MLAT_timestamp& mlat_timestamp = s_msg->mlat_timestamp;
        auto aircraft = ds->get_aircraft(s_msg->icao);
        if (output_format == Output_Data_Format::SBS) {
            bool sbs_out_put = false;
            if (feed_format.sbs_only_pos) {
                auto s = ds->get_aircraft(s_msg->icao);
                if (s) {
                    auto pos = s->pos();
                    if (pos.has_value()) {
                        if (feed->sbs_flm.test(s_msg->icao)) {
                            sbs_out_put = true;
                        }
                    }
                }
            }
            else {
                if (feed->sbs_flm.test(s_msg->icao)) {
                    sbs_out_put = true;
                }
            }
            if (!sbs_out_put)
                return std::nullopt;
            send_msg = "";
            SSR::SBS_MSG msg_base;
            if (!aircraft)
                return std::nullopt;
            msg_base.set(aircraft.get());
            for (int msg_t = 1; msg_t <= 8; ++msg_t) {
                send_msg->append(msg_base.to_msg(msg_t) + "\n");
            }
            send_msg->append(msg_base.to_ID() + "\n");
            send_msg->append(msg_base.to_AIR() + "\n");
            send_msg->append(msg_base.to_CLK() + "\n");
            send_msg->append(msg_base.to_SEL() + "\n");
            send_msg->append(msg_base.to_STA() + "\n");
            return send_msg;
        }
        bool DF_11_17_18 = df == SSR::Downlink_Format::All_Call_Reply_11 ||
        df == SSR::Downlink_Format::Extended_Squitter_17 ||
        df == SSR::Downlink_Format::Extended_Squitter_Non_Transponder_18;
        if (mode_s_output_type == Mode_S_Output_Type::DF_11_17_18 && !DF_11_17_18)
            return std::nullopt;
        if (mode_s_output_type == Mode_S_Output_Type::NO_POS_Mode_S) {
            if (aircraft && aircraft->pos().has_value()) {
                return std::nullopt;
            }
        }
        if (output_format == Output_Data_Format::BIN) {
            send_msg = SSR::packet_to_escape_format(s_msg->packet);
        }
        else if (output_format == Output_Data_Format::AVR) {
            send_msg = SSR::create_AVR_format(msg_hex);
        }
        else if (output_format == Output_Data_Format::AVR_MLAT) {
            send_msg = create_MLAT_AVR_format(msg_hex, &mlat_timestamp);
        }
        else {
            std::cerr << "Unknown output format  " << VAR_STR_1(output_format) << std::endl;
            return std::nullopt;
        }
    }
    return send_msg;
}
void Data_Source_Handler::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;
    }
    const auto [other_size, s_size] = Global::instance()->mode_acs.settings.read([](const auto& value) {
        return std::pair{value.mode_other_max_num, value.mode_s_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->enabled())
            continue;
        if (msg->type == SSR::Mode_Msg::T::S7 || msg->type == SSR::Mode_Msg::T::S14) {
            // static Frequency_Limit fl;
            // if (!fl.test()) {
            //     std::ostringstream oss;
            //     oss <<  "收到消息 ===" << VAR_STR_3(msg->type
            //     ,feed->msg_buffer.mode_s_msg_num, s_size) << std::endl; std::cout
            //     << oss.str() << std::endl;
            // }
            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());
        }
    }
}
Psc::JSON Data_Source_Handler::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_Handler::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;
}
