Skip to content

W
wj-datacenter-platform
Commit 5c7c87f4 authored by zhoushiguang's avatar zhoushiguang
Browse files
Options

路段路况计算优化

parent d6607311
Hide whitespace changes
Inline Side-by-side
Showing with 111 additions and 94 deletions
wj-realtime-computing/src/main/java/com/wanji/indicators/cache/BaseLaneCache.java
View file @ 5c7c87f4
......@@ -48,7 +48,7 @@ public class BaseLaneCache {
private static Map<String, Integer> segmentLaneNumberMap = new ConcurrentHashMap<>();
private static Map<String, Integer> ridMaxLaneNumberMap = new ConcurrentHashMap<>();
private static Map<String, Double[]> ridLaneLengthMap = new ConcurrentHashMap<>();
private static Map<String, String> refLaneIdMap = new ConcurrentHashMap<>();
......@@ -76,6 +76,7 @@ public class BaseLaneCache {
ApplicationContext beanConf = new ClassPathXmlApplicationContext("spring-container.xml");
BaseLaneInfoService baseLaneInfoService = beanConf.getBean(BaseLaneInfoServiceImpl.class);
String crossList = property.getProperties().getProperty("full.area.cross.list");
Map<String, BaseLaneInfo> map = baseLaneInfoService.findLaneInfo(crossList);
baseLaneInfoMap.putAll(map);
......@@ -91,9 +92,13 @@ public class BaseLaneCache {
segmentLaneNumberMap.put(entry.getKey(), entry.getValue().size());
}
Map<String, List<BaseLaneInfo>> ridLaneList = collection.stream().filter(o -> Objects.equals(2,o.getType())).collect(Collectors.groupingBy(o -> o.getRid()));
Map<String, List<BaseLaneInfo>> ridLaneList = collection.stream().collect(Collectors.groupingBy(o -> o.getRid()+"_"+o.getType()));
for (Map.Entry<String, List<BaseLaneInfo>> entry : ridLaneList.entrySet()) {
ridMaxLaneNumberMap.put(entry.getKey(), entry.getValue().size());
List<BaseLaneInfo> value = entry.getValue();
double totalLength = value.stream().mapToDouble(o->o.getLength()).summaryStatistics().getSum();
//路段上车道总长度
ridLaneLengthMap.put(entry.getKey(), new Double[]{totalLength,(double)value.size()});
}
}
......@@ -138,7 +143,7 @@ public class BaseLaneCache {
return ridSameDirOutLaneMap;
}
public static Map<String, Integer> getRidMaxLaneNumberMap() {
return ridMaxLaneNumberMap;
public static Map<String, Double[]> getRidLaneLengthMap() {
return ridLaneLengthMap;
}
}
wj-realtime-computing/src/main/java/com/wanji/indicators/task/AreaIndicatorMain.java
View file @ 5c7c87f4
......@@ -123,8 +123,7 @@ public class AreaIndicatorMain {
try {
//区域指数相关
AreaIndexAnalysisMainNew.init(env, "indexName",false).run(joinLaneStream);
//行人非机动车数量对机动车速度影响
//RidParticipantAnalysisMainNew.init(env,null,false).run(joinLaneStream);
//非机动车参与者类别数量统计
LaneNonMotorAnalysisMainNew.init(env,null,false).run(joinLaneStream);
......
wj-realtime-computing/src/main/java/com/wanji/indicators/task/OfflineTestIndicatorTaskMain.java
View file @ 5c7c87f4
......@@ -54,8 +54,8 @@ public class OfflineTestIndicatorTaskMain {
ParameterTool parameter = ParameterTool.fromArgs(args);
inPath = parameter.get("inPath","D:\\WanJi-Work\\07 项目\\02 济南\\数据");
outPath = parameter.get("outPath","d:/flink/out/");
startTime = parameter.get("startTime","2024-05-20_19:25:00");
endTime = parameter.get("endTime","2024-05-20_19:35:00");
startTime = parameter.get("startTime","2024-05-30_19:00:00");
endTime = parameter.get("endTime","2024-05-30_19:35:00");
// inPath = "d:/flink/workspace";
//读取离线数据
......
wj-realtime-computing/src/main/java/com/wanji/indicators/task/laneIndicators/func/LaneDataProcessWindow.java
View file @ 5c7c87f4
......@@ -23,10 +23,13 @@ import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;
......@@ -68,6 +71,7 @@ public class LaneDataProcessWindow extends ProcessWindowFunction<CrossFrameModel
long windowEndTime = context.window().getEnd();
String startTime = DateUtil.toDateTime(windowStartTime, DateUtil.YYYY_MM_DD_HH_MM_SS);
String endTime = DateUtil.toDateTime(windowEndTime, DateUtil.YYYY_MM_DD_HH_MM_SS);
long globalTime = windowEndTime;
Map<String, LaneResultModel> turnResultModelMap = new HashMap<>();
......@@ -89,7 +93,7 @@ public class LaneDataProcessWindow extends ProcessWindowFunction<CrossFrameModel
String laneStartPoint = carTrackModel.getRoadnet().getLaneStartPoint();
String carPoint = carTrackModel.getLongitude() + "," + carTrackModel.getLatitude();
double distance = GeomsConvertUtil.getDistance(laneStartPoint, carPoint);
CarTrackModel.RoadNet roadNet = BaseGisLaneCache.getInstance().queryRoadNet(carPoint);
String key = entry.getKey()+"_"+entry1.getKey();
if (distance >= (laneLength+1) && !distinctIdState.contains(key)) {
......@@ -133,5 +137,21 @@ public class LaneDataProcessWindow extends ProcessWindowFunction<CrossFrameModel
out.collect(crossTurnResultModel);
}
//状态过期清理
Set<String> expireKeys = new HashSet<>();
Iterator<Map.Entry<String,Long>> iterator = distinctIdState.iterator();
while (iterator.hasNext()) {
Map.Entry<String,Long> entry = iterator.next();
long timestamp = entry.getValue();
if (globalTime - timestamp > 5 * 60 * 1000) {
//缓存超过
expireKeys.add(entry.getKey());
}
}
if (!expireKeys.isEmpty()) {
for (String key : expireKeys) {
distinctIdState.remove(key);
}
}
}
}
wj-realtime-computing/src/main/java/com/wanji/indicators/task/trafficstate/func/RidProcessWindow.java
View file @ 5c7c87f4
......@@ -74,39 +74,49 @@ public class RidProcessWindow extends ProcessWindowFunction<CarTrackModel, LaneT
List<CarTrackModel> list = StreamSupport.stream(elements.spliterator(), false).collect(Collectors.toList());
//进口车道
List<CarTrackModel> inLaneList = list.stream().filter(o -> Objects.equals(2, o.getRoadnet().getLaneType())).collect(Collectors.toList());
// List<CarTrackModel> inLaneList = list.stream().filter(o -> Objects.equals(2, o.getRoadnet().getLaneType())).collect(Collectors.toList());
//出口道
// List<CarTrackModel> outLaneList = list.stream().filter(o -> Objects.equals(3, o.getRoadnet().getLaneType())).collect(Collectors.toList());
//保留路段上的车辆
list = list.stream().filter(o -> !Objects.equals(2, o.getRoadnet().getLaneType()) && !Objects.equals(3, o.getRoadnet().getLaneType())).collect(Collectors.toList());
list = list.stream().filter(o -> Objects.equals(1, o.getRoadnet().getLaneType())).collect(Collectors.toList());
Double freeSpeed = null;
String rid = null;
Integer ridDir = null;
double trafficIndex = 1;
if (!list.isEmpty()) {
if (!inLaneList.isEmpty()) {
freeSpeed = inLaneList.get(0).getRoadnet().getFreeSpeed();
rid = inLaneList.get(0).getRoadnet().getRid();
ridDir = inLaneList.get(0).getRoadnet().getRidDir8();
} else {
freeSpeed = list.get(0).getRoadnet().getFreeSpeed();
rid = list.get(0).getRoadnet().getRid();
ridDir = list.get(0).getRoadnet().getRidDir8();
}
freeSpeed = list.get(0).getRoadnet().getFreeSpeed();
rid = list.get(0).getRoadnet().getRid();
ridDir = list.get(0).getRoadnet().getRidDir8();
if (freeSpeed == null) {
freeSpeed = Double.parseDouble(this.properties.getProperty("rid.default.free.speed"));
}
double avgSpeed = getAvgSpeed(list, freeSpeed);
double laneCount = 0;
double totalLength = 0;
double carSizeThreshold = 0;
Double[] arr = BaseLaneCache.getInstance().getRidLaneLengthMap().get(key+"_1");
if (arr != null) {
laneCount = arr[1];
totalLength = arr[0];
carSizeThreshold = ArithOfBigDecmial.div(totalLength, 5, 0);
if (laneCount>1) {
carSizeThreshold = ArithOfBigDecmial.div(carSizeThreshold, laneCount);
} else {
carSizeThreshold = ArithOfBigDecmial.div(carSizeThreshold, 2);
}
}
long segmentCarSize = list.stream().mapToInt(o -> o.getId()).distinct().count();
if (avgSpeed > 0 && segmentCarSize > 10) {
//过滤轨迹覆盖度较低情况
if (avgSpeed > 0 && segmentCarSize > carSizeThreshold) {
trafficIndex = ArithOfBigDecmial.div(freeSpeed, avgSpeed);
}
String trafficState = "1";
Integer count = BaseLaneCache.getInstance().getRidMaxLaneNumberMap().get(key);
int ridLevel = RoadClassMap.MAIN_DAO.getType();
if (count != null && count >= 3) {
if (laneCount >= 3) {
ridLevel = RoadClassMap.MAIN_DAO.getType();
} else {
ridLevel = RoadClassMap.CIYAO_DAO.getType();
......@@ -118,6 +128,9 @@ public class RidProcessWindow extends ProcessWindowFunction<CarTrackModel, LaneT
trafficState = CongestEnum.NO_CONGEST.getType();
} else if (trafficIndex > 1.8 && trafficIndex <= 2.5) {
trafficState = CongestEnum.LIGHT_CONGEST.getType();
// if (key.equals("13NID0B5RM013NI00B5RM00")) {
// System.out.println();
// }
} else if (trafficIndex > 2.5 && trafficIndex <= 3.5) {
trafficState = CongestEnum.MODERATE_CONGEST.getType();
} else if (trafficIndex > 3.5) {
......@@ -161,14 +174,13 @@ public class RidProcessWindow extends ProcessWindowFunction<CarTrackModel, LaneT
out.collect(outModel);
outModel.setCreateTime(System.currentTimeMillis());
log.info("路段渠化RID:{},方向：{},持续时间：{}ms，拥堵指数：{}，拥堵状态：{}，平均速度：{}，车辆数：{},路段长度：{}m, 自由流速度：{}，开始时间：{},截止时间：{}", key, outModel.getRidDir(), duration, trafficIndex, trafficState, avgSpeed, segmentCarSize, outModel.getRidLength(), freeSpeed, outModel.getStartTime(), outModel.getEndTime());
} else {
} else {
outModel.setCreateTime(agoTimeStamp);
}
} else {
//第一次缓存创建时间
outModel.setCreateTime(System.currentTimeMillis());
}
roadStateMapState.put(outModel.getRid(), outModel);
}
......@@ -192,7 +204,8 @@ public class RidProcessWindow extends ProcessWindowFunction<CarTrackModel, LaneT
//米
double distance = CommonUtil.getTravelDistance(carList);
//秒
travelTime = (carList.get(carList.size() - 1).getGlobalTimeStamp() - carList.get(0).getGlobalTimeStamp()) / 1000;
travelTime = (carList.get(carList.size() - 1).getGlobalTimeStamp() - carList.get(0).getGlobalTimeStamp()) / 1000.0;
travelTime = ArithOfBigDecmial.round(travelTime, 2);
if (travelTime > 0) {//单车平均行驶速度km/h
travelSpeed = (distance / 1000) / (travelTime / 60 / 60.0);
}
......
wj-realtime-computing/src/main/java/com/wanji/indicators/task/turnIndicators/func/TurnDataProcessWindowByAngle.java
View file @ 5c7c87f4
......@@ -27,10 +27,13 @@ import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;
......@@ -73,6 +76,8 @@ public class TurnDataProcessWindowByAngle extends ProcessWindowFunction<CrossFra
String startTime = DateUtil.toDateTime(windowStartTime, DateUtil.YYYY_MM_DD_HH_MM_SS);
String endTime = DateUtil.toDateTime(windowEndTime, DateUtil.YYYY_MM_DD_HH_MM_SS);
long globalTime = windowEndTime;
Map<String, TurnResultModel> turnResultModelMap = new HashMap<>();
//按转向计算
for (Map.Entry<String, List<CarTrackModel>> entry : groupById.entrySet()) {
......@@ -131,5 +136,21 @@ public class TurnDataProcessWindowByAngle extends ProcessWindowFunction<CrossFra
out.collect(crossTurnResultModel);
}
//状态过期清理
Set<String> expireKeys = new HashSet<>();
Iterator<Map.Entry<String,Long>> iterator = distinctIdState.iterator();
while (iterator.hasNext()) {
Map.Entry<String,Long> entry = iterator.next();
long timestamp = entry.getValue();
if (globalTime - timestamp > 5 * 60 * 1000) {
//缓存超过
expireKeys.add(entry.getKey());
}
}
if (!expireKeys.isEmpty()) {
for (String key : expireKeys) {
distinctIdState.remove(key);
}
}
}
}
wj-realtime-computing/src/main/java/com/wanji/indicators/util/CommonUtil.java
View file @ 5c7c87f4
......@@ -50,7 +50,6 @@ public class CommonUtil {
Double ey = Double.parseDouble(laneEndPoint.split(",")[1]);
double laneAngle = PtInPolyUtil.getAngle(fx, fy, ex, ey);
double laneLength = GeomsConvertUtil.getDistance(laneStartPoint,laneEndPoint);
//2进口道 3出口道
Integer type = baseLaneInfo.getType();
carTrack.getRoadnet().setLaneType(type);
......@@ -60,7 +59,7 @@ public class CommonUtil {
carTrack.getRoadnet().setRidDir8(ridDir);
carTrack.getRoadnet().setLaneAngle(laneAngle);
carTrack.getRoadnet().setCrossId(baseLaneInfo.getCrossId());
carTrack.getRoadnet().setLaneLength(laneLength);
carTrack.getRoadnet().setLaneLength(baseLaneInfo.getLength());
carTrack.getRoadnet().setRidLength(baseLaneInfo.getRidLength());
//出口道对应的哪些进口转向可以进入
if (type.equals(3)) {
......
wj-realtime-computing/src/main/resources/mapper/BaseLaneInfoMapper.xml
View file @ 5c7c87f4
......@@ -131,72 +131,32 @@
<!-- 查询路口进口车道、出口车道信息 -->
<select id="findCrossLaneInfo" parameterType="Map" resultMap="BaseLaneInfoMap">
SELECT t.cross_id,
t.lane_id,
t.rid,
t.type,
t.dir,
t.turn,
t.cut_point as wkt,
t.lane_id,
t.start_point,
t.end_point,
t.road_wkt rid_wkt,
t.end_angle,
t.segment_id,
t.rid_length
FROM
( <!-- 进口方向RID车道 -->
select t1.id lane_id,
substring_index(t1.wkt,';',1) start_point,
substring_index(t1.wkt,';',-1) end_point,
substring_index(t1.wkt,';',-3) cut_point,
t1.rid,t2.cross_id,t2.wkt road_wkt,t1.type,
t2.dir,t2.road_id,t1.turn,t2.end_angle,
t1.segment_id,
t2.rid_length
from t_base_lane_info t1
JOIN (<!-- 取end_cross_id -->
select a.id,a.end_cross_id cross_id,a.in_dir as dir, 2 as type,a.wkt,if(a.road_id='#',concat_ws('',a.id,'#'),a.road_id) road_id,
a.end_angle,a.length rid_length
from t_base_rid_info a
JOIN t_base_cross_info b on a.end_cross_id=b.id
where b.is_signal=1
<if test="crossList != null and crossList.size>0">
and end_cross_id in
<foreach item="item" index="index" collection="crossList" separator="," open="(" close=")">
#{item}
</foreach>
</if>
) t2 on t1.rid=t2.id and t1.type=t2.type
UNION
<!-- 出口方向RID车道 -->
select t1.id lane_id,
substring_index(t1.wkt,';',1) start_point,
substring_index(t1.wkt,';',-1) end_point,
substring_index(t1.wkt,';',3) cut_point,
t1.rid,t2.cross_id,t2.wkt road_wkt,t1.type,
t2.dir,t2.road_id,t1.turn,t2.end_angle,
t1.segment_id,
t2.rid_length
from t_base_lane_info t1
JOIN (<!-- 取start_cross_id -->
select a.id,start_cross_id cross_id,a.out_dir as dir, 3 as type,a.wkt,if(a.road_id='#',concat_ws('',a.id,'#'),a.road_id) road_id,
a.end_angle,a.length rid_length
from t_base_rid_info a
JOIN t_base_cross_info b on a.start_cross_id=b.id
where b.is_signal=1
<if test="crossList != null and crossList.size>0">
and start_cross_id in
<foreach item="item" index="index" collection="crossList" separator="," open="(" close=")">
#{item}
</foreach>
</if>
) t2 on t1.rid=t2.id and t1.type=t2.type
) t
select t1.cross_id,
t1.id lane_id,
substring_index(t1.wkt,';',1) start_point,
substring_index(t1.wkt,';',-1) end_point,
t1.wkt wkt,
t1.rid,
t1.cross_id,
t2.wkt rid_wkt,
t1.type,
t1.dir,
t1.turn,
t2.end_angle,
t1.segment_id,
t2.length rid_length
from t_base_lane_info t1
join t_base_rid_info t2
on t2.end_cross_id=t1.cross_id
where 1=1
<if test="type !=null and type != ''">
and t.type = #{type}
and t1.type = #{type}
</if>
<if test="crossList != null and crossList.size>0">
and t1.cross_id in
<foreach item="item" index="index" collection="crossList" separator="," open="(" close=")">
#{item}
</foreach>
</if>
</select>
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment