gsd_dsm/plugins/gsd_MonitorPlugin.cc

/**
 *
 *  gsd_MonitorPlugin.cc
 *
 */

#include "gsd_MonitorPlugin.h"
#include "HttpClient.h"

using namespace drogon;
using namespace gsd;

void MonitorPlugin::initAndStart(const Json::Value &config)
{
    #ifdef KAFKA
        // Initialize and start the plugin
        this->monitor = std::make_shared<Monitor>();
        if(this->Init() != OK){
            ErrorL << "MonitorPlugin初始化失败" << endl;
        }
    #endif
}

void MonitorPlugin::shutdown() 
{
    this->stop_ = true;
}

/**
* @description: 初始化
* @return {*}
*/    
int8_t MonitorPlugin::Init(){
    if(!monitor->Init()) {
        ErrorL << "monitor初始化失败" << endl;
        return ERR;
    }
    this->pool = std::make_shared<ThreadPool>(1,ThreadPool::PRIORITY_HIGHEST, false);
    return OK;
}

/**
 * @description: 启动任务
 * @return {*}
 */
void MonitorPlugin::StartTask(){
    // 寒武纪加速卡版本
    #ifdef KAFKA
        this->pool->async([&](){
            while(!this->stop_){
                 this->MonitorProThrd();
            }
        });
        this->pool->start();
    #endif
}   

/**
 * @description: 处理任务
 * @return {*}
 */
void MonitorPlugin::MonitorProThrd(){
    FrameInferData::Ptr result = std::make_shared<FrameInferData>();
    if(!this->ConsumeData(result)) return;
    if(result->Num == 0) return;
    if(!app().getPlugin<TcpPlugin>()->getAlive()){
        result->uuid = uuid::generate();
        std::string sql = result->ObjectToSql();
        SqlWriter insert(sql.c_str());
        vector<vector<std::string>> ret;
        insert << ret;
        InfoL << "Save inferential Data, uuid:" << result->uuid << endl;
        return;
    }
    config::Ptr m_config = config::getPtr();
    // 发送鸟情数据至后端  
    SendBird sendBird;
    sendBird.birdImageBase64 = "";
    if(!config::getPtr()->debug){
        sendBird.birdImageBase64 = result->ImageBase64;
    }
    sendBird.birdNum = result->Num;
    sendBird.birdType = (NettyClientCommandEnum().bird_info);

    if(!config::getPtr()->LowVersion) sendBird.deviceUuid = m_config->SimCode;
    else sendBird.deviceUuid = m_config->vpnIP;

    sendBird.degree = 2;
    sendBird.remark = result->Remark;
    sendBird.heightRange = "0";
    sendBird.warningTime = result->WarningTime;
    sendBird.attr4 = result->VideoPath;
    sendBird.attr5 = "";

    HistoryVideo historyVideo;
    historyVideo.fileName = result->VideoPath;
    historyVideo.UpdateTime = result->WarningTime;
    this->SaveVideoRecord(historyVideo);

    NettyClientResultMsg <SendBird> nettyClientResultMsg;
    std::string RequestId = uuid::generate();
    result->uuid = RequestId;
    nettyClientResultMsg.setRequestId(RequestId);
    nettyClientResultMsg.setData(sendBird);
    nettyClientResultMsg.setDataType((NettyClientCommandEnum().bird_info));

    std::string json;
    nettyClientResultMsg.objectToJson(json);
    json += "\r\n";
    std::string sql = result->ObjectToSql();

    if(app().getPlugin<TcpPlugin>()->getAlive()){
        app().getPlugin<TcpPlugin>()->sendRequest(RequestId, NettyClientCommandEnum().bird_info, json, [&, sql](int status,std::string pbuf){
            if(!config::getPtr()->LowVersion){
                if(status == 200) return;
                SqlWriter insert(sql.c_str());
                vector<vector<std::string>> ret;
                insert << ret;
            }
        });
    }
}

/**
 * @description: 构建请求
* @param {CNStreamInferData::Ptr} data
 * @return {*}
 */
bool MonitorPlugin::sendRequest(CNStreamInferData::Ptr data){
    FrameInferData::Ptr result = std::make_shared<FrameInferData>();
    InfineFilter::Ptr infineFilter = InfineFilter::getPtr();
    config::Ptr m_config = config::getPtr();

    if(!config::getPtr()->debug){
        infineFilter->bboxSize = m_config->bboxSize;
        // 设置
        infineFilter->setSensitivity(m_config->sensitivity);
        infineFilter->setInterval(m_config->interval);
        infineFilter->setTimeOut(m_config->TimeOut);
        if(infineFilter != nullptr) {
            if(infineFilter->judgementResult(data) != OK){
                return false;
            }
        }
    }

    rapidjson::StringBuffer buffer;
    rapidjson::Writer<rapidjson::StringBuffer> writer(buffer);
    writer.StartArray();
    for(auto &obj : data->Objects)
    {
        writer.StartObject();
        writer.Key("x");
        writer.Double(obj.BBox.x * 1920);
        writer.Key("y");
        writer.Double(obj.BBox.y * 1080);
        writer.Key("width");
        writer.Double(obj.BBox.w * 1920);
        writer.Key("height");
        writer.Double(obj.BBox.h * 1080);
        writer.EndObject();
    }
    writer.EndArray();
    // 获取时间
    char ctime[80];
    time_t rawtime;
    struct tm *info;
    time(&rawtime);
    info = localtime(&rawtime);
    strftime(ctime, 80, "%Y-%m-%d %H:%M:%S", info);
    result->ClassId = 1;
    result->ImageBase64 = data->ImageBase64;
    result->Remark = buffer.GetString();
    result->Num = data->Objects.size();
    result->WarningTime = ctime;
    result->VideoPath = data->videoPath;
    if(result->Num == 0) return false;
    
    if(!app().getPlugin<TcpPlugin>()->getAlive()){
        result->uuid = uuid::generate();
        std::string sql = result->ObjectToSql();
        SqlWriter insert(sql.c_str());
        vector<vector<std::string>> ret;
        insert << ret;
        return true;
    }
    // 发送鸟情数据至后端  
    SendBird sendBird;
    sendBird.birdImageBase64 = "";
    if(!config::getPtr()->debug){
        sendBird.birdImageBase64 = result->ImageBase64;
    }
    sendBird.birdNum = result->Num;
    sendBird.birdType = (NettyClientCommandEnum().bird_info);
    sendBird.deviceUuid = "";

    if(!config::getPtr()->LowVersion) sendBird.deviceUuid = config::getPtr()->getSimCode();
    else sendBird.deviceUuid = config::getPtr()->vpnIP;
    
    sendBird.degree = 2;
    sendBird.remark = result->Remark;
    sendBird.heightRange = "0";
    sendBird.warningTime = result->WarningTime;
    sendBird.attr4 = result->VideoPath;
    sendBird.attr5 = "";
    HistoryVideo historyVideo;
    historyVideo.fileName = result->VideoPath;
    historyVideo.UpdateTime = result->WarningTime;
    this->SaveVideoRecord(historyVideo);

    NettyClientResultMsg <SendBird> nettyClientResultMsg;
    std::string RequestId = uuid::generate();
    result->uuid = RequestId;
    nettyClientResultMsg.setRequestId(RequestId);
    nettyClientResultMsg.setData(sendBird);
    nettyClientResultMsg.setDataType((NettyClientCommandEnum().bird_info));
    std::string json;
    nettyClientResultMsg.objectToJson(json);
    json += "\r\n";
    std::string sql = result->ObjectToSql();
    if(app().getPlugin<TcpPlugin>()->getAlive()){
        app().getPlugin<TcpPlugin>()->sendRequest(RequestId, NettyClientCommandEnum().bird_info, json, [&, sql](int status,std::string pbuf){
            if(!config::getPtr()->debug){
                if(status == 200) return;
                SqlWriter insert(sql.c_str());
                vector<vector<std::string>> ret;
                insert << ret;
            }
        });
    }else ErrorL;
    return true;
}

/**
 * @description: 
 * @param {Ptr&} result
 * @return {*}
 */
bool MonitorPlugin::ConsumeData(FrameInferData::Ptr& result){
    return monitor->ConsumeData(result);
}

/**
 * @description: 保存视频记录
 * @param {HistoryVideo&} historyVideo
 * @return {*}
 */
bool MonitorPlugin::SaveVideoRecord(HistoryVideo& historyVideo){
    vector<vector<std::string>> result;
    SqlWriter sqlSelect("SELECT Id FROM gsdDB.`HistoryVideo` where fileName = '?' and ISDELETE = 0");
    sqlSelect << historyVideo.fileName << result;
    if(result.empty()){
        char ctime[80];
        HttpClient::getDataTime(ctime);
        historyVideo.UpdateTime = ctime;
        std::string sql = historyVideo.ObjectToSql();
        SqlWriter write(sql.c_str());
        write << result;
    }
    return true;
}