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

修复部分bug，提交路口对比报告相关代码（未经测试且尚未完全完成，但不影响线上内容）

This commit is contained in:
wangxu 2026-05-18 16:23:15 +08:00
parent d352df3846
commit 9a6d936c73
3 changed files with 116 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
app/compare_views.py
View File

@ -4,6 +4,7 @@
# @Description: 路口对比评测相关接口
from flask import request
from app.cross_compare_report import *
from app.cross_compare_worker import *
from app.cross_eva_views import app
@ -56,3 +57,18 @@ def query_cross_survey_usable_dates_api():
@app.route('/api/cross_survey_pngs', methods=['GET'])
def cross_survey_pngs_api():
return query_cross_survey_pngs(request.args)
@app.route('/api/gen_compare_report', methods=['POST'])
def gen_compare_report_api():
return gen_cross_compare_report(request.json)
@app.route('/api/save_report_record', methods=['POST'])
def save_report_record_api():
return collect_report(request.args)
@app.route('/api/query_report_records', methods=['GET'])
def query_report_records_api():
return query_collect_report_record(request.args)

103
app/eva_common.py
View File

@ -1588,11 +1588,12 @@ def gen_operating_efficiency_problems(avg_cross_delay_info, roads_dir_dict, cros
operating_efficiency_problems = {
'item': '运行效率',
'values': [],
'total_num': 0
'total_num': 0,
'detail': []
}
road_delay_infos = avg_cross_delay_info.inroad_delay_infos
high_park_problems, high_park_suggestions, high_park_total_num = gen_high_park_problems(road_delay_infos, roads_dir_dict, cross_phase)
high_stop_times_problems, high_stop_times_suggestions, high_stop_times_total_num = gen_high_stop_time_problems(avg_cross_delay_info, is_peak)
high_park_problems, high_park_suggestions, high_park_total_num, high_park_problem_info_detail = gen_high_park_problems(road_delay_infos, roads_dir_dict, cross_phase)
high_stop_times_problems, high_stop_times_suggestions, high_stop_times_total_num, high_stop_time_problems_detail = gen_high_stop_time_problems(avg_cross_delay_info, is_peak)
sum_num = 0
if high_park_total_num > 0:
sum_num += high_park_total_num
@ -1600,7 +1601,8 @@ def gen_operating_efficiency_problems(avg_cross_delay_info, roads_dir_dict, cros
'item': '多次排队',
'detail': high_park_problems,
'reason': '某一进口道转向的多次停车率大于15%',
'suggestions': high_park_suggestions
'suggestions': high_park_suggestions,
'problem_infos': high_park_problem_info_detail
})
if high_stop_times_total_num > 0:
sum_num += high_stop_times_total_num
@ -1608,7 +1610,8 @@ def gen_operating_efficiency_problems(avg_cross_delay_info, roads_dir_dict, cros
'item': '停车较多',
'detail': high_stop_times_problems,
'reason': '高峰时段路口停车次数大于2次，非高峰时段停车次数大于1次',
'suggestions': high_stop_times_suggestions
'suggestions': high_stop_times_suggestions,
'problem_infos': high_stop_time_problems_detail
})
operating_efficiency_problems['total_num'] = sum_num
return operating_efficiency_problems
@ -1619,6 +1622,11 @@ def gen_high_park_problems(road_delay_infos, roads_dir_dict, cross_phase):
detail = []
suggestion, total_num = [], 0
err_src_dict = {}
problem_info_detail = {
'item_name': '多次排队',
'item_detail': [],
'item_detail_str': ''
}
src_list = list(roads_dir_dict.keys())
road_delays_dict = {item.inroadid: item for item in road_delay_infos}
for src_dir in src_list:
@ -1640,7 +1648,7 @@ def gen_high_park_problems(road_delay_infos, roads_dir_dict, cross_phase):
detail = []
for src_dir in err_src_dict.keys():
total_num = 1
src_detail, src_suggestions, turn_type_list = [], [], []
src_detail, src_suggestions, turn_type_list, item_detail = [], [], [], []
for turn_type in err_src_dict[src_dir]:
turn_type_list.append(int(turn_type.split(':')[0]))
flow_detail = {
@ -1648,6 +1656,7 @@ def gen_high_park_problems(road_delay_infos, roads_dir_dict, cross_phase):
'text': '车辆多次停车率过大（' + turn_type.split(':')[1] + '%），'
}
src_detail.append(flow_detail)
item_detail.append(f"{int_turn_type2str[int(turn_type.split(':')[0])]}车辆多次停车率过大（{turn_type.split(':')[1]}%）")
src_detail.append({
'text': '车辆需要多次排队才能通过'
})
@ -1657,6 +1666,8 @@ def gen_high_park_problems(road_delay_infos, roads_dir_dict, cross_phase):
'child_detail': src_detail
}
])
item_str = '、'.join([f"{dir_str_dict[src_dir]}进口道" + item for item in item_detail])
problem_info_detail['item_detail'].append(item_str)
turn_type_str = '、'.join([int_turn_type2str[int(item.split(':')[0])] for item in err_src_dict[src_dir]])
src_suggestions.append(
{
@ -1673,7 +1684,8 @@ def gen_high_park_problems(road_delay_infos, roads_dir_dict, cross_phase):
'next_line': phase_suggestions
})
# suggestion.extend(phase_suggestions)
return detail, suggestion, total_num
problem_info_detail['item_detail_str'] = '；'.join(problem_info_detail['item_detail']) + '车辆需要多次排队才能通过。'
return detail, suggestion, total_num, problem_info_detail
# 运行效率-停车较多
@ -1681,11 +1693,17 @@ def gen_high_stop_time_problems(avg_cross_delay_info, is_peak):
detail = []
suggestion, total_num = [], 0
desc, max_stop_times = '非高峰时段', 1
problem_info_detail = {
'item_name': '停车较多',
'item_detail': [],
'item_detail_str': ''
}
if is_peak:
# 表示为高峰时段，否则为平峰时段
desc, max_stop_times = '高峰时段', 2
if avg_cross_delay_info:
if avg_cross_delay_info.delay_info.stop_times > max_stop_times and avg_cross_delay_info.delay_info.car_num >= 10:
problem_info_detail['item_detail_str'] = '车辆多次停车时长过长（' + str(round(avg_cross_delay_info.delay_info.stop_times, 2)) + '），整体运行效率不高'
total_num = 1
detail = [[
{
@ -1710,15 +1728,15 @@ def gen_high_stop_time_problems(avg_cross_delay_info, is_peak):
]
}
]
return detail, suggestion, total_num
return detail, suggestion, total_num, problem_info_detail
# 均衡调控
def gen_balanced_control_problems(crossid, nodeid, date_list, avg_cross_delay_info, roads_dir_dict, inroad_static_info_dict, cross_phase, weekdays):
road_delay_infos = avg_cross_delay_info.inroad_delay_infos
cross_imbalance_detail, cross_imbalance_suggestions, cross_imbalance_total_num = gen_cross_imbalance_problems(road_delay_infos, roads_dir_dict, cross_phase)
turn_imbalance_detail, turn_imbalance_suggestions, turn_imbalance_total_num = gen_turn_imbalance_problems(road_delay_infos, roads_dir_dict, inroad_static_info_dict, cross_phase)
cross_tide_problems, cross_tide_suggestions, cross_tide_total_num = gen_cross_tide_problems(crossid, nodeid, date_list, roads_dir_dict, weekdays)
cross_imbalance_detail, cross_imbalance_suggestions, cross_imbalance_total_num, cross_imbalance_problem_info_detail = gen_cross_imbalance_problems(road_delay_infos, roads_dir_dict, cross_phase)
turn_imbalance_detail, turn_imbalance_suggestions, turn_imbalance_total_num, turn_imbalance_problem_info_detail = gen_turn_imbalance_problems(road_delay_infos, roads_dir_dict, inroad_static_info_dict, cross_phase)
cross_tide_problems, cross_tide_suggestions, cross_tide_total_num, cross_tide_problem_info_detail = gen_cross_tide_problems(crossid, nodeid, date_list, roads_dir_dict, weekdays)
balanced_control_problems = {
'item': '均衡调控',
'values': [],
@ -1732,7 +1750,8 @@ def gen_balanced_control_problems(crossid, nodeid, date_list, avg_cross_delay_in
'item': '路口失衡',
'detail': cross_imbalance_detail,
'reason': '路口存在某个流向绿灯时长不足而另一个方向绿灯存在空放的现象',
'suggestions': cross_imbalance_suggestions
'suggestions': cross_imbalance_suggestions,
'problem_infos': cross_imbalance_problem_info_detail
}
)
if turn_imbalance_total_num > 0:
@ -1742,7 +1761,8 @@ def gen_balanced_control_problems(crossid, nodeid, date_list, avg_cross_delay_in
'item': '转向失衡',
'detail': turn_imbalance_detail,
'reason': '同一进口道，直行与左转停车次数之差的绝对值大于0.5，且转向停车次数的最大值大于1',
'suggestions': turn_imbalance_suggestions
'suggestions': turn_imbalance_suggestions,
'problem_infos': turn_imbalance_problem_info_detail
}
)
if cross_tide_total_num > 0:
@ -1752,7 +1772,8 @@ def gen_balanced_control_problems(crossid, nodeid, date_list, avg_cross_delay_in
'item': '路口潮汐',
'detail': cross_tide_problems,
'reason': '对向进口道，早高峰其中一个方向的进口道与出口道流量比大于150%，晚高峰另一个方向进口道与出口道流量比大于150%',
'suggestions': cross_tide_suggestions
'suggestions': cross_tide_suggestions,
'problem_infos': cross_tide_problem_info_detail
}
)
balanced_control_problems['total_num'] = sum_num
@ -1764,6 +1785,11 @@ def gen_cross_imbalance_problems(road_delay_infos, roads_dir_dict, cross_phase):
detail = []
suggestion, total_num = [], 0
road_src_dict = {v['in']: k for k, v in roads_dir_dict.items()}
problem_info_detail = {
'item_name': '路口失衡',
'item_detail': [],
'item_detail_str': ''
}
max_stop_times_road = max(road_delay_infos, key=lambda x: x.delay_info.stop_times)
min_stop_times_road = min(road_delay_infos, key=lambda x: x.delay_info.stop_times)
if max_stop_times_road.delay_info.stop_times > 1 \
@ -1792,6 +1818,7 @@ def gen_cross_imbalance_problems(road_delay_infos, roads_dir_dict, cross_phase):
}
]
]
problem_info_detail['item_detail_str'] = f"{dir_str_dict[max_src]}进口的停车次数（{str(round(max_stop_times_road.delay_info.stop_times, 2))}）与{dir_str_dict[min_src]}进口的停车次数（{str(round(min_stop_times_road.delay_info.stop_times, 2))}）相差过大，两者之比为{rate}%，分配的绿灯时长不匹配"
is_tide = False
if max_src == src_reverse[min_src]:
is_tide = True
@ -1815,7 +1842,7 @@ def gen_cross_imbalance_problems(road_delay_infos, roads_dir_dict, cross_phase):
}
]
# suggestion.extend(phase_suggestions)
return detail, suggestion, total_num
return detail, suggestion, total_num, problem_info_detail
# 均衡调控-转向失衡问题诊断
@ -1823,6 +1850,11 @@ def gen_turn_imbalance_problems(road_delay_infos, roads_dir_dict, inroad_static_
detail = []
suggestion, total_num = [], 0
err_road_dict = {}
problem_info_detail = {
'item_name': '转向失衡',
'item_detail': [],
'item_detail_str': ''
}
road_src_dict = {v['in']: k for k, v in roads_dir_dict.items()}
for road_delay_info in road_delay_infos:
inroadid = road_delay_info.inroadid
@ -1863,6 +1895,7 @@ def gen_turn_imbalance_problems(road_delay_infos, roads_dir_dict, inroad_static_
]
}
]
problem_info_detail['item_detail_str'] = dir_str_dict[src_dir] + '进口的停车次数（' + str(err_road_dict[src_dir][0].split(':')[1]) + '）与' + err_road_dict[src_dir][1].split(':')[0] + '进口的停车次数（' + str(err_road_dict[src_dir][1].split(':')[1]) + '）相差过大，两者之比为' + str(int(float(err_road_dict[src_dir][0].split(':')[1]) / float(err_road_dict[src_dir][1].split(':')[1]) * 100)) + '%，分配的绿灯时长不匹配'
detail.append(item_detail)
phase_suggestion = get_flow_phase_detail(src_dir, [0, 1], cross_phase)
item_suggestion = [
@ -1904,12 +1937,17 @@ def gen_turn_imbalance_problems(road_delay_infos, roads_dir_dict, inroad_static_
]
suggestion = suggestion + lane_num_suggestion
return detail, suggestion, total_num
return detail, suggestion, total_num, problem_info_detail
# 均衡调控-路口潮汐问题诊断
def gen_cross_tide_problems(crossid, nodeid, date_list, roads_dir_dict, weekdays):
detail = []
problem_info_detail = {
'item_name': '路口潮汐',
'item_detail': [],
'item_detail_str': ''
}
suggestions, total_num = [], 0
am_tp_start, am_tp_end = 't700', '900'
pm_tp_start, pm_tp_end = 't1700', '1900'
@ -1950,7 +1988,7 @@ def gen_cross_tide_problems(crossid, nodeid, date_list, roads_dir_dict, weekdays
# 如果路段id为空或不在对向对里，则无意义
if max_pm_flow_road == '' or max_pm_flow_road not in subtend_road_pair.keys():
return detail, suggestions, total_num
subtend_road_pm_flow = pm_inroad_info_dict[subtend_road_pair[max_pm_flow_road]].delay_info.car_num
subtend_road_pm_flow = pm_inroad_info_dict[subtend_road_pair[max_pm_flow_road]].delay_info.car_num if subtend_road_pair[max_pm_flow_road] in pm_inroad_info_dict.keys() else 0
if min(subtend_road_pm_flow, max_pm_flow) < 20:
return detail, suggestions, total_num
Fam = max(max_am_flow_road, subtend_road_am_flow) / min(max_am_flow_road, subtend_road_am_flow)
@ -1977,6 +2015,7 @@ def gen_cross_tide_problems(crossid, nodeid, date_list, roads_dir_dict, weekdays
}
]
}]
problem_info_detail['item_detail_str'] = '早高峰' + am_src_dir + '进口道与出口道流量比为' + str(int(Fam * 100)) + '%，晚高峰' + pm_src_dir + '进口道与出口道流量比为' + str(int(Fpm * 100)) + '%，存在潮汐现象'
suggestions = [
{
'child_detail': [
@ -1994,7 +2033,7 @@ def gen_cross_tide_problems(crossid, nodeid, date_list, roads_dir_dict, weekdays
]
}
]
return detail, suggestions, total_num
return detail, suggestions, total_num, problem_info_detail
# 配时方案
@ -2088,8 +2127,8 @@ def gen_err_phase_problems(max_date, crossid, min_date, time_range):
# 路口渠化
def gen_cross_channelized_problems(avg_cross_delay_info, roads_dir_dict, inroad_static_info_dict, cross_ledger_info, time_range):
road_delay_infos = avg_cross_delay_info.inroad_delay_infos
inroad_num_detail, inroad_num_suggestion, inroad_num_err_total_num = gen_inroad_num_problems(road_delay_infos, inroad_static_info_dict, roads_dir_dict, time_range, avg_cross_delay_info.delay_info.car_num)
inout_lane_num_gap_problems, inout_lane_num_gap_suggestions, inout_lane_num_gap_total_num = gen_in_out_lane_num_gap_problems(cross_ledger_info)
inroad_num_detail, inroad_num_suggestion, inroad_num_err_total_num, inroad_num_err_problem_detail = gen_inroad_num_problems(road_delay_infos, inroad_static_info_dict, roads_dir_dict, time_range, avg_cross_delay_info.delay_info.car_num)
inout_lane_num_gap_problems, inout_lane_num_gap_suggestions, inout_lane_num_gap_total_num, inout_lane_num_gap_problem_detail = gen_in_out_lane_num_gap_problems(cross_ledger_info)
cross_channelized_problems = {
'item': '路口渠化',
'values': [],
@ -2103,7 +2142,8 @@ def gen_cross_channelized_problems(avg_cross_delay_info, roads_dir_dict, inroad_
'item': '车道资源不匹配',
'detail': inroad_num_detail,
'reason': '路口的单位小时流量大于20, 同一进口道，针对直行、左转，如果该流向的流量占比大于20%，且该流向的车道数量占比小于流量占比的一半（多方向放行车道每个方向各算0.5）',
'suggestions': inroad_num_suggestion
'suggestions': inroad_num_suggestion,
'problem_infos': inroad_num_err_problem_detail
}
)
if inout_lane_num_gap_total_num > 0:
@ -2113,7 +2153,8 @@ def gen_cross_channelized_problems(avg_cross_delay_info, roads_dir_dict, inroad_
'item': '进出口车道数不匹配',
'detail': inout_lane_num_gap_problems,
'reason': '进口道车道数比对向的出口道车道数多2',
'suggestions': inout_lane_num_gap_suggestions
'suggestions': inout_lane_num_gap_suggestions,
'problem_infos': inout_lane_num_gap_problem_detail
}
)
cross_channelized_problems['total_num'] = sum_num
@ -2123,6 +2164,11 @@ def gen_cross_channelized_problems(avg_cross_delay_info, roads_dir_dict, inroad_
# 路口渠化-车道资源不匹配
def gen_inroad_num_problems(road_delay_infos, inroad_static_info_dict, roads_dir_dict, time_range, car_num):
detail = []
problem_info_detail = {
'item_name': '车道资源不匹配',
'item_detail': [],
'item_detail_str': ''
}
suggestions, err_src_dirs, total_num = [], {}, 0
road_src_dict = {v['in']: k for k, v in roads_dir_dict.items()}
# 20251204 调整计算逻辑，取消相对流量占比大于60% 更换为 路口 单位小时流量大于20
@ -2188,6 +2234,7 @@ def gen_inroad_num_problems(road_delay_infos, inroad_static_info_dict, roads_dir
'text': '分流转向比过大（' + str(int((turn_type['flow_rate']) * 100)) + '%），但分配的转向车道占比不足（' + str(int((turn_type['lane_num_rate']))) + '%），分配车道资源不匹配'
}
)
problem_info_detail['item_detail'].append(f"{dir_str_dict[src_dir]}进口：" + turn_type['turn_type'] + "分流转向比过大（" + str(int((turn_type['flow_rate']) * 100)) + "%），但分配的转向车道占比不足（" + str(int((turn_type['lane_num_rate']))) + "%），分配车道资源不匹配")
detail.append(
[src_detail]
)
@ -2215,7 +2262,8 @@ def gen_inroad_num_problems(road_delay_infos, inroad_static_info_dict, roads_dir
# ]
# }
# )
return detail, suggestions, total_num
problem_info_detail['item_detail_str'] = '；'.join(problem_info_detail['item_detail'])
return detail, suggestions, total_num, problem_info_detail
def calculate_hours(time_range):
@ -2245,6 +2293,11 @@ def calculate_hours(time_range):
# 路口渠化-进出口道数不匹配
def gen_in_out_lane_num_gap_problems(cross_ledger_info):
detail = []
problem_info_detail = {
'item_name': '进出口道数不匹配',
'item_detail': [],
'item_detail_str': ''
}
suggestions, err_src_dict, total_num = [], {}, 0
road_infos = cross_ledger_info['roads']
for src_dir in road_infos.keys():
@ -2278,7 +2331,9 @@ def gen_in_out_lane_num_gap_problems(cross_ledger_info):
]
}
])
return detail, suggestions, total_num
problem_info_detail['item_detail'].append(f"{dir_str_dict[src_dir]}进口道车道数（{err_src_dict[src_dir]['in']}）比对向出口道车道数（{err_src_dict[src_dir]['out']}）多，车辆并道减速可能会造成路口车辆积压")
problem_info_detail['item_detail_str'] = '；'.join(problem_info_detail['item_detail'])
return detail, suggestions, total_num, problem_info_detail
def check_is_peak_tp(time_range, area_id, nodeid):

19
app/tmnet_db_func.py
View File

@ -1,3 +1,4 @@
import json
from datetime import datetime
from app.db_func_base import TableDbHelperBase
@ -789,3 +790,21 @@ class TmnetDbHelper(TableDbHelperBase):
res = self.do_select(sql)
cross_type_dict = {item['mapping_code']: item['web_tag'] for item in res}
return cross_type_dict
def save_collect_report(self, report_type, values):
ret = None
page_params = json.dumps(values['page_params'], ensure_ascii=False) if values.get('page_params') else None
if report_type == 0:
sql = """
insert into cross_doctor_config.collect_report_record (crossid, cross_name, date_range, compare_date_range, time_range, weekday_str, gen_report_dater, download_url, nodeid, area_id, page_params, report_type) value (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
""" % (values['crossid'], values['cross_name'], values['date_range'], values['compare_date_range'], values['time_range'], values['weekday_str'], values['gen_report_date'], values['download_url'], values['nodeid'], values['area_id'], page_params, int(values['report_type']))
ret = self.do_execute(sql)
if ret == 1:
return True
return False
def query_collect_report_records(self, nodeid, area_id, report_type):
sql = """
select * from cross_doctor_config.collect_report_record where nodeid = %s and area_id = %s and report_type = %s
""" % (nodeid, area_id, report_type)
return self.do_select(sql)