From dfdf575932a0bc3a7ee7e3581227d7a2f50cd6aa Mon Sep 17 00:00:00 2001 From: lixiaolong <702156524@qq.com> Date: Thu, 23 Jul 2026 13:52:37 +0800 Subject: [PATCH] =?UTF-8?q?feat(workflow):=20=E6=B7=BB=E5=8A=A0=E5=B7=A1?= =?UTF-8?q?=E6=A3=80=E6=8A=A5=E8=AD=A6=E7=9B=91=E5=90=AC=E5=92=8C=E6=9C=BA?= =?UTF-8?q?=E6=A2=B0=E8=87=82=E6=8E=A7=E5=88=B6=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 在ActionEnum中新增巡检报警监听相关枚举值 - 集成InspectionAlertListenService和EdgeArmService依赖 - 实现巡检报警监听开始和停止功能 - 添加机械臂移动到指定坐标点功能 - 在工作流执行器中增加报警事件类型解析逻辑 - 实现基于终端ID的边缘端IP解析和报警过滤机制 - 新增巡检报警监听工作流组件服务类 - 创建巡检报警监听状态管理服务 --- .../service/InspectionAlertListenService.java | 242 ++++++++++++++++++ cmvr-iot-inspection/pom.xml | 6 + .../InspectionDetectionAlertServiceImpl.java | 10 + cmvr-iot-test/pom.xml | 8 +- .../java/com/cmvr/test/enums/ActionEnum.java | 6 +- .../EdgeInspectionAlertOperateService.java | 105 ++++++++ .../service/FlowActionExecutorService.java | 91 +++++++ 7 files changed, 466 insertions(+), 2 deletions(-) create mode 100644 cmvr-iot-device/src/main/java/com/cmvr/device/service/InspectionAlertListenService.java create mode 100644 cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeInspectionAlertOperateService.java diff --git 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 new file mode 100644 index 0000000..1643068 --- /dev/null +++ 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; + +/** + * 巡检报警监听状态服务。 + * + *

工作流开始监听时，根据终端ID解析终端IP，并将 IP + 报警类型保存到内存。 + * cmvr_edge_ai 推送报警后，巡检报警入库服务通过该服务判断是否需要保存。 + * 当前状态只在本 JVM 内有效，应用重启后会清空。

+ */ +@Slf4j +@Service +@RequiredArgsConstructor +public class InspectionAlertListenService +{ + /** 当前支持的 PPE 违规类型。 */ + private static final Set SUPPORTED_EVENT_TYPES; + + static + { + Set 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> listenMap = new ConcurrentHashMap<>(); + + /** + * 开始监听指定终端的指定报警类型。 + * + * @param terminalId 终端ID + * @param eventTypes 报警类型，支持 No-Glove、No-Helmet，可多选 + * @return 当前终端IP下正在监听的报警类型 + */ + public ListenState startListen(String terminalId, Collection eventTypes) + { + String grpcIp = resolveTerminalIp(terminalId); + Set normalizedTypes = normalizeEventTypes(eventTypes, true); + + Set 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)); + } + + /** + * 结束监听指定终端的指定报警类型。 + * + *

移除不存在的 IP + 类型不抛异常，只记录警告日志，避免工作流重复清理时失败。

+ * + * @param terminalId 终端ID + * @param eventTypes 报警类型，支持 No-Glove、No-Helmet，可多选 + * @return 当前终端IP下剩余正在监听的报警类型 + */ + public ListenState stopListen(String terminalId, Collection eventTypes) + { + String grpcIp = resolveTerminalIp(terminalId); + Set normalizedTypes = normalizeEventTypes(eventTypes, true); + Set 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 labels) + { + String normalizedIp = StringUtils.trimToNull(grpcIp); + if (normalizedIp == null) + { + log.warn("忽略巡检报警：grpc_ip为空，labels={}", labels); + return false; + } + + Set listeningTypes = listenMap.get(normalizedIp); + if (listeningTypes == null || listeningTypes.isEmpty()) + { + log.info("忽略巡检报警：未开启该IP的报警监听，grpcIp={}，labels={}", normalizedIp, labels); + return false; + } + + Set 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 normalizeEventTypes(Collection eventTypes, boolean failOnEmpty) + { + Set 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 snapshot(Set 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 eventTypes; + } +} diff --git a/cmvr-iot-inspection/pom.xml b/cmvr-iot-inspection/pom.xml index 3f8d31a..b158199 100644 --- a/cmvr-iot-inspection/pom.xml +++ b/cmvr-iot-inspection/pom.xml @@ -29,5 +29,11 @@ cmvr-iot-test + + + com.cmvr + cmvr-iot-device + + diff --git 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 index 23acf7c..a6cce88 100644 --- 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); diff --git a/cmvr-iot-test/pom.xml b/cmvr-iot-test/pom.xml index d339b69..fae65ce 100644 --- a/cmvr-iot-test/pom.xml +++ b/cmvr-iot-test/pom.xml @@ -35,6 +35,12 @@ cmvr-iot-grpc-client + + + com.cmvr + cmvr-iot-device + + org.graalvm.js js @@ -47,4 +53,4 @@ - \ No newline at end of file + diff --git 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 index 047db03..d0f76d2 100644 --- 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", "结束监听报警事件") ; diff --git 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 new file mode 100644 index 0000000..39cf150 --- /dev/null +++ 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 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 resolveEventTypes(JSONObject inputParams) + { + List 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 eventTypes, String value) + { + if (StringUtils.isBlank(value)) + { + return; + } + String[] split = value.split(","); + for (String item : split) + { + eventTypes.add(StringUtils.trim(item)); + } + } +} diff --git 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 index b1ce6fa..a7a61e8 100644 --- 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 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 resolveInspectionAlertEventTypes(JSONObject payload) { + java.util.List 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 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);