[update] 态势监测，修改ES查询

signal-optimize-service/src/main/java/net/wanji/opt/controller/TrendController.java

@@ -179,7 +179,7 @@ public class TrendController {
     @ApiResponses({
             @ApiResponse(code = 200, message = "OK", response = CountRealTimeVO.class),
     })
-    public JsonViewObject countRealTime(@RequestBody CrossIdBO crossIdBO) {
+    public JsonViewObject countRealTime(@RequestBody CrossIdBO crossIdBO) throws Exception {
         String crossId = crossIdBO.getCrossId();
         CountRealTimeVO countRealTimeVO = trendService.countRealTime(crossId);
         return JsonViewObject.newInstance().success(countRealTimeVO);
@@ -192,7 +192,7 @@ public class TrendController {
     @ApiResponses({
             @ApiResponse(code = 200, message = "OK", response = TableQueryVO.class),
     })
-    public JsonViewObject tableQuery(@RequestBody CrossIdAndTimeSpanBO crossIdAndTimeSpanBO) {
+    public JsonViewObject tableQuery(@RequestBody CrossIdAndTimeSpanBO crossIdAndTimeSpanBO) throws Exception {
         TableQueryVO tableQueryVO = trendService.tableQuery(crossIdAndTimeSpanBO);
         return JsonViewObject.newInstance().success(tableQueryVO);
     }
@@ -204,7 +204,7 @@ public class TrendController {
     @ApiResponses({
             @ApiResponse(code = 200, message = "OK", response = TableQueryVO.class),
     })
-    public JsonViewObject tableRealTime(@RequestBody CrossIdAndIsFirstBO crossIdAndIsFirstBO) {
+    public JsonViewObject tableRealTime(@RequestBody CrossIdAndIsFirstBO crossIdAndIsFirstBO) throws Exception {
         TableQueryVO tableRealTimeVO = trendService.tableRealTime(crossIdAndIsFirstBO);
         return JsonViewObject.newInstance().success(tableRealTimeVO);
     }

signal-optimize-service/src/main/java/net/wanji/opt/dto/CrossLaneSnapshotDataDTO.java

+package net.wanji.opt.dto;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.elasticsearch.annotations.Document;
+import org.springframework.data.elasticsearch.annotations.Field;
+import org.springframework.data.elasticsearch.annotations.FieldType;
+
+/**
+ * @author duanruiming
+ * @date 2023/10/25 17:33
+ */
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+@Document(indexName = "datacenter_lane_snapshot_data", createIndex = false)
+public class CrossLaneSnapshotDataDTO {
+
+    @Id
+    private String id;
+    @Field(type = FieldType.Keyword, name = "crossId")
+    private String crossId;
+    private Integer dir;
+    private Integer turn;
+
+    @Field(type = FieldType.Keyword, name = "timeStamp")
+    private String timeStamp;
+
+    /**
+     * 车道号(路网数据)
+     */
+    private String laneId;
+    /**
+     * 静态排队长度(米)
+     */
+    private double staticQueueLength;
+    /**
+     * 队首距离(米)
+     */
+    private double teamHeadDistance;
+    /**
+     * 队尾距离(米)
+     */
+    private double teamTailDistance;
+    /**
+     * 动态排队长度(米)
+     */
+    private double dynamicQueueLength;
+    /**
+     * 平均车头间距(米)
+     */
+    private double meanSpaceHeadway;
+    /**
+     * 车头间距方差(-)
+     */
+    private double stdSpaceHeadway;
+    /**
+     * 通道排队数量(辆)
+     */
+    private int queueNums;
+    /**
+     * 车道车辆数(辆)
+     */
+    private int carNums;
+    /**
+     * 空间密度（车辆负荷比）
+     */
+    private double vehicleNumsRatio;
+    /**
+     * 空间密度（长度占比）
+     */
+    private double vehicleLengthRatio;
+    /**
+     * 平均速度(km/h)
+     */
+    private double meanV;
+    /**
+     * 头车位置(米)
+     */
+    private double headCarPosition;
+    /**
+     * 头车速度(km/h)
+     */
+    private double startV;
+    /**
+     * 末车位置(米)
+     */
+    private double tailCarPosition;
+    /**
+     * 末车速度(km/h)
+     */
+    private double endV;
+}

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

@@ -38,11 +38,11 @@ public interface TrendService {
 
     GreenwaveDetailVO currentGreenwaveDetail(GreenwaveIdBO bo) throws ParseException;
 
-    CountRealTimeVO countRealTime(String crossId);
+    CountRealTimeVO countRealTime(String crossId) throws Exception;
 
-    TableQueryVO tableQuery(CrossIdAndTimeSpanBO crossIdAndTimeSpanBO);
+    TableQueryVO tableQuery(CrossIdAndTimeSpanBO crossIdAndTimeSpanBO) throws Exception;
 
-    TableQueryVO tableRealTime(CrossIdAndIsFirstBO crossIdAndIsFirstBO);
+    TableQueryVO tableRealTime(CrossIdAndIsFirstBO crossIdAndIsFirstBO) throws Exception;
 
     OptTypeVO optType(CrossIdBO crossIdBO);
 }

signal-optimize-service/src/main/java/net/wanji/opt/service/es/LaneSnapshotDataQueryService.java

+package net.wanji.opt.service.es;
+
+import lombok.extern.slf4j.Slf4j;
+import net.wanji.common.constant.Constants;
+import net.wanji.common.utils.tool.JacksonUtils;
+import net.wanji.opt.dto.CrossLaneSnapshotDataDTO;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.RequestOptions;
+import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.index.query.BoolQueryBuilder;
+import org.elasticsearch.index.query.MatchQueryBuilder;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.index.query.RangeQueryBuilder;
+import org.elasticsearch.search.SearchHit;
+import org.elasticsearch.search.SearchHits;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * @author duanruiming
+ * @date 2023/10/26 14:13
+ */
+@Service
+@Slf4j
+public class LaneSnapshotDataQueryService implements LaneSnapshotService {
+    @Resource
+    RestHighLevelClient client;
+
+    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+
+    @Override
+    public List<CrossLaneSnapshotDataDTO> queryByCrossIdAndTimeSpan(
+            String crossId, int startTimeStamp, int endTimeStamp)
+            throws Exception {
+
+        List<CrossLaneSnapshotDataDTO> result = new ArrayList<>();
+
+        try {
+            SearchRequest searchRequest = new SearchRequest(Constants.LANE_SNAPSHOT_DATA_ES_INDEX);
+            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
+            
+            MatchQueryBuilder matchQuery = QueryBuilders.matchQuery("crossId", crossId);
+            RangeQueryBuilder rangeQuery = QueryBuilders.rangeQuery("timeStamp")
+                    .gte(startTimeStamp)
+                    .lte(endTimeStamp);
+
+            BoolQueryBuilder boolQuery = QueryBuilders.boolQuery()
+                    .must(matchQuery)
+                    .must(rangeQuery);
+
+            searchSourceBuilder.query(boolQuery);
+            searchRequest.source(searchSourceBuilder);
+
+            SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT);
+            SearchHits hits = response.getHits();
+
+            for (SearchHit hit : hits) {
+                String sourceAsString = hit.getSourceAsString();
+                CrossLaneSnapshotDataDTO crossLaneSnapshotDataDTO =
+                        JacksonUtils.getInstance().readValue(sourceAsString, CrossLaneSnapshotDataDTO.class);
+                result.add(crossLaneSnapshotDataDTO);
+            }
+        } catch (Exception e) {
+            log.error("ES数据获取错误", e);
+            throw new Exception(e);
+        }
+        return result;
+    }
+}

signal-optimize-service/src/main/java/net/wanji/opt/service/es/LaneSnapshotService.java

+package net.wanji.opt.service.es;
+
+
+import net.wanji.opt.dto.CrossLaneSnapshotDataDTO;
+
+import java.util.List;
+
+/**
+ * @author duanruiming
+ * @date 2023/10/26 14:12
+ */
+public interface LaneSnapshotService {
+    List<CrossLaneSnapshotDataDTO> queryByCrossIdAndTimeSpan(String crossId, int startTimeStamp, int endTimeStamp) throws Exception;
+}

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

@@ -19,6 +19,7 @@ import net.wanji.opt.dao.mapper.strategy.SceneStrategyMapper;
 import net.wanji.opt.dao.mapper.strategy.StrategyMapper;
 import net.wanji.opt.dao.mapper.trend.EventAlarmMapper;
 import net.wanji.opt.dao.mapper.trend.GreenwaveInfoMapper;
+import net.wanji.opt.dto.CrossLaneSnapshotDataDTO;
 import net.wanji.opt.dto.LineSchemeDTO;
 import net.wanji.opt.dto.trend.AbnormalCrossListDTO;
 import net.wanji.opt.dto.trend.GreenwaveListDTO;
@@ -27,7 +28,9 @@ import net.wanji.opt.po.strategy.ScenePO;
 import net.wanji.opt.po.strategy.StrategyPO;
 import net.wanji.opt.po.trend.EventAlarmPO;
 import net.wanji.opt.service.TrendService;
+import net.wanji.opt.service.es.LaneSnapshotDataQueryService;
 import net.wanji.opt.vo.*;
+import org.jetbrains.annotations.NotNull;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
@@ -81,6 +84,7 @@ public class TrendServiceImpl implements TrendService {
     private final BaseCrossPlanMapper baseCrossPlanMapper;
     private final MainlineEvaluateServiceImpl mainlineEvaluateServiceImpl;
     private final CrossLaneDataHistMapper crossLaneDataHistMapper;
+    private final LaneSnapshotDataQueryService laneSnapshotDataQueryService;
 
     // 用于计算路口状态,key为方向，value为状态
     private Map<Integer, Integer> preStatus = new HashMap<Integer, Integer>() {{
@@ -106,7 +110,7 @@ public class TrendServiceImpl implements TrendService {
                             CrossTurnDataRealtimeMapper crossTurnDataRealtimeMapper,
                             CrossDirDataHistMapper crossDirDataHistMapper, EventAlarmMapper eventAlarmMapper,
                             RidInfoMapper ridInfoMapper, BaseCrossDirInfoMapper baseCrossDirInfoMapper,
-                            CrossSchemeOptLogMapper crossSchemeOptLogMapper, GreenwaveCrossMapper greenwaveCrossMapper, CrossDirDataRealtimeMapper crossDirDataRealtimeMapper, GreenwaveHistMapper greenwaveHistMapper, GreenwaveRealtimeMapper greenwaveRealtimeMapper, SceneStrategyIdeaMapper strategyIdeaMapper, StrategyMapper strategyMapper, BaseCrossSchedulesMapper baseCrossSchedulesMapper, BaseCrossSchedulesPlanMapper baseCrossSchedulesPlanMapper, BaseCrossSectionMapper baseCrossSectionMapper, BaseCrossSchemeMapper crossSchemeMapper, BaseCrossSchemeMapper baseCrossSchemeMapper, BaseCrossPhaseMapper baseCrossPhaseMapper, BaseCrossPhaseLightsMapper crossPhaseLightsMapper, BaseCrossLaneLightsMapper baseCrossLaneLightsMapper, CrossBaseLaneInfoMapper crossBaseLaneInfoMapper, CrossTurnDataHistMapper crossTurnDataHistMapper, @Qualifier("greenwaveSceneMapper") GreenwaveSceneMapper greenwaveSceneMapper, @Qualifier("sceneStrategyMapper") SceneStrategyMapper sceneStrategyMapper, @Qualifier("sceneMapper") SceneMapper sceneMapper, @Qualifier("baseCrossPlanMapper") BaseCrossPlanMapper baseCrossPlanMapper, MainlineEvaluateServiceImpl mainlineEvaluateServiceImpl, CrossLaneDataHistMapper crossLaneDataHistMapper) {
+                            CrossSchemeOptLogMapper crossSchemeOptLogMapper, GreenwaveCrossMapper greenwaveCrossMapper, CrossDirDataRealtimeMapper crossDirDataRealtimeMapper, GreenwaveHistMapper greenwaveHistMapper, GreenwaveRealtimeMapper greenwaveRealtimeMapper, SceneStrategyIdeaMapper strategyIdeaMapper, StrategyMapper strategyMapper, BaseCrossSchedulesMapper baseCrossSchedulesMapper, BaseCrossSchedulesPlanMapper baseCrossSchedulesPlanMapper, BaseCrossSectionMapper baseCrossSectionMapper, BaseCrossSchemeMapper crossSchemeMapper, BaseCrossSchemeMapper baseCrossSchemeMapper, BaseCrossPhaseMapper baseCrossPhaseMapper, BaseCrossPhaseLightsMapper crossPhaseLightsMapper, BaseCrossLaneLightsMapper baseCrossLaneLightsMapper, CrossBaseLaneInfoMapper crossBaseLaneInfoMapper, CrossTurnDataHistMapper crossTurnDataHistMapper, @Qualifier("greenwaveSceneMapper") GreenwaveSceneMapper greenwaveSceneMapper, @Qualifier("sceneStrategyMapper") SceneStrategyMapper sceneStrategyMapper, @Qualifier("sceneMapper") SceneMapper sceneMapper, @Qualifier("baseCrossPlanMapper") BaseCrossPlanMapper baseCrossPlanMapper, MainlineEvaluateServiceImpl mainlineEvaluateServiceImpl, CrossLaneDataHistMapper crossLaneDataHistMapper, LaneSnapshotDataQueryService laneSnapshotDataQueryService) {
         this.greenwaveInfoMapper = greenwaveInfoMapper;
         this.baseCrossInfoMapper = baseCrossInfoMapper;
         this.crossDataRealtimeMapper = crossDataRealtimeMapper;
@@ -136,6 +140,7 @@ public class TrendServiceImpl implements TrendService {
         this.baseCrossPlanMapper = baseCrossPlanMapper;
         this.mainlineEvaluateServiceImpl = mainlineEvaluateServiceImpl;
         this.crossLaneDataHistMapper = crossLaneDataHistMapper;
+        this.laneSnapshotDataQueryService = laneSnapshotDataQueryService;
     }
 
     @Override
@@ -1047,17 +1052,21 @@ public class TrendServiceImpl implements TrendService {
     }
 
     @Override
-    public CountRealTimeVO countRealTime(String crossId) {
+    public CountRealTimeVO countRealTime(String crossId) throws Exception {
         CountRealTimeVO countRealTimeVO = new CountRealTimeVO();
-        // todo 统计实时条数
-//        countRealTimeVO.setRealTimeCount();
 
-        // 统计周期条数
         ZoneId zoneId = ZoneId.systemDefault();
         ZonedDateTime startOfDay = LocalDateTime.now().toLocalDate().atStartOfDay(zoneId);
         ZonedDateTime now = ZonedDateTime.now(zoneId);
         int startTimeStamp = (int) startOfDay.toEpochSecond();
         int endTimeStamp = (int) now.toEpochSecond();
+
+        // 统计实时条数
+        List<CrossLaneSnapshotDataDTO> dtoList =
+                laneSnapshotDataQueryService.queryByCrossIdAndTimeSpan(crossId, startTimeStamp, endTimeStamp);
+        countRealTimeVO.setRealTimeCount(dtoList.size());
+
+        // 统计周期条数
         Integer cycleCount = crossLaneDataHistMapper.
                 selectCountByCrossIdAndTimeSpan(crossId, startTimeStamp, endTimeStamp);
         countRealTimeVO.setCycleCount(cycleCount);
@@ -1065,28 +1074,50 @@ public class TrendServiceImpl implements TrendService {
     }
 
     @Override
-    public TableQueryVO tableQuery(CrossIdAndTimeSpanBO crossIdAndTimeSpanBO) {
+    public TableQueryVO tableQuery(CrossIdAndTimeSpanBO crossIdAndTimeSpanBO) throws Exception {
         String crossId = crossIdAndTimeSpanBO.getCrossId();
         Date startTime = crossIdAndTimeSpanBO.getStartTime();
         Date endTime = crossIdAndTimeSpanBO.getEndTime();
-
-        TableQueryVO tableQueryVO = new TableQueryVO();
-        // todo 实时数据
-//        tableQueryVO.setRealTimeData();
-
         // 将Date对象转换为10位时间戳
         int startTimeStamp = (int) (startTime.getTime() / 1000);
         int endTimeStamp = (int) (endTime.getTime() / 1000);
-        List<CrossLaneDataHistPOExt> poExtList = crossLaneDataHistMapper
-                .selectByCrossIdAndTimeSpan(crossId, startTimeStamp, endTimeStamp);
 
-        tableQueryVO.setCycleData(buildCycleData(poExtList));
+        return buildTableQueryVO(crossId, startTimeStamp, endTimeStamp);
+    }
 
-        return tableQueryVO;
+    private List<TableQueryVO.RealTimeDataElement> buildRealTimeList(List<CrossLaneSnapshotDataDTO> dtoList) {
+        List<TableQueryVO.RealTimeDataElement> res = new ArrayList<>();
+
+        for (CrossLaneSnapshotDataDTO dto : dtoList) {
+            TableQueryVO.RealTimeDataElement element = new TableQueryVO.RealTimeDataElement();
+
+            element.setTime(new Date()); // todo
+
+            element.setDir(dto.getDir());
+
+            String laneId = dto.getLaneId();
+            String laneSort = laneId.substring(laneId.length() - 2);
+            element.setLaneSort(Integer.valueOf(laneSort));
+
+            double staticQueueLength = dto.getStaticQueueLength();
+            int staticQueueLengthRound = (int) Math.round(staticQueueLength);
+            element.setQueueLength(staticQueueLengthRound);
+
+            double headCarPosition = dto.getHeadCarPosition();
+            int headCarPositionRound = (int) Math.round(headCarPosition);
+            element.setFirstCarPosition(headCarPositionRound);
+
+            double tailCarPosition = dto.getTailCarPosition();
+            int tailCarPositionRound = (int) Math.round(tailCarPosition);
+            element.setEndCarPosition(tailCarPositionRound);
+            res.add(element);
+        }
+
+        return res;
     }
 
     @Override
-    public TableQueryVO tableRealTime(CrossIdAndIsFirstBO crossIdAndIsFirstBO) {
+    public TableQueryVO tableRealTime(CrossIdAndIsFirstBO crossIdAndIsFirstBO) throws Exception {
         String crossId = crossIdAndIsFirstBO.getCrossId();
         Integer isFirstInvoke = crossIdAndIsFirstBO.getIsFirstInvoke();
         // 如果是初次调用，需初始化开始时间
@@ -1095,14 +1126,22 @@ public class TrendServiceImpl implements TrendService {
         }
         // 以当前时间为结束时间
         Date endDate = new Date();
+        int startTimeStamp = (int) (tableCycleFirstDate.getTime() / 1000);
+        int endTimeStamp = (int) (endDate.getTime() / 1000);
 
+        return buildTableQueryVO(crossId, startTimeStamp, endTimeStamp);
+    }
+
+    @NotNull
+    private TableQueryVO buildTableQueryVO(String crossId, int startTimeStamp, int endTimeStamp) throws Exception {
         TableQueryVO tableQueryVO = new TableQueryVO();
 
-        // todo 查询秒级数据
+        // 查询秒级数据
+        List<CrossLaneSnapshotDataDTO> dtoList =
+                laneSnapshotDataQueryService.queryByCrossIdAndTimeSpan(crossId, startTimeStamp, endTimeStamp);
+        tableQueryVO.setRealTimeData(buildRealTimeList(dtoList));
 
         // 查询周期数据
-        int startTimeStamp = (int) (tableCycleFirstDate.getTime() / 1000);
-        int endTimeStamp = (int) (endDate.getTime() / 1000);
         List<CrossLaneDataHistPOExt> poExtList = crossLaneDataHistMapper
                 .selectByCrossIdAndTimeSpan(crossId, startTimeStamp, endTimeStamp);
         tableQueryVO.setCycleData(buildCycleData(poExtList));

signal-optimize-service/src/main/resources/application-local.yaml

 spring:
+  elasticsearch:
+    username: elastic
+    password: Wanji300552
+    uris: http://37.12.182.31:9200
   application:
     name: opt
   datasource:

wj-common/pom.xml

@@ -63,10 +63,26 @@
             <version>2.0</version>
         </dependency>
         <!--es start-->
+        <!-- ElasticSearch依赖 -->
+        <dependency>
+            <groupId>org.elasticsearch.client</groupId>
+            <artifactId>elasticsearch-rest-high-level-client</artifactId>
+            <version>7.3.0</version>
+        </dependency>
+        <dependency>
+            <groupId>org.elasticsearch</groupId>
+            <artifactId>elasticsearch</artifactId>
+            <version>7.3.0</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
+            <version>2.6.3</version>
+        </dependency>
         <dependency>
             <groupId>org.elasticsearch.client</groupId>
             <artifactId>transport</artifactId>
-            <version>6.8.5</version>
+            <version>7.3.0</version>
             <exclusions>
                 <exclusion>
                     <artifactId>elasticsearch</artifactId>
@@ -74,11 +90,6 @@
                 </exclusion>
             </exclusions>
         </dependency>
-        <dependency>
-            <groupId>org.elasticsearch</groupId>
-            <artifactId>elasticsearch</artifactId>
-            <version>6.8.5</version>
-        </dependency>
         <!--es end-->
         <!-- Excel -->
         <dependency>

wj-common/src/main/java/net/wanji/common/constant/Constants.java

+package net.wanji.common.constant;
+
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * @author duanruiming
+ * @date 2023/09/22 9:50
+ */
+public class Constants {
+    public static final String underline = "_";
+
+    // 横向方向列表
+    public static final List<Integer> X_DIR_LIST = Arrays.asList(2, 3, 6, 7);
+
+    // 路口列表，后续可以缓存信控路口列表
+    public static final List<String> IS_SIGNAL_CROSS_LIST = Arrays.asList("13NF80B5QN0", "13NGH0B5RC0", "13NI00B5RM0");
+
+    public static final String LANE_SNAPSHOT_DATA_ES_INDEX = "datacenter_lane_snapshot_data";
+}
\ No newline at end of file

wj-common/src/main/java/net/wanji/common/tool/estool/EsUtil.java

@@ -498,7 +498,7 @@ public class EsUtil {
                     .setExplain(true)
                     .execute()
                     .actionGet();
-            count = response.getHits().getTotalHits();
+            count = response.getHits().getTotalHits().value;
         } catch (Exception e) {
             log.error("[indexName:" + indexName + "],[typeName:" + typeName + "],[name:" + typeName + "], EsUtil  searchDatasCount is error! ", e);
         }