package com.cmvr.test.service; import cn.hutool.http.HttpRequest; import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSONArray; import com.alibaba.fastjson2.JSONObject; import com.cmvr.common.exception.GlobalException; import com.cmvr.common.utils.http.CallAPIUtil; import com.cmvr.common.utils.uuid.IdUtils; import com.cmvr.edge.client.model.EdgeCommonVO; import com.cmvr.edge.client.model.humanoid.EdgeMoveJVO; import com.cmvr.edge.client.service.EdgeBioHeadService; import com.cmvr.edge.client.service.EdgeHumanoidRobotService; import com.cmvr.test.enums.FlowiseActionEnum; import com.cmvr.test.flow.context.TaskContext; import com.cmvr.test.flow.context.TaskContextManager; import com.cmvr.test.flow.control.FlowControlService; import com.cmvr.test.model.vo.FlowiseActionRequestVO; import com.cmvr.test.model.vo.FlowiseChatRequestVO; import com.cmvr.test.model.vo.FlowiseStartRequestVO; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Service; import javax.annotation.Resource; import java.util.Collections; import java.util.HashMap; import java.util.Map; import java.util.concurrent.atomic.AtomicBoolean; @Slf4j @Service @RequiredArgsConstructor public class FlowiseActionService { private final EdgeBioHeadService edgeBioHeadService; private final EdgeHumanoidRobotService edgeHumanoidRobotService; private final TaskContextManager taskContextManager; private final FlowControlService flowControlService; @Value("${flowise.tts}") private String flowiseTts; @Value("${flowise.query}") private String flowiseQuery; @Value("${flowise.start}") private String flowiseStart; @Value("${flowise.abort}") private String flowiseAbort; @Value("${flowise.api-key}") private String flowiseApiKey; @Resource(name = "threadPoolTaskExecutor") private final ThreadPoolTaskExecutor executor; /** * 整个动作是否在运行(张嘴 + 摆头) */ private final AtomicBoolean actionRunning = new AtomicBoolean(true); private final Object lock = new Object(); private String chatId = null; public String start(FlowiseStartRequestVO request) { log.info("flowise start: {}", System.currentTimeMillis()); abort(); synchronized (lock) { if (chatId != null) { throw new GlobalException("Flow is already running: " + chatId); } chatId = IdUtils.fastUUID(); JSONObject body = new JSONObject(); body.put("question", request.getQuestion()); body.put("chatId", chatId); body.put("streaming", true); executor.submit(() -> { try { HttpRequest.post(flowiseStart) .header("Content-Type", "application/json") .header("Authorization", "Bearer " + flowiseApiKey) .body(body.toJSONString()) .timeout(0) .execute() .body(); } catch (Exception e) { throw new GlobalException("flowise 流程执行失败: ", e.getMessage()); } }); return chatId; } } public String abort() { CallAPIUtil.doPostJson(flowiseTts + "stop", null, null); Map instContextMap = taskContextManager.getInstContextMap(); instContextMap.forEach((key, value) -> flowControlService.stop(key)); synchronized (lock) { if (chatId == null) { return "No flow running."; } String id = chatId; String result = HttpRequest.put(flowiseAbort + id) .header("Authorization", "Bearer " + flowiseApiKey) .execute() .body(); chatId = null; return result; } } public String command(FlowiseActionRequestVO request) { log.info("command start: {}", System.currentTimeMillis()); String action = request.getAction(); FlowiseActionEnum actionEnum = FlowiseActionEnum.fromAction(action); String terminalId = "4ed1246c465b97975f96c9ef8371a3bd"; String bioId = "bio_head"; switch (actionEnum) { // ==== 表情 ==== case SMILE: return edgeBioHeadService.specialExpression(terminalId, bioId, 1); case SURPRISED: return edgeBioHeadService.specialExpression(terminalId, bioId, 2); case ANGRY: return edgeBioHeadService.specialExpression(terminalId, bioId, 3); case SAD: return edgeBioHeadService.specialExpression(terminalId, bioId, 4); case SLEEP: return edgeBioHeadService.specialExpression(terminalId, bioId, 5); case YAWN: return edgeBioHeadService.specialExpression(terminalId, bioId, 6); // ==== 动作 ==== default: throw new UnsupportedOperationException("未实现的 Flowise Action: " + action); } } public String chat(FlowiseChatRequestVO requestVO) { log.info("chat start: {}", System.currentTimeMillis()); Map body = new HashMap<>(); body.put("text", requestVO.getText()); body.put("voice", "x4_yezi"); CallAPIUtil.doPostJson(flowiseTts + "play", null, body); return "ok"; } public JSONObject query(String chatId) { String body = HttpRequest.get(flowiseQuery + chatId) .header("Authorization", "Bearer " + flowiseApiKey) .execute() .body(); JSONObject root = JSON.parseObject(body); String execStr = root.getString("executionData"); JSONArray execJson = JSON.parseArray(execStr); root.put("executionData", execJson); return root; } public String actionStart() { // 防止重复启动 if (!actionRunning.compareAndSet(false, true)) { log.warn("action 已在运行中"); return "already running"; } String terminalId = "8c7c5d5d8c8b09df9d1e5fc9b76c46ab"; // String terminalId = "25d13e8ee0cbc2b1046e190f840b8ab0"; String bioId = "hc01"; String spkId = "spk1"; executor.execute(() -> { EdgeCommonVO speakVO = new EdgeCommonVO(); speakVO.setTerminalId(terminalId); speakVO.setDeviceId(spkId); try { // 张嘴 // edgeBioHeadService.speakStart(speakVO); log.info("张嘴动作触发完成"); // 摆头循环 boolean toRight = true; while (actionRunning.get()) { double rad = toRight ? 0.1d : -0.1d; EdgeMoveJVO moveJVO = buildHeadMoveJVO( terminalId, bioId, rad ); long t0 = System.currentTimeMillis(); log.info(">>> moveJ start, rad={},time={}", rad, t0); String res = edgeHumanoidRobotService.moveJ(moveJVO); long t1 = System.currentTimeMillis(); log.info(">>> moveJ end, rad={}, cost={}", rad, t1 - t0); // 判断返回 try { JSONObject resJson = JSON.parseObject(res); if (!resJson.getBooleanValue("success")) { log.warn("moveJ 返回失败,res={}", res); } } catch (Exception ignore) { } toRight = !toRight; } } catch (Exception e) { log.error("action 执行异常", e); } finally { // 回正 try { EdgeMoveJVO resetVO = buildHeadMoveJVO( terminalId, bioId, 0d ); log.info("结束动作,头部回正 rad=0"); edgeHumanoidRobotService.moveJ(resetVO); } catch (Exception e) { log.error("头部回正异常", e); } // 停止张嘴 try { // edgeBioHeadService.speakStop(speakVO); log.info("speakStop 已调用"); } catch (Exception e) { log.error("speakStop 调用失败", e); } actionRunning.set(false); log.info("action 线程结束"); } }); return "ok"; } public String actionStop() { if (!actionRunning.get()) { return "not running"; } log.info("actionStop 调用,准备停止所有动作"); actionRunning.set(false); return "ok"; } private EdgeMoveJVO buildHeadMoveJVO(String terminalId, String deviceId, double rad) { EdgeMoveJVO vo = new EdgeMoveJVO(); vo.setTerminalId(terminalId); vo.setDeviceId(deviceId); vo.setAcc(1d); vo.setVel(1d); EdgeMoveJVO.JointCmd headCmd = new EdgeMoveJVO.JointCmd(); headCmd.setJointName("HEAD_R"); headCmd.setRad(rad); headCmd.setVel(0d); vo.setCmds(Collections.singletonList(headCmd)); return vo; } }