From afd4bf8e69b74cef72781a6eef99ac8d074f1edd Mon Sep 17 00:00:00 2001 From: wyc <1104749580@qq.com> Date: Tue, 23 Jun 2026 14:01:23 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8F=90=E5=8F=96=E5=87=BA=20Data=5FSource=5FH?= =?UTF-8?q?andler?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Data_Feed/data_feed_thread.cpp | 12 +- .../Local_Server/Data_Source/Data_Source.cpp | 316 ++---------------- module/Local_Server/Data_Source/Data_Source.h | 108 +++--- .../Data_Source/Data_Source_Handler.cpp | 264 +++++++++++++++ .../Data_Source/Data_Source_Handler.h | 33 ++ module/Local_Server/Data_Source/Database.h | 1 - module/Local_Server/server/Global.cpp | 30 -- module/Local_Server_main.cpp | 5 +- 8 files changed, 380 insertions(+), 389 deletions(-) create mode 100644 module/Local_Server/Data_Source/Data_Source_Handler.cpp create mode 100644 module/Local_Server/Data_Source/Data_Source_Handler.h diff --git a/module/Local_Server/Data_Feed/data_feed_thread.cpp b/module/Local_Server/Data_Feed/data_feed_thread.cpp index 553805a..f1b5b92 100644 --- a/module/Local_Server/Data_Feed/data_feed_thread.cpp +++ b/module/Local_Server/Data_Feed/data_feed_thread.cpp @@ -226,6 +226,16 @@ ucoro::awaitable data_feed_thread_coro(std::atomic& running) { } void data_feed_thread(std::atomic& running) { - ucoro::sync_await(data_feed_thread_coro(running)); + + try { + ucoro::sync_await(data_feed_thread_coro(running)); + } catch (const std::exception& e) { + std::cerr << "data_feed_thread_coro exception: " << e.what() << std::endl; + Psc::fail_fast_core_dump(""); + } catch (...) { + std::cerr << "data_feed_thread_coro unknown exception" << std::endl; + Psc::fail_fast_core_dump(""); + } + } diff --git a/module/Local_Server/Data_Source/Data_Source.cpp b/module/Local_Server/Data_Source/Data_Source.cpp index 205b3e6..2106f49 100644 --- a/module/Local_Server/Data_Source/Data_Source.cpp +++ b/module/Local_Server/Data_Source/Data_Source.cpp @@ -51,6 +51,8 @@ std::shared_ptr create_from_json(const JSON* that_json) { } + + std::shared_ptr Data_Source::that() { return shared_from_this(); } @@ -62,6 +64,18 @@ Psc::JSON Data_Source::get_state() { return ret; } +Data_Source::Data_Source() { + // parse_format = std::make_shared(); + // 写入到配置文件是懒加载 其他保存时他跟着保存 + base_station.handle_when_updated = [this](SSR::HULC_Status_Message msg) { + if (this->update_form_gps) { + lat = msg.get_latitude(); + lon = msg.get_longitude(); + alt = msg.Alt; + } + }; +} + std::string Data_Source::thread_key() const { //return "Data_Source_Handle_Thread:[" + key + "]"; return key + "_DS_HT"; @@ -74,88 +88,7 @@ void Data_Source::test_and_attach_thread() { } -std::shared_ptr Data_Source::create_msg(const std::shared_ptr &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(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(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(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(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(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 msg) { +void File_Data_Source::before_handle_msg(std::shared_ptr& msg) { auto cur = std::dynamic_pointer_cast(msg); if (cur != nullptr) { auto& pre = last_prase_msg; @@ -170,60 +103,13 @@ void Data_Source::update_msg_day_time(std::shared_ptr msg) { } } -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; - } - process_mode_acs_data(mode_data); -} -void Data_Source::process_mode_acs_data(const std::string& mode_data) { - 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(msg)); - } - }); -} - -ucoro::awaitable Data_Source::handle_in_loop_coro() { - handle_in_loop(); - co_return; -} - -ucoro::awaitable Data_Source::recv_coro() { - co_return read(); -} ucoro::awaitable Data_Source::source_step_coro() { co_await handle_in_loop_coro(); - auto mode_data = co_await recv_coro(); + auto mode_data = co_await read_coro(); if (mode_data.empty()) { std::this_thread::sleep_for(std::chrono::milliseconds(10)); co_return; @@ -640,13 +526,13 @@ std::vector File_Data_Source::readBinaryFileAsString(const std::str } -std::string File_Data_Source::read() { +ucoro::awaitable File_Data_Source::read_coro() { std::lock_guard g(mtx); int ret_index = 0; if (play_mode != Play_Mode::analysis) { std::optional 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()) : ""; + co_return log_info.has_value() ? get_true_from_raw_line(this, data_type, ret_index, log_info.value()) : ""; } else { int num = 0; @@ -665,48 +551,26 @@ std::string File_Data_Source::read() { log_info = get_raw_line(ret_index); } // std::cout << "size == " << t.size() << std::endl; - return ret; + co_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(std::dynamic_pointer_cast(msg))); - } - }); +void File_Data_Source::handle_mode_s(std::shared_ptr msg) { + if (!msg) return; + cache_list.push_back(std::make_shared(std::move(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; @@ -718,24 +582,19 @@ void File_Data_Source::handle_mode_acs_source() { } cache_list.pop_front(); pre_cache = cur; - // 时间到了 回放这个文件 - handle_mode_s(cur->msg); + Data_Source::handle_mode_s(cur->msg); if (cur->time() > sys_time) { break; } } } -ucoro::awaitable File_Data_Source::source_step_coro() { - handle_mode_acs_source(); - co_return; -} -std::string Dll_Data_Source::read() { +ucoro::awaitable Dll_Data_Source::read_coro() { auto start = std::chrono::steady_clock::now(); - if (read_func_ptr == nullptr) return ""; + if (read_func_ptr == nullptr) co_return ""; auto buf = reinterpret_cast(buffer.data()); auto len = read_func_ptr(buf, buffer_size); @@ -747,7 +606,7 @@ std::string Dll_Data_Source::read() { auto end = std::chrono::steady_clock::now(); auto cost = std::chrono::duration_cast(end - start).count(); //std::cout << "Dll_Data_Source::read cost: " << cost << " us\n"; - return ""; + co_return ""; } auto ret = get_true_from_raw_line(this, data_type, -1, data); @@ -757,23 +616,10 @@ std::string Dll_Data_Source::read() { // std::cout << "Dll_Data_Source::read cost: " << cost << " us\n"; - return ret; + co_return ret; } -// std::string Dll_Data_Source::read() { -// if (read_func_ptr == nullptr) return ""; -// auto buf = reinterpret_cast(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(); sm->init(shared_memory_name, shared_memory_size); @@ -784,7 +630,7 @@ void Shared_Memory_Data_Source::_close() { } -std::string Shared_Memory_Data_Source::read() { +ucoro::awaitable Shared_Memory_Data_Source::read_coro() { auto size = sm->shm.size(); size_t length = size; std::string data; @@ -792,7 +638,7 @@ std::string Shared_Memory_Data_Source::read() { bool ok = sm->read((uint8_t *)data.data(), length); if (!ok) { - return ""; + co_return ""; } if (length != size) { @@ -802,7 +648,7 @@ std::string Shared_Memory_Data_Source::read() { std::cout << "read:" << data << std::endl; } auto ret = get_true_from_raw_line(this, data_type, -1, data); - return ret; + co_return ret; } void Data_Source_Config::server(Global* g) { @@ -1030,105 +876,3 @@ void Data_Source::push_to_feed(const std::shared_ptr& msg) { } } -void Data_Source::handle_mode_s(std::shared_ptr 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(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(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; - } -} - diff --git a/module/Local_Server/Data_Source/Data_Source.h b/module/Local_Server/Data_Source/Data_Source.h index c54bbd0..91e41ec 100644 --- a/module/Local_Server/Data_Source/Data_Source.h +++ b/module/Local_Server/Data_Source/Data_Source.h @@ -1,18 +1,8 @@ #pragma once -#include -#include "Core/Base/SM_RingBuffer.h" -#include "Core/Serial/Serial.h" -#include "Core/Statistics/Frequency_Limit.h" -#include "Core/Statistics/Statistics.h" -#include "Core/transmit_protocol/core/statistics.h" -#include "Database.h" - -#include "Local_Server/global_include.h" -#include "global.h" -#include "PlayBack.h" +#include "Data_Source_Handler.h" namespace Psc { class SM_RingBuffer; @@ -72,75 +62,47 @@ protected: std::vector test_v; }; -class Data_Source : public DataBase, public std::enable_shared_from_this { + + + + +class Data_Source : public std::enable_shared_from_this, public Data_Source_Handler{ public: ~Data_Source() override = default; - virtual std::string read() { return ""; } std::shared_ptr that(); + virtual ucoro::awaitable read_coro() = 0; + virtual Psc::JSON get_custom_state_json() = 0; Psc::JSON get_state(); - - std::string last_char; // 用于处理奇数字节的情况 - // std::shared_ptr parse_format; - Psc::Speed_Statistics read_speed; - Psc::Value_Statistics value_statistics; - Data_Source() { - // parse_format = std::make_shared(); - // 写入到配置文件是懒加载 其他保存时他跟着保存 - base_station.handle_when_updated = [this](SSR::HULC_Status_Message msg) { - if (this->update_form_gps) { - lat = msg.get_latitude(); - lon = msg.get_longitude(); - alt = msg.Alt; - } - }; - } + + Data_Source(); std::shared_ptr last_prase_msg = nullptr; + virtual ucoro::awaitable handle_in_loop_coro(){co_return;} - virtual void handle_mode_s(std::shared_ptr msg); - virtual void pre_handle(const std::string &data); - virtual std::shared_ptr create_msg(const std::shared_ptr& source, const std::string& packet); - virtual void update_msg_day_time(std::shared_ptr msg); - virtual void handle_mode_acs_source(); - virtual ucoro::awaitable handle_in_loop_coro(); - virtual ucoro::awaitable recv_coro(); virtual ucoro::awaitable source_step_coro(); - - - void process_mode_acs_data(const std::string& mode_data); - - void refresh_data_feed_key_list(); void push_to_feed(const std::shared_ptr &msg); std::atomic enable{}; std::string type; std::atomic base_station_show{}; std::atomic aircraft_show{}; - - std::atomic show_icao = true; std::atomic show_call_sign = false; std::atomic show_fly_status = false; std::atomic keep_mode = true; - - - double lat{}; double lon{}; double alt{}; std::atomic update_form_gps{}; std::string color = "#1677ff"; int aircraft_pixel_size{}; - Mode_AC_Statistic_Data mode_ac_statistic; - Mode_S_Statistic_Data mode_s_statistic; - Frequency_Limit statistic_fl; - virtual void handle_in_loop() {}; + Frequency_Limit statistic_fl; std::string thread_key() const; void test_and_attach_thread(); void test_and_stop_thread() const; @@ -230,9 +192,6 @@ protected: struct TCP_Client_Data_Source : Data_Source { - void handle_in_loop() override { - ucoro::sync_await(handle_in_loop_coro()); - } ucoro::awaitable handle_in_loop_coro() override { co_await cli.tick_coro(); co_return; @@ -273,10 +232,8 @@ struct TCP_Client_Data_Source : Data_Source { Get_J(port); } - std::string read() override { - return ucoro::sync_await(recv_coro()); - } - ucoro::awaitable recv_coro() override { + + ucoro::awaitable read_coro() override { co_return co_await cli.read_coro(); } }; @@ -288,7 +245,7 @@ struct Serial_Data_Source : Data_Source { } Serial_Data_Source() { this->type = "Serial_Data_Source"; } bool _open() override { - serial = new Psc::serial::Serial; + serial = std::make_unique(); serial->set_serial_name(port_name); serial->set_baud_rate(baud_rate); serial->set_parity(Psc::serial::Parity::NoParity); @@ -304,8 +261,20 @@ struct Serial_Data_Source : Data_Source { } return ok; } - void _close() override { serial->close(); } - std::string read() override { return serial->read(); } + void _close() override { if (serial) serial->close(); } + + ucoro::awaitable handle_in_loop_coro() override { + if (serial) { + co_await serial->tick_coro(); + } + co_return; + } + ucoro::awaitable read_coro() override { + if (!serial) { + co_return ""; + } + co_return co_await serial->read_coro(); + } Psc::JSON to_json() override { Psc::JSON ret = Data_Source::to_json(); Ret_J(port_name); @@ -316,11 +285,10 @@ struct Serial_Data_Source : Data_Source { Data_Source::from_json(that_json); Get_J(port_name) Get_J(baud_rate); } - Psc::serial::Serial *serial{}; + std::unique_ptr serial{}; ~Serial_Data_Source() override { if (serial) { serial->close(); - delete serial; } } }; @@ -353,9 +321,7 @@ struct File_Data_Source : public Data_Source { } File_Data_Source() { this->type = "File_Data_Source"; } ~File_Data_Source() override = default; - - - + void before_handle_msg(std::shared_ptr& msg) override; struct Cache_Msg { explicit Cache_Msg(std::shared_ptr msg) : msg(std::move(msg)) {} @@ -389,9 +355,10 @@ struct File_Data_Source : public Data_Source { Ret_J(play_mode); return ret; } - std::string read() override; - void handle_mode_acs_source() override; - ucoro::awaitable source_step_coro() override; + + ucoro::awaitable read_coro() override; + + void handle_mode_s(std::shared_ptr msg) override; protected: std::optional get_raw_line(int &ret_index); @@ -462,7 +429,8 @@ struct Dll_Data_Source : Data_Source { state = "加载成功"; return true; } - std::string read() override; + + ucoro::awaitable read_coro(); }; @@ -485,8 +453,8 @@ struct Shared_Memory_Data_Source : Data_Source { } bool _open() override; void _close() override; - std::string read() override; + ucoro::awaitable read_coro() override; protected: std::unique_ptr sm; std::string shared_memory_name; diff --git a/module/Local_Server/Data_Source/Data_Source_Handler.cpp b/module/Local_Server/Data_Source/Data_Source_Handler.cpp new file mode 100644 index 0000000..7930c3f --- /dev/null +++ b/module/Local_Server/Data_Source/Data_Source_Handler.cpp @@ -0,0 +1,264 @@ +#include "Data_Source_Handler.h" +#include "Data_Source.h" +#include "Local_Server/server/Global.h" +using namespace Psc; + + + +std::shared_ptr ds(Data_Source_Handler* dsh) { + return dynamic_cast(dsh)->that(); +} + + + +std::shared_ptr Data_Source_Handler::create_msg( + const std::string &packet) { + + auto source = ds(this); + if (packet[0] != 0x1a) + { + std::cout << "first:" << mem2hex(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) + { + 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(source, 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(source, 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(source, 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(source, 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(source->type) + " HULC size:" + + std::to_string(packet.size()) + " hex:" + mem2hex(packet)); + return nullptr; + } + return std::make_shared(source, packet); + } + // std::cout << "unknown_type:" << memory2hex(packet); + rotating_log("unknown_type", mem2hex(packet)); + return nullptr; +} + +void Data_Source_Handler::process_mode_acs_data(const std::string& mode_data) { + 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 (mode_data.empty()) return; + 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; + } + + SSR::Binary_Format_handle_buffer(source->buffer, mode_data, [this, source](std::string& packet) { + auto msg = create_msg(packet); + if (!msg) return; + source->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) { + source->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(msg)); + } + }); +} + + +void Data_Source_Handler::handle_mode_s(std::shared_ptr 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) { + 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; + } + + // TODO: 可能不需要push_to_feed + source->push_to_feed(mode_s_msg); + auto& cfg = Global::instance()->mode_acs; + + + + + SSR::parse_mode_s_bin(source.get(), mode_s_msg, source->base_station.get_pos(),cfg.max_speed_m_s, cfg.air_pos_timeout, cfg.surface_pos_timeout); + + + + + auto base = source->get_aircraft(mode_s_msg->icao); + + if (base) + { + auto derived = std::dynamic_pointer_cast(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(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(); + source->alt = o_alt.value(); + double x, y, z; + SSR::CPR::WGS84_LBH_to_XYZ(pos.lon, pos.lat, source->alt, x, y, z); + oss << "\tpos：[" << pos.lon << "," << pos.lat << "," << source->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_Handler::handle_HULC(const std::string& packet) { + std::uint8_t len = SSR::get_len(packet); + std::uint8_t id = SSR::get_id(packet); + auto msg = mem2hex(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(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", 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); + } +} diff --git a/module/Local_Server/Data_Source/Data_Source_Handler.h b/module/Local_Server/Data_Source/Data_Source_Handler.h new file mode 100644 index 0000000..be1522e --- /dev/null +++ b/module/Local_Server/Data_Source/Data_Source_Handler.h @@ -0,0 +1,33 @@ +#pragma once +#include + +#include "Core/Base/SM_RingBuffer.h" +#include "Core/Serial/Serial_Coro.h" + +#include "Core/Statistics/Frequency_Limit.h" +#include "Core/Statistics/Statistics.h" +#include "Core/transmit_protocol/core/statistics.h" + +#include "Database.h" +#include "Local_Server/global_include.h" +#include "global.h" +#include "PlayBack.h" + + +// 这个函数专门装cpu密集函数 放到线程池去处理 +class Data_Source_Handler : public DataBase +{ +public: + std::shared_ptr create_msg(const std::string& packet); + virtual void before_handle_msg(std::shared_ptr& msg){}; + virtual void handle_mode_s(std::shared_ptr msg); + void handle_HULC(const std::string& packet); + void process_mode_acs_data(const std::string& mode_data); + Mode_AC_Statistic_Data mode_ac_statistic; + Mode_S_Statistic_Data mode_s_statistic; + Psc::Speed_Statistics read_speed; + Psc::Value_Statistics value_statistics; + virtual ~Data_Source_Handler(){}; +}; + + diff --git a/module/Local_Server/Data_Source/Database.h b/module/Local_Server/Data_Source/Database.h index 4e8f39f..edb9901 100644 --- a/module/Local_Server/Data_Source/Database.h +++ b/module/Local_Server/Data_Source/Database.h @@ -211,7 +211,6 @@ public: #endif size_t get_aircraft_num(); void clear_aircraft() { aircraft_map.delete_all(); } - void handle_HULC(const std::string& packet); ~DataBase() override { aircraft_map.delete_all(); } diff --git a/module/Local_Server/server/Global.cpp b/module/Local_Server/server/Global.cpp index 7c9a94a..e54ea9d 100644 --- a/module/Local_Server/server/Global.cpp +++ b/module/Local_Server/server/Global.cpp @@ -210,37 +210,7 @@ void handle_buffer_head_tail(std::string& buffer, const std::string& data, const } } } -void DataBase::handle_HULC(const std::string& packet) { - std::uint8_t len = SSR::get_len(packet); - std::uint8_t id = SSR::get_id(packet); - auto msg = mem2hex(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(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", 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 read_ais_serial_data(std::atomic& running) { // static int t = config.ais.read_serial_milliseconds; diff --git a/module/Local_Server_main.cpp b/module/Local_Server_main.cpp index 2dfdc57..bc6bbd2 100644 --- a/module/Local_Server_main.cpp +++ b/module/Local_Server_main.cpp @@ -173,10 +173,13 @@ int wyc_main(int argc, char *argv[]) { }); manager.test_and_start_thread("data_feed_thread", data_feed_thread); static int t = Global::instance()->mode_acs.read_milliseconds; - for (auto &source : + for (std::shared_ptr &source : Global::instance()->mode_acs.data_source_config.map.list()) { + if (!source->enable) continue; + + source->test_and_attach_thread(); } g->dsp_config.init_env();