Hu Chunming / mvpt_dq_ascend_arm

#include "mvpt.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/str_helper.hpp"

#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 AUTHORIZATION
#define productSN "51C4B28135604F649671727185949A91" //linux 通途抓拍引擎产品序列号

using namespace std;


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(){
}

CMultiSourceProcess::~CMultiSourceProcess(){
  jpegUtil.jpeg_release();
}

int CMultiSourceProcess::InitAlgorthim(tsl_aiplatform_param vptParam){

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

#ifndef AUTHORIZATION
  if (!CheckTime()) { //时间限制
    LOG_ERROR("sy_licence_check failed.");
    return AUTHOR_ERROR;
  }
#else
  if (!CheckLabel(vptParam.gpuid)) { //机器授权
    LOG_ERROR("CheckLabel failed.");
    return AUTHOR_ERROR;
  }
#endif

  SourceSingleton::getInstance();

  skip_frame_ = 5; 
  m_batch_size = 20;

  m_devId = vptParam.gpuid;

  m_result_cbk = vptParam.result_cbk;

  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;

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

  ret = m_feature_analysis.init(m_devId, models_dir);
  if (ret < 0){
    return ret;
  }

  jpegUtil.jpeg_init(m_devId);

  m_snapshot_reprocessing = new snapshot_reprocessing(m_devId);

  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);

  LOG_INFO("InitAlgorthim succeed !");

  return 0;
}

/* 获取任务的状态 MQ返回 */
int CMultiSourceProcess::GetTaskStatus(const string taskID) {

  std::vector<std::string> taskids;
  std::vector<int> statues;
  int sta = 0;
  if(m_task_manager.IsPausing(taskID)){
    taskids.emplace_back(taskID);
    statues.emplace_back(2);
    sta = 2;
  } else if(m_task_manager.IsRunning(taskID)){
    taskids.emplace_back(taskID);
    statues.emplace_back(1);
    sta = 1;
  } else if(m_task_manager.IsFinished(taskID)){
    taskids.emplace_back(taskID);
    statues.emplace_back(0);
    sta = 0;
  }

  return sta;
}

bool CMultiSourceProcess::AddTask(task_param _cur_task_param)
{
	string task_id = _cur_task_param.task_id;

  decode_cbk_userdata* userPtr = new decode_cbk_userdata;
  userPtr->task_id = task_id;
  userPtr->opaque = this;

	FFDecConfig cfg;
	cfg.dec_name = task_id;
	cfg.uri = _cur_task_param.ipc_url;
	cfg.postDecArg = userPtr;
	cfg.post_decoded_cbk = post_decod_cbk;
	cfg.finishedDecArg = userPtr;
	cfg.decode_finished_cbk = decode_finished_cbk;
	cfg.force_tcp = true;  // rtsp用tcp
	cfg.gpuid = to_string(m_devId);
	cfg.skip_frame = _cur_task_param.skip_frame;

	if(!m_task_manager.AddTask(_cur_task_param, cfg)) {
		return false;
	}

	int input_image_width = 0;
  int input_image_height = 0;
  m_task_manager.GetResolution(cfg.dec_name, input_image_width, input_image_height);
  LOG_INFO("task_id: {} width: {} height:{}", task_id, input_image_width, input_image_height);

	// m_task_manager.SetDecKeyframe(cfg.dec_name, true); // 只对关键帧解码

  // 所有参数都准备好之后再启动解码
  bool bStart = m_task_manager.StartTask(task_id);
  if (!bStart){
    LOG_INFO("started task {} failed!", task_id);
    m_task_manager.FinishTask(task_id);
    return false;
  }

  // 人车物跟踪
	vpt_process.addTaskTracker(task_id);

	m_FinishedTaskMtx.lock();
  m_FinishedTaskMap[task_id] = false;
  m_FinishedTaskMtx.unlock();

  LOG_INFO("started task {} successed!", cfg.dec_name);

  return true;
}

int CMultiSourceProcess::CountRunningTask() {
  return m_task_manager.GetRunningTaskCount();
}

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){
  return m_task_manager.PauseTask(taskID);
}

bool CMultiSourceProcess::RestartTask(const string taskID){
    return m_task_manager.RestartTask(taskID);
}

bool CMultiSourceProcess::FinishTask(const string taskID){
    return m_task_manager.FinishTask(taskID);
}

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 = m_task_manager.SnapShot(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());

    bool bSaved = jpegUtil.jpeg_encode(vpcDesc, file_name);
    if(!bSaved){
      LOG_ERROR("jpeg_encode failed");
    }
    
    acldvppDestroyPicDesc(vpcDesc);

    jpegUtil.jpeg_release();

    delete devMem;
    devMem = nullptr;

    return SUCCESS;
  }

  return FAILED;
}

int CMultiSourceProcess::SnapShot(task_param _cur_task_param, char* result_folder) 
{
  LOG_INFO("begin SnapShot task: {}", _cur_task_param.task_id);

  /* 拷贝通用的算法参数 */
  if (result_folder) {
    CreateResultFolder(result_folder, "");
  } else {
    result_folder = "";
  }

  std::string cur_timestamp_ms = std::to_string(helpers::timer::get_timestamp<std::chrono::milliseconds>());
  std::string file_name = std::string(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){
    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));	//授权

    m_task_manager.Release();
    
    if(m_pAlgorthimThread){
        m_pAlgorthimThread->join();
        m_pAlgorthimThread = 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() {
  m_task_manager.Release();
}

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);
 
  m_snapshot_reprocessing->release_finished_locus_snapshot(taskID);

  return true;
}

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

    while (true){
#ifndef AUTHORIZATION
        if (!CheckTime()) { //时间限制
          LOG_FATAL("authority failed!");
          break;
        }
#else
        if (check_label == -1) { //机器授权
          LOG_FATAL("authority failed!");
          break;
        }
        if (!UpdateLabel()) {
          LOG_FATAL("authority failed!");
          break;
        }
#endif
        if(m_bfinish){
            break;
        }

        clear_finished_task(); 

        m_DataListMtx.lock();
        vector<DeviceMemory*> vec_gpuMem;
        
        while (!m_RgbDataList.empty()){
            DeviceMemory* gpuMem = m_RgbDataList.front();
            if(gpuMem->getMem() == nullptr) {
              LOG_WARN("mem is null");
            } 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;
        }

        // auto record_t1 = std::chrono::system_clock::now();
        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()));

        // 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();
        vector<DeviceMemory*>().swap(vec_gpuMem); // free.
    }
    
    LOG_INFO("algorthim_process_thread exit.");

    return 0;
}

int CMultiSourceProcess::algorthim_vpt(vector<DeviceMemory*> vec_gpuMem){

    vector<string> vpt_interest_task_id;
    vector<DeviceMemory*> vec_vptMem;
    for (int i = 0; i < vec_gpuMem.size(); i++) {
        DeviceMemory* mem = vec_gpuMem[i];

        vpt_interest_task_id.push_back(mem->getId());
        vec_vptMem.push_back(mem);
    }

    /* 待检测的图片不为空 开始检测 */
    if (!vec_vptMem.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.process(vec_vptMem, vptResult, deleteObjectID, unUsedResult); // do det & track.

      m_snapshot_reprocessing->screen_effective_snapshot(vptResult); 

      /* 快照优选（内部可实现不同的快照优选策略） */
      m_snapshot_reprocessing->update_bestsnapshot(vec_vptMem, vptResult);
      /*  for snapshot algorithm. 轨迹结束目标 做最后的结果返回（当前返回算法结果+快照保存路径）*/
      vehicle_snapshot(vpt_interest_task_id, deleteObjectID);

      vptResult.clear();
      unUsedResult.clear();
      deleteObjectID.clear();
    }

    vec_vptMem.clear();
    vpt_interest_task_id.clear();

    return 0;
}

// for snapshot algorithm. 轨迹结束目标 做最后的结果返回（当前返回算法结果+快照保存路径）
void CMultiSourceProcess::vehicle_snapshot(vector<string>& vpt_interest_task_id, vector<vector<int>> deleteObjectID) {

  for (int i = 0; i < deleteObjectID.size(); i++) // loop taskId.
  {
    string task_id = vpt_interest_task_id[i];
    for (int &j : deleteObjectID[i]) // loop algor type.
    {
      OBJ_KEY obj_key = {task_id, j};
      // 该路任务开启了抓拍功能 开始抓拍保存；若未开启抓拍，清空显存资源
      vehicle_locus_finished(obj_key);
    }
  }
}

void CMultiSourceProcess::vehicle_locus_finished(const OBJ_KEY obj_key) {

  map<OBJ_KEY, OBJ_VALUE> _total_snapshot_info = m_snapshot_reprocessing->get_total_snapshot_info();
  if(_total_snapshot_info.size() <= 0){
    LOG_WARN("_total_snapshot_info size: {}", _total_snapshot_info.size());
    return;
  }

  auto it = _total_snapshot_info.find(obj_key);
  if (it == _total_snapshot_info.end()) {
    // LOG_WARN("object not found: {}, {}", obj_key.video_id, obj_key.obj_id);
    return;
  }

  const OBJ_VALUE obj_value = it->second;

  TaskInfo* info = m_task_manager.GetTaskInfo(obj_key.video_id);
      
  string result_folder = info->result_folder + "/snapshot";
  string result_folder_little = info->result_folder_little + "/snapshot_lite";

  if (!result_folder.empty()) {
		CreateResultFolder(result_folder.c_str(), "");
	} else {
		LOG_ERROR("result_folder is empty!");
	}

  if (!result_folder_little.empty()) {
		CreateResultFolder(result_folder_little.c_str(), "");
	} else {
		LOG_ERROR("result_folder_little is empty!");
	}

  LOG_DEBUG("update task_id {} obj_id {}", obj_key.video_id, obj_key.obj_id);

  std::string cur_timestamp_ms =  std::to_string(helpers::timer::get_cur_time_ms());
  // 原图
  std::string ori_pic_path = result_folder + helpers::os::sep + obj_key.video_id + "_" + std::to_string(obj_key.obj_id) + "_" + cur_timestamp_ms + ".jpg";
  // 抠图
  string object_file_name = result_folder_little + helpers::os::sep + obj_key.video_id + "_" + std::to_string(obj_key.obj_id) + "_" + cur_timestamp_ms + ".jpg";

  ObjectInfo res_obj;
  res_obj.task_id = obj_key.video_id;              //该物体属于的任务ID号
  res_obj.task_frame_count = obj_value.frameCount;     //该物体当前出现的帧号
  res_obj.object_id = obj_key.obj_id;            //该物体的ID号
  res_obj.left = obj_value.obj_pos.left_;                 //该物体位置的左坐标
  res_obj.top = obj_value.obj_pos.top_;                  //该物体位置的上坐标
  res_obj.right = obj_value.obj_pos.left_ + obj_value.obj_pos.width_;                //该物体位置的右坐标
  res_obj.bottom = obj_value.obj_pos.top_ + obj_value.obj_pos.height_;               //该物体位置的下坐标
  res_obj.index = obj_value.index.index;                //该物体所属类别的编号
  res_obj.confidence = obj_value.confidence;        //该物体的置信度
  res_obj.ori_pic_path = ori_pic_path;
  res_obj.ori_img_info = obj_value.snapShot;
  res_obj.obj_pic_path = object_file_name;
  res_obj.obj_img_info = obj_value.snapShotLittle;
  m_feature_analysis.update_object_info(res_obj, obj_value.snapShotLittle);
 
  bool bSaved = save_obj_pic(res_obj);
  if (bSaved)
  {
    m_result_cbk(res_obj);
  }

  m_snapshot_reprocessing->release_finished_locus_snapshot(obj_key.video_id, obj_key.obj_id, true); 

}

bool CMultiSourceProcess::save_obj_pic(ObjectInfo& res_obj) 
{
  bool bSaved = false;

  do
  {
    // 原图
    if(!res_obj.ori_pic_path.empty()){
      bSaved = jpegUtil.jpeg_encode(res_obj.ori_img_info.pic_desc, res_obj.ori_pic_path);
      if(!bSaved){
        LOG_ERROR("jpeg_encode failed");
        break;
      }
    }
    // 抠图
    if(!res_obj.obj_pic_path.empty()){
      bSaved = jpegUtil.jpeg_encode(res_obj.obj_img_info.pic_desc, res_obj.obj_pic_path);
      if(!bSaved){
        LOG_ERROR("jpeg_encode failed");
        break;
      }
    }
  } while (0);
  
  return bSaved;
}

bool CMultiSourceProcess::CheckTime() {
    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 (nYear == 2025 && nMonth <= 12)
    {
        return true;
    }
    else
    {
        return false;
    }
}

bool CMultiSourceProcess::CheckLabel(int devId) {
  	const char *token_password = "mvpt_dq_ascend_arm_2025";
    std::string guid = std::string("mvpt_dq_ascend_arm")+std::to_string(devId);
    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 false;
    }
    //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 false;
    }
    std::cout << "sy_licence_check end." << std::endl;

    check_label = 0;//授权成功
    return true;
}

bool CMultiSourceProcess::UpdateLabel() {
    //获取系统时间，每个月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) {
        LOG_INFO("atlas_licence_check start.");
        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) {
          LOG_FATAL("sy_licence_check failed.");
          return false;
        }
        if(check_result.code!=0) {
          check_label = -1;
          LOG_FATAL("atlas_licence_check code:{}, msg:{}",check_result.code, check_result.msg);
          return false;
        }
        LOG_INFO("atlas_licence_check end.");
        check_label =1;
      }
    }
    else {
      check_label =0;
    }
    return true;
}