leida_simi/tcp_server.cpp

// 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 <condition_variable>
#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);
    }

    // ==================== 停止并取回结果 ====================
    // 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)
    {
        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)
        {
            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_text(target, 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)
    {
        // 输出目录 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_text(target, 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)
    {
        if (snap.function == "count")
            finish_count(snap, pts, target);
        else if (snap.function == "check")
            finish_check(snap, pts, target);
    }

    // ==================== 自动到时停止线程 ====================
    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;
                finish_and_reply(snap, pts, snap.owner);
            }
        }
    }

    // ==================== 命令处理 ====================
    static void handle_scan_start(const std::string& msg, SOCKET client)
    {
        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;
                ns.owner    = 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_text(client, "{\"code\":400,\"message\":\"" + fail + "\"}");
            return;
        }
        std::cout << "[TCP] 扫描开始: function=" << fn
                  << ", task_id=" << task_id
                  << (scan_time > 0 ? ", 到时自动停止" : ", 等待stop") << std::endl;
        send_text(client, ack.str());
    }

    static void handle_scan_stop(const std::string& msg, SOCKET client)
    {
        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_text(client, r.str());
            return;
        }
        std::cout << "[TCP] 收到stop: function=" << snap.function
                  << ", task_id=" << snap.task_id << std::endl;
        finish_and_reply(snap, pts, client);
    }

    static void handle_config(const std::string& msg, SOCKET client)
    {
        (void)msg;
        std::cout << "[TCP] 收到config命令，重新读取 config/config ..." << std::endl;
        load_config();
        std::cout << "[TCP] 配置已重载: 采集=" << Collect_Duration_S
                  << "s, 速度=" << Speed_Z << "cm/s, 插值=" << Z_Interpolate_Mode
                  << ", 保留=" << Log_Retain_Days << "天, 端口=" << Tcp_Port << std::endl;
        send_text(client, "{\"code\":200,\"message\":\"config reloaded\"}");
    }

    static void dispatch(const std::string& raw, SOCKET client)
    {
        std::string msg = json_trim(raw);
        if (msg.empty()) return;

        std::string fn;
        if (!json_get_string(msg, "function", fn)) return;   // 未知消息忽略

        if (fn == "config")
        {
            handle_config(msg, client);
            return;
        }
        if (fn != "count" && fn != "check") return;          // 未知功能忽略

        std::string scan_cmd;
        json_get_string(msg, "scan", scan_cmd);
        if (scan_cmd == "stop")
            handle_scan_stop(msg, client);
        else if (scan_cmd == "start")
            handle_scan_start(msg, client);
        // 其余（无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] 监听端口: " << Tcp_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(Tcp_Port);
    if (bind(listen_sock, (sockaddr*)&addr, sizeof(addr)) == SOCKET_ERROR)
    {
        std::cerr << "[错误] 绑定端口 " << Tcp_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();
    }
}