xuwang
/
cross_doctor
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

提交除配时方案异常外的问题诊断完成代码，未经完整测试

This commit is contained in:
wangxu 2025-10-28 15:49:58 +08:00
parent 6b70ef10b8
commit 8030756d29
2 changed files with 201 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
app/cross_evaluate_worker.py
View File

@ -207,7 +207,7 @@ def query_cross_problems(params):
cross_phase, err = QueryCrossRunningPhase(nodeid, [crossid], date_list, time_range)
if err or not cross_phase or cross_phase.code != 0:
logging.warning("路口未录入配时方案")
gen_cross_problems(avg_cross_delay_info, roads_dir_dict, inroad_static_info_dict, cross_phase, is_peak)
gen_cross_problems(crossid, nodeid, date_list, avg_cross_delay_info, roads_dir_dict, inroad_static_info_dict, cross_phase, is_peak, cross_ledger_info)
# 指标变化趋势接口

216
app/eva_common.py
View File

@ -822,14 +822,18 @@ def calc_overview_change_rate(overview_data, prev_overview_data):
# 变化率 0表示正常 1表示恶化 2表示优化
if i in (0, 1, 2, 3, 6, 7, 8):
# 指标提升代表恶化的有停车次数、多次停车率、转向失衡指数、停车时间、延误时间、相对流量、进口道流量占比
rate = (overview_data[i] - prev_overview_data[i]) / prev_overview_data[i] * 100 if prev_overview_data[i] != '-' and prev_overview_data[i] > 0 else 0
if overview_data[i] == '-' or prev_overview_data[i] == '-':
continue
rate = (overview_data[i] - prev_overview_data[i]) / prev_overview_data[i] * 100 if prev_overview_data[i] > 0 else 0
if rate < -20:
res_data[i + 10] = 1
elif rate > 20:
res_data[i + 10] = 2
else:
# 指标下降代表恶化的有平均速度、不停车速度
rate = (overview_data[i] - prev_overview_data[i]) / prev_overview_data[i] * 100 if prev_overview_data[i] != '-' and prev_overview_data[i] > 0 else 0
if overview_data[i] == '-' or prev_overview_data[i] == '-':
continue
rate = (overview_data[i] - prev_overview_data[i]) / prev_overview_data[i] * 100 if prev_overview_data[i] > 0 else 0
if rate > 20:
res_data[i + 10] = 1
elif rate < -20:
@ -909,12 +913,12 @@ def parse_data2pb(data_list):
return res_list
def gen_cross_problems(avg_cross_delay_info, roads_dir_dict, inroad_static_info_dict, cross_phase, is_peak):
def gen_cross_problems(crossid, nodeid, date_list, avg_cross_delay_info, roads_dir_dict, inroad_static_info_dict, cross_phase, is_peak, cross_ledger_info):
# 运行效率、均衡调控、配时方案、路口渠化
operating_efficiency_problems = gen_operating_efficiency_problems(avg_cross_delay_info, roads_dir_dict, cross_phase, is_peak)
balanced_control_problems = gen_balanced_control_problems(avg_cross_delay_info, roads_dir_dict, inroad_static_info_dict, cross_phase)
balanced_control_problems = gen_balanced_control_problems(crossid, nodeid, date_list, avg_cross_delay_info, roads_dir_dict, inroad_static_info_dict, cross_phase)
phase_problems = gen_phase_problems()
cross_channelized_problems = gen_cross_channelized_problems()
cross_channelized_problems = gen_cross_channelized_problems(avg_cross_delay_info, roads_dir_dict, inroad_static_info_dict, cross_ledger_info)
problems = [operating_efficiency_problems, balanced_control_problems, phase_problems, cross_channelized_problems]
return problems
@ -1050,11 +1054,11 @@ def gen_high_stop_time_problems(avg_cross_delay_info, is_peak):
return detail, suggestion
def gen_balanced_control_problems(avg_cross_delay_info, roads_dir_dict, inroad_static_info_dict, cross_phase):
def gen_balanced_control_problems(crossid, nodeid, date_list, avg_cross_delay_info, roads_dir_dict, inroad_static_info_dict, cross_phase):
road_delay_infos = avg_cross_delay_info.inroad_delay_infos
cross_imbalance_detail, cross_imbalance_suggestions = gen_cross_imbalance_problems(road_delay_infos, roads_dir_dict, cross_phase)
turn_imbalance_detail, turn_imbalance_suggestions = gen_turn_imbalance_problems(road_delay_infos, roads_dir_dict, inroad_static_info_dict)
# todo 需要与产品确认路口潮汐问题诊断的逻辑
turn_imbalance_detail, turn_imbalance_suggestions = gen_turn_imbalance_problems(road_delay_infos, roads_dir_dict, inroad_static_info_dict, cross_phase)
cross_tide_problems, cross_tide_suggestions = gen_cross_tide_problems(crossid, nodeid, date_list, roads_dir_dict)
balanced_control_problems = {
'item': '均衡调控',
'values': [
@ -1070,6 +1074,12 @@ def gen_balanced_control_problems(avg_cross_delay_info, roads_dir_dict, inroad_s
'reason': '同一进口道，直行与左转停车次数之差的绝对值大于0.5，且转向停车次数的最大值大于1',
'suggestions': turn_imbalance_suggestions
},
{
'item': '路口潮汐',
'detail': cross_tide_problems,
'reason': '对向进口道，早高峰其中一个方向的进口道与出口道流量比大于150%，晚高峰另一个方向进口道与出口道流量比大于150%',
'suggestions': cross_tide_suggestions
}
]
}
return balanced_control_problems
@ -1230,14 +1240,121 @@ def gen_turn_imbalance_problems(road_delay_infos, roads_dir_dict, inroad_static_
return detail, suggestion
# 路口潮汐问题判定
def gen_cross_tide_problems(crossid, nodeid, date_list, roads_dir_dict):
detail = [{
'child_detail': [
{
'text': '未见异常'
}
]
}]
suggestions = []
am_tp_start, am_tp_end = 't700', '900'
pm_tp_start, pm_tp_end = 't1700', '1900'
am_cross_delay_data_list = db_cross.query_cross_delay_info(crossid, nodeid, date_list, am_tp_start)
pm_cross_delay_data_list = db_cross.query_cross_delay_info(crossid, nodeid, date_list, pm_tp_start)
am_avg_cross_delay_info = gen_avg_cross_delay_pb(am_cross_delay_data_list)
pm_avg_cross_delay_info = gen_avg_cross_delay_pb(pm_cross_delay_data_list)
# 构建对向进口道对
subtend_road_pair = gen_subtend_road_pair(roads_dir_dict)
road_src_dict = {item['in']: item for item in roads_dir_dict}
max_am_flow, max_pm_flow, tide_index, max_am_flow_road, max_pm_flow_road = 0, 99999999999, 0, '', ''
am_inroad_infos = am_avg_cross_delay_info.inroad_delay_infos
pm_inroad_infos = pm_avg_cross_delay_info.inroad_delay_infos
am_inroad_info_dict = {item.inroadid: item for item in am_inroad_infos}
pm_inroad_info_dict = {item.inroadid: item for item in pm_inroad_infos}
# 找出早高峰最高流量的进口道
for am_inroad_info in am_inroad_infos:
am_flow = am_inroad_info.delay_info.car_num
if am_flow > max_am_flow:
max_am_flow = am_flow
max_am_flow_road = am_inroad_info.inroadid
subtend_road_am_flow = am_inroad_info_dict[subtend_road_pair[max_am_flow_road]].delay_info.car_num if subtend_road_pair[max_am_flow_road] in am_inroad_info_dict.keys() else 0
# 如果进口道流量小于20 则无意义
if min(subtend_road_am_flow, max_am_flow) < 20:
return detail, suggestions
# 找出晚高峰最高流量的进口道
for pm_inroad_info in pm_inroad_infos:
pm_flow = pm_inroad_info.delay_info.car_num
if pm_flow > max_pm_flow:
max_pm_flow = pm_flow
max_pm_flow_road = pm_inroad_info.inroadid
# 如果路段id为空或不在对向对里，则无意义
if max_pm_flow_road == '' or max_pm_flow_road not in subtend_road_pair.keys():
return detail, suggestions
subtend_road_pm_flow = pm_inroad_info_dict[subtend_road_pair[max_pm_flow_road]].delay_info.car_num
if min(subtend_road_pm_flow, max_pm_flow) < 20:
return detail, suggestions
Fam = max(max_am_flow_road, subtend_road_am_flow) / min(max_am_flow_road, subtend_road_am_flow)
Fpm = max(max_pm_flow_road, subtend_road_pm_flow) / min(max_pm_flow_road, subtend_road_pm_flow)
if Fam >= 1.5 and Fpm >= 1.5 and max_pm_flow_road == subtend_road_pair[max_am_flow_road]:
src_dir = road_src_dict[max_am_flow_road]
full_src_dir = dir_str_dict[src_dir] + dir_str_dict[src_reverse[src_dir]] + '方向'
am_src_dir = dir_str_dict[src_dir] + '向' + dir_str_dict[src_reverse[src_dir]]
pm_src_dir = dir_str_dict[src_reverse[src_dir]] + '向' + dir_str_dict[src_dir]
detail = [{
'src_dir': full_src_dir,
'child_detail': [
{
'text': '早高峰'
},
{
'src_dir': am_src_dir,
'text': '进口道与出口道流量比为' + str(int(Fam * 100)) + '%，晚高峰'
},
{
'src_dir': pm_src_dir,
'text': '进口道与出口道流量比为' + str(int(Fpm * 100)) + '%，存在潮汐现象'
}
]
}]
suggestions = [
{
'child_detail': [
{
'text': '根据主车流方向调整绿灯时长分配'
}
]
},
{
'child_detail': [
{
'src_dir': full_src_dir,
'text': '机动车车道数双向均为3车道及以上，可以考虑设置潮汐车道'
}
]
}
]
return detail, suggestions
def gen_phase_problems():
pass
def gen_cross_channelized_problems(inroad_static_info_dict):
pass
def gen_cross_channelized_problems(avg_cross_delay_info, roads_dir_dict, inroad_static_info_dict, cross_ledger_info):
road_delay_infos = avg_cross_delay_info.inroad_delay_infos
inroad_num_detail, inroad_num_suggestion = gen_inroad_num_problems(road_delay_infos, inroad_static_info_dict, roads_dir_dict)
inout_lane_num_gap_problems, inout_lane_num_gap_suggestions = gen_in_out_lane_num_gap_problems(cross_ledger_info)
cross_channelized_problems = {
'item': '路口渠化',
'values': [
{
'item': '车道资源不匹配',
'detail': inroad_num_detail,
'reason': '同一进口道，直行或左转的流量占比大于50%，且相对流量占比大于60%，且车道数量占比小于流量占比的一半（多方向放行车道每个方向各算0.5）',
'suggestions': inroad_num_suggestion
},
{
'item': '进出口车道数不匹配',
'detail': inout_lane_num_gap_problems,
'reason': '进口道车道数比对向的出口道车道数多2',
'suggestions': inout_lane_num_gap_suggestions
},
]
}
return cross_channelized_problems
def gen_inroad_num_problems(road_delay_infos, inroad_static_info_dict, roads_dir_dict):
@ -1265,6 +1382,7 @@ def gen_inroad_num_problems(road_delay_infos, inroad_static_info_dict, roads_dir
road_car_num = road_delay.delay_info.car_num
turn_delay_dict = {item.turn_type: item for item in flow_delay_infos}
left_lane_num_half, straight_lane_num_half, right_lane_num_half = count_lsr_half(lane_info)
lane_num_info = count_lsr(lane_info)
sum_half_num = left_lane_num_half + straight_lane_num_half + right_lane_num_half
left_lane_num_rate = round(left_lane_num_half / sum_half_num) if sum_half_num > 0 else 0
straight_lane_num_rate = round(straight_lane_num_half / sum_half_num) if sum_half_num > 0 else 0
@ -1285,13 +1403,15 @@ def gen_inroad_num_problems(road_delay_infos, inroad_static_info_dict, roads_dir
err_src_dirs[src_dir].append({
'turn_type': turn_type_str,
'flow_rate': flow_rate,
'lane_num_rate': lane_num_rate
'lane_num_rate': lane_num_rate,
'lane_num_info': lane_num_info
})
else:
err_src_dirs[src_dir] = [{
'turn_type': turn_type_str,
'flow_rate': flow_rate,
'lane_num_rate': lane_num_rate
'lane_num_rate': lane_num_rate,
'lane_num_info': lane_num_info
}]
if len(err_src_dirs.keys()) > 0:
detail = []
@ -1316,9 +1436,73 @@ def gen_inroad_num_problems(road_delay_infos, inroad_static_info_dict, roads_dir
'child_detail': src_detail
}
)
suggestions.append({
'child_detail': [
{
'text': '调整',
'src_dir': dir_str_dict[src_dir] + '进口'
},
{
'text': '车道分配情况'
}
]
})
suggestions.append(
{
'child_detail': [
{
'src_dir': dir_str_dict[src_dir] + '进口',
'text': f"左转/直行/右转现有车道数分别为{err_src_dirs[src_dir][0]['lane_num_info']}"
}
]
}
)
return detail, suggestions
pass
pass
def gen_in_out_lane_num_gap_problems(cross_ledger_info):
detail = [{
'child_detail': [
{
'text': '未见异常'
}
]
}]
suggestions, err_src_dict = [], {}
road_infos = cross_ledger_info['roads']
for src_dir in road_infos.keys():
entry_lane_num = road_infos[src_dir]['entry_lane_num']
exit_lane_num = '-'
if src_reverse[src_dir] in road_infos.keys():
exit_lane_num = road_infos[src_reverse[src_dir]]['exit_lane_num']
if exit_lane_num != '-' and entry_lane_num != '-' and entry_lane_num - exit_lane_num > 2:
err_src_dict[src_dir] = {
'in': entry_lane_num,
'out': exit_lane_num
}
if len(err_src_dict.keys()) > 0:
detail = []
suggestions = {
'child_detail': [
{
'text': '预留全红时间清空积压车辆'
}
]
}
for src_dir in err_src_dict.keys():
detail.append([
{
'src_dir': dir_str_dict[src_dir] + '进口'
},
{
'child_detail': [
{
'text': '进口道车道数（' + str(err_src_dict[src_dir]['in']) + '）比对向出口道车道数（' + str(err_src_dict[src_dir]['out']) + '）多，车辆冰岛减速可能会造成路口车辆积压'
}
]
}
])
return detail, suggestions
def check_is_peak_tp(time_range, area_id, nodeid):