// 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 #include #include #include #include #include #include #include #include #include #include #include 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* pts) { std::lock_guard 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 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 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& 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 work = pts; std::string detail; std::vector 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& 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& 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 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 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 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 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 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 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 msg = json_trim(body); // 完整请求体,用于分发 std::string brief = msg; // 仅用于日志,避免污染真实报文 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(msg, 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(); }