[update] 运行评价-干线路口评价增加绿灯空放时长

signal-optimize-service/src/main/java/net/wanji/opt/service/impl/MainlineEvaluateServiceImpl.java

@@ -14,16 +14,19 @@ import net.wanji.databus.dao.mapper.*;
 import net.wanji.databus.po.*;
 import net.wanji.databus.vo.CrossIdAndNameVO;
 import net.wanji.opt.bo.*;
+import net.wanji.opt.common.KafkaConsumerUtil;
 import net.wanji.opt.dao.mapper.strategy.SceneStrategyMapper;
 import net.wanji.opt.dao.mapper.strategy.StrategyMapper;
 import net.wanji.opt.dto.GreenwaveInSectionDTO;
 import net.wanji.opt.dto.LineSchemeDTO;
+import net.wanji.opt.dto.PhaseEmptyResult;
 import net.wanji.opt.po.strategy.SceneStrategyPO;
 import net.wanji.opt.po.strategy.StrategyPO;
 import net.wanji.opt.service.MainlineEvaluateService;
 import net.wanji.opt.vo.*;
 import org.jetbrains.annotations.NotNull;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 
 import java.math.BigDecimal;
@@ -42,6 +45,12 @@ import java.util.stream.Collectors;
 @Slf4j
 @Service
 public class MainlineEvaluateServiceImpl implements MainlineEvaluateService {
+    @Value("${spring.kafka.bootstrap-servers}")
+    private String bootstrapServers;
+
+    @Value("${spring.kafka.empty-phase-topic}")
+    private String emptyPhaseTopic;
+
     private final BaseCrossDirInfoMapper baseCrossDirInfoMapper;
     private final BaseCrossTurnInfoMapper baseCrossTurnInfoMapper;
     private final LaneInfoMapper laneInfoMapper;
@@ -881,7 +890,18 @@ public class MainlineEvaluateServiceImpl implements MainlineEvaluateService {
         List<BaseCrossDirInfoPO> baseCrossDirInfoPOList =
                 baseCrossDirInfoMapper.selectByInOutType(CrossInOutEnum.IN.getCode());
 
+        // 从 Kafka 获取绿灯空放数据
+        KafkaConsumerUtil kafkaConsumerUtil =
+                new KafkaConsumerUtil(bootstrapServers, "mainline-empty-phase-comsumer");
+        List<PhaseEmptyResult> phaseEmptyResults =
+                kafkaConsumerUtil.consumeEmptyPhaseForTimeRange(
+                        emptyPhaseTopic, 0, startDate.getTime(), endDate.getTime());
+
         for (String crossId : crossIdList) {
+            List<PhaseEmptyResult> crossEmptyPhaseList = phaseEmptyResults.stream().
+                    filter(p -> crossId.equals(p.getCrossId()))
+                    .collect(Collectors.toList());
+
             MainlineCrossEvaluateVO mainlineCrossEvaluateVO = new MainlineCrossEvaluateVO();
 
             Optional<BaseCrossInfoPO> result = baseCrossInfoPOList.stream()
@@ -929,7 +949,7 @@ public class MainlineEvaluateServiceImpl implements MainlineEvaluateService {
             mainlineCrossEvaluateVO.setNonCoordValue(
                     calcNonCoordValue(coordDirCodeList, crossId, crossDirDataHistPOList, baseCrossDirInfoPOList));
             mainlineCrossEvaluateVO.setDirElementList(buildDirElementList(
-                    crossId, metricName, baseCrossDirInfoPOList, crossDirDataHistPOList));
+                    crossId, metricName, baseCrossDirInfoPOList, crossDirDataHistPOList, crossEmptyPhaseList));
             mainlineCrossEvaluateVO.setCrossEvaluateList(buildCrossEvaluateList(
                     crossId, baseCrossDirInfoPOList, crossDirDataHistPOList));
             res.add(mainlineCrossEvaluateVO);
@@ -985,7 +1005,7 @@ public class MainlineEvaluateServiceImpl implements MainlineEvaluateService {
 
     private List<MainlineCrossEvaluateVO.DirElement> buildDirElementList(
             String crossId, String metricName, List<BaseCrossDirInfoPO> baseCrossDirInfoPOList,
-            List<CrossDirDataHistPO> crossDirDataHistPOList) {
+            List<CrossDirDataHistPO> crossDirDataHistPOList, List<PhaseEmptyResult> crossEmptyPhaseList) {
 
         // 获取路口所有方向
         List<BaseCrossDirInfoPO> collect = baseCrossDirInfoPOList.stream()
@@ -1005,14 +1025,20 @@ public class MainlineEvaluateServiceImpl implements MainlineEvaluateService {
                     .filter(item -> 1 == item.getInOutType())
                     .filter(item -> dirCode.equals(item.getDirType()))
                     .collect(Collectors.toList());
-            dirElement.setValue(calcDirValue(metricName, filteredList));
+
+            List<PhaseEmptyResult> dirPhaseEmptyList = crossEmptyPhaseList.stream()
+                    .filter(p -> dirCode.equals(Integer.parseInt(p.getDirection())))
+                    .collect(Collectors.toList());
+
+            dirElement.setValue(calcDirValue(metricName, filteredList, dirPhaseEmptyList));
 
             res.add(dirElement);
         }
         return res;
     }
 
-    private Integer calcDirValue(String metricName, List<CrossDirDataHistPO> crossDirDataHistPOList) {
+    private Integer calcDirValue(String metricName, List<CrossDirDataHistPO> crossDirDataHistPOList,
+                                 List<PhaseEmptyResult> dirPhaseEmptyList) {
         int dirValue = 0;
         if (Objects.equals(metricName, StrategyAndMetricsEnum.Metrics.GREEN_LIGHT_EFFICIENCY.getDescription())) {
             double average = crossDirDataHistPOList.stream()
@@ -1044,6 +1070,30 @@ public class MainlineEvaluateServiceImpl implements MainlineEvaluateService {
                     .max()
                     .orElse(0.0);
             dirValue = (int) Math.round(max);
+        } else if (Objects.equals(metricName, StrategyAndMetricsEnum.Metrics.EFFUSION_RATE.getDescription())) {
+            double average = crossDirDataHistPOList.stream()
+                    .mapToDouble(CrossDirDataHistPO::getEffusionRate)
+                    .average()
+                    .orElse(0.0);
+            dirValue = (int) Math.round(average * 100);
+        } else if (Objects.equals(metricName, StrategyAndMetricsEnum.Metrics.EMPTY_PHASE.getDescription())) {
+            for (CrossDirDataHistPO po : crossDirDataHistPOList) {
+                // 计算结束时间
+                Date endTime = new Date(po.getStartTime().getTime() + 5 * 60 * 1000);  // 增加5分钟
+
+                // 遍历每个 PhaseEmptyResult 来查找和累加 duration
+                for (PhaseEmptyResult phaseEmptyResult : dirPhaseEmptyList) {
+                    if (phaseEmptyResult.getDetectTime() * 1000 >= po.getStartTime().getTime()
+                            && phaseEmptyResult.getDetectTime() * 1000 <= endTime.getTime()) {
+                        // 累加 duration
+                        po.setEmptyPhase(po.getEmptyPhase() + phaseEmptyResult.getDuration());
+                    }
+                }
+            }
+            double sum = crossDirDataHistPOList.stream()
+                    .mapToDouble(CrossDirDataHistPO::getEmptyPhase)
+                    .sum();
+            dirValue = (int) Math.round(sum);
         }
         return dirValue;
     }

wj-databus/src/main/java/net/wanji/databus/po/CrossDirDataRealtimePO.java

@@ -65,4 +65,6 @@ public class CrossDirDataRealtimePO {
     public Double effusionRate;
     @ApiModelProperty(value = "绿灯有效利用率", notes = "")
     public Double greenLightEfficiency;
+    @ApiModelProperty(value = "绿灯空放时长")
+    private Integer emptyPhase = 0;
 }