Hu Chunming / src

  /*
   * @Description: 主流程类 + 任务管理相关操作
   */
  
  //#pragma once
  #ifndef __MutliSourceVideoProcess_H__
  #define __MutliSourceVideoProcess_H__
  
  #include <iostream>
  #include <stdlib.h>
  
  #include "../FFNvDecoder/FFNvDecoderManager.h"
  #include "../FFNvDecoder/cuda_kernels.h"
  
  
  #include "VPTProcess.h"
  #include "common.h"
  #include "nvml.h"
  #include "opencv2/highgui/highgui.hpp"
  #include "time.h"
  #include <boost/thread/thread.hpp>
  #include <queue>
  #include <set>

  #include <list>

  
  #include "mvpt_process_assist.h"
  #include <atomic>
  #include <boost/thread/thread_pool.hpp>
  #include <mutex>
  #include <thread>
  
  #include <pthread.h>
  
  #include "../reprocessing_module/save_snapshot_reprocessing.h"
  #include "../reprocessing_module/snapshot_reprocessing.h"
  #include "task_param_manager.h"
  
  #ifdef POST_USE_RABBITMQ
  #include "../reprocessing_module/mq_manager.hpp"
  #endif
  
  #include "../ai_engine_module/ai_engine_module.h"
  #ifdef WITH_SECOND_PROCESS
  #include "../ai_engine_module/human_gather_statistics.h"
  #include "../ai_engine_module/pedestrian_safety_det.hpp"
  #include "../ai_engine_module/pedestrian_vehicle_retrograde.hpp"
  #include "../ai_engine_module/pedestrian_vehicle_trespass.hpp"
  #endif
  
  #include "ErrorInfo.h"
  #include <condition_variable>
  
  #ifdef WITH_FACE_DET_SS
  #include "face_det_ai_engine.h"
  #endif
  
  #include "GpuRgbMemory.hpp"
  
  #ifdef _MSC_VER
  #ifdef _DEBUG
  #pragma comment(lib, "opencv_world310d.lib")
  #else
  #pragma comment(lib, "opencv_world310.lib")
  #endif
  #endif
  using namespace cv;
  // using namespace std;
  using std::map;
  using std::set;
  using std::vector;

  using std::list;

  
  #ifndef _MSC_VER
  #ifndef TRUE
  #define TRUE 1
  #endif
  #ifndef FALSE
  #define FALSE 0
  #endif
  #define Sleep(a) usleep((a)*1000)
  // typedef    int    BOOL;
  #define BOOL bool
  typedef unsigned int DWORD;
  typedef void *LPVOID;
  #endif
  
  #ifdef _DEBUG
  #define DEBUG_MSG(msg, ...)                                                                                            \
    {                                                                                                                    \
      printf("%s %s [%d]: ", __FILE__, __FUNCTION__, __LINE__);                                                          \
      printf(msg, ##__VA_ARGS__);                                                                                        \
      printf("\n");                                                                                                      \
    }
  #else
  #define DEBUG_MSG(msg, ...)
  #endif
  
  #define MAXLENGTH 416
  #define MINLENGTH 224
  
  //#define PROCESSHEIGHT 224
  //#define PROCESSWIDTH 416
  //#define DATASIZE PROCESSWIDTH * PROCESSHEIGHT * 3
  #define THREAD_COUNT 30
  #define SNAPSHOTFRAME 15
  #define LOSTMAXFRAMECCOUNT 4
  
  #define SCALE_OUT 10 // 判断目标框初始位置时，在最小距离的基础上适当外扩
  
  enum TaskState {
    PLAY,
    PAUSE,
    FINISH,
    DECODEERROR // 解码线程可能报错，报错之后直接结束掉该路解码
  };
  

  struct decode_cbk_userdata{
    task_param _cur_task_param;
    void* opaque;

    void* opaque1;

  };
  
  class CMultiSourceProcess {
  public:
    CMultiSourceProcess();
    ~CMultiSourceProcess();
  
    int InitAlgorthim(tsl_aiplatform_param vptParam);

  
  #ifdef POST_USE_RABBITMQ
    int AddMqConn(mq_type_t mq_type, rabbitmq_conn_params_t mq_conn_param);
    int GetTaskStatus(const string taskID);
  #endif
  
    /* task api */

    bool AddTask(task_param _cur_task_param);
    bool PauseTask(const string taskID);
    bool RestartTask(const string taskID);
    bool FinishTask(const string taskID);
    void CloseAllTask();

  
  public:

    // 生成线程的内部接口，外部勿调用

    void algorthim_process_thread();  // 算法处理线程

    void post_decode_thread(decode_cbk_userdata* userPtr, AVFrame * gpuFrame);  // 解码数据处理函数，在解码线程中执行

    void decode_finished_thread(task_param _cur_task_param);  // 解码线程结束后处理函数，在解码线程中执行
  
  #ifdef AUTHORIZATION
    void check_thread();  // 授权校验线程
  #endif

  
  private:

    void startProcessByGpuid();

    bool task_has_vpt_algor(const std::string &task_id);
    bool task_has_face_algor(const std::string &task_id);
    void cuda_free_wrap(sy_img &img);
  

    bool finish_task(const string taskID, const bool delete_snapshot);
  

    // VPT
    void algorthim_vpt(vector<string>& task_list, sy_img *batch_img);
    // 行人安全分析算法
    void algorthim_pedestrian_safety(vector<string>& task_list, vector<sy_img>& vpt_interest_imgs, vector<onelevel_det_result>& vptResult);
    // 逆行&非法闯入算法模块
    void algorthim_retrograde_trespass(vector<string>& vpt_interest_task_id, vector<sy_img>& vpt_interest_imgs, vector<onelevel_det_result>& vptResult
                                                          ,vector<vector<int>>& deleteObjectID);
    // for snapshot algorithm. 轨迹结束目标 做最后的结果返回（当前返回算法结果+快照保存路径）
    void algorithm_snapshot(vector<string>& vpt_interest_task_id, vector<vector<int>> deleteObjectID);
    // 打架跌倒算法模块
    void algorithm_fight_fall(vector<string>& vpt_interest_task_id, vector<sy_img>& vpt_interest_imgs, vector<onelevel_det_result>& vptResult);
    // 外卖员分析模块
    void algorithm_takeaway_member_cls(vector<string>& vpt_interest_task_id, vector<sy_img>& vpt_interest_imgs, vector<onelevel_det_result>& vptResult);
  
  
     // 人脸检测抓拍算法模块
    void algorthim_face_detect(vector<string>& task_list, sy_img *batch_img);
  
  
  // 轨迹结束帧需要做的算法模块
    int endframe_obj_process(const OBJ_KEY &obj_key, algorithm_type_t algor_type);
    /* 实现快照保存功能（还未真正保存 将显存图片cp到内存
   * 直接保存本地或者存入缓存队列异步保存，保存方式看需求，报警类需要同步保存报警，分析类可异步保存后返回）*/

    bool save_snapshot_process(const OBJ_KEY &obj_key, const algorithm_type_t &algorithm_type, const sy_img &ori_img,
                               const sy_img &roi_img, const long long id, const std::string &json_str,
                               bool enable_async = true, const bool ori_img_is_in_gpu = true,
                               const bool roi_img_is_in_gpu = true);

    int everyframe_process(vector<string> &task_in_play_id, sy_img *images, vector<onelevel_det_result> &ol_det_result);

  private:

    pthread_t m_gpuProcessthread;

    list<GpuRgbMemory*> m_RgbDataList;
    std::mutex m_QueueMtx; 
    bool m_bfinish;

    bool m_bProcessEnd;

    pthread_t m_authority_check_thread;

  
    void *VPT_Handle_{nullptr};
  
    int section_batch_size_;
    int licence_status_;

    string gpu_id_;

    snapshot_reprocessing *m_snapshot_reprocessing{nullptr};
    task_param_manager *m_task_param_manager{nullptr};
  
  #ifdef WITH_SECOND_PROCESS
    //! TODO: use model manager to replace.
    ai_engine_module::fight_fall_cls::FightfallCls fight_fall_cls_;
    ai_engine_module::takeaway_member_classification::TakeawayMemberCls takeaway_member_;
    ai_engine_module::human_gather_statistics::human_gather_statistics m_human_gather_statistics;
    ai_engine_module::pedestrian_safety_det::PedestrianSafetyDetector pedestrian_safety_detector_;
    ai_engine_module::pedestrian_vehicle_retrograde::PedestrianVehicleRetrograde pedestrian_vehicle_retrograde_;
    ai_engine_module::pedestrian_vehicle_trespass::PedestrianVehicleTrespass pedestrian_vehicle_trespass_;
  #endif
  
  #ifdef POST_USE_RABBITMQ
    mq::Manager *mq_manager_{nullptr};
  #endif
  
  #ifdef WITH_FACE_DET_SS
    face_det_ai_engine m_face_det_ai_engine; // 人脸检测
  #endif
  

    base_reprocessing_unit *m_save_snapshot_reprocessing{nullptr};
  

  };
  

  #if 0
  
  struct CUVID_USERDATA {
  	int		id;
  	void* opaque;
  };
  
  struct CUVID_DATA {
  	float* pData;
  	int		nWidth;
  	int		nHeight;
  	int		nDatasize;
  };
  
  
  struct SNAPSHOT_PROCESS_UNIT
  {
  	vector<DxGPUFrame> imgBig;
  	vector<DxGPUFrame> imgSmall;
  	vector<onelevel_det_result> imgVPTResult;
  };
  
  
  struct VideoHeightWidth
  {
  	double height;
  	double width;
  };
  
  struct VideoObjectSS
  {
  	unsigned char* obj_person;
  	unsigned char* obj_bike;
  	unsigned char* obj_car;
  
  	VideoObjectSS() : obj_person(NULL), obj_bike(NULL), obj_car(NULL) {}
  
  };
  template <typename T> class MyAtomic;
  template<typename T>
  bool operator == (MyAtomic<T>& d1, T& d2);
  
  template<typename T>
  bool operator == (MyAtomic<T>& d1, MyAtomic<T>& d2);
  template <typename T>
  class MyAtomic
  {
  public:
  	MyAtomic() {};
  	MyAtomic(const T& d) { data.store(d); };
  	MyAtomic(const MyAtomic& d) { data.store(d.data.load()); };
  	MyAtomic& operator =(T d) { data.store(d); return *this; };
  	MyAtomic& operator =(MyAtomic& d) { data.store(d.data.load()); return *this; };
  	MyAtomic& operator +=(T d) { data.fetch_add(d); return *this; };
  	MyAtomic& operator +=(MyAtomic& d) { data.fetch_add(d); return *this; };
  	operator int() { return data.load(); }
  	friend bool operator ==<T> (MyAtomic<T>& d1, T& d2);
  	friend bool operator ==<T> (MyAtomic<T>& d1, MyAtomic<T>& d2);
  private:
  	std::atomic<T> data;
  };
  template<typename T>
  bool operator == (MyAtomic<T>& d1, T& d2)
  {
  	if (d1.data.load() == d2)
  		return true;
  	else
  		return false;
  }
  
  template<typename T>
  bool operator == (MyAtomic<T>& d1, MyAtomic<T>& d2)
  {
  	if (d1.data.load() == d2.load())
  		return true;
  	else
  		return false;
  }
  
  struct Task {
  	int taskID;
  	const char* taskFileSource;
  	TaskState taskState;
  	DxDecoderWrap* taskTcuvid;
  	DxGPUFrame task_algorithm_data;      //针对新框架不做resize的处理，先暂时用backup的原图大小的图片送进算法参与计算
  	float* taskDataToRT;
  	bool taskHasBackup;
  	//VideoHeightWidth taskHeightWidthRatio;
  	VideoHeightWidth taskHeightWidth;
  	MyAtomic<int> taskFrameCount;
  	int taskTotalFrameCount;
  	SNAPSHOT_CALLBACK taskObjCallbackFunc;
  	REALTIME_CALLBACK taskRealTimeCallbackFunc;
  	cv::Mat frameImage;
  	char* folderNameLittle;
  	char* folderName;
  	sy_rect task_min_boxsize[DETECTTYPE];
  };
  #endif
  
  #endif