增加监控设备采集数据kafka推送

monitor-manage-service/pom.xml

@@ -56,6 +56,12 @@
            <version>2.1.5-SNAPSHOT</version>
        </dependency>
        <dependency>
            <groupId>com.ai.bss</groupId>
            <artifactId>components-ai</artifactId>
            <version>2.1.5-SNAPSHOT</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/commons-fileupload/commons-fileupload -->
        <dependency>
            <groupId>commons-fileupload</groupId>

monitor-manage-service/src/main/java/com/ai/bss/monitorManage/constant/CommonConsts.java

@@ -1,100 +0,0 @@
//package com.ai.bss.monitorManage.constant;
//
///**
// * 公共常量类
// *
// * @author chencai
// */
//public class CommonConsts {
//    /**
//     * 接口调用返回的成功编码
//     */
//    public static final String RESULT_CODE_SUCCESSFUL = "0";
//
//    /**
//     * 接口调用返回的失败编码
//     */
//    public static final String RESULT_CODE_FAILED = "1";
//
//    /**
//     * 接口调用返回的成功描述信息
//     */
//    public static final String RESULT_MSG_SUCCESSFUL = "成功";
//
//    /**
//     * 接口调用返回的失败的描述信息
//     */
//    public static final String RESULT_MSG_FAILED = "失败";
//
//    /**
//     * CSV文件列分隔符
//     */
//    public static final String CSV_COLUMN_SEPARATOR = ",";
//    public static final String CSV_COLUMN_SEPARATOR2 = "|";
//
//
//    /**
//     * CSV文件列分隔符
//     */
//    public static final String CSV_RN = "\r\n";
//
//    /**
//     * 最小误差
//     */
//    public static final double EPSILON = 0.00000000000000022204;
//
//    /**
//     * 时间格式转化 yyyyMMdd_HHmmss
//     */
//    public static final String DATE_FORMAT_YYYYMMDD_HHMMSS = "yyyyMMdd_HHmmss";
//
//    /**
//     * 时间格式转化 yyyyMM
//     */
//    public static final String DATE_FORMAT_YYYYMM = "yyyyMM";
//
//    /**
//     * 时间格式转化 yyyyMMdd
//     */
//    public static final String DATE_FORMAT_YYYYMMDD = "yyyyMMdd";
//
//    /**
//     *
//     */
//    public static final String RESULT_MSG_CREATE_CSV_FAILED = "生成csv文件失败";
//
//    /**
//     * uspa组织类型-基础平台
//     */
//    public static final String USPA_ORG_TYPE_PLATFORM_ORG = "PLATFORM_ORG";
//
//    /**
//     * uspa组织类型-行业云
//     */
//    public static final String USPA_ORG_TYPE_INDUSTRY_ORG = "INDUSTRY_ORG";
//
//    /**
//     * uspa组织类型-客户
//     */
//    public static final String USPA_ORG_TYPE_CUSTOMER_ORG = "CUSTOMER_ORG";
//
//
//
//    public static final String CHARSET_UTF8 = "UTF-8";
//
//
//    public static final String HTTP_CONTENT_TYPE_JSON = "application/json; charset=utf-8";
//
//    private CommonConsts() {}
//
//
////
////    /** 
////     * callsTimesMap存放方法被调用的次数O 
////     */
////    public  static ThreadLocal<Map<String, Long >> callsTimesMap = new ThreadLocal<Map<String, Long>>();// new ThreadLocal<>();
////    /** 
////     * methodTotalTmieMap存放方法总耗时 
////     */
////    public  static ThreadLocal<Map<String, Long >> methodTotalTmieMap = new ThreadLocal<Map<String, Long>>();
//}

monitor-manage-service/src/main/java/com/ai/bss/monitorManage/constant/MonitorManageConsts.java

@@ -0,0 +1,27 @@
package com.ai.bss.monitorManage.constant;
/**
 * 公共常量类
 *
 * @author chencai
 */
public class MonitorManageConsts {
    public static String FILE_TYPE_VIDEO = "videos";
    public static String FILE_TYPE_IMG = "images";
    public static String IdenResultType = "";
//    aiIdenModel   AI匹配模型
    //人脸识别
    public static String AI_MODEL_IDENTIFY_FACE =  "IDENTIFY_FACE" ;
    //多人体识别
    public static String AI_MODEL_IDENTIFY_MULTIPLE_HUMAN  = "IDENTIFY_MULTIPLE_HUMAN" ;
    //工作服识别
    public static String AI_MODEL_IDENTIFY_DRESS = "IDENTIFY_DRESS";
    //安全帽识别
    public static String AI_MODEL_IDENTIFY_HELMET = "IDENTIFY_HELMET";
    //吸烟识别
    public static String AI_MODEL_IDENTIFY_SMOKING = "IDENTIFY_SMOKING";
    // 仪表识别
    public static String AI_MODEL_IDENTIFY_INSTRUMENT = "IDENTIFY_INSTRUMENT";
}

monitor-manage-service/src/main/java/com/ai/bss/monitorManage/model/MonitorVideoLogDto.java

@@ -0,0 +1,46 @@
package com.ai.bss.monitorManage.model;
import lombok.Data;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
import org.hibernate.annotations.Where;
import javax.persistence.*;
import java.io.Serializable;
import java.util.Date;
@Data
public  class MonitorVideoLogDto implements Serializable {
	private static final long serialVersionUID = -1L;
 	/**
	* 设备视频文件记录标识
	*/
	private Long monitorVideoLogId;
 	/**
	* 工作工具标识
	*/
	private String resourceToolId;
	private String resourceToolCode;
 	/**
	* 视频地址（资源标识）（冗余）
	*/
	private String videoUrl;
 	/**
	* 视频采集开始时间
	*/
	private String beginTime;
 	/**
	* 视频采集结束时间
	*/
	private String endTime;
	private String topic;
/*
INSERT INTO se_monitor_video_log SET _MONITOR_VIDEO_LOG_ID='', RESOURCE_TOOL_ID='', VIDEO_URL='', BEGIN_TIME='', END_TIME='';
*/
}

monitor-manage-service/src/main/java/com/ai/bss/monitorManage/service/impl/KafkaProcessImpl.java

@@ -1,6 +1,7 @@
package com.ai.bss.monitorManage.service.impl;
import com.ai.bss.monitorManage.model.AiIdenLog;
import com.ai.bss.monitorManage.model.AiIdenLogDto;
import com.ai.bss.monitorManage.service.interfaces.KafkaProcess;
import com.ai.bss.monitorManage.utils.KafkaProducerConfig;
import com.alibaba.fastjson.JSONObject;
@@ -25,29 +26,37 @@ public class KafkaProcessImpl implements KafkaProcess {
//    private static String kafkaServers = "47.105.160.21:9090";
    private static String kafkaServers = "10.19.90.34:9090";
    @Value("${kafka.producer.servers}")
    private static String servers;
    @Async
    @Override
    public void processKafka(List<AiIdenLog> aiIdenLogList){
    public void processKafka(String topic,List<String> msgList){
//        List<DataPoint> dataPointList =  parseDataPoint(mockScenarioDataList,topic0,topic1);
        try {
            for (AiIdenLog dataPoint : aiIdenLogList) {
                sendKafkaDataPoint(dataPoint);
            for (String msg : msgList) {
                sendKafkaDataPoint(topic,msg);
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }
    @Async
    @Override
    public void processKafka(String topic,String msg){
        try {
            sendKafkaDataPoint(topic,msg);
        }catch (Exception e){
            e.printStackTrace();
        }
    }
    private Map<String, KafkaTemplate> kafkaTemplateMap = new HashMap<>();
    private boolean sendKafkaDataPoint(AiIdenLog aiIdenLog){
    private boolean sendKafkaDataPoint(String topic,String msgStr){
        KafkaTemplate kafkaTemplate = kafkaTemplateMap.get(kafkaServers);
        if(kafkaTemplate == null){
            //new 实例
@@ -57,7 +66,8 @@ public class KafkaProcessImpl implements KafkaProcess {
        }
        try {
            Object result = kafkaTemplate.send(aiIdenLog.getTopic(), JSONObject.toJSONString(aiIdenLog)).get();
            Object result = kafkaTemplate.send(topic, msgStr).get();
            System.out.println("发送kafka消息：topic="+topic + "msg: "+msgStr);
        } catch (Exception e) {
            log.error(e.getMessage());
            return false;

monitor-manage-service/src/main/java/com/ai/bss/monitorManage/service/impl/MonitorManageServiceImpl.java

@@ -1,16 +1,26 @@
package com.ai.bss.monitorManage.service.impl;
import cn.hutool.json.JSONUtil;
import com.ai.bss.components.common.util.ImageBase64Converter;
import com.ai.bss.components.ebcai.service.EbcAiService;
import com.ai.bss.monitorManage.constant.MonitorManageConsts;
import com.ai.bss.monitorManage.model.AiIdenLogDto;
import com.ai.bss.monitorManage.model.MonitorVideoLogDto;
import com.ai.bss.monitorManage.service.interfaces.FileManageService;
import com.ai.bss.monitorManage.service.interfaces.KafkaProcess;
import com.ai.bss.monitorManage.service.interfaces.ListeningFileDirectoryService;
import com.ai.bss.monitorManage.service.interfaces.MonitorManageService;
import com.ai.bss.monitorManage.utils.DateUtils;
import com.ai.bss.monitorManage.utils.FileUtils;
import com.ai.bss.monitorManage.utils.SecurityProtectionMinioConfig;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.fileupload.FileItem;
import org.apache.commons.fileupload.FileItemFactory;
import org.apache.commons.fileupload.disk.DiskFileItemFactory;
import org.apache.logging.log4j.util.Strings;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
@@ -28,7 +38,7 @@ import java.util.*;
public class MonitorManageServiceImpl implements MonitorManageService {
	private final static String SPLIT_CHAR = "_";
	private static String bucketName =  "";
//	private static String bucketName =  "";
	@Autowired
	private FileManageService fileManageService;
@@ -36,18 +46,34 @@ public class MonitorManageServiceImpl implements MonitorManageService {
	@Autowired
	private ListeningFileDirectoryService listeningFileDirectoryService;
	@Autowired
	private KafkaProcess kafkaProcess;
	@Autowired
	private EbcAiService ebcAiService;
	@Value("${camera.filePath}")
	private String filePath;
	@Value("${kafka.topic.aitask}")
	private String topicAiTask;
	@Value("${kafka.topic.aivideo}")
	private String topicAiVideo;
	@Autowired
	private SecurityProtectionMinioConfig minioConfig;
	@Override
	public void loadPictureFromDirectory() throws InterruptedException {
		//监听文件目录，加载文件
		loadFileList();
		//等到10秒
		Thread.sleep(10000);
//		listeningFileDirectoryService.getFile(filePath);
//		fileList.forEach( tmp -> {System.out.println(tmp.getName()+ ":" + tmp.isDirectory());});
		//保存文件并发送kafka
		saveFile();
	}
@@ -64,7 +90,7 @@ public class MonitorManageServiceImpl implements MonitorManageService {
			}
			File[] devicefiles =file1.listFiles();
			System.out.println("deviceName()   " + file1.getName());
//			System.out.println("deviceName()   " + file1.getName());
			//图片、视频目录
			for(File deviceFile:devicefiles){
				if(!deviceFile.isDirectory()) {
@@ -78,7 +104,6 @@ public class MonitorManageServiceImpl implements MonitorManageService {
	}
	private void saveFile(){
//		Timer timer = new Timer();
		new Timer().schedule(new TimerTask() {
			@SneakyThrows
			@Override
@@ -89,79 +114,74 @@ public class MonitorManageServiceImpl implements MonitorManageService {
				String fileType = "";
				String filePath = "";
				String fileName = "";
				String dateTime = "";
				String[] keyArr;
				MultipartFile multipartFile = null;
				Map<String, String> minioFileMap = null;
				AiIdenLogDto aiIdenLogDto = null;
				List<String> fileList = null;
				Iterator<String> iterator = null;
				MonitorVideoLogDto monitorVideoLogDto = null;
				String time = "";
				String bucketName =  "";
				String aiResponseData = "";
				for(String key:keyList){
					System.out.println("key== " + key);
					String[] keyArr = key.split("~!@");
					keyArr = key.split("~!@");
					deviceCode = keyArr[0];
					fileType = keyArr[1];
					System.out.println("keyArr[0]: "+ deviceCode +"  keyArr[1]: "+ fileType);
					List<String> fileList = fileMap.get(key);
					Iterator<String> iterator = fileList.iterator();
					fileList = fileMap.get(key);
					iterator = fileList.iterator();
					while (iterator.hasNext()) {
						filePath = iterator.next();
						aiIdenLogDto = new AiIdenLogDto();
						monitorVideoLogDto = new MonitorVideoLogDto();
						fileName = filePath.substring(filePath.lastIndexOf("\\")+1,filePath.length());
						System.out.println("fileName=== " + fileName);
						String time = "";
						if(fileType.equals("videos")){
							bucketName = "ai-video";
						if(fileType.equals(MonitorManageConsts.FILE_TYPE_VIDEO)){
							bucketName = minioConfig.getBucketAiVideo();
							time = fileName.substring(5,19);
							aiIdenLogDto.setTaskExecuteTime(DateUtils.formatDate(DateUtils.convertDate(time,DateUtils.PURE_DATETIMEHHMM_PATTERN),DateUtils.NORM_DATETIME_PATTERN));
							aiIdenLogDto.setTopic("topic_ai_video");//视频历史记录TOPIC
							dateTime = DateUtils.formatDate(DateUtils.convertDate(time,DateUtils.PURE_DATETIMEHHMM_PATTERN),DateUtils.NORM_DATETIME_HHMM_PATTERN);
							monitorVideoLogDto.setBeginTime(dateTime + ":00");
							monitorVideoLogDto.setEndTime(dateTime + ":59");
						}else{
							bucketName = "ai-image";
							time = fileName.substring(3,19);
							aiIdenLogDto.setTaskExecuteTime(DateUtils.formatDate(DateUtils.convertDate(fileName.substring(3,17),DateUtils.PURE_DATETIME_PATTERN),DateUtils.NORM_DATETIME_PATTERN));
							aiIdenLogDto.setTopic("topic_ai_task");//AI识别TOPIC
							bucketName = minioConfig.getBucketAiImage();
							time = fileName.substring(3,17);
							dateTime = DateUtils.formatDate(DateUtils.convertDate(time,DateUtils.PURE_DATETIMEHHMM_PATTERN),DateUtils.NORM_DATETIME_PATTERN);
							aiIdenLogDto.setTaskExecuteTime(dateTime);
						}
//						String minioFileName = bucketName + SPLIT_CHAR + deviceCode +"-"+ fileName;
//						multipartFile = new File(filePath);
						//文件上传到 minio文件服务器
						multipartFile = new CommonsMultipartFile(createFileItem(new File(filePath),fileName));
						minioFileMap = fileManageService.uploadFile(multipartFile,bucketName,deviceCode,null);
						System.out.println("fileId=== " + minioFileMap.get("fileId"));
						System.out.println("fileUrl=== " + minioFileMap.get("fileUrl"));
						minioFileMap = fileManageService.uploadFile(multipartFile,bucketName,deviceCode,null);
						aiIdenLogDto.setResourceToolCode(deviceCode);
						if(fileType.equals("videos")){
							aiIdenLogDto.setIdenVideoUrl(minioFileMap.get("fileId"));
						if(fileType.equals(MonitorManageConsts.FILE_TYPE_VIDEO)){
							monitorVideoLogDto.setTopic(topicAiVideo);//视频历史记录TOPIC
							monitorVideoLogDto.setVideoUrl(minioFileMap.get("fileId"));
							monitorVideoLogDto.setResourceToolCode(deviceCode);
							kafkaProcess.processKafka(monitorVideoLogDto.getTopic(), JSONObject.toJSONString(monitorVideoLogDto));
						}else{
							aiIdenLogDto.setIdenVideoUrl("");
							aiIdenLogDto.setIdenPictureUrl(minioFileMap.get("fileId"));
							aiIdenLogDto.setResourceToolCode(deviceCode);
							//安全帽识别
							aiIdenLogDto.setTopic(topicAiTask);//AI识别TOPIC
							aiResponseData = ebcAiService.helmetDetect(ImageBase64Converter.convertFileToBase64(filePath), Strings.EMPTY,Strings.EMPTY);
							System.out.println(JSONUtil.formatJsonStr(aiResponseData));
							log.info(JSONUtil.formatJsonStr(aiResponseData));
//							ebcAiService.helmetDetect()
							//AI模型
							aiIdenLogDto.setAiIdenModel(MonitorManageConsts.AI_MODEL_IDENTIFY_HELMET);
							aiIdenLogDto.setIdenResult(JSONUtil.formatJsonStr(aiResponseData));
							aiIdenLogDto.setIdenResultType("");
							kafkaProcess.processKafka(aiIdenLogDto.getTopic(), JSONObject.toJSONString(aiIdenLogDto));
						}
//						aiIdenLogDto.setTaskExecuteTime();
						iterator.remove();//使用迭代器的删除方法删除
					}
//					for(String filePath:fileList){
//						System.out.println("key: "+key+"  filePath: "+filePath);
//
//						String fileName = filePath.substring(filePath.lastIndexOf("\\"),filePath.length());
//						if(keyArr[1].equals("videos")){
//							bucketName = "ai-video";
//						}else{
//							bucketName = "ai-image";
//						}
//						String minioFileName = bucketName + SPLIT_CHAR + keyArr[0] +"-"+ fileName;
//
//						File multipartFile = new File(filePath);
//
//						MultipartFile mfile = new CommonsMultipartFile(createFileItem(multipartFile,minioFileName));
//						Map<String, String> minioFileMap = fileManageService.uploadFile(mfile,null);
//						System.out.println("pictureUrl=== " + minioFileMap.get("pictureUrl"));
//						fileList.remove(filePath);
//					}
				}
				System.out.println("-------设定要指定任务--------");
			}
@@ -174,7 +194,6 @@ public class MonitorManageServiceImpl implements MonitorManageService {
		while (iter.hasNext()) {
			Map.Entry entry = (Map.Entry) iter.next();
			list.add(entry.getKey());
//			list.add(entry.getValue());
		}
		return list;
	}

monitor-manage-service/src/main/java/com/ai/bss/monitorManage/service/interfaces/KafkaProcess.java

@@ -1,12 +1,12 @@
package com.ai.bss.monitorManage.service.interfaces;
import com.ai.bss.monitorManage.model.AiIdenLog;
import java.util.List;
public interface KafkaProcess {
//    void processMock(Long sId, List<AiIdenLog> mockScenarioDataList, Long frequency);
    void processKafka(List<AiIdenLog> aiIdenLogList);
    void processKafka(String topic,List<String> msgList);
    void processKafka(String topic,String msgStr);
}

monitor-manage-service/src/main/java/com/ai/bss/monitorManage/utils/DateUtils.java

@@ -16,6 +16,7 @@ import java.util.Date;
public class DateUtils {
	public static final String NORM_DATETIME_PATTERN = "yyyy-MM-dd HH:mm:ss";
	public static final String NORM_DATETIME_HHMM_PATTERN = "yyyy-MM-dd HH:mm";
	public static final String NORM_DATETIME_MS_PATTERN = "yyyy-MM-dd HH:mm:ss.SSS";
	public static final String NORM_YYYY_MM_PATTERN = "yyyy-MM";
	public static final String NORM_DATE_PATTERN = "yyyy-MM-dd";

monitor-manage-service/src/main/java/com/ai/bss/monitorManage/utils/SecurityProtectionMinioConfig.java

@@ -16,8 +16,8 @@ import org.springframework.stereotype.Component;
@ConfigurationProperties(prefix = "myminio")
public class SecurityProtectionMinioConfig {
	private String bucketWorkToolImage;
	private String bucketAiImage;
	private String bucketWorkToolView;
	private String bucketAiVideo;
}

monitor-manage-service/src/main/resources/application.properties

@@ -57,8 +57,8 @@ minio.faceAddServiceUrl=http://10.21.10.28:9018/api/face/add
minio.face-del-service-url=http://10.21.10.28:9018/api/face/del
minio.face-recog-service-url=http://10.21.10.28:9018/api/face/recog
myminio.bucketWorkToolImage=tool-image
myminio.bucketWorkToolView=tool-view
myminio.bucketAiImage=ai-image
myminio.bucketAiVideo=ai-video
# CACHE
#spring.cache.type=ehcache
@@ -70,3 +70,19 @@ logging.level.org.springframework.data=debug
camera.filePath=E:\\data\\DAHUA
#AIÖ´Ðнá¹û
kafka.topic.aitask=topicTaskTest
#kafka.topic.aitask=topic_ai_task
#AIÊÓƵ
kafka.topic.aivideo=topicVideoTest
#kafka.topic.aivideo=topic_ai_video
ebc.ai.head-det-url=http://10.1.251.211:7001/todo/api/v1.4/control/online/IMAGE/HEAD/PREDICT
ebc.ai.helmet-det-url=http://10.1.251.211:7003/todo/api/v1.4/control/online/IMAGE/HELMET/PREDICT
ebc.ai.smokingdet-url=http://10.1.251.213:5125/smokingdet
ebc.ai.instrument-det-url=http://10.1.251.213:19100/predict
ebc.ai.work-clothes-det-url=http://10.1.251.211:7003/todo/api/v1.4/control/online/IMAGE/HUMAN/PREDICT