diff --git a/cmvr-iot-admin/src/main/java/com/cmvr/web/controller/vi/ViProjectController.java b/cmvr-iot-admin/src/main/java/com/cmvr/web/controller/vi/ViProjectController.java index 249db2a..aaa7d6a 100644 --- a/cmvr-iot-admin/src/main/java/com/cmvr/web/controller/vi/ViProjectController.java +++ b/cmvr-iot-admin/src/main/java/com/cmvr/web/controller/vi/ViProjectController.java @@ -3,6 +3,7 @@ package com.cmvr.web.controller.vi; import com.cmvr.common.core.controller.BaseController; import com.cmvr.common.core.domain.AjaxResult; import com.cmvr.common.core.page.TableDataInfo; +import com.cmvr.test.model.vo.TeTaskExecuteProjectVO; import com.cmvr.vi.model.domain.ViProject; import com.cmvr.vi.service.IViProjectService; import io.swagger.annotations.Api; @@ -66,9 +67,9 @@ public class ViProjectController extends BaseController { return toAjax(viProjectService.deleteViProjectByProjectIds(projectIds)); } -// @ApiOperation("执行项目") + @ApiOperation("执行项目") @PostMapping("/execute") - public AjaxResult execute(@RequestParam("projectId") String projectId) { - return AjaxResult.ok(viProjectService.execute(projectId)); + public AjaxResult execute(@RequestBody TeTaskExecuteProjectVO taskExecuteProjectVO) { + return AjaxResult.ok(viProjectService.execute(taskExecuteProjectVO)); } } 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 3edec86..50ab9f0 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 @@ -41,10 +41,22 @@ public enum ActionEnum { // ---------------扬声器（人工嘴）--------------- SPEAKER_PLAYAUDIO("EDGE", "SPEAKER_PLAYAUDIO", "播放音频"), + CORPUS_PLAY("EDGE", "CORPUS_PLAY", "播放语料"), // 大模型行为 // ---------------获取触控坐标--------------- TOUCH_COORDINATES("LLM", "TOUCH_COORDINATES", "获取触控坐标"), + + // 语音交互 + // ---------------语料--------------- + VI_CORPUS_WAKE("VI", "VI_CORPUS_WAKE", "唤醒语料处理"), + VI_CORPUS_SINGLE("VI", "VI_CORPUS_SINGLE", "测试语料-单次对话语料处理"), + VI_CORPUS_CONTINUOUS("VI", "VI_CORPUS_CONTINUOUS", "测试语料-连续对话语料处理"), + VI_PLAY_CORPUS("VI", "VI_PLAY_CORPUS", "播放语料"), + // ---------------方案--------------- + VI_SCHEME("VI", "VI_SCHEME", "语音交互-方案处理"), + + ; diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/enums/LoopNodeTypeEnum.java b/cmvr-iot-test/src/main/java/com/cmvr/test/enums/LoopNodeTypeEnum.java new file mode 100644 index 0000000..affd95e --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/enums/LoopNodeTypeEnum.java @@ -0,0 +1,10 @@ +package com.cmvr.test.enums; + +/** + * 循环节点类型 + */ +public enum LoopNodeTypeEnum { + NORMAL, // 普通 + MULTIPLE_CORPUS, // 多条语料（单次和连续） + CONTINUOUS_MULTIPLE_CORPUS // 连续多条语料 +} diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/enums/RunModeEnum.java b/cmvr-iot-test/src/main/java/com/cmvr/test/enums/RunModeEnum.java index b132523..3de1995 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/enums/RunModeEnum.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/enums/RunModeEnum.java @@ -1,6 +1,7 @@ package com.cmvr.test.enums; public enum RunModeEnum { - NORMAL, //正常 - TRIAL //试运行 + NORMAL, //正常 + TRIAL, //试运行 + PROJECT //项目 } \ No newline at end of file diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowModelBuilder.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowModelBuilder.java index 669b84e..cdd9546 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowModelBuilder.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowModelBuilder.java @@ -17,7 +17,6 @@ import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; -import java.util.Optional; /** * 构建任务执行图（支持主流程 + 子图 + 循环） @@ -125,15 +124,6 @@ public class FlowModelBuilder { } } wrapper.setChildNodes(childList); - - // 提取 loopNum 参数 - Optional optional = wrapper.getNodeParams().stream() - .filter(x -> "loopNum".equalsIgnoreCase(x.getName()) && x.getScope() == ParamScope.NODE) - .findFirst(); - if (!optional.isPresent()) { - throw new IllegalArgumentException("loopNum 不存在"); - } - wrapper.setLoopCount(Integer.parseInt(optional.get().getInput())); } } return toRemove; diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowNodeWrapper.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowNodeWrapper.java index c0ff02b..0503560 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowNodeWrapper.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowNodeWrapper.java @@ -63,11 +63,6 @@ public class FlowNodeWrapper { */ private List childNodeEdges = new ArrayList<>(); - /** - * 循环次数（如果是循环节点，此字段用于控制循环次数） - */ - private Integer loopCount; - /** * 子流程（当节点为循环节点或包含子流程时，此字段存储子流程） */ diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowParamDef.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowParamDef.java index 0dd632e..54d09e7 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowParamDef.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowParamDef.java @@ -51,6 +51,13 @@ public class FlowParamDef { */ private List quote; + /** + * 引用的是输出参数还是输入参数 + * outputParams -> output + * nodeParams -> input + */ + private String quoteType; + /** * 参数描述（用于 UI 显示或生成 Schema） */ diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowParamHelper.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowParamHelper.java deleted file mode 100644 index 116faa8..0000000 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowParamHelper.java +++ /dev/null @@ -1,95 +0,0 @@ -package com.cmvr.test.flow.builder; - -import cn.hutool.core.collection.CollUtil; -import com.alibaba.fastjson2.JSONArray; -import com.alibaba.fastjson2.JSONObject; -import com.cmvr.test.enums.ParamScope; - -import java.util.List; - -public class FlowParamHelper { - - /** - * 将参数定义列表中的 input 值构建为 JSONObject - */ - public static JSONObject toJson(List defs) { - JSONObject result = new JSONObject(); - if (CollUtil.isEmpty(defs)) return result; - - for (FlowParamDef param : defs) { - if (param.getType().equalsIgnoreCase(ParamScope.INPUT.name())) { - result.put(param.getName(), param.getInput()); - continue; - } - - String name = param.getName(); - String type = param.getType(); - Object input = param.getInput(); - - if ("Object".equals(type) && CollUtil.isNotEmpty(param.getChildren())) { - result.put(name, toJson(param.getChildren())); - } else if ("Array".equals(type) && CollUtil.isNotEmpty(param.getChildren())) { - JSONArray arr = new JSONArray(); - // 默认数组内只有一个结构体模型 - arr.add(toJson(param.getChildren())); - result.put(name, arr); - } else { - result.put(name, input); - } - } - - return result; - } - - /** - * 从引用中解析 quote 值（会合并 input 和 quote 参数） - */ - public static JSONObject buildInputParams(List nodeParams, - String startNodeId, - List startInputDefs, - ParamProvider paramProvider) { - JSONObject input = new JSONObject(); - - for (FlowParamDef param : nodeParams) { - if (param.getScope() != ParamScope.NODE) continue; - - if ("input".equals(param.getType())) { - input.put(param.getName(), param.getInput()); - } else if ("quote".equals(param.getType()) && CollUtil.isNotEmpty(param.getQuote())) { - String refNodeId = param.getQuote().get(0); - List path = param.getQuote().subList(1, param.getQuote().size()); - - List nodeParams1 = paramProvider.getNodeParams(refNodeId); - JSONObject source = refNodeId.equals(startNodeId) - ? toJson(startInputDefs) - : toJson(nodeParams1); - - Object val = findNestedValue(source, path); - input.put(param.getName(), val); - } - } - - return input; - } - - - /** - * JSON 多层路径提取 - */ - public static Object findNestedValue(JSONObject source, List path) { - if (source == null || CollUtil.isEmpty(path)) return null; - Object curr = source; - for (String key : path) { - if (!(curr instanceof JSONObject)) return null; - curr = ((JSONObject) curr).get(key); - } - return curr; - } - - /** - * 参数提供者接口：用于从上下文获取 ref 节点的 paramDef 列表 - */ - public interface ParamProvider { - List getNodeParams(String nodeId); - } -} diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/TaskKeyBuilder.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/TaskKeyBuilder.java index 0fd9929..700ce8b 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/TaskKeyBuilder.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/TaskKeyBuilder.java @@ -1,31 +1,41 @@ package com.cmvr.test.flow.builder; -import java.util.Arrays; -import java.util.Objects; +import java.util.List; import java.util.stream.Collectors; public class TaskKeyBuilder { /** - * 构建 itemKey + * 创建scheduleCache key */ - public static String itemKey(String instId, String itemId) { - return buildKey(instId, itemId); + public static String buildKey(String instId, String nodeId, List iterations) { + StringBuilder sb = new StringBuilder(instId).append(":").append(nodeId); + for (int it : iterations) { + sb.append(":").append(it); + } + return sb.toString(); } /** - * 构建 nodeKey + * 线程注册-节点执行上下文key */ - public static String nodeKey(String instId, String itemId, String nodeId) { - return buildKey(instId, itemId, nodeId); + public static String buildKey(String instId, String itemId, String nodeId) { + return instId + "_" + itemId + "_" + nodeId; } /** - * 构建通用 key，自动跳过 null 或空字符串 + * 线程注册-loop节点执行上下文key */ - public static String buildKey(String... parts) { - return Arrays.stream(parts) - .filter(part -> Objects.nonNull(part) && !part.isEmpty()) - .collect(Collectors.joining("_")); + public static String buildLoopKey(String instId, String nodeId) { + return instId + "_" + nodeId; + } + + /** + * 节点输出参数 key + */ + public static String buildKey(String nodeId, List iterations) { + return nodeId + "@" + iterations.stream() + .map(String::valueOf) + .collect(Collectors.joining(".")); } } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskContext.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskContext.java index 26b8f38..8572c35 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskContext.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskContext.java @@ -1,14 +1,18 @@ package com.cmvr.test.flow.context; +import cn.hutool.core.util.ObjUtil; import com.alibaba.fastjson2.JSONObject; import com.cmvr.test.enums.RunModeEnum; import com.cmvr.test.enums.TaskStatusEnum; import com.cmvr.test.enums.TerminalStatusEnum; +import com.cmvr.test.flow.builder.TaskKeyBuilder; import lombok.Data; +import java.util.Collections; import java.util.Date; -import java.util.HashMap; +import java.util.List; import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; /** * 任务执行上下文，用于保存任务运行状态、控制标志等信息。 @@ -68,9 +72,9 @@ public class TaskContext { /** * 存储节点输出参数 - * key为nodeI，value为节点的输出参数 + * key = nodeId@iterationPath，例如 "nodeA@0.1.2" */ - private Map nodeOutParams = new HashMap<>(); + private final Map nodeOutputs = new ConcurrentHashMap<>(); /** * 运行任务时的初始参数 @@ -100,16 +104,27 @@ public class TaskContext { } /** - * 存储指定节点的输出参数 + * 保存节点输出 */ - public void setNodeOutParams(String nodeId, JSONObject params) { - nodeOutParams.put(nodeId, params); + public void setNodeOutput(String nodeId, List iterations, JSONObject outputParams) { + String key = TaskKeyBuilder.buildKey(nodeId, iterations); + nodeOutputs.put(key, outputParams != null ? outputParams : new JSONObject()); + updateTimestamp(); } /** - * 获取指定节点的输出参数 + * 获取节点输出 */ - public JSONObject getNodeOutParams(String nodeId) { - return nodeOutParams.get(nodeId); + public JSONObject getNodeOutput(String nodeId, List iterations) { + // 1. 精确匹配：带循环层级 + String keyWithIterations = TaskKeyBuilder.buildKey(nodeId, iterations); + JSONObject result = nodeOutputs.get(keyWithIterations); + if (ObjUtil.isNotEmpty(result)) { + return result; + } + + // 2. 回退匹配：循环外（全局） + String keyWithoutIterations = TaskKeyBuilder.buildKey(nodeId, Collections.emptyList()); + return nodeOutputs.get(keyWithoutIterations); } } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskContextManager.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskContextManager.java index 664b685..527ab48 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskContextManager.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskContextManager.java @@ -6,6 +6,7 @@ import com.cmvr.test.enums.TerminalStatusEnum; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; +import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @@ -97,20 +98,20 @@ public class TaskContextManager { /** * 存储某个节点的输出参数 */ - public void setNodeOutParams(String instId, String nodeId, JSONObject params) { + public void setNodeOutParams(String instId, String nodeId, List iterations, JSONObject params) { TaskContext context = instContextMap.get(instId); if (context != null) { - context.setNodeOutParams(nodeId, params); + context.setNodeOutput(nodeId, iterations, params); } } /** - * 根据 nodeId 和 action 获取节点的输出参数 + * 根据 nodeId 和 iterations 获取节点的输出参数 */ - public JSONObject getNodeOutParams(String instId, String nodeId) { + public JSONObject getNodeOutParams(String instId, String nodeId, List iterations) { TaskContext context = instContextMap.get(instId); if (context != null) { - return context.getNodeOutParams(nodeId); + return context.getNodeOutput(nodeId, iterations); } return null; } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskInstHolder.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskInstHolder.java index 632137e..fff8232 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskInstHolder.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskInstHolder.java @@ -88,14 +88,14 @@ public class TaskInstHolder { /** * 存储当前节点的输出参数 */ - public void setNodeOutParams(String instId, String nodeId, JSONObject params) { - taskContextManager.setNodeOutParams(instId, nodeId, params); + public void setNodeOutParams(String instId, String nodeId, List iterations, JSONObject params) { + taskContextManager.setNodeOutParams(instId, nodeId, iterations, params); } /** - * 根据 nodeId 和 action 获取节点的输出参数 + * 根据 nodeId 和 iterations 获取节点的输出参数 */ - public JSONObject getNodeOutParams(String instId, String nodeId) { - return taskContextManager.getNodeOutParams(instId, nodeId); + public JSONObject getNodeOutParams(String instId, String nodeId, List iterations) { + return taskContextManager.getNodeOutParams(instId, nodeId, iterations); } } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskThreadRegistry.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskThreadRegistry.java index 126c53a..b14e31f 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskThreadRegistry.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskThreadRegistry.java @@ -2,6 +2,7 @@ package com.cmvr.test.flow.context; import com.cmvr.test.enums.NodeTypeEnum; import com.cmvr.test.flow.builder.FlowNodeWrapper; +import com.cmvr.test.flow.builder.TaskKeyBuilder; import com.cmvr.test.flow.runtime.message.TaskNodeExecuteContext; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; @@ -27,20 +28,11 @@ public class TaskThreadRegistry { // 结构为 instId_loopNodeId -> CountDownLatch private final Map loopLatchMap = new ConcurrentHashMap<>(); - - private String buildKey(String instId, String itemId, String nodeId) { - return instId + "_" + itemId + "_" + nodeId; - } - - private String buildLoopKey(String instId, String nodeId) { - return instId + "_" + nodeId; - } - /** * 注册节点执行上下文 */ public void registerNodeContext(String instId, String itemId, String nodeId, TaskNodeExecuteContext context) { - String key = buildKey(instId, itemId, nodeId); + String key = TaskKeyBuilder.buildKey(instId, itemId, nodeId); nodeContextMap.put(key, context); log.info("注册节点上下文: key={}, thread={}", key, context.getThread().getName()); } @@ -49,7 +41,7 @@ public class TaskThreadRegistry { * 获取上下文 */ public TaskNodeExecuteContext getNodeContext(String instId, String itemId, String nodeId) { - return nodeContextMap.get(buildKey(instId, itemId, nodeId)); + return nodeContextMap.get(TaskKeyBuilder.buildKey(instId, itemId, nodeId)); } /** @@ -102,7 +94,7 @@ public class TaskThreadRegistry { * 移除节点上下文（节点执行成功时调用） */ public void unregisterNodeContext(String instId, String itemId, String nodeId) { - String key = buildKey(instId, itemId, nodeId); + String key = TaskKeyBuilder.buildKey(instId, itemId, nodeId); nodeContextMap.remove(key); log.info("移除节点上下文: key={}", key); } @@ -118,7 +110,7 @@ public class TaskThreadRegistry { * 注册 loop 节点 latch（每次只允许一个 latch） */ public void registerLatch(String instId, String loopNodeId, CountDownLatch latch) { - String key = buildLoopKey(instId, loopNodeId); + String key = TaskKeyBuilder.buildLoopKey(instId, loopNodeId); loopLatchMap.put(key, latch); } @@ -126,7 +118,7 @@ public class TaskThreadRegistry { * 获取 latch（可用于测试或回调判断） */ public CountDownLatch getLatch(String instId, String loopNodeId) { - String key = buildLoopKey(instId, loopNodeId); + String key = TaskKeyBuilder.buildLoopKey(instId, loopNodeId); boolean flag = loopLatchMap.containsKey(key); System.out.println("这个latch是存在的吗："+flag); return loopLatchMap.get(key); @@ -136,7 +128,7 @@ public class TaskThreadRegistry { * 主动释放 latch（用于暂停任务时中断 loop await） */ public void releaseLatch(String instId, String loopNodeId) { - String key = buildLoopKey(instId, loopNodeId); + String key = TaskKeyBuilder.buildLoopKey(instId, loopNodeId); CountDownLatch latch = loopLatchMap.remove(key); if (latch != null) { try { diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/control/FlowControlService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/control/FlowControlService.java index c12f004..ef33b4a 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/control/FlowControlService.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/control/FlowControlService.java @@ -25,7 +25,7 @@ import java.util.concurrent.CountDownLatch; import java.util.stream.Collectors; /** - * 流程控制服务 + * 流程控制服务(检测项内部) * 终止/暂停/继续 */ @Slf4j @@ -78,7 +78,7 @@ public class FlowControlService { if (ctx.isPaused()) { throw new GlobalException("任务已处于暂停状态"); } - ctx.setPaused(false); + ctx.setPaused(true); ctx.setTerminalStatus(TerminalStatusEnum.RUNNING); ctx.setStatus(TaskStatusEnum.PAUSED); ctx.updateTimestamp(); diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/dispatcher/FlowLoopNodeHandler.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/dispatcher/FlowLoopNodeHandler.java index 24e8838..7b11a01 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/dispatcher/FlowLoopNodeHandler.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/dispatcher/FlowLoopNodeHandler.java @@ -1,6 +1,7 @@ package com.cmvr.test.flow.runtime.dispatcher; import cn.hutool.core.bean.BeanUtil; +import com.alibaba.fastjson2.JSONObject; import com.cmvr.common.utils.spring.SpringUtils; import com.cmvr.test.flow.builder.FlowGraph; import com.cmvr.test.flow.builder.FlowNodeWrapper; @@ -27,25 +28,26 @@ public class FlowLoopNodeHandler implements FlowNodeTypeHandler { public TaskNodeExecuteResult handle(TaskNodeExecuteMessage message) { String instId = message.getInstId(); FlowGraph graph = message.getGraph(); + JSONObject inputParams = message.getInputParams(); // 当前从第几次循环开始 - int loopNum = Math.max(message.getLoopNum(), 1); + int loopStartNum = Math.max(message.getLoopNum(), 1); // 父级迭代路径 List parentIterations = message.getIterations(); String nodeId = message.getNodeId(); FlowNodeWrapper nodeWrapper = graph.getNode(nodeId); - Integer loopCount = nodeWrapper.getLoopCount(); + int loopCount = inputParams.getIntValue("loopNum"); // 获取循环次数 // 如果没有设置循环次数或者循环次数为 0，则跳过处理 - if (loopCount == null || loopCount <= 0) { + if (loopCount <= 0) { log.warn("循环节点 {} 没有有效的 loopCount 参数，跳过处理", nodeId); - return TaskNodeExecuteResult.success(message.getInputParams()); + return TaskNodeExecuteResult.success(inputParams); } FlowGraph subGraph = nodeWrapper.getSubGraph(); subGraph.setSub(true); // 循环执行子图 - for (int i = loopNum; i <= loopCount; i++) { + for (int i = loopStartNum; i <= loopCount; i++) { // 如果子图设置了跳出标志，则结束循环 if (subGraph.isBreakLoop()) { log.info("检测到子图设置跳出标志，提前结束循环：instId={}, nodeId={}", instId, nodeId); @@ -56,16 +58,6 @@ public class FlowLoopNodeHandler implements FlowNodeTypeHandler { taskThreadRegistry.registerLatch(instId, nodeId, latch); FlowItemExecutor flowItemExecutor = SpringUtils.getBean(FlowItemExecutor.class); - // 构造新的迭代路径 -// List newIterations = new ArrayList<>(parentIterations); -// newIterations.add(i); -// -// // 克隆 message，避免覆盖父级 -// TaskNodeExecuteMessage subMessage = new TaskNodeExecuteMessage(); -// BeanUtil.copyProperties(message, subMessage); -//// subMessage.setLoopNum(i); // 当前层循环次数 -// subMessage.setIterations(newIterations); // 完整迭代路径 - // 构造新的迭代路径 List newIterations = new ArrayList<>(parentIterations); newIterations.add(i); @@ -90,6 +82,6 @@ public class FlowLoopNodeHandler implements FlowNodeTypeHandler { } // 返回执行结果 - return TaskNodeExecuteResult.success(message.getInputParams()); + return TaskNodeExecuteResult.success(inputParams); } } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowItemExecutor.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowItemExecutor.java index b797aa3..bae68cb 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowItemExecutor.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowItemExecutor.java @@ -15,7 +15,6 @@ import com.cmvr.test.flow.context.TaskThreadRegistry; import com.cmvr.test.flow.runtime.dispatcher.FlowNodeTypeDispatch; import com.cmvr.test.flow.runtime.engine.support.FlowBranchEvaluator; import com.cmvr.test.flow.runtime.engine.support.FlowExecutionChainBuilder; -import com.cmvr.test.flow.runtime.engine.support.FlowNodeParamPreparer; import com.cmvr.test.flow.runtime.interceptor.FlowMsgPreInterceptor; import com.cmvr.test.flow.runtime.message.TaskNodeExecuteContext; import com.cmvr.test.flow.runtime.message.TaskNodeExecuteMessage; @@ -39,7 +38,6 @@ public class FlowItemExecutor { private final List interceptors; private final FlowNodeTypeDispatch dispatchNodeHandler; private final TaskInstHolder taskInstHolder; - private final FlowNodeParamPreparer paramPreparer; private final FlowExecutionChainBuilder chainBuilder; private final TaskThreadRegistry taskThreadRegistry; @@ -112,11 +110,10 @@ public class FlowItemExecutor { registerThread(graph, node, startNodeId, startInputDefs, rootMessage, onFinished, iterations, instId, itemId, nodeId); try { - // 准备输入参数 - JSONObject inputParams = paramPreparer.prepare(graph, node, startNodeId, startInputDefs, rootMessage); - - // 创建带上下文的执行消息（关键：注入 graph + iterations） - TaskNodeExecuteMessage message = buildTaskNodeExecuteMsg(graph, node, rootMessage, nodeId, inputParams, iterations); + // 创建带上下文的执行消息 + TaskNodeExecuteMessage message = buildTaskNodeExecuteMsg( + graph, node, rootMessage, nodeId, iterations + ); // 执行责任链 TaskNodeExecuteResult result = chainBuilder.build(interceptors, dispatchNodeHandler).apply(message); @@ -128,7 +125,8 @@ public class FlowItemExecutor { // 分支节点处理 if (node.getNodeType() == NodeTypeEnum.BRANCH) { - postHandleBranchNode(graph, node, startNodeId, startInputDefs, rootMessage, onFinished, iterations, nodeId, inputParams, message); + postHandleBranchNode(graph, node, startNodeId, startInputDefs, + rootMessage, onFinished, iterations, nodeId, message.getInputParams(), message); return; } @@ -160,13 +158,13 @@ public class FlowItemExecutor { @NotNull private static TaskNodeExecuteMessage buildTaskNodeExecuteMsg(FlowGraph graph, FlowNodeWrapper node, TaskNodeExecuteMessage rootMessage, String nodeId, - JSONObject inputParams, List iterations) { + List iterations) { TaskNodeExecuteMessage message = new TaskNodeExecuteMessage(); BeanUtil.copyProperties(rootMessage, message); message.setNodeId(nodeId); message.setNodeType(node.getNodeType().name()); message.setAction(node.getAction()); - message.setInputParams(inputParams); +// message.setInputParams(inputParams); message.setGraph(graph); // 只设置 iterations message.setIterations(new ArrayList<>(iterations)); diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskEngine.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskEngine.java index 8f40178..d1e8982 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskEngine.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskEngine.java @@ -42,6 +42,9 @@ public class FlowTaskEngine { message.setTaskId(taskId); message.setTerminalId(terminalId); message.setItemId(itemId); + if (runMode.equals(RunModeEnum.PROJECT)) { + message.setLoopDetail(item.getSchemeInfo()); + } int pendingItemCount = totalItemCount - i - 1; message.setPendingItemCount(pendingItemCount); diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskRuntimeEntry.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskRuntimeEntry.java index 46597ec..fb8e4e7 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskRuntimeEntry.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskRuntimeEntry.java @@ -5,6 +5,7 @@ import cn.hutool.core.exceptions.ExceptionUtil; import cn.hutool.core.util.ObjUtil; import cn.hutool.core.util.StrUtil; import com.alibaba.fastjson2.JSON; +import com.alibaba.fastjson2.JSONArray; import com.alibaba.fastjson2.JSONObject; import com.cmvr.common.exception.GlobalException; import com.cmvr.test.enums.ActionEnum; @@ -16,6 +17,7 @@ import com.cmvr.test.model.domain.TeTaskInst; import com.cmvr.test.model.dto.TeQueryTaskDetailDTO; import com.cmvr.test.model.vo.TeDetectItemDeployFlowVO; import com.cmvr.test.model.vo.TeTaskExecuteNormalVO; +import com.cmvr.test.model.vo.TeTaskExecuteProjectVO; import com.cmvr.test.model.vo.TeTaskExecuteTrailVO; import com.cmvr.test.service.ITeDetectionItemService; import com.cmvr.test.service.ITeTaskInstService; @@ -133,6 +135,50 @@ public class FlowTaskRuntimeEntry implements FlowTaskRuntimeService { } } + @Transactional(rollbackFor = Exception.class) + public String executeProjectTask(TeTaskExecuteProjectVO taskExecuteProjectVO) { + JSONObject runParams = taskExecuteProjectVO.getRunParams(); + String terminalId = runParams.getString("terminalId"); + if (StrUtil.isEmpty(terminalId)) { + throw new GlobalException("终端ID不能为空"); + } + if (taskInstHolder.isTerminalLocked(terminalId)) { + throw new GlobalException("当前终端正在执行其他任务，请稍后再试"); + } + + String taskId = taskExecuteProjectVO.getProjectId(); + + // 查询任务详情（方案对应流程信息） + List details = taskOrchestrationService.queryTaskDetail(taskId); + JSONArray prjInfo = taskExecuteProjectVO.getPrjInfo(); + for (int i = 0; i < details.size(); i++) { + JSONObject jsonObject = prjInfo.getJSONObject(i); + TeQueryTaskDetailDTO detail = details.get(i); + detail.setSchemeInfo(jsonObject); + } + + // 创建并保存任务实例 + TeTaskInst instance = createAndSaveTaskInstance(taskId, terminalId, RunModeEnum.PROJECT); + String instId = instance.getId(); + + try { + // 注册上下文 + registerTaskContext(instId, taskId, terminalId, null, RunModeEnum.PROJECT, taskExecuteProjectVO.getRunParams()); + + // 异步提交执行 + flowTaskAsyncDispatcher.submit(instId, taskId, terminalId, RunModeEnum.PROJECT, details); + return instId; + } catch (Exception e) { + taskInstHolder.markFailed( + instId, taskId, null, null, null, + ActionEnum.NONE.getOperate(), ActionEnum.NONE.getAction(), + JSON.toJSONString(taskExecuteProjectVO), + "项目执行异常：" + StrUtil.sub(ExceptionUtil.getMessage(e), 0, 480) + ); + throw new GlobalException("项目执行异常：" + e.getMessage()); + } + } + /** * 创建并保存任务实例 */ diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskRuntimeService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskRuntimeService.java index fc39884..6135f0f 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskRuntimeService.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskRuntimeService.java @@ -1,7 +1,7 @@ package com.cmvr.test.flow.runtime.engine; -import com.cmvr.test.enums.TaskStatusEnum; import com.cmvr.test.model.vo.TeTaskExecuteNormalVO; +import com.cmvr.test.model.vo.TeTaskExecuteProjectVO; import com.cmvr.test.model.vo.TeTaskExecuteTrailVO; /** @@ -25,4 +25,10 @@ public interface FlowTaskRuntimeService { * @return 返回任务实例ID（instId） */ String executeTrialTask(TeTaskExecuteTrailVO taskExecuteTrailVO); + + /** + * 执行（语音交互-项目）任务 + * 将方案注册成流程，并进行编排，然后串行执行 + */ + String executeProjectTask(TeTaskExecuteProjectVO taskExecuteProjectVO); } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskScheduler.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskScheduler.java index 56e5a6b..b291a53 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskScheduler.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskScheduler.java @@ -3,6 +3,7 @@ package com.cmvr.test.flow.runtime.engine; import cn.hutool.core.exceptions.ExceptionUtil; import com.cmvr.test.flow.builder.FlowGraph; import com.cmvr.test.flow.builder.FlowNodeWrapper; +import com.cmvr.test.flow.builder.TaskKeyBuilder; import com.cmvr.test.flow.context.TaskContext; import com.cmvr.test.flow.runtime.engine.support.FlowSchedulerCache; import lombok.RequiredArgsConstructor; @@ -71,33 +72,6 @@ public class FlowTaskScheduler { } } - /** - * 某节点执行完成后，判断并调度其所有后继节点 - */ -/* public void markCompleted(String instId, - String nodeId, - FlowGraph graph, - FlowItemExecutor.NodeExecutor executorFunc, - int loopIteration) { - - List nextNodes = graph.getNextNodes(nodeId); - - for (String nextId : nextNodes) { - String key = FlowSchedulerCache.buildKey(instId, nextId, loopIteration); - - // 使用 Set 来记录已完成的前驱节点 - Set completedPreNodes = schedulerCache.getOrInitCompletedPre(key); - // 添加当前节点作为已完成的前驱节点 - completedPreNodes.add(nodeId); - - int preCount = graph.getPredecessors().getOrDefault(nextId, Collections.emptyList()).size(); - - // 如果当前节点的所有前驱节点都已完成，则调度后继节点 - if (completedPreNodes.size() == preCount) { - scheduleNode(instId, nextId, graph, executorFunc, loopIteration); - } - } - }*/ /** * 某节点执行完成后，判断并调度其所有后继节点 */ @@ -109,7 +83,7 @@ public class FlowTaskScheduler { List nextNodes = graph.getNextNodes(nodeId); for (String nextId : nextNodes) { - String key = FlowSchedulerCache.buildKey(instId, nextId, iterations); + String key = TaskKeyBuilder.buildKey(instId, nextId, iterations); // 使用 Set 来记录已完成的前驱节点 Set completedPreNodes = schedulerCache.getOrInitCompletedPre(key); @@ -124,33 +98,6 @@ public class FlowTaskScheduler { } } } - - /** - * 提交节点执行任务（只调度一次） - */ -/* private void scheduleNode(String instId, - String nodeId, - FlowGraph graph, - FlowItemExecutor.NodeExecutor executorFunc, - int loopIteration) { - - String key = FlowSchedulerCache.buildKey(instId, nodeId, loopIteration); - - // 如果已经调度过，则不再重复调度 - if (schedulerCache.trySchedule(key)) { - executor.submit(() -> { - Thread.currentThread().setName("node-thread-" + nodeId); - try { - executorFunc.execute(graph.getNode(nodeId)); - } catch (Exception e) { - if (ExceptionUtil.getRootCause(e) instanceof InterruptedException) { - return; - } - log.error("节点执行异常：instId={}, nodeId={}, err={}", instId, nodeId, e.getMessage(), e); - } - }); - } - }*/ /** * 提交节点执行任务（除循环只调度一次） */ @@ -159,7 +106,7 @@ public class FlowTaskScheduler { FlowGraph graph, FlowItemExecutor.NodeExecutor executorFunc, List iterations) { - String key = FlowSchedulerCache.buildKey(instId, nodeId, iterations); + String key = TaskKeyBuilder.buildKey(instId, nodeId, iterations); // 如果已经调度过，则不再重复调度 if (schedulerCache.trySchedule(key)) { @@ -183,7 +130,7 @@ public class FlowTaskScheduler { FlowItemExecutor.NodeExecutor executorFunc, List iterations) { - String key = FlowSchedulerCache.buildKey(instId, nodeId, iterations); + String key = TaskKeyBuilder.buildKey(instId, nodeId, iterations); if (schedulerCache.trySchedule(key)) { executor.submit(() -> { try { diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowNodeParamPreparer.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowNodeParamPreparer.java index f43cab0..3949605 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowNodeParamPreparer.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowNodeParamPreparer.java @@ -1,17 +1,22 @@ package com.cmvr.test.flow.runtime.engine.support; +import cn.hutool.core.collection.CollUtil; +import com.alibaba.fastjson2.JSONArray; import com.alibaba.fastjson2.JSONObject; +import com.cmvr.common.exception.GlobalException; import com.cmvr.test.enums.NodeTypeEnum; +import com.cmvr.test.enums.ParamScope; import com.cmvr.test.flow.builder.FlowGraph; import com.cmvr.test.flow.builder.FlowNodeWrapper; import com.cmvr.test.flow.builder.FlowParamDef; -import com.cmvr.test.flow.builder.FlowParamHelper; import com.cmvr.test.flow.context.TaskContext; import com.cmvr.test.flow.context.TaskInstHolder; import com.cmvr.test.flow.runtime.message.TaskNodeExecuteMessage; import lombok.RequiredArgsConstructor; import org.springframework.stereotype.Component; +import java.lang.reflect.Array; +import java.util.Collection; import java.util.List; @Component @@ -20,35 +25,144 @@ public class FlowNodeParamPreparer { private final TaskInstHolder taskInstHolder; - /** - * 准备节点输入参数（自动处理 START 类型） - */ - public JSONObject prepare(FlowGraph graph, FlowNodeWrapper node, String startNodeId, - List startInputDefs, TaskNodeExecuteMessage rootMessage) { + public JSONObject prepare(FlowGraph graph, + FlowNodeWrapper node, + String startNodeId, + List startInputDefs, + TaskNodeExecuteMessage rootMessage) { - JSONObject inputParams = FlowParamHelper.buildInputParams( - node.getNodeParams(), - startNodeId, - startInputDefs, - refNodeId -> { + + JSONObject input = new JSONObject(); + TaskContext ctx = taskInstHolder.getContext(rootMessage.getInstId()); + + // 遍历当前节点定义的参数 + for (FlowParamDef param : node.getNodeParams()) { + if (param.getScope() != ParamScope.NODE) continue; + + // 1 静态 input 参数 + if ("input".equals(param.getType())) { + input.put(param.getName(), param.getInput()); + } + + // 2 引用参数（quote） + else if ("quote".equals(param.getType()) && CollUtil.isNotEmpty(param.getQuote())) { + String refNodeId = param.getQuote().get(0); // 被引用的上游节点 ID + List path = param.getQuote().subList(2, param.getQuote().size()); // 取值路径 + + JSONObject source; + if ("input".equalsIgnoreCase(param.getQuoteType())) { if (refNodeId.equals(startNodeId)) { - return startInputDefs; + // Start 节点：优先取 runParams（实际运行时输入） + source = ctx.getRunParams(); + if (source == null || source.isEmpty()) { + // 如果运行时没有，就回退到定义的 startInputDefs + source = toJson(startInputDefs); + } + } else { + // 普通节点：取定义参数 + source = toJson(graph.getNodeParamsDefs().get(refNodeId)); } - return graph.getNodeParamsDefs().get(refNodeId); + } else if ("output".equalsIgnoreCase(param.getQuoteType())) { + // 2.2 引用上游的 output 参数（运行时实际执行结果） + // 这里支持循环：用 nodeId + iterations 作为 key，避免覆盖 + source = ctx.getNodeOutput(refNodeId, rootMessage.getIterations()); + } else { + throw new GlobalException("参数应用类型出错！quoteType:" + param.getQuoteType()); } - ); - if (node.getNodeType() == NodeTypeEnum.START || node.getNodeType() == NodeTypeEnum.BRANCH) { - TaskContext ctx = taskInstHolder.getContext(rootMessage.getInstId()); + // 按路径取嵌套值，例如 ["result", "image", "url"] + Object val = findNestedValue(source, path); + input.put(param.getName(), val); + } + } + + // 3 START 节点：用 runParams 覆盖默认参数 + if (node.getNodeType() == NodeTypeEnum.START) { + JSONObject merged = new JSONObject(input); JSONObject runParams = ctx.getRunParams(); - JSONObject merged = new JSONObject(); - for (String key : FlowParamHelper.toJson(startInputDefs).keySet()) { - if (key.equals("terminalId")) continue; - Object override = runParams.get(key); - merged.put(key, override != null ? override : inputParams.get(key)); + for (String key : runParams.keySet()) { + // runParams 里有值，就覆盖掉定义里的值 + merged.put(key, runParams.get(key)); } return merged; } - return inputParams; + + // 4 LOOP 节点：动态计算 loopCount + if (node.getNodeType() == NodeTypeEnum.LOOP) { + Object loopNumVal = input.get("loopNum"); + int loopCount = 0; + + if (loopNumVal instanceof Number) { + loopCount = ((Number) loopNumVal).intValue(); + } else if (loopNumVal instanceof JSONArray) { + // 支持 JSON 数组 + loopCount = ((JSONArray) loopNumVal).size(); + } + else if (loopNumVal instanceof Collection) { + // 支持所有 Java Collection + loopCount = ((Collection) loopNumVal).size(); + } + else if (loopNumVal != null && loopNumVal.getClass().isArray()) { + // 支持 Java 数组 + loopCount = Array.getLength(loopNumVal); + } else if (loopNumVal != null) { + try { + loopCount = Integer.parseInt(loopNumVal.toString()); + } catch (NumberFormatException e) { + throw new GlobalException("loopNum 参数格式错误: " + loopNumVal); + } + } + + // 写回 inputParams，供 LoopHandler 使用 + input.put("loopNum", loopCount); + } + + return input; + } + + + /** + * 将参数定义列表中的 input 值构建为 JSONObject + */ + private JSONObject toJson(List defs) { + JSONObject result = new JSONObject(); + if (CollUtil.isEmpty(defs)) return result; + + for (FlowParamDef param : defs) { + if (param.getType().equalsIgnoreCase(ParamScope.INPUT.name())) { + result.put(param.getName(), param.getInput()); + continue; + } + + String name = param.getName(); + String type = param.getType(); + Object input = param.getInput(); + + if ("Object".equals(type) && CollUtil.isNotEmpty(param.getChildren())) { + result.put(name, toJson(param.getChildren())); + } else if ("Array".equals(type) && CollUtil.isNotEmpty(param.getChildren())) { + JSONArray arr = new JSONArray(); + // 默认数组内只有一个结构体模型 + arr.add(toJson(param.getChildren())); + result.put(name, arr); + } else { + result.put(name, input); + } + } + + return result; + } + + /** + * JSON 多层路径提取 + */ + private Object findNestedValue(JSONObject source, List path) { + if (source == null || CollUtil.isEmpty(path)) return null; + Object curr = source; + for (String key : path) { + if (!(curr instanceof JSONObject)) return null; + curr = ((JSONObject) curr).get(key); + } + return curr; } } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowSchedulerCache.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowSchedulerCache.java index 4b08217..e2977bc 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowSchedulerCache.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowSchedulerCache.java @@ -1,9 +1,7 @@ package com.cmvr.test.flow.runtime.engine.support; -import cn.hutool.core.util.StrUtil; import org.springframework.stereotype.Component; -import java.util.List; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -29,18 +27,4 @@ public class FlowSchedulerCache { scheduled.keySet().removeIf(k -> k.startsWith(instId + "_")); completedPreMap.keySet().removeIf(k -> k.startsWith(instId + "_")); } - - public static String buildKey(String instId, String nodeId, int loopIteration) { - return loopIteration >= 0 - ? StrUtil.format("{}_{}_{}", instId, nodeId, loopIteration) - : StrUtil.format("{}_{}", instId, nodeId); - } - - public static String buildKey(String instId, String nodeId, List iterations) { - StringBuilder sb = new StringBuilder(instId).append(":").append(nodeId); - for (int it : iterations) { - sb.append(":").append(it); - } - return sb.toString(); - } } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowInputPrepareInterceptor.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowInputPrepareInterceptor.java new file mode 100644 index 0000000..49fa8fb --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowInputPrepareInterceptor.java @@ -0,0 +1,53 @@ +package com.cmvr.test.flow.runtime.interceptor; + +import com.alibaba.fastjson2.JSONObject; +import com.cmvr.test.flow.builder.FlowGraph; +import com.cmvr.test.flow.builder.FlowNodeWrapper; +import com.cmvr.test.flow.builder.FlowParamDef; +import com.cmvr.test.flow.runtime.engine.support.FlowNodeParamPreparer; +import com.cmvr.test.flow.runtime.message.TaskNodeExecuteMessage; +import com.cmvr.test.flow.runtime.message.TaskNodeExecuteResult; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.core.annotation.Order; +import org.springframework.stereotype.Component; + +import java.util.List; +import java.util.function.Function; + +/** + * 节点输入参数准备拦截器 + * 在执行 OperatorService 前，利用 FlowNodeParamPreparer 生成 inputParams + */ +@Slf4j +@Component +@Order(2) +@RequiredArgsConstructor +public class FlowInputPrepareInterceptor implements FlowMsgPreInterceptor { + + private final FlowNodeParamPreparer paramPreparer; + + @Override + public TaskNodeExecuteResult intercept(TaskNodeExecuteMessage message, + Function next) { + FlowGraph graph = message.getGraph(); + FlowNodeWrapper node = graph.getNode(message.getNodeId()); + String startNodeId = graph.isSub() ? graph.findSubStartNodeId().getNodeId(): graph.findStartNodeId(); + List startInputDefs = graph.getStartInputParamsDefs(); + + // 调用已有的 prepare 方法 + JSONObject inputParams = paramPreparer.prepare( + graph, + node, + startNodeId, + startInputDefs, + message + ); + + message.setInputParams(inputParams); + log.debug("[InputPrepare] 节点 {} 准备输入参数: {}", message.getNodeId(), inputParams); + + // 执行下一个拦截器 / OperatorService + return next.apply(message); + } +} diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowLoggingInterceptor.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowLoggingInterceptor.java index f568a19..9f0e665 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowLoggingInterceptor.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowLoggingInterceptor.java @@ -15,7 +15,7 @@ import java.util.function.Function; @Slf4j @Component -@Order(2) +@Order(3) @RequiredArgsConstructor public class FlowLoggingInterceptor implements FlowMsgPreInterceptor { diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowOutputStoreInterceptor.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowOutputStoreInterceptor.java new file mode 100644 index 0000000..0a7287b --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowOutputStoreInterceptor.java @@ -0,0 +1,44 @@ +package com.cmvr.test.flow.runtime.interceptor; + +import com.alibaba.fastjson2.JSONObject; +import com.cmvr.test.flow.context.TaskContext; +import com.cmvr.test.flow.context.TaskInstHolder; +import com.cmvr.test.flow.runtime.message.TaskNodeExecuteMessage; +import com.cmvr.test.flow.runtime.message.TaskNodeExecuteResult; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.core.annotation.Order; +import org.springframework.stereotype.Component; + +import java.util.function.Function; + +/** + * 节点输出统一存储拦截器 + * 在节点执行结束后，把输出参数存入 TaskContext，支持循环迭代路径 + */ +@Slf4j +@Component +@Order(4) +@RequiredArgsConstructor +public class FlowOutputStoreInterceptor implements FlowMsgPreInterceptor { + + private final TaskInstHolder taskInstHolder; + + @Override + public TaskNodeExecuteResult intercept(TaskNodeExecuteMessage message, + Function next) { + + TaskNodeExecuteResult result = next.apply(message); + + if (result != null && result.isSuccess()) { + TaskContext ctx = taskInstHolder.getContext(message.getInstId()); + if (ctx != null) { + JSONObject outputParams = result.getOutputParams(); + ctx.setNodeOutput(message.getNodeId(), message.getIterations(), outputParams); + log.debug("[OutputStore] 保存节点输出: instId={}, nodeId={}, iterations={}, output={}", + message.getInstId(), message.getNodeId(), message.getIterations(), outputParams); + } + } + return result; + } +} diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/message/TaskNodeExecuteMessage.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/message/TaskNodeExecuteMessage.java index 1d81d69..b887a37 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/message/TaskNodeExecuteMessage.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/message/TaskNodeExecuteMessage.java @@ -90,4 +90,9 @@ public class TaskNodeExecuteMessage { */ private JSONObject upStreamOutput; + /** + * 循环参数(项目执行时用到) + */ + private JSONObject loopDetail; + } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeCameraOperateService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeCameraOperateService.java index fb3f3d1..2b9d8a9 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeCameraOperateService.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeCameraOperateService.java @@ -1,14 +1,11 @@ package com.cmvr.test.flow.runtime.operator.edge; -import cn.hutool.core.util.ObjUtil; import cn.hutool.core.util.StrUtil; -import com.alibaba.fastjson2.JSONArray; import com.alibaba.fastjson2.JSONObject; import com.cmvr.common.enums.FileType; import com.cmvr.common.exception.GlobalException; import com.cmvr.edge.client.service.EdgeCameraService; import com.cmvr.test.enums.ActionEnum; -import com.cmvr.test.flow.builder.FlowNodeWrapper; import com.cmvr.test.flow.context.TaskInstHolder; import com.cmvr.test.flow.runtime.message.TaskNodeExecuteMessage; import com.cmvr.test.flow.runtime.message.TaskNodeExecuteResult; @@ -29,6 +26,11 @@ public class EdgeCameraOperateService implements EdgeOperateService { @Override public TaskNodeExecuteResult execute(TaskNodeExecuteMessage message) { +// try { +// Thread.sleep(3000); +// } catch (InterruptedException e) { +// throw new RuntimeException(e); +// } String instId = message.getInstId(); String nodeId = message.getNodeId(); ActionEnum action = message.getAction(); @@ -67,35 +69,34 @@ public class EdgeCameraOperateService implements EdgeOperateService { String imageUrl = "http://192.168.1.100:9000/cmvr-iot/IMAGE/20250901/1756718594725.jpg"; // String imageUrl = edgeCameraService.getRGBDImages(terminalId, deviceId); - JSONArray imageUrls = new JSONArray(); - if (ObjUtil.isNotEmpty(upstreamOutput)) { - JSONArray upstream = upstreamOutput.getJSONArray("imageUrl"); - if (ObjUtil.isNotEmpty(upstream)) { - imageUrls.addAll(upstream); - } - } - - imageUrls.add(imageUrl); - output.put("imageUrl", imageUrls); +// JSONArray imageUrls = new JSONArray(); +// if (ObjUtil.isNotEmpty(upstreamOutput)) { +// JSONArray upstream = upstreamOutput.getJSONArray("imageUrl"); +// if (ObjUtil.isNotEmpty(upstream)) { +// imageUrls.addAll(upstream); +// } +// } +// +// imageUrls.add(imageUrl); + output.put("imageUrl", imageUrl); output.put("type", FileType.IMAGE.code()); break; } - case CAMERA_RECORDING_START: { String videoUrl = edgeCameraService.startRecording(terminalId, deviceId); videoUrl = StrUtil.format("{}/{}", "http://10.148.108.162/system/video", StrUtil.subAfter(videoUrl, "/", true)); // String videoUrl = "http://10.148.108.162/system/video/cam4_1752225205349.mp4"; - JSONArray videoUrls = new JSONArray(); - if (ObjUtil.isNotEmpty(upstreamOutput)) { - JSONArray upstream = upstreamOutput.getJSONArray("videoUrl"); - if (ObjUtil.isNotEmpty(upstream)) { - videoUrls.addAll(upstream); - } - } - - videoUrls.add(videoUrl); - output.put("videoUrl", videoUrls); +// JSONArray videoUrls = new JSONArray(); +// if (ObjUtil.isNotEmpty(upstreamOutput)) { +// JSONArray upstream = upstreamOutput.getJSONArray("videoUrl"); +// if (ObjUtil.isNotEmpty(upstream)) { +// videoUrls.addAll(upstream); +// } +// } +// +// videoUrls.add(videoUrl); + output.put("videoUrl", videoUrl); output.put("isPlay", false); output.put("type", FileType.VIDEO.code()); @@ -104,29 +105,28 @@ public class EdgeCameraOperateService implements EdgeOperateService { case CAMERA_RECORDING_STOP: { edgeCameraService.stopRecording(terminalId, deviceId); - - // 获取视频录制开始节点的id - FlowNodeWrapper startRecordNode = message.getGraph() - .getNodeMap() - .values() - .stream() - .filter(node -> ActionEnum.CAMERA_RECORDING_START.equals(node.getAction())) - .findFirst() - .orElseThrow(() -> new IllegalStateException("未找到视频录制开始节点")); - - // 从上下文获取上游节点存储的参数 获取视频地址 - JSONObject pre = taskInstHolder.getNodeOutParams(instId, startRecordNode.getNodeId()); - JSONArray videoUrl = pre.getJSONArray("videoUrl"); - output.put("videoUrl", videoUrl); - output.put("isPlay", true); - output.put("type", FileType.VIDEO.code()); + // todo 结束录像获取视频路径输出 +// // 获取视频录制开始节点的id +// FlowNodeWrapper startRecordNode = message.getGraph() +// .getNodeMap() +// .values() +// .stream() +// .filter(node -> ActionEnum.CAMERA_RECORDING_START.equals(node.getAction())) +// .findFirst() +// .orElseThrow(() -> new IllegalStateException("未找到视频录制开始节点")); +// +// // 从上下文获取上游节点存储的参数 获取视频地址 +// JSONObject pre = taskInstHolder.getNodeOutParams(instId, startRecordNode.getNodeId()); +// JSONArray videoUrl = pre.getJSONArray("videoUrl"); +// output.put("videoUrl", videoUrl); +// output.put("isPlay", true); +// output.put("type", FileType.VIDEO.code()); break; } default: throw new GlobalException("不支持的相机操作类型: " + action); } - taskInstHolder.setNodeOutParams(instId, nodeId, output); return TaskNodeExecuteResult.success(output); } } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeMicrophoneOperateService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeMicrophoneOperateService.java index fbe7800..5dffd28 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeMicrophoneOperateService.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeMicrophoneOperateService.java @@ -16,7 +16,6 @@ import org.springframework.stereotype.Service; public class EdgeMicrophoneOperateService implements EdgeOperateService { private final EdgeMicrophoneService edgeMicrophoneService; - private final TaskInstHolder taskInstHolder; @Override public boolean supports(ActionEnum action) { @@ -31,7 +30,6 @@ public class EdgeMicrophoneOperateService implements EdgeOperateService { JSONObject inputParams = message.getInputParams(); String deviceId = inputParams.getString("deviceId"); String terminalId = message.getTerminalId(); -// JSONObject upstreamOutput = message.getUpStreamOutput(); JSONObject output = new JSONObject(); @@ -53,7 +51,6 @@ public class EdgeMicrophoneOperateService implements EdgeOperateService { throw new GlobalException("不支持的麦克风操作类型: " + action); } - taskInstHolder.setNodeOutParams(instId, nodeId, output); return TaskNodeExecuteResult.success(output); } } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeSpeakerOperateService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeSpeakerOperateService.java index 6c2fc0e..e75ccfc 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeSpeakerOperateService.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeSpeakerOperateService.java @@ -1,6 +1,5 @@ package com.cmvr.test.flow.runtime.operator.edge; -import cmvr.api.SpeakerCommand; import com.alibaba.fastjson2.JSONObject; import com.cmvr.common.exception.GlobalException; import com.cmvr.edge.client.service.EdgeSpeakerService; @@ -11,9 +10,6 @@ import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; -import java.util.List; -import java.util.Random; - @Slf4j @Service @RequiredArgsConstructor @@ -34,32 +30,33 @@ public class EdgeSpeakerOperateService implements EdgeOperateService { String terminalId = message.getTerminalId(); switch (action) { case SPEAKER_PLAYAUDIO: { + log.info("speaker play audio"); // todo 音频暂时随机播放 - List all = edgeSpeakerService.getAll(); +// List all = edgeSpeakerService.getAll(); - for (int i = 0; i < 3; i++) { - int index = new Random().nextInt(all.size()); - String audioUrl = "/home/share/assets/upload/" + all.get(index); - edgeSpeakerService.playAudio(terminalId, deviceId, audioUrl); - // 等待播放完成 - while (true) { - try { - Thread.sleep(200); - } catch (InterruptedException e) { - throw new RuntimeException(e); - } - SpeakerCommand.GetSpeakerStateCommand.Feedback status = edgeSpeakerService.getStatus(terminalId, deviceId); - boolean isRunning = status.getState().getIsRunning(); - if (!isRunning) { - break; - } - } - try { - Thread.sleep(500); - } catch (InterruptedException e) { - throw new RuntimeException(e); - } - } +// for (int i = 0; i < 3; i++) { +// int index = new Random().nextInt(all.size()); +// String audioUrl = "/home/share/assets/upload/" + all.get(index); +// edgeSpeakerService.playAudio(terminalId, deviceId, audioUrl); +// // 等待播放完成 +// while (true) { +// try { +// Thread.sleep(200); +// } catch (InterruptedException e) { +// throw new RuntimeException(e); +// } +// SpeakerCommand.GetSpeakerStateCommand.Feedback status = edgeSpeakerService.getStatus(terminalId, deviceId); +// boolean isRunning = status.getState().getIsRunning(); +// if (!isRunning) { +// break; +// } +// } +// try { +// Thread.sleep(500); +// } catch (InterruptedException e) { +// throw new RuntimeException(e); +// } +// } break; } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/llm/LLMTouchOperateService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/llm/LLMTouchOperateService.java index d9f6cd0..eafc5f7 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/llm/LLMTouchOperateService.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/llm/LLMTouchOperateService.java @@ -9,7 +9,6 @@ import com.cmvr.common.exception.GlobalException; import com.cmvr.edge.client.service.EdgeHlcService; import com.cmvr.llm.service.LLMTouchService; import com.cmvr.test.enums.ActionEnum; -import com.cmvr.test.flow.builder.FlowNodeWrapper; import com.cmvr.test.flow.context.TaskInstHolder; import com.cmvr.test.flow.runtime.message.TaskNodeExecuteMessage; import com.cmvr.test.flow.runtime.message.TaskNodeExecuteResult; @@ -38,7 +37,6 @@ public class LLMTouchOperateService implements LLMOperateService { @Override public TaskNodeExecuteResult execute(TaskNodeExecuteMessage message) { ActionEnum action = message.getAction(); - String instId = message.getInstId(); JSONObject inputParams = message.getInputParams(); String deviceId = inputParams.getString("deviceId"); String terminalId = message.getTerminalId(); @@ -49,17 +47,23 @@ public class LLMTouchOperateService implements LLMOperateService { String targetFeature = inputParams.getString("targetFeature"); String manufacturer = inputParams.getString("manufacturer"); String vehType = inputParams.getString("vehType"); - // 图片节点的id - FlowNodeWrapper imageNode = message.getGraph() - .getNodeMap() - .values() - .stream() - .filter(node -> ActionEnum.CAMERA_GETRGBIMAGE.equals(node.getAction())) - .findFirst() - .orElseThrow(() -> new IllegalStateException("未找到获取图片节点")); - JSONObject pre = taskInstHolder.getNodeOutParams(instId, imageNode.getNodeId()); - JSONArray imageUrl = pre.getJSONArray("imageUrl"); - String url = imageUrl.getString(imageUrl.size() - 1); +// // 图片节点的id +// FlowNodeWrapper imageNode = message.getGraph() +// .getNodeMap() +// .values() +// .stream() +// .filter(node -> ActionEnum.CAMERA_GETRGBIMAGE.equals(node.getAction())) +// .findFirst() +// .orElseThrow(() -> new IllegalStateException("未找到获取图片节点")); +// JSONObject pre = taskInstHolder.getNodeOutParams(instId, imageNode.getNodeId()); +// JSONArray imageUrl = pre.getJSONArray("imageUrl"); +// String url = imageUrl.getString(imageUrl.size() - 1); + + // 直接从 inputParams 获取上游相机的输出（单个 imageUrl） + String url = inputParams.getString("imageUrl"); + if (StrUtil.isEmpty(url)) { + throw new IllegalStateException("未找到上游相机输出的图片地址"); + } String bucketName = minioProps.getBucketName(); String prefix = "/" + bucketName + "/"; String objectName = StrUtil.removePrefix(url, StrUtil.subBefore(url, prefix, true) + prefix); diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/vi/VIOperateService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/vi/VIOperateService.java new file mode 100644 index 0000000..eef9c6d --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/vi/VIOperateService.java @@ -0,0 +1,12 @@ +package com.cmvr.test.flow.runtime.operator.vi; + +import com.cmvr.test.enums.ActionEnum; +import com.cmvr.test.flow.runtime.message.TaskNodeExecuteMessage; +import com.cmvr.test.flow.runtime.message.TaskNodeExecuteResult; + +public interface VIOperateService { + + boolean supports(ActionEnum action); + + TaskNodeExecuteResult execute(TaskNodeExecuteMessage message); +} diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/vi/ViCorpusOperateService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/vi/ViCorpusOperateService.java new file mode 100644 index 0000000..3ba345b --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/vi/ViCorpusOperateService.java @@ -0,0 +1,62 @@ +package com.cmvr.test.flow.runtime.operator.vi; + +import com.alibaba.fastjson2.JSONArray; +import com.alibaba.fastjson2.JSONObject; +import com.cmvr.common.exception.GlobalException; +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 lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; + +import java.util.List; + +@Slf4j +@Service +@RequiredArgsConstructor +public class ViCorpusOperateService implements VIOperateService { + + @Override + public boolean supports(ActionEnum action) { + return action.name().startsWith("VI_CORPUS"); + } + + @Override + public TaskNodeExecuteResult execute(TaskNodeExecuteMessage message) { + ActionEnum action = message.getAction(); + JSONObject inputParams = message.getInputParams(); + List iterations = message.getIterations(); + + String audioPath; + switch (action) { + case VI_CORPUS_WAKE: + JSONObject wakeCorpus = inputParams.getJSONObject("wakeCorpus"); + audioPath = wakeCorpus.getString("audioPath"); + break; + case VI_CORPUS_SINGLE: + JSONArray singleCorpus = inputParams.getJSONArray("testCorpus"); + // 单次对话只有一层循环 + Integer i = iterations.get(0); + int index = i - 1; + JSONObject singleCorpusObj = singleCorpus.getJSONObject(index); + audioPath = singleCorpusObj.getString("audioPath"); + break; + case VI_CORPUS_CONTINUOUS: + // List> + JSONArray continuousCorpus = inputParams.getJSONArray("testCorpus"); + + int idx1 = iterations.get(0) - 1; // 外层 + int idx2 = iterations.get(1) - 1; // 内层 + JSONArray continuous = continuousCorpus.getJSONArray(idx1); + audioPath = continuous.getJSONObject(idx2).getString("audioPath"); + break; + default: + throw new GlobalException("不存在的语音交互action=" + action); + + + } + JSONObject output = new JSONObject().fluentPut("audioPath", audioPath); + return TaskNodeExecuteResult.success(output); + } +} \ No newline at end of file diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/vi/ViNodeOperateHandler.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/vi/ViNodeOperateHandler.java new file mode 100644 index 0000000..36c8681 --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/vi/ViNodeOperateHandler.java @@ -0,0 +1,28 @@ +package com.cmvr.test.flow.runtime.operator.vi; + +import com.cmvr.common.exception.GlobalException; +import com.cmvr.test.flow.runtime.message.TaskNodeExecuteMessage; +import com.cmvr.test.flow.runtime.message.TaskNodeExecuteResult; +import com.cmvr.test.flow.runtime.operator.AbstractNodeOperateHandler; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; + +import java.util.List; + +@Slf4j +@Component("VI") +@RequiredArgsConstructor +public class ViNodeOperateHandler extends AbstractNodeOperateHandler { + + private final List viOperateServices; + + @Override + protected TaskNodeExecuteResult doExecute(TaskNodeExecuteMessage message) { + return viOperateServices.stream() + .filter(s -> s.supports(message.getAction())) + .findFirst() + .map(s -> s.execute(message)) + .orElseThrow(() -> new GlobalException("不支持的语音交互操作: " + message.getAction())); + } +} diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/vi/ViPlayOperateService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/vi/ViPlayOperateService.java new file mode 100644 index 0000000..7fd33bb --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/vi/ViPlayOperateService.java @@ -0,0 +1,54 @@ +package com.cmvr.test.flow.runtime.operator.vi; + +import cmvr.api.SpeakerCommand; +import com.alibaba.fastjson2.JSONObject; +import com.cmvr.common.exception.GlobalException; +import com.cmvr.edge.client.service.EdgeSpeakerService; +import com.cmvr.test.enums.ActionEnum; +import com.cmvr.test.flow.runtime.message.TaskNodeExecuteMessage; +import com.cmvr.test.flow.runtime.message.TaskNodeExecuteResult; +import com.cmvr.test.flow.runtime.operator.edge.EdgeOperateService; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; + +@Slf4j +@Service +@RequiredArgsConstructor +public class ViPlayOperateService implements VIOperateService { + + private final EdgeSpeakerService edgeSpeakerService; + + @Override + public boolean supports(ActionEnum action) { + return action.name().startsWith("VI_PLAY"); + } + + @Override + public TaskNodeExecuteResult execute(TaskNodeExecuteMessage message) { + ActionEnum action = message.getAction(); + if (!action.equals(ActionEnum.VI_PLAY_CORPUS)) { + throw new GlobalException("播放语料的action不是VI_PLAY_CORPUS"); + } + JSONObject inputParams = message.getInputParams(); + String deviceId = inputParams.getString("deviceId"); + String audioPath = inputParams.getString("audioPath"); + String terminalId = message.getTerminalId(); + log.info("播放语料成功,deviceId={},audioPath={},terminalId={}", deviceId, audioPath, terminalId); +// edgeSpeakerService.playAudio(terminalId, deviceId, audioPath); +// // 等待播放完成 +// while (true) { +// try { +// Thread.sleep(1000); +// } catch (InterruptedException e) { +// throw new RuntimeException(e); +// } +// SpeakerCommand.GetSpeakerStateCommand.Feedback status = edgeSpeakerService.getStatus(terminalId, deviceId); +// boolean isRunning = status.getState().getIsRunning(); +// if (!isRunning) { +// break; +// } +// } + return TaskNodeExecuteResult.success(); + } +} diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/vi/ViSchemeOperateService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/vi/ViSchemeOperateService.java new file mode 100644 index 0000000..cc70591 --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/vi/ViSchemeOperateService.java @@ -0,0 +1,100 @@ +package com.cmvr.test.flow.runtime.operator.vi; + +import cn.hutool.core.collection.CollUtil; +import cn.hutool.core.util.ObjUtil; +import cn.hutool.core.util.StrUtil; +import com.alibaba.fastjson2.JSONArray; +import com.alibaba.fastjson2.JSONObject; +import com.cmvr.common.exception.GlobalException; +import com.cmvr.test.enums.ActionEnum; +import com.cmvr.test.flow.runtime.message.TaskNodeExecuteMessage; +import com.cmvr.test.flow.runtime.message.TaskNodeExecuteResult; +import com.cmvr.test.service.ex.ExViCorpusService; +import com.cmvr.test.service.ex.ExViSchemeService; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.jetbrains.annotations.NotNull; +import org.springframework.stereotype.Service; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +@Slf4j +@Service +@RequiredArgsConstructor +public class ViSchemeOperateService implements VIOperateService { + + private final ExViSchemeService exViSchemeService; + private final ExViCorpusService exViCorpusService; + + @Override + public boolean supports(ActionEnum action) { + return action.name().startsWith("VI_SCHEME"); + } + + @Override + public TaskNodeExecuteResult execute(TaskNodeExecuteMessage message) { + ActionEnum action = message.getAction(); + JSONObject inputParams = message.getInputParams(); + String schemeId = inputParams.getString("schemeId"); + + JSONObject scheme = exViSchemeService.queryById(schemeId); + if (ObjUtil.isEmpty(scheme)) { + throw new GlobalException(StrUtil.format("id为 [{}] 的方案为空", schemeId)); + } + int sceneType = scheme.getIntValue("sceneType"); + String wakeId = scheme.getString("wakeId"); + JSONObject wakeCorpus = exViCorpusService.queryById(wakeId); + if (ObjUtil.isEmpty(wakeCorpus)) { + throw new GlobalException(StrUtil.format("id为 [{}] 的唤醒语料为空", wakeId)); + } + + JSONObject output = new JSONObject(); + output.put("wakeCorpus", wakeCorpus); + switch (sceneType) { + case 1: + break; + case 2: + JSONArray singleCorpus = getTestCorpus(scheme, schemeId); + output.put("testCorpus", singleCorpus); + break; + case 3: + JSONArray continuousCorpus = getTestCorpus(scheme, schemeId); + // 转成 List + List list = continuousCorpus.stream() + .map(o -> (JSONObject) o) + .collect(Collectors.toList()); + + // 按 parentId 分组（保留整个 JSONObject） + Map> grouped = list.stream() + .collect(Collectors.groupingBy( + obj -> obj.getString("parentId") + )); + + // 转成 List> + List> continuousCorpusGroup = new ArrayList<>(grouped.values()); + + output.put("testCorpus", continuousCorpusGroup); + break; + default: + throw new GlobalException("不存在的语音交互action=" + action); + } + return TaskNodeExecuteResult.success(output); + } + + @NotNull + private JSONArray getTestCorpus(JSONObject scheme, String schemeId) { + String corpusIdsStr = scheme.getString("corpusIds"); + if (StrUtil.isEmpty(corpusIdsStr)) { + throw new GlobalException(StrUtil.format("id为 [{}] 的方案下,corpusIds为空", schemeId)); + } + List corpusIds = StrUtil.split(corpusIdsStr, StrUtil.COMMA); + JSONArray testCorpus = exViCorpusService.queryInIds(corpusIds); + if (CollUtil.isEmpty(testCorpus)) { + throw new GlobalException(StrUtil.format("id为 [{}] 的测试语料不存在", corpusIdsStr)); + } + return testCorpus; + } +} \ No newline at end of file diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/model/dto/TeQueryTaskDetailDTO.java b/cmvr-iot-test/src/main/java/com/cmvr/test/model/dto/TeQueryTaskDetailDTO.java index 17673bb..eefefd5 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/model/dto/TeQueryTaskDetailDTO.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/model/dto/TeQueryTaskDetailDTO.java @@ -1,5 +1,6 @@ package com.cmvr.test.model.dto; +import com.alibaba.fastjson2.JSONObject; import lombok.Data; @Data @@ -9,4 +10,6 @@ public class TeQueryTaskDetailDTO { private String config; private Integer orderNum; private String isDeploy; + private String sceneType; + private JSONObject schemeInfo; } diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/model/vo/TeTaskExecuteProjectVO.java b/cmvr-iot-test/src/main/java/com/cmvr/test/model/vo/TeTaskExecuteProjectVO.java new file mode 100644 index 0000000..467ed06 --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/model/vo/TeTaskExecuteProjectVO.java @@ -0,0 +1,24 @@ +package com.cmvr.test.model.vo; + +import com.alibaba.fastjson2.JSONArray; +import com.alibaba.fastjson2.JSONObject; +import io.swagger.annotations.ApiModel; +import io.swagger.annotations.ApiModelProperty; +import lombok.Data; + +import javax.validation.constraints.NotEmpty; + +@Data +@ApiModel("项目执行VO") +public class TeTaskExecuteProjectVO { + + @ApiModelProperty("项目ID") + @NotEmpty(message = "项目ID不能为空") + private String projectId; + + @ApiModelProperty(value = "运行时参数") + private JSONObject runParams = new JSONObject(); + + @ApiModelProperty(value = "项目运行时方案及语料信息") + private JSONArray prjInfo; +} diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/service/ex/ExViCorpusService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/service/ex/ExViCorpusService.java new file mode 100644 index 0000000..0cfaf49 --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/service/ex/ExViCorpusService.java @@ -0,0 +1,16 @@ +package com.cmvr.test.service.ex; + +import com.alibaba.fastjson2.JSONArray; +import com.alibaba.fastjson2.JSONObject; + +import java.util.List; + +/** + * 外部-语音交互-语料管理接口 + */ +public interface ExViCorpusService { + + public JSONObject queryById(String id); + + public JSONArray queryInIds(List corpusIds); +} diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/service/ex/ExViSchemeService.java b/cmvr-iot-test/src/main/java/com/cmvr/test/service/ex/ExViSchemeService.java new file mode 100644 index 0000000..47ebebc --- /dev/null +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/service/ex/ExViSchemeService.java @@ -0,0 +1,11 @@ +package com.cmvr.test.service.ex; + +import com.alibaba.fastjson2.JSONObject; + +/** + * 外部-语音交互-方案管理接口 + */ +public interface ExViSchemeService { + + public JSONObject queryById(String id); +} diff --git a/cmvr-iot-test/src/main/java/com/cmvr/test/service/impl/TeTaskOrchestrationServiceImpl.java b/cmvr-iot-test/src/main/java/com/cmvr/test/service/impl/TeTaskOrchestrationServiceImpl.java index 74a1ed2..0b55048 100644 --- a/cmvr-iot-test/src/main/java/com/cmvr/test/service/impl/TeTaskOrchestrationServiceImpl.java +++ b/cmvr-iot-test/src/main/java/com/cmvr/test/service/impl/TeTaskOrchestrationServiceImpl.java @@ -94,6 +94,7 @@ public class TeTaskOrchestrationServiceImpl extends ServiceImpl wrapper = new MPJLambdaWrapper<>(); wrapper .selectAs(TeTaskOrchestration::getItemId, TeQueryTaskDetailDTO::getDetectItemId) + .selectAs(TeDetectionItem::getSceneCode, TeQueryTaskDetailDTO::getSceneType) .select(TeTaskOrchestration::getOrderNum) .select(TeDetectionItem::getFlowData) .eq(TeTaskOrchestration::getTaskId, taskId) diff --git a/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/IViCorpusService.java b/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/IViCorpusService.java index 884a27a..c16de53 100644 --- a/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/IViCorpusService.java +++ b/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/IViCorpusService.java @@ -85,4 +85,15 @@ public interface IViCorpusService extends IService { int updateContinuousViCorpus(ViContinuousCorpusVO viContinuousCorpusVO); int deleteViCorpusByParentIds(String[] parentIds); + + + /** + * 根据parentId eq查询 + */ + public ViCorpus queryCorpusByParentId(String parentId); + + /** + * 根据parentId in查询 + */ + public List queryCorpusInParentIds(List parentIds); } diff --git a/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/IViProjectService.java b/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/IViProjectService.java index 4c371bd..61ddddd 100644 --- a/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/IViProjectService.java +++ b/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/IViProjectService.java @@ -1,6 +1,8 @@ package com.cmvr.vi.service; import com.baomidou.mybatisplus.extension.service.IService; +import com.cmvr.test.model.vo.TeTaskExecuteNormalVO; +import com.cmvr.test.model.vo.TeTaskExecuteProjectVO; import com.cmvr.vi.model.domain.ViProject; import java.util.List; @@ -54,5 +56,5 @@ public interface IViProjectService extends IService { /** * 执行项目 */ - public String execute(String projectId); + public String execute(TeTaskExecuteProjectVO taskExecuteProjectVO); } diff --git a/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/IViSchemeService.java b/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/IViSchemeService.java index e6fdfda..1359043 100644 --- a/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/IViSchemeService.java +++ b/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/IViSchemeService.java @@ -50,4 +50,9 @@ public interface IViSchemeService extends IService { * @return 结果 */ public int deleteViSchemeBySchemeIds(String[] schemeIds); + + /** + * 根据projectId查询方案列表 + */ + public List querySchemesByProjectId(String projectId); } diff --git a/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/ex/ExViCorpusServiceImpl.java b/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/ex/ExViCorpusServiceImpl.java new file mode 100644 index 0000000..4d01f1f --- /dev/null +++ b/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/ex/ExViCorpusServiceImpl.java @@ -0,0 +1,30 @@ +package com.cmvr.vi.service.ex; + +import com.alibaba.fastjson2.JSONArray; +import com.alibaba.fastjson2.JSONObject; +import com.cmvr.test.service.ex.ExViCorpusService; +import com.cmvr.vi.model.domain.ViCorpus; +import com.cmvr.vi.service.IViCorpusService; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; + +import java.util.List; + +@Service +@RequiredArgsConstructor +public class ExViCorpusServiceImpl implements ExViCorpusService { + + private final IViCorpusService viCorpusService; + + @Override + public JSONObject queryById(String id) { + ViCorpus viCorpus = viCorpusService.queryCorpusByParentId(id); + return JSONObject.from(viCorpus); + } + + @Override + public JSONArray queryInIds(List corpusIds) { + List viCorpus = viCorpusService.queryCorpusInParentIds(corpusIds); + return JSONArray.from(viCorpus); + } +} diff --git a/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/ex/ExViSchemeServiceImpl.java b/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/ex/ExViSchemeServiceImpl.java new file mode 100644 index 0000000..5f0f7ee --- /dev/null +++ b/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/ex/ExViSchemeServiceImpl.java @@ -0,0 +1,21 @@ +package com.cmvr.vi.service.ex; + +import com.alibaba.fastjson2.JSONObject; +import com.cmvr.test.service.ex.ExViSchemeService; +import com.cmvr.vi.model.domain.ViScheme; +import com.cmvr.vi.service.IViSchemeService; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; + +@Service +@RequiredArgsConstructor +public class ExViSchemeServiceImpl implements ExViSchemeService { + + private final IViSchemeService ViSchemeService; + + @Override + public JSONObject queryById(String id) { + ViScheme viScheme = ViSchemeService.selectViSchemeBySchemeId(id); + return JSONObject.from(viScheme); + } +} diff --git a/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/impl/IViCorpusServiceImpl.java b/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/impl/IViCorpusServiceImpl.java index d7d0620..73a0c8e 100644 --- a/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/impl/IViCorpusServiceImpl.java +++ b/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/impl/IViCorpusServiceImpl.java @@ -155,7 +155,7 @@ public class IViCorpusServiceImpl extends ServiceImpl String parentId = IdUtils.fastSimpleUUID(); List children = viContinuousCorpusVO.getChildren(); children.forEach(v -> v.setParentId(parentId)); - return this.insertBatch(children); + return this.saveBatch(children); } @Override @@ -220,4 +220,18 @@ public class IViCorpusServiceImpl extends ServiceImpl ); } + @Override + public ViCorpus queryCorpusByParentId(String parentId) { + LambdaQueryWrapper wrapper = Wrappers.lambdaQuery(); + wrapper.eq(ViCorpus::getParentId, parentId); + return this.getOne(wrapper); + } + + @Override + public List queryCorpusInParentIds(List parentIds) { + LambdaQueryWrapper wrapper = Wrappers.lambdaQuery(); + wrapper.in(ViCorpus::getParentId, parentIds) + .orderByAsc(ViCorpus::getSortOrder); + return this.list(wrapper); + } } diff --git a/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/impl/ViProjectServiceImpl.java b/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/impl/ViProjectServiceImpl.java index 74658cb..edecc52 100644 --- a/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/impl/ViProjectServiceImpl.java +++ b/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/impl/ViProjectServiceImpl.java @@ -1,15 +1,30 @@ package com.cmvr.vi.service.impl; +import cn.hutool.core.collection.CollUtil; import cn.hutool.core.util.ObjUtil; +import cn.hutool.core.util.StrUtil; +import com.alibaba.fastjson2.JSONArray; +import com.alibaba.fastjson2.JSONObject; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.baomidou.mybatisplus.core.toolkit.Wrappers; import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; +import com.cmvr.test.enums.RunModeEnum; +import com.cmvr.test.flow.runtime.engine.FlowTaskRuntimeService; +import com.cmvr.test.model.domain.TeTaskConfigInfo; +import com.cmvr.test.model.vo.TeTaskExecuteProjectVO; +import com.cmvr.test.service.ITeTaskConfigInfoService; import com.cmvr.vi.mapper.ViProjectMapper; +import com.cmvr.vi.model.domain.ViCorpus; import com.cmvr.vi.model.domain.ViProject; +import com.cmvr.vi.model.domain.ViScheme; +import com.cmvr.vi.service.IViCorpusService; import com.cmvr.vi.service.IViProjectService; +import com.cmvr.vi.service.IViSchemeService; +import lombok.RequiredArgsConstructor; import org.springframework.stereotype.Service; import java.util.Arrays; +import java.util.Collections; import java.util.List; /** @@ -18,8 +33,14 @@ import java.util.List; * @author cmvr-iot */ @Service +@RequiredArgsConstructor public class ViProjectServiceImpl extends ServiceImpl implements IViProjectService { + private final FlowTaskRuntimeService flowTaskRuntimeService; + private final IViSchemeService viSchemeService; + private final IViCorpusService viCorpusService; + private final ITeTaskConfigInfoService taskConfigInfoService; + @Override public ViProject selectViProjectByProjectId(String projectId) { return this.getById(projectId); @@ -35,7 +56,14 @@ public class ViProjectServiceImpl extends ServiceImpl viSchemes = viSchemeService.querySchemesByProjectId(projectId); - return ""; + JSONArray jsonArray = new JSONArray(); + for (ViScheme scheme : viSchemes) { + // 唤醒语料 + ViCorpus wakeCorpus = viCorpusService.queryCorpusByParentId(scheme.getWakeId()); + + // 测试语料 + List testCorpus = Collections.emptyList(); + if (scheme.getSceneType() != 1) { + List corpusIds = StrUtil.split(scheme.getCorpusIds(), StrUtil.COMMA); + if (CollUtil.isNotEmpty(corpusIds)) { + testCorpus = viCorpusService.queryCorpusInParentIds(corpusIds); + } + } + + JSONObject object = new JSONObject() + .fluentPut("scheme", scheme) + .fluentPut("wake", wakeCorpus) + .fluentPut("test", testCorpus); + + jsonArray.fluentAdd(object); + } + taskExecuteProjectVO.setPrjInfo(jsonArray); + return flowTaskRuntimeService.executeProjectTask(taskExecuteProjectVO); } } diff --git a/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/impl/ViSchemeServiceImpl.java b/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/impl/ViSchemeServiceImpl.java index 465cf09..5b51735 100644 --- a/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/impl/ViSchemeServiceImpl.java +++ b/cmvr-iot-vi/src/main/java/com/cmvr/vi/service/impl/ViSchemeServiceImpl.java @@ -1,5 +1,6 @@ package com.cmvr.vi.service.impl; +import cn.hutool.core.collection.CollUtil; import cn.hutool.core.util.ObjUtil; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.baomidou.mybatisplus.core.toolkit.Wrappers; @@ -17,6 +18,7 @@ import com.cmvr.vi.service.IViSchemeService; import lombok.RequiredArgsConstructor; import org.springframework.stereotype.Service; +import java.util.ArrayList; import java.util.Arrays; import java.util.Comparator; import java.util.List; @@ -77,10 +79,13 @@ public class ViSchemeServiceImpl extends ServiceImpl i TeQueryTaskOrchestraItemVO itemVO = new TeQueryTaskOrchestraItemVO(); itemVO.setItemId(itemId); itemVO.setOrderNum(sort); + if (CollUtil.isEmpty(items)) { + items = new ArrayList<>(); + } items.add(itemVO); - // 落库并归一化 - normalizeTaskItems(taskId); + // 保存并归一化 + saveAndNormalizeItems(taskId, items); return insert; } @@ -91,8 +96,9 @@ public class ViSchemeServiceImpl extends ServiceImpl i String taskId = viScheme.getProjectId(); - // 这里不管 sort 是否改动，都做一次归一化，避免乱序 - normalizeTaskItems(taskId); + // 更新后，直接查询并归一化 + List items = teTaskOrchestrationService.queryByTaskId(taskId); + saveAndNormalizeItems(taskId, items); return update; } @@ -130,19 +136,26 @@ public class ViSchemeServiceImpl extends ServiceImpl i // 移除这些 item items.removeIf(i -> deleteItemIds.contains(i.getItemId())); - // 覆盖保存并归一化 - normalizeTaskItems(taskId); + // 保存并归一化 + saveAndNormalizeItems(taskId, items); } } return delete; } + @Override + public List querySchemesByProjectId(String projectId) { + LambdaQueryWrapper wrapper = Wrappers.lambdaQuery(); + wrapper.eq(ViScheme::getProjectId, projectId) + .orderByAsc(ViScheme::getSort); + return this.list(wrapper); + } + /** - * 根据 taskId 对编排表的检测项重新排序（1..N） + * 保存并归一化 taskId 的编排项 */ - private void normalizeTaskItems(String taskId) { - List items = teTaskOrchestrationService.queryByTaskId(taskId); + private void saveAndNormalizeItems(String taskId, List items) { if (items == null || items.isEmpty()) { return; } @@ -167,4 +180,5 @@ public class ViSchemeServiceImpl extends ServiceImpl i teTaskOrchestrationService.insertTeTaskOrchestration(orchestraVO); } + }