feat: 场景联动显示告警详情

芯火源
1 parent 89167b44
Showing 4 changed files with 148 additions and 93 deletions
common/data/src/main/java/org/thingsboard/server/common/data/yunteng/constant/FastIotConstants.java
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/yunteng/scene/ReactState.java
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/yunteng/scene/TriggerState.java
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/yunteng/utils/TriggerRuleState.java
--- a/common/data/src/main/java/org/thingsboard/server/common/data/yunteng/constant/FastIotConstants.java
View file @fc0d47a
+++ b/common/data/src/main/java/org/thingsboard/server/common/data/yunteng/constant/FastIotConstants.java
View file @fc0d47a
@@ -175,4 +175,17 @@ public interface FastIotConstants {
     /**RPC单项双向*/
     public static String ONEWAY = "oneway";
   }
+  class Alarm{
+
+    /**遥测指标标识符*/
+    public static String KEY = "key";
+    /**遥测指标值*/
+    public static String VALUE = "value";
+    /**触发器逻辑关系*/
+    public static String PREDICATE = "logic";
+    /**触发器*/
+    public static String TRIGGER = "trigger";
+    /**执行条件*/
+    public static String CONDITION = "condition";
+  }
 }
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/yunteng/scene/ReactState.java
View file @fc0d47a
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/yunteng/scene/ReactState.java
View file @fc0d47a
@@ -89,24 +89,27 @@ class ReactState {
    *
    * @param ctx
    * @param msg 设备推送的遥测数据
+   * @param prefixId 运算规则ID，例如：触发器、执行条件
    * @param deviceId 遥测数据的来源设备的TB设备ID
    * @throws ExecutionException
    * @throws InterruptedException
    */
-  public void process(TbContext ctx, TbMsg msg,  String prefixId,String deviceId)
+  public void process(TbContext ctx, TbMsg msg, String prefixId, String deviceId)
       throws ExecutionException, InterruptedException {
-    StringBuilder detail = new StringBuilder();
+    /** 场景联动告警详情 */
+    ObjectNode detail = JacksonUtil.newObjectNode();
     if (actions == null) {
       ctx.tellSuccess(msg);
     }
+    /** 1、单个触发器内只要有1个设备满足条件即为true 2、多个触发器只要有1个触发器满足条件即为true */
     AtomicBoolean triggerMatched = new AtomicBoolean(true);
     Optional.ofNullable(triggers)
         .ifPresent(
-            t -> {
+            all -> {
               triggerMatched.set(false);
-              t.forEach(
+              all.forEach(
                   trigger -> {
                     ScopeEnum entityType = trigger.getEntityType();
                     List<String> trifggerDevices = trigger.getEntityId();
@@ -119,24 +122,30 @@ class ReactState {
                     triggerMatched.set(
                         trifggerDevices.stream()
                             .anyMatch(
-                                id -> {
+                                devId -> {
                                   TriggerState triggerState =
-                                      getOrCreateTriggerState(trigger, tkProjectId, id);
+                                      getOrCreateTriggerState(trigger, tkProjectId, devId);
                                   if (triggerState == null) {
                                     return false;
                                   }
                                   try {
                                     boolean fresh = false;
-                                    if(trigger.getId().equals(prefixId)&&msg.getOriginator().getId().toString().equals(id)){
-                                      fresh=true;
+                                    if (trigger.getId().equals(prefixId)
+                                        && msg.getOriginator().getId().toString().equals(devId)) {
+                                      fresh = true;
                                     }
-                                    boolean result = triggerState.process(ctx, msg,fresh);
-                                    log.error(String.format("触发器【%s】刷新【%s】结果【%s】触发器设备【%s】数据设备【%s】数据内容【%s】",trigger.getId(),fresh,result,id,msg.getOriginator(),msg.getData()));
-                                    if (result) {
-                                      detail.append(
-                                          triggerState.getAlarmDetails() == null
-                                              ? ""
-                                              : triggerState.getAlarmDetails());
+                                    ObjectNode result = triggerState.process(ctx, msg, fresh);
+                                    log.error(
+                                        String.format(
+                                            "触发器【%s】刷新【%s】结果【%s】触发器设备【%s】数据设备【%s】数据内容【%s】",
+                                            trigger.getId(),
+                                            fresh,
+                                            result,
+                                            devId,
+                                            msg.getOriginator(),
+                                            msg.getData()));
+                                    if (!result.isEmpty()) {
+                                      detail.set(FastIotConstants.Alarm.TRIGGER,result);
                                       return true;
                                     } else if (currentAlarms.containsKey(deviceId)) {
                                       // 清除设备告警
@@ -155,51 +164,65 @@ class ReactState {
                   });
             });
-    /** 执行条件的所有设备都满足才为true */
+    /** 1、单个执行条件内全部设备满足条件才为true 2、多个执行条件全部执行条件满足条件才为true */
     AtomicBoolean conditionMatched = new AtomicBoolean(true);
     Optional.ofNullable(conditions)
-            .ifPresent(
-                    t -> {
-                      t.forEach(
-                              condition -> {
-                                ScopeEnum entityType = condition.getEntityType();
-                                List<String> conditionDevices = condition.getEntityId();
-                                String tkProjectId = condition.getDeviceProfileId();
-                                if (ScopeEnum.ALL.equals(entityType)) {
-                                  conditionDevices =
-                                          ytDeviceService.findTbDeviceIdsByDeviceProfileId(
-                                                  tkProjectId, condition.getTenantId());
-                                }
-                                conditionMatched.set(
-                                        !conditionDevices.stream()
-                                                .anyMatch(
-                                                        id -> {
-                                                          TriggerState conditionState =
-                                                                  getOrCreateConditionState(condition, tkProjectId, id);
-                                                          try {
-                                                            boolean fresh = false;
-                                                            if(msg.getOriginator().getId().toString().equals(id)){
-                                                              fresh=true;
-                                                            }
-                                                            boolean result = conditionState.process(ctx, msg,fresh);
-                                                            log.warn(String.format("执行器【%s】刷新【%s】结果【%s】执行器设备【%s】数据设备【%s】数据内容【%s】",condition.getId(),fresh,result,id,msg.getOriginator(),msg.getData()));
-                                                            return !result;
-                                                          } catch (ExecutionException e) {
-                                                            throw new RuntimeException(e);
-                                                          } catch (InterruptedException e) {
-                                                            throw new RuntimeException(e);
-                                                          }
-                                                        }));
-                              });
-                    });
+        .ifPresent(
+            all -> {
+              conditionMatched.set(
+                  all.stream()
+                      .allMatch(
+                          condition -> {
+                            ScopeEnum entityType = condition.getEntityType();
+                            List<String> conditionDevices = condition.getEntityId();
+                            String tkProjectId = condition.getDeviceProfileId();
+                            if (ScopeEnum.ALL.equals(entityType)) {
+                              conditionDevices =
+                                  ytDeviceService.findTbDeviceIdsByDeviceProfileId(
+                                      tkProjectId, condition.getTenantId());
+                            }
+
+                            return conditionDevices.stream()
+                                .allMatch(
+                                    devId -> {
+                                      TriggerState conditionState =
+                                          getOrCreateConditionState(condition, tkProjectId, devId);
+                                      try {
+                                        boolean fresh = false;
+                                        if (msg.getOriginator().getId().toString().equals(devId)) {
+                                          fresh = true;
+                                        }
+                                        ObjectNode result = conditionState.process(ctx, msg, fresh);
+                                        log.warn(
+                                            String.format(
+                                                "执行器【%s】刷新【%s】结果【%s】执行器设备【%s】数据设备【%s】数据内容【%s】",
+                                                condition.getId(),
+                                                fresh,
+                                                result,
+                                                devId,
+                                                msg.getOriginator(),
+                                                msg.getData()));
+                                        if (!result.isEmpty()) {
+                                          detail.set(FastIotConstants.Alarm.CONDITION,result);
+                                          return true;
+                                        }
+                                      } catch (ExecutionException e) {
+                                        throw new RuntimeException(e);
+                                      } catch (InterruptedException e) {
+                                        throw new RuntimeException(e);
+                                      }
+                                      return false;
+                                    });
+                          }));
+            });
     if (triggerMatched.get() && conditionMatched.get()) {
-      log.error(String.format("设备【%s】的消息内容【%s】触发动作",deviceId,msg.getData()));
+      log.error(String.format("设备【%s】的消息内容【%s】触发动作", deviceId, msg.getData()));
       for (TkDoActionEntity item : actions) {
         if (ActionTypeEnum.MSG_NOTIFY.equals(item.getOutTarget())) {
-          noticeMsg(ctx, msg, item, deviceId, detail.toString(), msg.getTs());
+          noticeMsg(ctx, msg, item, deviceId, detail, msg.getTs());
         } else {
-          pushMsg(ctx, msg, item, detail.toString());
+          pushMsg(ctx, msg, item);
         }
       }
     } else {
@@ -226,7 +249,7 @@ class ReactState {
         || (trigger.getEntityType().equals(ScopeEnum.ALL)
             && trigger.getDeviceProfileId().equals(profileId))) {
       TriggerState state = createTriggerState(deviceId, trigger.getTriggerCondition());
-      log.error(String.format("新建设备【%s】的触发器",deviceId));
+      log.error(String.format("新建设备【%s】的触发器", deviceId));
       triggerState.put(cacheKey, state);
       return state;
     }
@@ -290,12 +313,12 @@ class ReactState {
     for (AlarmConditionFilter filter : rule.getCondition().getCondition()) {
       filterKeys.add(filter.getKey());
     }
-    TriggerState state = new TriggerState(deviceId, rule, filterKeys, rule.getAlarmDetails(), null);
+    TriggerState state = new TriggerState(deviceId, rule, filterKeys, null);
     return state;
   }
-  private void pushMsg(TbContext ctx, TbMsg msg, TkDoActionEntity action, String detail) {
+  private void pushMsg(TbContext ctx, TbMsg msg, TkDoActionEntity action) {
     switch (action.getOutTarget()) {
       case DEVICE_OUT:
         List<String> rpcDevices = action.getDeviceId();
@@ -354,7 +377,7 @@ class ReactState {
       TbMsg msg,
       TkDoActionEntity action,
       String deviceId,
-      String detailStr,
+      ObjectNode detail,
       long startTs) {
     DeviceId entityId = new DeviceId(UUID.fromString(deviceId));
@@ -370,12 +393,7 @@ class ReactState {
     }
     currentAlarm.setStartTs(startTs);
     currentAlarm.setEndTs(currentAlarm.getStartTs());
-    ObjectNode detailData = JacksonUtil.newObjectNode();
-    if (StringUtils.isNotEmpty(detailStr)) {
-      detailData.put("msg", detailStr);
-    }
-    detailData.put("data", JacksonUtil.toJsonNode(msg.getData()));
-    currentAlarm.setDetails(detailData);
+    currentAlarm.setDetails(detail);
     currentAlarm.setOriginator(entityId);
     currentAlarm.setTenantId(ctx.getTenantId());
     currentAlarm.setPropagate(false);
@@ -393,7 +411,7 @@ class ReactState {
     AlarmInfoDTO formData = new AlarmInfoDTO();
     formData.setDeviceName(msg.getMetaData().getData().get("deviceName"));
-    formData.setDetails(JacksonUtil.toString(detailData));
+    formData.setDetails(JacksonUtil.toString(detail));
     formData.setType(currentAlarm.getType());
     formData.setCreateTs(currentAlarm.getCreatedTime());
     formData.setStartTs(currentAlarm.getStartTs());
@@ -408,16 +426,19 @@ class ReactState {
   private void clearAlarm(TbContext ctx, TbMsg msg, String deviceId, String key)
       throws ExecutionException, InterruptedException {
     TriggerState clearState = getOrCreateClearState(deviceId, key);
-    if (clearState != null && clearState.process(ctx, msg,true)) {
-      ctx.getAlarmService()
-          .clearAlarmForResult(
-              ctx.getTenantId(),
-              currentAlarms.get(deviceId).getId(),
-              null,
-              System.currentTimeMillis());
-      ytDeviceService.freshAlarmStatus(new DeviceId(UUID.fromString(deviceId)), 0);
-      alarmMsg(ctx, msg, currentAlarms.get(deviceId), "Alarm Cleared");
-      currentAlarms.remove(deviceId);
+    if(clearState !=null){
+      ObjectNode clearResult = clearState.process(ctx, msg, true);
+      if(!clearResult.isEmpty()){
+        ctx.getAlarmService()
+                .clearAlarmForResult(
+                        ctx.getTenantId(),
+                        currentAlarms.get(deviceId).getId(),
+                        clearResult,
+                        System.currentTimeMillis());
+        ytDeviceService.freshAlarmStatus(new DeviceId(UUID.fromString(deviceId)), 0);
+        alarmMsg(ctx, msg, currentAlarms.get(deviceId), "Alarm Cleared");
+        currentAlarms.remove(deviceId);
+      }
     }
   }
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/yunteng/scene/TriggerState.java
View file @fc0d47a
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/yunteng/scene/TriggerState.java
View file @fc0d47a
@@ -13,6 +13,7 @@
  */
 package org.thingsboard.rule.engine.yunteng.scene;
+import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.google.gson.JsonParser;
 import java.util.*;
 import java.util.concurrent.ExecutionException;
@@ -32,6 +33,7 @@ import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.kv.AttributeKvEntry;
 import org.thingsboard.server.common.data.kv.KvEntry;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
+import org.thingsboard.server.common.data.yunteng.utils.JacksonUtil;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.TbMsgMetaData;
 import org.thingsboard.server.common.msg.session.SessionMsgType;
@@ -50,18 +52,14 @@ class TriggerState {
   private final DynamicPredicateValueCtx dynamicPredicateValueCtx;
   private DataSnapshot latestValues;
+  /**触发器的遥测指标*/
   private final Set<AlarmConditionFilterKey> entityKeys;
-  private final String alarmDetails;
-  @Getter
-  /** 触发条件执行结果 */
-  private boolean ruleMatched = false;
   TriggerState(
       String originator,
       AlarmRule rule,
       Set<AlarmConditionFilterKey> filterKeys,
-      String alarmDetails,
       DynamicPredicateValueCtx dynamicPredicateValueCtx) {
     this.originator = originator;
@@ -70,12 +68,12 @@ class TriggerState {
         new TriggerRuleState(
             rule.getCondition(), filterKeys, new PersistedAlarmRuleState(), rule.getSchedule());
     this.entityKeys = filterKeys;
-    this.alarmDetails = alarmDetails;
   }
-  public boolean process(TbContext ctx, TbMsg msg,boolean needFresh) throws ExecutionException, InterruptedException {
+  public ObjectNode process(TbContext ctx, TbMsg msg, boolean needFresh)
+      throws ExecutionException, InterruptedException {
     if (!needFresh) {
-      return ruleMatched;
+      return ruleState.getDetailInform();
     }
     if (latestValues == null) {
       latestValues = fetchLatestValues(ctx, originator);
@@ -89,35 +87,33 @@ class TriggerState {
     }
     if (update != null && update.hasUpdate()) {
-      ruleMatched = createOrClearAlarms(ctx, msg, latestValues, update, TriggerRuleState::eval);
-      return ruleMatched;
+      return createOrClearAlarms(ctx, msg, latestValues, update, TriggerRuleState::eval);
     }
-    return false;
+    return JacksonUtil.newObjectNode();
   }
-  public boolean process(TbContext ctx, long ts) throws ExecutionException, InterruptedException {
+  public ObjectNode process(TbContext ctx, long ts) throws ExecutionException, InterruptedException {
     return createOrClearAlarms(
         ctx, null, ts, null, (alarmState, tsParam) -> alarmState.eval(tsParam, latestValues));
   }
-  public <T> boolean createOrClearAlarms(
+  public <T> ObjectNode createOrClearAlarms(
       TbContext ctx,
       TbMsg msg,
       T data,
       SnapshotUpdate update,
       BiFunction<TriggerRuleState, T, AlarmEvalResult> evalFunction) {
-    boolean stateUpdate = false;
     if (!validateUpdate(update, ruleState)) {
-      return false;
+      return JacksonUtil.newObjectNode();
     }
+//    ruleState.getCondition().
     AlarmEvalResult evalResult = evalFunction.apply(ruleState, data);
     if (AlarmEvalResult.TRUE.equals(evalResult)) {
-      stateUpdate = true;
-      clearAlarmState(stateUpdate, ruleState);
+      clearAlarmState(true, ruleState);
     } else if (AlarmEvalResult.FALSE.equals(evalResult)) {
-      clearAlarmState(stateUpdate, ruleState);
+      clearAlarmState(false, ruleState);
     }
-    return stateUpdate;
+    return ruleState.getDetailInform();
   }
   public boolean clearAlarmState(boolean stateUpdate, TriggerRuleState state) {
@@ -128,6 +124,12 @@ class TriggerState {
     return stateUpdate;
   }
+  /**
+   * 快照中有更新的遥测指标是否包含在触发器中
+   * @param update 遥测数据的快照
+   * @param state 触发器状态
+   * @return
+   */
   public boolean validateUpdate(SnapshotUpdate update, TriggerRuleState state) {
     if (update != null) {
       // Check that the update type and that keys match.
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/yunteng/utils/TriggerRuleState.java
View file @fc0d47a
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/yunteng/utils/TriggerRuleState.java
View file @fc0d47a
@@ -3,7 +3,9 @@
  */
 package org.thingsboard.rule.engine.yunteng.utils;
+import com.fasterxml.jackson.databind.node.ObjectNode;
 import lombok.Data;
+import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.rule.engine.profile.AlarmEvalResult;
 import org.thingsboard.rule.engine.profile.DataSnapshot;
@@ -12,6 +14,8 @@ import org.thingsboard.rule.engine.profile.EntityKeyValue;
 import org.thingsboard.rule.engine.profile.state.PersistedAlarmRuleState;
 import org.thingsboard.server.common.data.device.profile.*;
 import org.thingsboard.server.common.data.query.*;
+import org.thingsboard.server.common.data.yunteng.constant.FastIotConstants;
+import org.thingsboard.server.common.data.yunteng.utils.JacksonUtil;
 import org.thingsboard.server.common.msg.tools.SchedulerUtils;
 import java.time.Instant;
@@ -33,6 +37,8 @@ public class TriggerRuleState {
     private final DynamicPredicateValueCtx dynamicPredicateValueCtx =null;
+    /**告警触发详情*/
+    private ObjectNode detailInform = null;
@@ -46,6 +52,7 @@ public class TriggerRuleState {
             this.state = new PersistedAlarmRuleState(0L, 0L, 0L);
         }
         this.spec = getSpec(condition);
+        this.detailInform = JacksonUtil.newObjectNode();
     }
@@ -279,7 +286,11 @@ public class TriggerRuleState {
             if (value == null) {
                 return false;
             }
+            detailInform.put(FastIotConstants.Alarm.KEY,filter.getKey().getKey());
             eval = eval && eval(data, value, filter.getPredicate(), filter);
+            if(!eval){
+                detailInform.removeAll();
+            }
         }
         return eval;
     }
@@ -320,6 +331,8 @@ public class TriggerRuleState {
     }
     private boolean evalComplexPredicate(DataSnapshot data, EntityKeyValue ekv, ComplexFilterPredicate predicate, AlarmConditionFilter filter) {
+        detailInform.put(FastIotConstants.Alarm.VALUE,ekv.getDataType().toString());
+        detailInform.put(FastIotConstants.Alarm.PREDICATE,predicate.getOperation().toString());
         switch (predicate.getOperation()) {
             case OR:
                 for (KeyFilterPredicate kfp : predicate.getPredicates()) {
@@ -349,6 +362,8 @@ public class TriggerRuleState {
         if (predicateValue == null) {
             return false;
         }
+        detailInform.put(FastIotConstants.Alarm.VALUE,predicateValue);
+        detailInform.put(FastIotConstants.Alarm.PREDICATE,predicate.getOperation().toString());
         switch (predicate.getOperation()) {
             case EQUAL:
                 return val.equals(predicateValue);
@@ -368,6 +383,8 @@ public class TriggerRuleState {
         if (predicateValue == null) {
             return false;
         }
+        detailInform.put(FastIotConstants.Alarm.VALUE,predicateValue);
+        detailInform.put(FastIotConstants.Alarm.PREDICATE,predicate.getOperation().toString());
         switch (predicate.getOperation()) {
             case NOT_EQUAL:
                 return !val.equals(predicateValue);
@@ -399,6 +416,8 @@ public class TriggerRuleState {
             val = val.toLowerCase();
             predicateValue = predicateValue.toLowerCase();
         }
+        detailInform.put(FastIotConstants.Alarm.VALUE,predicateValue);
+        detailInform.put(FastIotConstants.Alarm.PREDICATE,predicate.getOperation().toString());
         switch (predicate.getOperation()) {
             case CONTAINS:
                 return val.contains(predicateValue);