写入redis处理加上失效ttl

admin

2025-04-29 8ece1bcf75056d8006e2d56721d1a6498e7d273c

 PreProcessFn.cpp

@@ -9,7 +9,7 @@
#include <mysql/mysql.h> 
#include <future>
#include "iostream"

#include <hiredis.h>
#include <string>
#include <locale>
#include <codecvt>
@@ -255,7 +255,7 @@
   //cout << current_time << endl;
   // 如果录制时间超过 10 秒，则停止录制
   if (seconds >= _recordtime) {
      cout << "已录制 10 秒视频，停止录制" << endl;
      std::cout << "已录制 10 秒视频，停止录制" << endl;
      video_writer.release();
   }
   else
@@ -284,7 +284,7 @@
      if (mysql_query(mysql, query.c_str()))
      {
         string err_string = mysql_error(mysql);
         cout << err_string << endl;
         std::cout << err_string << endl;
      }      
   }
}
@@ -321,7 +321,7 @@
      if (mysql_query(mysql, query.c_str()))
      {
         string err_string = mysql_error(mysql);
         cout << err_string << endl;
         std::cout << err_string << endl;
      }
   }
}
@@ -330,6 +330,38 @@
size_t WriteCallback(void* contents, size_t size, size_t nmemb, void* userp) {
    ((std::string*)userp)->append((char*)contents, size * nmemb);
    return size * nmemb;
}

string join(const vector<string>& sequence, const string& separator)
{
   std::string result;
   for (size_t i = 0; i < sequence.size(); ++i)
      result += sequence[i] + ((i != sequence.size() - 1) ? separator : "");
   return result;
}

/// <summary>
/// 写入数据至redis模块
/// </summary>
void fnWriteToRedis(redisContext*& context, string ipccode)
{
   redisReply* reply;

   //检测redis是否连接，报警写入redis
   if (!context == NULL && !context->err) {

      // 记录报警的时候，同时写入redis数据库
      std::string key = "camera::run::" + ipccode;
      std::string value = "1";

      //写入redis
      reply = (redisReply*)redisCommand(context, "SET %s %s EX %d", key.c_str(), value.c_str(),3);
      if (reply == NULL) {
         printf("信息写入redis失败: %s\n", context->errstr);
         redisFree(context);
      }
      freeReplyObject(reply);
   }
}

/// @brief 视频流拉取处理
@@ -455,6 +487,44 @@
   rtspStream.release();   
}

/// <summary>
/// 连接并校验 Redis 服务器密码
/// </summary>
/// <param name="server"></param>
/// <param name="port"></param>
/// <param name="password"></param>
/// <param name="context"></param>
/// <returns></returns>
bool connectToRedis(const char* server, int port, const char* password, redisContext*& context) {
   // 连接 Redis 服务器
   context = redisConnect(server, port);
   if (context == nullptr || context->err) {
      std::cerr << "连接redis服务端失败！" << std::endl;
      return false;
   }

   //// 校验密码
   //redisReply* reply = (redisReply*)redisCommand(context, "AUTH %s", password);
   //if (reply == NULL) {
   //   std::cerr << "连接校验失败！" << std::endl;
   //   redisFree(context);
   //   return false;
   //}
   //else {
   //   // 检查回复以确定是否鉴权成功
   //   if (reply->type == REDIS_REPLY_ERROR && strcmp(reply->str, "OK") != 0) {
   //      std::cerr << "密码错误: " << reply->str << std::endl;
   //      freeReplyObject(reply);
   //      redisFree(context);
   //      return false;
   //   }
   //   freeReplyObject(reply);
   //}

   return true;
}

//json转换字符串
std::string jsontostr(Json::Value& json)
{
   string return_str;
@@ -502,6 +572,18 @@
      mysql_set_character_set(mysql, "utf8mb4");
      //mysql_query(mysql, "SET NAMES GB2312");//解决中文乱码问题       
   }   


   //初始化redis链接
   redisContext* context;	
   if (!connectToRedis(redispath.c_str(), stoi(redisport), redispass.c_str(), context))
   {
      cout << "redis连接初始化失败！" << endl;
   }
   else
   {
      cout << "redis连接初始化成功！" << endl;
   }

   std::string limitDM = "1";//堆煤限定阈值，超过阈值进行报警，配置文件赋值，每个摄像头都不一样

@@ -677,132 +759,125 @@
      std::string modelAnalysis ="";
      color_result = modelNode["color_result"].asString();//颜色反正值

      // std::string area = modelNode["threshold"].asString();//区域一
      // std::string area2 = modelNode.get("area2", "").asString();//区域二
      Json::Value rects = modelNode["point_rects"];////模型分析类型编码  
      for (auto k = 0; k < rects.size(); ++k) 
      {
         Json::Value pointNode = rects[k];
         lable_title=pointNode["title"].asString();
         std::string point_type=pointNode["type"].asString();
         Json::Value point_Arr= pointNode["points"];
         for (auto x = 0; x < point_Arr.size(); ++x) 
         {
            Json::Value pointNode = rects[k];
            lable_title=pointNode["title"].asString();
            std::string point_type=pointNode["type"].asString();
            Json::Value point_Arr= pointNode["points"];
            for (auto x = 0; x < point_Arr.size(); ++x) 
            int croodx = stoi(point_Arr[x]["x"].asString());
            int croody = stoi(point_Arr[x]["y"].asString());
            if(point_type=="detectarea")
            {
               int croodx = stoi(point_Arr[x]["x"].asString());
               int croody = stoi(point_Arr[x]["y"].asString());
               if(point_type=="detectarea")
               {
                  detectarea.push_back(cv::Point(croodx, croody));
               }
               if (point_type=="flowarea")
               {
                  flowarea.push_back(cv::Point(croodx, croody));
               }
               if (point_type=="flowrelate")
               {
                  flowrelate.push_back(cv::Point(croodx, croody));
               }
               if (point_type=="leftroller")
               {
                  leftarea.push_back(cv::Point(croodx, croody));
               }
               if (point_type=="rightroller")
               {
                  rightarea.push_back(cv::Point(croodx, croody));
               }
               if (point_type=="beltarea")
               {
                  beltarea.push_back(cv::Point(croodx, croody));
               }
               if (point_type=="workarea")
               {
                  workarea.push_back(cv::Point(croodx, croody));
               }
						
               detectarea.push_back(cv::Point(croodx, croody));
            }
            if (point_type=="flowarea")
            {
               flowarea.push_back(cv::Point(croodx, croody));
            }
            if (point_type=="flowrelate")
            {
               flowrelate.push_back(cv::Point(croodx, croody));
            }
            if (point_type=="leftroller")
            {
               leftarea.push_back(cv::Point(croodx, croody));
            }
            if (point_type=="rightroller")
            {
               rightarea.push_back(cv::Point(croodx, croody));
            }
            if (point_type=="beltarea")
            {
               beltarea.push_back(cv::Point(croodx, croody));
            }   
         }
         }	
      }

      int* arr = new int[detectarea.size() * 2];
      //加载模型处理------
      if(modelCode=="1") //皮带状态
      {
            if (!rBelt.initConfig(modelPath.c_str(),arr))
            {
               printf("皮带运行模型初始化失败。\n");
            }
            else
            {
               printf("皮带运行模型初始化成功。\n");
               rBelt.isLoad = true;
            }
            areaBelt =detectarea;
            videoPathBelt = coalCode + ipccode + modelType + modelAnalysis;
            wfBelt.setCoalCode(coalCode);
            wfBelt.setCameraCode(ipccode);
            wfBelt.setCameraName(ipcname);
            wfBelt.setRtspUrl(ipcrtsppath);
            wfBelt.setAnalyse(modelType);
            //wfBelt.setAnalyseResult(modelAnalysis);
         if (!rBelt.initConfig(modelPath.c_str(),arr))
         {
            printf("皮带运行模型初始化失败。\n");
         }
         else
         {
            printf("皮带运行模型初始化成功。\n");
            rBelt.isLoad = true;
         }
         areaBelt =detectarea;
         videoPathBelt = coalCode + ipccode + modelType + modelAnalysis;
         wfBelt.setCoalCode(coalCode);
         wfBelt.setCameraCode(ipccode);
         wfBelt.setCameraName(ipcname);
         wfBelt.setRtspUrl(ipcrtsppath);
         wfBelt.setAnalyse(modelType);
         wfBelt.setAnalyseResult(modelAnalysis);
      }
      if(modelCode=="2") //提升机运行检测模型
      {
         if (!rTSJ.initConfig(modelPath.c_str(), arr))
            {
               printf("提升机模型初始化失败。\n");
            }
            else
            {
               printf("提升机模型初始化成功。\n");
               rTSJ.isLoad = true;
            }
            areaTSJ = detectarea;
            videoPathTSJ = coalCode + ipccode + modelType;
            wfTSJ.setCoalCode(coalCode);
            wfTSJ.setCameraCode(ipccode);
            wfTSJ.setCameraName(ipcname);
            wfTSJ.setRtspUrl(ipcrtsppath);
            wfTSJ.setAnalyse(modelType);
            wfTSJ.setAnalyseResult(modelAnalysis);
            break;
         {
            printf("提升机模型初始化失败。\n");
         }
         else
         {
            printf("提升机模型初始化成功。\n");
            rTSJ.isLoad = true;
         }
         areaTSJ = detectarea;
         videoPathTSJ = coalCode + ipccode + modelType;
         wfTSJ.setCoalCode(coalCode);
         wfTSJ.setCameraCode(ipccode);
         wfTSJ.setCameraName(ipcname);
         wfTSJ.setRtspUrl(ipcrtsppath);
         wfTSJ.setAnalyse(modelType);
         wfTSJ.setAnalyseResult(modelAnalysis);
         break;
      }
      if(modelCode=="3") //人员区域闯入模型
      {
         if (!hatJC.initConfig(modelPath.c_str(), 0.5, 0.5))
            {
               printf("目标检测模型初始化失败\n");
            }
            else
            {
               printf("目标检测模型初始化成功。\n");
               hatJC.isLoad = true;
            }
            areaPerson = detectarea;//危险区域
            workareaPerson = workarea;//工作区域
            videoPathPerson = coalCode + ipccode + modelType + modelAnalysis;
            wfPerson.setCoalCode(coalCode);
            wfPerson.setCameraCode(ipccode);
            wfPerson.setCameraName(ipcname);
            wfPerson.setRtspUrl(ipcrtsppath);
            wfPerson.setAnalyse(modelType);
            wfPerson.setAnalyseResult(modelAnalysis);
         {
            printf("目标检测模型初始化失败\n");
         }
         else
         {
            printf("目标检测模型初始化成功。\n");
            hatJC.isLoad = true;
         }
         areaPerson = detectarea;//危险区域
         workareaPerson = workarea;//工作区域
         videoPathPerson = coalCode + ipccode + modelType + modelAnalysis;
         wfPerson.setCoalCode(coalCode);
         wfPerson.setCameraCode(ipccode);
         wfPerson.setCameraName(ipcname);
         wfPerson.setRtspUrl(ipcrtsppath);
         wfPerson.setAnalyse(modelType);
         wfPerson.setAnalyseResult(modelAnalysis);
      }
      if(modelCode=="4") //摄像头遮挡算法
      {
         if (!cover.initConfig()) {
               printf("摄像头遮挡算法初始化失败\n");
            }
            else
            {
               std::cout << "摄像头遮挡算法初始化成功" << std::endl;
               cover.isLoad = true;
            }
            videoPathCameraCover = coalCode + ipccode + modelType + modelAnalysis;
            wfCameraCover.setCoalCode(coalCode);
            wfCameraCover.setCameraCode(ipccode);
            wfCameraCover.setCameraName(ipcname);
            wfCameraCover.setRtspUrl(ipcrtsppath);
            wfCameraCover.setAnalyse(modelType);
            wfCameraCover.setAnalyseResult(modelAnalysis);
            printf("摄像头遮挡算法初始化失败\n");
         }
         else
         {
            std::cout << "摄像头遮挡算法初始化成功" << std::endl;
            cover.isLoad = true;
         }
         videoPathCameraCover = coalCode + ipccode + modelType + modelAnalysis;
         wfCameraCover.setCoalCode(coalCode);
         wfCameraCover.setCameraCode(ipccode);
         wfCameraCover.setCameraName(ipcname);
         wfCameraCover.setRtspUrl(ipcrtsppath);
         wfCameraCover.setAnalyse(modelType);
         wfCameraCover.setAnalyseResult(modelAnalysis);
      }
      if(modelCode=="5") //摄像头移动算法
      {
@@ -825,24 +900,25 @@
      }
      if(modelCode=="6") //堆煤检测模型
      {
         if (!dm.initConfig(modelPath.c_str(), arr)) {
               printf("堆煤模型初始化失败\n");
            }
            else
            {
               std::cout << "堆煤模型初始化成功" << std::endl;
               dm.isLoad = true;
            }
            videoPathDM = coalCode + ipccode + modelType + modelAnalysis;
            areaDM = detectarea;
            limitDM = modelLimit;//堆煤限定煤量占比
         if (!dm.initConfig(modelPath.c_str(), arr)) 
         {
            printf("堆煤模型初始化失败\n");
         }
         else
         {
            std::cout << "堆煤模型初始化成功" << std::endl;
            dm.isLoad = true;
         }
         videoPathDM = coalCode + ipccode + modelType + modelAnalysis;
         areaDM = detectarea;
         limitDM = modelLimit;//堆煤限定煤量占比

            wfDM.setCoalCode(coalCode);
            wfDM.setCameraCode(ipccode);
            wfDM.setCameraName(ipcname);
            wfDM.setRtspUrl(ipcrtsppath);
            wfDM.setAnalyse(modelType);
            wfDM.setAnalyseResult(modelAnalysis);
         wfDM.setCoalCode(coalCode);
         wfDM.setCameraCode(ipccode);
         wfDM.setCameraName(ipcname);
         wfDM.setRtspUrl(ipcrtsppath);
         wfDM.setAnalyse(modelType);
         wfDM.setAnalyseResult(modelAnalysis);
      }
      if(modelCode=="7") //皮带跑偏和异物检测
      {
@@ -958,11 +1034,7 @@
         wfFire.setAnalyse(modelType);
         wfFire.setAnalyseResult(modelAnalysis);
      }
      if (modelCode == "12")//人员跌倒
      {

      }
   }	
   }

   std::queue<int> quTSJ;//记录提升机运动的判断结果
   std::queue<int> quBelt;//记录皮带运动判断结果
@@ -1710,7 +1782,9 @@
            currentFrame = 0;
         }

         currentFrame++;			
         currentFrame++;		
			
         fnWriteToRedis(context,ipccode);
      }
      catch (const std::exception& ex)
      {         
@@ -1867,6 +1941,37 @@
   //定义接受帧
   Mat frame;

   vector<string> arguments = {
      "ffmpeg "
      "-hwaccel","cuvid",
      "-hwaccel_output_format","cuda",
      "-y", "-an",			
      "-f", "rawvideo",
      "-vcodec", "rawvideo",
      "-pix_fmt", "bgr24",
      "-s",  "640x480",
      "-r", "15",			
      "-i", "-",	
      "-pix_fmt", "yuv420p",
      "-f", "flv",
      "-max_delay", "1000",
      "-flvflags", "no_duration_filesize",
      "-c:v","h264_nvenc",
      "-b:v", "3M",
      "-g:v", "15",
      "-bf", "0",
      "-bufsize", "50000000",
      "-rtbufsize", "50000000",	
      "rtmp://192.168.1.8:1935/live/camera1" };

   string ffmpeg_command = join(arguments, " ");

   // 打开FFmpeg进程
   FILE* pipe = popen(ffmpeg_command.c_str(), "w");
   if (!pipe) {
      std::cerr << "无法启动FFmpeg" << std::endl;
   }
		
   while (!asyncStop) {
      //std::cout << ipccode + "当前线程Rabbitmq数据数量2222:" << queJC2.size() << std::endl;      
      try
@@ -1885,20 +1990,9 @@
         queJC2.pop();

         if (frame.empty())//帧为空，则舍弃
            continue;
            continue;	

         // FFmpeg推流命令
         std::string ffmpeg_command = 
            "ffmpeg -y -f rawvideo -pixel_format bgr24 -video_size 640x480 "
              "-framerate 30 -i - -c:v libx264 -pix_fmt yuv420p -f rtsp "
                "rtsp://192.168.1.188:8554/live/stream";

         // 打开FFmpeg进程
         FILE* pipe = popen(ffmpeg_command.c_str(), "w");
         if (!pipe) {
            std::cerr << "无法启动FFmpeg" << std::endl;
         }

			
         // 将帧写入到FFmpeg管道中
         fwrite(frame.data, 1, frame.total() * frame.elemSize(), pipe);

@@ -1926,11 +2020,42 @@
/// @brief 推流到流媒体服务器
/// @param queJC2 处理完的视频流帧队列
/// @param ipccode 摄像机编号
void PreProcessFn::fnPushVideoToUrl(queue<Mat>& queJC2,string toRtsp, string fps, string ipccode)
void PreProcessFn::fnPushVideoToUrl(queue<Mat>& queJC2,string toRtsp, string fps, string video_size,string ipccode)
{   
   //定义接受帧
   Mat frame;

   vector<string> arguments = {
      "ffmpeg "
      "-hwaccel","cuvid",
      "-hwaccel_output_format","cuda",
      "-y", "-an",			
      "-f", "rawvideo",
      "-vcodec", "rawvideo",
      "-pix_fmt", "bgr24",
      "-s",  video_size,
      "-r", fps,			
      "-i", "-",	
      "-pix_fmt", "yuv420p",
      "-f", "flv",
      "-max_delay", "1000",
      "-flvflags", "no_duration_filesize",
      "-c:v","h264_nvenc",
      "-b:v", "3M",
      "-g:v", "15",
      "-bf", "0",
      "-bufsize", "50000000",
      "-rtbufsize", "50000000",	
      toRtsp };

   string ffmpeg_command = join(arguments, " ");

   // 打开FFmpeg进程
   FILE* pipe = popen(ffmpeg_command.c_str(), "w");
   if (!pipe) {
      std::cerr << "无法启动FFmpeg" << std::endl;
   }
		
   while (!asyncStop) {
      //std::cout << ipccode + "当前线程Rabbitmq数据数量2222:" << queJC2.size() << std::endl;      
      try
@@ -1949,20 +2074,8 @@
         queJC2.pop();

         if (frame.empty())//帧为空，则舍弃
            continue;

         // FFmpeg推流命令
         std::string ffmpeg_command = 
            "ffmpeg -y -f rawvideo -pixel_format bgr24 -video_size 640x480 "
              "-framerate 30 -i - -c:v libx264 -pix_fmt yuv420p -f rtsp "
                "rtsp://192.168.1.8:8554/live/stream";

         // 打开FFmpeg进程
         FILE* pipe = popen(ffmpeg_command.c_str(), "w");
         if (!pipe) {
            std::cerr << "无法启动FFmpeg" << std::endl;
         }

            continue;	
			
         // 将帧写入到FFmpeg管道中
         fwrite(frame.data, 1, frame.total() * frame.elemSize(), pipe);