Hu Chunming / vpt_ascend_arm

  #include "MultiSourceProcess.h"
  #include "../common/logger.hpp"
  
  #include <string>
  #include <queue>
  #include <mutex>
  #include <chrono>
  #include <thread>
  
  #include "../decoder/interface/DecoderManager.h"
  #include "../decoder/interface/utiltools.hpp"
  #include "../helpers/time_helper.hpp"
  #include "../helpers/os_helper.hpp"
  #include "../helpers/gen_json.hpp"

  #include "../helpers/str_helper.hpp"

  #include "../reprocessing_module/save_snapshot_reprocessing.h"
  
  #include "macro_definition.h"
  #include "ErrorInfo.h"
  #include "SourceSingleton.hpp"
  #include "mvpt_process_assist.h"
  
  #include "../util/vpc_util.h"
  
  // #define VEHICLE_MULTI_BOXES
  
  #define WITH_FACE_DET_SS
  

  #define productSN "51C4B28135604F649671727185949A91" //linux 通途抓拍引擎产品序列号
  

  using namespace std;
  
  map<int, algo_type> index_to_algo_type = {{0, algorithm_type_t::HUMAN_SNAPSHOT},
                                            {1, algorithm_type_t::NONMOTOR_VEHICLE_SNAPSHOT},
                                            {2, algorithm_type_t::NONMOTOR_VEHICLE_SNAPSHOT},
                                            {3, algorithm_type_t::NONMOTOR_VEHICLE_SNAPSHOT},
                                            {4, algorithm_type_t::VEHICLE_SNAPSHOT},
                                            {5, algorithm_type_t::VEHICLE_SNAPSHOT},
                                            {6, algorithm_type_t::VEHICLE_SNAPSHOT},
                                            {7, algorithm_type_t::VEHICLE_SNAPSHOT},
                                            {8, algorithm_type_t::VEHICLE_SNAPSHOT},
                                            {9, algorithm_type_t::FACE_SNAPSHOT}};
  
  
  struct decode_cbk_userdata{
    string task_id;
    void* opaque;
    void* opaque1;
  };
  
  
  /**
   * 注意： gpuFrame 在解码器设置的显卡上，后续操作要十分注意这一点，尤其是多线程情况
   * */
  void post_decod_cbk(const void * userPtr, DeviceMemory* devFrame){
      decode_cbk_userdata* ptr = (decode_cbk_userdata*)userPtr;
      if (ptr!= nullptr)
      {
          CMultiSourceProcess* _this = (CMultiSourceProcess*)ptr->opaque;
          if(nullptr != _this){
              _this->decoded_cbk(devFrame);
          }
      }
  }
  
  void decode_finished_cbk(const void * userPtr){
      decode_cbk_userdata* ptr = (decode_cbk_userdata*)userPtr;
      if (ptr!= nullptr)
      {
          CMultiSourceProcess* _this = (CMultiSourceProcess*)ptr->opaque;
          if(nullptr != _this){
              _this->task_finished(ptr->task_id);
          }
      }
      delete ptr;
      ptr = nullptr;
  }
  
  CMultiSourceProcess::CMultiSourceProcess(){
    m_timing_snapshot_thread = nullptr;
  }
  
  CMultiSourceProcess::~CMultiSourceProcess(){
  }
  
  int CMultiSourceProcess::InitAlgorthim(tsl_aiplatform_param vptParam){

  #ifdef USE_VILLAGE

    if (!CheckTime()) {

      std::cout << "sy_licence_check failed." << std::endl;

      return AUTHOR_ERROR;
    }

  #else
    //授权相关------------------------------------
  	//授权init-----------
  	const char *token_password = "Capture_atlas_arm_2023";
  	std::string guid = std::string("Capture_atlas_arm")+std::to_string(vptParam.gpuid);
  	atlas_licence_param lic_param;
  	memset(&lic_param, 0, sizeof(atlas_licence_param));
  	sprintf(lic_param.product_id, "%s", productSN);
  	sprintf(lic_param.guid, "%s",guid.c_str());
  	sprintf(lic_param.token_pwd, "%s", token_password);
  	lic_param.channel_num = 1;
  	atlas_licence_token param_token;
  	memset(&param_token, 0, sizeof(atlas_licence_token));
  
  	int result = atlas_licence_connect(&(skt_handle), lic_param, &param_token);
  	//printf("result:%d,code:%d,msg:%s \n", result, param_token.code, param_token.msg);
  	//printf("token:%s\n", param_token.token);
  	std::string recv_token = std::string(param_token.token);
  
  	//atlas_licence_check_param check_param;
  	memset(&(check_param), 0, sizeof(atlas_licence_check_param));
  	sprintf(check_param.token_pwd, "%s", token_password);
  	sprintf(check_param.time, "%s", "2023-01-10 20:00:00");
  	sprintf(check_param.token, "%s", recv_token .c_str());
  	check_param.consume = 2;
  
  	//授权check------------
  	check_label = -1;//初始值
  	std::cout << "sy_licence_check start."  << std::endl;
  	atlas_licence_check_result check_result;
  	memset(&check_result, 0, sizeof(atlas_licence_check_result));
  	check_result.code = -1;
  
  	int res = atlas_licence_check(skt_handle, check_param, &check_result);
  	if(res!=0) {
  		std::cout << "sy_licence_check failed." << std::endl;
  		return AUTHOR_ERROR;
  	}
  	//std::cout << "code:" << check_result.code << ",msg:" << check_result.msg << std::endl;
  	if(check_result.code!=0) {
  		std::cout << "code:" << check_result.code << ",msg:" << check_result.msg << std::endl;
  		return AUTHOR_ERROR;
  	}
  	std::cout << "sy_licence_check end."  << std::endl;
  
  	check_label = 0;//授权成功
  	//授权相关------------------------------------
  #endif

    set_default_logger(LogLevel(vptParam.log_level), "multi_source_process", vptParam.log_path, vptParam.log_mem, vptParam.log_mem);
    LOG_INFO("编译时间：{} {}", __DATE__, __TIME__);

    SourceSingleton::getInstance();

    skip_frame_ = 5; 
    m_batch_size = 20;

    m_devId = vptParam.gpuid;

    string models_dir = vptParam.models_dir;

    VPTProcess_PARAM vparam;
    vparam.gpuid = m_devId;
    vparam.max_batch = m_batch_size;
    vparam.threshold = 0.4;
    vparam.model_dir = models_dir;

    aclrtSetDevice(m_devId);

    int ret = vpt_process.init(vparam);
    if (ret < 0){
      return ret;
    }

  #ifdef USE_VILLAGE 

    //道路分割
    if(seg_process.init(vptParam.gpuid, models_dir) != 0) {
      LOG_FATAL("Init road_seg failed");
      return -1;
    }
    if(seg3cls_process.init(vptParam.gpuid, models_dir) != 0) {
      LOG_FATAL("Init road_3clsseg failed");
      return -1;
    }
    
  

    //三轮车头肩检测
    if (!tricycle_manned_.init(vptParam.gpuid, models_dir)) {
      LOG_FATAL("Init tricycle_hs failed");
      return -1;
    }

    //货车头肩检测
    if (!truck_manned_.init(vptParam.gpuid, models_dir)) {
      LOG_FATAL("Init truck_hs failed");
      return -1;
    }

    //二轮车头肩检测
    if (!motor_hsprocess_.init(vptParam.gpuid, models_dir)) {
      LOG_FATAL("Init motor_hs failed");
      return -1;
    }

  
    //二轮车使用手机及加雨棚
    if (!motor_refit_phoneprocess_.init(vptParam.gpuid, models_dir)) {
      LOG_FATAL("Init motor_phone failed");
      return -1;
    }
    /*

    //二轮车使用手机检测
    if (!motor_phoneprocess_.init(vptParam.gpuid, models_dir)) {
      LOG_FATAL("Init motor_phone failed");
      return -1;

    }*/

    //电动车改装(加雨棚)
    if (!motor_refitprocess_.init(vptParam.gpuid, models_dir)) {
      LOG_FATAL("Init motor_refit failed");
      return -1;

    }

    // 行人/二轮车闯红灯
    if (!traffic_lightprocess_.init(vptParam.gpuid, models_dir)) {
      LOG_FATAL("Init traffic_light failed");
      return -1;
    }

    // 与道路分割相关的算法
    if (!roadseg_corrprocess_.init(vptParam.gpuid, models_dir)) {
      LOG_FATAL("Init roadseg_corrprocess failed");
      return -1;
    }

  #endif

    #ifdef WITH_FACE_DET_SS
        // 人脸检测初始化
        facedet_ai_engine_param fd_param;
        char model_path_yolov5s[100];
        strcpy(model_path_yolov5s, (models_dir + "/models/face_detect/face_det_yolov5s_310p.om").c_str());
        fd_param.sdk_param.det_modelNames = model_path_yolov5s;
        char model_path_ldmk[100];
        strcpy(model_path_ldmk, (models_dir + "/models/face_detect/face_ldmk_310p.om").c_str());
        fd_param.sdk_param.ldmk_modelNames = model_path_ldmk;
        char model_path_pose[100];
        strcpy(model_path_pose, (models_dir + "/models/face_detect/face_pose_310p.om").c_str());
        fd_param.sdk_param.pose_modelNames = model_path_pose;
        char model_path_score[100];
        strcpy(model_path_score, (models_dir + "/models/face_detect/face_score_310p.om").c_str());
        fd_param.sdk_param.score_modelNames = model_path_score;
        char model_path_fuzzy[100];
        strcpy(model_path_fuzzy, (models_dir + "/models/face_detect/face_fuzzy_310p.om").c_str());
        fd_param.sdk_param.fuzzy_modelNames = model_path_fuzzy;
        char model_path_occlusion[100];
        strcpy(model_path_occlusion, (models_dir + "/models/face_detect/face_occlusion_310p.om").c_str());
        fd_param.sdk_param.occlusion_modelNames = model_path_occlusion;
        fd_param.sdk_param.thresld = 0.6;
        fd_param.sdk_param.devId = m_devId;
        fd_param.sdk_param.auth_license = "sy_tongtu_aiplatform_sdk_2023";
        fd_param.sdk_param.facial_fea_point_config = SY_CONFIG_OPEN;		//是否启动关键点检测
        fd_param.sdk_param.pose_config = SY_CONFIG_OPEN;					//是否启动姿态角
        fd_param.sdk_param.quality_config = SY_CONFIG_OPEN;				//是否启动质量检测
        fd_param.sdk_param.score_config = SY_CONFIG_OPEN;				//是否启动人脸置信度  //SY_CONFIG_OPEN  SY_CONFIG_CLOSE
        fd_param.sdk_param.max_result_count = 50;
        ret = m_face_det_ai_engine.init_ai_engine(fd_param);
        if (ret < 0 ) {
          LOG_FATAL("Init face detection failed");
          return ret;
        }
  #endif

    m_task_param_manager = task_param_manager::getInstance();
    m_snapshot_reprocessing = new snapshot_reprocessing(m_devId);
    m_save_snapshot_reprocessing = new save_snapshot_reprocessing(m_devId);

  #ifdef POST_USE_RABBITMQ
      mq_manager_ = new mq::Manager();
  #endif

    VPCUtil* pVpcUtil = VPCUtil::getInstance();
    pVpcUtil->init(m_devId);

    m_pAlgorthimThread = new thread([](void* arg) {
        CMultiSourceProcess* process = (CMultiSourceProcess*)arg ;
            process->algorthim_process_thread();
        return (void*)0;

      }

      , this);

    m_recode_thread = new thread([](void* arg) {
        CMultiSourceProcess* process = (CMultiSourceProcess*)arg ;
            process->recode_thread();
        return (void*)0;
      }
      , this);

    m_timing_snapshot_thread = new std::thread(
          [](void* arg)
          {
              CMultiSourceProcess* _this=(CMultiSourceProcess*)arg;
              if(_this != nullptr){
                _this->timing_snapshot_thread();
              }else{
                LOG_ERROR("定时抓拍线程启动失败 !");
              }
              return (void*)0;
          }, this);
          
    if(nullptr == m_timing_snapshot_thread){
        LOG_ERROR("定时抓拍线程启动失败 !");

    }

  
    LOG_INFO("InitAlgorthim succeed !");
  
    return 0;

  }
  
  #ifdef POST_USE_RABBITMQ
  /* MQ队列的初始化 */
  int CMultiSourceProcess::AddMqConn(mq_type_t mq_type, rabbitmq_conn_params_t mq_conn_param) {
    /* 初始化MQ队列 */
    LOG_INFO("初始化MQ队列");
    if(mq_manager_ == nullptr){
      LOG_ERROR("mq_manager_ is null");
      return FAILED;
    }
  
    if (!mq_manager_->add_conn(mq_type, mq_conn_param)) {
      LOG_ERROR("Connection MQ failed, ip: {} port: {} uname: {} passwd: {}", mq_conn_param.ip, mq_conn_param.port,
                mq_conn_param.uname, mq_conn_param.passwd);
      return MQ_CONN_ERROR;
    }
  
    LOG_INFO("为报警类 绑定MQ回调");
    /* 为报警类 绑定回调 传入mq_manager_.publish 内部直接调用*/
    if (m_save_snapshot_reprocessing!= nullptr && mq_type_t::ALARM_MQ == mq_type) {
      m_save_snapshot_reprocessing->set_callback(
          std::bind(&mq::Manager::publish, mq_manager_, mq_type, std::placeholders::_1, true));
      LOG_INFO("为报警类 绑定MQ回调 成功！");
    }
  
    LOG_INFO("初始化MQ队列成功！");
  
    return SUCCESS;
  }
  
  /* 获取任务的状态 MQ返回 */
  int CMultiSourceProcess::GetTaskStatus(const string taskID) {
  
    DecoderManager* pDecManager = DecoderManager::getInstance();
  
    std::vector<std::string> taskids;
    std::vector<int> statues;
    if(pDecManager->isPausing(taskID)){
      taskids.emplace_back(taskID);
      statues.emplace_back(2);
    }else if(pDecManager->isRunning(taskID)){
      taskids.emplace_back(taskID);
      statues.emplace_back(1);
    }
  
    if (!taskids.empty()) {
      auto json_str = helpers::gen_json::gen_task_status_json(taskids, statues);
      mq_manager_->publish(mq_type_t::GET_TASK_MQ, json_str.c_str(),true);
    }
  
    return SUCCESS;
  }
  #endif
  
  bool CMultiSourceProcess::AddTask(task_param _cur_task_param){
      DecoderManager* pDecManager = DecoderManager::getInstance();
  
      const char* task_id = _cur_task_param.task_id;
  
      MgrDecConfig config;
      config.name = task_id;
      config.cfg.uri = _cur_task_param.ipc_url;
      config.cfg.post_decoded_cbk = post_decod_cbk;
      config.cfg.decode_finished_cbk = decode_finished_cbk;
      config.cfg.force_tcp = true;  // rtsp用tcp
      config.cfg.gpuid = to_string(m_devId);
      config.cfg.skip_frame = skip_frame_;
  
      if (1 == _cur_task_param.dec_type){
        config.cfg.port = _cur_task_param.port;
        config.dec_type = DECODER_TYPE_GB28181;
        config.cfg.uri = task_id;
        if(_cur_task_param.protocal == 0){
          // 指定用udp协议
          config.cfg.force_tcp = false;
        }
        config.cfg.request_stream_cbk = _cur_task_param.gb28181_request_stream_callback ;
      }else if (2 == _cur_task_param.dec_type){
        config.dec_type = DECODER_TYPE_DVPP;
      }else {
          config.dec_type = DECODER_TYPE_FFMPEG;
      }
      
      AbstractDecoder* dec = pDecManager->createDecoder(config);
      if (!dec){
          return false;
      }
  
      decode_cbk_userdata* userPtr = new decode_cbk_userdata;
      userPtr->task_id = string(task_id);
      userPtr->opaque = this;
      userPtr->opaque1 = dec;
      pDecManager->setPostDecArg(config.name, userPtr);
      pDecManager->setFinishedDecArg(config.name, userPtr);
      
      // pDecManager->setDecKeyframe(config.name, true); // 只对关键帧解码
      
  
      // 保存新添加任务的配置参数
      m_task_param_manager->add_task_param(task_id, _cur_task_param);
  
      int input_image_width = 0;
      int input_image_height = 0;
      pDecManager->getResolution(config.name, input_image_width, input_image_height);
      LOG_INFO("task_id: {} width: {} height:{}", task_id, input_image_width, input_image_height);
  
       // 所有参数都准备好之后再启动解码
      bool bStart = pDecManager->startDecodeByName(config.name);
      if (!bStart){
        LOG_INFO("started task {} failed!", config.name);
        pDecManager->closeDecoderByName(config.name);
        return false;
      }
  

  #ifdef POST_USE_RABBITMQ

      mq_type_t mq_type = mq_type_t::ALARM_MQ;

      pDecManager->set_mq_callback(config.name, std::bind(&mq::Manager::publish, mq_manager_, mq_type, std::placeholders::_1, true));
  #endif
  

      // 人车物跟踪
      if (task_has_vpt_algor(task_id))
        vpt_process.addTaskTracker(task_id, 1, 1, skip_frame_);
  
          // 人脸跟踪
  #ifdef WITH_FACE_DET_SS
      if (task_has_face_algor(task_id))
        m_face_det_ai_engine.add_tracker(task_id, skip_frame_); // 跳帧数暂时写死
  #endif
  
      m_FinishedTaskMtx.lock();
      m_FinishedTaskMap[task_id] = false;
      m_FinishedTaskMtx.unlock();
  
      auto algor_map = m_task_param_manager->get_task_other_param(task_id);
      if (algor_map != nullptr){
        /* 如果开启了行人 机动车非法闯入功能 生成闯入区域mask */
        if (algor_map->find(algorithm_type_t::PEDESTRIAN_TRESPASS) != algor_map->end()) {
          pedestrian_vehicle_trespass_.pedestrianvehicletrespass_init_region(
              task_id, algorithm_type_t::PEDESTRIAN_TRESPASS, input_image_width, input_image_height);
        }
  
        if (algor_map->find(algorithm_type_t::VEHICLE_TRESPASS) != algor_map->end()) {
          pedestrian_vehicle_trespass_.pedestrianvehicletrespass_init_region(
              task_id, algorithm_type_t::VEHICLE_TRESPASS, input_image_width, input_image_height);
        }
  
        // 设置定时截图的时间间隔
        auto algor = algor_map->find(algorithm_type_t::VIDEO_TIMING_SNAPSHOT);
        if(algor != algor_map->end()){
          task_param_manager::algo_param_type_t_* cur_task_params = algor->second;
          if(cur_task_params){
            int frame_stride = ((algor_config_video_timing_snapshot*)cur_task_params->algor_param)->frame_stride;
            LOG_INFO("task {} snap time interval: {} s", config.name, frame_stride);
            // 单位是ms, 乘以 1000 编成s
            dec->setSnapTimeInterval(frame_stride * 1000);
          }
        }

        // 测试代码
        // auto algor_face = algor_map->find(algorithm_type_t::FACE_SNAPSHOT);
        // if(algor_face != algor_map->end()){
        //   const algor_basic_config_param_t *cur_param = ((algor_init_config_param_t *)(algor_face->second))->basic_param;
        //   LOG_INFO("face_snapshot, result_folder: {}  result_folder_little: {}", cur_param->result_folder, cur_param->result_folder_little);

      // }

      }

      LOG_INFO("started task {} successed!", config.name);
  
      return true;
  }
  

  int CMultiSourceProcess::CountRunningTask() {
    DecoderManager* pDecManager = DecoderManager::getInstance();
    return pDecManager->count();
  }
  

  bool CMultiSourceProcess::task_has_vpt_algor(const std::string &task_id){
      //! TODO: create enum iterator.
      auto algor_map = m_task_param_manager->get_task_other_param(task_id);
      if (algor_map == nullptr)
        return false;
  
      return (algor_map->find(algorithm_type_t::HUMAN_GATHER) != algor_map->end() ||
              algor_map->find(algorithm_type_t::HUMAN_SNAPSHOT) != algor_map->end() ||
              algor_map->find(algorithm_type_t::NONMOTOR_VEHICLE_SNAPSHOT) != algor_map->end() ||
              algor_map->find(algorithm_type_t::SMOKING_DET) != algor_map->end() ||
              algor_map->find(algorithm_type_t::NO_REFLECTIVE_CLOTHING) != algor_map->end() ||
              algor_map->find(algorithm_type_t::NO_SAFETY_HELMET) != algor_map->end() ||
              algor_map->find(algorithm_type_t::CALL_PHONE_DET) != algor_map->end() ||
              algor_map->find(algorithm_type_t::VEHICLE_SNAPSHOT) != algor_map->end() ||
              algor_map->find(algorithm_type_t::TAKEAWAY_MEMBER_CLASSIFICATION) != algor_map->end() ||
              algor_map->find(algorithm_type_t::PEDESTRIAN_FALL) != algor_map->end() ||
              algor_map->find(algorithm_type_t::PEDESTRIAN_FIGHT) != algor_map->end() ||
              algor_map->find(algorithm_type_t::PEDESTRIAN_RETROGRADE) != algor_map->end() ||
              algor_map->find(algorithm_type_t::VEHICLE_RETROGRADE) != algor_map->end() ||
              algor_map->find(algorithm_type_t::PEDESTRIAN_TRESPASS) != algor_map->end() ||

              algor_map->find(algorithm_type_t::VEHICLE_TRESPASS) != algor_map->end() ||
              algor_map->find(algorithm_type_t::NONMOTOR_VEHICLE_NOHELMET) != algor_map->end() ||
              algor_map->find(algorithm_type_t::NONMOTOR_VEHICLE_OVERMAN) != algor_map->end() ||

              algor_map->find(algorithm_type_t::NONMOTOR_VEHICLE_USEPHONE) != algor_map->end() ||
              algor_map->find(algorithm_type_t::NONMOTOR_VEHICLE_REFIT) != algor_map->end() ||

              algor_map->find(algorithm_type_t::NONMOTOR_RUNNING_REDLIGHTS) != algor_map->end() ||
              algor_map->find(algorithm_type_t::NONMOTOR_CEOSSPARKLINE) != algor_map->end() ||
              algor_map->find(algorithm_type_t::NONMOTOR_IN_VEHICLELANE) != algor_map->end() ||

              algor_map->find(algorithm_type_t::NONMOTOR_WRONGDIRECTION) != algor_map->end() ||

              algor_map->find(algorithm_type_t::PERSON_IN_VEHICLELANE) != algor_map->end() ||
              algor_map->find(algorithm_type_t::PERSON_RUNNING_REDLIGHTS) != algor_map->end() ||
              algor_map->find(algorithm_type_t::PERSON_CROSS) != algor_map->end() ||

              algor_map->find(algorithm_type_t::VEHICLE_WRONGDIRECTION) != algor_map->end() ||

              algor_map->find(algorithm_type_t::VEHICLE_SOLIDLINETURNAROUND) != algor_map->end() ||
              algor_map->find(algorithm_type_t::VEHICLE_NOTGIVEWAY) != algor_map->end() ||
              algor_map->find(algorithm_type_t::VEHICLE_NOTDECELERATION) != algor_map->end() ||

              algor_map->find(algorithm_type_t::TRICYCLE_MANNED) != algor_map->end() ||
              algor_map->find(algorithm_type_t::TRUCK_MANNED) != algor_map->end());

  }
  
  bool CMultiSourceProcess::task_has_face_algor(const std::string &task_id){
      auto algor_map = m_task_param_manager->get_task_other_param(task_id);
      if (nullptr != algor_map){
        return algor_map->find(algorithm_type_t::FACE_SNAPSHOT) != algor_map->end();
      }
      
      return false;
  }
  
  void CMultiSourceProcess::decoded_cbk(DeviceMemory* devFrame){
  
    do{
      if(m_bfinish){
        break;
      }
      m_DataListMtx.lock();

      if(m_RgbDataList.size() >= 100){ //缓存100帧

        m_DataListMtx.unlock();
        std::this_thread::sleep_for(std::chrono::milliseconds(3));
        continue;
      }
      m_RgbDataList.push_back(devFrame);
      m_DataListMtx.unlock();
      break;
    }while (true);
  }
  
  void CMultiSourceProcess::task_finished(const string task_id){
  
    std::lock_guard<std::mutex> l(m_FinishedTaskMtx);
  
    m_FinishedTaskMap[task_id] = true;
  
    LOG_INFO("task {} finished!", task_id);
  }
  
  bool CMultiSourceProcess::PauseTask(const string taskID){
      DecoderManager* pDecManager = DecoderManager::getInstance();
      return pDecManager->pauseDecoder(taskID);
  }
  
  bool CMultiSourceProcess::RestartTask(const string taskID){
      DecoderManager* pDecManager = DecoderManager::getInstance();
      return pDecManager->resumeDecoder(taskID);
  }
  
  bool CMultiSourceProcess::FinishTask(const string taskID){
      DecoderManager* pDecManager = DecoderManager::getInstance();

      // return pDecManager->closeDecoderByName(taskID);
  
      // 记录主动结束的任务
      bool flag = pDecManager->closeDecoderByName(taskID);
  #ifdef POST_USE_RABBITMQ
      if (flag) {
        std::lock_guard<std::mutex> l1(m_ActiveFinishedTaskMtx);
        m_ActiveFinishedTaskMap[taskID] = true;
      }
  #endif
      return flag;

  }
  
  int CMultiSourceProcess::snapshot_task(std::string& uri_or_name, const std::string& file_name, bool bInTask) {
    DecoderManager* pDecManager = DecoderManager::getInstance();
  
    DeviceMemory* devMem = nullptr;
    if(bInTask) {
      LOG_INFO("snap in task");
      devMem = pDecManager->snapshot_in_task(uri_or_name);
    } else {
      LOG_INFO("snap out task");
      devMem = pDecManager->snapshot_out_task(uri_or_name, m_devId);
    }
    if(devMem != nullptr) {
      JpegUtil jpegUtil;
      jpegUtil.jpeg_init(m_devId);
  
      acldvppPicDesc *vpcDesc= acldvppCreatePicDesc();
      acldvppSetPicDescData(vpcDesc, devMem->getMem()); 
      acldvppSetPicDescFormat(vpcDesc, PIXEL_FORMAT_YUV_SEMIPLANAR_420);
      acldvppSetPicDescWidth(vpcDesc, devMem->getWidth());
      acldvppSetPicDescHeight(vpcDesc, devMem->getHeight());
      acldvppSetPicDescWidthStride(vpcDesc, devMem->getWidthStride());
      acldvppSetPicDescHeightStride(vpcDesc, devMem->getHeightStride());
      acldvppSetPicDescSize(vpcDesc, devMem->getSize());
  
      jpegUtil.jpeg_encode(vpcDesc, file_name);
      
      acldvppDestroyPicDesc(vpcDesc);
  
      jpegUtil.jpeg_release();
  
      delete devMem;
      devMem = nullptr;
  

      return SUCCESS;
    }
  

    return FAILED;
  }
  
  int CMultiSourceProcess::SnapShot(task_param _cur_task_param) {
    LOG_INFO("begin SnapShot task: {}", _cur_task_param.task_id);
  
    /* 添加任务参数 */
    auto &algor_config_param = _cur_task_param.algor_config_params[0]; // 默认算法数量为1
    
    /* 拷贝通用的算法参数 */
    auto src_basic_param = algor_config_param.algor_init_config_param->basic_param;
    if (src_basic_param->result_folder) {
      CreateResultFolder(src_basic_param->result_folder, "");
    } else {
      src_basic_param->result_folder = "";
    }
  
    std::string cur_timestamp_ms = std::to_string(helpers::timer::get_timestamp<std::chrono::milliseconds>());
    std::string file_name = std::string(src_basic_param->result_folder) + helpers::os::sep + _cur_task_param.task_id  + "_snap_" + cur_timestamp_ms + ".jpg";
  
    // file_name = "/mnt/data/cmhu/tongtu/Linux_3rdparty/tsl_sdk_20220801/res/snap.jpg";
    string task_id = _cur_task_param.task_id;
    int bSnapShot = snapshot_task(task_id, file_name, true);
    if (FAILED == bSnapShot){
      string url = _cur_task_param.ipc_url;
      bSnapShot = snapshot_task(url, file_name, false);
    }
    if(SUCCESS == bSnapShot){
  #ifdef POST_USE_RABBITMQ
    auto json_str = helpers::gen_json::gen_screen_json(task_id, file_name);
    // mq_manager_->publish(mq_type_t::GET_TASK_MQ, json_str.c_str(), true);
    mq_manager_->publish(mq_type_t::SCREENSHORT_TASK_MQ, json_str.c_str(), true);
  #endif
      LOG_INFO("SnapShot task: {} success", task_id);

    } else {
      LOG_ERROR("SnapShot task: {} failed !", task_id);

    }
  
    return bSnapShot;
  }
  
  void CMultiSourceProcess::CloseAllTask(){
      m_bfinish = true;

      atlas_licence_close(&(skt_handle));	//授权

      DecoderManager* pDecManager = DecoderManager::getInstance();
      pDecManager->closeAllDecoder();
      
      if(m_pAlgorthimThread){
          m_pAlgorthimThread->join();
          m_pAlgorthimThread = nullptr;
      }
  

      if(m_recode_thread){
          m_recode_thread->join();
          m_recode_thread = nullptr;
      }
  

      if(m_timing_snapshot_thread != nullptr && m_timing_snapshot_thread->joinable()){
        m_timing_snapshot_thread->join();
        m_timing_snapshot_thread = nullptr;
      }
  
      m_DataListMtx.lock();
      while (!m_RgbDataList.empty()){
          DeviceMemory* gpuMem = m_RgbDataList.front();
          delete gpuMem;
          gpuMem = nullptr;
          m_RgbDataList.pop_front();
      }
      m_DataListMtx.unlock();
  
      int size = m_RgbDataList.size();
      bool bEmpty = m_RgbDataList.empty();
      
      LOG_INFO("CloseAllTask exit.");
  }
  

  void CMultiSourceProcess::CloseAllTask2() {
    DecoderManager* pDecManager = DecoderManager::getInstance();
    pDecManager->closeAllDecoder();
  }
  

  void CMultiSourceProcess::clear_finished_task(){// 清理已经结束的任务
  
    std::lock_guard<std::mutex> l1(m_FinishedTaskMtx);
    std::lock_guard<std::mutex> l2(m_DataListMtx);
  
    for (auto iter_finished = m_FinishedTaskMap.begin(); iter_finished!=m_FinishedTaskMap.end(); ){
      if(iter_finished->second){
        // 解码已经结束
        // 判断数据对列中是否还有数据
        string task_id = iter_finished->first;
        bool bFinished = true;
        for (auto iter = m_RgbDataList.begin(); iter!=m_RgbDataList.end(); ++ iter){
          DeviceMemory* gpuMem = *iter;
          if(task_id == gpuMem->getId()){
            bFinished = false;
            break;
          }
        }
  
        if (bFinished){
          // 解码器已经结束，且数据队列中没有改任务的数据，则做最后任务清理工作
          finish_task(task_id,false);
          iter_finished = m_FinishedTaskMap.erase(iter_finished);
          continue;
        }
      }
  
      ++ iter_finished;

    } 

  }
  
  bool CMultiSourceProcess::finish_task(const string taskID, const bool delete_snapshot){
  
    // 任务结束，关闭跟踪
    if (!vpt_process.finishTaskTracker(taskID))
      LOG_ERROR("Finish VPT Tracker failed, task_id: {}", taskID);

  #ifdef POST_USE_RABBITMQ

    {
      // 外部主动结束的任务不推送mq
      std::lock_guard<std::mutex> mlock(m_ActiveFinishedTaskMtx);
      if (!(m_ActiveFinishedTaskMap.count(taskID) && m_ActiveFinishedTaskMap[taskID])) {
        auto json_str = helpers::gen_json::gen_office_task_heart_beat_json({taskID});
        mq_manager_->publish(mq_type_t::HEART_BEAT_MQ, json_str.c_str(), true);
      }
      if (m_ActiveFinishedTaskMap.count(taskID))  m_ActiveFinishedTaskMap.erase(taskID);
    }

  #endif
  

  // #ifdef WITH_SECOND_PROCESS

    pedestrian_vehicle_retrograde_.force_release_result(taskID); //221024 byzsh

  // #endif

  #ifdef USE_VILLAGE 

    tricycle_manned_.force_release_result(taskID);
    truck_manned_.force_release_result(taskID);
    motor_hsprocess_.force_release_result(taskID);

    motor_refit_phoneprocess_.force_release_result(taskID);
    // motor_phoneprocess_.force_release_result(taskID);

    motor_refitprocess_.force_release_result(taskID);

    traffic_lightprocess_.force_release_result(taskID);

    roadseg_corrprocess_.force_release_result(taskID);

    m_RoadSegTaskMtx.lock();
    if (m_RoadSegTaskMap.count(taskID)) {
      for (int i = 0; i < m_RoadSegTaskMap[taskID].traffic_region.size(); i++) {
        vector<int>().swap(m_RoadSegTaskMap[taskID].traffic_region[i]);
      }
      for (int i = 0; i < m_RoadSegTaskMap[taskID].fence_region.size(); i++) {
        vector<int>().swap(m_RoadSegTaskMap[taskID].fence_region[i]);
      }
      vector<int>().swap(m_RoadSegTaskMap[taskID].labels); 
      vector<int>().swap(m_RoadSegTaskMap[taskID].fence_labels);
      vector<vector<int>>().swap(m_RoadSegTaskMap[taskID].traffic_region); 
      vector<vector<int>>().swap(m_RoadSegTaskMap[taskID].fence_region); 
      m_RoadSegTaskMap.erase(taskID);
    }
    m_RoadSegTaskMtx.unlock();
    m_TotalObjMtx.lock();
    for(auto ss = m_total_obj_info.begin(); ss != m_total_obj_info.end(); ) {
  		if (strcmp(ss->first.video_id.c_str(), taskID.c_str()) == 0) {
  			vector<sy_point>().swap(ss->second.center_points);
  			m_total_obj_info.erase(ss++);
  		}
  		else ss++;
  	}
    m_TotalObjMtx.unlock();

  #endif

    m_task_param_manager->delete_task_param(taskID);

    m_snapshot_reprocessing->release_finished_locus_snapshot(taskID);

    m_snapshot_reprocessing->release_finished_face_locus_snapshot(taskID);

    m_snapshot_reprocessing->release_village_finished_locus_snapshot(taskID);
  

    return true;
  }
  
  int CMultiSourceProcess::algorthim_process_thread(){
      LOG_INFO("algorthim_process_thread start...");
  

      ACL_CALL(aclrtSetDevice(m_devId), ACL_SUCCESS, 1);

      aclrtContext ctx;

      ACL_CALL(aclrtCreateContext(&ctx, m_devId), ACL_SUCCESS, 1);

  
      while (true){

  #ifdef USE_VILLAGE

          if (!CheckTime()) {
            LOG_FATAL("authority failed!");
            break;
          }

  #else
          //授权相关------------------------------------
          if(check_label == -1) {
            printf("sy_atlas_licence_check failed.\n");
            break;
          }
          //获取系统时间，每个月1号check一次授权
          struct tm* info;
          int nYear, nMonth, nDay;
          time_t raw;
          time(&raw);
          info = localtime(&raw);
          //nYear = info->tm_year + 1900;
          //nMonth = info->tm_mon + 1;
          nDay = info->tm_mday;
          if(nDay==1) {
            if(check_label ==0) {
              std::cout << "atlas_licence_check start."  << std::endl;
              atlas_licence_check_result check_result;
              memset(&check_result, 0, sizeof(atlas_licence_check_result));
              check_result.code = -1;
  
              int res = atlas_licence_check(skt_handle, check_param, &check_result);
              if(res!=0) {
                std::cout << "sy_licence_check failed." << std::endl;
                break;
              }
              //std::cout << "code:" << check_result.code << ",msg:" << check_result.msg << std::endl;
              if(check_result.code!=0) {
                check_label = -1;
                std::cout << "code:" << check_result.code << ",msg:" << check_result.msg << std::endl;
                break;
              }
              std::cout << "atlas_licence_check end."  << std::endl;
              check_label =1;
            }
          }
          else {
            check_label =0;
          }
  	      //授权相关------------------------------------
  #endif

          if(m_bfinish){
              break;
          }
  

          clear_finished_task(); 

  
          vector<DeviceMemory*> vec_gpuMem;
          m_DataListMtx.lock();

          while (!m_RgbDataList.empty()){
              DeviceMemory* gpuMem = m_RgbDataList.front();

              // if(gpuMem->getMem() == nullptr) {
              if(gpuMem->getMem() == nullptr || gpuMem->getFrameNb() % skip_frame_) { //跳帧

                // 错误数据，直接删除

                if(gpuMem->getMem() == nullptr) LOG_WARN("mem is null");

                delete gpuMem;
                gpuMem = nullptr;

              } else {
                vec_gpuMem.push_back(gpuMem);
              }

              m_RgbDataList.pop_front();
              if(vec_gpuMem.size() >= m_batch_size){
                  break;
              }
          }
          m_DataListMtx.unlock();

          
          if(vec_gpuMem.size() <= 0){ 

              std::this_thread::sleep_for(std::chrono::milliseconds(3));
              continue;
          }
  
          aclrtSetCurrentContext(ctx);

          // auto record_t1 = std::chrono::system_clock::now();

  #ifdef USE_VILLAGE
          algorthim_road_seg(vec_gpuMem);
  #endif

          auto vptprocess1 = std::chrono::system_clock::now();

          algorthim_vpt(vec_gpuMem); 

          auto vptprocess2 = std::chrono::system_clock::now(); 
          auto vptprocess_time = std::chrono::duration_cast<std::chrono::milliseconds>(vptprocess2 - vptprocess1).count(); 
          LOG_INFO("Process per frame cost time: {} ms fps: {}", vptprocess_time/vec_gpuMem.size(), 1000/(vptprocess_time/vec_gpuMem.size()));

  
  #ifdef WITH_FACE_DET_SS
          algorthim_face_detect(vec_gpuMem);
  #endif

          // auto record_t2 = std::chrono::system_clock::now();
          // auto record_time = std::chrono::duration_cast<std::chrono::milliseconds>(record_t2 - record_t1).count(); 
          // LOG_INFO("Record per frame cost time: {} ms fps: {}", record_time/vec_gpuMem.size(), 1000/(record_time/vec_gpuMem.size()));

          for(int i=0;i < vec_gpuMem.size(); i++){
              DeviceMemory* mem = vec_gpuMem[i];
              if(mem->getSize() <= 0){
                continue;
              }
              delete mem;
              mem = nullptr;
          }
          vec_gpuMem.clear();
  
      }
      
      aclrtDestroyContext(ctx);
  
      LOG_INFO("algorthim_process_thread exit.");
  
      return 0;
  }
  

  
  int CMultiSourceProcess::algorthim_road_seg(vector<DeviceMemory*> vec_gpuMem) {
  
      vector<string> roadSeg_task_id;
      vector<sy_img> roadSeg_imgs;
      vector<DeviceMemory*> vec_segMem;
      m_RoadSegTaskMtx.lock();
      for (int i = 0; i < vec_gpuMem.size(); i++) {
          DeviceMemory* mem = vec_gpuMem[i];
          long long frame_number = mem->getFrameNb();
          if (m_RoadSegTaskMap[mem->getId()].isseg && frame_number % 15000) { //10min重复一次
              continue;
          }
          LOG_INFO("task: {} frame_num: {} perform road_seg.", mem->getId(), mem->getFrameNb());
          sy_img img;
          img.w_ = mem->getWidth();     img.h_ = mem->getHeight();
          img.c_ = mem->getChannel();   img.data_ = mem->getMem();
          roadSeg_imgs.push_back(img);
          roadSeg_task_id.push_back(mem->getId());
          vec_segMem.push_back(mem);
          m_RoadSegTaskMap[mem->getId()].isseg = true; //
      }
      m_RoadSegTaskMtx.unlock();
  
      /* 待检测的图片不为空 开始检测 */
      if (!roadSeg_imgs.empty()) {
          vector<vector<vector<int>>> traffic_region, fence_region; //道路区域(多个)
          vector<vector<int>> labels, fence_labels; //道路区域对应的类别：机动车道、非机动车道等	
          seg_process.process_gpu(roadSeg_imgs.data(), vec_segMem, roadSeg_task_id, traffic_region, labels);
          seg3cls_process.process_gpu(roadSeg_imgs.data(), vec_segMem, roadSeg_task_id, fence_region, fence_labels);
  
          int bidx = 0;
          m_RoadSegTaskMtx.lock();
          for (auto _task_id_iter = roadSeg_task_id.begin(); _task_id_iter != roadSeg_task_id.end(); ++_task_id_iter, ++ bidx) {
  					auto task_id = *_task_id_iter;
            m_RoadSegTaskMap[task_id].traffic_region.assign(traffic_region[bidx].begin(), traffic_region[bidx].end());
            m_RoadSegTaskMap[task_id].labels.assign(labels[bidx].begin(), labels[bidx].end());
            m_RoadSegTaskMap[task_id].fence_region.assign(fence_region[bidx].begin(), fence_region[bidx].end());
            m_RoadSegTaskMap[task_id].fence_labels.assign(fence_labels[bidx].begin(), fence_labels[bidx].end());
            m_RoadSegTaskMap[task_id].isseg = true;
          }
          m_RoadSegTaskMtx.unlock();
      }
      vec_segMem.clear();
      roadSeg_task_id.clear();
      roadSeg_imgs.clear();
  
      return 0;
  }