ECAP_Server/module/Local_Server/DSP/DSP_Config.cpp

#include "Local_Server/DSP/DSP_Config.h"
#include "../server/Global.h"
#include "Psc_Cpp_Core/socket/ByteOrder.h"
#include <drogon/HttpAppFramework.h>
#include <drogon/PubSubService.h>
#include <drogon/WebSocketController.h>
#if !USE_PROCESS
// adsb_set_less_30Mhz_ddc_param_t set_less_30Mhz_ddc_param;
// adsb_set_less_30Mhz_fft_param_t set_less_30Mhz_fft_param;
decltype(&adsb_set_less_30Mhz_ddc_param) psc_set_less_30Mhz_ddc_param;
decltype(&adsb_set_less_30Mhz_fft_param) psc_set_less_30Mhz_fft_param;
#endif
using namespace drogon;
// ws://127.0.0.1:8848/chat
BaseLogger* dsp_logger = nullptr;
DSP_Config::~DSP_Config() = default;

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

void DSP_Config::server(Global* g) {
	auto& svr = g->svr;
	auto& api = g->api;
	std::string name = "dsp";
	drogon::app().registerWebSocketControllerRegex(R"(^/websocket/.*$)",
	                                               "DSP_Ws_Mgr", {Get});
	ws_iq = Ws_Mgr::instance()->register_ws("/download_iq");
	ws_fft = Ws_Mgr::instance()->register_ws("/get_fft");
	svr.Post(api + svr.update + name, [this, g](HTTP_Param) {
		CHECK_JSON_PARAM
		init_update(&params);
#if USE_PROCESS
		adsb_set_less_30Mhz_ddc_param(this->center_freq, this->sample_rate,
		                              this->band_width, 1);
		adsb_set_less_30Mhz_fft_param(fft_win_type, fft_point_number, fft_period_ms,
		                              1);
#else
		psc_set_less_30Mhz_ddc_param(this->center_freq, this->sample_rate, this->band_width, 1);
		psc_set_less_30Mhz_fft_param(fft_win_type, fft_point_number, fft_period_ms, 1);
#endif
		set_fft_point_number(this->fft_point_number);
		g->save();
		res->setBody(warp(to_base_json()).to_json_string());
	});
	svr.Post(api + "get_dsp_config", [this](HTTP_Param) {
		res->setBody(warp(to_base_json()).to_json_string());
	});
	drogon::app().registerHandler(
		api + "start_record_iq_stream",
		[this](const drogon::HttpRequestPtr& request,
		       std::function<void(const drogon::HttpResponsePtr&)>&& callback) {
			this->record_iq_stream_ing = true;
			auto res = drogon::HttpResponse::newHttpResponse();
			res->setBody("{}");
			callback(res);
		},
		{drogon::Post});
	svr.Post(api + "end_record_iq_stream", [this](HTTP_Param) {
		this->record_iq_stream_ing = false;
		res->setBody("{}");
	});
	svr.Post(api + "restart_device", [this](HTTP_Param) {
#if USE_PROCESS
		restart_device();
#endif
	});
}

void DSP_Config::set_fft_point_number(uint64_t new_fft_point_number,
                                      bool force) {
	fft_point_number = new_fft_point_number;
#if !USE_PROCESS
	std::lock_guard<std::mutex> lock(fft_mtx);
	if (!force && fft_data.size() == (new_fft_point_number * sizeof(short))) {
		return;
	}
	fft_data.resize(new_fft_point_number * sizeof(short));
#endif
}
#if USE_PROCESS
void DSP_Config::adsb_set_less_30Mhz_ddc_param(double center_freq,
                                               double sample_rate,
                                               double band_width, int sw) {
	JSON ret = JSON::object();
	ret.append(JSON("method", "adsb_set_less_30Mhz_ddc_param"));
	ret.append(JSON("center_freq", center_freq));
	ret.append(JSON("sample_rate", sample_rate));
	ret.append(JSON("band_width", band_width));
	ret.append(JSON("sw", sw));
	auto d = ret.to_json_string();
	json_func_call.write((uint8_t*)d.data(), d.size());
}
void DSP_Config::adsb_set_less_30Mhz_fft_param(UHD_FFTWin fft_win_type,
                                               size_t fft_point_number,
                                               size_t fft_period_ms, int sw) {
	JSON ret = JSON::object();
	ret.append(JSON("method", "adsb_set_less_30Mhz_fft_param"));
	ret.append(JSON("fft_win_type", fft_win_type));
	ret.append(JSON("fft_point_number", fft_point_number));
	ret.append(JSON("fft_period_ms", fft_period_ms));
	ret.append(JSON("sw", sw));
	auto d = ret.to_json_string();
	json_func_call.write((uint8_t*)d.data(), d.size());
}
void DSP_Config::restart_device() {
	JSON ret = JSON::object();
	ret.append(JSON("method", "restart_device"));
	auto d = ret.to_json_string();
	json_func_call.write((uint8_t*)d.data(), d.size());
}
#endif
#define USE_ERROR_CBK

void DSP_Config::init_env() {
	set_fft_point_number(fft_point_number, true);
	auto g = Global::instance();
	Detach_Thread_Manager& manager = g->thread_manager;
	auto handle_iq = [this](void* data) {
		uint8_t* pdata = (uint8_t*)data;
		auto wi = Ws_Mgr::instance();
		if (!wi)
			return;
		auto d = (UHD_Struct_KFFT*)(pdata);
		if (debug) {
			std::ostringstream oss;
			oss << "UHD_Struct_KFFT{"
			<< mem2hex(std::string((char*)pdata, 64), true, " ") << std::endl;
			oss << "DataType{" << (int)d->DataType << "}  接收到数据:["
			<< "] == " << mem2hex(std::string((char*)pdata + 64, 20), true, " ")
			<< std::endl;
			server_logger->c_debug({}, {}, oss.str());
		}
		if (d->DataType == zd_ddc || d->DataType == kd_ddc) {
			auto len = d->UsefullLen;
			bool cross_the_boundary = len > 2048;
			if (cross_the_boundary || debug) {
				auto show_len = (size_t)std::min((decltype(len))20, len);
				std::ostringstream oss;
				oss << "iq 有效数据长度 【" << len << "】 有效数据： 【"
				<< std::string((char*)pdata + 64, show_len) << "】" << std::endl;
				server_logger->c_debug({}, {}, oss.str());
			}
			if (cross_the_boundary) {
				std::ostringstream oss;
				oss << LOG_POS << "ERROR: ddc iq 回调返回的有效长度越界！" << std::endl;
				server_logger->c_debug({}, {}, oss.str());
				return;
			}
			if (record_iq_stream_ing) {
				ws_iq->push_data((const char*)pdata, 2112);
			}
		}
		else if (d->DataType == kd_fft) {
#if !USE_PROCESS
			auto idx = d->Param.FFT.SubFrame;
			auto max = fft_point_number / 1024;
			if (idx >= max) {
				std::ostringstream oss;
				oss << "收到了不合法kd_fft  Param.FFT.SubFrame  " << idx << std::endl;
				server_logger->c_debug({}, {}, oss.str());
				return;
			}
			std::lock_guard<std::mutex> lock(fft_mtx);
			auto pos = fft_data.data() + idx * 2048;
			memcpy(pos, pdata + 64, 2048);
#endif
		}
	};
#if USE_PROCESS
	manager.test_and_start_thread(
		"读取iq进程数据线程", [this, handle_iq](std::atomic<bool>& running) {
			while (running) {
				uint8_t buf[3000];
				size_t length = 3000;
				while (rb.read(buf, length)) {
					Frequency_Limit lm(0.1);
					if (lm.test()) {
						std::cout << "读取到iq数据" << std::endl;
					}
					handle_iq(buf);
					// auto data = std::string((char *) buf, length);
					//  if (length != 0) {
					//      std::cout << "read:【" << length << "】  【" <<
					//      mem2hex(data.substr(64, 20)) << "】" << std::endl;
					//  }
					if (!running) {
						break;
					}
				}
				std::this_thread::sleep_for(std::chrono::microseconds(1));
			}
		});
#else
	manager.test_and_start_thread("iq回调处理", [this, handle_iq](
		                              std::atomic<bool>& running) {
		                              Frequency_Limit_ST limit(frame_rate);
		                              while (running) {
			                              auto list = iq_memory_buffer.get_all();
			                              if (debug && list.size() > 1000) {
				                              std::ostringstream oss;
				                              oss << "警告： 数据过多 当前数据队列长度:" << list.size() << std::endl;
				                              server_logger->c_debug({}, {}, oss.str());
			                              }
			                              if (list.empty()) {
				                              std::this_thread::sleep_for(std::chrono::microseconds(1));
				                              continue;
			                              }
			                              auto wi = Ws_Mgr::instance();
			                              if (!wi)
				                              continue;
			                              auto frame_number = list.size();
			                              for (int i = 0; i < frame_number; i++) {
				                              auto pdata = list[i].data();
				                              handle_iq(pdata);
			                              }
		                              }
	                              });
#endif
	auto interval = 1.0 / (double)frame_rate;
	app().getLoop()->runEvery(interval, [this] {
#if USE_PROCESS
		wi->ws_get_fft->push_data((const char*)psm.shm.data(),
		                          fft_point_number * 2);
#else
		std::lock_guard<std::mutex> lock(fft_mtx);
		ws_fft->push_data((const char*)fft_data.data(), fft_data.size());
#endif
	});
#if !USE_PROCESS
	{
		auto r = try_load_library(Psc::get_abs_path(library_path));
		if (!r) {
			Psc::fail_fast();
		}
		lib = r.value();
	}
	{
		auto r = Psc::try_load_function(lib, "adsb_set_less_30Mhz_ddc_param");
		if (!r) {
			Psc::fail_fast();
		}
		psc_set_less_30Mhz_ddc_param = (adsb_set_less_30Mhz_ddc_param_t)r.value();
	}
	{
		auto r = Psc::try_load_function(lib, "adsb_set_less_30Mhz_fft_param");
		if (!r) {
			Psc::fail_fast();
		}
		psc_set_less_30Mhz_fft_param = (adsb_set_less_30Mhz_fft_param_t)r.value();
	}
	using set_iq_cbk_t = decltype(&adsb_set_iq_cbk);
	set_iq_cbk_t set_iq_cbk;
	{
		auto r = Psc::try_load_function(lib, "adsb_set_iq_cbk");
		if (!r) {
			Psc::fail_fast();
		}
		set_iq_cbk = (set_iq_cbk_t)r.value();
	}
	using adsb_set_error_cbk_t = decltype(&set_error_cbk);
	adsb_set_error_cbk_t adsb_set_error_cbk;
	{
		auto r = Psc::try_load_function(lib, "set_error_cbk");
		if (!r) {
			Psc::fail_fast();
		}
		adsb_set_error_cbk = (adsb_set_error_cbk_t)r.value();
	}
	psc_set_less_30Mhz_ddc_param(center_freq, sample_rate, band_width, 1);
	psc_set_less_30Mhz_fft_param(fft_win_type, fft_point_number, fft_period_ms,
	                             1);
	set_iq_cbk(
		[](void* phandle, void* pdata, uint64_t length) {
			if (stop_program != 0) {
				static Frequency_Limit fl;
				if (fl.test()) {
					std::ostringstream oss;
					oss << "set_iq_cbk 回调退出：" << signal_to_string(stop_program)
					<< "  " << VAR_STR_1(stop_program) << LOG_POS_SIMPLE << std::endl;
					dsp_logger->c_debug({}, {}, oss.str());
				}
			}
			if (length != 2112) {
				std::ostringstream oss;
				oss << LOG_POS << "  IQ回调传入的长度不对!" << std::endl;
				dsp_logger->c_debug({}, {}, oss.str());
				return;
			}
			auto that = (DSP_Config*)phandle;
			that->iq_memory_buffer.push(pdata, length); // length 永远等于2112
		},
		(void*)&Global::instance()->dsp_config);
#ifdef USE_ERROR_CBK
	adsb_set_error_cbk([](const char* file_name, const char* func_name,
	                      const char* err_msg) {
		if (stop_program != 0) {
			static Frequency_Limit fl(3.0);
			if (fl.test()) {
				std::ostringstream oss;
				oss << "adsb_set_error_cbk 回调退出：" << signal_to_string(stop_program)
				<< "  " << VAR_STR_1(stop_program) << LOG_POS << std::endl;
				dsp_logger->c_debug({}, {}, oss.str());
			}
			return;
		}
		Log_Type lt{
			{},
			{
				{"file_name", std::string(file_name)},
				{"func_name", std::string(func_name)},
			}
		};
		dsp_logger->debug({}, lt, std::string(err_msg));
	});
#endif
#else
	adsb_set_less_30Mhz_ddc_param(center_freq, sample_rate, band_width, 1);
	adsb_set_less_30Mhz_fft_param(fft_win_type, fft_point_number, fft_period_ms,
	                              1);
#endif
}