新增 气象站单条和多条件预警功能

src/main/java/com/yunfeiyun/agmp/iots/device/serviceImp/YfQxzDeviceImpl.java

@@ -275,6 +275,9 @@ public class YfQxzDeviceImpl extends DeviceAbstractImpl implements IYfQxzDevice
             reportDt = DateUtils.parseDate(reportTime);
         }*/
 
+        String yuLiangEnum = EnumYfQxzElement.TYPE4.geteNum();
+        IotYfqxzdata iotYuLiangData = null;
+        Set<String> enumSet = new HashSet<>();
         // 更新设备数据信息到数据库 mongodb
         List<IotYfqxzdata> iotYfqxzdataList = new ArrayList<>();
         for (Object item: jarrData) {
@@ -289,7 +292,6 @@ public class YfQxzDeviceImpl extends DeviceAbstractImpl implements IYfQxzDevice
             iotYfqxzdata.setEName(jobj.getString("eName"));
             iotYfqxzdata.setEKey(eKey);
             iotYfqxzdata.setEValue(eValue);
-
             iotYfqxzdata.setTime(reportDt);
 
             iotYfqxzdataList.add(iotYfqxzdata);
@@ -311,32 +313,42 @@ public class YfQxzDeviceImpl extends DeviceAbstractImpl implements IYfQxzDevice
                 if(ev < 0){
                     ev = 0.0;
                 }
-                IotYfqxzdata iotYfqxzdata2 = new IotYfqxzdata();
-                iotYfqxzdata2.setCId(iotDeviceOld.getTid());
-                iotYfqxzdata2.setDevBid(iotDeviceOld.getDevBid());
-                iotYfqxzdata2.setENum(EnumYfQxzElement.TYPE4.geteNum());
-                iotYfqxzdata2.setEName(EnumCommonDataFactor.TYPE_104.getCode());
-                iotYfqxzdata2.setEKey("e" + EnumYfQxzElement.TYPE4.geteNum());
-                iotYfqxzdata2.setEValue(new DecimalFormat("0.00").format(ev));
-                iotYfqxzdata2.setTime(reportDt);
-
-                iotYfqxzdataList.add(iotYfqxzdata2);
+
+                iotYuLiangData = new IotYfqxzdata();
+                iotYuLiangData.setCId(iotDeviceOld.getTid());
+                iotYuLiangData.setDevBid(iotDeviceOld.getDevBid());
+                iotYuLiangData.setENum(yuLiangEnum);
+                iotYuLiangData.setEName(EnumCommonDataFactor.TYPE_104.getCode());
+                iotYuLiangData.setEKey("e" + yuLiangEnum);
+                iotYuLiangData.setEValue(new DecimalFormat("0.00").format(ev));
+                iotYuLiangData.setTime(reportDt);
             }
 
+            enumSet.add(eNum);
             //把“上报时间”放里面
             jobj.put("time",DateUtils.parseDateToStr("yyyy-MM-dd'T'HH:mm:ss.SSSX",reportDt));
             jobj.put("devBid",iotYfqxzdata.getDevBid());
             jobj.put("cId",iotYfqxzdata.getCId());
         }
+
+        if(!enumSet.contains(yuLiangEnum) && iotYuLiangData != null){
+            iotYfqxzdataList.add(iotYuLiangData);
+
+            JSONObject yuliangObj = new JSONObject();
+            yuliangObj.put("eNum", iotYuLiangData.getENum());
+            yuliangObj.put("eName", iotYuLiangData.getEName());
+            yuliangObj.put("eKey", iotYuLiangData.getEKey());
+            yuliangObj.put("eValue", iotYuLiangData.getEValue());
+            ext.getJSONArray("data").add(yuliangObj);
+        }
+
         if(!iotYfqxzdataList.isEmpty()){
             mongoService.insertList(IotYfqxzdata.class, iotYfqxzdataList);
         }
         iIotDevicelasteddataService.updateDeviceLastedData(iotDeviceOld,jarrData.toString(), devUpdateddate);
 
         // 发送告警消息
-        String devBid = iotDeviceOld.getDevBid();
-        JSONObject jsonObject = JSONObject.from(iotYfqxzdataList);
-        warnService.processWarningReportData(iotDeviceOld, jsonObject);
+        warnService.processWarningReportData(iotDeviceOld, ext);
 
         //预警
         //云飞 气象站 和 墒情站  都从这里进数据

src/main/java/com/yunfeiyun/agmp/iots/warn/mapper/IotWarnBussinessMapper.java

@@ -11,20 +11,20 @@ public interface IotWarnBussinessMapper {
     /**
      * 查询设备+配置的重复次数
      *
-     * @param devId
+     * @param devBid
      * @param configId
      * @return
      */
-    IotWarncount selectIotWarnCountByDevAndConfig(@Param("devId") String devId, @Param("wcBid") String configId);
+    IotWarncount selectIotWarnCountByDevAndConfig(@Param("devBid") String devBid, @Param("wcBid") String configId);
 
     /**
      * 增加重复次数
      *
-     * @param devId
+     * @param devBid
      * @param configId
      * @return
      */
-    int incrementReCount(@Param("devId") String devId, @Param("wcBid") String configId);
+    int incrementReCount(@Param("devBid") String devBid, @Param("wcBid") String configId);
 
 
     /**
@@ -41,11 +41,11 @@ public interface IotWarnBussinessMapper {
      */
     int insertWarnRecord(IotWarnlog iotWarnlog);
 
-    int updateIncrementReCount(@Param("devId") String devId, @Param("wcBid") String configId);
+    int updateIncrementReCount(@Param("devBid") String devBid, @Param("wcBid") String configId);
 
     int insertIncrementReCount(IotWarncount iotWarncount);
 
     List<WarnConfigInfo> selectIotWarnConfigInfoList(WarnConfigInfo warnConfigInfo);
 
-    int resetReCountByDevIdAndConfigId(@Param("devId") String devId, @Param("wcBid") String configId);
+    int resetReCountByDevIdAndConfigId(@Param("devBid") String devBid, @Param("wcBid") String configId);
 }

src/main/java/com/yunfeiyun/agmp/iots/warn/model/WarnConfigInfo.java

@@ -9,7 +9,7 @@ public class WarnConfigInfo {
     private String wcLevel;
     private String wcTouchtype;
     private String wcCondition;
-    private String wcRepeatnum;
+    private int wcRepeatnum;
     private String wiBid;
     private String wiAddress;
     private String wiCode;

src/main/java/com/yunfeiyun/agmp/iots/warn/model/WarnResult.java

@@ -1,5 +1,6 @@
 package com.yunfeiyun.agmp.iots.warn.model;
 
+import com.yunfeiyun.agmp.iot.common.domain.IotBaseEntity;
 import com.yunfeiyun.agmp.iot.common.domain.IotWarnconfig;
 import lombok.Data;
 
@@ -7,7 +8,7 @@ import lombok.Data;
  * 预警结果，不是对应数据库层面的
  */
 @Data
-public class WarnResult {
+public class WarnResult extends IotBaseEntity {
     /**
      * 是否触发预警
      */

src/main/java/com/yunfeiyun/agmp/iots/warn/service/IotWarnBussinessService.java

@@ -36,7 +36,7 @@ public class IotWarnBussinessService {
             log.info("从缓存中获取到重复次数: {}", count);
             return count;
         }
-        IotWarncount iotWarncount = iotWarncountMapper.selectIotWarnCountByDevAndConfig(devId, configId);
+        IotWarncount iotWarncount  = iotWarncountMapper.selectIotWarnCountByDevAndConfig(devId, configId);
         if (iotWarncount == null) {
             log.warn("数据库中未找到设备ID: {}, 配置ID: {} 的重复次数数据", devId, configId);
             return null;

src/main/java/com/yunfeiyun/agmp/iots/warn/service/WarnService.java

@@ -8,6 +8,7 @@ import com.yunfeiyun.agmp.iot.common.constant.devicetype.IotDeviceDictEnum;
 import com.yunfeiyun.agmp.iot.common.constant.devicetype.IotDeviceTypeLv1Enum;
 import com.yunfeiyun.agmp.iot.common.domain.IotDevice;
 import com.yunfeiyun.agmp.iot.common.domain.IotDevicefactor;
+import com.yunfeiyun.agmp.iot.common.domain.IotWarnconfig;
 import com.yunfeiyun.agmp.iot.common.enums.EnumWarnRuleOp;
 import com.yunfeiyun.agmp.iots.service.IIotDevicefactorService;
 import com.yunfeiyun.agmp.iots.warn.model.WarnConfigInfo;
@@ -15,6 +16,7 @@ import com.yunfeiyun.agmp.iots.warn.model.WarnResult;
 import com.yunfeiyun.agmp.iots.warn.util.CompareUtil;
 import com.yunfeiyun.agmp.iots.warn.util.WarnMessageBuilderUtil;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.stereotype.Service;
@@ -62,8 +64,8 @@ public class WarnService {
     /**
      * 统一处理上报数据
      *
-     * @param devId 设备id
      * @param data  上报的数据对象
+     * param ext   上报的原始数据
      */
     public void processWarningReportData(IotDevice iotDevice, JSONObject data) {
         //转异步处理
@@ -74,16 +76,15 @@ public class WarnService {
     /**
      * 统一处理上报数据：异步处理
      *
-     * @param devId 设备id
      * @param data  上报的数据对象
      */
     private void processWarningReportDataSyn(IotDevice iotDevice, JSONObject data) {
-        String devId = iotDevice.getDevBid();
+        String devBid = iotDevice.getDevBid();
         String devtypeBid = iotDevice.getDevtypeBid();
         String devClass =  IotDeviceDictEnum.getLv1CodeByCode(devtypeBid);
         IotDeviceTypeLv1Enum iotDeviceTypeLv1Enum = IotDeviceTypeLv1Enum.findEnumByCode(devClass);
         if (iotDeviceTypeLv1Enum == null) {
-            log.error("[设备告警] 设备大类不存在，devId:{}, devtypeBid:{}", devId, devtypeBid);
+            log.error("[设备告警] 设备大类不存在，devBid:{}, devtypeBid:{}", devBid, devtypeBid);
             return;
         }
 
@@ -91,55 +92,38 @@ public class WarnService {
             // 获取该设备有哪些告警配置
             WarnConfigInfo warnConfigInfo = new WarnConfigInfo();
             warnConfigInfo.setTid(iotDevice.getTid());
-            warnConfigInfo.setDevBid(devId);
+            warnConfigInfo.setDevBid(devBid);
 
             Map<String, List<WarnConfigInfo>> configMap = iotWarnBussinessService.selectIotWarnConfigInfoMap(warnConfigInfo);
             //配置一个个检查
             for(Map.Entry<String, List<WarnConfigInfo>> entry : configMap.entrySet()) {
                 List<WarnConfigInfo> configList = entry.getValue();
                 WarnResult warnResult = null;
-                switch (iotDeviceTypeLv1Enum) {
-                    case QXZ: {
-                        warnResult = comparableQxzReportData(iotDevice, configList, data);
-                        break;
-                    }
+                try{
+                    switch (iotDeviceTypeLv1Enum) {
+                        case QXZ: {
+                            warnResult = comparableQxzReportData(iotDevice, configList, data);
+                            break;
+                        }
 //                    case SQZ: {
-//                        warnResult = comparableSqzReportData(devId, config, data);
+//                        warnResult = comparableSqzReportData(devBid, config, data);
 //                        break;
 //                    }
 //                    case "病虫害": {
-//                        warnResult = comparableBchReportData(devId, config, data);
+//                        warnResult = comparableBchReportData(devBid, config, data);
 //                        break;
 //                    }
-                    default:
-                        break;
+                        default:
+                            break;
+                    }
+                }catch (Exception e){
+                    log.error("[设备告警] 设备上报数据 异常，devBid:{}, config:{}, data:{}", devBid, configList, data, e);
                 }
+
                 if (warnResult != null) {
                     handleWarnRecord(warnResult);
                 }
             }
-//            for (Object config : objects) {
-//                WarnResult warnResult = null;
-//                switch (iotDeviceTypeLv1Enum) {
-//                    case QXZ: {
-//                        warnResult = comparableQxzReportData(devId, config, data);
-//                        break;
-//                    }
-//                    case SQZ: {
-//                        warnResult = comparableSqzReportData(devId, config, data);
-//                        break;
-//                    }
-////                    case "病虫害": {
-////                        warnResult = comparableBchReportData(devId, config, data);
-////                        break;
-////                    }
-//                    default:
-//                        break;
-//                }
-//                if (warnResult != null) {
-//                    handleWarnRecord(warnResult);
-//                }
-//            }
         }, threadPoolTaskExecutor);
     }
 
@@ -156,160 +140,156 @@ public class WarnService {
     /**
      * 根据设备id获取策略，可能有多个
      *
-     * @param devId
+     * @param devBid
      * @return
      */
-    List<Object> getConfigByDevId(String devId) {
+    List<Object> getConfigByDevId(String devBid) {
 
         return null;
     }
 
-    /**
-     * 【气象站】比较该设备上报的要素和配置是否达到预警条件
-     *
-     * @param devId      设备id
-     * @param config     对应的配置
-     * @param jsonObject 上报的数据
-     */
-    WarnResult comparableQxzReportData(IotDevice iotDevice, List<WarnConfigInfo> configList, JSONObject jsonObject) {
+    private Map<String, String> getQxzCurrentValueMap(IotDevice iotDevice, JSONArray jsonArray) {
         String devCode = iotDevice.getDevCode();
-        String devBid = iotDevice.getDevBid();
-        WarnConfigInfo configInfo = configList.get(0);
-        String wcCondition = configInfo.getWcCondition();
-        Map<String, String> elementMap = new HashMap<>();
-        JSONArray jsonArray = JSONArray.from(jsonObject);
         BigDecimal errorValue = new BigDecimal("-99");
+        Map<String, String> currentValueMap = new HashMap<>();
         for(Object obj : jsonArray) {
-            JSONObject ob = (JSONObject) obj;
+            JSONObject ob;
+            try {
+                ob = JSONObject.from(obj);
+            } catch (Exception e) {
+                log.error("[设备告警] 数据类型转换异常，devCode:{}, obj:{}", devCode, obj);
+                continue;
+            }
+
             String eNum = ob.getString("eNum");
             String eName = ob.getString("eName");
             String eKey = ob.getString("eKey");
             String address = ElementFormatUtil.getAddress(eName, eKey);
             String key = address + eNum;
+
             BigDecimal eValue = null;
-            try{
+            try {
                 eValue = ob.getBigDecimal("eValue");
-                if(eValue.compareTo(errorValue) <= 0){
+                if (eValue.compareTo(errorValue) <= 0) {
                     throw new Exception();
                 }
-            }catch (Exception e){
-                log.error("[设备告警] 数据类型转换异常，devId:{}, eNum:{}, eName:{}, eKey:{}", devCode, eNum, eName, eKey);
+            } catch (Exception e) {
+                log.error("[设备告警] 数据类型转换异常，devCode:{}, eNum:{}, eName:{}, eKey:{}", devCode, eNum, eName, eKey);
                 continue;
             }
-            elementMap.put(key, String.valueOf(eValue));
+            currentValueMap.put(key, String.valueOf(eValue));
         }
+        return currentValueMap;
+    }
+
+    /**
+     * 【气象站】检测该设备上报的要素和配置是否达到预警条件
+     * @param config
+     * @param factorMap
+     * @param currentValueMap
+     * @param devCode
+     * @return
+     */
+    private Map<String, Object> getQxzCondition(WarnConfigInfo config, Map<String, IotDevicefactor> factorMap,
+                                     Map<String, String> currentValueMap, String devCode) {
+        String targetValue = config.getWiValue();
+        String wiAddress = config.getWiAddress();
+        String wiCode = config.getWiCode();
+        String key = wiAddress + wiCode;
+
+        if (!currentValueMap.containsKey(key)) {
+            return null;
+        }
+
+        String wiName = config.getWiName();
+        if(factorMap.containsKey(key)){
+            IotDevicefactor iotDevicefactor = factorMap.get(key);
+            String displayname = iotDevicefactor.getDfDisplayname();
+            if(StringUtils.isNotEmpty(displayname)){
+                wiName = displayname;
+            }
+        }
+        String wiUnit = config.getWiUnit();
+
+        String currentValue = currentValueMap.get(key);
+        String expression = config.getWiExpression();
+        EnumWarnRuleOp warnRuleOp = EnumWarnRuleOp.findEnumByCode(expression);
+        String ruleName = warnRuleOp.getName();
+
+        boolean tempSuccess = CompareUtil.comp(currentValue, expression, targetValue);
+        String warnMessage = WarnMessageBuilderUtil.buildQxzWarningMessage(
+                "气象站设备", devCode, wiName, currentValue, wiUnit, ruleName, targetValue);
+
+        Map<String, Object> map = new HashMap<>();
+        map.put("warnMessage", warnMessage);
+        map.put("tempSuccess", tempSuccess);
+        return map;
+    }
+
+    /**
+     * 【气象站】比较该设备上报的要素和配置是否达到预警条件
+     *
+     * @param devBid      设备id
+     * @param config     对应的配置
+     * @param jsonObject 上报的数据
+     */
+    private WarnResult comparableQxzReportData(IotDevice iotDevice, List<WarnConfigInfo> configList, JSONObject jsonObject) {
+        String devCode = iotDevice.getDevCode();
+        String devBid = iotDevice.getDevBid();
+        WarnConfigInfo configInfo = configList.get(0);
+        String wcCondition = configInfo.getWcCondition();
+        JSONArray jsonArray = jsonObject.getJSONArray("data");
+
+        Map<String, String> currentValueMap = getQxzCurrentValueMap(iotDevice, jsonArray);
+
+        IotWarnconfig iotWarnconfig = new IotWarnconfig();
+        BeanUtils.copyProperties(configInfo, iotWarnconfig);
+
+        WarnResult warnResult = new WarnResult();
+        warnResult.setMessageId(warnResult.getUUId());
+        warnResult.setDevId(devBid);
+        warnResult.setTid(iotDevice.getTid());
+        warnResult.setConfigId(configInfo.getWcBid());
+        warnResult.setReportData(jsonObject.toJSONString());
+        warnResult.setTargetReCount(configInfo.getWcRepeatnum());
+        warnResult.setDevtypeBid(iotDevice.getDevtypeBid());
+        warnResult.setConfig(iotWarnconfig);
+        warnResult.setTriggered(false);
+
         Map<String, IotDevicefactor> factorMap = getDevicefactorMap(devBid);
         String message = null;
         StringBuilder messageBuilder = new StringBuilder();
         for (WarnConfigInfo config : configList) {
-            String currentValue = config.getWiValue();
-            String wiAddress = config.getWiAddress();
-            String wiCode = config.getWiCode();
-            String key = wiAddress + wiCode;
-            String wiName = config.getWiName();
-            if(factorMap.containsKey(key)){
-                IotDevicefactor iotDevicefactor = factorMap.get(key);
-                String displayname = iotDevicefactor.getDfDisplayname();
-                if(StringUtils.isNotEmpty(displayname)){
-                    wiName = displayname;
-                }
-            }
-            String wiUnit = config.getWiUnit();
-            if (!elementMap.containsKey(key)) {
+            Map<String, Object> conditionResult = getQxzCondition(config, factorMap, currentValueMap, devCode);
+            if (conditionResult == null) {
                 continue;
             }
-            String targetValue = elementMap.get(key);
-            String expression = config.getWiExpression();
-            EnumWarnRuleOp warnRuleOp = EnumWarnRuleOp.findEnumByCode(expression);
-            String ruleName = warnRuleOp.getName();
-            String warnMessage = WarnMessageBuilderUtil.buildQxzWarningMessage(
-                    "气象站设备", devCode, wiName, currentValue, wiUnit, ruleName, targetValue);
-
-            boolean tempSuccess = CompareUtil.comp(currentValue, expression, targetValue);
+            String warnMessage = (String) conditionResult.get("warnMessage");
+            boolean tempSuccess = (boolean) conditionResult.get("tempSuccess");
+
             if ("0".equals(wcCondition)) {
                 if (tempSuccess) {
                     message = warnMessage;
-                    return new WarnResult(true, message);
+                    warnResult.setMessage(message);
+                    warnResult.setTriggered(true);
+                    return warnResult;
                 }
             } else {
                 if (!tempSuccess) {
-                    return new WarnResult(false, null);
+                    return warnResult;
                 }
                 messageBuilder.append(warnMessage);
             }
         }
 
         if(messageBuilder.length() > 0){
-            return new WarnResult(true, messageBuilder.toString());
+            warnResult.setMessage(messageBuilder.toString());
+            warnResult.setTriggered(true);
+            return warnResult;
         }
-        return new WarnResult(false, null);
+        return warnResult;
     }
 
-    /**
-     * 处理单一指标是否达到预警条件。
-     *
-     * @param devId      设备ID
-     * @param config     配置对象
-     * @param jsonObject 上报的数据
-     * @return WarnResult 包含是否触发告警的信息
-     */
-//    public WarnResult comparableQxzSingleIndicator(String devId, Object config, JSONObject jsonObject) {
-//        //从config取出需要匹配的要素列表 todo
-//        List<Object> items = new ArrayList<>();
-//        //循环要素，一个个对比
-//        for (Object item : items) {
-//            // 取出来需要比较的要素 todo
-//            String comparableItem = "item.getField";
-//            //要对比的目标值 todo
-//            String targetValue = "item.targetValue";
-//            //取出来当前的最新值 todo
-//            String currentValue = "jsonObject.get[item.getField]";
-//            // 字段表达式比较
-//            boolean result = CompareUtil.comp(currentValue, EnumWarnRuleOp.EQUAL.getCode(), targetValue);
-//            if (result) {
-//                String message = WarnMessageBuilderUtil.buildQxzWarningMessage("气象站设备", devId, comparableItem, Double.parseDouble(currentValue), "℃", "超过", targetValue);
-//                return new WarnResult(true, message);
-//            }
-//        }
-//        return new WarnResult(false, null);
-//    }
-
-    /**
-     * 处理多个指标是否同时满足预警条件。
-     *
-     * @param devId      设备ID
-     * @param config     告警规则
-     * @param jsonObject 上报的数据
-     * @return WarnResult 包含是否触发告警的信息
-     */
-//    public WarnResult comparableQxzMultipleIndicators(String devId, Object config, JSONObject jsonObject) {
-//        //从config取出需要匹配的要素列表 todo
-//        List<Object> items = new ArrayList<>();
-//        int successCount = 0;
-//        StringBuilder messages = new StringBuilder();
-//        //循环要素，一个个对比
-//        for (Object item : items) {
-//            // 取出来需要比较的要素 todo
-//            String comparableItem = "item.getField";
-//            //要对比的目标值 todo
-//            String targetValue = "item.targetValue";
-//            //取出来当前的最新值 todo
-//            String currentValue = "jsonObject.get[item.getField]";
-//            // 字段表达式比较
-//            boolean result = CompareUtil.comp(currentValue, EnumWarnRuleOp.EQUAL.getCode(), targetValue);
-//            if (result) {
-//                messages.append(WarnMessageBuilderUtil.buildQxzWarningMessage("气象站设备", devId, comparableItem, Double.parseDouble(currentValue), "℃", "超过", targetValue));
-//                successCount += 1;
-//            }
-//        }
-//
-//        if (successCount == items.size()) {
-//            return new WarnResult(true, messages.toString());
-//        } else {
-//            return new WarnResult(false, null);
-//        }
-//    }
-
 
     /**
      * 【墒情站】比较该设备上报的要素和配置是否达到预警条件，暂不实现，预留

src/main/resources/mapper/IotWarnBusinessMapper.xml

@@ -77,8 +77,8 @@
         select  * from IotWarncount where devBid=#{devBid} and wcBid=#{wcBid};
     </select>
 
-    <select id="selectIotWarnConfigInfoList" parameterType="WarnConfigInfo" resultType="com.yunfeiyun.agmp.iots.warn.model.WarnConfigInfo">
-        SELECT wc.wcBid, wc.wcName, wc.wcLevel, wc.wcTouchtype, wc.wcCondition, wc.wcRepeatnum, wi.wiBid, wi.wiAddress,
+    <select id="selectIotWarnConfigInfoList" parameterType="com.yunfeiyun.agmp.iots.warn.model.WarnConfigInfo" resultType="com.yunfeiyun.agmp.iots.warn.model.WarnConfigInfo">
+        SELECT wc.*, wi.wiBid, wi.wiAddress,
             wi.wiCode, wi.wiName, wi.wiUnit, wi.wiExpression, wi.wiValue
         FROM IotWarnconfig AS wc
             LEFT JOIN IotWarnindicator AS wi ON wi.wcBid = wc.wcBid