[update] 数据中心转换保存逻辑

signal-datacenter-service/src/main/java/net/wanji/datacenter/pojo/convert/LanePeriodicDataEventListConvert.java

+package net.wanji.datacenter.pojo.convert;
+
+import net.wanji.common.framework.Constants;
+import net.wanji.datacenter.pojo.dto.LanePeriodicDataDTO;
+import net.wanji.datacenter.pojo.po.*;
+
+import java.util.Date;
+
+/**
+ * @author duanruiming
+ * @date 2023/03/11 14:04
+ */
+public class LanePeriodicDataEventListConvert {
+    
+    public CrossDataRealTimePO convert2CrossDataRealTimePO(LanePeriodicDataDTO.EventList eventList, CrossBaseLaneInfoPO laneInfoPO) {
+        String id = laneInfoPO.getId();
+        String crossId = laneInfoPO.getCrossId();
+        // todo 路口状态
+        int trafficFlow = eventList.getTrafficFlow();
+        double meanV = eventList.getMeanV();
+        double staticQueueLengthMax = eventList.getStaticQueueLengthMax();
+        int stopTimes = eventList.getMeanStopsNumber();
+        double meanDelay = eventList.getMeanDelay();
+        Integer meanDelayInt = Double.valueOf(meanDelay).intValue();
+        double laneSaturation = eventList.getLaneSaturation();
+        Date now = new Date();
+        int batchTime = (int)now.getTime();
+        String dir = String.valueOf(laneInfoPO.getDir());
+
+        CrossDataRealTimePO crossDataRealTimePO = new CrossDataRealTimePO();
+        crossDataRealTimePO.setCrossId(crossId);
+        crossDataRealTimePO.setStatus(0000); // 路口状态
+        crossDataRealTimePO.setType(1); // 1常规 2异常
+        crossDataRealTimePO.setStartTime(now);
+        crossDataRealTimePO.setDuration(5); // todo 通过第二批次判断
+
+        crossDataRealTimePO.setUnbalanceStatus(0);
+        crossDataRealTimePO.setUnbalanceIndex(1.0);
+        crossDataRealTimePO.setUnbalanceDirs(dir);
+        crossDataRealTimePO.setCongestionStatus(0);
+        crossDataRealTimePO.setCongestionIndex(1.0);
+        crossDataRealTimePO.setCongestionDirs(dir);
+        crossDataRealTimePO.setSpilloverStatus(0);
+        crossDataRealTimePO.setSpilloverIndex(1.0);
+        crossDataRealTimePO.setSpilloverDirs(dir);
+
+        crossDataRealTimePO.setTrafficIndex(1.0);// 交通指数
+        crossDataRealTimePO.setFlow(trafficFlow);
+        crossDataRealTimePO.setFlowRate(Double.valueOf(trafficFlow / 5 * 60));
+        crossDataRealTimePO.setSpeed(meanV);
+        crossDataRealTimePO.setQueueLength(staticQueueLengthMax);
+        crossDataRealTimePO.setStopTimes(stopTimes);
+        crossDataRealTimePO.setDelayTime(meanDelayInt);
+        crossDataRealTimePO.setSturation(laneSaturation);
+        crossDataRealTimePO.setBatchTime(batchTime);
+        crossDataRealTimePO.setGmtCreate(now);
+        crossDataRealTimePO.setGmtModified(now);
+        return crossDataRealTimePO;
+    }
+
+    public CrossDirDataRealTimePO convert2CrossDirDataRealTimePO(LanePeriodicDataDTO.EventList eventList, CrossBaseLaneInfoPO laneInfoPO) {
+        String crossId = laneInfoPO.getCrossId();
+        // todo 路口状态
+        int trafficFlow = eventList.getTrafficFlow();
+        double meanV = eventList.getMeanV();
+        double staticQueueLengthMax = eventList.getStaticQueueLengthMax();
+        int stopTimes = eventList.getMeanStopsNumber();
+        double meanDelay = eventList.getMeanDelay();
+        Integer meanDelayInt = Double.valueOf(meanDelay).intValue();
+        double laneSaturation = eventList.getLaneSaturation();
+        Date now = new Date();
+        int batchTime = (int)now.getTime();
+        Integer dir = laneInfoPO.getDir();
+        Integer type = laneInfoPO.getType();
+
+        CrossDirDataRealTimePO crossDirDataRealTimePO = new CrossDirDataRealTimePO();
+        // 路口方向ID（路口ID_方向_进出口_主辅路序号）
+        crossDirDataRealTimePO.setId(String.join(Constants.SystemParam.SEPARATOR_UNDER_LINE, String.valueOf(dir), String.valueOf(type)));
+        crossDirDataRealTimePO.setDirType(dir);
+        crossDirDataRealTimePO.setInOutType(type);
+        crossDirDataRealTimePO.setCrossId(crossId);
+        crossDirDataRealTimePO.setLength(staticQueueLengthMax);
+        crossDirDataRealTimePO.setStatus(0000);
+        crossDirDataRealTimePO.setTrafficIndex(0000.00);
+        crossDirDataRealTimePO.setStartTime(now);
+        crossDirDataRealTimePO.setDuration(0000);
+        crossDirDataRealTimePO.setFlow(trafficFlow);
+        crossDirDataRealTimePO.setSpeed(meanV);
+        crossDirDataRealTimePO.setQueueLength(staticQueueLengthMax);
+        crossDirDataRealTimePO.setStopTimes(stopTimes);
+        crossDirDataRealTimePO.setDelayTime(meanDelayInt);
+        crossDirDataRealTimePO.setSturation(laneSaturation);
+        crossDirDataRealTimePO.setBatchTime(batchTime);
+        crossDirDataRealTimePO.setGmtCreate(now);
+        crossDirDataRealTimePO.setGmtModified(now);//
+        crossDirDataRealTimePO.setEffusionTime(0000.0);//
+        return crossDirDataRealTimePO;
+    }
+
+    public CrossTurnDataRealtimePO convert2CrossTurnDataRealtimePO(LanePeriodicDataDTO.EventList eventList, CrossBaseLaneInfoPO laneInfoPO) {
+        String crossId = laneInfoPO.getCrossId();
+        // todo 路口状态
+        int trafficFlow = eventList.getTrafficFlow();
+        double meanV = eventList.getMeanV();
+        double staticQueueLengthMax = eventList.getStaticQueueLengthMax();
+        int stopTimes = eventList.getMeanStopsNumber();
+        double meanDelay = eventList.getMeanDelay();
+        Integer meanDelayInt = Double.valueOf(meanDelay).intValue();
+        double laneSaturation = eventList.getLaneSaturation();
+        Date now = new Date();
+        int batchTime = (int)now.getTime();
+        Integer dir = laneInfoPO.getDir();
+        Integer type = laneInfoPO.getType();
+        Integer turn = laneInfoPO.getTurn();
+
+        CrossTurnDataRealtimePO crossTurnDataRealtimePO = new CrossTurnDataRealtimePO();
+        crossTurnDataRealtimePO.setId(String.join(Constants.SystemParam.SEPARATOR_UNDER_LINE, crossId, null, String.valueOf(turn)));
+        crossTurnDataRealtimePO.setTurnType("");
+        crossTurnDataRealtimePO.setInDir(0000);
+        crossTurnDataRealtimePO.setOutDir(0000);
+        crossTurnDataRealtimePO.setCrossId(crossId);
+        crossTurnDataRealtimePO.setFlow(trafficFlow);
+        crossTurnDataRealtimePO.setStatus(00000);
+        crossTurnDataRealtimePO.setTrafficIndex(0000.0);
+        crossTurnDataRealtimePO.setSpeed(meanV);
+        crossTurnDataRealtimePO.setInSpeed(0000.0);
+        crossTurnDataRealtimePO.setOutSpeed(0000.0);
+        crossTurnDataRealtimePO.setQueueLength(staticQueueLengthMax);
+        crossTurnDataRealtimePO.setStopTimes(stopTimes);
+        crossTurnDataRealtimePO.setDelayTime(meanDelayInt);
+        crossTurnDataRealtimePO.setSturation(laneSaturation);
+        crossTurnDataRealtimePO.setVehheadDist(0000.0);
+        crossTurnDataRealtimePO.setVehheadTime(0000.0);
+        crossTurnDataRealtimePO.setQuality(0000.0);
+        crossTurnDataRealtimePO.setBatchTime(batchTime);
+        crossTurnDataRealtimePO.setGmtCreate(now);
+        crossTurnDataRealtimePO.setGmtModified(now);
+        return crossTurnDataRealtimePO;
+    }
+
+    public CrossLaneDataRealTimePO convert2CrossLaneDataRealTimePO(LanePeriodicDataDTO.EventList eventList, CrossBaseLaneInfoPO laneInfoPO) {
+        String crossId = laneInfoPO.getCrossId();
+        // todo 路口状态
+        int trafficFlow = eventList.getTrafficFlow();
+        double meanV = eventList.getMeanV();
+        double staticQueueLengthMax = eventList.getStaticQueueLengthMax();
+        int stopTimes = eventList.getMeanStopsNumber();
+        double meanDelay = eventList.getMeanDelay();
+        Integer meanDelayInt = Double.valueOf(meanDelay).intValue();
+        double laneSaturation = eventList.getLaneSaturation();
+        Date now = new Date();
+        int batchTime = (int)now.getTime();
+
+        CrossLaneDataRealTimePO crossLaneDataRealTimePO = new CrossLaneDataRealTimePO();
+        crossLaneDataRealTimePO.setId(laneInfoPO.getId());
+        crossLaneDataRealTimePO.setCrossId(crossId);
+        crossLaneDataRealTimePO.setFlow(trafficFlow);
+        crossLaneDataRealTimePO.setSpeed(meanV);
+        crossLaneDataRealTimePO.setInSpeed(0000.0);
+        crossLaneDataRealTimePO.setOutSpeed(0000.0);
+        crossLaneDataRealTimePO.setQueueLength(staticQueueLengthMax);
+        crossLaneDataRealTimePO.setStopTimes(stopTimes);
+        crossLaneDataRealTimePO.setDelayTime(meanDelayInt);
+        crossLaneDataRealTimePO.setCapacity(trafficFlow);
+        crossLaneDataRealTimePO.setSturation(laneSaturation);
+        crossLaneDataRealTimePO.setVehheadTime(0000.0);
+        crossLaneDataRealTimePO.setVehheadDist(0000.0);
+        crossLaneDataRealTimePO.setQuality(0000.0);
+        crossLaneDataRealTimePO.setBatchTime(batchTime);
+        crossLaneDataRealTimePO.setGmtCreate(now);
+        crossLaneDataRealTimePO.setGmtModified(now);
+
+        return crossLaneDataRealTimePO;
+    }
+}

signal-datacenter-service/src/main/java/net/wanji/datacenter/pojo/dto/LanePeriodicDataDTO.java

@@ -38,11 +38,11 @@ public class LanePeriodicDataDTO {
         /**
          * 最小动态排队长度
          */
-        private Integer dynamicQueueLengthMin;
+        private double dynamicQueueLengthMin;
         /**
          * 绿灯有效利用率
          */
-        private Integer greenLightEfficiency;
+        private double greenLightEfficiency;
         /**
          * 车道号(路网数据)
          */
@@ -50,19 +50,19 @@ public class LanePeriodicDataDTO {
         /**
          * 车道通行能力
          */
-        private Integer laneCapacity;
+        private int laneCapacity;
         /**
          * 车道流率
          */
-        private Integer laneFlowRate;
+        private double laneFlowRate;
         /**
          * 不停车率
          */
-        private Integer laneNoStopRate;
+        private double laneNoStopRate;
         /**
          * 一次停车率
          */
-        private Integer laneOneStopRate;
+        private double laneOneStopRate;
         /**
          * 车道饱和度
          */
@@ -70,15 +70,15 @@ public class LanePeriodicDataDTO {
         /**
          * 车道饱和流率
          */
-        private Integer laneSaturationFlowRate;
+        private int laneSaturationFlowRate;
         /**
          * 三次停车率
          */
-        private Integer laneThreeStopRate;
+        private double laneThreeStopRate;
         /**
          * 二次停车率
          */
-        private Integer laneTwoStopRate;
+        private double laneTwoStopRate;
         /**
          * 绿灯结束排队长度
          */
@@ -86,15 +86,15 @@ public class LanePeriodicDataDTO {
         /**
          * 绿灯起亮排队长度
          */
-        private Integer lightGreenStartQueueLength;
+        private double lightGreenStartQueueLength;
         /**
          * 平均延误
          */
-        private Integer meanDelay;
+        private double meanDelay;
         /**
          * 平均停车次数
          */
-        private Integer meanStopsNumber;
+        private int meanStopsNumber;
         /**
          * 平均点速度
          */
@@ -102,11 +102,11 @@ public class LanePeriodicDataDTO {
         /**
          * 最大静态排队长度
          */
-        private Integer staticQueueLengthMax;
+        private double staticQueueLengthMax;
         /**
          * 最小静态排队长度
          */
-        private Integer staticQueueLengthMin;
+        private double staticQueueLengthMin;
         /**
          * 时间占有率
          */
@@ -114,11 +114,11 @@ public class LanePeriodicDataDTO {
         /**
          * 自然流量
          */
-        private Integer trafficFlow;
+        private int trafficFlow;
         /**
          * 小车当量
          */
-        private Integer trolleyEquivalent;
+        private int trolleyEquivalent;
         /**
          * 平均空间密度（长度占比）
          */

signal-datacenter-service/src/main/java/net/wanji/datacenter/pojo/po/CrossDirDataRealTimePO.java

@@ -21,7 +21,7 @@ public class CrossDirDataRealTimePO {
     private Integer dirType;
     @ApiModelProperty(name = "进出口类型：1进口；2出口",notes = "")
     private Integer inOutType;
-    @ApiModelProperty(name = "进出口类型：1进口；2出口",notes = "")
+    @ApiModelProperty(name = "路口编号",notes = "")
     private String crossId;
     @ApiModelProperty(name = "路段长度",notes = "")
     private Double length;

signal-datacenter-service/src/main/java/net/wanji/datacenter/pojo/po/CrossLaneDataRealTimePO.java

@@ -20,31 +20,31 @@ public class CrossLaneDataRealTimePO {
     @ApiModelProperty(name = "路口编号",notes = "")
     private String crossId;
     @ApiModelProperty(name = "交通流量（辆）",notes = "")
-    private String flow;
+    private Integer flow;
     @ApiModelProperty(name = "平均速度（km/h）",notes = "")
-    private String speed;
+    private Double speed;
     @ApiModelProperty(name = "驶入速度（km/h）",notes = "")
-    private String in_speed;
+    private Double inSpeed;
     @ApiModelProperty(name = "驶出速度（km/h）",notes = "")
-    private String out_speed;
+    private Double outSpeed;
     @ApiModelProperty(name = "排队长度（米）",notes = "")
-    private String queue_length;
+    private Double queueLength;
     @ApiModelProperty(name = "停车次数",notes = "")
-    private String stop_times;
+    private Integer stopTimes;
     @ApiModelProperty(name = "延误时间",notes = "")
-    private String delay_time;
+    private Integer delayTime;
     @ApiModelProperty(name = "通行能力（辆/某一时段内最大过车数）,通行能力=自然流量",notes = "")
-    private String capacity;
+    private Integer capacity;
     @ApiModelProperty(name = "饱和度",notes = "")
-    private String sturation;
+    private Double sturation;
     @ApiModelProperty(name = "车头间距（米）",notes = "")
-    private String vehhead_dist;
+    private Double vehheadDist;
     @ApiModelProperty(name = "车头时距（秒）",notes = "")
-    private String vehhead_time;
+    private Double vehheadTime;
     @ApiModelProperty(name = "可信度（0-1）",notes = "")
-    private String quality;
+    private Double quality;
     @ApiModelProperty(name = "采集时间（10位时间戳）",notes = "")
-    private String batch_time;
+    private Integer batchTime;
     @ApiModelProperty(name = "创建时间",notes = "")
     private Date gmtCreate ;
     @ApiModelProperty(name = "修改时间",notes = "")

signal-datacenter-service/src/main/java/net/wanji/datacenter/pojo/po/CrossTurnDataRealtimePO.java

@@ -52,7 +52,7 @@ public class CrossTurnDataRealtimePO {
     private Double queueLength ;
     /** 停车次数（次） */
     @ApiModelProperty(name = "停车次数（次）",notes = "")
-    private Double stopTimes ;
+    private Integer stopTimes ;
     /** 延误时间（秒） */
     @ApiModelProperty(name = "延误时间（秒）",notes = "")
     private Integer delayTime ;

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

@@ -5,6 +5,7 @@ import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import net.wanji.datacenter.dao.mapper.*;
 import net.wanji.datacenter.kafka.ConsumerHandler;
+import net.wanji.datacenter.pojo.convert.LanePeriodicDataEventListConvert;
 import net.wanji.datacenter.pojo.dto.LanePeriodicDataDTO;
 import net.wanji.datacenter.pojo.po.*;
 import net.wanji.datacenter.service.dataconvert.DataProcessService;
@@ -13,7 +14,6 @@ import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
 
 import java.util.ArrayList;
-import java.util.Date;
 import java.util.List;
 import java.util.stream.Collectors;
 
@@ -33,6 +33,7 @@ public class LanePeriodicDataProcessServiceImpl implements DataProcessService {
     private final CrossTurnDataRealTimeMapper crossTurnDataRealTimeMapper;
     private final CrossLaneDataRealTimeMapper crossLaneDataRealTimeMapper;
     private final CrossDataRealTimeMapper crossDataRealTimeMapper;
+    private final LanePeriodicDataEventListConvert lanePeriodicDataEventListConvert;
 
     @Override
     public String receive(String topic) {
@@ -77,66 +78,22 @@ public class LanePeriodicDataProcessServiceImpl implements DataProcessService {
 
         for (LanePeriodicDataDTO.EventList laneRealTimeItem : eventLists) {
             String laneId = laneRealTimeItem.getLaneId();
-            Date now = new Date();
             for (CrossBaseLaneInfoPO laneInfoPO : crossBaseLaneInfoPOS) {
                 String id = laneInfoPO.getId();
-                String crossId = laneInfoPO.getCrossId();
-                // todo 路口状态
-                Double trafficFlow = Double.valueOf(laneRealTimeItem.getTrafficFlow());
-                String dir = String.valueOf(laneInfoPO.getDir());
-                String turn = String.valueOf(laneInfoPO.getTurn());
-
-
                 if (StringUtils.endsWithIgnoreCase(laneId, id)) {
-                    CrossDataRealTimePO crossDataRealTimePO = new CrossDataRealTimePO();
-                    crossDataRealTimePO.setCrossId(crossId);
-                    crossDataRealTimePO.setStatus(0000);
-                    crossDataRealTimePO.setType(1); // 1常规 2异常
-                    crossDataRealTimePO.setStartTime(now);
-                    crossDataRealTimePO.setDuration(5); // todo 通过第二批次判断
-
-                    crossDataRealTimePO.setUnbalanceStatus(0);
-                    crossDataRealTimePO.setUnbalanceIndex(1.0);
-                    crossDataRealTimePO.setUnbalanceDirs(dir);
-                    crossDataRealTimePO.setCongestionStatus(0);
-                    crossDataRealTimePO.setCongestionIndex(1.0);
-                    crossDataRealTimePO.setCongestionDirs(dir);
-                    crossDataRealTimePO.setSpilloverStatus(0);
-                    crossDataRealTimePO.setSpilloverIndex(1.0);
-                    crossDataRealTimePO.setSpilloverDirs(dir);
-
-
-                    crossDataRealTimePO.setTrafficIndex(trafficFlow);
-
-                    crossDataRealTimePOS.add(crossDataRealTimePO);
 
-                    CrossDirDataRealTimePO crossDirDataRealTimePO = new CrossDirDataRealTimePO();
-                    crossDirDataRealTimePO.setCrossId(crossId);
-                    crossDirDataRealTimePO.setStatus(0000);
-                    crossDirDataRealTimePO.setTrafficIndex(trafficFlow);
+                    crossDataRealTimePOS.add(lanePeriodicDataEventListConvert.convert2CrossDataRealTimePO(laneRealTimeItem, laneInfoPO));
 
+                    crossDirDataRealTimePOS.add(lanePeriodicDataEventListConvert.convert2CrossDirDataRealTimePO(laneRealTimeItem, laneInfoPO));
 
-                    crossDirDataRealTimePOS.add(crossDirDataRealTimePO);
+                    crossTurnDataRealtimePOS.add(lanePeriodicDataEventListConvert.convert2CrossTurnDataRealtimePO(laneRealTimeItem, laneInfoPO));
 
-                    CrossTurnDataRealtimePO crossTurnDataRealtimePO = new CrossTurnDataRealtimePO();
-                    crossTurnDataRealtimePO.setCrossId(crossId);
-                    crossTurnDataRealtimePO.setStatus(0000);
-
-                    crossTurnDataRealtimePOS.add(crossTurnDataRealtimePO);
-
-
-                    CrossLaneDataRealTimePO crossLaneDataRealTimePO = new CrossLaneDataRealTimePO();
-                    crossLaneDataRealTimePO.setCrossId(crossId);
-                    crossLaneDataRealTimePO.setId(id);
-
-                    crossLaneDataRealTimePOS.add(crossLaneDataRealTimePO);
+                    crossLaneDataRealTimePOS.add(lanePeriodicDataEventListConvert.convert2CrossLaneDataRealTimePO(laneRealTimeItem, laneInfoPO));
 
                 }
-
             }
         }
 
-
         crossDataRealTimeMapper.insert(crossDataRealTimePOS);
         crossDirDataRealTimeMapper.insert(crossDirDataRealTimePOS);
         crossTurnDataRealTimeMapper.insert(crossTurnDataRealtimePOS);