You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

681 lines
26 KiB
C++

This file contains ambiguous Unicode characters!

This file contains ambiguous Unicode characters that may be confused with others in your current locale. If your use case is intentional and legitimate, you can safely ignore this warning. Use the Escape button to highlight these characters.

// tcp_server.cpp - TCP 服务端(新版统一扫描协议)
//
// 消息以 '>' 结尾(不是换行)。每条命令是一个 JSON:
// {"function":"count|check|config",
// "scan":"start|stop",
// "scanTime":N, // >0: 到时自动停止,无需再发 stop(单位:秒)
// "box":{"l":..,"w":..,"h":..,"maxPerLayer":..},
// "task_id":".."}
// 语义:
// - function=count/check + scan=start:启动统一后台扫描;若扫描已在运行,则
// 强制重启(旧扫描结果丢弃);scanTime>0 时到点自动停止并返回结果。
// - function=count/check + scan=stop:立即停止当前扫描并返回结果。
// - function=config:重新加载配置文件。
// count/check 共用同一个 ScanSession,同一时刻只有一个采集线程访问雷达。
#include "tcp_server.h"
#include "config.h"
#include "json_utils.h"
#include "lidar_utils.h"
#include "count.h"
#include "detect.h"
#include "io_utils.h"
#include <winsock2.h>
#include <ws2tcpip.h>
#include <chrono>
#include <cctype>
#include <condition_variable>
#include <cstdlib>
#include <cstring>
#include <iostream>
#include <iomanip>
#include <mutex>
#include <sstream>
#include <thread>
namespace
{
// ==================== 会话状态(全局单雷达) ====================
struct SessionState
{
bool active = false; // 是否有扫描在运行
bool auto_stop = false; // 是否到时自动停止
std::string function; // "count" / "check"
CountParams box; // count 的 box 参数
std::string task_id;
SOCKET owner = INVALID_SOCKET; // 发起者(自动停止结果回发)
std::chrono::steady_clock::time_point deadline; // 自动停止时刻
};
static ScanSession g_scan; // 统一采集会话
static pavo2::pavo2_driver* g_driver = nullptr;
static SessionState g_state;
static std::mutex g_life; // 会话生命周期锁
static std::mutex g_cv_mtx; // 自动停止等待锁
static std::condition_variable g_cv;
static bool g_armed = false; // 是否有到点任务
static std::chrono::steady_clock::time_point g_deadline;
// ==================== 发送 ====================
// 响应统一以 '>' 结尾
static void send_text(SOCKET s, const std::string& msg)
{
std::string out = msg;
if (!out.empty() && out.back() != '>') out += '>';
send(s, out.c_str(), (int)out.size(), 0);
}
// 响应出口:resp==nullptr 直接发到 client(TCP 模式);否则收集到 resp(HTTP 模式)。
static void send_reply(SOCKET client, std::string* resp, const std::string& msg)
{
if (resp) *resp = msg;
else send_text(client, msg);
}
// ==================== 停止并取回结果 ====================
// need_auto=true 表示"自动到时"触发,只允许在 auto_stop 仍武装时抢到;
// 否则(手动 stop)只要 active 即可。
static bool claim_stop_and_fetch(bool need_auto, SessionState* snap,
std::vector<CountPoint>* pts)
{
std::lock_guard<std::mutex> lk(g_life);
if (!g_state.active) return false;
if (need_auto && !g_state.auto_stop) return false; // 已被手动 stop 抢先
*snap = g_state;
g_state.active = false;
g_state.auto_stop = false;
{
std::lock_guard<std::mutex> ck(g_cv_mtx);
g_armed = false;
}
g_cv.notify_all();
g_scan.stop(); // 等待采集线程退出(≤100ms轮询粒度)
*pts = g_scan.snapshot();
g_scan.reset(); // 清空点集,为下一次 start 准备
return true;
}
// 仅停止当前扫描,不返回结果(用于 start 时的强制重启)。
// 调用方需已持有 g_life。
static bool discard_current_locked()
{
if (!g_state.active) return false;
g_state.active = false;
g_state.auto_stop = false;
{
std::lock_guard<std::mutex> ck(g_cv_mtx);
g_armed = false;
}
g_cv.notify_all();
g_scan.stop();
g_scan.reset();
return true;
}
// ==================== 结果计算与回包 ====================
static void finish_count(const SessionState& snap,
const std::vector<CountPoint>& pts,
SOCKET target, std::string* resp = nullptr)
{
bool valid_box = snap.box.l > 0 && snap.box.w > 0 &&
snap.box.h > 0 && snap.box.maxPerLayer > 0;
std::vector<CountPoint> work = pts;
std::string detail;
std::vector<uint8_t> top_img;
int total = 0, tw = 0, th = 0, tlv = 0;
double tarea = 0.0;
if (!pts.empty() && valid_box)
{
sparse_density_filter(work); // 点云级稀疏拖影过滤(默认开, 配置可调)
z_gap_interpolate(work); // z空隙插值:只在停止后执行一次
total = calc_item_count(work, snap.box, &detail,
&top_img, &tw, &th, &tlv, &tarea);
}
std::cout << "[TCP] count结束: task_id=" << snap.task_id
<< ", 点数=" << pts.size() << ", 结果=" << total << std::endl;
std::ostringstream r;
r << "{\"code\":200,\"message\":\"count done\",\"count\":" << total;
if (!snap.task_id.empty())
r << ",\"task_id\":\"" << snap.task_id << "\"";
r << "}";
send_reply(target, resp, r.str());
// 后台保存 PCD/图片/日志(不阻塞响应)
if (!pts.empty() && valid_box)
save_count_files_async(work, top_img, tw, th, snap.box, detail,
total, tlv, tarea);
}
static void finish_check(const SessionState& snap,
const std::vector<CountPoint>& pts,
SOCKET target, std::string* resp = nullptr)
{
// 输出目录 data/日期/时间_taskId(由 io_utils 统一创建)
std::string out_dir = make_task_dir(snap.task_id);
DetectResult res;
detect_analyze_slices(pts, out_dir, &res);
std::ostringstream r;
r << "{\"code\":200,\"message\":\"detect done\"";
if (res.valid)
{
if (res.start_fill_pct >= 0.0)
r << ",\"start\":{\"fill_pct\":"
<< std::fixed << std::setprecision(2) << res.start_fill_pct
<< ",\"y_max\":" << std::setprecision(1) << res.start_y_max
<< ",\"y_median\":" << res.start_y_med << "}";
if (res.end_fill_pct >= 0.0)
r << ",\"end\":{\"fill_pct\":"
<< std::fixed << std::setprecision(2) << res.end_fill_pct
<< ",\"y_max\":" << std::setprecision(1) << res.end_y_max
<< ",\"y_median\":" << res.end_y_med << "}";
}
if (!snap.task_id.empty())
r << ",\"task_id\":\"" << snap.task_id << "\"";
r << "}";
send_reply(target, resp, r.str());
std::cout << "[TCP] check结束: task_id=" << snap.task_id
<< ", 点数=" << pts.size()
<< (res.valid ? ", 检测有效" : ", 检测无效") << std::endl;
}
static void finish_and_reply(const SessionState& snap,
const std::vector<CountPoint>& pts,
SOCKET target, std::string* resp = nullptr)
{
if (snap.function == "count")
finish_count(snap, pts, target, resp);
else if (snap.function == "check")
finish_check(snap, pts, target, resp);
}
// ==================== 自动到时停止线程 ====================
static void auto_stop_loop()
{
for (;;)
{
std::unique_lock<std::mutex> lk(g_cv_mtx);
if (!g_armed)
{
g_cv.wait(lk);
continue;
}
auto deadline = g_deadline;
if (g_cv.wait_until(lk, deadline) != std::cv_status::timeout)
continue; // 被通知:可能已取消/重设,回到顶部
g_armed = false; // 到点
lk.unlock();
SessionState snap;
std::vector<CountPoint> pts;
if (claim_stop_and_fetch(true, &snap, &pts))
{
std::cout << "[TCP] 扫描到时自动停止: function=" << snap.function
<< ", task_id=" << snap.task_id << std::endl;
if (snap.owner != INVALID_SOCKET)
finish_and_reply(snap, pts, snap.owner);
else
{
// HTTP 模式无长连接可推送,结果只打印并留痕
std::string out;
finish_and_reply(snap, pts, INVALID_SOCKET, &out);
std::cout << "[HTTP] scanTime到时自动停止, 无连接可推送, 结果: "
<< out << std::endl;
}
}
}
}
// ==================== 命令处理 ====================
static void handle_scan_start(const std::string& msg, SOCKET client,
std::string* resp = nullptr)
{
std::string fn;
json_get_string(msg, "function", fn);
std::string task_id;
json_get_string(msg, "task_id", task_id);
double scan_time = 0.0;
json_get_number(msg, "scanTime", scan_time);
CountParams box;
json_get_string(msg, "task_id", box.task_id);
json_get_number(msg, "l", box.l);
json_get_number(msg, "w", box.w);
json_get_number(msg, "h", box.h);
double d = 0.0;
if (json_get_number(msg, "maxPerLayer", d) ||
json_get_number(msg, "max", d))
box.maxPerLayer = (int)d;
std::ostringstream ack;
ack << "{\"code\":200,\"message\":\"scan start\",\"function\":\"" << fn << "\"";
if (!task_id.empty())
ack << ",\"task_id\":\"" << task_id << "\"";
if (scan_time > 0)
ack << ",\"scanTime\":" << (int)scan_time;
ack << "}";
bool started = false;
std::string fail;
{
std::lock_guard<std::mutex> lk(g_life);
// 扫描运行中收到新的 start:强制重启(旧扫描结果丢弃)
if (discard_current_locked())
std::cout << "[TCP] 扫描运行中被新命令强制重启: function=" << fn
<< ", task_id=" << task_id << std::endl;
g_scan.reset();
if (!g_scan.start(*g_driver))
{
fail = "scan start failed";
}
else
{
SessionState ns;
ns.active = true;
ns.function = fn;
ns.box = box;
ns.task_id = task_id;
// TCP 用连接作 owner(自动到时结果回发);HTTP 无长连接,置无效不推送
ns.owner = resp ? INVALID_SOCKET : client;
if (scan_time > 0)
{
ns.auto_stop = true;
ns.deadline = std::chrono::steady_clock::now() +
std::chrono::milliseconds((long)(scan_time * 1000.0));
}
g_state = ns;
if (ns.auto_stop)
{
std::lock_guard<std::mutex> ck(g_cv_mtx);
g_armed = true;
g_deadline = ns.deadline;
}
g_cv.notify_all();
started = true;
}
}
if (!started)
{
std::cerr << "[TCP] " << fail << std::endl;
send_reply(client, resp, "{\"code\":400,\"message\":\"" + fail + "\"}");
return;
}
std::cout << "[TCP] 扫描开始: function=" << fn
<< ", task_id=" << task_id
<< (scan_time > 0 ? ", 到时自动停止" : ", 等待stop") << std::endl;
send_reply(client, resp, ack.str());
}
static void handle_scan_stop(const std::string& msg, SOCKET client,
std::string* resp = nullptr)
{
std::string fn;
json_get_string(msg, "function", fn);
SessionState snap;
std::vector<CountPoint> pts;
if (!claim_stop_and_fetch(false, &snap, &pts))
{
// 未在扫描(从未 start / 已停止 / 自动到时已完成)
std::ostringstream r;
r << "{\"code\":200,\"message\":\"scan not running\"";
if (fn == "count")
r << ",\"count\":0";
r << "}";
send_reply(client, resp, r.str());
return;
}
std::cout << "[TCP] 收到stop: function=" << snap.function
<< ", task_id=" << snap.task_id << std::endl;
finish_and_reply(snap, pts, client, resp);
}
static void handle_config(const std::string& msg, SOCKET client,
std::string* resp = nullptr)
{
(void)msg;
std::cout << "[TCP] 收到config命令,重新读取 config/config ..." << std::endl;
load_config();
reload_speed_profile(); // cfg.scan.z_speed_type=configuration 时重新读取 speed.txt
std::cout << "[TCP] 配置已重载: 采集=" << cfg.scan.collect_duration_s
<< "s, 速度模式=" << cfg.scan.z_speed_type
<< ", 速度=" << cfg.scan.z_speed << "cm/s, 插值=" << cfg.count.z_interpolate_mode
<< ", 保留=" << cfg.log.retain_days << "天, 端口=" << cfg.server.port << std::endl;
std::cout << "[TCP] count容差: 基准=" << cfg.count.base_tolerance
<< "mm + 每层" << cfg.count.per_level_tolerance << "mm"
<< ", 地板高=" << cfg.count.floor_height << "mm"
<< ", SparseCut=" << (cfg.filter.sparse_cut.enable ? "开" : "关")
<< "(X±" << cfg.filter.sparse_cut.x_radius << "mm,同圈>=" << cfg.filter.sparse_cut.circle_min
<< ",跨圈>=" << cfg.filter.sparse_cut.run_min << ")"
<< ", TailCut=" << (cfg.filter.tail_cut.enable ? "开" : "关") << std::endl;
send_reply(client, resp, "{\"code\":200,\"message\":\"config reloaded\"}");
}
// ConfigureSpeed: 标定生成 config/speed.txt(雷达横放时,每帧Y最大值=行进距离)。
// 请求: {"function":"ConfigureSpeed","scanTime":N},N缺省用 cfg.scan.collect_duration_s。
// 标定期间与扫描会话互斥(全程持 g_life,其他命令排队)。
static void handle_configure_speed(const std::string& msg, SOCKET client,
std::string* resp = nullptr)
{
double scan_time = 0.0;
json_get_number(msg, "scanTime", scan_time);
double seconds = (scan_time > 0) ? scan_time : (double)cfg.scan.collect_duration_s;
if (seconds <= 0.0) seconds = 5.0;
std::cout << "[TCP] 收到ConfigureSpeed: 采集" << seconds
<< "s 标定变速剖面(每帧Y最大值) ..." << std::endl;
int rows = -1;
std::string err;
{
std::lock_guard<std::mutex> lk(g_life);
if (discard_current_locked()) // 扫描运行中先强制停止(本函数持锁)
std::cout << "[TCP] 扫描运行中被ConfigureSpeed中断" << std::endl;
rows = collect_speed_profile(*g_driver, seconds, err);
}
if (rows < 0)
{
std::cerr << "[TCP] ConfigureSpeed失败: " << err << std::endl;
send_reply(client, resp, "{\"code\":400,\"message\":\"configure speed failed\"}");
return;
}
reload_speed_profile(); // 重新载入新生成的剖面(若已是configuration则立即生效)
std::cout << "[TCP] speed.txt 已更新: " << rows
<< " 帧样本 (采集" << seconds << "s)" << std::endl;
std::ostringstream r;
r << "{\"code\":200,\"message\":\"configure speed done\",\"rows\":" << rows
<< ",\"duration\":" << (int)seconds
<< ",\"file\":\"config/speed.txt\"}";
send_reply(client, resp, r.str());
}
// 处理一条命令。resp==nullptr: 直接发送(TCP);resp!=nullptr: 响应写入该串(HTTP)。
// 返回值: 是否产生响应。
static bool dispatch(const std::string& raw, SOCKET client,
std::string* resp = nullptr)
{
std::string msg = json_trim(raw);
if (msg.empty()) return false;
std::string fn;
if (!json_get_string(msg, "function", fn)) return false; // 未知消息忽略
if (fn == "config")
{
handle_config(msg, client, resp);
return true;
}
if (fn == "ConfigureSpeed" || fn == "configurespeed")
{
handle_configure_speed(msg, client, resp);
return true;
}
if (fn != "count" && fn != "check") return false; // 未知功能忽略
std::string scan_cmd;
json_get_string(msg, "scan", scan_cmd);
if (scan_cmd == "stop")
{
handle_scan_stop(msg, client, resp);
return true;
}
else if (scan_cmd == "start")
{
handle_scan_start(msg, client, resp);
return true;
}
return false; // 无scan/未知scan值忽略
}
static void handle_client(SOCKET client)
{
char buf[4096];
std::string rbuf;
while (true)
{
int n = recv(client, buf, sizeof(buf) - 1, 0);
if (n <= 0) break; // 连接关闭或出错
buf[n] = '\0';
rbuf.append(buf, n);
// 按 '>' 分帧处理(一条消息以 '>' 结尾)
size_t pos;
while ((pos = rbuf.find('>')) != std::string::npos)
{
std::string raw = rbuf.substr(0, pos);
rbuf.erase(0, pos + 1);
dispatch(raw, client);
}
}
// 客户端断开:若它是某自动扫描的 owner,扫描仍会到时自动停止;
// 未设置 scanTime 的扫描则持续到下一次 start 强制重启或 stop。
closesocket(client);
}
} // namespace
void run_tcp_server(pavo2::pavo2_driver& driver)
{
g_driver = &driver;
std::cout << "\n========== TCP服务端模式 ==========" << std::endl;
std::cout << "[TCP] 监听端口: " << cfg.server.port
<< " | 消息以 '>' 结尾" << std::endl;
SOCKET listen_sock = socket(AF_INET, SOCK_STREAM, 0);
if (listen_sock == INVALID_SOCKET)
{
std::cerr << "[错误] 创建监听socket失败!" << std::endl;
return;
}
sockaddr_in addr;
addr.sin_family = AF_INET;
addr.sin_addr.s_addr = INADDR_ANY;
addr.sin_port = htons(cfg.server.port);
if (bind(listen_sock, (sockaddr*)&addr, sizeof(addr)) == SOCKET_ERROR)
{
std::cerr << "[错误] 绑定端口 " << cfg.server.port << " 失败!" << std::endl;
closesocket(listen_sock);
return;
}
if (listen(listen_sock, 5) == SOCKET_ERROR)
{
std::cerr << "[错误] 监听失败!" << std::endl;
closesocket(listen_sock);
return;
}
// 自动到时停止线程(scanTime 模式)
std::thread(auto_stop_loop).detach();
std::cout << "[TCP] 初始化完成,等待命令 "
"{\"function\":\"count\",\"scan\":\"start\",...} ..." << std::endl;
while (true)
{
SOCKET client = accept(listen_sock, nullptr, nullptr);
if (client == INVALID_SOCKET)
{
std::cout << "[TCP] accept失败,重试..." << std::endl;
continue;
}
std::thread(handle_client, client).detach();
}
}
// ==================== HTTP 服务端模式 ====================
// 请求: POST /<任意路径>, body = 与 TCP 协议相同的 JSON(不带结尾 '>')
// 响应: HTTP 200 + application/json, body = 与 TCP 协议相同的响应文本(不带 '>')
// 注: HTTP 请求是"无连接"的。scanTime>0 的扫描到点自动停止后无法主动推送
// 结果(结果打印到日志);客户端应在需要结果时再发一次 stop(会返回
// "scan not running" + count:0) 或通过文件目录查看。建议流程:
// start(不带scanTime) → 采集完成 → stop 拿结果。
namespace
{
static bool http_read_head(SOCKET c, std::string& head, std::string& rest)
{
char buf[8192];
head.clear();
rest.clear();
while (head.size() < 65536) // 头上限 64KB,防呆
{
int n = recv(c, buf, (int)sizeof(buf), 0);
if (n <= 0) return false;
head.append(buf, n);
size_t p = head.find("\r\n\r\n");
if (p != std::string::npos)
{
rest = head.substr(p + 4); // 头部之后可能已带部分 body
head = head.substr(0, p);
return true;
}
}
return false;
}
static long http_find_content_length(const std::string& head)
{
std::istringstream iss(head);
std::string line;
while (std::getline(iss, line))
{
if (!line.empty() && line.back() == '\r') line.pop_back();
size_t c = line.find(':');
if (c == std::string::npos) continue;
std::string key = line.substr(0, c);
for (auto& ch : key) ch = (char)tolower((unsigned char)ch);
if (key == "content-length")
return strtol(line.c_str() + c + 1, nullptr, 10);
}
return -1;
}
static void http_send_response(SOCKET c, int code, const char* reason,
const std::string& body)
{
std::ostringstream h;
h << "HTTP/1.1 " << code << " " << reason << "\r\n"
<< "Content-Type: application/json; charset=utf-8\r\n"
<< "Content-Length: " << body.size() << "\r\n"
<< "Connection: close\r\n\r\n";
std::string out = h.str() + body;
int off = 0;
while (off < (int)out.size())
{
int n = send(c, out.c_str() + off, (int)out.size() - off, 0);
if (n <= 0) break;
off += n;
}
}
// 每个连接处理一个请求后即关闭(Connection: close)
static void handle_http_client(SOCKET c)
{
std::string head, rest;
if (!http_read_head(c, head, rest))
{
closesocket(c);
return;
}
std::string line0 = head.substr(0, head.find("\r\n"));
std::istringstream iss(line0);
std::string method, path;
iss >> method >> path;
if (method != "POST")
{
http_send_response(c, 405, "Method Not Allowed",
"{\"code\":405,\"message\":\"use POST, body same as TCP protocol json\"}");
closesocket(c);
return;
}
long cl = http_find_content_length(head);
if (cl < 0 || cl > 4 * 1024 * 1024) // body 上限 4MB
{
http_send_response(c, 411, "Length Required",
"{\"code\":411,\"message\":\"missing/invalid Content-Length\"}");
closesocket(c);
return;
}
std::string body = rest;
while ((long)body.size() < cl)
{
char buf[8192];
int n = recv(c, buf, (int)sizeof(buf), 0);
if (n <= 0) break;
body.append(buf, n);
}
if ((long)body.size() > cl) body.resize(cl);
std::string brief = json_trim(body);
if (brief.size() > 120) brief = brief.substr(0, 120) + "...";
std::cout << "[HTTP] POST " << path << " (" << cl << "B): "
<< brief << std::endl;
std::string out;
bool handled = dispatch(brief, INVALID_SOCKET, &out);
if (handled)
http_send_response(c, 200, "OK", out);
else
http_send_response(c, 400, "Bad Request",
"{\"code\":400,\"message\":\"unknown command, see TCP protocol\"}");
closesocket(c);
}
static void run_http_server_internal()
{
SOCKET listen_sock = socket(AF_INET, SOCK_STREAM, 0);
if (listen_sock == INVALID_SOCKET)
{
std::cerr << "[错误] 创建HTTP监听socket失败!" << std::endl;
return;
}
sockaddr_in addr;
addr.sin_family = AF_INET;
addr.sin_addr.s_addr = INADDR_ANY;
addr.sin_port = htons(cfg.server.port);
if (bind(listen_sock, (sockaddr*)&addr, sizeof(addr)) == SOCKET_ERROR)
{
std::cerr << "[错误] 绑定HTTP端口 " << cfg.server.port << " 失败!" << std::endl;
closesocket(listen_sock);
return;
}
if (listen(listen_sock, 5) == SOCKET_ERROR)
{
std::cerr << "[错误] HTTP监听失败!" << std::endl;
closesocket(listen_sock);
return;
}
// 自动到时停止线程(scanTime 模式; HTTP 下结果打印日志不推送)
std::thread(auto_stop_loop).detach();
std::cout << "[HTTP] 初始化完成,等待 POST 命令 "
"{\"function\":\"count\",\"scan\":\"start\",...} ..." << std::endl;
while (true)
{
SOCKET client = accept(listen_sock, nullptr, nullptr);
if (client == INVALID_SOCKET)
{
std::cout << "[HTTP] accept失败,重试..." << std::endl;
continue;
}
std::thread(handle_http_client, client).detach();
}
}
}
void run_http_server(pavo2::pavo2_driver& driver)
{
g_driver = &driver;
std::cout << "\n========== HTTP服务端模式 ==========" << std::endl;
std::cout << "[HTTP] 监听端口: " << cfg.server.port
<< " | POST json body(与TCP协议相同, 不带结尾'>')" << std::endl;
run_http_server_internal();
}