mlx

simulation-resource-scheduler/pom.xml

@@ -119,6 +119,12 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.13.2</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
     <build>
         <plugins>

simulation-resource-scheduler/src/main/java/com/css/simulation/resource/scheduler/consumer/ManualProjectConsumer.java

@@ -37,6 +37,8 @@ import java.util.List;
 @Slf4j
 public class ManualProjectConsumer {
 
+    private final String USER_ID = "simulation-resource-scheduler";
+
     @Autowired
     ProjectMapper projectMapper;
     @Autowired
@@ -73,6 +75,16 @@ public class ManualProjectConsumer {
     @KafkaListener(groupId = "simulation-resource-scheduler", topics = "manualProject")
     public void parseProject(ConsumerRecord<String, String> projectRecord) throws IOException, ApiException {
         //1 读取 kafka 的 project 信息
+        /*
+            {
+                "projectId": "sadfasdfs",	// 项目 id
+                "algorithmId": "sadfasdfs",	// 算法 id
+                "vehicleId": "sadfasdfs",	// 车辆 id
+                "scenePackageId": "sadfasdfs",	// 场景包 id
+                "maxSimulationTime": 11111,	// 最大仿真时间
+                "parallelism": 30		// 并行度
+            }
+         */
         String projectJson = projectRecord.value();
         ProjectMessageDTO projectMessageDTO = JsonUtil.jsonToBean(projectJson, ProjectMessageDTO.class);
         String projectId = projectMessageDTO.getProjectId();    // 项目 id
@@ -133,9 +145,9 @@ public class ManualProjectConsumer {
                     .runResult(resultPath)
                     .build();
             taskPO.setCreateTime(TimeUtil.getNowForMysql());
-            taskPO.setCreateUserId("simulation-resource-scheduler");
+            taskPO.setCreateUserId(USER_ID);
             taskPO.setModifyTime(TimeUtil.getNowForMysql());
-            taskPO.setModifyUserId("simulation-resource-scheduler");
+            taskPO.setModifyUserId(USER_ID);
             taskPO.setModifyTime(TimeUtil.getNowForMysql());
             taskPO.setIsDeleted("0");
             taskMapper.insert(taskPO);

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

@@ -17,6 +17,7 @@ import com.css.simulation.resource.scheduler.pojo.po.TaskIndexPO;
 import com.css.simulation.resource.scheduler.pojo.po.TaskPO;
 import com.css.simulation.resource.scheduler.pojo.to.ScoreTO;
 import feign.Response;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
@@ -28,6 +29,7 @@ import java.util.*;
 import java.util.stream.Collectors;
 
 @Service
+@Slf4j
 public class TaskService {
 
     private final String USER_ID = "simulation-resource-scheduler";
@@ -79,16 +81,16 @@ public class TaskService {
         String scenePackageId = projectPO.getScenePackageId();  // 场景测试包 id，指标根 id
         int taskNum = projectMapper.selectTaskNumById(projectId);
         int completedTaskNum = projectMapper.selectTaskNumByProjectIdAndState(projectId, DictConstants.TASK_COMPLETED);
-        if (taskNum != completedTaskNum) {
+        if (taskNum != completedTaskNum) {  // 已完成任务数等于所有任务数量，才会准备打分；否则退出。
             return;
-        }  //3 如果已完成任务数等于所有任务数量，调用打分程序，对项目中每个指标进行打分，
+        }
         List<TaskPO> taskList = taskMapper.selectTaskListByProjectId(projectId);  // 所有任务信息
+        // -------------------------------- 查询叶子指标 --------------------------------
         List<IndexTemplatePO> leafIndexTemplateList = indexTemplateMapper.selectLeafIndexWithRuleDetailsByPackageId(scenePackageId);
         List<TaskIndexPO> leafTaskIndexList = new ArrayList<>();
-
-        // 计算所有叶子节点指标的得分
         for (IndexTemplatePO indexTemplatePO : leafIndexTemplateList) {
-            String ruleDetails = indexTemplatePO.getRuleDetails();
+//            String ruleDetails = indexTemplatePO.getRuleDetails();    // 打分脚本路径
+            // -------------------------------- 查询每个叶子指标包括的场景 --------------------------------
             Set<String> sceneIdSet = new HashSet<>();
             String naturalIds = indexTemplatePO.getSceneNaturalIds();
             String standardIds = indexTemplatePO.getSceneStatueIds();
@@ -106,22 +108,27 @@ public class TaskService {
                 sceneIdSet.addAll(Arrays.asList(accidentIdArray));
             }
             int resultNumberOfCurrentIndex = sceneIdSet.size();
+            // -------------------------------- 计算叶子指标的得分 --------------------------------
             double sum = taskList.stream()
                     .filter(task1 -> sceneIdSet.contains(task1.getSceneId()))
                     .mapToDouble(task2 -> {
                         // 计算每个任务的得分
-                        ScoreTO score = new ScoreTO();
+                        ScoreTO score;
+                        String runResultMinio = task2.getRunResult();
+                        String runResultLinux = manualProjectResultPathLinux + runResultMinio.substring(1);
+                        String command = "python3 " + pyPath + " " + runResultLinux + " " + task2.getSceneType();  // 默认使用场景名称找打分脚本
+//                        String command = "python3 " + pyPath + " " + runResultLinux + " " + task2.getSceneType() + " " + ruleDetails; // 指定打分脚本
                         try {
-                            String runResultMinio = task2.getRunResult();
-                            String runResultLinux = manualProjectResultPathLinux + runResultMinio.substring(1);
                             Response download = commonService.download(MinioParameter.builder().objectName(runResultMinio).build());
                             Response.Body body = download.body();
                             InputStream inputStream = body.asInputStream();
                             FileUtil.writeInputStreamToLocalFile(inputStream, runResultLinux);
-                            String executeResult = SshUtil.execute(hostname, username, password, "python3 " + pyPath + " " + runResultLinux + " " + task2.getSceneType());
-                            score = JsonUtil.jsonToBean(executeResult, ScoreTO.class);
+                            log.info("------- 开始执行打分命令：" + command);
+                            score = JsonUtil.jsonToBean(SshUtil.execute(hostname, username, password, command), ScoreTO.class);
+                            log.info("------- 打分结束，结果为：" + score);
                         } catch (IOException e) {
-                            e.printStackTrace();
+                            log.error("------- 打分出错，命令为：" + command);
+                            throw new RuntimeException();
                         }
                         task2.setReturnSceneId(score.getUnit_scene_ID());
                         task2.setScore(score.getUnit_scene_score());
@@ -131,10 +138,11 @@ public class TaskService {
                         task2.setModifyTime(TimeUtil.getNowForMysql());
                         return score.getUnit_scene_score();
                     }).sum();
-            long notStandardSceneNum = taskList.stream()    // 计算不合格的任务数（不到100分就是不合格）
-                    .filter(task1 -> sceneIdSet.contains(task1.getSceneId()) && task1.getScore() < 100)
-                    .count();
-            double leafIndexScore = sum / resultNumberOfCurrentIndex;   // 叶子指标的得分
+            // 计算不合格的任务数（不到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())
@@ -151,14 +159,14 @@ public class TaskService {
             leafTaskIndexList.add(leafTaskIndex);
         }
 
-        //4 根据每个指标的得分和权重算出 project 的总得分。
+        // 根据每个指标的得分和权重算出 project 的总得分。
         double totalScore = compute(leafIndexTemplateList);
-        //5 保存分数
-        //5-1 保存任务分数
+        // 保存分数
+        // 保存任务分数
         taskManager.batchUpdateByScoreResult(taskList);
-        //5-2 保存指标分数
+        // 保存指标分数
         taskIndexManager.batchInsertLeafIndex(leafTaskIndexList);
-        //5-3 保存总分数
+        // 保存总分数
         TaskIndexPO totalTaskIndex = TaskIndexPO.builder()
                 .id(StringUtil.getRandomUUID())
                 .pId(projectId)

simulation-resource-scheduler/src/test/java/com/css/simulation/resource/scheduler/SchedulerTest.java

@@ -0,0 +1,16 @@
+package com.css.simulation.resource.scheduler;
+
+
+import org.junit.Test;
+
+import java.io.IOException;
+
+public class SchedulerTest {
+
+    @Test
+    public void test() throws IOException {
+
+    }
+
+
+}