Hu Chunming / src

  #include "save_snapshot_reprocessing.h"
  #include <helpers/helpers.h>
  #include <thread>
  

  #include "opencv2/opencv.hpp"
  #include <opencv2/imgproc/types_c.h>
  #include <algorithm>
  

  const bool DRAW_ON_IMG = false;
  

  DWORD save_image_thread_process(LPVOID param);

  
  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}};
  
  // 初始化快照保存模块 开启图片保存线程
  save_snapshot_reprocessing::save_snapshot_reprocessing() {
  #ifdef POST_USE_RABBITMQ
    callback_ = nullptr;
  #endif
  

    compression_params.push_back(cv::IMWRITE_JPEG_QUALITY);                                                                             

    compression_params.push_back(30);
  
    _snapshot_reprocessing = snapshot_reprocessing::getInstance();
    _task_param_manager = task_param_manager::getInstance();
  

    bFinish = false;

    m_save_ss_thread = std::thread(save_image_thread_process, this);
    // m_save_ss_thread.detach();

  }
  

  save_snapshot_reprocessing::~save_snapshot_reprocessing(){

    // 结束线程
    bFinish = true;

    m_save_ss_thread.join();
  }
  
  // 释放资源
  void save_snapshot_reprocessing::save_snapshot_reprocessing_release() {

    std::unique_lock<std::mutex> l(waitforsave_image_queue_mutex);
  
    while (!waitforsave_image_queue.empty()) {
      SNAPSHOT_IMAGE_UNIT &cur_image = waitforsave_image_queue.front();
  
      if (cur_image.ori_image.data != nullptr) {
        delete[] cur_image.ori_image.data;
        cur_image.ori_image.data = nullptr;
      }
  
      if (cur_image.image.data != nullptr) {
        delete[] cur_image.image.data;
        cur_image.image.data = nullptr;
      }
      cur_image.image.release();
  
      waitforsave_image_queue.pop();
    }
  
    l.unlock();
  }
  
  #ifdef POST_USE_RABBITMQ
  
  // 设置MQ返回回调函数 方便内部调用MQ推送结果
  void save_snapshot_reprocessing::set_callback(callback_t cb) {
    callback_ = cb;
  }
  
  #endif // #ifdef POST_USE_RABBITMQ
  
  // 快照保存模块，把待保存图片cp回内存，同步或者异步保存；JSON结果字符串替换正确，MQ返回结果
  void save_snapshot_reprocessing::reprocessing_process_wo_locus(const OBJ_KEY &obj_key, const multi_image_t &m_image,
                                                                 bool enable_async
  
  #ifdef POST_USE_RABBITMQ
                                                                 ,
                                                                 const std::string &json_str
  #endif
  ) {
    auto func = [](sy_img &dst_img, const data_device_t &data_device, const sy_img &src_img) {
      auto size = src_img.c_ * src_img.h_ * src_img.w_;
      // dst_img.data_ = new unsigned char[size]{};
      {
        dst_img.c_ = src_img.c_;
        dst_img.h_ = src_img.h_;
        dst_img.w_ = src_img.w_;
      }
      if (data_device_t::GPU == data_device) {
        dst_img.data_ = new unsigned char[size]{};
        CHECK(cudaMemcpy(dst_img.data_, src_img.data_, size * sizeof(unsigned char), cudaMemcpyDeviceToHost));
      } else {
        dst_img.data_ = src_img.data_;
        // memcpy(dst_img.data_, src_img.data_, size * sizeof(unsigned char));
      }
    };
  
    sy_img ori_img, roi_img;
    {
      ori_img.data_ = nullptr;
      roi_img.data_ = nullptr;
    }
  
    if (m_image.ori_fpath != "") {
      func(ori_img, m_image.ori_data_device, m_image.ori_img);
    }
  
    if (m_image.roi_fpath != "") {
      func(roi_img, m_image.roi_data_device, m_image.roi_img);
    }
  
    SNAPSHOT_IMAGE_UNIT cur_image;
    {
      if (ori_img.data_ != nullptr)
        cur_image.ori_image = cv::Mat(ori_img.h_, ori_img.w_, ori_img.c_ == 3 ? CV_8UC3 : CV_8UC1, ori_img.data_);
  
      if (roi_img.data_ != nullptr)
        cur_image.image = cv::Mat(roi_img.h_, roi_img.w_, roi_img.c_ == 3 ? CV_8UC3 : CV_8UC1, roi_img.data_);
    }
  
    cur_image.ori_file_name = m_image.ori_fpath;
    cur_image.file_name = m_image.roi_fpath;
    cur_image.obj_key = obj_key;
  
  #ifdef POST_USE_RABBITMQ
    auto json_str_tmp =
        helpers::string::replace_all(json_str, helpers::gen_json::ORI_IMAGE_PATH_PLACEHOLDER, cur_image.ori_file_name);
    cur_image.json_str =
        helpers::string::replace_all(json_str_tmp, helpers::gen_json::ROI_IMAGE_PATH_PLACEHOLDER, cur_image.file_name);
  #endif
  
    LOG_DEBUG("task_id {} do push image path {}", obj_key.video_id, cur_image.ori_file_name);
  
    if (enable_async) {
      std::lock_guard<std::mutex> lock(waitforsave_image_queue_mutex);
      waitforsave_image_queue.push(cur_image);
      LOG_DEBUG("task_id {} waitforsave image path {}", obj_key.video_id, cur_image.ori_file_name);
    } else {
      if (!cur_image.ori_image.empty())
        if (!cv::imwrite(cur_image.ori_file_name, cur_image.ori_image))
          LOG_ERROR("save image to {} failed.", cur_image.ori_file_name.c_str());
  
      if (!cur_image.image.empty())
        if (!cv::imwrite(cur_image.file_name, cur_image.image))
          LOG_ERROR("save image to {} failed.", cur_image.file_name.c_str());
  
  #ifdef POST_USE_RABBITMQ
      if (callback_ != nullptr)
        callback_(cur_image.json_str.c_str());
  #endif
    }
  }
  
  // 快照保存模块，把待保存图片cp回内存，异步保存
  void save_snapshot_reprocessing::reprocessing_process(const OBJ_KEY &obj_key) {
    map<OBJ_KEY, OBJ_VALUE> *_total_snapshot_info = _snapshot_reprocessing->get_total_snapshot_info();
  
    if ((*_total_snapshot_info).find(obj_key) == (*_total_snapshot_info).end()) {
  #ifdef MTASK_DEBUG_
      LOG_DEBUG("task_id {} reprocessing_process obj_id {}", obj_key.video_id, obj_key.obj_id);
  #endif
      return;
    }
  
  #ifdef MTASK_DEBUG_
      LOG_DEBUG("[info] task_id {} reprocessing_process obj_id {}", obj_key.video_id, obj_key.obj_id);
  #endif
    const OBJ_VALUE &obj_value = (*_total_snapshot_info)[obj_key];
    auto task_other_param = _task_param_manager->get_task_other_params()[obj_key.video_id];
  #ifdef MTASK_DEBUG_
      LOG_DEBUG("[info] task_id {} reprocessing_process obj_id {}", obj_key.video_id, obj_key.obj_id);
  #endif
    auto algor_type = index_to_algo_type[obj_value.index.index];
  
    {
  #ifdef MTASK_DEBUG_
      LOG_DEBUG("[info] algor_type {} ", algor_type);
  #endif
      const auto algor_param = (algor_config_param_snapshot *)task_other_param[algor_type]->algor_param;
      if (obj_value.confidence <= algor_param->threshold) {
        LOG_DEBUG("Snapshot conf filter ({} vs {})", obj_value.confidence, algor_param->threshold);
        _snapshot_reprocessing->delete_finishtask_snapshot(obj_key.video_id, obj_key.obj_id); // modified by zsh 220708
        return;
      }
  #ifdef MTASK_DEBUG_
      LOG_DEBUG("[info] obj_value.snapShotLittle.height {} obj_value.snapShotLittle.width {} ", obj_value.snapShotLittle.height,obj_value.snapShotLittle.width);
  #endif
      const algor_basic_config_param_t *cur_param = task_other_param[algor_type]->basic_param;
      unsigned char *temp_ss_host = new unsigned char[3 * obj_value.snapShotLittle.height *
                                                      obj_value.snapShotLittle.width * sizeof(unsigned char)]{};
      CHECK(cudaMemcpy(temp_ss_host, obj_value.snapShotLittle.frame,
                       3 * obj_value.snapShotLittle.height * obj_value.snapShotLittle.width * sizeof(unsigned char),
                       cudaMemcpyDeviceToHost));  
      unsigned char *temp_img_host =
          new unsigned char[3 * obj_value.snapShot.height * obj_value.snapShot.width * sizeof(unsigned char)]{};
      CHECK(cudaMemcpy(temp_img_host, obj_value.snapShot.frame,
                       3 * obj_value.snapShot.height * obj_value.snapShot.width * sizeof(unsigned char),
                       cudaMemcpyDeviceToHost));
  #ifdef MTASK_DEBUG_
      LOG_DEBUG("[info] obj_value.snapShot.height {} obj_value.snapShot.width {} ", obj_value.snapShot.height,obj_value.snapShot.width);
  #endif
      _snapshot_reprocessing->delete_finishtask_snapshot(obj_key.video_id, obj_key.obj_id);
      SNAPSHOT_IMAGE_UNIT cur_image(obj_value.snapShotLittle.height, obj_value.snapShotLittle.width, CV_8UC3,
                                    temp_ss_host, obj_value.snapShot.height, obj_value.snapShot.width, CV_8UC3,
                                    temp_img_host, algorithm_type_t::UNKNOWN);
      // cur_image.ori_file_name = std::string(cur_param->result_folder) + helpers::os::sep + obj_key.video_id + "_" +
      //                           std::to_string(obj_key.obj_id) + ".jpg";
      // cur_image.file_name = std::string(cur_param->result_folder_little) + helpers::os::sep + obj_key.video_id + "_" +
      //                       std::to_string(obj_key.obj_id) + ".jpg";
      // modified by zsh 图片名添加时间戳---------------------------------------------------------------------------------
      std::string cur_timestamp_ms =  std::to_string(helpers::timer::get_timestamp<std::chrono::milliseconds>());
      cur_image.ori_file_name = std::string(cur_param->result_folder) + helpers::os::sep + obj_key.video_id + "_" +
                                std::to_string(obj_key.obj_id) + "_" + cur_timestamp_ms + ".jpg";
      cur_image.file_name = std::string(cur_param->result_folder_little) + helpers::os::sep + obj_key.video_id + "_" +
                            std::to_string(obj_key.obj_id) + "_" + cur_timestamp_ms + ".jpg";
      //-----------------------------------------------------------------------------------------------------------------
      cur_image.obj_key = obj_key;
      cur_image.obj_rect = obj_value.obj_pos;
  
  #ifdef POST_USE_RABBITMQ
      {
        video_object_snapshot new_obj_ss_info;
        new_obj_ss_info.analysisRes = nullptr;
        new_obj_ss_info.object_id = cur_image.obj_key.obj_id;
        new_obj_ss_info.obj_info.set_data(obj_value.index.index, obj_value.confidence, obj_value.obj_pos.left_,
                                          obj_value.obj_pos.top_, obj_value.obj_pos.left_ + obj_value.obj_pos.width_,
                                          obj_value.obj_pos.top_ + obj_value.obj_pos.height_);
        strcpy(new_obj_ss_info.task_id, cur_image.obj_key.video_id.c_str());
        strcpy(new_obj_ss_info.video_image_path, cur_image.ori_file_name.c_str());
        strcpy(new_obj_ss_info.snapshot_image_path, cur_image.file_name.c_str());
        cur_image.json_str = helpers::gen_json::gen_snapshot_json(algor_type, new_obj_ss_info);
      }
  #endif
      {
  #ifdef MTASK_DEBUG_
        LOG_DEBUG("task_id {} do push image path {} {}", obj_key.video_id, cur_image.ori_file_name, cur_image.file_name);
  #endif
  
        std::lock_guard<std::mutex> lock(waitforsave_image_queue_mutex);
        waitforsave_image_queue.push(cur_image);
      }
    }
  #ifdef MTASK_DEBUG_
      LOG_DEBUG("6666666666666666666\n");
      // save_snapshot_info_count[obj_key]++;
      // printf("save_snapshot_info_count[obj_key]:%d %s %d\n",save_snapshot_info_count[obj_key],obj_key.video_id.c_str(), obj_key.obj_id);
      // if(save_snapshot_info_count[obj_key] > 1) {
      //   printf("!!!!!!!!!!!!!!!!!!!!!!save_snapshot_info_count[obj_key]:%d %s %d\n",save_snapshot_info_count[obj_key],obj_key.video_id.c_str(), obj_key.obj_id);
      // }
  #endif
  }
  
  // 人脸快照后处理函数
  void save_snapshot_reprocessing::reprocessing_process_face(const OBJ_KEY &obj_key) {
    map<OBJ_KEY, OBJ_VALUE> *_total_face_snapshot_info = _snapshot_reprocessing->get_total_face_snapshot_info();
  
  
    LOG_TRACE("face reprocessing {}:{}.", obj_key.video_id, obj_key.obj_id);
  
    if ((*_total_face_snapshot_info).find(obj_key) == (*_total_face_snapshot_info).end()) {
      LOG_TRACE("face post {}:{} can not found.", obj_key.video_id, obj_key.obj_id);
      return;
    }
  
    OBJ_VALUE &obj_value = (*_total_face_snapshot_info)[obj_key];
    auto task_other_params = _task_param_manager->get_task_other_params();
  
    const algor_basic_config_param_t *cur_param =
        ((algor_init_config_param_t *)(task_other_params[obj_key.video_id][algorithm_type_t::FACE_SNAPSHOT]))
            ->basic_param;
  
    // 抠图
    unsigned char *temp_ss_host =
        new unsigned char[3 * obj_value.snapShotLittle.height * obj_value.snapShotLittle.width * sizeof(unsigned char)]{};
    CHECK(cudaMemcpy(temp_ss_host, obj_value.snapShotLittle.frame,
                     3 * obj_value.snapShotLittle.height * obj_value.snapShotLittle.width * sizeof(unsigned char),
                     cudaMemcpyDeviceToHost));
    {
      CHECK(cudaFree(obj_value.snapShotLittle.frame));
      obj_value.snapShotLittle.frame = nullptr;
    }
  
    // 原图
    unsigned char *temp_ss_host_ori =
        new unsigned char[3 * obj_value.snapShot.height * obj_value.snapShot.width * sizeof(unsigned char)]{};
    CHECK(cudaMemcpy(temp_ss_host_ori, obj_value.snapShot.frame,
                     3 * obj_value.snapShot.height * obj_value.snapShot.width * sizeof(unsigned char),
                     cudaMemcpyDeviceToHost));
    {
      CHECK(cudaFree(obj_value.snapShot.frame));
      obj_value.snapShot.frame = nullptr;
    }
  
    SNAPSHOT_IMAGE_UNIT cur_image(obj_value.snapShotLittle.height, obj_value.snapShotLittle.width, CV_8UC3, temp_ss_host,
                                  obj_value.snapShot.height, obj_value.snapShot.width, CV_8UC3, temp_ss_host_ori,
                                  algorithm_type_t::FACE_SNAPSHOT);
    // modified by zsh 图片名添加时间戳---------------------------------------------------------------------------------
    std::string cur_timestamp_ms =  std::to_string(helpers::timer::get_timestamp<std::chrono::milliseconds>());
    cur_image.ori_file_name = std::string(cur_param->result_folder) + helpers::os::sep + obj_key.video_id + "_" +
                              std::to_string(obj_key.obj_id) + "_" + cur_timestamp_ms + ".jpg";
    cur_image.file_name = std::string(cur_param->result_folder_little) + helpers::os::sep + obj_key.video_id + "_" +
                          std::to_string(obj_key.obj_id) + "_" + cur_timestamp_ms + ".jpg";
    //-----------------------------------------------------------------------------------------------------------------
    cur_image.obj_key = obj_key;
  
  #ifdef POST_USE_RABBITMQ
    cur_image.json_str = helpers::gen_json::gen_face_detection_json(
        cur_image.obj_key.video_id, cur_image.obj_key.obj_id, cur_image.file_name, cur_image.ori_file_name,
        obj_value.position, obj_value.confidence, obj_value.landmark_point, 25);
  #endif
  
    {
      std::lock_guard<std::mutex> lock(waitforsave_image_queue_mutex);
      waitforsave_image_queue.push(cur_image);
    }
  }
  

  DWORD save_image_thread_process(LPVOID param) {
    save_snapshot_reprocessing *pThreadParam = (save_snapshot_reprocessing *)param;

    if (pThreadParam != nullptr){
      pThreadParam->save_image_process();
    }
    return 0;
  }
  
  // 快照保存子线程 依次从缓存队列中取数据 然后保存图片 然后MQ返回结果
  void save_snapshot_reprocessing::save_image_process() {

  
    while (true) {

      if (bFinish){
        break;
      }
      

      std::unique_lock<std::mutex> l(waitforsave_image_queue_mutex);
      std::unique_lock<std::mutex> l_taskid(waitforchange_taskid_mutex);
      if (!waitforsave_image_queue.empty()) {
        SNAPSHOT_IMAGE_UNIT cur_image = waitforsave_image_queue.front();
        waitforsave_image_queue.pop();
        if (task_on_play.find(cur_image.obj_key.video_id) == task_on_play.end()) {

          LOG_DEBUG("opencv nosave img {}", cur_image.ori_file_name.c_str());
          if (cur_image.ori_image.data != nullptr) {
            delete[] cur_image.ori_image.data;
            cur_image.ori_image.data = nullptr;
          }
  
          if (cur_image.image.data != nullptr) {
            delete[] cur_image.image.data;
            cur_image.image.data = nullptr;
          }
          cur_image.image.release();
          l.unlock();
          l_taskid.unlock();
          continue;
        }
  
        l.unlock();
  
        if (!cur_image.image.empty()) {
          // if (!cv::imwrite(cur_image.file_name, cur_image.image, compression_params))
          if (!cv::imwrite(cur_image.file_name, cur_image.image))
            LOG_ERROR("opencv save to {} failed", cur_image.file_name.c_str());
          LOG_DEBUG("opencv save to {}", cur_image.file_name.c_str());
        }
        // if (!cur_image.image.empty()) 
        //   // if (!cv::imwrite(cur_image.file_name, cur_image.image, compression_params))
        //   if (!cv::imwrite(cur_image.file_name, cur_image.image))
        //     LOG_ERROR("opencv save to {} failed", cur_image.file_name.c_str());
        
  
        if (!cur_image.ori_image.empty()) {
          if (DRAW_ON_IMG)
            cv::rectangle(cur_image.ori_image,
                          cv::Rect(cur_image.obj_rect.left_, cur_image.obj_rect.top_, cur_image.obj_rect.width_,
                                   cur_image.obj_rect.height_),
                          cv::Scalar(0, 255, 0), 1, 1, 0);
  
          LOG_DEBUG("opencv save to {}", cur_image.ori_file_name.c_str());
          if (!cv::imwrite(cur_image.ori_file_name, cur_image.ori_image))
          // if (!cv::imwrite(cur_image.ori_file_name, cur_image.ori_image, compression_params))
            LOG_ERROR("opencv save to {} failed", cur_image.ori_file_name.c_str());
        }
  
  #ifdef POST_USE_RABBITMQ

        if (callback_ != nullptr) {

          // LOG_DEBUG("mq publish process 00000000000000000");

          callback_(cur_image.json_str.c_str());

          // LOG_DEBUG("mq publish process 11111111111111111");
        }
  
  #endif
  
        if (cur_image.ori_image.data != nullptr) {
          delete[] cur_image.ori_image.data;
          cur_image.ori_image.data = nullptr;
        }
  
        if (cur_image.image.data != nullptr) {
          delete[] cur_image.image.data;
          cur_image.image.data = nullptr;
        }
        cur_image.image.release();
        l_taskid.unlock();
      } else {
        l.unlock();
        l_taskid.unlock();
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
      }
    }

  }
  
  void save_snapshot_reprocessing::add_newtask(const string taskid) {
    std::unique_lock<std::mutex> l_taskid(waitforchange_taskid_mutex);
    task_on_play.insert(taskid);
  }
  
  void save_snapshot_reprocessing::delete_finishtask(const string taskid) {
    std::unique_lock<std::mutex> l_taskid(waitforchange_taskid_mutex);
    task_on_play.erase(taskid);
  }