601 lines
32 KiB
Python
601 lines
32 KiB
Python
# -*- coding: utf-8 -*-
|
||
# @Author: Owl
|
||
# @Date: 2026/5/11 14:37
|
||
# @Description: 路口对比报告文件必须数据产出代码
|
||
import os
|
||
|
||
from app.cross_compare_common import *
|
||
from app.report_common import *
|
||
from tool.compare_report import *
|
||
from tool.qcos_func import get_client, CosFolderManager
|
||
g_cos_root = 'https://xinglu-1324629296.cos.ap-beijing.myqcloud.com'
|
||
g_cos_bucket = 'xinglu-1324629296'
|
||
|
||
|
||
weekday2Str = {
|
||
1: '周一',
|
||
2: '周二',
|
||
3: '周三',
|
||
4: '周四',
|
||
5: '周五',
|
||
6: '周六',
|
||
7: '周日'
|
||
}
|
||
|
||
|
||
def gen_cross_compare_report(params):
|
||
crossid = check_param(params, 'crossid')
|
||
if not crossid:
|
||
return json.dumps(make_common_res(1, '缺少crossid， 请刷新后重试'))
|
||
nodeid = check_param(params, 'nodeid')
|
||
if not nodeid:
|
||
return json.dumps(make_common_res(2, '缺少nodeid， 请刷新后重试'))
|
||
area_id = check_param(params, 'area_id')
|
||
if not area_id:
|
||
return json.dumps(make_common_res(3, '缺少area_id， 请刷新后重试'))
|
||
userid = check_param(params, 'userid')
|
||
if not userid:
|
||
return json.dumps(make_common_res(4, '缺少userid， 请刷新后重试'))
|
||
area_list = db_user.query_areaid_list(userid)
|
||
if not area_list or len(area_list) < 1:
|
||
return json.dumps(make_common_res(5, '用户信息异常'))
|
||
area_list = map(int, area_list)
|
||
if not str(area_id).lstrip('-').isdigit() or int(area_id) not in area_list:
|
||
return json.dumps(make_common_res(5, '辖区id异常，请检查后重试'))
|
||
date_list = check_param(params, 'date_list') # 优化后
|
||
if not date_list or len(date_list) < 1:
|
||
return json.dumps(make_common_res(7, '缺少日期参数，请最少选择一天作为查询日期'))
|
||
date_list = list(map(str, date_list))
|
||
compare_date_list = check_param(params, 'compare_date_list') # 优化前
|
||
if not compare_date_list or len(compare_date_list) < 1:
|
||
return json.dumps(make_common_res(6, '缺少对比日期参数，请最少选择一天作为对比日期'))
|
||
compare_date_list = list(map(str, compare_date_list))
|
||
query_type = check_param(params, 'query_type')
|
||
if not query_type:
|
||
query_type = 0
|
||
time_range = check_param(params, 'time_range')
|
||
if not time_range:
|
||
return json.dumps(make_common_res(8, '缺少时段范围，请选择时段范围'))
|
||
weekdays = check_param(params, 'weekdays')
|
||
if not weekdays:
|
||
weekdays = '1,2,3,4,5,6,7'
|
||
tp_start = int(str(time_range.split('-')[0]).split(':')[0]) * 100 + int(str(time_range.split('-')[0]).split(':')[1])
|
||
if query_type == 1:
|
||
tp_start = 't' + str(tp_start)
|
||
elif query_type == 2:
|
||
tp_start = 'h' + str(tp_start)
|
||
parts = check_param(params, 'parts')
|
||
if not parts or len(parts) < 1:
|
||
return json.dumps(make_common_res(11, '缺少章节信息，请选择输出章节内容信息'))
|
||
cut_images = check_param(params, 'cut_images')
|
||
if not cut_images:
|
||
cut_images = 0
|
||
|
||
# 路口静态信息及台账信息
|
||
cross_ledger_info_dict = query_cross_ledger_info(crossid, nodeid, area_id, userid)
|
||
if not cross_ledger_info_dict:
|
||
return json.dumps(make_common_res(10, '查询路口信息失败'))
|
||
cross_static_info, cross_ledger_info = gen_cross_static_info(crossid, nodeid, area_id, cross_ledger_info_dict)
|
||
roads_dir_dict = gen_road_dir_dict(cross_ledger_info)
|
||
gen_report_date = datetime.now().strftime('%Y年%m月%d日')
|
||
file_name = gen_report_file_name(cross_static_info['name'] + '路口优化对比报告' + str(cut_images), date_list + compare_date_list, time_range, parts, gen_report_date)
|
||
# 优化后
|
||
data_range = gen_date_range_str(date_list)
|
||
# 优化前
|
||
compare_date_range = gen_date_range_str(compare_date_list)
|
||
# 验证文件是否已经存在
|
||
cos_client = get_client()
|
||
folder_manager = CosFolderManager(cos_client, g_cos_bucket)
|
||
download_url = g_cos_root + f"/user/cross_doctor/platform_report/{nodeid}/{area_id}/cross_compare_report/" + file_name
|
||
if folder_manager.file_exists(download_url):
|
||
res = make_common_res(0, 'ok')
|
||
res['data'] = {
|
||
'crossid': crossid,
|
||
'nodeid': nodeid,
|
||
'area_id': area_id,
|
||
'name': cross_static_info['name'],
|
||
'file_name': file_name,
|
||
'gen_report_date': gen_report_date,
|
||
'date_range': data_range,
|
||
'weekdays': weekdays,
|
||
'compare_date_range': compare_date_range,
|
||
'weekday_str': gen_weekday_str(weekdays),
|
||
'download_url': download_url
|
||
}
|
||
return json.dumps(res, ensure_ascii=False)
|
||
|
||
dir = os.path.dirname(os.path.abspath(__file__))
|
||
localfile_path = f'{dir}/../temp/{file_name}.lock'
|
||
if os.path.exists(localfile_path):
|
||
return json.dumps(make_common_res(12, '正在生成中，请勿重复提交'))
|
||
with open(localfile_path, 'w') as f:
|
||
pass
|
||
doc1 = DocDocumentContrastCtx('9660')
|
||
cut_image_params = {
|
||
'crossid': crossid,
|
||
'nodeid': nodeid,
|
||
'area_id': area_id,
|
||
'date_list': date_list,
|
||
'compare_date_list': compare_date_list,
|
||
'query_type': query_type,
|
||
'time_range': time_range,
|
||
'weekdays': weekdays,
|
||
'userid': userid,
|
||
'cross_name': cross_static_info['name'],
|
||
}
|
||
# 必要数据获取
|
||
cross_delay_data_list = db_cross.query_cross_delay_info(crossid, nodeid, date_list, tp_start)
|
||
avg_cross_delay_info = gen_avg_cross_delay_pb(cross_delay_data_list, weekdays)
|
||
if not avg_cross_delay_info:
|
||
return json.dumps(make_common_res(9, '当前所选日期范围内该评测时段无可用数据'))
|
||
comp_cross_delay_data_list = db_cross.query_cross_delay_info(crossid, nodeid, compare_date_list, tp_start)
|
||
avg_comp_cross_delay_info = gen_avg_cross_delay_pb(comp_cross_delay_data_list, weekdays)
|
||
if not avg_comp_cross_delay_info:
|
||
return json.dumps(make_common_res(10, '当前所选对比日期范围内该评测时段无可用数据'))
|
||
cross_inroads = db_tmnet.query_cross_inroads(crossid, nodeid)
|
||
inroad_static_info_dict = {item['roadid']: item for item in cross_inroads}
|
||
overview_res = gen_overview_index(avg_cross_delay_info, inroad_static_info_dict, nodeid, date_list, roads_dir_dict)
|
||
comp_overview_res = gen_overview_index(avg_comp_cross_delay_info, inroad_static_info_dict, nodeid, compare_date_list, roads_dir_dict)
|
||
road_flow_turn_rate = gen_flow_turn_rate_index(avg_cross_delay_info, roads_dir_dict)
|
||
is_peak = check_is_peak_tp(time_range, area_id, nodeid)
|
||
cross_phase, err = QueryCrossRunningPhase(int(nodeid), [crossid], [str(item) for item in date_list], time_range)
|
||
# 整体指标
|
||
final_overview = gen_compare_overview_res(overview_res, comp_overview_res)
|
||
# 路段指标
|
||
road_flow_delay_infos = gen_road_delay_index(avg_cross_delay_info, roads_dir_dict)
|
||
comp_road_flow_delay_infos = gen_road_delay_index(avg_comp_cross_delay_info, roads_dir_dict)
|
||
compared_inroad_delay_infos = parse_comp_inroad_delay_infos(road_flow_delay_infos, comp_road_flow_delay_infos)
|
||
# 装配数据
|
||
title = f"{cross_static_info['name']}路口优化对比报告"
|
||
doc1.title = title
|
||
part1_data = gen_compare_report_part1_data(data_range, compare_date_range, time_range, cross_static_info['name'], final_overview)
|
||
if 'part1' in parts:
|
||
doc1.tpl_paragraph['part1']['visible'] = 1
|
||
doc1.tpl_data['part1_1'].item1 = part1_data['data_range']
|
||
doc1.tpl_data['part1_1'].item2 = part1_data['compare_date_range']
|
||
doc1.tpl_data['part1_1'].item3 = part1_data['cross_name']
|
||
doc1.tpl_data['part1_1'].item4 = part1_data['time_range']
|
||
for item in part1_data['data_list']:
|
||
item_detail = PartDetail()
|
||
item_detail.detail = item
|
||
doc1.tpl_data['part1_1'].table.append(item_detail)
|
||
if cut_images == 1:
|
||
part1_data['cut_images'] = True
|
||
part1_data['cut_image_params'] = cut_image_params
|
||
part1_data['cut_image_params']['image_part'] = 'part1'
|
||
# todo 补充自动截图逻辑
|
||
if 'part2' in parts:
|
||
doc1.tpl_paragraph['part2']['visible'] = 1
|
||
if 'part2_1' in parts:
|
||
doc1.tpl_paragraph['part2']['part2_1'] = 1
|
||
part2_1_data = gen_compare_report_part2_1_data(cross_static_info, cross_ledger_info)
|
||
doc1.tpl_data['part2_1'].item1 = part2_1_data['cross_name']
|
||
doc1.tpl_data['part2_1'].item2 = part2_1_data['division']
|
||
doc1.tpl_data['part2_1'].item3 = part2_1_data['cross_model'] + '的' + part2_1_data['cross_type']
|
||
doc1.tpl_data['part2_1'].item4 = part2_1_data['slc_company']
|
||
if cut_images == 1:
|
||
part2_1_data['cut_images'] = True
|
||
part2_1_data['cut_image_params'] = cut_image_params
|
||
part2_1_data['cut_image_params']['image_part'] = 'part2_1'
|
||
# todo 补充自动截图逻辑
|
||
if 'part2_2' in parts:
|
||
doc1.tpl_paragraph['part2']['part2_2'] = 1
|
||
part2_2_data = gen_compare_report_part2_2_data(road_flow_delay_infos, roads_dir_dict)
|
||
detail1 = PartDetail()
|
||
detail1.detail = part2_2_data['detail']
|
||
doc1.tpl_data['part2_2'].table = [detail1]
|
||
if cut_images == 1:
|
||
part2_2_data['cut_images'] = True
|
||
part2_2_data['cut_image_params'] = cut_image_params
|
||
part2_2_data['cut_image_params']['image_part'] = 'part2_2'
|
||
# todo 补充自动截图逻辑
|
||
if 'part2_3' in parts:
|
||
doc1.tpl_paragraph['part2']['part2_3'] = 1
|
||
part2_3_data = gen_compare_report_part2_3_data(road_flow_turn_rate, avg_cross_delay_info, inroad_static_info_dict)
|
||
detail1 = PartDetail()
|
||
detail1.detail = "根据车辆分流转向比分析，?进口道流量占比最大为?，?进口道流量占比最小为?。路口的主要车流方向为?，？。"
|
||
detail1.detail_value = [part2_3_data['max_src_dir'], part2_3_data['max_flow_rate'], part2_3_data['min_src_dir'], part2_3_data['min_flow_rate'], part2_3_data['main_flow_str'], part2_3_data['main_flow_detail']]
|
||
doc1.tpl_data['part2_3'].table = [detail1]
|
||
if cut_images == 1:
|
||
part2_3_data['cut_images'] = True
|
||
part2_3_data['cut_image_params'] = cut_image_params
|
||
part2_3_data['cut_image_params']['image_part'] = 'part2_3'
|
||
# todo 补充自动截图逻辑
|
||
if 'part3' in parts:
|
||
doc1.tpl_paragraph['part3']['visible'] = 1
|
||
part3_data = gen_compare_report_part3_data(crossid, nodeid, area_id, time_range, tp_start, date_list, avg_cross_delay_info, roads_dir_dict, inroad_static_info_dict, cross_phase, is_peak, cross_ledger_info, weekdays)
|
||
for problem_key in part3_data.keys():
|
||
class_part = PartDetail()
|
||
class_part.item1 = part3_data[problem_key]['class_name']
|
||
for item in part3_data[problem_key]['detail']:
|
||
item_part = PartDetail()
|
||
item_part.item1 = item['item_name']
|
||
child_item_part = PartDetail()
|
||
child_item_part.detail = item['item_detail']
|
||
item_part.table = [child_item_part]
|
||
class_part.table.append(item_part)
|
||
doc1.tpl_data['part3_1'].table.append(class_part)
|
||
if 'part4' in parts:
|
||
doc1.tpl_paragraph['part4']['visible'] = 1
|
||
if 'part4_1' in parts:
|
||
doc1.tpl_paragraph['part4']['part4_1'] = 1
|
||
if 'part4_2' in parts:
|
||
doc1.tpl_paragraph['part4']['part4_2'] = 1
|
||
if 'part5' in parts:
|
||
doc1.tpl_paragraph['part5']['visible'] = 1
|
||
part5_data = gen_compare_report_part5_data(part1_data, compared_inroad_delay_infos)
|
||
detail5_1 = PartDetail()
|
||
detail5_1.item1 = '路口方面'
|
||
for item in part5_data['cross_data']['data_list']:
|
||
item_detail = PartDetail()
|
||
item_detail.detail = item
|
||
detail5_1.table.append(item_detail)
|
||
table_value = [detail5_1]
|
||
for src_dir in part5_data['src_dir_data_dict'].keys():
|
||
detail5_2 = PartDetail()
|
||
detail5_2.item1 = src_dir + '进口方面'
|
||
for item in part5_data['src_dir_data_dict'][src_dir]['data_list']:
|
||
item_detail = PartDetail()
|
||
item_detail.detail = item
|
||
detail5_2.table.append(item_detail)
|
||
table_value.append(detail5_2)
|
||
doc1.tpl_data['part5_1'].item1 = '方案下发后通过平台对路口优化效果进行对比，优化后路口及进口道效果提升较明显的指标如下：'
|
||
doc1.tpl_data['part5_1'].table = table_value
|
||
# 清除临时本地文件
|
||
os.remove(localfile_path)
|
||
file_stream = io.BytesIO()
|
||
doc1.build_template('')
|
||
doc1.doc.save(file_stream)
|
||
file_stream.seek(0)
|
||
cos_path = f'user/cross_doctor/platform_report/{nodeid}/{area_id}/cross_compare_report'
|
||
folder_manager.ensure_folder(cos_path)
|
||
cos_key = f"{cos_path}/{cross_static_info['name']}路口优化对比报告.docx"
|
||
cos_client.put_object(Bucket=g_cos_bucket, Key=cos_key, Body=file_stream)
|
||
download_url = f'{g_cos_root}/{cos_key}'
|
||
res = make_common_res(0, 'ok')
|
||
res['data'] = {
|
||
'crossid': crossid,
|
||
'nodeid': nodeid,
|
||
'area_id': area_id,
|
||
'name': cross_static_info['name'],
|
||
'file_name': file_name,
|
||
'gen_report_date': gen_report_date,
|
||
'date_range': data_range,
|
||
'weekdays': weekdays,
|
||
'compare_date_range': compare_date_range,
|
||
'weekday_str': gen_weekday_str(weekdays),
|
||
'download_url': download_url
|
||
}
|
||
return json.dumps(res, ensure_ascii=False)
|
||
|
||
|
||
def gen_compare_report_part1_data(data_range, compare_date_range, time_range, cross_name, final_overview):
|
||
part1_data = {
|
||
'data_range': data_range,
|
||
'compare_date_range': compare_date_range,
|
||
'cross_name': cross_name,
|
||
'time_range': time_range,
|
||
'data_list': []
|
||
}
|
||
index_dict = {
|
||
'service_level': '路口服务水平',
|
||
'jam_index': '路口拥堵指数',
|
||
'stop_times': '路口停车次数',
|
||
'high_park_percent': '路口多次停车率',
|
||
'speed': '路口平均速度',
|
||
'delay_time': '路口延误时间'
|
||
}
|
||
for key in ('service_level', 'jam_index', 'stop_times', 'high_park_percent', 'speed', 'delay_time'):
|
||
compare_key = 'comp_' + key
|
||
if key == 'service_level':
|
||
service_level_compare_res = compare_service_level(final_overview[key], final_overview[compare_key])
|
||
if service_level_compare_res and service_level_compare_res > 0:
|
||
service_level_compare_str = f"{index_dict[key]}由{final_overview[compare_key]}提升为{final_overview[key]}， 提升{service_level_compare_res}个等级；"
|
||
part1_data['data_list'].append(service_level_compare_str)
|
||
elif key in ('jam_index', 'stop_times', 'high_park_percent', 'delay_time'):
|
||
if key == 'high_park_percent':
|
||
final_overview[key] = int(final_overview[key].replace('%', ''))
|
||
final_overview[compare_key] = int(final_overview[compare_key].replace('%', ''))
|
||
if final_overview[compare_key] and final_overview[key] and final_overview[compare_key] > final_overview[key]:
|
||
rate = round((final_overview[key] - final_overview[compare_key]) / final_overview[compare_key] * 100, 2) if final_overview[compare_key] != '-' and final_overview[compare_key] > 0 else 0
|
||
compare_res_str = f"{index_dict[key]}由{final_overview[compare_key]}下降为{final_overview[key]}， 减少{round(final_overview[key] - final_overview[compare_key], 2)}，优化率为{rate}%；"
|
||
part1_data['data_list'].append(compare_res_str)
|
||
else:
|
||
if final_overview[compare_key] and final_overview[key] and final_overview[compare_key] < final_overview[key]:
|
||
rate = round((final_overview[compare_key] - final_overview[key]) / final_overview[compare_key] * 100, 2) if final_overview[compare_key] != '-' and final_overview[compare_key] > 0 else 0
|
||
compare_res_str = f"{index_dict[key]}由{final_overview[compare_key]}Km/h提升为{final_overview[key]}km/h， 提升{round(final_overview[key] - final_overview[compare_key], 2)}km/h，提升率为{rate}%；"
|
||
part1_data['data_list'].append(compare_res_str)
|
||
return part1_data
|
||
|
||
|
||
def gen_compare_report_part2_1_data(cross_static_info, cross_ledger_info):
|
||
cross_type_dict = db_tmnet.gen_cross_type_dict()
|
||
cross_type = cross_type_dict[int(cross_ledger_info['ledger']['cross_type'])] if cross_ledger_info['ledger']['cross_type'] and int(cross_ledger_info['ledger']['cross_type']) in cross_type_dict.keys() else ''
|
||
part2_1_data = {
|
||
'cross_name': cross_static_info['name'],
|
||
'division': cross_ledger_info['ledger']['division'],
|
||
'cross_model': cross_static_info['cross_model'],
|
||
'cross_type': cross_type,
|
||
'slc_company': cross_static_info['slc_company']
|
||
}
|
||
return part2_1_data
|
||
|
||
|
||
def gen_compare_report_part2_2_data(road_flow_delay_infos, roads_dir_dict):
|
||
road_dir = {v['in']: k for k, v in roads_dir_dict.items()}
|
||
part2_2_data = {
|
||
'total_num': 0,
|
||
'detail': ''
|
||
}
|
||
err_dir_list = []
|
||
for roadid in road_flow_delay_infos.keys():
|
||
if road_flow_delay_infos[roadid]['service_level'] in ('E', 'F'):
|
||
item_str = f"{srcDir_toStr(road_dir[roadid])}进口的延误时间为{road_flow_delay_infos[roadid]['delay_time']}s"
|
||
part2_2_data['total_num'] += 1
|
||
err_dir_list.append(item_str)
|
||
total_str = ''
|
||
if err_dir_list:
|
||
total_str = '，'.join(err_dir_list)
|
||
total_str += '。'
|
||
part2_2_data['detail'] = total_str
|
||
return part2_2_data
|
||
|
||
|
||
def gen_compare_report_part2_3_data(road_flow_turn_rate, avg_cross_delay_info, inroad_static_info_dict):
|
||
flow_rate_dict = {
|
||
'max': {
|
||
'roadid': '',
|
||
'flow_rate': 0,
|
||
'src_dir': ''
|
||
},
|
||
'min': {
|
||
'roadid': '',
|
||
'flow_rate': 101,
|
||
'src_dir': ''
|
||
}
|
||
}
|
||
main_flow_src_list, main_flow_str_list = [], []
|
||
car_num = avg_cross_delay_info.delay_info.car_num
|
||
for inroad_info in avg_cross_delay_info.inroad_delay_infos:
|
||
inroadid = inroad_info.inroadid
|
||
if inroadid not in inroad_static_info_dict.keys():
|
||
continue
|
||
inroad_car_num = inroad_info.delay_info.car_num
|
||
src_dir = inroad_static_info_dict[inroadid]['src_direct']
|
||
if inroad_car_num / car_num > 0.25:
|
||
check_res = found_max_car_num_flow(inroad_car_num, inroad_info)
|
||
if check_res != -1:
|
||
main_flow_src_list.append(inroadid + ' ' + src_dir + ' ' +str(check_res))
|
||
main_flow_str_list.append(dir_str_dict[src_dir] + int_turn_type2str[check_res])
|
||
for roadid in road_flow_turn_rate.keys():
|
||
if road_flow_turn_rate[roadid]['in_flow_rate'] > flow_rate_dict['max']['flow_rate']:
|
||
flow_rate_dict['max']['roadid'] = roadid
|
||
flow_rate_dict['max']['flow_rate'] = road_flow_turn_rate[roadid]['in_flow_rate']
|
||
flow_rate_dict['max']['src_dir'] = dir_str_dict[road_flow_turn_rate[roadid]['src_dir']]
|
||
if road_flow_turn_rate[roadid]['in_flow_rate'] < flow_rate_dict['min']['flow_rate']:
|
||
flow_rate_dict['min']['roadid'] = roadid
|
||
flow_rate_dict['min']['flow_rate'] = road_flow_turn_rate[roadid]['in_flow_rate']
|
||
flow_rate_dict['min']['src_dir'] = dir_str_dict[road_flow_turn_rate[roadid]['src_dir']]
|
||
main_flow_str = '、'.join(main_flow_str_list)
|
||
main_flow_detail = []
|
||
for item in main_flow_src_list:
|
||
roadid = item.split(' ')[0]
|
||
src_dir = item.split(' ')[1]
|
||
turn_type = int(item.split(' ')[2])
|
||
if turn_type == 0:
|
||
turn_type_str = 's_rate'
|
||
elif turn_type == 1:
|
||
turn_type_str = 'l_rate'
|
||
else:
|
||
turn_type_str = 'r_rate'
|
||
main_flow_detail.append(f"{dir_str_dict[src_dir]}进口道以{int_turn_type2str[turn_type]}车流为主，占比为{road_flow_turn_rate[roadid][turn_type_str]}%")
|
||
part2_3_data = {
|
||
'max_src_dir': flow_rate_dict['max']['src_dir'],
|
||
'max_flow_rate': str(flow_rate_dict['max']['flow_rate']) + '%',
|
||
'min_src_dir': flow_rate_dict['min']['src_dir'],
|
||
'min_flow_rate': str(flow_rate_dict['min']['flow_rate']) + '%',
|
||
'main_flow_str': main_flow_str,
|
||
'main_flow_detail': '；'.join(main_flow_detail)
|
||
}
|
||
return part2_3_data
|
||
|
||
|
||
def gen_compare_report_part3_data(crossid, nodeid, area_id, time_range, tp_start, date_list, avg_cross_delay_info, roads_dir_dict, inroad_static_info_dict, cross_phase, is_peak, cross_ledger_info, weekdays):
|
||
cross_problems = gen_cross_problems(crossid, nodeid, area_id, time_range, str(tp_start).replace('h', '').replace('t', ''), date_list, avg_cross_delay_info, roads_dir_dict, inroad_static_info_dict, cross_phase, is_peak, cross_ledger_info, weekdays)
|
||
problem_key_dict = {
|
||
'operating_efficiency_problems': '运行效果',
|
||
'balanced_control_problems': '均衡调控',
|
||
'phase_problems': '配时方案',
|
||
'cross_channelized_problems': '路口渠化'
|
||
}
|
||
problem_detail_dict = {}
|
||
for problem_key in cross_problems.keys():
|
||
if problem_key != 'phase_problems':
|
||
if 'total_num' in cross_problems[problem_key].keys() and cross_problems[problem_key]['total_num'] > 0:
|
||
if problem_key not in problem_detail_dict.keys():
|
||
problem_detail_dict[problem_key] = {
|
||
'class_name': problem_key_dict[problem_key],
|
||
'detail': []
|
||
}
|
||
problem_infos = cross_problems[problem_key]['values']
|
||
for problem_item in problem_infos:
|
||
problem_detail_dict[problem_key]['detail'].append({
|
||
'item_name': problem_item['problem_infos']['item_name'],
|
||
'item_detail': problem_item['problem_infos']['item_detail_str']
|
||
})
|
||
else:
|
||
if 'total_num' in cross_problems[problem_key].keys() and cross_problems[problem_key]['total_num'] > 0:
|
||
if problem_key not in problem_detail_dict.keys():
|
||
problem_detail_dict[problem_key] = {
|
||
'class_name': problem_key_dict[problem_key],
|
||
'detail': []
|
||
}
|
||
for phase_item in cross_problems[problem_key]['values']:
|
||
item_list = []
|
||
if len(phase_item['detail']) > 0:
|
||
if phase_item['item'] == '路口方案异常':
|
||
for item in phase_item['detail']:
|
||
item_detail = item['child_detail']
|
||
item_str = f"{item_detail[0]['text']}{item_detail[1]['level_color']}{item_detail[1]['text']}{item_detail[1]['final_state']}"
|
||
item_list.append(item_str)
|
||
elif phase_item['item'] == '调度计划单一':
|
||
item_detail = phase_item['detail'][0]['child_detail']
|
||
item_str = f"{item_detail[0]['text']}{item_detail[1]['scheduleid']}{item_detail[2]['text']}"
|
||
item_list.append(item_str)
|
||
elif phase_item['item'] == '时段划分单一':
|
||
for item in phase_item['detail']:
|
||
item_detail = item['child_detail']
|
||
item_str = f"{item_detail[0]['text']}{item_detail[1]['scheduleid']}{item_detail[2]['text']}"
|
||
item_list.append(item_str)
|
||
elif phase_item['item'] in ('方案运行时段过短', '方案周期过大', '方案周期过小', '黄灯时间过短', '全红时间不合理', '行人过街时间不足'):
|
||
for item in phase_item['detail']:
|
||
item_detail = item['child_detail']
|
||
item_str = f"{item_detail[0]['plan_name']}{item_detail[1]['text']}"
|
||
item_list.append(item_str)
|
||
problem_detail_dict[problem_key]['detail'].append({
|
||
'item_name': phase_item['item'],
|
||
'item_detail': '；'.join(item_list)
|
||
})
|
||
return problem_detail_dict
|
||
|
||
|
||
def gen_compare_report_part5_data(part1_data, compared_inroad_delay_infos):
|
||
part5_data = {
|
||
'cross_data': part1_data,
|
||
'src_dir_data_dict': {}
|
||
}
|
||
index_dict = {
|
||
'service_level': '路口服务水平',
|
||
'jam_index': '路口拥堵指数',
|
||
'stop_times': '路口停车次数',
|
||
'high_park_percent': '路口多次停车率',
|
||
'speed': '路口平均速度',
|
||
'delay_time': '路口延误时间'
|
||
}
|
||
for roadid in compared_inroad_delay_infos.keys():
|
||
src_dir = compared_inroad_delay_infos[roadid]['src_dir']
|
||
src_dir_data = []
|
||
compare_data = compared_inroad_delay_infos[roadid]['dir_data']['comp_data']
|
||
item_data = compared_inroad_delay_infos[roadid]['dir_data']['item_data']
|
||
diff_data = compared_inroad_delay_infos[roadid]['dir_data']['diff_data']
|
||
for key in ('service_level', 'stop_times', 'high_park_percent', 'speed', 'delay_time'):
|
||
if key == 'service_level':
|
||
service_level_compare_res = compare_service_level(item_data[key], compare_data[key])
|
||
if service_level_compare_res and service_level_compare_res > 0:
|
||
service_level_compare_str = f"{index_dict[key]}由{compare_data[key]}提升为{item_data[key]}， 提升{service_level_compare_res}个等级；"
|
||
src_dir_data.append(service_level_compare_str)
|
||
elif key in ('stop_times', 'high_park_percent', 'delay_time'):
|
||
if compare_data[key] and item_data[key] and compare_data[key] > item_data[key]:
|
||
if key == 'high_park_percent':
|
||
item_data[key] = int(item_data[key].replace('%', ''))
|
||
compare_data[key] = int(compare_data[key].replace('%', ''))
|
||
rate = round((item_data[key] - compare_data[key]) / compare_data[key] * 100, 2) if compare_data[key] != '-' and compare_data[key] > 0 else 0
|
||
compare_res_str = f"{index_dict[key]}由{compare_data[key]}下降为{item_data[key]}， 减少{diff_data[key]}，优化率为{rate}%；"
|
||
src_dir_data.append(compare_res_str)
|
||
else:
|
||
if compare_data[key] and item_data[key] and compare_data[key] < item_data[key]:
|
||
rate = round((compare_data[key] - item_data[key]) / compare_data[key] * 100, 2) if compare_data[key] != '-' and compare_data[key] > 0 else 0
|
||
compare_res_str = f"{index_dict[key]}由{compare_data[key]}Km/h提升为{item_data[key]}km/h， 提升{diff_data[key]}km/h，提升率为{rate}%；"
|
||
src_dir_data.append(compare_res_str)
|
||
if len(src_dir_data) > 0:
|
||
part5_data['src_dir_data_dict'][srcDir_toStr(src_dir)] = {
|
||
'src_dir': src_dir,
|
||
'src_dir_str': srcDir_toStr(src_dir),
|
||
'data_list': src_dir_data
|
||
}
|
||
return part5_data
|
||
|
||
|
||
def gen_weekday_str(weekday):
|
||
if weekday == '1,2,3,4,5':
|
||
weekday_str = '工作日'
|
||
elif weekday == '6,7':
|
||
weekday_str = '节假日'
|
||
elif weekday == '1,2,3,4,5,6,7':
|
||
weekday_str = '全周'
|
||
else:
|
||
weekday_list = weekday.split(',')
|
||
weekday_str = weekday2Str[int(weekday_list[0])]
|
||
for i in range(1, len(weekday_list)):
|
||
weekday_str += '、' + weekday2Str[int(weekday_list[i])]
|
||
return weekday_str
|
||
|
||
|
||
def collect_report(params):
|
||
nodeid = check_param(params, 'nodeid')
|
||
if not nodeid:
|
||
return json.dumps(make_common_res(2, '缺少nodeid， 请刷新后重试'))
|
||
area_id = check_param(params, 'area_id')
|
||
if not area_id:
|
||
return json.dumps(make_common_res(3, '缺少area_id， 请刷新后重试'))
|
||
userid = check_param(params, 'userid')
|
||
if not userid:
|
||
return json.dumps(make_common_res(4, '缺少userid， 请刷新后重试'))
|
||
area_list = db_user.query_areaid_list(userid)
|
||
if not area_list or len(area_list) < 1:
|
||
return json.dumps(make_common_res(5, '用户信息异常'))
|
||
area_list = map(int, area_list)
|
||
if not str(area_id).lstrip('-').isdigit() or int(area_id) not in area_list:
|
||
return json.dumps(make_common_res(5, '辖区id异常，请检查后重试'))
|
||
crossid = check_param(params, 'crossid')
|
||
if not crossid:
|
||
return json.dumps(make_common_res(6, '缺少路口id，请刷新后重试'))
|
||
cross_name = check_param(params, 'cross_name')
|
||
if not cross_name:
|
||
return json.dumps(make_common_res(7, '缺少路口名称，请刷新后重试'))
|
||
download_url = check_param(params, 'download_url')
|
||
if not download_url:
|
||
return json.dumps(make_common_res(8, '缺少下载地址，请刷新后重试'))
|
||
report_type = check_param(params, 'report_type')
|
||
if not report_type or report_type not in(0, 1, 2, 3):
|
||
# 0 对比报告 ，其余为预留
|
||
return json.dumps(make_common_res(7, '缺少报告类型，请刷新后重试'))
|
||
report_type = int(report_type)
|
||
ret = None
|
||
if report_type == 0:
|
||
date_range = check_param(params, 'date_range')
|
||
if not date_range:
|
||
return json.dumps(make_common_res(8, '缺少时间范围， 请刷新后重试'))
|
||
compare_date_range = check_param(params, 'compare_date_range')
|
||
if not compare_date_range:
|
||
return json.dumps(make_common_res(9, '缺少对比时间范围， 请刷新后重试'))
|
||
time_range = check_param(params, 'time_range')
|
||
if not time_range:
|
||
return json.dumps(make_common_res(10, '缺少时间范围， 请刷新后重试'))
|
||
weekday_str = check_param(params, 'weekday_str')
|
||
if not weekday_str:
|
||
return json.dumps(make_common_res(11, '缺少周信息， 请刷新后重试'))
|
||
gen_report_date = check_param(params, 'gen_report_date')
|
||
if not gen_report_date:
|
||
return json.dumps(make_common_res(12, '缺少生成时间， 请刷新后重试'))
|
||
ret = db_tmnet.save_collect_report(report_type, params)
|
||
if ret:
|
||
return json.dumps(make_common_res(0, 'ok'))
|
||
|
||
|
||
def query_collect_report_record(params):
|
||
nodeid = check_param(params, 'nodeid')
|
||
if not nodeid:
|
||
return json.dumps(make_common_res(2, '缺少nodeid， 请刷新后重试'))
|
||
area_id = check_param(params, 'area_id')
|
||
if not area_id:
|
||
return json.dumps(make_common_res(3, '缺少area_id， 请刷新后重试'))
|
||
userid = check_param(params, 'userid')
|
||
if not userid:
|
||
return json.dumps(make_common_res(4, '缺少userid， 请刷新后重试'))
|
||
area_list = db_user.query_areaid_list(userid)
|
||
if not area_list or len(area_list) < 1:
|
||
return json.dumps(make_common_res(5, '用户信息异常'))
|
||
area_list = map(int, area_list)
|
||
if not str(area_id).lstrip('-').isdigit() or int(area_id) not in area_list:
|
||
return json.dumps(make_common_res(5, '辖区id异常，请检查后重试'))
|
||
report_type = check_param(params, 'report_type')
|
||
if not report_type or report_type not in(0, 1, 2, 3):
|
||
# 0 对比报告 ，其余为预留
|
||
return json.dumps(make_common_res(7, '缺少报告类型，请刷新后重试'))
|
||
report_type = int(report_type)
|
||
records = db_tmnet.query_collect_report_records(nodeid, area_id, report_type)
|
||
res = make_common_res(0, 'ok')
|
||
res['data'] = records
|
||
return json.dumps(res, ensure_ascii=False)
|