ae/ECAP_Server
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

更新

Browse Source
This commit is contained in:
种太阳的热带美短
2026-08-05 11:59:45 +08:00
parent 18e64dfe22
commit 587c1d6d01
22 changed files with 2517 additions and 3049 deletions
Download Patch File Download Diff File Expand all files Collapse all files
module/Local_Server/Data_Source/Data_Source.cpp
+38 -61
View File
@@ -1,10 +1,9 @@
#include "Data_Source.h"
#include <codecvt>
#include <string_view>
#include "../server/Global.h"
#include "Database.h"
#include <string_view>
Psc::serial::Serial* create_serial(std::string_view serial_name,
Baud_Rate_Type baud_rate) {
Psc::serial::Serial* create_serial(std::string_view 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);
@@ -14,8 +13,7 @@ Psc::serial::Serial* create_serial(std::string_view serial_name,
serial->set_flow_control(Psc::serial::FlowControl::HardwareControl);
serial->set_buffer_byte_size(10 * 1024);
if (!serial->open()) {
std::cerr << "createSerial " + std::string(serial_name) + ":" +
std::to_string(baud_rate) + " 打开串口失败！\n";
std::cerr << "createSerial " + std::string(serial_name) + ":" + std::to_string(baud_rate) + " 打开串口失败！\n";
Psc::fail_fast();
return nullptr;
}
@@ -25,14 +23,10 @@ Psc::serial::Serial* create_serial(std::string_view serial_name,
}
return serial;
}
std::shared_ptr<Data_Source> create_from_json(const JSON* that_json) {
std::shared_ptr<Data_Source> create_from_json(const JSON* that_json, bool not_exist_use_default_value) {
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();
// }
ret->from_json(that_json, not_exist_use_default_value);
return ret;
}
std::shared_ptr<Data_Source> Data_Source::that() {
@@ -87,8 +81,7 @@ void File_Data_Source::before_handle_msg(std::shared_ptr<SSR::Mode_Msg>& msg) {
std::vector<std::string> readLines(std::string_view path) {
std::vector<std::string> lines;
auto path_string = std::string(path);
std::filesystem::path p =
std::filesystem::path((const char8_t*)path_string.c_str());
std::filesystem::path p = std::filesystem::path((const char8_t*)path_string.c_str());
// std::filesystem::path p = std::filesystem::u8path(path);
std::ifstream istream(p);
if (!istream) {
@@ -134,9 +127,8 @@ std::vector<std::string> readLines(std::string_view path) {
lines.push_back(line);
}
std::ostringstream oss;
oss << path_string + " 总行数：" + std::to_string(lines.size()) + " 有效行数：" +
std::to_string(lines.size()) + "\n"
<< std::flush;
oss << path_string + " 总行数：" + std::to_string(lines.size()) + " 有效行数：" + std::to_string(lines.size()) +
"\n" << std::flush;
std::cout << oss.str() << std::flush;
return lines;
}
@@ -168,17 +160,13 @@ std::tuple<std::string, time_t> extractID2(std::string_view logLine) {
size_t lastSpacePos = logLine.rfind(" "); // 查找最后一个空格
time_t t = convert_to_timestamp(logLine.substr(0, 19));
if (lastSpacePos != std::string::npos) {
return {
std::string(logLine.substr(lastSpacePos + 1)),
t
}; // 从最后一个空格之后提取字符串
return {std::string(logLine.substr(lastSpacePos + 1)), t}; // 从最后一个空格之后提取字符串
}
return {std::string(logLine), t};
}
std::string bin_format(std::string_view hex, time_t t) {
std::tm* currentTime = std::localtime(&t);
auto sec = currentTime->tm_hour * 3600 + currentTime->tm_min * 60 +
currentTime->tm_sec;
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);
}
@@ -196,9 +184,8 @@ std::optional<std::string> File_Data_Source::get_raw_line(int& ret_index) {
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::this_thread::get_id() << "] " << file_path + " 全部加载成功！\n"
<< std::flush;
std::cout << oss.str() << std::flush;
have_report_play_back_all_success = true;
}
@@ -210,8 +197,7 @@ std::optional<std::string> File_Data_Source::get_raw_line(int& ret_index) {
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, std::string_view log_info) {
std::string get_true_from_raw_line(Data_Source* ds, File_Data_Type data_type, int index, std::string_view log_info) {
std::string ret;
// 2025-01-06 07:20:17 [info] dc2541e65001401a3119b2dd1c7af67132201
if (data_type == File_Data_Type::BIN) {
@@ -242,8 +228,7 @@ std::string get_true_from_raw_line(Data_Source* ds, File_Data_Type data_type,
}
else {
std::ostringstream oss;
oss << "存在非法字符[" << c << "]: value:" << (int)c << " in："
<< log_info << LOG_POS;
oss << "存在非法字符[" << c << "]: value:" << (int)c << " in：" << log_info << LOG_POS;
server_logger->c_debug("非法字符：", {}, oss.str());
return "";
}
@@ -287,8 +272,7 @@ std::string get_true_from_raw_line(Data_Source* ds, File_Data_Type data_type,
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;
<< " 不为偶数,或不为2，7，14 不能转换成二进制格式：" << result << std::endl;
return "";
}
std::string head("\x1a");
@@ -304,8 +288,7 @@ std::string get_true_from_raw_line(Data_Source* ds, File_Data_Type data_type,
else if (size == 14 * 2) {
type = R"(3)";
}
ret = SSR::packet_to_escape_format(head + type + time_stamp + signal_level +
hex2mem(result));
ret = SSR::packet_to_escape_format(head + type + time_stamp + signal_level + hex2mem(result));
// std::cout << ret.size() << std::endl;
}
else {
@@ -337,8 +320,7 @@ asio::awaitable<void> File_Data_Source::_open() {
state = "已加载" + to_string(part_infos.size()) + "长度数据!";
co_return;
}
std::vector<std::string> File_Data_Source::readBinaryFileAsString(std::string_view filepath,
size_t part_size) {
std::vector<std::string> File_Data_Source::readBinaryFileAsString(std::string_view filepath, size_t part_size) {
// 打开文件（以二进制模式）
std::ifstream file(filepath.data(), std::ios::binary);
if (!file) {
@@ -376,10 +358,7 @@ void File_Data_Source::origin_data_transform_mode_data(std::string& mode_data) {
int ret_index = 0;
if (play_mode != Play_Mode::analysis) {
std::optional<std::string> log_info = get_raw_line(ret_index);
mode_data = log_info.has_value()
? get_true_from_raw_line(this, data_type, ret_index,
log_info.value())
: "";
mode_data = log_info.has_value() ? get_true_from_raw_line(this, data_type, ret_index, log_info.value()) : "";
}
else {
int num = 0;
@@ -400,10 +379,12 @@ void File_Data_Source::origin_data_transform_mode_data(std::string& mode_data) {
mode_data = ret;
}
}
void File_Data_Source::handle_mode_s(std::shared_ptr<SSR::Mode_S_Msg> msg) {
if (!msg) return;
void File_Data_Source::handle_mode_s(std::shared_ptr<SSR::Mode_S_Msg>& msg) {
if (!msg)
return;
cache_list.push_back(std::make_shared<Cache_Msg>(std::move(msg)));
if (cache_list.empty()) return;
if (cache_list.empty())
return;
if (pre_cache == nullptr) {
pre_cache = cache_list.front();
cache_list.pop_front();
@@ -413,7 +394,8 @@ void File_Data_Source::handle_mode_s(std::shared_ptr<SSR::Mode_S_Msg> msg) {
}
auto sys_time = player_clock.get_cur_time_point();
while (true) {
if (cache_list.empty()) break;
if (cache_list.empty())
break;
auto pre = pre_cache;
auto cur = cache_list.front();
cur->day_num = pre->day_num;
@@ -430,23 +412,21 @@ void File_Data_Source::handle_mode_s(std::shared_ptr<SSR::Mode_S_Msg> msg) {
}
void Dll_Data_Source::origin_data_transform_mode_data(std::string& mode_data) {
auto start = std::chrono::steady_clock::now();
if (read_func_ptr == nullptr) return;
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();
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();
auto cost = std::chrono::duration_cast<std::chrono::microseconds>(end - start).count();
// std::cout << "Dll_Data_Source::read cost: " << cost << " us\n";
mode_data = ret;
}
@@ -459,8 +439,7 @@ asio::awaitable<void> Shared_Memory_Data_Source::_close() {
sm.reset();
co_return;
}
void Shared_Memory_Data_Source::origin_data_transform_mode_data(
std::string& mode_data) {
void Shared_Memory_Data_Source::origin_data_transform_mode_data(std::string& mode_data) {
auto size = sm->shm.size();
size_t length = size;
std::string data;
@@ -503,36 +482,37 @@ void Data_Source_Config::server(Global* g) {
}
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 + "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
std::cout << "Data_Source:" << req->path() << that_json->to_json_string();
HTTP_REQUIRE_VALUE(key, params.try_get_string("key"))
HTTP_REQUIRE_VALUE(ds, map.get(key))
ds->from_json(&params);
ds->from_json(&params, true);
g->save();
g->mode_acs.source_feed_relation_config.set_need_refresh();
res->setBody(warp(ds->to_json()).to_json_string());
});
svr.Post(api + svr.insert + name, [g, this](HTTP_Param) {
CHECK_JSON_PARAM
std::cout << "Data_Source:" << req->path() << that_json->to_json_string();
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);
ds->from_json(data, true);
HTTP_REQUIRE_VALUE(enable, data->try_get_bool("enable"))
(void)enable;
HTTP_REQUIRE_TRUE(map.insert(index, ds), "index")
// std::cout << params.to_json_string() << std::endl;
HTTP_REQUIRE_TRUE(map.insert(index, ds), "index");
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
std::cout << "Data_Source:" << req->path() << that_json->to_json_string();
HTTP_REQUIRE_VALUE(index, params.try_get_number<int>("index"))
HTTP_REQUIRE_VALUE(ds, map.try_get(index))
ds->async_stop();
@@ -558,6 +538,3 @@ void Data_Source_Config::server(Global* g) {
g->mode_acs.source_feed_relation_config.set_need_refresh();
});
}
module/Local_Server/Data_Source/Data_Source.h
+59 -63
View File
@@ -5,7 +5,7 @@ namespace Psc {
class SM_RingBuffer;
}
class Data_Source;
std::shared_ptr<Data_Source> create_from_json(const Psc::JSON* that_json);
std::shared_ptr<Data_Source> create_from_json(const Psc::JSON* that_json, bool not_exist_use_default_value);
class Input_Format {
public:
Input_Format() {
@@ -17,7 +17,8 @@ public:
for (auto e : magic_enum::enum_values<SSR::Downlink_Format>()) {
std::string_view value_view = magic_enum::enum_name(e);
std::string value(value_view.begin(), value_view.end());
if (value == Psc::to_string(SSR::Downlink_Format::Unknown)) continue;
if (value == Psc::to_string(SSR::Downlink_Format::Unknown))
continue;
auto ev = static_cast<std::uint8_t>(e);
bool it = test_v[ev];
ret.append({value, it});
@@ -29,7 +30,8 @@ public:
for (auto e : magic_enum::enum_values<SSR::Downlink_Format>()) {
std::string_view value_view = magic_enum::enum_name(e);
std::string value(value_view.begin(), value_view.end());
if (value == Psc::to_string(SSR::Downlink_Format::Unknown)) continue;
if (value == Psc::to_string(SSR::Downlink_Format::Unknown))
continue;
auto use = json->get_bool(value);
auto ev = static_cast<std::uint8_t>(e);
test_v[ev] = use;
@@ -38,12 +40,14 @@ public:
bool test(SSR::Downlink_Format df) {
std::lock_guard g(mtx);
auto dfv = static_cast<std::uint8_t>(df);
if (dfv >= static_cast<std::uint8_t>(SSR::Downlink_Format::Unknown)) return false;
if (dfv >= static_cast<std::uint8_t>(SSR::Downlink_Format::Unknown))
return false;
return test_v[dfv];
}
bool test(std::uint8_t dfv) {
std::lock_guard g(mtx);
if (dfv >= static_cast<std::uint8_t>(SSR::Downlink_Format::Unknown)) return false;
if (dfv >= static_cast<std::uint8_t>(SSR::Downlink_Format::Unknown))
return false;
return test_v[dfv];
}
protected:
@@ -77,12 +81,12 @@ public:
ret.append({"map3d", map3d.to_base_json()});
return ret;
}
void from_base_json(const Psc::JSON* that_json) {
void from_base_json(const Psc::JSON* that_json, bool not_exist_use_default_value) {
if (that_json == nullptr || that_json->valueType != Psc::Object) {
throw Psc::json_assign_error(std::make_error_code(std::errc::invalid_argument), "map_display");
}
map2d.from_base_json(that_json->get("map2d"));
map3d.from_base_json(that_json->get("map3d"));
map2d.from_base_json(that_json->get("map2d"), not_exist_use_default_value);
map3d.from_base_json(that_json->get("map3d"), not_exist_use_default_value);
}
};
class Data_Source_Data {
@@ -97,29 +101,18 @@ public:
Psc::Copyable_Atomic<bool> keep_mode = true;
[[nodiscard]] Psc::JSON to_base_json() const {
auto ret = Psc::JSON::object();
Ret_J(base_station_has_valid_position)
ret.append({"map_display", map_display.to_base_json()});
Ret_J(lat)
Ret_J(lon)
Ret_J(alt)
Ret_J(ignore_msg_time)
Ret_J(update_form_gps)
Ret_J(keep_mode)
return ret;
Ret_J(base_station_has_valid_position) ret.append({"map_display", map_display.to_base_json()});
Ret_J(lat) Ret_J(lon) Ret_J(alt) Ret_J(ignore_msg_time) Ret_J(update_form_gps) Ret_J(keep_mode) return ret;
}
void from_base_json(const Psc::JSON* that_json) {
void from_base_json(const Psc::JSON* that_json, bool not_exist_use_default_value) {
Get_J(base_station_has_valid_position)
map_display.from_base_json(that_json->get("map_display"));
Get_J(lat)
Get_J(lon)
Get_J(alt)
Get_J(ignore_msg_time)
Get_J(update_form_gps)
Get_J(keep_mode)
map_display.from_base_json(that_json->get("map_display"), not_exist_use_default_value);
Get_J(lat) Get_J(lon) Get_J(alt) Get_J(ignore_msg_time) Get_J(update_form_gps) Get_J(keep_mode)
}
};
class Data_Source : public std::enable_shared_from_this<Data_Source>,
public Data_Source_Handler, public Data_Source_Data {
public Data_Source_Handler,
public Data_Source_Data {
public:
std::shared_ptr<Data_Source> that();
virtual asio::awaitable<std::string> read_coro() {
@@ -149,9 +142,9 @@ public:
virtual Psc::JSON to_json() {
return With_Loop_Coro::to_base_json() += Data_Source_Data::to_base_json();
}
virtual void from_json(const Psc::JSON* that_json) {
With_Loop_Coro::from_base_json(that_json);
Data_Source_Data::from_base_json(that_json);
virtual void from_json(const Psc::JSON* that_json, bool not_exist_use_default_value) {
With_Loop_Coro::from_base_json(that_json, not_exist_use_default_value);
Data_Source_Data::from_base_json(that_json, not_exist_use_default_value);
}
std::string buffer;
std::atomic_bool ask_sleep = false;
@@ -193,9 +186,9 @@ public:
Psc::JSON to_json() override {
return Data_Source::to_json() += TCP_Client_Data_Source_Data::to_base_json();
}
void from_json(const Psc::JSON* that_json) override {
Data_Source::from_json(that_json);
TCP_Client_Data_Source_Data::from_base_json(that_json);
void from_json(const Psc::JSON* that_json, bool not_exist_use_default_value) override {
Data_Source::from_json(that_json, not_exist_use_default_value);
TCP_Client_Data_Source_Data::from_base_json(that_json, not_exist_use_default_value);
}
asio::awaitable<std::string> read_coro() override {
co_return co_await cli.read_coro();
@@ -226,8 +219,7 @@ public:
serial->set_buffer_byte_size(10 * 1024);
bool ok = serial->open();
if (!ok) {
std::cerr << "createSerial " + port_name + ":" +
std::to_string(baud_rate) + " 打开串口失败！\n";
std::cerr << "createSerial " + port_name + ":" + std::to_string(baud_rate) + " 打开串口失败！\n";
}
else {
// std::cerr << "createSerial " + serial_name + ":" +
@@ -236,7 +228,8 @@ public:
co_return;
}
asio::awaitable<void> _close() override {
if (serial) serial->close();
if (serial)
serial->close();
co_return;
}
asio::awaitable<void> handle_in_loop_coro() override {
@@ -254,9 +247,9 @@ public:
Psc::JSON to_json() override {
return Data_Source::to_json() += Serial_Data_Source_Data::to_base_json();
}
void from_json(const Psc::JSON* that_json) override {
Data_Source::from_json(that_json);
Serial_Data_Source_Data::from_base_json(that_json);
void from_json(const Psc::JSON* that_json, bool not_exist_use_default_value) override {
Data_Source::from_json(that_json, not_exist_use_default_value);
Serial_Data_Source_Data::from_base_json(that_json, not_exist_use_default_value);
}
std::unique_ptr<Psc::serial::Serial_Coro> serial{};
~Serial_Data_Source() override {
@@ -291,8 +284,7 @@ public:
Psc::JSON get_custom_state_json() override {
auto cur_virtual_time = player_clock.get_cur_time_point();
auto playback_speed_rate = player_clock.playback_speed_rate;
return VAR_JSON_5(cur_virtual_time, playback_speed_rate, state, index,
part_infos.size());
return VAR_JSON_5(cur_virtual_time, playback_speed_rate, state, index, part_infos.size());
}
File_Data_Source() {
this->type = "File_Data_Source";
@@ -300,7 +292,8 @@ public:
~File_Data_Source() override = default;
void before_handle_msg(std::shared_ptr<SSR::Mode_Msg>& msg) override;
struct Cache_Msg {
explicit Cache_Msg(std::shared_ptr<SSR::Mode_S_Msg> msg) : msg(std::move(msg)) {}
explicit Cache_Msg(std::shared_ptr<SSR::Mode_S_Msg> msg) : msg(std::move(msg)) {
}
SSR::Play_Back_Time_Point time() const {
return SSR::Play_Back_Time_Point{day_num, msg->mlat_timestamp.daysec};
}
@@ -315,17 +308,16 @@ public:
}
asio::awaitable<void> _close() override;
asio::awaitable<void> _open() override;
std::vector<std::string> readBinaryFileAsString(std::string_view filepath,
size_t part_size);
void from_json(const Psc::JSON* that_json) override {
Data_Source::from_json(that_json);
File_Data_Source_Data::from_base_json(that_json);
std::vector<std::string> readBinaryFileAsString(std::string_view filepath, size_t part_size);
void from_json(const Psc::JSON* that_json, bool not_exist_use_default_value) override {
Data_Source::from_json(that_json, not_exist_use_default_value);
File_Data_Source_Data::from_base_json(that_json, not_exist_use_default_value);
}
Psc::JSON to_json() override {
return Data_Source::to_json() += File_Data_Source_Data::to_base_json();
}
void origin_data_transform_mode_data(std::string& data) override;
void handle_mode_s(std::shared_ptr<SSR::Mode_S_Msg> msg) override;
void handle_mode_s(std::shared_ptr<SSR::Mode_S_Msg>& msg) override;
protected:
std::optional<std::string> get_raw_line(int& ret_index);
long long index = 0;
@@ -353,9 +345,9 @@ public:
}
~Dll_Data_Source() override = default;
std::vector<std::uint8_t> buffer;
void from_json(const Psc::JSON* that_json) override {
Data_Source::from_json(that_json);
Dll_Data_Source_Data::from_base_json(that_json);
void from_json(const Psc::JSON* that_json, bool not_exist_use_default_value) override {
Data_Source::from_json(that_json, not_exist_use_default_value);
Dll_Data_Source_Data::from_base_json(that_json, not_exist_use_default_value);
buffer.resize(buffer_size);
}
Psc::JSON to_json() override {
@@ -385,8 +377,7 @@ public:
auto r = Psc::try_load_function(lib, function_name);
if (!r) {
state = r.error().message();
std::cout << LOG_POS << " [" << function_name << "] 函数指针加载失败!"
<< std::endl;
std::cout << LOG_POS << " [" << function_name << "] 函数指针加载失败!" << std::endl;
co_return;
}
read_func_ptr = (Func_Type)r.value();
@@ -408,9 +399,9 @@ public:
Psc::JSON get_custom_state_json() override {
return Psc::JSON::object();
}
void from_json(const Psc::JSON* that_json) override {
Data_Source::from_json(that_json);
Shared_Memory_Data_Source_Data::from_base_json(that_json);
void from_json(const Psc::JSON* that_json, bool not_exist_use_default_value) override {
Data_Source::from_json(that_json, not_exist_use_default_value);
Shared_Memory_Data_Source_Data::from_base_json(that_json, not_exist_use_default_value);
}
Psc::JSON to_json() override {
return Data_Source::to_json() += Shared_Memory_Data_Source_Data::to_base_json();
@@ -423,13 +414,18 @@ protected:
};
inline std::shared_ptr<Data_Source> create_data_source_from_type(std::string_view t) {
std::shared_ptr<Data_Source> ret{};
if (t == "Serial_Data_Source") ret = std::make_shared<Serial_Data_Source>();
else if (t == "TCP_Client_Data_Source") ret = std::make_shared<TCP_Client_Data_Source>();
else if (t == "File_Data_Source") ret = std::make_shared<File_Data_Source>();
else if (t == "Dll_Data_Source") ret = std::make_shared<Dll_Data_Source>();
else if (t == "Shared_Memory_Data_Source") ret = std::make_shared<Shared_Memory_Data_Source>();
if (t == "Serial_Data_Source")
ret = std::make_shared<Serial_Data_Source>();
else if (t == "TCP_Client_Data_Source")
ret = std::make_shared<TCP_Client_Data_Source>();
else if (t == "File_Data_Source")
ret = std::make_shared<File_Data_Source>();
else if (t == "Dll_Data_Source")
ret = std::make_shared<Dll_Data_Source>();
else if (t == "Shared_Memory_Data_Source")
ret = std::make_shared<Shared_Memory_Data_Source>();
else {
std::cout << "未知�?Data_Source type类型!" << std::endl;
std::cout << "未知 Data_Source type类型!" << std::endl;
throw std::invalid_argument("unknown Data_Source type: " + std::string(t));
}
return ret;
@@ -437,10 +433,10 @@ inline std::shared_ptr<Data_Source> create_data_source_from_type(std::string_vie
struct Data_Source_Config {
Ordered_Map<std::string, std::shared_ptr<Data_Source>> map;
Data_Source_Config() = default;
void init(const Psc::JSON* that_json) {
void init(const Psc::JSON* that_json, bool not_exist_use_default_value) {
auto list = that_json->get("list");
for (auto& it : list->children) {
auto ds = create_from_json(&it);
auto ds = create_from_json(&it, not_exist_use_default_value);
map.push_back(ds);
}
}
module/Local_Server/Data_Source/Data_Source_Handler.cpp
+10 -16
View File
@@ -136,7 +136,9 @@ size_t Data_Source_Handler::process_mode_acs_data(std::string_view origin_data)
}
else if (mode_s) {
// 拓展点
handle_mode_s(std::dynamic_pointer_cast<SSR::Mode_S_Msg>(msg));
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);
@@ -156,17 +158,16 @@ size_t Data_Source_Handler::process_mode_acs_data(std::string_view origin_data)
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) {
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) {
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;
auto data = std::format("编解码出问题了(组合的数据，拼接的数据) hex:[{}] [{}] bin:[{}] [{}]",
mem2hex(t, true, " "), mem2hex(mode_s_msg->packet, true, " "), t, mode_s_msg->packet);
Psc::fail_fast(data);
}
source->push_to_feed(mode_s_msg);
auto& cfg = Global::instance()->mode_acs;
bool time_space_filter = cfg.time_space_filter.load();
bool speed_filter = cfg.speed_filter.load();
@@ -180,16 +181,9 @@ void Data_Source_Handler::handle_mode_s(std::shared_ptr<SSR::Mode_S_Msg> mode_s_
SSR::ADS_B_T::Constraint air_constraint{cfg.max_speed_m_s, cfg.air_pos_timeout, SSR::cpr_cb, time_space_filter,
speed_filter, use_system_time, base_station_pos, source->alt,
range_filter, range_factor};
SSR::ADS_B_T::Constraint surface_constraint{cfg.max_speed_m_s,
cfg.surface_pos_timeout,
SSR::cpr_cb,
time_space_filter,
speed_filter,
use_system_time,
base_station_pos,
source->alt,
range_filter,
range_factor};
SSR::ADS_B_T::Constraint surface_constraint{
cfg.max_speed_m_s, cfg.surface_pos_timeout, SSR::cpr_cb, time_space_filter, speed_filter,
use_system_time, base_station_pos, source->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) {
module/Local_Server/Data_Source/Data_Source_Handler.h
+25 -27
View File
@@ -8,35 +8,33 @@
#include "Core/transmit_protocol/core/statistics.h"
#include "Database.h"
#include "Local_Server/global_include.h"
#include "global.h"
#include "PlayBack.h"
#include "global.h"
// 这个函数专门装cpu密集函数 放到线程池去处理
class Data_Source_Handler : public DataBase {
public:
std::shared_ptr<SSR::Msg> create_msg(std::string_view packet);
virtual void origin_data_transform_mode_data(std::string& origin_data) {};
virtual void before_handle_msg(std::shared_ptr<SSR::Mode_Msg>& msg) {};
virtual void handle_mode_s(std::shared_ptr<SSR::Mode_S_Msg> msg);
void handle_HULC(std::string_view packet);
size_t process_mode_acs_data(std::string_view 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;
~Data_Source_Handler() override = default;
// 推送到 data_feed 相关代码
void push_to_feed(const std::shared_ptr<SSR::Msg>& msg);
void refresh_data_feed_key_list();
struct Cached_Source_Info {
std::string key;
Psc::Value_Statistics mode_s_cache_num;
Psc::Value_Statistics mode_other_cache_num;
};
std::atomic<bool> need_refresh_data_feed_key_list = true;
std::vector<Cached_Source_Info> cached_data_feed_key_list{};
Psc::JSON get_all_connect_feed_status();
Psc::JSON get_all_connect_feed();
std::mutex cdf_mtx;
std::shared_ptr<SSR::Msg> create_msg(std::string_view packet);
virtual void origin_data_transform_mode_data(std::string& origin_data) {};
virtual void before_handle_msg(std::shared_ptr<SSR::Mode_Msg>& msg) {};
virtual void handle_mode_s(std::shared_ptr<SSR::Mode_S_Msg>& msg);
void handle_HULC(std::string_view packet);
size_t process_mode_acs_data(std::string_view 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;
~Data_Source_Handler() override = default;
// 推送到 data_feed 相关代码
void push_to_feed(const std::shared_ptr<SSR::Msg>& msg);
void refresh_data_feed_key_list();
struct Cached_Source_Info {
std::string key;
Psc::Value_Statistics mode_s_cache_num;
Psc::Value_Statistics mode_other_cache_num;
};
std::atomic<bool> need_refresh_data_feed_key_list = true;
std::vector<Cached_Source_Info> cached_data_feed_key_list{};
Psc::JSON get_all_connect_feed_status();
Psc::JSON get_all_connect_feed();
std::mutex cdf_mtx;
};