Hu Chunming / src

  #include "MultiSourceProcess.h"
  

  #ifdef POST_USE_RABBITMQ
    #include "../helpers/gen_json.hpp"
  #endif
  
  #include "../helpers/logger.hpp"
  #include "../helpers/os_helper.hpp"

  
  #include "authority.h"
  

  #include "NvJpegEncoder.hpp"
  

  #ifdef AUTHORIZATION
  #include <boost/thread.hpp>
  #ifdef _MSC_VER
  
  #define productSN "4ACFCBE67EF645AB8F0B4CFCDD1926F1" // WINDOWS
  #else
  #define productSN "4FD45501D5104F0C8C4BE530FC872F46" // LINUX
  #endif
  #endif
  
  #define SKIP_FRAME 5
  
  /**
   * 注意： gpuFrame 在解码器设置的显卡上，后续操作要十分注意这一点，尤其是多线程情况
   * */
  void decoded_cbk(const void * userPtr, AVFrame * gpuFrame){
      decode_cbk_userdata* ptr = (decode_cbk_userdata*)userPtr;
      if (ptr!= nullptr)
      {   
          CMultiSourceProcess* _this = (CMultiSourceProcess*)ptr->opaque;
          if(nullptr != _this){

              _this->post_decode_thread(ptr, gpuFrame);

          }
      }
  }
  
  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->decode_finished_thread(ptr->_cur_task_param);
        }
    }
  }
  

  CMultiSourceProcess::CMultiSourceProcess(){
    m_bfinish = false;

    m_bProcessEnd = false;

  }
  
  CMultiSourceProcess::~CMultiSourceProcess(){
  
  }
  
  /* 授权校验线程 */
  #ifdef AUTHORIZATION
  void CMultiSourceProcess::check_thread() {
    int res = -1;
  #ifndef _MSC_VER
    char wtime[15];
    memset(wtime, 0, 15);
    char *time = wtime;
  #endif
  
    while (1) {
      if(m_bfinish){
        break;
      }
  #ifdef _MSC_VER
      res = sy_licence(productSN);
  #else
      res = sy_licence(productSN, &time);
  #endif
      if (res < 0) {
        licence_status = licence_status - 1;
      } else {
        if (licence_status < 0) {
          licence_status = 0;
        }
      }
      std::this_thread::sleep_for(std::chrono::seconds(300));
    }
  }
  #endif
  

  /* @InitAlgorthim
   * @Description: 初始化全局参数 + 根据配置初始化算法模型
   */
  int CMultiSourceProcess::InitAlgorthim(tsl_aiplatform_param vptParam) {
    licence_status_ = -1;

  
    m_bfinish = false;
  

    int ret = SUCCESS;
  
  #ifdef AUTHORIZATION
  #ifdef _WIN32
    if (SUCCESS == (ret = sy_licence(productSN)))
  #elif __linux__
    char wtime[15];
    memset(wtime, 0, 15);
    char *time = wtime;
    if (SUCCESS == (ret = sy_licence(productSN, &time)))
  #endif //
  #else  // ifdef AUTHORIZATION
    ret = sy_time_check(2023, 2); // license_check(param.auth_license, productSN);//
    if (ret == SUCCESS)
  #endif // ifdef AUTHORIZATION

    {
      /*初始化全局参数*/

      set_default_logger(LogLevel(vptParam.log_level), "multi_source_video_process", vptParam.log_path, vptParam.log_mem,
                         vptParam.log_mem);
  
      cuInit(0);
      int device_count = 0;
      cuDeviceGetCount(&device_count);
  
      if (vptParam.gpuid >= device_count) {
        LOG_ERROR("gpu id ge host gpu device num ({} vs {})", vptParam.gpuid, device_count);
        return GPUID_PARAM_ERROR;
      }
  

      // 设置gpuid
      gpu_id_ = to_string(vptParam.gpuid);
  
      cudaSetDevice(atoi(gpu_id_.c_str()));
      cuda_common::setColorSpace( ITU_709, 0 );
  

      CUdevice dev = 0;
      size_t memSize = 0;
      dev = vptParam.gpuid;
  
      CUresult rlt = CUDA_SUCCESS;
      rlt = cuDeviceTotalMem(&memSize, dev);
  

      double gpu_total_memory = (float)memSize / (1024 * 1024);

  
      if (gpu_total_memory < 9000) // small gpu memory
        section_batch_size_ = 10;
      else
        section_batch_size_ = 20;
  
      /*初始化各个模块：任务管理、快照处理、快照保存、MQ*/
      m_task_param_manager = task_param_manager::getInstance();
      m_snapshot_reprocessing = snapshot_reprocessing::getInstance();
      m_save_snapshot_reprocessing = new save_snapshot_reprocessing();
  #ifdef POST_USE_RABBITMQ
      mq_manager_ = new mq::Manager();
  #endif
  
      /*初始化各个算法模块*/
      /* Init Model */
      {
        VPTProcess_PARAM param{};
        {
          param.gpuid = vptParam.gpuid;
          // param.threshold = 0.6;
          param.threshold = 0.45;
          param.max_batch = section_batch_size_;
          param.serialize_file = "./serialize_file/FPN_VPT";
          param.auth_license = /*vptParam.auth_license*/ "sy_tsl_aiplatform_sdk_2021";

        }
  
        VPT_Handle_ = nullptr;
  
        if (0 != (ret = VPT_Init(VPT_Handle_, param))) {
          LOG_FATAL("Init VPT failed");
          return ret;
        }
  
  #ifdef WITH_SECOND_PROCESS
        if (!takeaway_member_.init(vptParam.gpuid)) {
          LOG_FATAL("Init TakeAwayMember failed");
          return FALSE;
        }
  
        if (!fight_fall_cls_.init(vptParam.gpuid)) {
          LOG_FATAL("Init FightFellCls failed");
          return FALSE;
        }
  
        if (m_human_gather_statistics.human_gather_statistics_init() != 0) {
          LOG_FATAL("Init Human Gather failed");
          return FALSE;
        }
  
        {
          hat_callsmoke_vestuniform_det_param init_param;
          {
            init_param.gpuid = vptParam.gpuid;
            init_param.mode = init_param.gpuid >= 0 ? DEVICE_GPU : DEVICE_CPU;
  
            init_param.engine = ENGINE_TENSORRT;
            init_param.max_batch = section_batch_size_;
            init_param.auth_license = "shiyu_zhian_2021_jetson_sdk";
            init_param.serialize_file = "./serialize_file/pedestrian_safety_det";
          }
  
          if (!pedestrian_safety_detector_.init(init_param)) {
            LOG_FATAL("Init Pedestrian Safety failure");
            return FALSE;
          }
        }
  
        {
          // pedestrian_vehicle_retrograde_.init()
        }
  #endif
  #ifdef WITH_FACE_DET_SS
        // 人脸检测初始化
        facedet_ai_engine_param fd_param;
        fd_param.sdk_param.gpuid = vptParam.gpuid;
        fd_param.sdk_param.mode = DEVICE_GPU;
        fd_param.sdk_param.log = SY_CONFIG_CLOSE;
        fd_param.sdk_param.thresld = 0.6;
        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_CLOSE;
        fd_param.sdk_param.score_config = SY_CONFIG_OPEN;
        fd_param.sdk_param.max_result_count = 50;
        fd_param.sdk_param.max_batch_size_detect = section_batch_size_;
        fd_param.sdk_param.max_batch_size_ldmk = section_batch_size_;
        fd_param.sdk_param.max_batch_size_score = section_batch_size_;
        fd_param.sdk_param.max_batch_size_pose = section_batch_size_;
        fd_param.sdk_param.serialize_file = "./serialize_file/FD";
        fd_param.sdk_param.auth_license = "sy_tsl_aiplatform_sdk_2021";
        if (0 > m_face_det_ai_engine.init_ai_engine(fd_param)) {
          LOG_FATAL("Init face detection failed");
          return FALSE;
        }
  #endif
      }
  
      if (ret == SUCCESS) //
      {
        licence_status_ = 0;
  #ifdef AUTHORIZATION

        pthread_create(&m_authority_check_thread,0,
          [](void* arg)
          {
              CMultiSourceProcess* a=(CMultiSourceProcess*)arg;
              a->check_thread();
              return (void*)0;
          }
      ,this);

  #endif

      }
    } else {
      return AUTHOR_ERROR;
    }
  
  #ifdef AUTHORIZATION
  #ifdef __linux__
    if (wtime) {
      delete[] wtime;
      wtime = NULL;
    }
  #endif // #ifdef __linux__
  #endif // #ifdef AUTHORIZATION
  

    // 启动算法处理线程
    startProcessByGpuid();
  

    return ret;
  }
  

  bool CMultiSourceProcess::AddTask(task_param _cur_task_param){

  
      FFNvDecoderManager* pDecManager = FFNvDecoderManager::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 = decoded_cbk;
      config.cfg.decode_finished_cbk = decode_finished_cbk;
      config.cfg.force_tcp = true;  // rtsp用tcp

      config.cfg.gpuid = gpu_id_;

      FFNvDecoder* dec = pDecManager->createDecoder(config);
      if (!dec)
      {
          return false;
      }
  
      decode_cbk_userdata* userPtr = new decode_cbk_userdata;
      userPtr->_cur_task_param = _cur_task_param;
      userPtr->opaque = this;

      userPtr->opaque1 = dec;

      pDecManager->setUserPtr(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);

      cout << "task_id: " << task_id << " width: " << input_image_width << " height: " << input_image_height << endl;

  #ifdef WITH_SECOND_PROCESS
    /* 如果开启了行人 机动车非法闯入功能 生成闯入区域mask */
    auto new_task_algor_param = m_task_param_manager->get_task_other_param(task_id);

    if (new_task_algor_param->find(algorithm_type_t::PEDESTRIAN_TRESPASS) != new_task_algor_param->end()) {
      pedestrian_vehicle_trespass_.pedestrianvehicletrespass_init_region(
          task_id, algorithm_type_t::PEDESTRIAN_TRESPASS, input_image_width, input_image_height);
    }

    if (new_task_algor_param->find(algorithm_type_t::VEHICLE_TRESPASS) != new_task_algor_param->end()) {
      pedestrian_vehicle_trespass_.pedestrianvehicletrespass_init_region(
          task_id, algorithm_type_t::VEHICLE_TRESPASS, input_image_width, input_image_height);
    }
  #endif
  
      ((save_snapshot_reprocessing *)m_save_snapshot_reprocessing)->add_newtask(task_id);

      // 人车物跟踪
      if (task_has_vpt_algor(task_id))
        AddTaskTracker(VPT_Handle_, task_id);

      // 人脸跟踪
  #ifdef WITH_FACE_DET_SS
      if (task_has_face_algor(task_id))
        m_face_det_ai_engine.operator_tracker(task_id, ADDTASK, SKIP_FRAME); // 跳帧数暂时写死
  #endif

      // 所有参数都准备好之后再启动解码
      pDecManager->startDecodeByName(config.name);
      cout << "started task: " << config.name << endl;

  }
  

  // 启动算法处理线程

  void CMultiSourceProcess::startProcessByGpuid(){
    pthread_create(&m_gpuProcessthread,0,
        [](void* arg)
        {
            CMultiSourceProcess* process = (CMultiSourceProcess*)arg ;
            process->algorthim_process_thread();
            return (void*)0;
        }
    ,this);

  }
  

  void CMultiSourceProcess::post_decode_thread(decode_cbk_userdata* userPtr, AVFrame * gpuFrame){

    decode_cbk_userdata* ptr = (decode_cbk_userdata*)userPtr;
  
    FFNvDecoder* dec = (FFNvDecoder*)ptr->opaque1;
  
    if (gpuFrame->format == AV_PIX_FMT_CUDA && dec != nullptr){

      // cout << "decode task: " << dec->getName() << " gpuid: " << gpu_id_ << endl;

      GpuRgbMemory* gpuMem = new GpuRgbMemory(3, gpuFrame->width, gpuFrame->height, dec->getName(), gpu_id_ , true);

  
      if (gpuMem->getMem() == nullptr){
        cout << "new GpuRgbMemory failed !!!" << endl;
        return;
      }
      

      
      // cudaSetDevice(atoi(gpu_id_.c_str()));
      // cuda_common::setColorSpace( ITU_709, 0 );
      cudaError_t cudaStatus = cuda_common::CUDAToBGR((CUdeviceptr)gpuFrame->data[0],(CUdeviceptr)gpuFrame->data[1], gpuFrame->linesize[0], gpuFrame->linesize[1], gpuMem->getMem(), gpuFrame->width, gpuFrame->height);
      cudaDeviceSynchronize();
      if (cudaStatus != cudaSuccess) {
          cout << "CUDAToBGR failed !!!" << endl;
          return;
      }
  
      do{
        if(m_bfinish){

          break;

        }
        // TODO 本循环需要一个可以手动终止的开关
        m_QueueMtx.lock();
        if(m_RgbDataList.size() >= 40){
          m_QueueMtx.unlock();
          std::this_thread::sleep_for(std::chrono::milliseconds(1));
          continue;
        }
        m_RgbDataList.push_back(gpuMem);
        m_QueueMtx.unlock();
        break;
      }while (true);

    }
  }
  

  // 解码线程退出的最后回调的实际任务函数

  void CMultiSourceProcess::decode_finished_thread(task_param t_param){

    FFNvDecoderManager* pDecManager = FFNvDecoderManager::getInstance();
  
    while(true){
  
      if(m_bfinish){
        break;
      }
  
      std::this_thread::sleep_for(std::chrono::milliseconds(5));
  
      std::lock_guard<std::mutex> l(m_QueueMtx);
  
      // 确保 m_RgbDataList 中没有任务数据
      for (auto iter = m_RgbDataList.begin(); iter!=m_RgbDataList.end(); ++ iter){
        GpuRgbMemory* gpuMem = *iter;
        if(gpuMem->getId() == t_param.task_id){
          continue;
        }
      }
  
      // 确保 对应解码器中没有缓存的数据
      if (pDecManager->getCachedQueueLength(t_param.task_id) > 0){
        continue;
      }
      
      // 确保对应任务没有正在执行算法的数据
      if (m_bProcessEnd){
        // 任务结束，关闭跟踪
        if (!FinishTaskTracker(VPT_Handle_, t_param.task_id))
          LOG_ERROR("Finish VPT Tracker failed, task_id: {}", t_param.task_id);
  
        finish_task(t_param.task_id,false);
  
        break;
      }
    }

  }
  
  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());
  }
  
  bool CMultiSourceProcess::task_has_face_algor(const std::string &task_id){
      auto algor_map = m_task_param_manager->get_task_other_param(task_id);
      return algor_map->find(algorithm_type_t::FACE_SNAPSHOT) != algor_map->end();
  }
  
  void CMultiSourceProcess::cuda_free_wrap(sy_img &img) {
    CHECK(cudaFree(img.data_)) {
      img.c_ = 0;
      img.h_ = 0;
      img.w_ = 0;
      img.data_ = nullptr;
    }
  }
  
  // 算法处理函数，由算法线程调用

  void CMultiSourceProcess::algorthim_process_thread(){

    cudaSetDevice(atoi(gpu_id_.c_str()));
  

    map<string, int> task_id_to_n_frame;

  
    int sum = 0;

    while(true){

      m_bProcessEnd = true;

      /* step1. 授权check */
      if (licence_status_ <= -3) {
        LOG_FATAL("authority failed!");
        break;
      }
  

      if(m_bfinish){
        break;
      }
  

      /* step5. 凑齐的解码数据 拼batch */
      m_QueueMtx.lock();

      m_bProcessEnd = false;

      vector<GpuRgbMemory*> vec_gpuMem;

      for (auto iter = m_RgbDataList.begin(); iter!=m_RgbDataList.end(); ++ iter){

        GpuRgbMemory* gpuMem = *iter;

        vec_gpuMem.push_back(gpuMem);
        iter = m_RgbDataList.erase(iter);
        if(vec_gpuMem.size() >= 20){
          break;

        }

      }

  
      int batch_size = vec_gpuMem.size();

      if (batch_size <= 0){
        m_QueueMtx.unlock();
        std::this_thread::sleep_for(std::chrono::milliseconds(1));
        continue;
      }

      vector<string> task_list;
      sy_img *batch_img = new sy_img[batch_size];

      for (size_t i = 0; i < batch_size; i++){

        GpuRgbMemory* gpuMem = vec_gpuMem[i];

        batch_img[i].set_data(gpuMem->getWidth(), gpuMem->getHeight(), gpuMem->getChannel(), gpuMem->getMem());
        task_list.push_back(gpuMem->getId());
        ++task_id_to_n_frame[gpuMem->getId()];

  
        // string path = "/home/cmhu/data2/test/" + gpuMem->getId() + "_" + to_string(sum) + ".jpg";
        // saveJpeg(path.c_str(), gpuMem->getMem(), gpuMem->getWidth(), gpuMem->getHeight(), nullptr);  // 验证 CUDAToRGB 
        // sum ++;

      }
  
      m_QueueMtx.unlock();
  
      // VPT 检测
      algorthim_vpt(task_list, batch_img);
  
    #ifdef WITH_FACE_DET_SS
      algorthim_face_detect(task_list, batch_img);
    #endif
  
      // 清理显存
      for (size_t i = 0; i < vec_gpuMem.size(); i++){
        GpuRgbMemory* gpuMem = vec_gpuMem[i];
        delete gpuMem;
        gpuMem = nullptr;
      }
      vec_gpuMem.clear();
  
      if (batch_img != nullptr) {
        delete[] batch_img;
        batch_img = nullptr;
      }
    }
  
  #ifdef WITH_SECOND_PROCESS
    {
      std::string msg("taskId to n_frame pair has");
      for (auto &iter : task_id_to_n_frame)
        msg.append(fmt::format("({}: {})", iter.first.c_str(), iter.second));
  
      for (auto &iter : task_id_to_n_frame) {
        auto &task_id = iter.first;
        pedestrian_safety_detector_.force_release_result(task_id);
        pedestrian_vehicle_retrograde_.force_release_result(task_id);
        takeaway_member_.force_release_result(task_id);
      }
  
      LOG_DEBUG("{}", msg);
    }
  #endif

    m_bProcessEnd = true;

    cout << "algorthim_process_thread end. " << endl;

  }
  
  // VPT 检测
  void CMultiSourceProcess::algorthim_vpt(vector<string>& task_list, sy_img *batch_img){
    
    vector<string> vpt_interest_task_id;
    std::vector<sy_img> vpt_interest_imgs(0);
  
    for (size_t i = 0; i < task_list.size(); i++){
      string task_id = task_list[i];
      if (!task_has_vpt_algor(task_id))
        continue;
  
      vpt_interest_task_id.push_back(task_id);

      vpt_interest_imgs.emplace_back(batch_img[i]);

    }
  
    /* 待检测的图片不为空 开始检测 */
    if (!vpt_interest_imgs.empty()) {
      vector<vector<int>> deleteObjectID;
      deleteObjectID.resize(vpt_interest_task_id.size());
      vector<vector<onelevel_det_result>> unUsedResult;
      vector<onelevel_det_result> vptResult(0);
  
      /* 一级检测器，内部已完成跟踪操作 */
      VPT_Process_GPU2(VPT_Handle_, vpt_interest_imgs.data(), vpt_interest_task_id, vptResult, deleteObjectID, unUsedResult); // do det & track.
      /* 每帧都需要进行的操作函数（如人数统计功能） */
      everyframe_process(vpt_interest_task_id, vpt_interest_imgs.data(), vptResult); // modified by zsh
      /* 根据任务的配置 对目标进行筛选 */
      m_snapshot_reprocessing->screen_effective_snapshot2(vpt_interest_task_id, vptResult); //
  
  #ifdef MTASK_DEBUG_
      int idx_tmp = 0;
      for (auto task_id : vpt_interest_task_id)
        LOG_DEBUG("after screen_effective_snapshot task_id {} obj count {}", task_id, vptResult[idx_tmp++].obj_count);
  #endif
      /* 快照优选（内部可实现不同的快照优选策略） */
      m_snapshot_reprocessing->update_bestsnapshot2(vpt_interest_task_id, vpt_interest_imgs.data(),
                                                                  vptResult, deleteObjectID);
  #ifdef WITH_SECOND_PROCESS
      /* for pedestrian safety det. 行人安全分析算法模块 */
      algorthim_pedestrian_safety(vpt_interest_task_id, vpt_interest_imgs,vptResult);
  
      /* for retrograde & trespass algor 逆行&非法闯入算法模块 */
      algorthim_retrograde_trespass(vpt_interest_task_id, vpt_interest_imgs,vptResult,deleteObjectID);
  #endif
      /*  for snapshot algorithm. 轨迹结束目标 做最后的结果返回（当前返回算法结果+快照保存路径）*/
      algorithm_snapshot(vpt_interest_task_id, deleteObjectID);
  
  #ifdef WITH_SECOND_PROCESS
      /*  for fight fall cls. sync code. 打架跌倒算法模块 */
      algorithm_fight_fall(vpt_interest_task_id, vpt_interest_imgs, vptResult);
  
      /*  for takeaway member. 外卖员分析模块 */
      algorithm_takeaway_member_cls(vpt_interest_task_id, vpt_interest_imgs, vptResult);
  #endif
    }
  }
  
  // 行人安全分析算法模块
  void CMultiSourceProcess::algorthim_pedestrian_safety(vector<string>& vpt_interest_task_id, vector<sy_img>& vpt_interest_imgs, vector<onelevel_det_result>& vptResult){

  
  #ifdef WITH_SECOND_PROCESS
  

      vector<string> interest_task_id;

      vector<onelevel_det_result> interest_vpt_result;

      vector<sy_img> interest_imgs(0);
  
      int _idx = 0;
      for (auto _task_id_iter = vpt_interest_task_id.begin(); _task_id_iter != vpt_interest_task_id.end();
            ++_task_id_iter, ++_idx) // loop task_id;
      {
        auto task_id = *_task_id_iter;
        auto algor_map = m_task_param_manager->get_task_other_param(task_id);
  
        if (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::SMOKING_DET) != algor_map->end()) {
          interest_task_id.emplace_back(task_id);
          interest_imgs.emplace_back(vpt_interest_imgs[_idx]);
          interest_vpt_result.emplace_back(vptResult[_idx]);
        }
      }
  
      if (!interest_imgs.empty())
        pedestrian_safety_detector_.update_mstreams2(interest_task_id, interest_imgs.data(), interest_vpt_result);

  
  #endif

  }
  
  // 逆行&非法闯入算法模块

  void CMultiSourceProcess::algorthim_retrograde_trespass(vector<string>& vpt_interest_task_id, vector<sy_img>& vpt_interest_imgs
        , vector<onelevel_det_result>& vptResult ,vector<vector<int>>& deleteObjectID){

  #ifdef WITH_SECOND_PROCESS

    vector<string> interest_task_id;

    vector<onelevel_det_result> interest_vpt_result;

    vector<sy_img> interest_imgs(0);
  
    vector<string> trespass_interest_task_id;

    vector<onelevel_det_result> trespass_interest_vpt_result;
    vector<vector<int>> trespass_interest_deleteobjs;

    vector<sy_img> trespass_interest_imgs(0);
  
    int _idx = 0;
    for (auto _task_id_iter = vpt_interest_task_id.begin(); _task_id_iter != vpt_interest_task_id.end();
          ++_task_id_iter, ++_idx) // loop task_id;
    {
      auto task_id = *_task_id_iter;
      auto algor_map = m_task_param_manager->get_task_other_param(task_id);
  
      if (algor_map->find(algorithm_type_t::PEDESTRIAN_RETROGRADE) != algor_map->end() ||
          algor_map->find(algorithm_type_t::VEHICLE_RETROGRADE) != algor_map->end()) {
        interest_task_id.emplace_back(task_id);
        interest_imgs.emplace_back(vpt_interest_imgs[_idx]);
        interest_vpt_result.emplace_back(vptResult[_idx]);
      }
  
      if (algor_map->find(algorithm_type_t::PEDESTRIAN_TRESPASS) != algor_map->end() ||
          algor_map->find(algorithm_type_t::VEHICLE_TRESPASS) != algor_map->end()) {
        trespass_interest_task_id.emplace_back(task_id);
        trespass_interest_imgs.emplace_back(vpt_interest_imgs[_idx]);
        trespass_interest_vpt_result.emplace_back(vptResult[_idx]);
        trespass_interest_deleteobjs.emplace_back(deleteObjectID[_idx]);
      }
    }
  
    if (!interest_imgs.empty())
      pedestrian_vehicle_retrograde_.update_mstreams2(interest_task_id, interest_imgs.data(),
                                                                    interest_vpt_result);
  
    if (!trespass_interest_imgs.empty()) {
      pedestrian_vehicle_trespass_.update_mstreams2(
          trespass_interest_task_id, trespass_interest_imgs.data(), trespass_interest_vpt_result,
          trespass_interest_deleteobjs);
    }

  #endif

  }
  
  // for snapshot algorithm. 轨迹结束目标 做最后的结果返回（当前返回算法结果+快照保存路径）
  void CMultiSourceProcess::algorithm_snapshot(vector<string>& vpt_interest_task_id, vector<vector<int>> deleteObjectID) {
    auto task_iter = vpt_interest_task_id.begin();
  
    for (int i = 0; i < deleteObjectID.size(); i++, ++task_iter) // loop taskId.
    {
      for (int &j : deleteObjectID[i]) // loop algor type.
      {
        OBJ_KEY obj_key = {*task_iter, j};
        endframe_obj_process(obj_key, algorithm_type_t::PLACEHOLDER);
      }
      vector<int>().swap(deleteObjectID[i]); // free.
    }
    vector<vector<int>>().swap(deleteObjectID); // free.
  }
  
  /* 轨迹结束帧需要做的算法模块 */
  int CMultiSourceProcess::endframe_obj_process(const OBJ_KEY &obj_key, algorithm_type_t algor_type) {
    auto task_param_ptr = m_task_param_manager->get_task_algor_param(obj_key.video_id);

  
    // 该路任务开启了抓拍功能 开始抓拍保存；若未开启抓拍，清空显存资源
    if ((task_param_ptr->human_algors.find(algorithm_type_t::HUMAN_SNAPSHOT) != task_param_ptr->human_algors.end() ||
         task_param_ptr->vehicle_algors.find(algorithm_type_t::VEHICLE_SNAPSHOT) !=
             task_param_ptr->vehicle_algors.end() ||
         task_param_ptr->nonmotor_vehicle_algors.find(algorithm_type_t::NONMOTOR_VEHICLE_SNAPSHOT) !=
             task_param_ptr->nonmotor_vehicle_algors.end())) {
  #ifdef MTASK_DEBUG_
      LOG_DEBUG("task_id {} endframe_obj_process obj_id {}", obj_key.video_id, obj_key.obj_id);
  #endif
      m_save_snapshot_reprocessing->reprocessing_process(obj_key);
    } else {
      // printf("delete snapshot\n");
      m_snapshot_reprocessing->delete_finishtask_snapshot(obj_key.video_id, obj_key.obj_id);
    }
  
  #ifdef WITH_SECOND_PROCESS
    /* 开启外卖员算法模块，获取该目标的算法分析结果 返回结果+快照 最后释放资源 */
    if (task_param_ptr->nonmotor_vehicle_algors.find(algorithm_type_t::TAKEAWAY_MEMBER_CLASSIFICATION) !=
        task_param_ptr->nonmotor_vehicle_algors.end()) {
      auto result = takeaway_member_.get_result_by_objectid(ai_engine_module::obj_key_t{
          obj_key.obj_id, obj_key.video_id, algorithm_type_t::TAKEAWAY_MEMBER_CLASSIFICATION});
      if (result.get()) {
        auto json_str = helpers::gen_json::gen_takeaway_member_cls_json(obj_key.video_id, obj_key.obj_id, result->box,
                                                                        result->category);
        save_snapshot_process(obj_key, algorithm_type_t::TAKEAWAY_MEMBER_CLASSIFICATION, result->ori_img, result->roi_img,
                              0, json_str);
  
        CHECK(cudaFree((void *)(result->roi_img.data_)));
        CHECK(cudaFree((void *)(result->ori_img.data_)));
      }
    }
  
    /* 开启行人&机动车逆行算法模块，获取该目标的算法分析结果 返回结果+快照 最后释放资源 */
    if (task_param_ptr->human_algors.find(algorithm_type_t::PEDESTRIAN_RETROGRADE) !=
            task_param_ptr->human_algors.end() ||
        task_param_ptr->vehicle_algors.find(algorithm_type_t::VEHICLE_RETROGRADE) !=
            task_param_ptr->vehicle_algors.end()) {
      auto func_wrap = [&](const algorithm_type_t &algor_type) {
        auto results = pedestrian_vehicle_retrograde_.get_results_by_id(
            ai_engine_module::obj_key_t{obj_key.obj_id, obj_key.video_id, algor_type});
        if (results) {
          bool flag_pRetroGrade{true}, flag_vRetroGrade{true}; // modified by zsh
          for (unsigned idx = 0; idx < results->size(); ++idx) {
            auto &result = results->at(idx);
            // modified by zsh 一个id一种事件只报警一次------------------------------------------------------------------
            if(algor_type==algorithm_type_t::PEDESTRIAN_RETROGRADE && flag_pRetroGrade == true) {
                auto &&json_str = helpers::gen_json::gen_retrograde_json(obj_key.video_id, obj_key.obj_id, result.box, algor_type);
                save_snapshot_process(obj_key, algor_type, result.ori_img, result.roi_img, /*id=*/idx, json_str, true, result.ori_img_is_in_gpu, result.roi_img_is_in_gpu);
                flag_pRetroGrade = false;
            }
            if(algor_type==algorithm_type_t::VEHICLE_RETROGRADE && flag_vRetroGrade == true) {
                auto &&json_str = helpers::gen_json::gen_retrograde_json(obj_key.video_id, obj_key.obj_id, result.box, algor_type);
                save_snapshot_process(obj_key, algor_type, result.ori_img, result.roi_img, /*id=*/idx, json_str, true, result.ori_img_is_in_gpu, result.roi_img_is_in_gpu);  
                flag_vRetroGrade = false;
            }
            // -------------------------------------------------------------------------------------------------------
            if(0) {
            auto &&json_str =
                helpers::gen_json::gen_retrograde_json(obj_key.video_id, obj_key.obj_id, result.box, algor_type);
  
            save_snapshot_process(obj_key, algor_type, result.ori_img, result.roi_img, /*id=*/idx, json_str, true,
                                  result.ori_img_is_in_gpu, result.roi_img_is_in_gpu);
            }
            if (result.roi_img.data_ != nullptr) {
              if (result.roi_img_is_in_gpu) {
                LOG_DEBUG("free roi gpu memory.");
                CHECK(cudaFree((void *)(result.roi_img.data_)))
              }
              result.roi_img.data_ = nullptr;
            }
  
            if (result.ori_img.data_ != nullptr) {
              if (result.ori_img_is_in_gpu) {
                LOG_DEBUG("free ori gpu memory.");
                CHECK(cudaFree((void *)(result.ori_img.data_)))
              }
              result.ori_img.data_ = nullptr;
            }
          }
        }
      };
  
      if (task_param_ptr->human_algors.find(algorithm_type_t::PEDESTRIAN_RETROGRADE) !=
          task_param_ptr->human_algors.end())
        func_wrap(algorithm_type_t::PEDESTRIAN_RETROGRADE);
  
      if (task_param_ptr->vehicle_algors.find(algorithm_type_t::VEHICLE_RETROGRADE) !=
          task_param_ptr->vehicle_algors.end())
        func_wrap(algorithm_type_t::VEHICLE_RETROGRADE);
    }
  
    /* 开启行人&机动车非法闯入算法模块，获取该目标的算法分析结果 返回结果+快照 最后释放资源 */
    if (task_param_ptr->human_algors.find(algorithm_type_t::PEDESTRIAN_TRESPASS) != task_param_ptr->human_algors.end() ||
        task_param_ptr->vehicle_algors.find(algorithm_type_t::VEHICLE_TRESPASS) != task_param_ptr->vehicle_algors.end()) {
      auto func_wrap = [&](const algorithm_type_t &algor_type) {
        auto results = pedestrian_vehicle_trespass_.get_results_by_id(
            ai_engine_module::obj_key_t{obj_key.obj_id, obj_key.video_id, algor_type});
        if (results) {
          bool flag_pThePass{true}, flag_vThePass{true}; // modified by zsh 
          for (unsigned idx = 0; idx < results->size(); ++idx) {
            auto &result = results->at(idx);
            // modified by zsh 一个id一种事件只报警一次------------------------------------------------------------------
            if(algor_type==algorithm_type_t::PEDESTRIAN_TRESPASS && flag_pThePass == true) {
                auto &&json_str = helpers::gen_json::gen_retrograde_json(obj_key.video_id, obj_key.obj_id, result.box, algor_type);
                save_snapshot_process(obj_key, algor_type, result.ori_img, result.roi_img, /*id=*/idx, json_str, true, result.ori_img_is_in_gpu, result.roi_img_is_in_gpu); 
                flag_pThePass = false;
            }
            if(algor_type==algorithm_type_t::VEHICLE_TRESPASS && flag_vThePass == true) {
                auto &&json_str = helpers::gen_json::gen_retrograde_json(obj_key.video_id, obj_key.obj_id, result.box, algor_type);
                save_snapshot_process(obj_key, algor_type, result.ori_img, result.roi_img, /*id=*/idx, json_str, true, result.ori_img_is_in_gpu, result.roi_img_is_in_gpu);  
                flag_vThePass = false;
            }
            // -------------------------------------------------------------------------------------------------------
            if(0) {
            auto &&json_str =
                helpers::gen_json::gen_retrograde_json(obj_key.video_id, obj_key.obj_id, result.box, algor_type);
            // printf("%s\n", json_str.c_str());
            save_snapshot_process(obj_key, algor_type, result.ori_img, result.roi_img, /*id=*/idx, json_str, true,
                                  result.ori_img_is_in_gpu, result.roi_img_is_in_gpu);
            }
            if (result.roi_img.data_ != nullptr) {
              if (result.roi_img_is_in_gpu)
                CHECK(cudaFree((void *)(result.roi_img.data_)))
              result.roi_img.data_ = nullptr;
            }
  
            if (result.ori_img.data_ != nullptr) {
              if (result.ori_img_is_in_gpu)
                CHECK(cudaFree((void *)(result.ori_img.data_)))
              result.ori_img.data_ = nullptr;
            }
          }
        }
      };
  
      if (task_param_ptr->human_algors.find(algorithm_type_t::PEDESTRIAN_TRESPASS) != task_param_ptr->human_algors.end())
        func_wrap(algorithm_type_t::PEDESTRIAN_TRESPASS);
  
      if (task_param_ptr->vehicle_algors.find(algorithm_type_t::VEHICLE_TRESPASS) != task_param_ptr->vehicle_algors.end())
        func_wrap(algorithm_type_t::VEHICLE_TRESPASS);
    }
  
    /* 开启行人安全分析算法模块，获取该目标的算法分析结果 返回结果+快照 最后释放资源 */
    if (task_param_ptr->human_algors.find(algorithm_type_t::NO_REFLECTIVE_CLOTHING) !=
            task_param_ptr->human_algors.end() ||
        task_param_ptr->human_algors.find(algorithm_type_t::NO_SAFETY_HELMET) != task_param_ptr->human_algors.end() ||
        task_param_ptr->human_algors.find(algorithm_type_t::CALL_PHONE_DET) != task_param_ptr->human_algors.end() ||
        task_param_ptr->human_algors.find(algorithm_type_t::SMOKING_DET) != task_param_ptr->human_algors.end()) {
  
      LOG_TRACE("get result task_id {} obj_id {}", obj_key.video_id, obj_key.obj_id);
      auto results = pedestrian_safety_detector_.get_results_by_id(
          ai_engine_module::unique_obj_id_t{obj_key.obj_id, obj_key.video_id});
      // auto results = std::shared_ptr<ai_engine_module::pedestrian_safety_det::results_data_t>(nullptr);
  
      if (results) {
        //! loop per pedestrian safety attribute.
        long id = 0;
        bool flag_nReflect{true}, flag_nHelmet{true}, flag_cPhone{true}, flag_smoking{true}; // modified by zsh 
        LOG_DEBUG("  size of results is {}", results->size());
        for (auto result : *results) {
          for (auto algor_type_iter = result.algorithm_type_seq.begin();
               algor_type_iter != result.algorithm_type_seq.end(); ++algor_type_iter) {
            auto algor_type = *algor_type_iter;
            // modified by zsh 一个id一种事件只报警一次------------------------------------------------------------------
            if(algor_type==algorithm_type_t::NO_REFLECTIVE_CLOTHING && flag_nReflect == true) {
                auto json_str = helpers::gen_json::gen_pedestrian_safety_json(obj_key.video_id, obj_key.obj_id, result.box, algor_type);
                save_snapshot_process(obj_key, algor_type, result.ori_img, result.roi_img, /*id=*/id++, json_str, true, result.ori_img_is_in_gpu, result.roi_img_is_in_gpu); 
                flag_nReflect = false;
                LOG_TRACE("PEDESTRIAN task_id {} obj_id {} algor_type {}", obj_key.video_id, obj_key.obj_id, int(algor_type));
            }
            if(algor_type==algorithm_type_t::NO_SAFETY_HELMET && flag_nHelmet == true) {
                auto json_str = helpers::gen_json::gen_pedestrian_safety_json(obj_key.video_id, obj_key.obj_id, result.box, algor_type);
                save_snapshot_process(obj_key, algor_type, result.ori_img, result.roi_img, /*id=*/id++, json_str, true, result.ori_img_is_in_gpu, result.roi_img_is_in_gpu); 
                flag_nHelmet = false;
                LOG_TRACE("PEDESTRIAN task_id {} obj_id {} algor_type {}", obj_key.video_id, obj_key.obj_id, int(algor_type));
            }
            if(algor_type==algorithm_type_t::CALL_PHONE_DET && flag_cPhone == true) {
                auto json_str = helpers::gen_json::gen_pedestrian_safety_json(obj_key.video_id, obj_key.obj_id, result.box, algor_type);
                save_snapshot_process(obj_key, algor_type, result.ori_img, result.roi_img, /*id=*/id++, json_str, true, result.ori_img_is_in_gpu, result.roi_img_is_in_gpu); 
                flag_cPhone = false;
                LOG_TRACE("PEDESTRIAN task_id {} obj_id {} algor_type {}", obj_key.video_id, obj_key.obj_id, int(algor_type));
            }
            if(algor_type==algorithm_type_t::SMOKING_DET && flag_smoking == true) {
                auto json_str = helpers::gen_json::gen_pedestrian_safety_json(obj_key.video_id, obj_key.obj_id, result.box, algor_type);
                save_snapshot_process(obj_key, algor_type, result.ori_img, result.roi_img, /*id=*/id++, json_str, true, result.ori_img_is_in_gpu, result.roi_img_is_in_gpu); 
                flag_smoking = false;
                LOG_TRACE("PEDESTRIAN task_id {} obj_id {} algor_type {}", obj_key.video_id, obj_key.obj_id, int(algor_type));
            }
            // -------------------------------------------------------------------------------------------------------
            if(0) {
            LOG_TRACE("PEDESTRIAN task_id {} obj_id {} algor_type {}", obj_key.video_id, obj_key.obj_id, int(algor_type));
            auto json_str =
                helpers::gen_json::gen_pedestrian_safety_json(obj_key.video_id, obj_key.obj_id, result.box, algor_type);
            save_snapshot_process(obj_key, algor_type, result.ori_img, result.roi_img, /*id=*/id++, json_str, true,
                                  result.ori_img_is_in_gpu, result.roi_img_is_in_gpu); }
          }
  
          if (result.roi_img.data_ != nullptr) {
            if (result.roi_img_is_in_gpu)
              CHECK(cudaFree((void *)(result.roi_img.data_)));
            result.roi_img.data_ = nullptr;
          }
  
          if (result.ori_img.data_ != nullptr) {
            if (result.ori_img_is_in_gpu)
              CHECK(cudaFree((void *)(result.ori_img.data_)));
            result.ori_img.data_ = nullptr;
          }
        }
      }
    }
    #endif
  

    auto task_other_param_ptr = m_task_param_manager->get_task_other_param(obj_key.video_id);

    /* 开启人脸抓拍分析算法模块，获取该目标的算法分析结果 返回结果+快照 最后释放资源 */
    if (task_other_param_ptr->find(algorithm_type_t::FACE_SNAPSHOT) != task_other_param_ptr->end() &&
        algor_type == algorithm_type_t::FACE_SNAPSHOT) {
      LOG_TRACE("--- face snapshot will save snapshot IAMGE, task: {} obj: {}.", obj_key.video_id, obj_key.obj_id);
      m_save_snapshot_reprocessing->reprocessing_process_face(obj_key);
    }
  
    return 0;
  }
  
  // 打架跌倒算法模块
  void CMultiSourceProcess::algorithm_fight_fall(vector<string>& vpt_interest_task_id, vector<sy_img>& vpt_interest_imgs, vector<onelevel_det_result>& vptResult){

  
  #ifdef WITH_SECOND_PROCESS

    /* a. find fight or fall classification taskId. */
    vector<string> interest_task_id;

    vector<onelevel_det_result> interest_vpt_result;