mlx

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/mapper/TaskIndexMapper.java

@@ -4,6 +4,8 @@ package com.css.simulation.resource.scheduler.mapper;
 import com.css.simulation.resource.scheduler.pojo.po.TaskIndexPO;
 import org.apache.ibatis.annotations.*;
 
+import java.util.List;
+
 /**
  * 任务指标
  * simulation_mpt_first_target_score
@@ -68,7 +70,7 @@ public interface TaskIndexMapper {
             "    or scene_traffic_ids like #{idExtend}\n" +
             "    or scene_statue_ids like #{idExtend}\n" +
             "    )\n")
-    String selectLeafIndexIdByProjectAndSceneId(@Param("projectId") String projectId, @Param("idExtend") String idExtend);
+    List<String> selectLeafIndexIdByProjectAndSceneId(@Param("projectId") String projectId, @Param("idExtend") String idExtend);
 
 
     @Delete("delete from simulation_mpt_first_target_score where p_id = #{projectId}")

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/mapper/TaskMapper.java

@@ -92,7 +92,12 @@ public interface TaskMapper {
     @Select("select id\n" +
             "from simulation_manual_project_task\n" +
             "where p_id = #{projectId} and scene_id = #{sceneId}")
-    String selectIdByProjectIdAndSceneId(@Param("projectId")String projectId,@Param("sceneId") String sceneId);
+    List<String> selectIdByProjectIdAndSceneId(@Param("projectId")String projectId,@Param("sceneId") String sceneId);
+
+    @Select("select last_targer_id\n" +
+            "from simulation_manual_project_task\n" +
+            "where id = #{taskId}")
+    String selectLastTargetIdById(@Param("taskId") String taskId);
 
     @Update("update simulation_manual_project_task\n" +
             "set run_state = #{runState}\n" +

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/service/TaskService.java

@@ -159,27 +159,12 @@ public class TaskService {
             String ruleFilePath = pyPath + "script/" + ruleName.split("_")[0] + "/" + ruleName + ".py";
             log.info("------- /state 将叶子节点 " + indexId + " 对应的打分规则保存到临时目录：" + ruleFilePath);
             FileUtil.writeInputStreamToLocalFile(IoUtil.stringToInputStream(ruleDetails), ruleFilePath);
-            Set<String> sceneIdSet = new HashSet<>();
-            String naturalIds = indexTemplatePO.getSceneNaturalIds();
-            String standardIds = indexTemplatePO.getSceneStatueIds();
-            String accidentIds = indexTemplatePO.getSceneTrafficIds();
-            if (StringUtil.isNotEmpty(naturalIds)) {
-                String[] naturalIdArray = naturalIds.split(",");
-                sceneIdSet.addAll(Arrays.asList(naturalIdArray));
-            }
-            if (StringUtil.isNotEmpty(standardIds)) {
-                String[] standardArray = standardIds.split(",");
-                sceneIdSet.addAll(Arrays.asList(standardArray));
-            }
-            if (StringUtil.isNotEmpty(accidentIds)) {
-                String[] accidentIdArray = accidentIds.split(",");
-                sceneIdSet.addAll(Arrays.asList(accidentIdArray));
-            }
-            int resultNumberOfCurrentIndex = sceneIdSet.size();
-            log.info("------- /state 叶子节点 " + indexId + " 包括 " + resultNumberOfCurrentIndex + " 个场景！");
+
+            List<TaskPO> taskListOfLeafIndex = taskList.stream().filter(task -> indexId.equals(task.getLastTargetId())).collect(Collectors.toList());
+            int resultNumberOfCurrentIndex = taskListOfLeafIndex.size();
+            log.info("------- /state 叶子节点 " + indexId + " 包括 " + resultNumberOfCurrentIndex + " 个任务！");
             log.info("------- /state 计算叶子节点 " + indexId + " 的得分！");
-            double sum = taskList.stream()
-                    .filter(task1 -> sceneIdSet.contains(task1.getSceneId()))
+            double sum = taskListOfLeafIndex.stream()
                     .mapToDouble(task2 -> {
                         String task2Id = task2.getId();
                         taskMapper.updateState(task2Id, DictConstants.TASK_ANALYSING);
@@ -231,7 +216,7 @@ public class TaskService {
                         return score.getUnit_scene_score();
                     }).sum();
             // 计算不合格的任务数（不到100分就是不合格）
-            long notStandardSceneNum = taskList.stream().filter(task1 -> sceneIdSet.contains(task1.getSceneId()) && task1.getScore() < 100).count();
+            long notStandardSceneNum = taskListOfLeafIndex.stream().filter(task -> task.getScore() < 100).count();
             // 叶子指标的得分（叶子指标下所有场景得分的平均值）
             double leafIndexScore = sum / resultNumberOfCurrentIndex;
             // -------------------------------- 保存叶子指标得分 --------------------------------
@@ -279,12 +264,194 @@ public class TaskService {
         Map<String, String> params = new HashMap<>();
         params.put("id", projectId);
         String post = HttpUtil.post(closeableHttpClient, requestConfig, evaluationLevelUri, headers, params);
-        log.info("------- /state 访问仿真云平台评价等级接口：" + evaluationLevelUri +"，请求头为：" + headers +"，请求体为：" + params + "结果为：" + post);
+        log.info("------- /state 访问仿真云平台评价等级接口：" + evaluationLevelUri + "，请求头为：" + headers + "，请求体为：" + params + "结果为：" + post);
         log.info("------- /state 项目 " + projectId + " 打分完成");
 
 
     }
 
+//    @SneakyThrows
+//    public void taskState(String taskId, String state, String podName) {
+//        log.info("------- /state 修改任务 " + taskId + "的状态：" + state + "，pod 名称为：" + podName);
+//        if ("Running".equals(state)) {
+//            taskMapper.updateStateWithStartTime(taskId, state, TimeUtil.getNowForMysql());
+//        } else if ("Aborted".equals(state) || "Terminated".equals(state)) {
+//            LinuxUtil.execute("kubectl delete pod " + podName);
+//            taskMapper.updateFailStateWithStopTime(taskId, state, TimeUtil.getNowForMysql());
+//        } else if ("PendingAnalysis".equals(state)) {
+//            LinuxUtil.execute("kubectl delete pod " + podName);
+//            taskMapper.updateSuccessStateWithStopTime(taskId, state, TimeUtil.getNowForMysql());
+//        } else {
+//            taskMapper.updateState(taskId, state);
+//        }
+//        ProjectPO projectPO = projectMapper.selectById(taskId);
+//        String projectId = projectPO.getId();
+//        String scenePackageId = projectPO.getScenePackageId();  // 场景测试包 id，指标根 id
+//        log.info("------- /state 任务 " + taskId + " 的父项目为：" + projectId);
+//        int taskNum = projectMapper.selectTaskNumById(projectId);
+//        int endTaskNum = projectMapper.selectEndTaskNum(projectId);    // 查询已结束的任务 'Aborted', 'PendingAnalysis', 'Terminated'
+//        projectMapper.updateTaskCompleted(projectId, endTaskNum);
+//        log.info("------- /state 项目 " + projectId + " 完成进度为：" + endTaskNum + "/" + taskNum);
+//        if (taskNum != endTaskNum) {  // 已结束任务数等于所有任务数量，才会准备打分；否则退出。
+//            return;
+//        }
+//        projectMapper.updateProjectState(projectId, DictConstants.PROJECT_COMPLETED, TimeUtil.getNowForMysql());   // 修改该 project 的状态为已完成
+//        LinuxUtil.execute("kubectl delete job project-" + projectId);
+//        SshClient clientKafka = SshUtil.getClient();
+//        ClientSession sessionKafka = SshUtil.getSession(clientKafka, hostnameKafka, usernameKafka, passwordKafka);
+//        String topicDeleteCommand = StringUtil.replace(kafkaDeleteCommand, "topicName", projectId);
+//        SshUtil.execute(sessionKafka, topicDeleteCommand);
+//        SshUtil.stop(clientKafka, sessionKafka);
+//        List<TaskPO> taskList = taskMapper.selectTaskListByProjectId(projectId);  // 所有任务信息
+//        log.info("------- /state 共有 " + taskList.size() + "个任务！");
+//        taskIndexMapper.deleteFirstByProjectId(projectId);
+//        taskIndexMapper.deleteLastByProjectId(projectId);
+//        // -------------------------------- 查询叶子指标 --------------------------------
+//        List<IndexTemplatePO> leafIndexTemplateList = indexTemplateMapper.selectLeafIndexWithRuleDetailsByPackageId(scenePackageId);
+//        List<TaskIndexPO> leafTaskIndexList = new ArrayList<>();
+//        log.info("------- /state 共有 " + leafIndexTemplateList.size() + "个叶子节点！");
+//
+//        SshClient clientScore = SshUtil.getClient();
+//        ClientSession sessionScore = SshUtil.getSession(clientScore, hostnameScore, usernameScore, passwordScore);
+//        for (int i = 0; i < leafIndexTemplateList.size(); i++) {
+//            AtomicReference<String> scoreExplain = new AtomicReference<>(); // 每个叶子指标下的任务的得分说明一样和叶子指标一致
+//            IndexTemplatePO indexTemplatePO = leafIndexTemplateList.get(i);
+//            String indexId = indexTemplatePO.getIndexId();
+//            String parentId = indexTemplatePO.getParentId();
+//            String rootId = indexTemplatePO.getRootId();
+//            String weight = indexTemplatePO.getWeight();
+//            log.info("------- /state 开始执行对第 " + (i + 1) + " 个叶子节点 " + indexId + " 进行打分！");
+//            String ruleName = indexTemplatePO.getRuleName();    // 打分脚本名称，例如 AEB_1-1
+//            String ruleDetails = indexTemplatePO.getRuleDetails();    // 打分脚本内容
+//            String ruleFilePath = pyPath + "script/" + ruleName.split("_")[0] + "/" + ruleName + ".py";
+//            log.info("------- /state 将叶子节点 " + indexId + " 对应的打分规则保存到临时目录：" + ruleFilePath);
+//            FileUtil.writeInputStreamToLocalFile(IoUtil.stringToInputStream(ruleDetails), ruleFilePath);
+//            Set<String> sceneIdSet = new HashSet<>();
+//            String naturalIds = indexTemplatePO.getSceneNaturalIds();
+//            String standardIds = indexTemplatePO.getSceneStatueIds();
+//            String accidentIds = indexTemplatePO.getSceneTrafficIds();
+//            if (StringUtil.isNotEmpty(naturalIds)) {
+//                String[] naturalIdArray = naturalIds.split(",");
+//                sceneIdSet.addAll(Arrays.asList(naturalIdArray));
+//            }
+//            if (StringUtil.isNotEmpty(standardIds)) {
+//                String[] standardArray = standardIds.split(",");
+//                sceneIdSet.addAll(Arrays.asList(standardArray));
+//            }
+//            if (StringUtil.isNotEmpty(accidentIds)) {
+//                String[] accidentIdArray = accidentIds.split(",");
+//                sceneIdSet.addAll(Arrays.asList(accidentIdArray));
+//            }
+//            int resultNumberOfCurrentIndex = sceneIdSet.size();
+//            log.info("------- /state 叶子节点 " + indexId + " 包括 " + resultNumberOfCurrentIndex + " 个场景！");
+//            log.info("------- /state 计算叶子节点 " + indexId + " 的得分！");
+//            double sum = taskList.stream()
+//                    .filter(task1 -> sceneIdSet.contains(task1.getSceneId()))
+//                    .mapToDouble(task2 -> {
+//                        String task2Id = task2.getId();
+//                        taskMapper.updateState(task2Id, DictConstants.TASK_ANALYSING);
+//                        // 计算每个任务的得分
+//
+//                        String runResultMinio = task2.getRunResultFilePath() + "/Ego.csv";
+//                        String runResultLinux = linuxTempPath + runResultMinio;
+//
+////                        python3 /home/ubuntu/test/Evaluate/main.py /home/ubuntu/test/test_data.csv 4 AEB_1-2
+////                        String command = "python3 " + pyPath + " " + runResultLinux + " " + task2.getSceneType();  // 默认使用场景名称找打分脚本
+//                        String command = "python3 " + pyPath + "main.py " + runResultLinux + " " + task2.getSceneType() + " " + ruleName; // 指定打分脚本
+//                        String scoreResult;
+//                        try {
+//                            try {
+//                                log.info("------- /state 下载 minio 上的结果文件 " + runResultMinio + " 到本地：" + runResultLinux);
+//                                MinioUtil.downloadToFile(minioClient, bucketName, runResultMinio, runResultLinux);  // 也可改成下载到指定ip的服务器上，需要保证和打分脚本在一台机器上。
+//                            } catch (Exception e) {
+//                                throw new RuntimeException("------- /state 下载 minio 上的结果文件出错：" + e.getMessage());
+//                            }
+//                            try {
+//                                log.info("------- /state 开始执行打分命令：" + command);
+//                                scoreResult = SshUtil.execute(sessionScore, command);
+//                                log.info("------- /state 打分结束，结果为：" + scoreResult);
+//                            } catch (IOException e) {
+//                                throw new RuntimeException("------- /state 任务 " + task2Id + " 打分出错，命令为：" + command + " 修改状态为：" + DictConstants.TASK_ABORTED + "\n" + e.getMessage());
+//                            }
+//                        } catch (Exception e) {
+//                            taskMapper.updateState(task2Id, DictConstants.TASK_ABORTED);
+//                            throw new RuntimeException(e.getMessage());
+//                        }
+//                        ScoreTO score = null;
+//                        try {
+//                            String replace = StringUtil.replace(scoreResult, "'", "\"");
+//                            score = JsonUtil.jsonToBean(replace, ScoreTO.class);
+//                        } catch (JsonProcessingException e) {
+//                            e.printStackTrace();
+//                        }
+//                        assert score != null;
+//                        task2.setReturnSceneId(score.getUnit_scene_ID());
+//                        task2.setScore(score.getUnit_scene_score());
+//                        task2.setTargetEvaluate(score.getEvaluate_item());
+//                        task2.setScoreExplain(score.getScore_description());
+//                        task2.setModifyUserId(USER_ID);
+//                        task2.setModifyTime(TimeUtil.getNowForMysql());
+//                        scoreExplain.set(score.getScore_description());
+//
+//                        taskMapper.updateState(task2Id, DictConstants.TASK_COMPLETED);
+//
+//                        return score.getUnit_scene_score();
+//                    }).sum();
+//            // 计算不合格的任务数（不到100分就是不合格）
+//            long notStandardSceneNum = taskList.stream().filter(task1 -> sceneIdSet.contains(task1.getSceneId()) && task1.getScore() < 100).count();
+//            // 叶子指标的得分（叶子指标下所有场景得分的平均值）
+//            double leafIndexScore = sum / resultNumberOfCurrentIndex;
+//            // -------------------------------- 保存叶子指标得分 --------------------------------
+//            indexTemplatePO.setTempScore(leafIndexScore);
+//            TaskIndexPO leafTaskIndex = TaskIndexPO.builder()
+//                    .id(StringUtil.getRandomUUID())
+//                    .pId(projectId)
+//                    .target(indexTemplatePO.getIndexId())
+//                    .notStandardSceneNum((int) notStandardSceneNum)
+//                    .score(leafIndexScore)
+//                    .indexId(indexId)
+//                    .parentId(parentId)
+//                    .rootId(rootId)
+//                    .weight(weight)
+//                    .scoreExplain(scoreExplain.get())
+//                    .build();
+//            leafTaskIndex.setCreateUserId(USER_ID);
+//            leafTaskIndex.setCreateTime(TimeUtil.getNowForMysql());
+//            leafTaskIndex.setModifyUserId(USER_ID);
+//            leafTaskIndex.setModifyTime(TimeUtil.getNowForMysql());
+//            leafTaskIndex.setIsDeleted("0");
+//            leafTaskIndexList.add(leafTaskIndex);
+//        }
+//        SshUtil.stop(clientScore, sessionScore);
+//        // 保存任务分数
+//        taskManager.batchUpdateByScoreResult(taskList);
+//        // 保存末级指标分数
+//        taskIndexManager.batchInsertLeafIndex(leafTaskIndexList);
+//        // 保存一级指标分数
+//        log.info("------- /state 根据每个指标的得分和权重算出各个一级指标的得分（即 project 对应的场景测试包下的一级指标）！");
+//        computeFirst(leafTaskIndexList, projectId);
+//
+//        // 调用 server 的接口，计算评价等级
+//        String tokenUrl = tokenUri + "?grant_type=client_credentials"
+//                + "&client_id=" + clientId
+//                + "&client_secret=" + clientSecret;
+//        log.info("------- /state 获取仿真云平台 token：" + tokenUrl);
+//        String response = HttpUtil.get(closeableHttpClient, requestConfig, tokenUrl);
+//        ObjectMapper objectMapper = new ObjectMapper();
+//        JsonNode jsonNode = objectMapper.readTree(response);
+//        String accessToken = jsonNode.path("access_token").asText();
+//        log.info("------- /state 仿真云平台 token 为：" + accessToken);
+//        Map<String, String> headers = new HashMap<>();
+//        headers.put("Authorization", "Bearer " + accessToken);
+//        Map<String, String> params = new HashMap<>();
+//        params.put("id", projectId);
+//        String post = HttpUtil.post(closeableHttpClient, requestConfig, evaluationLevelUri, headers, params);
+//        log.info("------- /state 访问仿真云平台评价等级接口：" + evaluationLevelUri +"，请求头为：" + headers +"，请求体为：" + params + "结果为：" + post);
+//        log.info("------- /state 项目 " + projectId + " 打分完成");
+//
+//
+//    }
+
 
     public Boolean taskConfirm(String taskId) {
         // 查询 task 如果不是 pending 则不执行