feat(workflow): 添加巡检报警监听和机械臂控制功能

- 在ActionEnum中新增巡检报警监听相关枚举值 - 集成InspectionAlertListenService和EdgeArmService依赖 - 实现巡检报警监听开始和停止功能 - 添加机械臂移动到指定坐标点功能 - 在工作流执行器中增加报警事件类型解析逻辑 - 实现基于终端ID的边缘端IP解析和报警过滤机制 - 新增巡检报警监听工作流组件服务类 - 创建巡检报警监听状态管理服务
2026-07-23 13:52:37 +08:00 · 2026-07-23 13:52:37 +08:00 · dfdf575932
commit dfdf575932
parent 922d30632b
7 changed files with 466 additions and 2 deletions
--- a/cmvr-iot-device/src/main/java/com/cmvr/device/service/InspectionAlertListenService.java
+++ b/cmvr-iot-device/src/main/java/com/cmvr/device/service/InspectionAlertListenService.java
@ -0,0 +1,242 @@
+package com.cmvr.device.service;
+
+import com.cmvr.common.exception.GlobalException;
+import com.cmvr.device.domain.DeDeviceTerminalConfig;
+import lombok.Data;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.stereotype.Service;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+/**
+ * 巡检报警监听状态服务。
+ *
+ * <p>工作流开始监听时，根据终端ID解析终端IP，并将 IP + 报警类型保存到内存。
+ * cmvr_edge_ai 推送报警后，巡检报警入库服务通过该服务判断是否需要保存。
+ * 当前状态只在本 JVM 内有效，应用重启后会清空。</p>
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class InspectionAlertListenService
+{
+    /** 当前支持的 PPE 违规类型。 */
+    private static final Set<String> SUPPORTED_EVENT_TYPES;
+
+    static
+    {
+        Set<String> values = new LinkedHashSet<>();
+        values.add("No-Glove");
+        values.add("No-Helmet");
+        SUPPORTED_EVENT_TYPES = Collections.unmodifiableSet(values);
+    }
+
+    private final IDeDeviceTerminalConfigService terminalConfigService;
+
+    /**
+     * key 为边缘端 gRPC IP，value 为当前需要保存的报警类型集合。
+     */
+    private final ConcurrentMap<String, Set<String>> listenMap = new ConcurrentHashMap<>();
+
+    /**
+     * 开始监听指定终端的指定报警类型。
+     *
+     * @param terminalId 终端ID
+     * @param eventTypes 报警类型，支持 No-Glove、No-Helmet，可多选
+     * @return 当前终端IP下正在监听的报警类型
+     */
+    public ListenState startListen(String terminalId, Collection<String> eventTypes)
+    {
+        String grpcIp = resolveTerminalIp(terminalId);
+        Set<String> normalizedTypes = normalizeEventTypes(eventTypes, true);
+
+        Set<String> target = listenMap.computeIfAbsent(grpcIp, key -> ConcurrentHashMap.newKeySet());
+        target.addAll(normalizedTypes);
+
+        log.info("开始监听巡检报警，terminalId={}，grpcIp={}，eventTypes={}，currentTypes={}",
+                terminalId, grpcIp, normalizedTypes, target);
+        return new ListenState(terminalId, grpcIp, snapshot(target));
+    }
+
+    /**
+     * 结束监听指定终端的指定报警类型。
+     *
+     * <p>移除不存在的 IP + 类型不抛异常，只记录警告日志，避免工作流重复清理时失败。</p>
+     *
+     * @param terminalId 终端ID
+     * @param eventTypes 报警类型，支持 No-Glove、No-Helmet，可多选
+     * @return 当前终端IP下剩余正在监听的报警类型
+     */
+    public ListenState stopListen(String terminalId, Collection<String> eventTypes)
+    {
+        String grpcIp = resolveTerminalIp(terminalId);
+        Set<String> normalizedTypes = normalizeEventTypes(eventTypes, true);
+        Set<String> current = listenMap.get(grpcIp);
+
+        if (current == null)
+        {
+            log.warn("结束监听巡检报警时未找到IP监听状态，terminalId={}，grpcIp={}，eventTypes={}",
+                    terminalId, grpcIp, normalizedTypes);
+            return new ListenState(terminalId, grpcIp, Collections.emptyList());
+        }
+
+        for (String eventType : normalizedTypes)
+        {
+            if (!current.remove(eventType))
+            {
+                log.warn("结束监听巡检报警时事件类型不存在，terminalId={}，grpcIp={}，eventType={}",
+                        terminalId, grpcIp, eventType);
+            }
+        }
+
+        if (current.isEmpty())
+        {
+            listenMap.remove(grpcIp, current);
+        }
+
+        log.info("结束监听巡检报警，terminalId={}，grpcIp={}，eventTypes={}，remainingTypes={}",
+                terminalId, grpcIp, normalizedTypes, current);
+        return new ListenState(terminalId, grpcIp, snapshot(current));
+    }
+
+    /**
+     * 判断报警是否命中当前监听条件。
+     *
+     * @param grpcIp 边缘端 gRPC IP
+     * @param labels 报警标签列表，来自 payload.labels
+     * @return true 表示允许入库，false 表示忽略
+     */
+    public boolean shouldStore(String grpcIp, Collection<String> labels)
+    {
+        String normalizedIp = StringUtils.trimToNull(grpcIp);
+        if (normalizedIp == null)
+        {
+            log.warn("忽略巡检报警：grpc_ip为空，labels={}", labels);
+            return false;
+        }
+
+        Set<String> listeningTypes = listenMap.get(normalizedIp);
+        if (listeningTypes == null || listeningTypes.isEmpty())
+        {
+            log.info("忽略巡检报警：未开启该IP的报警监听，grpcIp={}，labels={}", normalizedIp, labels);
+            return false;
+        }
+
+        Set<String> alertTypes = normalizeEventTypes(labels, false);
+        for (String alertType : alertTypes)
+        {
+            if (listeningTypes.contains(alertType))
+            {
+                return true;
+            }
+        }
+
+        log.info("忽略巡检报警：事件类型未命中监听条件，grpcIp={}，labels={}，listeningTypes={}",
+                normalizedIp, labels, listeningTypes);
+        return false;
+    }
+
+    /**
+     * 查询指定终端当前监听状态，便于调试或前端展示。
+     */
+    public ListenState getListenState(String terminalId)
+    {
+        String grpcIp = resolveTerminalIp(terminalId);
+        return new ListenState(terminalId, grpcIp, snapshot(listenMap.get(grpcIp)));
+    }
+
+    private String resolveTerminalIp(String terminalId)
+    {
+        String normalizedTerminalId = StringUtils.trimToNull(terminalId);
+        if (normalizedTerminalId == null)
+        {
+            throw new GlobalException("terminalId不能为空");
+        }
+
+        DeDeviceTerminalConfig terminalConfig =
+                terminalConfigService.selectDeDeviceTerminalConfigById(normalizedTerminalId);
+        if (terminalConfig == null)
+        {
+            throw new GlobalException("未找到终端配置，terminalId=" + normalizedTerminalId);
+        }
+
+        String grpcIp = StringUtils.trimToNull(terminalConfig.getHost());
+        if (grpcIp == null)
+        {
+            throw new GlobalException("终端配置host不能为空，terminalId=" + normalizedTerminalId);
+        }
+        return grpcIp;
+    }
+
+    private Set<String> normalizeEventTypes(Collection<String> eventTypes, boolean failOnEmpty)
+    {
+        Set<String> result = new LinkedHashSet<>();
+        if (eventTypes != null)
+        {
+            for (String eventType : eventTypes)
+            {
+                String normalized = normalizeEventType(eventType);
+                if (normalized != null)
+                {
+                    result.add(normalized);
+                }
+            }
+        }
+
+        if (failOnEmpty && result.isEmpty())
+        {
+            throw new GlobalException("eventTypes不能为空，当前支持No-Glove、No-Helmet");
+        }
+        return result;
+    }
+
+    private String normalizeEventType(String eventType)
+    {
+        String value = StringUtils.trimToNull(eventType);
+        if (value == null)
+        {
+            return null;
+        }
+
+        for (String supported : SUPPORTED_EVENT_TYPES)
+        {
+            if (supported.equalsIgnoreCase(value))
+            {
+                return supported;
+            }
+        }
+
+        log.warn("忽略不支持的巡检报警类型，eventType={}，supported={}", value, SUPPORTED_EVENT_TYPES);
+        return null;
+    }
+
+    private List<String> snapshot(Set<String> eventTypes)
+    {
+        if (eventTypes == null || eventTypes.isEmpty())
+        {
+            return Collections.emptyList();
+        }
+        return new ArrayList<>(eventTypes);
+    }
+
+    /**
+     * 当前监听状态。
+     */
+    @Data
+    public static class ListenState
+    {
+        private final String terminalId;
+        private final String grpcIp;
+        private final List<String> eventTypes;
+    }
+}
--- a/cmvr-iot-inspection/pom.xml
+++ b/cmvr-iot-inspection/pom.xml
@ -29,5 +29,11 @@
            <artifactId>cmvr-iot-test</artifactId>
        </dependency>

+        <!-- 复用终端报警监听状态，按 grpc_ip + 事件类型过滤入库 -->
+        <dependency>
+            <groupId>com.cmvr</groupId>
+            <artifactId>cmvr-iot-device</artifactId>
+        </dependency>
+
    </dependencies>
 </project>
--- a/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/impl/InspectionDetectionAlertServiceImpl.java
+++ b/cmvr-iot-inspection/src/main/java/com/cmvr/inspection/service/impl/InspectionDetectionAlertServiceImpl.java
@ -3,6 +3,7 @@ package com.cmvr.inspection.service.impl;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.cmvr.common.config.properties.MinioProperties;
 import com.cmvr.common.core.minio.MinioService;
+import com.cmvr.device.service.InspectionAlertListenService;
 import com.cmvr.inspection.domain.InspectionDetectionAlert;
 import com.cmvr.inspection.domain.dto.alert.AlertEnvelope;
 import com.cmvr.inspection.domain.dto.alert.AlertImage;
@ -47,6 +48,7 @@ public class InspectionDetectionAlertServiceImpl implements IInspectionDetection
    private final MinioService minioService;
    private final MinioProperties minioProperties;
    private final ObjectMapper objectMapper;
+    private final InspectionAlertListenService inspectionAlertListenService;

    @Override
    @Transactional(rollbackFor = Exception.class)
@ -55,6 +57,14 @@ public class InspectionDetectionAlertServiceImpl implements IInspectionDetection
        String eventId = validateAndResolveEventId(envelope);
        String idempotencyKey = trimToNull(requestIdempotencyKey);

+        // 工作流未开启对应 IP + 事件类型监听时，报警只确认接收，不做持久化。
+        if (!inspectionAlertListenService.shouldStore(envelope.getGrpcIp(), envelope.getPayload().getLabels()))
+        {
+            log.info("忽略未监听的PPE报警，eventId={}，grpcIp={}，labels={}",
+                    eventId, envelope.getGrpcIp(), envelope.getPayload().getLabels());
+            return false;
+        }
+
        if (exists(eventId))
        {
            log.info("忽略重复PPE报警，eventId={}", eventId);
--- a/cmvr-iot-test/pom.xml
+++ b/cmvr-iot-test/pom.xml
@ -35,6 +35,12 @@
            <artifactId>cmvr-iot-grpc-client</artifactId>
        </dependency>

+        <!-- 设备终端配置，用于工作流报警监听组件按终端ID解析边缘端IP -->
+        <dependency>
+            <groupId>com.cmvr</groupId>
+            <artifactId>cmvr-iot-device</artifactId>
+        </dependency>
+
        <dependency>
            <groupId>org.graalvm.js</groupId>
            <artifactId>js</artifactId>
--- a/cmvr-iot-test/src/main/java/com/cmvr/test/enums/ActionEnum.java
+++ b/cmvr-iot-test/src/main/java/com/cmvr/test/enums/ActionEnum.java
@ -79,7 +79,11 @@ public enum ActionEnum {

    // 触控交互
    TI_PATH_SEARCH("EDGE", "TI_PATH_SEARCH", "路径搜索"),
-    TI_TOUCH_COORDINATES("EDGE", "TI_TOUCH_COORDINATES", "获取触控二维坐标")
+    TI_TOUCH_COORDINATES("EDGE", "TI_TOUCH_COORDINATES", "获取触控二维坐标"),
+
+    // 巡检报警
+    INSPECTION_ALERT_LISTEN_START("EDGE", "INSPECTION_ALERT_LISTEN_START", "开始监听报警事件"),
+    INSPECTION_ALERT_LISTEN_STOP("EDGE", "INSPECTION_ALERT_LISTEN_STOP", "结束监听报警事件")
    ;


--- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeInspectionAlertOperateService.java
+++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeInspectionAlertOperateService.java
@ -0,0 +1,105 @@
+package com.cmvr.test.flow.runtime.operator.edge;
+
+import com.alibaba.fastjson2.JSONArray;
+import com.alibaba.fastjson2.JSONObject;
+import com.cmvr.common.exception.GlobalException;
+import com.cmvr.device.service.InspectionAlertListenService;
+import com.cmvr.test.enums.ActionEnum;
+import com.cmvr.test.flow.runtime.message.TaskNodeExecuteMessage;
+import com.cmvr.test.flow.runtime.message.TaskNodeExecuteResult;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.stereotype.Service;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+/**
+ * 巡检报警监听工作流组件。
+ */
+@Service
+@RequiredArgsConstructor
+public class EdgeInspectionAlertOperateService implements EdgeOperateService
+{
+    private final InspectionAlertListenService inspectionAlertListenService;
+
+    @Override
+    public boolean supports(ActionEnum action)
+    {
+        return ActionEnum.INSPECTION_ALERT_LISTEN_START.equals(action)
+                || ActionEnum.INSPECTION_ALERT_LISTEN_STOP.equals(action);
+    }
+
+    @Override
+    public TaskNodeExecuteResult execute(TaskNodeExecuteMessage message)
+    {
+        JSONObject inputParams = message.getInputParams() == null ? new JSONObject() : message.getInputParams();
+        String terminalId = StringUtils.defaultIfBlank(inputParams.getString("terminalId"), message.getTerminalId());
+        Collection<String> eventTypes = resolveEventTypes(inputParams);
+
+        InspectionAlertListenService.ListenState state;
+        switch (message.getAction())
+        {
+            case INSPECTION_ALERT_LISTEN_START:
+                state = inspectionAlertListenService.startListen(terminalId, eventTypes);
+                break;
+            case INSPECTION_ALERT_LISTEN_STOP:
+                state = inspectionAlertListenService.stopListen(terminalId, eventTypes);
+                break;
+            default:
+                throw new GlobalException("不支持的巡检报警监听动作: " + message.getAction());
+        }
+
+        JSONObject output = new JSONObject();
+        output.put("terminalId", state.getTerminalId());
+        output.put("grpcIp", state.getGrpcIp());
+        output.put("eventTypes", state.getEventTypes());
+        return TaskNodeExecuteResult.success(output);
+    }
+
+    private Collection<String> resolveEventTypes(JSONObject inputParams)
+    {
+        List<String> eventTypes = new ArrayList<>();
+
+        Object rawEventTypes = inputParams.get("eventTypes");
+        if (rawEventTypes instanceof JSONArray)
+        {
+            JSONArray array = (JSONArray) rawEventTypes;
+            for (int i = 0; i < array.size(); i++)
+            {
+                eventTypes.add(array.getString(i));
+            }
+        }
+        else if (rawEventTypes instanceof Collection)
+        {
+            for (Object item : (Collection<?>) rawEventTypes)
+            {
+                if (item != null)
+                {
+                    eventTypes.add(String.valueOf(item));
+                }
+            }
+        }
+        else if (rawEventTypes != null)
+        {
+            addDelimited(eventTypes, String.valueOf(rawEventTypes));
+        }
+
+        addDelimited(eventTypes, inputParams.getString("eventType"));
+        return eventTypes;
+    }
+
+    private void addDelimited(List<String> eventTypes, String value)
+    {
+        if (StringUtils.isBlank(value))
+        {
+            return;
+        }
+        String[] split = value.split(",");
+        for (String item : split)
+        {
+            eventTypes.add(StringUtils.trim(item));
+        }
+    }
+}
--- a/cmvr-iot-test/src/main/java/com/cmvr/test/service/FlowActionExecutorService.java
+++ b/cmvr-iot-test/src/main/java/com/cmvr/test/service/FlowActionExecutorService.java
@ -14,6 +14,8 @@ import com.cmvr.edge.client.service.EdgeHlcService;
 import com.cmvr.edge.client.service.EdgeMicrophoneService;
 import com.cmvr.edge.client.service.EdgeSpeakerService;
 import com.cmvr.edge.client.service.EdgeAgvService;
+import com.cmvr.edge.client.service.EdgeArmService;
+import com.cmvr.device.service.InspectionAlertListenService;
 import com.cmvr.llm.service.LLMAiAgentPlatformService;
 import com.cmvr.test.enums.ActionEnum;
 import com.cmvr.test.model.vo.FlowActionRequestVO;
@ -32,6 +34,8 @@ public class FlowActionExecutorService {
    private final EdgeHlcService edgeHlcService;
    private final LLMAiAgentPlatformService llmAiAgentPlatformService;
    private final EdgeAgvService edgeAgvService;
+    private final InspectionAlertListenService inspectionAlertListenService;
+    private final EdgeArmService edgeArmService;

    public String actionExecute(FlowActionRequestVO req) {

@ -58,6 +62,12 @@ public class FlowActionExecutorService {
        String terminalId = payload.getString("terminalId");
        String deviceId = payload.getString("deviceId");

+        // 巡检报警监听组件只需要终端ID和事件类型，不需要设备ID。
+        if (ActionEnum.INSPECTION_ALERT_LISTEN_START.equals(action)
+                || ActionEnum.INSPECTION_ALERT_LISTEN_STOP.equals(action)) {
+            return executeInspectionAlertAction(action, payload);
+        }
+
        if (StrUtil.isEmpty(terminalId) || StrUtil.isEmpty(deviceId)) {
            throw new GlobalException("EDGE 类型动作必须提供 terminalId 和 deviceId");
        }
@ -89,6 +99,10 @@ public class FlowActionExecutorService {
                EdgeTouchVO edgeTouchVO = payload.to(EdgeTouchVO.class);
                return edgeHlcService.touch(edgeTouchVO);

+            // ==== 机械臂 ====
+            case ARM_MOVE_TO_POINT:
+                return executeArmMoveToPoint(edgeCommonVO, payload);
+
            // ==== 语料 ====
            case VI_PLAY_CORPUS:
                String audioPath = payload.getString("audioPath");
@ -145,6 +159,83 @@ public class FlowActionExecutorService {
        }
    }

+    private String executeArmMoveToPoint(EdgeCommonVO edgeCommonVO, JSONObject payload) {
+        Double x = payload.getDouble("x");
+        Double y = payload.getDouble("y");
+        Double z = payload.getDouble("z");
+        Double rx = payload.getDouble("rx");
+        Double ry = payload.getDouble("ry");
+        Double rz = payload.getDouble("rz");
+        String frame = payload.getString("frame");
+        Double velocity = payload.getDouble("velocity");
+        Double acceleration = payload.getDouble("acceleration");
+        Double blendRadius = payload.getDouble("blendRadius");
+
+        if (x == null || y == null || z == null) {
+            throw new GlobalException("X、Y、Z坐标不能为空");
+        }
+        if (rx == null || ry == null || rz == null) {
+            throw new GlobalException("RX、RY、RZ旋转角度不能为空");
+        }
+
+        // 单节点执行和正式工作流保持一致：先开启力矩，再下发笛卡尔直线运动。
+        edgeArmService.torqueOn(edgeCommonVO);
+        edgeArmService.moveL(
+                edgeCommonVO,
+                x,
+                y,
+                z,
+                rx,
+                ry,
+                rz,
+                frame,
+                velocity,
+                acceleration,
+                blendRadius
+        );
+        return "机械臂末端运动任务下发成功";
+    }
+
+    private String executeInspectionAlertAction(ActionEnum action, JSONObject payload) {
+        String terminalId = payload.getString("terminalId");
+        java.util.List<String> types = resolveInspectionAlertEventTypes(payload);
+
+        InspectionAlertListenService.ListenState state;
+        if (ActionEnum.INSPECTION_ALERT_LISTEN_START.equals(action)) {
+            state = inspectionAlertListenService.startListen(terminalId, types);
+        } else {
+            state = inspectionAlertListenService.stopListen(terminalId, types);
+        }
+        return JSONObject.toJSONString(state);
+    }
+
+    private java.util.List<String> resolveInspectionAlertEventTypes(JSONObject payload) {
+        java.util.List<String> types = new java.util.ArrayList<>();
+        Object eventTypes = payload.get("eventTypes");
+
+        if (eventTypes instanceof Iterable) {
+            for (Object item : (Iterable<?>) eventTypes) {
+                if (item != null) {
+                    types.add(String.valueOf(item));
+                }
+            }
+        } else if (eventTypes != null) {
+            addInspectionAlertEventTypes(types, String.valueOf(eventTypes));
+        }
+
+        addInspectionAlertEventTypes(types, payload.getString("eventType"));
+        return types;
+    }
+
+    private void addInspectionAlertEventTypes(java.util.List<String> types, String value) {
+        if (StrUtil.isBlank(value)) {
+            return;
+        }
+        for (String item : value.split(",")) {
+            types.add(item.trim());
+        }
+    }
+
    private String executeLLMAction(ActionEnum action, FlowActionRequestVO req) {
        log.info("LLM 执行动作: {}", action);