herb.git - Gitblit

			@@ -1,23 +1,35 @@
			package org.jeecg.modules.dry.service.impl;

			import cn.hutool.core.bean.BeanUtil;
			import cn.hutool.core.thread.ThreadUtil;
			import com.alibaba.fastjson.JSON;
			import com.alibaba.fastjson.JSONArray;
			import com.alibaba.fastjson.JSONObject;
			import com.alibaba.fastjson.TypeReference;
			import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
			import lombok.extern.slf4j.Slf4j;
			import org.apache.commons.lang3.StringUtils;
			import org.apache.mina.core.service.IoAcceptor;
			import org.apache.mina.core.service.IoHandler;
			import org.apache.mina.core.session.IoSession;
			import org.apache.shiro.SecurityUtils;
			import org.eclipse.paho.client.mqttv3.MqttClient;
			import org.eclipse.paho.client.mqttv3.MqttException;
			import org.eclipse.paho.client.mqttv3.MqttMessage;
			import org.jeecg.common.api.CommonAPI;
			import org.jeecg.common.api.vo.Result;
			import org.jeecg.common.config.TenantContext;
			import org.jeecg.common.config.mqtoken.UserTokenContext;
			import org.jeecg.common.constant.CommonConstant;
			import org.jeecg.common.constant.MqttConstant;
			import org.jeecg.common.system.util.JwtUtil;
			import org.jeecg.common.system.vo.LoginUser;
			import org.jeecg.common.util.DateUtils;
			import org.jeecg.common.util.RedisUtil;
			import org.jeecg.common.util.SpringContextUtils;
			import org.jeecg.modules.dry.common.CacheConstants;
			import org.jeecg.modules.dry.entity.*;
			import org.jeecg.modules.dry.mqtt.MqttUtil;
			import org.jeecg.modules.dry.service.*;
			import org.jeecg.modules.dry.socket.ServerHandler;
			import org.jeecg.modules.dry.socket.SocketServerConfig;
			@@ -25,17 +37,17 @@
			import org.jeecg.modules.dry.vo.*;
			import org.jetbrains.annotations.NotNull;
			import org.springframework.beans.factory.annotation.Autowired;
			import org.springframework.beans.factory.annotation.Value;
			import org.springframework.stereotype.Service;
			import org.springframework.transaction.annotation.Transactional;

			import javax.security.auth.login.LoginContext;
			import java.io.IOException;
			import java.io.ObjectOutputStream;
			import java.net.Socket;
			import java.text.DecimalFormat;
			import java.util.ArrayList;
			import java.util.List;
			import java.util.Map;
			import java.util.Set;
			import java.util.*;
			import java.util.stream.Collectors;

			@Slf4j
			@Service
			@@ -60,10 +72,17 @@
			@Autowired
			private IDryProdRecordService prodRecordService;

			@Autowired
			private IDryFaultRecordService faultRecordService;


			private String token;

			@Value(value = "${jeecg.mqtt.role}")
			private String role;

			@Autowired
			private MqttUtil mqttUtil;

			public String getTemporaryToken() {
			if (token == null) {
			@@ -97,6 +116,10 @@
			// 根据租户id和工单号查询数据库是否有记录，有则返回，没有则新增一条再返回
			orderVo = getOrSaveDryOrderVoDB(realTimeDataVo);
			}
			if (orderVo == null) {
			log.error("工单不存在，工单号："+realTimeDataVo.getWorkorder()+",设备：" + realTimeDataVo.getMachineid() +",药材：" + realTimeDataVo.getName());
			return Result.error("工单不存在");
			}

			// 2 更新工单实时数据
			// 2.1 将工单中的数据替换为最新数据
			@@ -126,7 +149,7 @@
			orderVo.setState_valve(realTimeDataVo.getState_valve());
			orderVo.setOrderStatus(realTimeDataVo.getWorkorder_status());
			orderVo.setEqp_status(realTimeDataVo.getEqp_status());
			orderVo.setEqp_state(realTimeDataVo.getEqp_state());
			// orderVo.setEqp_state(realTimeDataVo.getEqp_state());
			orderVo.setWarning(realTimeDataVo.getEqp_warning());
			orderVo.setFault(realTimeDataVo.getEqp_fault());
			orderVo.setLevel(realTimeDataVo.getLevel());
			@@ -232,16 +255,25 @@
			* @param realTimeDataVo
			* @return
			*/
			@NotNull
			private DryOrderVo saveNewOrder(RealTimeDataVo realTimeDataVo) {
			DryOrderVo orderVo;
			// 查询药材
			DryHerbFormula herbFormula = queryHerbByIndexTenant(realTimeDataVo);

			// 查询设备
			DryEquipment equ = queryEquipmentByCodeTenant(realTimeDataVo);
			if (equ == null) {
			log.error("未找到设备："+realTimeDataVo.getIndex() +","+realTimeDataVo.getName() +",机台：" + realTimeDataVo.getMachineid());
			return null;
			}
			// 查询药材
			DryHerbFormula herbFormula = queryHerbByIndexTenant(realTimeDataVo);

			if (herbFormula == null) {
			log.error("未找到药材："+realTimeDataVo.getIndex() +","+realTimeDataVo.getName() +",机台：" + realTimeDataVo.getMachineid());
			return null;
			}
			// 创建新工单
			orderVo = new DryOrderVo(realTimeDataVo);

			orderVo.setHerbId(herbFormula.getId());
			orderVo.setEquId(equ.getId());
			DryOrder dryOrder = BeanUtil.toBean(orderVo, DryOrder.class);
			@@ -261,8 +293,24 @@
			queryWrapper.eq(DryEquipment::getCode, realTimeDataVo.getMachineid());
			DryEquipment one = equipmentService.getOne(queryWrapper);
			if (one == null) {
			one = new DryEquipment(realTimeDataVo);
			equipmentService.save(one);
			log.error(role+"保存实时数据，未找到设备："+realTimeDataVo.getMachineid());
			// one = new DryEquipment(realTimeDataVo);
			// equipmentService.save(one);
			if (MqttConstant.ROLE_ADMIN.equals(role)) {
			MqttMessage mqttMessage = new MqttMessage();
			mqttMessage.setQos(0);
			JSONObject object = new JSONObject();
			object.put("code", realTimeDataVo.getMachineid());
			object.put("tenantId", realTimeDataVo.getTenantid());
			mqttMessage.setPayload(object.toJSONString().getBytes());
			try {
			mqttUtil.getMqttClient().publish(MqttConstant.SERVICE_REQ_PREFIX ,mqttMessage);
			}catch (MqttException e) {
			e.printStackTrace();
			}

			}
			return null;
			}
			return one;
			}
			@@ -275,12 +323,14 @@
			private DryHerbFormula queryHerbByIndexTenant(RealTimeDataVo realTimeDataVo) {
			LambdaQueryWrapper<DryHerbFormula> queryWrapper = new LambdaQueryWrapper<>();
			queryWrapper.eq(DryHerbFormula::getTenantId, realTimeDataVo.getTenantid());
			queryWrapper.eq(DryHerbFormula::getCode, realTimeDataVo.getIndex());
			queryWrapper.eq(DryHerbFormula::getName, realTimeDataVo.getName());
			DryHerbFormula one = dryHerbFormulaService.getOne(queryWrapper);
			// if (one == null) {
			// one = new DryHerb(realTimeDataVo);
			// herbService.save(one);
			// }
			if (one == null) {
			one = new DryHerbFormula(realTimeDataVo);
			DryEquipment dryEquipment = equipmentService.selectByTenantIdEquipmentId(realTimeDataVo.getTenantid() + "", realTimeDataVo.getMachineid());
			one.setEqpType(dryEquipment.getType());
			dryHerbFormulaService.save(one);
			}
			return one;
			}

			@@ -292,7 +342,7 @@
			private void saveOrderTrendVo(DryOrderTrendVo trendVo, DryOrderVo orderVo) {
			//判断实时含水率或实时重量有没有变化，有变化则更新
			if(orderVo.getTrendVo() == null && trendVo != null && trendVo.getWeight() > 0
			\|\| trendVo.getWeight() < orderVo.getTrendVo().getWeight()
			\|\| orderVo.getTrendVo()!=null && trendVo.getWeight() < orderVo.getTrendVo().getWeight()
			) {
			DryOrder byId = dryOrderService.getById(orderVo.getId());
			// 将最新结果更新到工单
			@@ -318,7 +368,7 @@
			public Result<?> queryMachineRealTImeData(RealTimeDataVo realTimeDataVo) {
			TenantContext.setTenant(realTimeDataVo.getTenantid()+"");

			// 查询所有机台查询组装
			// 查询所有机台,查询语句组装
			LambdaQueryWrapper<DryEquipment> queryWrapper = new LambdaQueryWrapper<>();
			queryWrapper.eq(DryEquipment::getTenantId, realTimeDataVo.getTenantid());
			queryWrapper.eq(DryEquipment::getEnable, "Y");
			@@ -351,9 +401,11 @@
			dList.add(50d);
			}

			} else {
			// 如果没有生产，效率置0
			dList.add(0d);
			}
			// 如果没有生产，效率置0
			dList.add(0d);

			});
			}

			@@ -439,4 +491,112 @@
			public Result<?> statisticsDataHandle(StatisticsDataVo statsDataVo) {
			return null;
			}

			@Override
			public Result<?> fitFultRecord(RealTimeDataVo vo) {
			ThreadUtil.execute(() -> {
			try {
			//解析存储报警数据
			List<DryFaultRecord> faultRecords1 = fitFault(vo.getEqp_fault(), vo.getWorkorder(), vo.getTenantid(), vo.getMachineid(), 1);
			List<DryFaultRecord> faultRecords2 = fitFault(vo.getEqp_warning(), vo.getWorkorder(), vo.getTenantid(), vo.getMachineid(), 2);
			faultRecords1.addAll(faultRecords2);
			JSONObject json = new JSONObject();
			json.put("data",JSON.toJSONString(faultRecords1));

			if(faultRecords1.isEmpty()) return;
			MqttMessage mqttMessage = new MqttMessage();
			mqttMessage.setQos(0);
			mqttMessage.setPayload((JSON.toJSONString(json).getBytes()));
			mqttUtil.getMqttClient().publish(MqttConstant.TENANT_UP_PREFIX_FAULT_DATA,mqttMessage);

			} catch (Exception e) {
			e.printStackTrace();
			}
			});

			return null;
			}


			/**
			* 解析存储故障数据
			* TODO 保证原子性
			* @param fault 故障数据
			* @param orderId 工单
			* @param tenantId 租户
			* @param machineId 设备
			* @param faultType 故障类型
			* @return 组装好故障数据
			*/
			private List<DryFaultRecord> fitFault(String fault, String orderId,Integer tenantId,String machineId,Integer faultType){
			List<DryFaultRecord> result = new ArrayList<>();
			//数据样本："eqp_fault": "滚筒降超时-报警,风机过流报警,滚筒升超时-报警,风箱升报警",
			System.err.println((faultType == 1 ? "类型：故障" : "类型：报警") + DateUtils.formatDateTime()+"--"+fault);
			//redis中的故障
			Map<Object, Object> rFauMap = redisUtil.hmget(MqttConstant.MQTT_EQP_FAULT);
			Map<String, Object> redFauMap = rFauMap.entrySet().stream()
			.collect(Collectors.toMap(
			entry -> entry.getKey().toString(), // 键转换为字符串
			entry -> entry.getValue()
			));

			//没有生成工单的故障数据不存储
			if(StringUtils.isEmpty(orderId)){
			return result;
			}

			if(StringUtils.isEmpty(fault) && rFauMap.isEmpty()){
			return result;
			}
			//1.解析数据
			String[] eqpFaults = fault.split(",");
			Map<String,DryFaultRecord> realFauMap = new HashMap<>();
			for (int i = 0; i < eqpFaults.length; i++) {
			String eqpFault = eqpFaults[i];
			//避免空字符串
			if(StringUtils.isEmpty(eqpFault)) continue;
			//1.1检查mqtt中是否已存在这个故障
			String redisKey = String.format("%s_%s_%s", tenantId, machineId,eqpFault);
			String rFault = (String) redisUtil.get(redisKey);
			//1.2如果redis不存在则存入（存故障开始）
			if(rFault ==null){
			//组装缓存数据
			DryFaultRecord faultRecord = new DryFaultRecord(orderId,eqpFault,faultType,new Date(),null);
			realFauMap.put(redisKey,faultRecord);
			}

			}

			//1.3缓存至redis
			//合并数据
			realFauMap.forEach((key, value) -> redFauMap.putIfAbsent(key, value));
			//没有新故障数据不用覆盖
			if(!realFauMap.isEmpty()){
			redisUtil.hmset(MqttConstant.MQTT_EQP_FAULT,redFauMap);
			}



			//2检测已结束的故障
			//2.1如果实时数据不存在redis存在则代表故障结束，存入数据库
			Map<Object, Object> curFauMap = redisUtil.hmget(MqttConstant.MQTT_EQP_FAULT);
			curFauMap.keySet().stream()
			//特别注意，多个报警类型共用方法需要区分类型
			.filter(key -> !realFauMap.containsKey(key) && ((DryFaultRecord)curFauMap.get(key)).getFaultType() == faultType)
			.forEach(key -> {
			System.err.println((faultType == 1 ? "类型：故障" : "类型：报警") + DateUtils.formatDateTime()+"存入数据库");
			DryFaultRecord record = (DryFaultRecord)redFauMap.get(key);
			record.setEndTime(new Date());
			faultRecordService.save(record);
			redisUtil.hdel(MqttConstant.MQTT_EQP_FAULT,key);

			result.add(record);

			});

			return result;
			}



			}