Merge remote-tracking branch 'origin/master'

wj-datacenter-service/src/main/java/net/wanji/datacenter/cache/CrossGreenWaveInfoCache.java

@@ -11,10 +11,7 @@ import org.springframework.util.CollectionUtils;
 
 import javax.annotation.Resource;
 import java.time.LocalTime;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.stream.Collectors;
 
 /**
@@ -51,7 +48,8 @@ public class CrossGreenWaveInfoCache implements InitializingBean {
                 List<GreenwaveCrossPO> greenWaveCrossPOS = greenwaveCrossMapper.selectByGreenwaveId(greenWaveId);
                 if (!CollectionUtils.isEmpty(greenWaveCrossPOS)) {
                     List<GreenwaveCrossPO> greenWaveCrossIds = greenWaveCrossPOS.stream()
-                            .filter(po -> LocalTime.parse(po.getStartTime()).isBefore(now) && LocalTime.parse(po.getEndTime()).isAfter(now))
+                            .filter(po -> LocalTime.parse(po.getStartTime()).isBefore(now)
+                                    && LocalTime.parse(po.getEndTime()).isAfter(now) && Objects.equals(0, po.getWeek()))
                             .sorted(Comparator.comparingInt(GreenwaveCrossPO::getSort))
                             .collect(Collectors.toList());
                     greenWaveInfoMap.put(greenWaveId, greenWaveCrossIds);

wj-datacenter-service/src/main/java/net/wanji/datacenter/cache/CrossSchemePhaseLightsCache.java

@@ -48,6 +48,24 @@ public class CrossSchemePhaseLightsCache implements CommandLineRunner {
                     Integer schemeId = baseCrossSchemePO.getId();
                     List<CrossPhasePO> crossPhasePOList = baseCrossPhaseMapper.selectByCrossIdAndSchemeId(crossId, schemeId);
                     if (!CollectionUtils.isEmpty(crossPhasePOList)) {
+                        buildData(crossId, schemeNo, crossPhasePOList);
+                    }
+                }
+            }
+            log.info("路口缓存方案相位方向转向数据：{}", crossSchemePhaseDirTurnDTOList);
+        } catch (Exception e) {
+            log.error("路口缓存方案相位方向转向关系数据初始化失败", e);
+            throw new Exception(e);
+        }
+    }
+
+    /**
+     * 构建路口方案相位方向转向实体
+     * @param crossId 路口编号
+     * @param schemeNo
+     * @param crossPhasePOList
+     */
+    private void buildData(String crossId, String schemeNo, List<CrossPhasePO> crossPhasePOList) {
         for (CrossPhasePO crossPhasePO : crossPhasePOList) {
             Integer phaseId = crossPhasePO.getId();
             String phaseNo = crossPhasePO.getPhaseNo();
@@ -78,14 +96,6 @@ public class CrossSchemePhaseLightsCache implements CommandLineRunner {
             }
         }
     }
-                }
-            }
-            log.info("路口缓存方案相位方向转向数据：{}", crossSchemePhaseDirTurnDTOList);
-        } catch (Exception e) {
-            log.error("路口缓存方案相位方向转向关系数据初始化失败", e);
-            throw new Exception(e);
-        }
-    }
 
     @Override
     public void run(String... args) throws Exception {

wj-datacenter-service/src/main/java/net/wanji/datacenter/cache/CrossTurnWeekMaxFlowCache.java

@@ -43,6 +43,17 @@ public class CrossTurnWeekMaxFlowCache implements CommandLineRunner {
             queryWrapper.between(CrossTurnDataHistPO::getStartTime, lastWeek, lastDay);
             List<CrossTurnDataHistPO> crossTurnDataHistPOS = crossTurnDataHistMapper.selectList(queryWrapper);
             if (!CollectionUtils.isEmpty(crossTurnDataHistPOS)) {
+                calculateData(crossId, crossTurnDataHistPOS);
+            }
+        }
+    }
+
+    /**
+     * 计算转向一周最大流量
+     * @param crossId
+     * @param crossTurnDataHistPOS
+     */
+    private static void calculateData(String crossId, List<CrossTurnDataHistPO> crossTurnDataHistPOS) {
         Map<Integer, List<CrossTurnDataHistPO>> dirDataPOsMap = crossTurnDataHistPOS.stream().collect(Collectors.groupingBy(CrossTurnDataHistPO::getInDir));
         for (Map.Entry<Integer, List<CrossTurnDataHistPO>> entry : dirDataPOsMap.entrySet()) {
             Integer dir = entry.getKey();
@@ -63,8 +74,6 @@ public class CrossTurnWeekMaxFlowCache implements CommandLineRunner {
             }
         }
     }
-        }
-    }
 
     @Override
     public void run(String... args) throws Exception {

wj-datacenter-service/src/main/java/net/wanji/datacenter/cache/WeekDirFreeFlowSpeedCache.java

@@ -48,6 +48,21 @@ public class WeekDirFreeFlowSpeedCache implements CommandLineRunner {
             queryWrapper.between(CrossDirDataHistPO::getStartTime, lastWeek, lastDay);
             List<CrossDirDataHistPO> crossDirDataHistPOS = crossDirDataHistMapper.selectList(queryWrapper);
             if (!CollectionUtils.isEmpty(crossDirDataHistPOS)) {
+                calculateData(crossDirFreeFlowSpeedDTOS, crossId, crossDirDataHistPOS);
+            }
+        }
+        Map<String, List<CrossDirFreeFlowSpeedDTO>> map = new HashMap<>(1);
+        map.put(Constants.CROSS_FREE_FLOW_SPEED_KEY, crossDirFreeFlowSpeedDTOS);
+        producerHandler.send2FirstKafka(Constants.CROSS_FREE_FLOW_SPEED_TOPIC, map);
+    }
+
+    /**
+     * 计算前一周方向自由流速度85~95分位速度
+     * @param crossDirFreeFlowSpeedDTOS
+     * @param crossId
+     * @param crossDirDataHistPOS
+     */
+    private static void calculateData(List<CrossDirFreeFlowSpeedDTO> crossDirFreeFlowSpeedDTOS, String crossId, List<CrossDirDataHistPO> crossDirDataHistPOS) {
         Map<Integer, List<CrossDirDataHistPO>> dirDataMap = crossDirDataHistPOS.stream().collect(Collectors.groupingBy(CrossDirDataHistPO::getDirType));
         if (!dirDataMap.isEmpty()) {
             for (Map.Entry<Integer, List<CrossDirDataHistPO>> entry : dirDataMap.entrySet()) {
@@ -69,11 +84,6 @@ public class WeekDirFreeFlowSpeedCache implements CommandLineRunner {
             }
         }
     }
-        }
-        Map<String, List<CrossDirFreeFlowSpeedDTO>> map = new HashMap<>(1);
-        map.put(Constants.CROSS_FREE_FLOW_SPEED_KEY, crossDirFreeFlowSpeedDTOS);
-        producerHandler.send2FirstKafka(Constants.CROSS_FREE_FLOW_SPEED_TOPIC, map);
-    }
 
     @Override
     public void run(String... args) throws Exception {

wj-datacenter-service/src/main/java/net/wanji/datacenter/cache/WeekTurnFreeFlowSpeedCache.java

@@ -43,6 +43,16 @@ public class WeekTurnFreeFlowSpeedCache implements CommandLineRunner {
             queryWrapper.between(CrossTurnDataHistPO::getStartTime, lastWeek, lastDay);
             List<CrossTurnDataHistPO> crossTurnDataHistPOS = crossTurnDataHistMapper.selectList(queryWrapper);
             if (!CollectionUtils.isEmpty(crossTurnDataHistPOS)) {
+                calculateData(crossTurnDataHistPOS);
+            }
+        }
+    }
+
+    /**
+     * 计算前一周自由流速度85~95分位速度
+     * @param crossTurnDataHistPOS
+     */
+    private static void calculateData(List<CrossTurnDataHistPO> crossTurnDataHistPOS) {
         Map<Integer, List<CrossTurnDataHistPO>> dirDataMap = crossTurnDataHistPOS.stream().collect(Collectors.groupingBy(CrossTurnDataHistPO::getInDir));
         for (Map.Entry<Integer, List<CrossTurnDataHistPO>> dirEntry : dirDataMap.entrySet()) {
             Integer dir = dirEntry.getKey();
@@ -69,8 +79,6 @@ public class WeekTurnFreeFlowSpeedCache implements CommandLineRunner {
             }
         }
     }
-        }
-    }
 
     @Override
     public void run(String... args) throws Exception {

wj-datacenter-service/src/main/java/net/wanji/datacenter/config/RedisConfig.java

@@ -112,14 +112,10 @@ public class RedisConfig extends CachingConfigurerSupport {
         return new FastJson2JsonRedisSerializer(Object.class);
     }
 
-    @Value("${spring.redis.host}")
-    private String redisHost;
-    @Value("${spring.redis.port}")
-    private Integer redisPort;
     @Bean
     public Jedis jedis() {
-        Jedis jedis = new Jedis(redisHost, redisPort);
-        jedis.auth("Wanji300552");
+        Jedis jedis = new Jedis(host, port);
+        jedis.auth(password);
         return jedis;
     }
 

wj-datacenter-service/src/main/java/net/wanji/datacenter/constant/Constant.java

@@ -32,4 +32,6 @@ public class Constant {
     public static final String REAL_TIME_TURN_DATA_TOPIC= "cross_real_time_turn_data";
     public static final String REAL_TIME_CROSS_DATA_KEY= "crossRealTimeData";
     public static final String REAL_TIME_CROSS_DATA_TOPIC= "cross_real_time_data";
+
+    private Constant() {}
 }

wj-datacenter-service/src/main/java/net/wanji/datacenter/service/impl/LanePeriodicDataProcessServiceImpl.java

@@ -524,7 +524,7 @@ public class LanePeriodicDataProcessServiceImpl implements DataProcessService {
             crossDataIndexDTO.setLoadBalance(Double.valueOf(decimalFormat.format(loadBalance)));
             CrossRealTimeCache.crossDataIndexMap.put(crossId, crossDataIndexDTO);
         }
-        saveGreenWaveRealTime(insertCrossDirDataRealtimePOS, decimalFormat);
+        saveGreenWaveRealTime(insertCrossDirDataRealtimePOS, decimalFormat, startDate);
         return insertCrossDirDataRealtimePOS;
     }
 
@@ -690,7 +690,7 @@ public class LanePeriodicDataProcessServiceImpl implements DataProcessService {
         }
     }
 
-    private void saveGreenWaveRealTime(List<CrossDirDataRealtimePO> crossDirDataRealtimePOS, DecimalFormat decimalFormat) throws Exception {
+    private void saveGreenWaveRealTime(List<CrossDirDataRealtimePO> crossDirDataRealtimePOS, DecimalFormat decimalFormat, Date date) throws Exception {
         try {
             Map<Integer, List<GreenwaveCrossPO>> greenWaveInfoMap = CrossGreenWaveInfoCache.greenWaveInfoMap;
             if (!greenWaveInfoMap.isEmpty()) {
@@ -699,7 +699,6 @@ public class LanePeriodicDataProcessServiceImpl implements DataProcessService {
                 for (Map.Entry<Integer, List<GreenwaveCrossPO>> entry : greenWaveInfoMap.entrySet()) {
                     List<GreenwaveCrossPO> greenWaveCrossIds = entry.getValue();
                     if (!CollectionUtils.isEmpty(greenWaveCrossIds)) {
-                        int status = 1;
                         double trafficIndex = 1.0;
                         Double speedSum = 0.0;
                         Double stopTimes = 0.0;
@@ -734,10 +733,7 @@ public class LanePeriodicDataProcessServiceImpl implements DataProcessService {
                             for (CrossDirDataRealtimePO dirDataRealtimePO : crossDirDataRealtimePOS) {
                                 String crossIdDir = dirDataRealtimePO.getCrossId();
                                 Integer dirType = dirDataRealtimePO.getDirType();
-                                startTime = crossDirDataRealtimePOS.get(0).getStartTime();
-
                                 if (StringUtils.equals(crossIdDir, crossId) && Objects.equals(inDir, dirType)) {
-                                    status = status > dirDataRealtimePO.getStatus() ? status : dirDataRealtimePO.getStatus();
                                     trafficIndex = trafficIndex > dirDataRealtimePO.getTrafficIndex() ? trafficIndex : dirDataRealtimePO.getTrafficIndex();
                                     speedSum += dirDataRealtimePO.getSpeed();
                                     stopTimes += dirDataRealtimePO.getStopTimes();
@@ -757,10 +753,10 @@ public class LanePeriodicDataProcessServiceImpl implements DataProcessService {
                             noparkPassRateAvg = noparkPassRate / (greenwaveCount);
                             congRate = congCount / greenwaveCount;
                         }
-                        setTraffixIndex(decimalFormat, speedAvg, freeFlowSpeedAvg, greenwaveRealtimePO);
+                        // 计算交通指数，并设置status
+                        setTraffixIndexAndStatus(decimalFormat, speedAvg, freeFlowSpeedAvg, greenwaveRealtimePO);
                         greenwaveRealtimePO.setGreenId(greenWaveId);
-                        greenwaveRealtimePO.setStartTime(startTime);
-                        greenwaveRealtimePO.setStatus(status);
+                        greenwaveRealtimePO.setStartTime(date);
                         greenwaveRealtimePO.setType(1);
                         greenwaveRealtimePO.setSpeed(speedAvg);
                         if (speedAvg != 0) {
@@ -797,7 +793,7 @@ public class LanePeriodicDataProcessServiceImpl implements DataProcessService {
         }
     }
 
-    private static void setTraffixIndex(DecimalFormat decimalFormat, double speedAvg, Double freeFlowSpeedAvg, GreenwaveRealtimePO greenwaveRealtimePO) {
+    private static void setTraffixIndexAndStatus(DecimalFormat decimalFormat, double speedAvg, Double freeFlowSpeedAvg, GreenwaveRealtimePO greenwaveRealtimePO) {
         Double greenwaveIndex = Double.valueOf(decimalFormat.format(freeFlowSpeedAvg / speedAvg));
         if (greenwaveIndex >= 10) {
             greenwaveRealtimePO.setTrafficIndex(10.0);
@@ -806,6 +802,18 @@ public class LanePeriodicDataProcessServiceImpl implements DataProcessService {
         } else {
             greenwaveRealtimePO.setTrafficIndex(1.0);
         }
+        // 根据指数设置status
+        int status = 1;
+        if (greenwaveIndex <= 1.8) {
+            status = 1;
+        } else if (greenwaveIndex > 1.8 && greenwaveIndex <= 2.5) {
+            status = 2;
+        } else if (greenwaveIndex > 2.5 && greenwaveIndex <= 3.5) {
+            status = 3;
+        } else if (greenwaveIndex > 3.5) {
+            status = 4;
+        }
+        greenwaveRealtimePO.setStatus(status);
     }
 
     private static Double getFreeFlowSpeed(Double freeFlowSpeed, String crossId, Map<String, Double> crossFreeFlowSpeedMap, Integer dirType) {
@@ -860,8 +868,8 @@ public class LanePeriodicDataProcessServiceImpl implements DataProcessService {
         // 替换干线旅行速度
         po.setSpeed(lineCongestion.getSpeed());
         // 替换状态
-        String congestionCode = lineCongestion.getCongestionCode();
-        if (EventAbnormalEnum.NO_CONGEST.getType().equals(congestionCode)) {
+        String congestionType = lineCongestion.getCongestionType();
+        if (EventAbnormalEnum.NO_CONGEST.getType().equals(congestionType)) {
             po.setStatus(1);
         } else {
             po.setStatus(3);

wj-datacenter-service/src/main/java/net/wanji/datacenter/task/DataCenterTask.java

@@ -62,6 +62,17 @@ public class DataCenterTask {
                 int greenDelayTime = 0; // 每个路口绿波开始时间
                 List<GreenwaveCrossPO> greenwaveCrossPOList = entry.getValue();
                 if (!CollectionUtils.isEmpty(greenwaveCrossPOList)) {
+                    greenwaveWidth = calculate(greenwaveWidth, lastCrossDistance, crossTimeOffset, greenDelayTime, greenwaveCrossPOList);
+                }
+                log.info("当前绿波编号:{}, 绿波带宽:{}", greenId, greenwaveWidth);
+            }
+        } catch (Exception e) {
+            log.error("获取绿波带宽异常", e);
+            throw new Exception("获取绿波带宽异常", e);
+        }
+    }
+
+    private int calculate(int greenwaveWidth, Double lastCrossDistance, int crossTimeOffset, int greenDelayTime, List<GreenwaveCrossPO> greenwaveCrossPOList) {
         for (GreenwaveCrossPO greenwaveCrossPO : greenwaveCrossPOList) {
             String crossId = greenwaveCrossPO.getCrossId();
             Integer sectionId = greenwaveCrossPO.getSectionId();
@@ -101,12 +112,6 @@ public class DataCenterTask {
             }
             lastCrossDistance = greenwaveCrossPO.getNextCrossLen();
         }
-                }
-                log.info("当前绿波编号:{}, 绿波带宽:{}", greenId, greenwaveWidth);
-            }
-        } catch (Exception e) {
-            log.error("获取绿波带宽异常", e);
-            throw new Exception("获取绿波带宽异常", e);
-        }
+        return greenwaveWidth;
     }
 }

wj-datacenter-service/src/main/java/net/wanji/datacenter/task/DataProcessTask.java

@@ -12,7 +12,6 @@ import net.wanji.datacenter.cache.CrossSchemePhaseLightsCache;
 import net.wanji.datacenter.kafka.ConsumerHandler;
 import net.wanji.datacenter.pojo.dto.CrossSchemePhaseDirTurnDTO;
 import net.wanji.datacenter.pojo.dto.CrossSnapshotDataDTO;
-import net.wanji.datacenter.pojo.dto.LanePeriodicDataDTO;
 import net.wanji.datacenter.service.impl.LanePeriodicDataProcessServiceImpl;
 import net.wanji.feign.service.UtcFeignClients;
 import org.apache.commons.lang.StringUtils;
@@ -52,8 +51,14 @@ public class DataProcessTask {
                 return;
             }
 
-            List<LightsStatusVO> lightsStatusVOS = JacksonUtils.getInstance().convertValue(jsonViewObject.getContent(), new TypeReference<List<LightsStatusVO>>() {
-            });
+            List<LightsStatusVO> lightsStatusVOS = JacksonUtils.getInstance().convertValue(jsonViewObject.getContent(), new TypeReference<List<LightsStatusVO>>() {});
+            calculateData(lightsStatusVOS);
+        }catch (Exception e) {
+           // log.error("",e);
+        }
+    }
+
+    private void calculateData(List<LightsStatusVO> lightsStatusVOS) throws Exception {
         for (LightsStatusVO lightsStatusVO : lightsStatusVOS) {
             String crossId = lightsStatusVO.getCrossId();
             String schemeNo = lightsStatusVO.getSchemeId();
@@ -62,21 +67,8 @@ public class DataProcessTask {
             ConcurrentHashMap<String, Integer> crossIdPhaseIdClearGreenMap = CrossRealTimeCache.crossPhaseCarNumsMap;
             getClearRateCache(crossId, currentPhaseId, crossIdPhaseIdClearGreenMap);
             Map<String, Object> dirLampGroupMap = lightsStatusVO.getDirLampGroupMap();
-                boolean isAllRed = true;
-                if (!dirLampGroupMap.isEmpty()) {
-                    for (Map.Entry<String, Object> entry : dirLampGroupMap.entrySet()) {
-                        Map<String, String> turnLightColorMap = (Map<String, String>) entry.getValue();
-                        for (Map.Entry<String, String> colorEntry : turnLightColorMap.entrySet()) {
-                            String lightColor = colorEntry.getValue();
-                            // 全场红，如果有一个绿灯就不计算 20行人跳过
-                            String key = colorEntry.getKey();
-                            if (Objects.equals(lightColor, "green") && !StringUtils.equals(key, "20")) {
-                                isAllRed = false;
-                            }
-                        }
-                    }
-                }
-
+            // 是否是全场红
+            boolean isAllRed = getAllRed(dirLampGroupMap);
             // 如果全场红
             if (isAllRed) {
                 List<CrossSchemePhaseDirTurnDTO> crossSchemePhaseDirTurnDTOList = CrossSchemePhaseLightsCache.crossSchemePhaseDirTurnDTOList;
@@ -102,9 +94,24 @@ public class DataProcessTask {
                 LanePeriodicDataProcessServiceImpl.periodHaveAllRedMap.put(crossId, 1);
             }
         }
-        }catch (Exception e) {
-           // log.error("",e);
     }
+
+    private static boolean getAllRed(Map<String, Object> dirLampGroupMap) {
+        boolean isAllRed = true;
+        if (!dirLampGroupMap.isEmpty()) {
+            for (Map.Entry<String, Object> entry : dirLampGroupMap.entrySet()) {
+                Map<String, String> turnLightColorMap = (Map<String, String>) entry.getValue();
+                for (Map.Entry<String, String> colorEntry : turnLightColorMap.entrySet()) {
+                    String lightColor = colorEntry.getValue();
+                    // 全场红，如果有一个绿灯就不计算 20行人跳过
+                    String key = colorEntry.getKey();
+                    if (Objects.equals(lightColor, "green") && !StringUtils.equals(key, "20")) {
+                        isAllRed = false;
+                    }
+                }
+            }
+        }
+        return isAllRed;
     }
 
     private void getClearRateCache(String crossId, String currentPhaseId, Map<String, Integer> crossIdPhaseIdClearGreenMap) throws JsonProcessingException {