SmartBench/CMVR-IOT
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 语音交互接入流程

Browse Source
This commit is contained in:
stream 2025-09-23 12:35:27 +08:00
parent fa63a8e039
commit 43a3c416a7
48 changed files with 977 additions and 386 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
cmvr-iot-admin/src/main/java/com/cmvr/web/controller/vi/ViProjectController.java
View File

@ -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));
}
}

12
cmvr-iot-test/src/main/java/com/cmvr/test/enums/ActionEnum.java
View File

@ -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", "语音交互-方案处理"),
;

10
cmvr-iot-test/src/main/java/com/cmvr/test/enums/LoopNodeTypeEnum.java Normal file
View File

@ -0,0 +1,10 @@
package com.cmvr.test.enums;
/**
* 循环节点类型
*/
public enum LoopNodeTypeEnum {
NORMAL, // 普通
MULTIPLE_CORPUS, // 多条语料（单次和连续）
CONTINUOUS_MULTIPLE_CORPUS // 连续多条语料
}

5
cmvr-iot-test/src/main/java/com/cmvr/test/enums/RunModeEnum.java
View File

@ -1,6 +1,7 @@
package com.cmvr.test.enums;
public enum RunModeEnum {
NORMAL, //正常
TRIAL //试运行
NORMAL, //正常
TRIAL, //试运行
PROJECT //项目
}

10
cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowModelBuilder.java
View File

@ -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<FlowParamDef> 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;

5
cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowNodeWrapper.java
View File

@ -63,11 +63,6 @@ public class FlowNodeWrapper {
*/
private List<FlowEdge> childNodeEdges = new ArrayList<>();
/**
* 循环次数（如果是循环节点，此字段用于控制循环次数）
*/
private Integer loopCount;
/**
* 子流程（当节点为循环节点或包含子流程时，此字段存储子流程）
*/

7
cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowParamDef.java
View File

@ -51,6 +51,13 @@ public class FlowParamDef {
*/
private List<String> quote;
/**
* 引用的是输出参数还是输入参数
* outputParams -> output
* nodeParams -> input
*/
private String quoteType;
/**
* 参数描述（用于 UI 显示或生成 Schema）
*/

95
cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/FlowParamHelper.java
View File

@ -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<FlowParamDef> 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<FlowParamDef> nodeParams,
String startNodeId,
List<FlowParamDef> 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<String> path = param.getQuote().subList(1, param.getQuote().size());
List<FlowParamDef> 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<String> 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<FlowParamDef> getNodeParams(String nodeId);
}
}

36
cmvr-iot-test/src/main/java/com/cmvr/test/flow/builder/TaskKeyBuilder.java
View File

@ -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<Integer> 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<Integer> iterations) {
return nodeId + "@" + iterations.stream()
.map(String::valueOf)
.collect(Collectors.joining("."));
}
}

33
cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskContext.java
View File

@ -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<String, JSONObject> nodeOutParams = new HashMap<>();
private final Map<String, JSONObject> 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<Integer> 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<Integer> 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);
}
}

11
cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskContextManager.java
View File

@ -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<Integer> 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<Integer> iterations) {
TaskContext context = instContextMap.get(instId);
if (context != null) {
return context.getNodeOutParams(nodeId);
return context.getNodeOutput(nodeId, iterations);
}
return null;
}

10
cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskInstHolder.java
View File

@ -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<Integer> 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<Integer> iterations) {
return taskContextManager.getNodeOutParams(instId, nodeId, iterations);
}
}

22
cmvr-iot-test/src/main/java/com/cmvr/test/flow/context/TaskThreadRegistry.java
View File

@ -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<String, CountDownLatch> 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 {

4
cmvr-iot-test/src/main/java/com/cmvr/test/flow/control/FlowControlService.java
View File

@ -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();

24
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/dispatcher/FlowLoopNodeHandler.java
View File

@ -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<Integer> 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<Integer> newIterations = new ArrayList<>(parentIterations);
// newIterations.add(i);
//
// // 克隆 message，避免覆盖父级
// TaskNodeExecuteMessage subMessage = new TaskNodeExecuteMessage();
// BeanUtil.copyProperties(message, subMessage);
//// subMessage.setLoopNum(i); // 当前层循环次数
// subMessage.setIterations(newIterations); // 完整迭代路径
// 构造新的迭代路径
List<Integer> 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);
}
}

18
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowItemExecutor.java
View File

@ -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<FlowMsgPreInterceptor> 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<Integer> iterations) {
List<Integer> 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));

3
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskEngine.java
View File

@ -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);

46
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskRuntimeEntry.java
View File

@ -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<TeQueryTaskDetailDTO> 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());
}
}
/**
* 创建并保存任务实例
*/

8
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskRuntimeService.java
View File

@ -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);
}

61
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/FlowTaskScheduler.java
View File

@ -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<String> nextNodes = graph.getNextNodes(nodeId);
for (String nextId : nextNodes) {
String key = FlowSchedulerCache.buildKey(instId, nextId, loopIteration);
// 使用 Set 来记录已完成的前驱节点
Set<String> 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<String> nextNodes = graph.getNextNodes(nodeId);
for (String nextId : nextNodes) {
String key = FlowSchedulerCache.buildKey(instId, nextId, iterations);
String key = TaskKeyBuilder.buildKey(instId, nextId, iterations);
// 使用 Set 来记录已完成的前驱节点
Set<String> 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<Integer> 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<Integer> iterations) {
String key = FlowSchedulerCache.buildKey(instId, nodeId, iterations);
String key = TaskKeyBuilder.buildKey(instId, nodeId, iterations);
if (schedulerCache.trySchedule(key)) {
executor.submit(() -> {
try {

158
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowNodeParamPreparer.java
View File

@ -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<FlowParamDef> startInputDefs, TaskNodeExecuteMessage rootMessage) {
public JSONObject prepare(FlowGraph graph,
FlowNodeWrapper node,
String startNodeId,
List<FlowParamDef> 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<String> 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<FlowParamDef> 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<String> 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;
}
}

16
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/engine/support/FlowSchedulerCache.java
View File

@ -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<Integer> iterations) {
StringBuilder sb = new StringBuilder(instId).append(":").append(nodeId);
for (int it : iterations) {
sb.append(":").append(it);
}
return sb.toString();
}
}

53
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowInputPrepareInterceptor.java Normal file
View File

@ -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<TaskNodeExecuteMessage, TaskNodeExecuteResult> next) {
FlowGraph graph = message.getGraph();
FlowNodeWrapper node = graph.getNode(message.getNodeId());
String startNodeId = graph.isSub() ? graph.findSubStartNodeId().getNodeId(): graph.findStartNodeId();
List<FlowParamDef> 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);
}
}

2
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowLoggingInterceptor.java
View File

@ -15,7 +15,7 @@ import java.util.function.Function;
@Slf4j
@Component
@Order(2)
@Order(3)
@RequiredArgsConstructor
public class FlowLoggingInterceptor implements FlowMsgPreInterceptor {

44
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/interceptor/FlowOutputStoreInterceptor.java Normal file
View File

@ -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<TaskNodeExecuteMessage, TaskNodeExecuteResult> 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;
}
}

5
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/message/TaskNodeExecuteMessage.java
View File

@ -90,4 +90,9 @@ public class TaskNodeExecuteMessage {
*/
private JSONObject upStreamOutput;
/**
* 循环参数(项目执行时用到)
*/
private JSONObject loopDetail;
}

82
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeCameraOperateService.java
View File

@ -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);
}
}

3
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeMicrophoneOperateService.java
View File

@ -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);
}
}

53
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/edge/EdgeSpeakerOperateService.java
View File

@ -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<String> all = edgeSpeakerService.getAll();
// List<String> 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;
}

30
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/llm/LLMTouchOperateService.java
View File

@ -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);

12
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/vi/VIOperateService.java Normal file
View File

@ -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);
}

62
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/vi/ViCorpusOperateService.java Normal file
View File

@ -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<Integer> 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<List<JSONObject>>
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);
}
}

28
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/vi/ViNodeOperateHandler.java Normal file
View File

@ -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<VIOperateService> 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()));
}
}

54
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/vi/ViPlayOperateService.java Normal file
View File

@ -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();
}
}

100
cmvr-iot-test/src/main/java/com/cmvr/test/flow/runtime/operator/vi/ViSchemeOperateService.java Normal file
View File

@ -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<JSONObject>
List<JSONObject> list = continuousCorpus.stream()
.map(o -> (JSONObject) o)
.collect(Collectors.toList());
// 按 parentId 分组（保留整个 JSONObject）
Map<String, List<JSONObject>> grouped = list.stream()
.collect(Collectors.groupingBy(
obj -> obj.getString("parentId")
));
// 转成 List<List<JSONObject>>
List<List<JSONObject>> 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<String> corpusIds = StrUtil.split(corpusIdsStr, StrUtil.COMMA);
JSONArray testCorpus = exViCorpusService.queryInIds(corpusIds);
if (CollUtil.isEmpty(testCorpus)) {
throw new GlobalException(StrUtil.format("id为 [{}] 的测试语料不存在", corpusIdsStr));
}
return testCorpus;
}
}

3
cmvr-iot-test/src/main/java/com/cmvr/test/model/dto/TeQueryTaskDetailDTO.java
View File

@ -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;
}

24
cmvr-iot-test/src/main/java/com/cmvr/test/model/vo/TeTaskExecuteProjectVO.java Normal file
View File

@ -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;
}

16
cmvr-iot-test/src/main/java/com/cmvr/test/service/ex/ExViCorpusService.java Normal file
View File

@ -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<String> corpusIds);
}

11
cmvr-iot-test/src/main/java/com/cmvr/test/service/ex/ExViSchemeService.java Normal file
View File

@ -0,0 +1,11 @@
package com.cmvr.test.service.ex;
import com.alibaba.fastjson2.JSONObject;
/**
* 外部-语音交互-方案管理接口
*/
public interface ExViSchemeService {
public JSONObject queryById(String id);
}

1
cmvr-iot-test/src/main/java/com/cmvr/test/service/impl/TeTaskOrchestrationServiceImpl.java
View File

@ -94,6 +94,7 @@ public class TeTaskOrchestrationServiceImpl extends ServiceImpl<TeTaskOrchestrat
MPJLambdaWrapper<TeTaskOrchestration> wrapper = new MPJLambdaWrapper<>();
wrapper
.selectAs(TeTaskOrchestration::getItemId, TeQueryTaskDetailDTO::getDetectItemId)
.selectAs(TeDetectionItem::getSceneCode, TeQueryTaskDetailDTO::getSceneType)
.select(TeTaskOrchestration::getOrderNum)
.select(TeDetectionItem::getFlowData)
.eq(TeTaskOrchestration::getTaskId, taskId)

11
cmvr-iot-vi/src/main/java/com/cmvr/vi/service/IViCorpusService.java
View File

@ -85,4 +85,15 @@ public interface IViCorpusService extends IService<ViCorpus> {
int updateContinuousViCorpus(ViContinuousCorpusVO viContinuousCorpusVO);
int deleteViCorpusByParentIds(String[] parentIds);
/**
* 根据parentId eq查询
*/
public ViCorpus queryCorpusByParentId(String parentId);
/**
* 根据parentId in查询
*/
public List<ViCorpus> queryCorpusInParentIds(List<String> parentIds);
}

4
cmvr-iot-vi/src/main/java/com/cmvr/vi/service/IViProjectService.java
View File

@ -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<ViProject> {
/**
* 执行项目
*/
public String execute(String projectId);
public String execute(TeTaskExecuteProjectVO taskExecuteProjectVO);
}

5
cmvr-iot-vi/src/main/java/com/cmvr/vi/service/IViSchemeService.java
View File

@ -50,4 +50,9 @@ public interface IViSchemeService extends IService<ViScheme> {
* @return 结果
*/
public int deleteViSchemeBySchemeIds(String[] schemeIds);
/**
* 根据projectId查询方案列表
*/
public List<ViScheme> querySchemesByProjectId(String projectId);
}

30
cmvr-iot-vi/src/main/java/com/cmvr/vi/service/ex/ExViCorpusServiceImpl.java Normal file
View File

@ -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<String> corpusIds) {
List<ViCorpus> viCorpus = viCorpusService.queryCorpusInParentIds(corpusIds);
return JSONArray.from(viCorpus);
}
}

21
cmvr-iot-vi/src/main/java/com/cmvr/vi/service/ex/ExViSchemeServiceImpl.java Normal file
View File

@ -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);
}
}

16
cmvr-iot-vi/src/main/java/com/cmvr/vi/service/impl/IViCorpusServiceImpl.java
View File

@ -155,7 +155,7 @@ public class IViCorpusServiceImpl extends ServiceImpl<ViCorpusMapper, ViCorpus>
String parentId = IdUtils.fastSimpleUUID();
List<ViCorpus> 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<ViCorpusMapper, ViCorpus>
);
}
@Override
public ViCorpus queryCorpusByParentId(String parentId) {
LambdaQueryWrapper<ViCorpus> wrapper = Wrappers.lambdaQuery();
wrapper.eq(ViCorpus::getParentId, parentId);
return this.getOne(wrapper);
}
@Override
public List<ViCorpus> queryCorpusInParentIds(List<String> parentIds) {
LambdaQueryWrapper<ViCorpus> wrapper = Wrappers.lambdaQuery();
wrapper.in(ViCorpus::getParentId, parentIds)
.orderByAsc(ViCorpus::getSortOrder);
return this.list(wrapper);
}
}

60
cmvr-iot-vi/src/main/java/com/cmvr/vi/service/impl/ViProjectServiceImpl.java
View File

@ -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<ViProjectMapper, ViProject> 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<ViProjectMapper, ViProject
@Override
public int insertViProject(ViProject viProject) {
return this.baseMapper.insert(viProject);
int insert = this.baseMapper.insert(viProject);
TeTaskConfigInfo taskConfigInfo = new TeTaskConfigInfo();
taskConfigInfo.setRunMode(RunModeEnum.PROJECT.name());
taskConfigInfo.setId(viProject.getProjectId());
taskConfigInfo.setStatus("0");
taskConfigInfo.setTaskName(StrUtil.format("{}_{}", RunModeEnum.PROJECT, viProject.getProjectName()));
taskConfigInfoService.save(taskConfigInfo);
return insert;
}
@Override
@ -45,12 +73,38 @@ public class ViProjectServiceImpl extends ServiceImpl<ViProjectMapper, ViProject
@Override
public int deleteViProjectByProjectIds(String[] projectIds) {
taskConfigInfoService.deleteTeTaskConfigInfoByIds(projectIds);
return this.baseMapper.deleteBatchIds(Arrays.asList(projectIds));
}
@Override
public String execute(String projectId) {
public String execute(TeTaskExecuteProjectVO taskExecuteProjectVO) {
String projectId = taskExecuteProjectVO.getProjectId();
// 查询方案
List<ViScheme> viSchemes = viSchemeService.querySchemesByProjectId(projectId);
return "";
JSONArray jsonArray = new JSONArray();
for (ViScheme scheme : viSchemes) {
// 唤醒语料
ViCorpus wakeCorpus = viCorpusService.queryCorpusByParentId(scheme.getWakeId());
// 测试语料
List<ViCorpus> testCorpus = Collections.emptyList();
if (scheme.getSceneType() != 1) {
List<String> 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);
}
}

32
cmvr-iot-vi/src/main/java/com/cmvr/vi/service/impl/ViSchemeServiceImpl.java
View File

@ -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<ViSchemeMapper, ViScheme> 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<ViSchemeMapper, ViScheme> i
String taskId = viScheme.getProjectId();
// 这里不管 sort 是否改动，都做一次归一化，避免乱序
normalizeTaskItems(taskId);
// 更新后，直接查询并归一化
List<TeQueryTaskOrchestraItemVO> items = teTaskOrchestrationService.queryByTaskId(taskId);
saveAndNormalizeItems(taskId, items);
return update;
}
@ -130,19 +136,26 @@ public class ViSchemeServiceImpl extends ServiceImpl<ViSchemeMapper, ViScheme> i
// 移除这些 item
items.removeIf(i -> deleteItemIds.contains(i.getItemId()));
// 覆盖保存并归一化
normalizeTaskItems(taskId);
// 保存并归一化
saveAndNormalizeItems(taskId, items);
}
}
return delete;
}
@Override
public List<ViScheme> querySchemesByProjectId(String projectId) {
LambdaQueryWrapper<ViScheme> 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<TeQueryTaskOrchestraItemVO> items = teTaskOrchestrationService.queryByTaskId(taskId);
private void saveAndNormalizeItems(String taskId, List<TeQueryTaskOrchestraItemVO> items) {
if (items == null || items.isEmpty()) {
return;
}
@ -167,4 +180,5 @@ public class ViSchemeServiceImpl extends ServiceImpl<ViSchemeMapper, ViScheme> i
teTaskOrchestrationService.insertTeTaskOrchestration(orchestraVO);
}
}