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

@@ -7,6 +7,8 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import com.ai.bss.monitorManage.utils.*;
import com.sun.javafx.collections.MappingChange;
import org.apache.commons.lang.StringUtils;
import org.apache.logging.log4j.util.Strings;
import org.slf4j.Logger;
@@ -29,11 +31,6 @@ import com.ai.bss.monitorManage.model.IdenResultDto;
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.utils.DateUtils;
import com.ai.bss.monitorManage.utils.FileUtils;
import com.ai.bss.monitorManage.utils.HttpServiceUtil;
import com.ai.bss.monitorManage.utils.KafkaProducerConfig;
import com.ai.bss.monitorManage.utils.SecurityProtectionMinioConfig;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
@@ -320,6 +317,10 @@ public class KafkaProcessImpl implements KafkaProcess {
							idenResultDto.setAlarmEmployeeRoleId(MonitorManageConsts.alarmEmployeeRoleId);
							idenResultDto.setAlarmType(aiClassName);
							idenResultDto.setAlarmMemo(aiClassName);
							Map resultMap = (Map)jsonoArray.get(i);
							int[] boxIntArr = ArrayUtils.JsonToArray((JSONArray)resultMap.get("box"));
							idenResultDto.setBox(boxIntArr);
							idenResultDtoList.add(idenResultDto);
							// aiIdenLogDto.setIdenResult(idenResultDtoList);
							aiIdenLogDto.setIdenResult(JSON.toJSONString(idenResultDtoList));
@@ -331,6 +332,11 @@ public class KafkaProcessImpl implements KafkaProcess {
							idenResultDto.setAlarmEmployeeRoleId(MonitorManageConsts.alarmEmployeeRoleId);
							idenResultDto.setAlarmType(MonitorManageConsts.AI_ALARM_TYPE_WEA);
							idenResultDto.setAlarmMemo("未戴安全帽");
							Map resultMap = (Map)jsonoArray.get(i);
							int[] boxIntArr = ArrayUtils.JsonToArray((JSONArray)resultMap.get("box"));
							idenResultDto.setBox(boxIntArr);
							aiIdenLogDto.setIdenResultType(MonitorManageConsts.AI_IDENTIFY_RESULT_ALARM);
							idenResultDtoList.add(idenResultDto);
							// aiIdenLogDto.setIdenResult(idenResultDtoList);
@@ -434,12 +440,7 @@ public class KafkaProcessImpl implements KafkaProcess {
					resultDto.setSimi(score.setScale(4,BigDecimal.ROUND_HALF_UP).floatValue());
					resultDto.setIsMatch(0);
					JSONArray boxJSONArray = (JSONArray) resultMap.get("box");
					int[] boxIntArr=new int[boxJSONArray.size()];
					for(int i = 0;i<boxJSONArray.size();i++){
						BigDecimal boxValue=(BigDecimal) boxJSONArray.get(i);
						boxIntArr[i] =boxValue.intValue();
					}
					int[] boxIntArr = ArrayUtils.JsonToArray((JSONArray) resultMap.get("box"));
					resultDto.setBox(boxIntArr);
					aiIdenLogDto.setIdenResultType(MonitorManageConsts.AI_IDENTIFY_RESULT);
@@ -449,12 +450,7 @@ public class KafkaProcessImpl implements KafkaProcess {
					resultDto.setSimi(score.setScale(4,BigDecimal.ROUND_HALF_UP).floatValue());
					resultDto.setIsMatch(1);
					JSONArray boxJSONArray = (JSONArray) resultMap.get("box");
					int[] boxIntArr=new int[boxJSONArray.size()];
					for(int i = 0;i<boxJSONArray.size();i++){
						BigDecimal boxValue=(BigDecimal) boxJSONArray.get(i);
						boxIntArr[i] =boxValue.intValue();
					}
					int[] boxIntArr = ArrayUtils.JsonToArray((JSONArray) resultMap.get("box"));
					resultDto.setBox(boxIntArr);
					aiIdenLogDto.setRelateEmployeeRoleId(faceId);
@@ -512,12 +508,7 @@ public class KafkaProcessImpl implements KafkaProcess {
					idenResultDto.setAlarmType(MonitorManageConsts.AI_ALARM_TYPE_STA);
					idenResultDto.setAlarmMemo("陌生人");
					JSONArray boxJSONArray = (JSONArray) resultMap.get("box");
					int[] boxIntArr=new int[boxJSONArray.size()];
					for(int i = 0;i<boxJSONArray.size();i++){
						BigDecimal boxValue=(BigDecimal) boxJSONArray.get(i);
						boxIntArr[i] =boxValue.intValue();
					}
					int[] boxIntArr = ArrayUtils.JsonToArray((JSONArray) resultMap.get("box"));
					idenResultDto.setBox(boxIntArr);
					idenResultDtoList.add(idenResultDto);
@@ -571,6 +562,11 @@ public class KafkaProcessImpl implements KafkaProcess {
                                    idenResultDto.setAlarmEmployeeRoleId(MonitorManageConsts.alarmEmployeeRoleId);
                                    idenResultDto.setAlarmType(MonitorManageConsts.AI_ALARM_TYPE_VIO);
                                    idenResultDto.setAlarmMemo("违规吸烟");
								    Map resultMap = (Map) jsonArray.get(i);
									int[] boxIntArr = ArrayUtils.JsonToArray((JSONArray) resultMap.get("box"));
									idenResultDto.setBox(boxIntArr);
                                    log.info("检测到违规吸烟");
                                    aiIdenLogDto.setIdenResultType(MonitorManageConsts.AI_IDENTIFY_RESULT_ALARM);
                                    idenResultDtoList.add(idenResultDto);
@@ -621,6 +617,10 @@ private AiIdenLogDto workClothesDetect(AiIdenLogDto aiIdenLogDto, String aifileP
						idenResultDto.setAlarmType(MonitorManageConsts.AI_ALARM_TYPE_WEA);
						idenResultDto.setAlarmMemo("未穿工作服");
						log.info("未穿工作服");
						int[] boxIntArr = ArrayUtils.JsonToArray((JSONArray) resultList.get(i).get("box"));
						idenResultDto.setBox(boxIntArr);
						aiIdenLogDto.setIdenResultType(MonitorManageConsts.AI_IDENTIFY_RESULT_ALARM);
						idenResultDtoList.add(idenResultDto);
						aiIdenLogDto.setIdenResult(JSON.toJSONString(idenResultDtoList));
@@ -630,6 +630,10 @@ private AiIdenLogDto workClothesDetect(AiIdenLogDto aiIdenLogDto, String aifileP
						idenResultDto.setAlarmEmployeeRoleId(MonitorManageConsts.alarmEmployeeRoleId);
						idenResultDto.setAlarmType(aiClassName);
						idenResultDto.setAlarmMemo(aiClassName);
						int[] boxIntArr = ArrayUtils.JsonToArray((JSONArray) resultList.get(i).get("box"));
						idenResultDto.setBox(boxIntArr);
						idenResultDtoList.add(idenResultDto);
						aiIdenLogDto.setIdenResult(JSON.toJSONString(idenResultDtoList));
						aiIdenLogDto.setIdenResultType(MonitorManageConsts.AI_IDENTIFY_RESULT);
@@ -671,6 +675,8 @@ private AiIdenLogDto instrumentDetect(AiIdenLogDto aiIdenLogDto, String aifilePa
				aiIdenLogDto.setRelateEmployeeRoleId(MonitorManageConsts.alarmEmployeeRoleId);
				log.info("返回的仪表信息为："+jsonObjectpredictions.getString("Items"));
				aiIdenLogDto.setIdenResult(jsonObjectpredictions.getString("Items"));
				log.info("仪表识别成功，推送KAFKA消息");
				processKafka(aiIdenLogDto.getTopic(), JSONObject.toJSONString(aiIdenLogDto));
				log.info("消息推送完成");

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

@@ -0,0 +1,32 @@
package com.ai.bss.monitorManage.utils;
import com.alibaba.fastjson.JSONArray;
import java.math.BigDecimal;
import java.util.Calendar;
import java.util.Date;
/**
 * @Auther: 王超
 * @Date: 2021/2/3 17:49
 * @Description:
 */
public class ArrayUtils {
    /**
     * JSONArray转int[]
     * @param jsonArray 原json
     *
     * @return boxIntArr
     */
    public static int[] JsonToArray(JSONArray jsonArray) {
        int[] boxIntArr=new int[jsonArray.size()];
        for(int i = 0;i<jsonArray.size();i++){
            BigDecimal boxValue=(BigDecimal) jsonArray.get(i);
            boxIntArr[i] =boxValue.intValue();
        }
        return boxIntArr;
    }
}

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

@@ -74,7 +74,7 @@ logging.level.org.springframework.data=debug
camera.filePath=/home/puaiuc/data/DAHUA
#camera.filePath=E:\\data\\DAHUA
#camera.filePath=D:\\data\\DAHUA
#AI\u05b4\ufffd\u043d\ufffd\ufffd
#kafka.topic.aitask=topicTaskTest

security-protection-service/pom.xml

@@ -186,6 +186,18 @@
            <version>2.6.12</version>
        </dependency>
        <!--  IPU 镜屏 -->
        <dependency>
            <groupId>com.ai.ipu.server</groupId>
            <artifactId>iot-stomp-server</artifactId>
            <version>3.1-SNAPSHOT</version>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.logging.log4j</groupId>
                    <artifactId>log4j-slf4j-impl</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
    </dependencies>
    <repositories>

security-protection-service/src/main/java/com/ai/bss/security/protection/SecurityProtectionApp.java

@@ -1,5 +1,8 @@
package com.ai.bss.security.protection;
import com.ai.ipu.server.connect.ConnectServerStart;
import com.ai.ipu.server.connect.util.ConnectServerManager;
import com.ai.ipu.server.stomp.WebSocketStompServer;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.domain.EntityScan;
@@ -18,6 +21,13 @@ import org.springframework.scheduling.annotation.EnableAsync;
@SpringBootApplication
public class SecurityProtectionApp {
    public static void main(String[] args) throws Exception {
        System.setProperty("websocket.port", "7200");
        //注册镜屏服务
        ConnectServerManager.registerServer("websocket.port", new WebSocketStompServer("/stomp"));
        //镜屏服务启动
        ConnectServerStart.start(args, false);
        SpringApplication.run(SecurityProtectionApp.class, args);
    }
}

security-protection-service/src/main/java/com/ai/bss/security/protection/config/MyCorsFilter.java

@@ -4,6 +4,8 @@ import com.ai.sso.filter.SessionUserFilter;
import org.springframework.boot.web.servlet.FilterRegistrationBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.web.cors.CorsConfiguration;
import org.springframework.web.cors.UrlBasedCorsConfigurationSource;
import org.springframework.web.filter.CorsFilter;
@@ -11,6 +13,7 @@ import org.springframework.web.filter.CorsFilter;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ThreadPoolExecutor;
/**
 * 解决接口跨域问题
@@ -58,4 +61,16 @@ public class MyCorsFilter {
        registrationBean.setInitParameters(initParameters);
        return registrationBean;
    }
    @Bean
    public ThreadPoolTaskExecutor executor(){
        ThreadPoolTaskExecutor executor=new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10); //核心线程数
        executor.setMaxPoolSize(20);  //最大线程数
        executor.setQueueCapacity(1000); //队列大小
        executor.setKeepAliveSeconds(300); //线程最大空闲时间
        executor.setThreadNamePrefix("fsx-Executor-"); //指定用于新创建的线程名称的前缀。
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        return executor;
    }
}

security-protection-service/src/main/java/com/ai/bss/security/protection/service/task/AiResultRecordKafkaTask.java

@@ -2,11 +2,14 @@ package com.ai.bss.security.protection.service.task;
import java.util.List;
import java.util.Map;
import java.util.concurrent.FutureTask;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.TaskExecutor;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
@@ -47,8 +50,11 @@ public class AiResultRecordKafkaTask {
	@Autowired
	private EmployeeService employeeService;
	@KafkaListener(containerFactory = "kafkaBatchListener6", topics = "${kafka.topic.aitask:topic_ai_task}", groupId = "ai_group")
	public void AiResultRecordListener(ConsumerRecord<String, String> records, Acknowledgment ack) throws Throwable {
	/*@KafkaListener(containerFactory = "kafkaBatchListener6", topics = "${kafka.topic.aitask:topicTaskTest}", groupId = "ai_group")
	public void AiResultRecordListener(ConsumerRecord<String, String> records, Acknowledgment ack) throws Throwable {*/
		try {
			log.debug("----------------AI任务执行结果信息消费开始---------------------------");
@@ -58,6 +64,10 @@ public class AiResultRecordKafkaTask {
			AiIdenLog aiIdenLog = JSON.parseObject(message, new TypeReference<AiIdenLog>() {
			});
			//将结果使用一步方式用镜屏推送到前台
			FutureTask<Integer> integerFutureTask = new FutureTask<>(new MirrorSendAiDataThread(message));
			new Thread(integerFutureTask).start();
			JSONObject messageJson = JSONObject.parseObject(message);
			String resourceToolId = messageJson.getString("resourceToolId");
			String idenResultType = messageJson.getString("idenResultType");

security-protection-service/src/main/java/com/ai/bss/security/protection/service/task/MirrorManagement.java

@@ -0,0 +1,41 @@
package com.ai.bss.security.protection.service.task;
import com.ai.ipu.server.connect.ps.EventChannelTool;
import com.ai.ipu.server.connect.ps.EventChannelTool.PublisherEvent;
import com.ai.ipu.server.connect.ps.ISubscriber;
import com.ai.ipu.server.connect.ps.impl.Publisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
 * 镜屏推送消息
 * @author konghl@asiainfo.com
 * 2020-10-21
 */
public class MirrorManagement {
	private static Logger logger = LoggerFactory.getLogger(MirrorManagement.class);
	// ai返回结果topic
	public static final String SEND_AI_DATA = "sendAiData";
	public static void sendData(String topic, String msg) {
		logger.debug("推送镜屏数据: topic=" + topic + " , data=" + msg);
		if (!EventChannelTool.isExistSubscriber(topic)) {
			// 如果没有订阅 先不管
			logger.debug("未推送镜屏数据，原因是没有订阅：topic="+topic);
			return;
		}
		// 你发送的内容
		final Publisher publisher = new Publisher(topic);
		PublisherEvent<?> publisherEvent = new PublisherEvent<String>(msg) {
			@Override
			public void callback(String topic, ISubscriber... subscribers) {
				// 可记录消息发布后的统计日志
			}
		};
		publisher.publish(topic, publisherEvent);
		logger.debug("推送镜屏数据成功");
	}
}

security-protection-service/src/main/java/com/ai/bss/security/protection/service/task/MirrorSendAiDataThread.java

@@ -0,0 +1,54 @@
package com.ai.bss.security.protection.service.task;
import com.ai.abc.api.model.CommonRequest;
import com.ai.abc.api.model.CommonResponse;
import com.ai.bss.security.protection.service.interfaces.CharSpecService;
import com.ai.bss.work.task.model.common.WorkTask;
import com.ai.bss.worker.service.api.EntityAlarmHandle;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.collections4.map.HashedMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;
import java.util.Map;
import java.util.concurrent.Callable;
public class MirrorSendAiDataThread implements Callable<Integer> {
	private Logger logger = LoggerFactory.getLogger(MirrorSendAiDataThread.class);
	private String message;
	public MirrorSendAiDataThread(String aimessage) {
		this.message = aimessage;
	}
	@Override
	public Integer call() throws Exception {
		System.out.println("执行call。。。。。。。。。。。。。。方法");
		mirrorSendAiData(message);
		return null;
	}
	/**
	 * 镜屏推送ai识别数据
	 *
	 * @param message
	 * @return
	 */
	public Object mirrorSendAiData(String message) {
		logger.debug("====================实时报警接收镜屏数据: " + message);
		try {
			logger.info("实时报警推送镜屏数据: " + message);
			MirrorManagement.sendData(MirrorManagement.SEND_AI_DATA, message);
		} catch (Exception e) {
			logger.error("实时报警推送镜屏失败: " + e.getMessage());
		}
		return null;
	}
}