Skip to content

Z
ZHHT-IRN-BD-ANALYSIS
Commit 46d6ddbe by 黄准
Options

车道指标相关计算逻辑

parent 2930fc23
Showing with 64 additions and 16 deletions
realtime/hologram-streaming/src/main/java/com/zhht/irn/functions/LaneNormProcessFunction.java
......@@ -40,6 +40,8 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
private ValueState<Boolean> processCycle;
private Connection connection;
//每个路口允许的最大时间数据
private static Long maxTravel=50000L;
@Override
public void open(Configuration parameters) throws Exception {
......@@ -112,6 +114,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
List<TravelEvent> car = getCar(travelEvents,cycle);
//根据车道分割过车数据
Map<String, List<TravelEvent>> carForLane = getCarForLane(car);
//获取周期相位数据
List<Stage> stageList = f1.getStageList();
//获取相位关系
for (Stage stage : stageList) {
......@@ -124,6 +127,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
List<Map> positionList = dict.get("positionList").stream().filter(recode -> f1.getCrossCode().equals(recode.get("cross_id"))).collect(Collectors.toList());
List<Map> positionNameList = dict.get("positionNameList").stream().filter(recode -> f1.getCrossCode().equals(recode.get("cross_id"))).collect(Collectors.toList());
for (Map map : dictResult) {
//根据车道字典计算车道指标
LaneNorm laneNorm = getLaneNorm(map, f1, carForLane, laneList);
list.add(laneNorm);
//交通流量为0得不参与计算
......@@ -138,12 +142,16 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
if(laneNorms.size()>0){
out.collect(laneNorms);
}
//本次计算中未发生进入事件的数据
List<TravelEvent> collect = car.stream().filter(recode -> StringUtils.isEmpty(recode.getInCorssTime())).collect(Collectors.toList());
//计算中进入的数据
List<TravelEvent> travelEvents1 = cycleTravelTemp.get(cycle.getCrossCode())==null?new ArrayList<>():cycleTravelTemp.get(cycle.getCrossCode());
//插入计算过程中缓存的数据
for (TravelEvent travelEvent:travelEvents1){
collect.add(travelEvent);
}
//未参与计算的事件数据
List<TravelEvent> futureEvents = travelEvents.stream().filter(a -> Long.parseLong(a.getEventTime()) > Long.parseLong(cycle.getEndDateTime())).collect(Collectors.toList());
//计算中进入的数据 未参与计算的事件数据 本次计算中未发生进入事件的数据进行合并 放入内存中 下周期继续计算
collect.addAll(travelEvents1);
collect.addAll(futureEvents);
cycleTravel.put(f1.getCrossCode(),collect);
cycleTravelTemp.put(f1.getCrossCode(),new ArrayList<>());
processCycle.update(false);
......@@ -160,13 +168,17 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
if (cycleTravelTemp.get(event.getCrossId()) == null) {
cycleTravelTemp.put(event.getCrossId(), new ArrayList<>());
}
//周期数据处理时，讲数据塞入临时内存中
//如果当前路口存留的事件数据积压 超过阈值 则进行清理操作 十分钟之前的数据丢弃
if(!processCycle.value() && cycleTravel.get(event.getCrossId()).size()>maxTravel){
List<TravelEvent> collect = cycleTravel.get(event.getCrossId()).stream().filter(a -> DateUtils.compareDataIsBefore10Min(a.getEventTime())).collect(Collectors.toList());
cycleTravelTemp.put(event.getCrossId(),collect);
}
//当周期数据处理时，将数据塞入临时内存中
if (processCycle.value()) {
cycleTravelTemp.get(event.getCrossId()).add(event);
} else {
cycleTravel.get(event.getCrossId()).add(event);
}
}
......@@ -174,7 +186,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
public void onTimer(long timestamp, OnTimerContext ctx, Collector<List<LaneNorm>> out) throws Exception {
super.onTimer(timestamp, ctx, out);
System.out.println("触发定时器了。。。。");
//通过定时器清空字典表
//通过定时器清空字典表 防止字典表数据有更新
dict.remove("phaseDirection");
dict.remove("laneList");
dict.remove("dictResult");
......@@ -197,6 +209,21 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
return AllGreenTime + "";
}
private static String gettEffectGreenTimeForDirction(List<Stage> stageList, String position,String direction) {
int AllGreenTime = 0;
for (Stage stage : stageList) {
List<StageForDirection> stageForDirectionList = stage.getStageForDirectionList();
for (StageForDirection dir : stageForDirectionList) {
if (position.equals(dir.getDirection()) && direction.equals(dir.getTraffic_flow_direction())) {
AllGreenTime += stage.getValidGreen();
//同一个相位累加一次即可
break;
}
}
}
return AllGreenTime + "";
}
/**
* 提前计算所有相位允许的流向
*
......@@ -257,8 +284,8 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
//剩余承载力求和
double sum2 = collect.stream().mapToDouble(LaneNorm::getResidualCapacity).sum();
laneNorm.setResidualCapacity(new BigDecimal(sum2).setScale(5, BigDecimal.ROUND_HALF_UP).doubleValue());
//有效绿灯时间 相位下此方向非右转得绿灯有效时间之和
laneNorm.setEffectGreenTime(gettEffectGreenTimeForPosition(cycle.getStageList(), position));
//有效绿灯时间 相位下此方向非右转的绿灯有效时间之和
laneNorm.setEffectGreenTime(gettEffectGreenTimeForDirction(cycle.getStageList(), position,name));
//识别空间空间占有率 车道长度需要交通流量为0得数据
int sum = list.stream().filter(recode -> position.equals(recode.getDirection()) && recode.getLaneName().indexOf(name) != -1).mapToInt(LaneNorm::getLaneLength).sum();
int sum1 = collect.stream().mapToInt(LaneNorm::getMaxQueueLength).sum();
......@@ -485,7 +512,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
if (laneCarMap.containsKey(laneId)) {
//获取这个车道所有的 过车数据
List<TravelEvent> travelEvents = laneCarMap.get(laneId);
//去周期内 每一秒的排队长度 取最大值
//周期内 每一秒的排队长度 取最大值
for(Long start = cycleStart; start <= cycleEnd; start += 1000) {
long start1 = start;
//通过每一秒分析出该车道停留的车数量
......@@ -508,9 +535,10 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
private static DruidPooledConnection getConnection() throws Exception {
//修改为从连接池中取连接
DruidPooledConnection connection = DruidConnectPoolUtils.getConnection();
//DruidPooledConnection connection = DruidConnectPoolUtils.getConnection();
//Connection connection = DorisUtils.getConnection();
return connection;
//return connection;
return null;
}
private static List<Map> getDictResult(Connection connection, String sql) throws Exception {
......@@ -548,6 +576,10 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
if (!DateUtils.compareDataIsBefore10Min(event.getEventTime())) {
continue;
}
//事件数据发生在周期结束时间之后的 不参与计算
if (Long.parseLong(event.getEventTime())>Long.parseLong(cycle.getEndDateTime())) {
continue;
}
if (!map.containsKey(event.getRecordId())) {
map.put(event.getRecordId(), event);
}
......@@ -575,6 +607,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
return lists;
}
//将过车数据根据车道分组
private static Map<String, List<TravelEvent>> getCarForLane(List<TravelEvent> cars) {
Map<String, List<TravelEvent>> stringListHashMap = new HashMap<>();
for (TravelEvent ev : cars) {
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/functions/TravelEventAndCycleCoProcessFunction.java
......@@ -648,14 +648,12 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
* @return
*/
private Double getAverageSpeed(Map<String, TravelEvent> allEventOfCar) {
//根据当前车辆的所有事件数据，计算平均速度
Iterator<TravelEvent> iterator = allEventOfCar.values().iterator();
double sumSpeed =0.0 ;
while(iterator.hasNext()){
sumSpeed= sumSpeed+iterator.next().getEventSpeed();
}
return sumSpeed/allEventOfCar.keySet().size() ;
}
......@@ -665,8 +663,6 @@ public class TravelEventAndCycleCoProcessFunction extends CoProcessFunction<Trav
* @return
*/
private Long getStopDelayTime(List<TravelEvent> travelEvents) {
//按照事件发送的事件顺序进行排序后，再计算停车延误时间
Set<TravelEvent> set = new TreeSet<>();
for (TravelEvent t:travelEvents) {
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/DateUtils.java
......@@ -365,6 +365,25 @@ public class DateUtils {
return 0;
}
public static int compare_dateTime(String DATE1, String DATE2) {
SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd");
try {
Date dt1 = df.parse(DATE1);
Date dt2 = df.parse(DATE2);
if (dt1.getTime() > dt2.getTime()) {
return 1;
} else if (dt1.getTime() < dt2.getTime()) {
return -1;
} else {
return 0;
}
} catch (Exception exception) {
exception.printStackTrace();
}
return 0;
}
public static String getBeforeTime(Calendar beforeTime,int before){
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
......
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