Compare commits

7 Commits

6 changed files with 325 additions and 47 deletions

View File

@@ -3037,39 +3037,57 @@ bool extract_timestamp_from_cfg_file(const std::string& cfg_path, long long& sta
return start_tm > 0 && trig_tm > 0;
}
bool compare_qvvr_and_file(const std::string& cfg_path, std::vector<qvvr_data>& data_list,qvvr_data& matched_data) {
bool compare_qvvr_and_file(const std::string& cfg_path,
std::vector<qvvr_data>& data_list,
qvvr_data& matched_data) {
long long start_tm = 0;
long long trig_tm = 0;
// 提取 .cfg 文件中的时间戳
if (!extract_timestamp_from_cfg_file(cfg_path, start_tm, trig_tm)) {
std::cerr << "Failed to extract timestamp from cfg file: " << cfg_path << "\n";
return false;
}
//打印提取到的时间戳
std::cout << "[调试] 提取到的起始时间戳: " << start_tm << ", 触发时间戳: " << trig_tm << "\n";
std::cout << "[调试] 提取到的起始时间戳: " << start_tm
<< ", 触发时间戳: " << trig_tm << "\n";
// 遍历所有暂态事件,查找与 trig_tm 匹配的
for (auto& data : data_list) {
long long diff = static_cast<long long>(data.QVVR_time) - trig_tm;
const long long EIGHT_HOURS_MS = 8LL * 60 * 60 * 1000;
std::cout << "[调试] QVVR_time=" << data.QVVR_time
<< ", trig_tm=" << trig_tm
<< ", diff=" << diff << "\n";
std::vector<long long> candidates;
candidates.push_back(trig_tm); // 原始时间
candidates.push_back(trig_tm - EIGHT_HOURS_MS); // 减 8 小时
candidates.push_back(trig_tm + EIGHT_HOURS_MS); // 加 8 小时
if (std::abs(diff) <= 1) {
for (size_t i = 0; i < candidates.size(); ++i) {
long long cmp_tm = candidates[i];
data.is_pair = true; // 标记为已匹配
std::cout << "[调试] 开始匹配候选触发时间: " << cmp_tm
<< " offset_ms=" << (cmp_tm - trig_tm) << "\n";
matched_data = data; // 返回匹配到的事件
for (auto& data : data_list) {
long long diff =
static_cast<long long>(data.QVVR_time) - cmp_tm;
std::cout << "[调试] 匹配成功diff=" << diff << "\n";
std::cout << "[调试] QVVR_time=" << data.QVVR_time
<< ", cmp_tm=" << cmp_tm
<< ", diff=" << diff << "\n";
return true;
if (std::abs(diff) <= 1) {
data.is_pair = true;
matched_data = data;
std::cout << "[调试] 匹配成功cmp_tm=" << cmp_tm
<< ", offset_ms=" << (cmp_tm - trig_tm)
<< ", diff=" << diff << "\n";
return true;
}
}
}
std::cout << "[调试] 匹配失败,原始 trig_tm=" << trig_tm
<< ",已尝试 trig_tm、trig_tm-8h、trig_tm+8h\n";
return false;
}
@@ -3674,7 +3692,8 @@ void check_device_busy_timeout()
{
dev.busytimecount++;
if (dev.busytype == static_cast<int>(DeviceState::READING_FILEDATA) || dev.busytype == static_cast<int>(DeviceState::READING_STATSFILE)) //下载文件业务
if (dev.busytype == static_cast<int>(DeviceState::READING_FILEDATA) || dev.busytype == static_cast<int>(DeviceState::READING_STATSFILE)
|| dev.busytype == static_cast<int>(DeviceState::SET_PREUPGRADE) || dev.busytype == static_cast<int>(DeviceState::SET_UPGRADE)) //下载文件业务
{
if (dev.busytimecount > 600)
{
@@ -4925,7 +4944,7 @@ void check_recall_file() {
std::vector<RecallTask> tasks; // 本轮要发送的“每终端一条”(目录请求 或 文件下载 请求)
// ★修改开始:新增“待上传动作”容器与两个小工具(局部作用域,函数私有)
// “待上传动作”容器与两个小工具(局部作用域,函数私有)
struct PendingUpload {
std::string terminal_id;
unsigned short logical_seq = 0;
@@ -5577,11 +5596,17 @@ void check_recall_file() {
}
//更新事件
// ★修改开始替换“assign_qvvr_file_list + update_qvvr_file_download(有锁)”
// 组装完整路径列表
std::vector<std::string> fullFilenames;
fullFilenames.reserve(front.required_files.size());
for (const auto& f : front.required_files) fullFilenames.push_back(f);
const std::string base_dir = std::string("download/") + sanitize(dev.addr_str);
for (const auto& remote_file : front.required_files) {
std::string fname = sanitize(extract_filename1(remote_file));
if (!fname.empty()) {
fullFilenames.push_back(base_dir + "/" + fname);
}
}
// 仅对“当前监测点”添加一组 qvvr_filefile_name=仅文件名,下载标记留到下行)
{
@@ -7851,7 +7876,11 @@ bool SendFileWebAuto(const std::string& id,
<< " -> use default upload URL (cloud path=" << file_cloudpath << ")\n";
// 实际上传调用
SendFileWeb(WEB_FILEUPLOAD, local_path, file_cloudpath, out_filename,2);
if (!SendFileWeb(WEB_FILEUPLOAD, local_path, file_cloudpath, out_filename,2)) {
std::cerr << "[SendFileWebAuto][ERROR] File upload failed: "
<< local_path << std::endl;
return false;
}
std::cout << "[SendFileWebAuto] File upload complete: " << out_filename << std::endl;

View File

@@ -47,6 +47,14 @@ std::vector<terminal_dev> terminal_devlist;
//台账锁
std::mutex ledgermtx;
///////////////////////////////////异步接口队列
std::mutex g_qvvr_async_mtx;
std::list<QvvrFileDownloadTask> g_qvvr_file_download_list;
std::list<QvvrJsonTask> g_qvvr_json_list;
///////////////////////////////////
/////////////////////////////////////////////////////////////////////////////////////////////////////////////////
extern int g_front_seg_num;
@@ -211,7 +219,7 @@ bool DownloadFileAPI_web(const std::string& strUrl, // 接口路径
////////////////////////////////////////////////////////////////////////////////////////////////////////上传文件接口
//处理文件上传响应
void handleUploadResponse(const std::string& response, std::string& wavepath, int type) {
bool handleUploadResponse(const std::string& response, std::string& wavepath, int type) {
using nlohmann::json; //把 nlohmann::json 这个名字带到当前作用域
@@ -223,27 +231,34 @@ void handleUploadResponse(const std::string& response, std::string& wavepath, in
catch (const json::parse_error& e) {
std::cerr << "Error parsing response: " << e.what() << std::endl;
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON, "文件上传接口响应异常,上传文件失败");
return;
return false;
}
// 提取字段
if (!json_data.contains("code") || !json_data.contains("data")) {
std::cerr << "Error: Missing expected fields in JSON response." << std::endl;
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON, "文件上传接口响应异常,上传文件失败");
return;
return false;
}
std::string code = json_data["code"].get<std::string>();
std::cout << "Response Code: " << code << std::endl;
std::string msg = json_data.value("msg", std::string{"not found"});
std::string msg = json_data.value("message",
json_data.value("msg", std::string{"not found"}));
std::cout << "Message: " << msg << std::endl;
if (code != "A0000") {
std::cerr << "File upload failed, response code=" << code
<< ", message=" << msg << std::endl;
return false;
}
auto& data = json_data["data"];
if (!data.contains("name") || !data.contains("fileName") || !data.contains("url")) {
std::cerr << "Error: Missing expected fields in JSON data object." << std::endl;
DIY_ERRORLOG_CODE("process",0,LOG_CODE_JSON, "文件上传接口响应异常,上传文件失败");
return;
return false;
}
std::string name = data["name"].get<std::string>();
@@ -280,20 +295,24 @@ void handleUploadResponse(const std::string& response, std::string& wavepath, in
std::cout << "wavepath: " << wavepath << std::endl;
DIY_INFOLOG_CODE("process",0,LOG_CODE_FILE, "上传文件成功,远端文件名:%s", wavepath.c_str());
return true;
}
//上传文件
void SendFileWeb(const std::string& strUrl, const std::string& localpath, const std::string& cloudpath, std::string& wavepath, int type) {
bool SendFileWeb(const std::string& strUrl, const std::string& localpath, const std::string& cloudpath, std::string& wavepath, int type) {
wavepath.clear();
// 基本存在性检查
if (access(localpath.c_str(), F_OK) != 0) {
std::cerr << "Local file does not exist: " << localpath << std::endl;
return;
return false;
}
// ★新增stat 打印大小,便于快速确认读源
struct stat st {};
if (stat(localpath.c_str(), &st) != 0) {
perror("stat");
return false;
} else {
std::cout << "[debug] upload file: " << localpath
<< ", size=" << static_cast<long long>(st.st_size) << " bytes\n";
@@ -351,6 +370,7 @@ void SendFileWeb(const std::string& strUrl, const std::string& localpath, const
curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, req_reply_web);
// 执行请求
bool upload_ok = false;
CURLcode res = curl_easy_perform(curl);
if (res != CURLE_OK) {
const char* em = errbuf[0] ? errbuf : curl_easy_strerror(res);
@@ -358,16 +378,18 @@ void SendFileWeb(const std::string& strUrl, const std::string& localpath, const
DIY_ERRORLOG_CODE("process",0,LOG_CODE_CONFIG, "通过文件接口上传文件 %s 失败",localpath.c_str());
} else {
std::cout << "http web success, response: " << resPost0 << std::endl;
handleUploadResponse(resPost0, wavepath, type); // 处理响应
upload_ok = handleUploadResponse(resPost0, wavepath, type); // 处理响应
}
// 清理
curl_formfree(formpost); // 释放表单数据
//curl_slist_free_all(header_list); // 释放头部列表
curl_easy_cleanup(curl);
return upload_ok;
} else {
std::cerr << ">>> curl init failed" << std::endl;
}
return false;
}
//上传暂态文件
@@ -1699,3 +1721,148 @@ bool get_monitor_name_by_monitor_id(const std::string& monitor_id,
out_monitor_name.clear();
return false;
}
/////////////////////////////////////////////////异步接口
// ★新增:代替直接调用 update_qvvr_file_download()
void enqueue_qvvr_file_download(const std::string& file_path,
const std::string& terminal_id)
{
std::lock_guard<std::mutex> lk(g_qvvr_async_mtx);
QvvrFileDownloadTask task;
task.file_path = file_path;
task.terminal_id = terminal_id;
g_qvvr_file_download_list.push_back(task);
std::cout << "[QVVR_ASYNC] enqueue file_download terminal="
<< terminal_id
<< " file=" << file_path
<< " size=" << g_qvvr_file_download_list.size()
<< std::endl;
}
// ★新增:代替直接调用 transfer_json_qvvr_data()
void enqueue_qvvr_json_update(const std::string& terminal_id,
int line_id,
double amplitude,
double persist_time,
long long trigger_time_ms,
int type,
int phase,
const std::string& wavepath)
{
std::lock_guard<std::mutex> lk(g_qvvr_async_mtx);
QvvrJsonTask task;
task.terminal_id = terminal_id;
task.line_id = line_id;
task.amplitude = amplitude;
task.persist_time = persist_time;
task.trigger_time_ms = trigger_time_ms;
task.type = type;
task.phase = phase;
task.wavepath = wavepath;
g_qvvr_json_list.push_back(task);
std::cout << "[QVVR_ASYNC] enqueue json terminal="
<< terminal_id
<< " line=" << line_id
<< " time=" << trigger_time_ms
<< " size=" << g_qvvr_json_list.size()
<< std::endl;
}
// ★新增FrontThread 中调用,消费异步接口任务
void process_qvvr_async_tasks()
{
// 每轮最多处理几个,避免一个接口卡住太久
const int MAX_FILE_TASK_PER_ROUND = 1;
const int MAX_JSON_TASK_PER_ROUND = 3;
// 1. 处理 update_qvvr_file_download 队列
for (int n = 0; n < MAX_FILE_TASK_PER_ROUND; ++n) {
QvvrFileDownloadTask task;
{
std::lock_guard<std::mutex> lk(g_qvvr_async_mtx);
if (g_qvvr_file_download_list.empty()) {
break;
}
task = g_qvvr_file_download_list.front();
g_qvvr_file_download_list.pop_front();
}
try {
std::cout << "[QVVR_ASYNC] call update_qvvr_file_download terminal="
<< task.terminal_id
<< " file=" << task.file_path
<< std::endl;
update_qvvr_file_download(task.file_path, task.terminal_id);
}
catch (const std::exception& e) {
std::cerr << "[QVVR_ASYNC][ERROR] update_qvvr_file_download exception: "
<< e.what()
<< " terminal=" << task.terminal_id
<< " file=" << task.file_path
<< std::endl;
}
catch (...) {
std::cerr << "[QVVR_ASYNC][ERROR] update_qvvr_file_download unknown exception"
<< " terminal=" << task.terminal_id
<< " file=" << task.file_path
<< std::endl;
}
}
// 2. 处理 transfer_json_qvvr_data 队列
for (int n = 0; n < MAX_JSON_TASK_PER_ROUND; ++n) {
QvvrJsonTask task;
{
std::lock_guard<std::mutex> lk(g_qvvr_async_mtx);
if (g_qvvr_json_list.empty()) {
break;
}
task = g_qvvr_json_list.front();
g_qvvr_json_list.pop_front();
}
try {
std::cout << "[QVVR_ASYNC] call transfer_json_qvvr_data terminal="
<< task.terminal_id
<< " line=" << task.line_id
<< " time=" << task.trigger_time_ms
<< " wavepath=" << task.wavepath
<< std::endl;
transfer_json_qvvr_data(task.terminal_id,
task.line_id,
task.amplitude,
task.persist_time,
task.trigger_time_ms,
task.type,
task.phase,
task.wavepath);
}
catch (const std::exception& e) {
std::cerr << "[QVVR_ASYNC][ERROR] transfer_json_qvvr_data exception: "
<< e.what()
<< " terminal=" << task.terminal_id
<< " line=" << task.line_id
<< std::endl;
}
catch (...) {
std::cerr << "[QVVR_ASYNC][ERROR] transfer_json_qvvr_data unknown exception"
<< " terminal=" << task.terminal_id
<< " line=" << task.line_id
<< std::endl;
}
}
}

View File

@@ -32,6 +32,44 @@ class Front;
#define RECALL_HIS_DATA_BASE_NODE_ID 600
#define RECALL_ALL_DATA_BASE_NODE_ID 700*/
///////////////////////////////////////////////////////////////////////////////////////////异步接口
struct QvvrFileDownloadTask {
std::string file_path;
std::string terminal_id;
};
struct QvvrJsonTask {
std::string terminal_id;
int line_id;
double amplitude;
double persist_time;
long long trigger_time_ms;
int type;
int phase;
std::string wavepath;
};
// ★声明
extern std::mutex g_qvvr_async_mtx;
extern std::list<QvvrFileDownloadTask> g_qvvr_file_download_list;
extern std::list<QvvrJsonTask> g_qvvr_json_list;
extern void process_qvvr_async_tasks();
extern void enqueue_qvvr_file_download(const std::string& file_path,
const std::string& terminal_id);
extern void enqueue_qvvr_json_update(const std::string& terminal_id,
int line_id,
double amplitude,
double persist_time,
long long trigger_time_ms,
int type,
int phase,
const std::string& wavepath);
///////////////////////////////////////////////////////////////////////////////////////////
//单条补招时间结构
@@ -759,7 +797,7 @@ bool save_internal_value(const std::string &dev_id, const std::vector<ushort> &f
bool save_set_value(const std::string &dev_id, unsigned char mp_index, const std::vector<float> &fabsf);
//发送文件
void SendFileWeb(const std::string& strUrl, const std::string& localpath, const std::string& cloudpath, std::string& wavepath,int Type);
bool SendFileWeb(const std::string& strUrl, const std::string& localpath, const std::string& cloudpath, std::string& wavepath,int Type);
//状态翻转
void connect_status_update(const std::string& id, int status);
@@ -1110,4 +1148,4 @@ bool runninginfo_cache_take(const std::string& dev_id, RunningInformation& out);
void versioninfo_cache_put(const std::string& dev_id, const DeviceVersionInfo& info);
bool versioninfo_cache_take(const std::string& dev_id, DeviceVersionInfo& out);
#endif
#endif

View File

@@ -473,6 +473,9 @@ void Front::FrontThread() {
check_recall_event(); // 处理补招事件从list中读取然后直接调用接口,每一条可能都不同测点,每个测点自己做好记录
check_recall_file(); //处理补招文件-稳态和暂态
// ★新增:异步处理 QVVR 接口任务
process_qvvr_async_tasks();
std::this_thread::sleep_for(std::chrono::milliseconds(100));
}

View File

@@ -38,7 +38,7 @@ AccessKey=rmqroot
SecretKey=001@#njcnmq
Topic_Test=lnk_Topic
Tag_Test=Test_Tag
Tag_Test=914b94563ca7f272c90ee8580ed6adc6
Key_Test=Test_Keys
Testflag=1
Testnum=0
@@ -48,7 +48,7 @@ TestList=
consumer=Group_consumer
ConsumerIpport=192.168.1.103:9876
ConsumerTopicRT=ask_real_data_topic
ConsumerTopicRT=ask_real_data_Topic
ConsumerTagRT=Test_Tag
ConsumerKeyRT=Test_Keys
ConsumerAccessKey=rmqroot
@@ -57,7 +57,7 @@ ConsumerChannel=
ConsumerTopicUD=control_Topic
ConsumerTagUD=Test_Tag
ConsumerKeyUD=Test_Keys
ConsumerTopicRC=recall_Topic
ConsumerTopicRC=ask_recall_Topic
ConsumerTagRC=Test_Tag
ConsumerKeyRC=Test_Keys
ConsumerTopicSET=process_Topic
@@ -75,6 +75,15 @@ CONNECTKey=Test_Keys
Heart_Beat_Topic=Heart_Beat_Topic
Heart_Beat_Tag=Test_Tag
Heart_Beat_Key=Test_Key
Topic_Reply_Topic=Topic_Reply_Topic
Topic_Reply_Topic=Reply_Topic
Topic_Reply_Tag=Test_Tag
Topic_Reply_Key=Test_Key
ConsumerTopicLOG=ask_log_Topic
ConsumerTagLOG=Test_Tag
ConsumerKeyLOG=Test_Keys
ConsumerTopicCLOUD=Cloud_Topic
ConsumerTagCLOUD=Cloud_Tag
ConsumerKeyCLOUD=Cloud_Keys
Cloud_Reply_Topic=Cloud_Reply_Topic
Cloud_Reply_Tag=Cloud_Reply_Tag
Cloud_Reply_Key=Cloud_Reply_Key

View File

@@ -350,11 +350,14 @@ void process_received_message(string mac, string id,const char* data, size_t len
<< ", 时间戳: " << record.triggerTimeMs << "ms" << std::endl;
//lnk20250805 事件上送先记录,录波文件上传结束后再更新文件
append_qvvr_event(id,event.head.name,
if(record.nType != 0){
append_qvvr_event(id,event.head.name,
record.nType,record.fPersisstime,record.fMagntitude,record.triggerTimeMs,record.phase);
transfer_json_qvvr_data(id,event.head.name,
/*transfer_json_qvvr_data(id,event.head.name,
record.fMagntitude,record.fPersisstime,record.triggerTimeMs,record.nType,record.phase,
"");
"");*/
enqueue_qvvr_json_update(id, event.head.name, record.fMagntitude, record.fPersisstime, record.triggerTimeMs, record.nType, record.phase, "");
}
//事件主动上送处理完成,不需要通知状态机
}
@@ -462,6 +465,9 @@ void process_received_message(string mac, string id,const char* data, size_t len
);
std::cout << "Added download request for: " << filename << std::endl;
DIY_INFOLOG_CODE(id, 1, static_cast<int>(LogCode::LOG_CODE_DEV_ALARM),
"添加装置暂态波形文件下载请求: %s",
filename.c_str());
}
//最后报文收发处理逻辑(如果当前装置空闲则尝试执行后续动作)(如果当前装置存在其他状态则直接退出,不要干扰装置后续执行)
@@ -840,10 +846,20 @@ void process_received_message(string mac, string id,const char* data, size_t len
if (out_file) {
out_file.write(reinterpret_cast<const char*>(file_data.data()),
file_data.size());
std::cout << "File saved: " << file_path << std::endl;
// SendFileWebAuto reopens this path. Close first so data buffered by
// ofstream (especially for small files) is visible to stat/libcurl.
out_file.close();
if (!out_file) {
std::cerr << "Failed to write/close file: " << file_path << std::endl;
ClientManager::instance().change_device_state(id, DeviceState::IDLE);
return;
}
std::cout << "File saved: " << file_path
<< ", size=" << file_data.size() << " bytes" << std::endl;
//lnk20250805文件保存成功调用录波文件上送接口
update_qvvr_file_download(file_path, id);
//update_qvvr_file_download(file_path, id);
enqueue_qvvr_file_download(file_path, id);
}
else {
std::cerr << "Failed to save file: " << file_path
@@ -860,6 +876,9 @@ void process_received_message(string mac, string id,const char* data, size_t len
}
else {
// 装置答非所问异常
DIY_INFOLOG_CODE(id, 1, static_cast<int>(LogCode::LOG_CODE_DEV_ALARM),
"装置答非所问异常,接收波形文件数据错误,错误码:%d",
static_cast<int>(udata[8]));
//on_device_response_minimal(static_cast<int>(ResponseCode::INTERNAL_ERROR), id, 0, static_cast<int>(DeviceState::READING_EVENTFILE));
// 接收波形文件数据错误,调整为空闲状态,处理下一项工作。
ClientManager::instance().change_device_state(id, DeviceState::IDLE);
@@ -1161,7 +1180,18 @@ void process_received_message(string mac, string id,const char* data, size_t len
if (out_file) {
out_file.write(reinterpret_cast<const char*>(file_data.data()),
file_data.size());
std::cout << "File saved: " << file_path << std::endl;
// The uploader reopens this path. Close first so buffered data is
// visible to stat/libcurl; otherwise small files can upload as 0 bytes.
out_file.close();
if (!out_file) {
std::cerr << "Failed to write/close file: " << file_path << std::endl;
on_device_response_minimal(static_cast<int>(ResponseCode::INTERNAL_ERROR),
id, 0, static_cast<int>(DeviceState::READING_FILEDATA));
ClientManager::instance().change_device_state(id, DeviceState::IDLE);
return;
}
std::cout << "File saved: " << file_path
<< ", size=" << file_data.size() << " bytes" << std::endl;
//调试用
// 若是 .cfg先查看并打印内容限长
@@ -1199,8 +1229,7 @@ void process_received_message(string mac, string id,const char* data, size_t len
//使用接口上送文件lnk20250826
std::string filename;
if(SendFileWebAuto(id, file_path, file_path, filename)){//如果是补招文件的下载则不会在这边上送
std::cout << "File upload success: " << filename << std::endl;
//通知文件上传
on_device_response_minimal(static_cast<int>(ResponseCode::OK), id, 0, static_cast<int>(DeviceState::READING_FILEDATA));
}
@@ -2449,12 +2478,15 @@ void process_received_message(string mac, string id,const char* data, size_t len
<< ", 时间戳: " << record.triggerTimeMs << "ms" << std::endl;
//记录补招上来的暂态事件
append_qvvr_event(id,event.head.name,
record.nType,record.fPersisstime,record.fMagntitude,record.triggerTimeMs,record.phase);
if(record.nType != 0){
append_qvvr_event(id,event.head.name,
record.nType,record.fPersisstime,record.fMagntitude,record.triggerTimeMs,record.phase);
//直接发走暂态事件
transfer_json_qvvr_data(id,event.head.name,
record.fMagntitude,record.fPersisstime,record.triggerTimeMs,record.nType,record.phase,"");
/*transfer_json_qvvr_data(id,event.head.name,
record.fMagntitude,record.fPersisstime,record.triggerTimeMs,record.nType,record.phase,"");*/
enqueue_qvvr_json_update(id, event.head.name, record.fMagntitude, record.fPersisstime, record.triggerTimeMs, record.nType, record.phase, "");
}
//通知状态机补招暂态事件成功
on_device_response_minimal(static_cast<int>(ResponseCode::OK), id, 0, static_cast<int>(DeviceState::READING_EVENTLOG));