Skip to content

Z
ZHHT-IRN-BD-ANALYSIS
Commit 3463a1c9 by 黄准
Options

车道指标增加日志级别配置 方便现场调试程序

parent 0f11e5b1
Showing with 129 additions and 25 deletions
realtime/hologram-streaming/src/main/java/com/zhht/irn/functions/LaneNormProcessFunction.java
package com.zhht.irn.functions;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSON;
import com.zhht.irn.entity.*;
import com.zhht.irn.enums.EventType;
import com.zhht.irn.job.LaneNormJob;
import com.zhht.irn.utils.*;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
......@@ -158,9 +160,9 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
}
}
//获取所有方向的指标
getPositionNorm(list, positionList, cycle, laneNorms);
getPositionNorm(list, positionList, f1, laneNorms);
//获取所有流向的指标
getPositionNameNorm(list, positionNameList, cycle, laneNorms);
getPositionNameNorm(list, positionNameList, f1, laneNorms,loggerLevel);
if(laneNorms.size()>0){
out.collect(laneNorms);
}
......@@ -170,7 +172,6 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
List<TravelEvent> travelEvents1 = cycleTravelTemp.get(cycle.getCrossCode())==null?new ArrayList<>():cycleTravelTemp.get(cycle.getCrossCode());
//未参与计算的事件数据
List<TravelEvent> futureEvents = travelEvents.stream().filter(a -> Long.parseLong(a.getEventTime()) > Long.parseLong(cycle.getEndDateTime())).collect(Collectors.toList());
//计算中进入的数据 未参与计算的事件数据 本次计算中未发生进入事件的数据进行合并 放入内存中 下周期继续计算
collect.addAll(travelEvents1);
collect.addAll(futureEvents);
......@@ -231,18 +232,32 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
return AllGreenTime + "";
}
private static String gettEffectGreenTimeForDirction(List<Stage> stageList, String position,String direction) {
/**
* 北 直行
* @param stageList
* @param position
* @param direction
* @return
*/
private static String gettEffectGreenTimeForDirction(List<Stage> stageList, String position,String direction,String loggerLevel) {
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();
if("debug".equals(loggerLevel)) {
System.out.println(position + direction+"命中->相位"+stage.getPhaseValue() + "相位有效绿灯时间为->"+stage.getValidGreen() );
}
//同一个相位累加一次即可
break;
}
}
}
if("debug".equals(loggerLevel)){
System.out.println(position+direction+"流向有效绿灯时间为->"+AllGreenTime);
}
return AllGreenTime + "";
}
......@@ -276,7 +291,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
* @param cycle
* @param collectList
*/
private static void getPositionNameNorm(List<LaneNorm> list, List<Map> positionNameList, Cycle cycle, List<LaneNorm> collectList) {
private static void getPositionNameNorm(List<LaneNorm> list, List<Map> positionNameList, Cycle cycle, List<LaneNorm> collectList,String loggerLevel) {
if (list.size() > 0) {
for (Map map : positionNameList) {
String position = String.valueOf(map.get("position"));
......@@ -308,8 +323,8 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
//double sum2 = collect.stream().mapToDouble(LaneNorm::getResidualCapacity).sum();
laneNorm.setResidualCapacity(new BigDecimal(sum2<=0?0d:sum2).setScale(5, BigDecimal.ROUND_HALF_UP).doubleValue());
//laneNorm.setResidualCapacity(new BigDecimal(sum2).setScale(5, BigDecimal.ROUND_HALF_UP).doubleValue());
//有效绿灯时间 相位下此方向非右转的绿灯有效时间之和
laneNorm.setEffectGreenTime(gettEffectGreenTimeForDirction(cycle.getStageList(), position,name));
//有效绿灯时间 相位下此方向此流向有效绿灯时间之和
laneNorm.setEffectGreenTime(gettEffectGreenTimeForDirction(cycle.getStageList(), position,name,loggerLevel));
//识别空间空间占有率 车道长度需要交通流量为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();
......@@ -376,17 +391,108 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
}
public static void main(String[] args) {
List<LaneNorm> objects = new ArrayList<LaneNorm>();
LaneNorm laneNorm = new LaneNorm();
laneNorm.setTrafficCapacity(100d);
laneNorm.setPassCapacity(50d);
LaneNorm laneNorm1 = new LaneNorm();
laneNorm1.setTrafficCapacity(100d);
laneNorm1.setPassCapacity(130d);
objects.add(laneNorm);
objects.add(laneNorm1);
Double reduce = objects.stream().reduce(0.0, (x, y) -> x + (y.getPassCapacity() - y.getTrafficCapacity()), Double::sum);
System.out.println(reduce);
try {
Connection connection = getConnection();
List<Map> dictResult = getDictResult(connection, "SELECT\n" +
" a.ID id,\n" +
" a.Direction direction,\n" +
" b.Name traffic_flow_direction,\n" +
" c.PID phase_no,\n" +
" c.PName phase_name\n" +
"FROM dim_phase_movement a\n" +
"LEFT JOIN dim_traffic_flow_direction b\n" +
" on a.TrafficFlowId = b.ID\n" +
"LEFT JOIN dim_dic_phase c\n" +
" on a.PhaseId = c.ID");
connection.close();
Cycle cycle = setCycle();
Cycle cycle1 = cycle;
for (Stage stage : cycle1.getStageList()) {
stage.setStageForDirectionList(getStageForDiretion(stage, dictResult));
}
String s = gettEffectGreenTimeForDirction(cycle.getStageList(), "北", "直行","debug");
System.out.println(s);
}catch (Exception e){
System.out.println(e);
}
}
public static Cycle setCycle()throws Exception{
String cycleJson="{\n" +
" \"id\": 39240,\n" +
" \"crossCode\": \"13070200022\",\n" +
" \"signalCode\": \"22\",\n" +
" \"beginControlModeCode\": \"21\",\n" +
" \"endControlModeCode\": \"21\",\n" +
" \"beginPlanId\": 33638,\n" +
" \"endPlanId\": 0,\n" +
" \"beginDateTime\": 1673084326000,\n" +
" \"endDateTime\": 1673084481000,\n" +
" \"duration\": 155,\n" +
" \"cycleOrder\": 2695,\n" +
" \"stageList\": [{\n" +
" \"id\": 97300,\n" +
" \"phaseValue\": \"5\",\n" +
" \"beginDateTime\": 1673084326000,\n" +
" \"endDateTime\": 1673084370000,\n" +
" \"beginControlModeCode\": \"21\",\n" +
" \"endControlModeCode\": \"21\",\n" +
" \"beginPlanId\": 33638,\n" +
" \"endPlanId\": 0,\n" +
" \"duration\": 44,\n" +
" \"green\": 41,\n" +
" \"yellow\": 3,\n" +
" \"allRed\": 0,\n" +
" \"phaseOrder\": 1\n" +
" }, {\n" +
" \"id\": 97302,\n" +
" \"phaseValue\": \"6\",\n" +
" \"beginDateTime\": 1673084370000,\n" +
" \"endDateTime\": 1673084417000,\n" +
" \"beginControlModeCode\": \"21\",\n" +
" \"endControlModeCode\": \"21\",\n" +
" \"beginPlanId\": 33638,\n" +
" \"endPlanId\": 0,\n" +
" \"duration\": 46,\n" +
" \"green\": 43,\n" +
" \"yellow\": 3,\n" +
" \"allRed\": 0,\n" +
" \"phaseOrder\": 2\n" +
" }, {\n" +
" \"id\": 97303,\n" +
" \"phaseValue\": \"8\",\n" +
" \"beginDateTime\": 1673084417000,\n" +
" \"endDateTime\": 1673084455000,\n" +
" \"beginControlModeCode\": \"21\",\n" +
" \"endControlModeCode\": \"21\",\n" +
" \"beginPlanId\": 33638,\n" +
" \"endPlanId\": 0,\n" +
" \"duration\": 39,\n" +
" \"green\": 36,\n" +
" \"yellow\": 3,\n" +
" \"allRed\": 0,\n" +
" \"phaseOrder\": 3\n" +
" }, {\n" +
" \"id\": 97305,\n" +
" \"phaseValue\": \"0\",\n" +
" \"beginDateTime\": 1673084455000,\n" +
" \"endDateTime\": 1673084481000,\n" +
" \"beginControlModeCode\": \"21\",\n" +
" \"endControlModeCode\": \"21\",\n" +
" \"beginPlanId\": 33638,\n" +
" \"endPlanId\": 0,\n" +
" \"duration\": 26,\n" +
" \"green\": 23,\n" +
" \"yellow\": 3,\n" +
" \"allRed\": 0,\n" +
" \"phaseOrder\": 4\n" +
" }]\n" +
"}";
Cycle cycle = JSON.parseObject(cycleJson, Cycle.class);
Cycle cycle1 = LaneNormJob.getCycle(cycle);
return cycle1;
}
/**
......@@ -458,7 +564,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
//计算空间占有长度
laneNorm.setSpotSpaceLength(maxQueueLength);
if("debug".equals(loggerLevel)){
System.out.println("车道指标信息->"+laneNorm.toString()+" 车道饱和度->"+saturation_flow_rate+" 有效绿灯时间为->"+effcetGreenTime[1]+" 周期有效绿灯时间->"+cycle.getDuration());
//System.out.println("车道指标信息->"+laneNorm.toString()+" 车道饱和度->"+saturation_flow_rate+" 有效绿灯时间为->"+effcetGreenTime[1]+" 周期有效绿灯时间->"+cycle.getDuration());
}
return laneNorm;
}
......@@ -582,12 +688,10 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
return (maxQueneLength * 6);
}
private static DruidPooledConnection getConnection() throws Exception {
private static Connection getConnection() throws Exception {
//修改为从连接池中取连接
//DruidPooledConnection connection = DruidConnectPoolUtils.getConnection();
//Connection connection = DorisUtils.getConnection();
//return connection;
return null;
Connection connection = DorisUtils.getConnection();
return connection;
}
private static List<Map> getDictResult(Connection connection, String sql) throws Exception {
......
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