# -*- coding: utf-8 -*- # @Author: Owl # @Date: 2025/11/27 11:17 # @Description: 路口优化对比页面相关公共函数 from pypinyin import lazy_pinyin from app.eva_common import * def gen_compare_overview_res(eva_overview, comp_eva_overview): # comp_index 为优化前指标 main_flows_num = len(eva_overview['main_flow_src_desc'].split('、')) comp_main_flows_num = len(comp_eva_overview['main_flow_src_desc'].split('、')) high_stop_turns_num = sum([len(item.split('/')) for item in eva_overview['high_stop_turn_ratio_desc'].split('、')]) comp_high_stop_turns_num = sum([len(item.split('/')) for item in comp_eva_overview['high_stop_turn_ratio_desc'].split('、')]) overview = { 'stop_times': eva_overview['stop_times'], 'comp_stop_times': comp_eva_overview['stop_times'], 'stop_times_color': calc_index_color(eva_overview['stop_times'], comp_eva_overview['stop_times'], 'stop_times'), 'jam_index': eva_overview['jam_index'], 'comp_jam_index': comp_eva_overview['jam_index'], 'jam_index_color': calc_index_color(eva_overview['jam_index'], comp_eva_overview['jam_index'], 'jam_index'), 'high_park_percent': eva_overview['high_park_percent'].replace('%', ''), 'comp_high_park_percent': comp_eva_overview['high_park_percent'].replace('%', ''), 'high_park_percent_color': calc_index_color(float(eva_overview['high_park_percent'].replace('%', '')), float(comp_eva_overview['high_park_percent'].replace('%', '')), 'high_park_percent'), 'imbalance_index': eva_overview['imbalance_index'], 'comp_imbalance_index': comp_eva_overview['imbalance_index'], 'imbalance_index_color': calc_index_color(eva_overview['imbalance_index'], comp_eva_overview['imbalance_index'], 'imbalance_index'), 'speed': eva_overview['speed'], 'comp_speed': comp_eva_overview['speed'], 'speed_color': calc_index_color(eva_overview['speed'], comp_eva_overview['speed'], 'speed'), 'move_speed': eva_overview['move_speed'], 'comp_move_speed': comp_eva_overview['move_speed'], 'move_speed_color': calc_index_color(eva_overview['move_speed'], comp_eva_overview['move_speed'], 'move_speed'), 'delay_time': eva_overview['delay_time'], 'comp_delay_time': comp_eva_overview['delay_time'], 'delay_time_color': calc_index_color(eva_overview['delay_time'], comp_eva_overview['delay_time'], 'delay_time'), 'park_time': eva_overview['park_time'], 'comp_park_time': comp_eva_overview['park_time'], 'park_time_color': calc_index_color(eva_overview['park_time'], comp_eva_overview['park_time'], 'park_time'), 'service_level': eva_overview['service_level'], 'comp_service_level': comp_eva_overview['service_level'], 'main_flow_nums': main_flows_num, 'comp_main_flow_nums': comp_main_flows_num, 'high_stop_turns_num': high_stop_turns_num, 'comp_high_stop_turns_num': comp_high_stop_turns_num } return overview def calc_index_color(index, comp_index, key): color = 0 if key not in ('speed', 'move_speed'): rate = 0 if (comp_index in (0, '-') or index == '-') else round((index - comp_index) / comp_index * 100, 2) if rate > 20: color = 1 elif rate < -20: color = 2 else: rate = 0 if (comp_index in (0, '-') or index == '-') else round((index - comp_index) / comp_index * 100, 2) if rate < -20: color = 1 elif rate > 20: color = 2 return color def parse_comp_inroad_delay_infos(inroad_delay_infos, comp_inroad_delay_infos): inroad_data_keys = inroad_delay_infos.keys() comp_inroad_data_keys = comp_inroad_delay_infos.keys() all_keys = inroad_data_keys | comp_inroad_data_keys res = {} for key in all_keys: comp_data = { 'item': '优化前', 'service_level': comp_inroad_delay_infos[key]['service_level'] if key in comp_inroad_data_keys else '-', 'stop_times': comp_inroad_delay_infos[key]['stop_times'] if key in comp_inroad_data_keys else '-', 'high_park_percent': str(comp_inroad_delay_infos[key]['high_park_percent']) + '%' if key in comp_inroad_data_keys else '-', 'imbalance_index': comp_inroad_delay_infos[key]['imbalance_index'] if key in comp_inroad_data_keys else '-', 'park_time': comp_inroad_delay_infos[key]['park_time'] if key in comp_inroad_data_keys else '-', 'delay_time': comp_inroad_delay_infos[key]['delay_time'] if key in comp_inroad_data_keys else '-', 'speed': comp_inroad_delay_infos[key]['speed'] if key in comp_inroad_data_keys else '-', 'move_speed': comp_inroad_delay_infos[key]['move_speed'] if key in comp_inroad_data_keys else '-' } item_data = { 'item': '优化后', 'service_level': inroad_delay_infos[key]['service_level'] if key in inroad_data_keys else '-', 'stop_times': inroad_delay_infos[key]['stop_times'] if key in inroad_data_keys else '-', 'high_park_percent': str(inroad_delay_infos[key]['high_park_percent']) + '%' if key in inroad_data_keys else '-', 'imbalance_index': inroad_delay_infos[key]['imbalance_index'] if key in inroad_data_keys else '-', 'park_time': inroad_delay_infos[key]['park_time'] if key in inroad_data_keys else '-', 'delay_time': inroad_delay_infos[key]['delay_time'] if key in inroad_data_keys else '-', 'speed': inroad_delay_infos[key]['speed'] if key in inroad_data_keys else '-', 'move_speed': inroad_delay_infos[key]['move_speed'] if key in inroad_data_keys else '-' } diff_data = { 'item': '变化量', 'service_level': '-' if item_data['service_level'] == '-' or comp_data['service_level'] == '-' else compare_level(comp_data['service_level'], item_data['service_level']), 'stop_times': '-' if item_data['stop_times'] == '-' or comp_data['stop_times'] == '-' else round(item_data['stop_times'] - comp_data['stop_times'], 2), 'high_park_percent': '-' if item_data['high_park_percent'] == '-' or comp_data['high_park_percent'] == '-' else str(round(float(item_data['high_park_percent'].replace('%', '')) - float(comp_data['high_park_percent'].replace('%', '')), 2)) + '%', 'imbalance_index': '-' if item_data['imbalance_index'] == '-' or comp_data['imbalance_index'] == '-' else round(item_data['imbalance_index'] - comp_data['imbalance_index'], 2), 'park_time': '-' if item_data['park_time'] == '-' or comp_data['park_time'] == '-' else item_data['park_time'] - comp_data['park_time'], 'delay_time': '-' if item_data['delay_time'] == '-' or comp_data['delay_time'] == '-' else item_data['delay_time'] - comp_data['delay_time'], 'speed': '-' if item_data['speed'] == '-' or comp_data['speed'] == '-' else round(item_data['speed'] - comp_data['speed'], 2), 'move_speed': '-' if item_data['move_speed'] == '-' or comp_data['move_speed'] == '-' else round(item_data['move_speed'] - comp_data['move_speed'], 2), 'stop_times_color': '-' if item_data['stop_times'] == '-' or comp_data['stop_times'] == '-' else calc_index_color(item_data['stop_times'], comp_data['stop_times'], 'stop_times'), 'high_park_percent_color': '-' if item_data['high_park_percent'] == '-' or comp_data['high_park_percent'] == '-' else calc_index_color(float(item_data['high_park_percent'].replace('%', '')), float(comp_data['high_park_percent'].replace('%', '')), 'high_park_percent'), 'imbalance_index_color': '-' if item_data['imbalance_index'] == '-' or comp_data['imbalance_index'] == '-' else calc_index_color(item_data['imbalance_index'], comp_data['imbalance_index'], 'imbalance_index'), 'park_time_color': '-' if item_data['park_time'] == '-' or comp_data['park_time'] == '-' else calc_index_color(item_data['park_time'], comp_data['park_time'], 'park_time'), 'delay_time_color': '-' if item_data['delay_time'] == '-' or comp_data['delay_time'] == '-' else calc_index_color(item_data['delay_time'], comp_data['delay_time'], 'delay_time'), 'speed_color': '-' if item_data['speed'] == '-' or comp_data['speed'] == '-' else calc_index_color(item_data['speed'], comp_data['speed'], 'speed'), 'move_speed_color': '-' if item_data['move_speed'] == '-' or comp_data['move_speed'] == '-' else calc_index_color(item_data['move_speed'], comp_data['move_speed'], 'move_speed') } turn_type = inroad_delay_infos[key]['flow_delays'].keys() if key in inroad_delay_infos.keys() else set() comp_turn_type = comp_inroad_delay_infos[key]['flow_delays'].keys() if key in comp_inroad_delay_infos.keys() else set() all_turn_type = turn_type | comp_turn_type flow_delay_datas = {} for item_turn in all_turn_type: flow_comp_data = { 'item': '优化前', 'service_level': comp_inroad_delay_infos[key]['flow_delays'][item_turn]['service_level'] if key in comp_inroad_delay_infos.keys() and item_turn in comp_turn_type else '-', 'stop_times': comp_inroad_delay_infos[key]['flow_delays'][item_turn]['stop_times'] if key in comp_inroad_delay_infos.keys() and item_turn in comp_turn_type else '-', 'high_park_percent': str(comp_inroad_delay_infos[key]['flow_delays'][item_turn]['high_park_percent']) + '%' if key in comp_inroad_delay_infos.keys() and item_turn in comp_turn_type else '-', 'park_time': comp_inroad_delay_infos[key]['flow_delays'][item_turn]['park_time'] if key in comp_inroad_delay_infos.keys() and item_turn in comp_turn_type else '-', 'delay_time': comp_inroad_delay_infos[key]['flow_delays'][item_turn]['delay_time'] if key in comp_inroad_delay_infos.keys() and item_turn in comp_turn_type else '-', 'speed': comp_inroad_delay_infos[key]['flow_delays'][item_turn]['speed'] if key in comp_inroad_delay_infos.keys() and item_turn in comp_turn_type else '-', 'move_speed': comp_inroad_delay_infos[key]['flow_delays'][item_turn]['move_speed'] if key in comp_inroad_delay_infos.keys() and item_turn in comp_turn_type else '-' } flow_item_data = { 'item': '优化后', 'service_level': inroad_delay_infos[key]['flow_delays'][item_turn]['service_level'] if key in inroad_delay_infos.keys() and item_turn in turn_type else '-', 'stop_times': inroad_delay_infos[key]['flow_delays'][item_turn]['stop_times'] if key in inroad_delay_infos.keys() and item_turn in turn_type else '-', 'high_park_percent': str(inroad_delay_infos[key]['flow_delays'][item_turn]['high_park_percent']) + '%' if key in inroad_delay_infos.keys() and item_turn in turn_type else '-', 'park_time': inroad_delay_infos[key]['flow_delays'][item_turn]['park_time'] if key in inroad_delay_infos.keys() and item_turn in turn_type else '-', 'delay_time': inroad_delay_infos[key]['flow_delays'][item_turn]['delay_time'] if key in inroad_delay_infos.keys() and item_turn in turn_type else '-', 'speed': inroad_delay_infos[key]['flow_delays'][item_turn]['speed'] if key in inroad_delay_infos.keys() and item_turn in turn_type else '-', 'move_speed': inroad_delay_infos[key]['flow_delays'][item_turn]['move_speed'] if key in inroad_delay_infos.keys() and item_turn in turn_type else '-' } flow_diff_data = { 'item': '变化量', 'service_level': '-' if flow_comp_data['service_level'] == '-' or flow_item_data['service_level'] == '-' else compare_level(flow_comp_data['service_level'], flow_item_data['service_level']), 'stop_times': '-' if flow_comp_data['stop_times'] == '-' or flow_item_data['stop_times'] == '-' else round(flow_comp_data['stop_times'] - flow_item_data['stop_times'], 2), 'high_park_percent': '-' if flow_comp_data['high_park_percent'] == '-' or flow_item_data['high_park_percent'] == '-' else str(round(float(flow_comp_data['high_park_percent'].replace('%', '')) - float(flow_item_data['high_park_percent'].replace('%', '')), 2)) + '%', 'park_time': '-' if flow_comp_data['park_time'] == '-' or flow_item_data['park_time'] == '-' else round(flow_comp_data['park_time'] - flow_item_data['park_time'], 2), 'delay_time': '-' if flow_comp_data['delay_time'] == '-' or flow_item_data['delay_time'] == '-' else round(flow_comp_data['delay_time'] - flow_item_data['delay_time'], 2), 'speed': '-' if flow_comp_data['speed'] == '-' or flow_item_data['speed'] == '-' else round(flow_comp_data['speed'] - flow_item_data['speed'], 2), 'move_speed': '-' if flow_comp_data['move_speed'] == '-' or flow_item_data['move_speed'] == '-' else round(flow_comp_data['move_speed'] - flow_item_data['move_speed'], 2), 'stop_times_color': '-' if flow_item_data['stop_times'] == '-' or flow_comp_data['stop_times'] == '-' else calc_index_color(flow_item_data['stop_times'], flow_comp_data['stop_times'], 'stop_times'), 'high_park_percent_color': '-' if flow_item_data['high_park_percent'] == '-' or flow_comp_data['high_park_percent'] == '-' else calc_index_color(float(flow_item_data['high_park_percent'].replace('%', '')), float(comp_data['high_park_percent'].replace('%', '')), 'high_park_percent'), 'park_time_color': '-' if flow_item_data['park_time'] == '-' or flow_comp_data['park_time'] == '-' else calc_index_color(flow_item_data['park_time'], flow_comp_data['park_time'], 'park_time'), 'delay_time_color': '-' if flow_item_data['delay_time'] == '-' or flow_comp_data['delay_time'] == '-' else calc_index_color(flow_item_data['delay_time'], flow_comp_data['delay_time'], 'delay_time'), 'speed_color': '-' if flow_item_data['speed'] == '-' or flow_comp_data['speed'] == '-' else calc_index_color(flow_item_data['speed'], flow_comp_data['speed'], 'speed'), 'move_speed_color': '-' if flow_item_data['move_speed'] == '-' or flow_comp_data['move_speed'] == '-' else calc_index_color(flow_item_data['move_speed'], flow_comp_data['move_speed'], 'move_speed') } flow_delay_datas[item_turn] = { 'item_data': flow_item_data, 'comp_data': flow_comp_data, 'diff_data': flow_diff_data } res[key] = { 'src_dir': inroad_delay_infos[key]['src_dir'] if key in inroad_delay_infos.keys() and 'src_dir' in inroad_delay_infos[key] else comp_inroad_delay_infos[key]['src_dir'], 'dir_data': { 'item_data': item_data, 'comp_data': comp_data, 'diff_data': diff_data, 'flow_delay_datas': flow_delay_datas } } tmp_dict = {v['src_dir']: k for k, v in res.items()} sorted_tmp_dict = sort_dict_by_clockwise(tmp_dict) sorted_key = list(sorted_tmp_dict.values()) final_res = {k: res[k] for k in sorted_key} return final_res def compare_level(new, old): if new == '-' or old == '-': return 0 scores = {'A': 0, 'B': 1, 'C': 2, 'D': 3, 'E': 4, 'F': 5} new_score, old_score = scores.get(new, 99), scores.get(old, 99) if new_score == old_score: return 0 return 2 if new_score < old_score else 1 def find_job_info(key, value_list): result = [] # 遍历名字列表 for value in value_list: # 检查是否完全匹配 if key in value['cross_name']: result.append(value) if key == value['cross_name']: result = [value] break # 检查是否是首字母缩写匹配 elif len(key) >= 2 and len(value['cross_name']) >= 2: # 将名字转换为拼音并提取首字母 first_pinyin = lazy_pinyin(value['cross_name'][0])[0] second_pinyin = lazy_pinyin(value['cross_name'][1])[0] if key[0].lower() == first_pinyin[0].lower() and key[1].lower() == second_pinyin[0].lower(): result.append(value) # # 检查是否包含关键字 # elif key in value['wave_name']: # result.append(value) return result def query_compare_data_export_excel(compared_inroad_delay_infos, cross_name,compare_date_list,date_list,time_range): wb = Workbook() sheet1 = wb.active sheet1.title = "调度计划表" # 给第一个 sheet 命名 table_head = [ {"range": "A1:A2", "value": "路口名称"}, {"range": "B1:B2", "value": "进口道"}, {"range": "C1:C2", "value": "对比"}, {"range": "D1:D2", "value": "服务水平"}, {"range": "E1:F1", "value": "转向服务水平"}, {"range": "E2", "value": "左转", "width": 10}, {"range": "F2", "value": "直行", "width": 10}, {"range": "G1:G2", "value": "停车次数"}, {"range": "H1:I1", "value": "转向停车次数"}, {"range": "H2", "value": "左转"}, {"range": "I2", "value": "直行"}, {"range": "J1:J2", "value": "多次停车率", "width": 20}, {"range": "K1:L1", "value": "转向多次停车率"}, {"range": "K2", "value": "左转", "width": 10}, {"range": "L2", "value": "直行", "width": 10}, # {"range": "M1:M2", "value": "行程时间(s)", "width": 20}, # {"range": "N1:O1", "value": "转向行程时间(s)"}, # {"range": "N2", "value": "左转", "width": 10}, # {"range": "O2", "value": "直行", "width": 10}, {"range": "M1:M2", "value": "延误时间(s)", "width": 20}, {"range": "N1:O1", "value": "转向延误时间(s)"}, {"range": "N2", "value": "左转", "width": 10}, {"range": "O2", "value": "直行", "width": 10}, {"range": "P1:P2", "value": "平均速度(Km/h)", "width": 20}, {"range": "Q1:R1", "value": "转向平均速度(Km/h)"}, {"range": "Q2", "value": "左转", "width": 10}, {"range": "R2", "value": "直行", "width": 10}, {"range": "S1:S2", "value": "不停车速度(Km/h)", "width": 20}, {"range": "T1:U1", "value": "转向不停车速度(KM/h)"}, {"range": "T2", "value": "左转", "width": 15}, {"range": "U2", "value": "直行", "width": 15}, {"range": "V1:V2", "value": "转向失衡系数", "width": 20}, ] for item_head in table_head: sheet1.merge_cells(item_head['range']) cell = item_head['range'].split(":")[0] sheet1[cell] = item_head['value'] sheet1[cell].alignment = Alignment(horizontal='center', vertical='center') if item_head.get('width'): sheet1.column_dimensions[cell[:-1]].width = item_head['width'] in_road_num = len(compared_inroad_delay_infos) sheet1.merge_cells(f"A3:A{in_road_num * 3 + 2}") sheet1["A3"] = cross_name sheet1["A3"].alignment = Alignment(horizontal='center', vertical='center', wrap_text=True) index = 3 for road, road_value in compared_inroad_delay_infos.items(): start_row = index middle_row = index + 1 end_row = index + 2 sheet1.merge_cells(f"B{start_row}:B{end_row}") sheet1[f"B{index}"] = srcDir_toStr(road_value['src_dir']) sheet1[f"B{index}"].alignment = Alignment(horizontal='center', vertical='center') index = end_row + 1 sheet1[f"C{start_row}"] = "优化前" sheet1[f"D{start_row}"] = road_value['dir_data']['comp_data']['service_level'] sheet1[f"E{start_row}"] = road_value['dir_data']['flow_delay_datas'][1]['comp_data']['service_level'] if road_value['dir_data']['flow_delay_datas'].get(1) else '' sheet1[f"F{start_row}"] = road_value['dir_data']['flow_delay_datas'][0]['comp_data']['service_level'] if road_value['dir_data']['flow_delay_datas'].get(0) else '' sheet1[f"G{start_row}"] = road_value['dir_data']['comp_data']['stop_times'] sheet1[f"H{start_row}"] = road_value['dir_data']['flow_delay_datas'][1]['comp_data']['stop_times'] if road_value['dir_data']['flow_delay_datas'].get(1) else '' sheet1[f"I{start_row}"] = road_value['dir_data']['flow_delay_datas'][0]['comp_data']['stop_times'] if road_value['dir_data']['flow_delay_datas'].get(0) else '' sheet1[f"J{start_row}"] = road_value['dir_data']['comp_data']['high_park_percent'] sheet1[f"K{start_row}"] = road_value['dir_data']['flow_delay_datas'][1]['comp_data']['high_park_percent'] if road_value['dir_data']['flow_delay_datas'].get(1) else '' sheet1[f"L{start_row}"] = road_value['dir_data']['flow_delay_datas'][0]['comp_data']['high_park_percent'] if road_value['dir_data']['flow_delay_datas'].get(0) else '' sheet1[f"M{start_row}"] = road_value['dir_data']['comp_data']['delay_time'] sheet1[f"M{start_row}"] = road_value['dir_data']['flow_delay_datas'][1]['comp_data']['delay_time'] if road_value['dir_data']['flow_delay_datas'].get(1) else '' sheet1[f"N{start_row}"] = road_value['dir_data']['flow_delay_datas'][0]['comp_data']['delay_time'] if road_value['dir_data']['flow_delay_datas'].get(0) else '' sheet1[f"P{start_row}"] = road_value['dir_data']['comp_data']['speed'] sheet1[f"Q{start_row}"] = road_value['dir_data']['flow_delay_datas'][1]['comp_data']['speed'] if road_value['dir_data']['flow_delay_datas'].get(1) else '' sheet1[f"R{start_row}"] = road_value['dir_data']['flow_delay_datas'][0]['comp_data']['speed'] if road_value['dir_data']['flow_delay_datas'].get(0) else '' sheet1[f"S{start_row}"] = road_value['dir_data']['comp_data']['move_speed'] sheet1[f"T{start_row}"] = road_value['dir_data']['flow_delay_datas'][1]['comp_data']['move_speed'] if road_value['dir_data']['flow_delay_datas'].get(1) else '' sheet1[f"U{start_row}"] = road_value['dir_data']['flow_delay_datas'][0]['comp_data']['move_speed'] if road_value['dir_data']['flow_delay_datas'].get(0) else '' sheet1[f"V{start_row}"] = road_value['dir_data']['comp_data']['imbalance_index'] sheet1[f"C{middle_row}"] = "优化后" sheet1[f"D{middle_row}"] = road_value['dir_data']['item_data']['service_level'] sheet1[f"E{middle_row}"] = road_value['dir_data']['flow_delay_datas'][1]['item_data']['service_level'] if road_value['dir_data']['flow_delay_datas'].get(1) else '' sheet1[f"F{middle_row}"] = road_value['dir_data']['flow_delay_datas'][0]['item_data']['service_level'] if road_value['dir_data']['flow_delay_datas'].get(0) else '' sheet1[f"G{middle_row}"] = road_value['dir_data']['item_data']['stop_times'] sheet1[f"H{middle_row}"] = road_value['dir_data']['flow_delay_datas'][1]['item_data']['stop_times'] if road_value['dir_data']['flow_delay_datas'].get(1) else '' sheet1[f"I{middle_row}"] = road_value['dir_data']['flow_delay_datas'][0]['item_data']['stop_times'] if road_value['dir_data']['flow_delay_datas'].get(0) else '' sheet1[f"J{middle_row}"] = road_value['dir_data']['item_data']['high_park_percent'] sheet1[f"K{middle_row}"] = road_value['dir_data']['flow_delay_datas'][1]['item_data']['high_park_percent'] if road_value['dir_data']['flow_delay_datas'].get(1) else '' sheet1[f"L{middle_row}"] = road_value['dir_data']['flow_delay_datas'][0]['item_data']['high_park_percent'] if road_value['dir_data']['flow_delay_datas'].get(0) else '' sheet1[f"M{middle_row}"] = road_value['dir_data']['item_data']['delay_time'] sheet1[f"N{middle_row}"] = road_value['dir_data']['flow_delay_datas'][1]['item_data']['delay_time'] if road_value['dir_data']['flow_delay_datas'].get(1) else '' sheet1[f"O{middle_row}"] = road_value['dir_data']['flow_delay_datas'][0]['item_data']['delay_time'] if road_value['dir_data']['flow_delay_datas'].get(0) else '' sheet1[f"P{middle_row}"] = road_value['dir_data']['item_data']['speed'] sheet1[f"Q{middle_row}"] = road_value['dir_data']['flow_delay_datas'][1]['item_data']['speed'] if road_value['dir_data']['flow_delay_datas'].get(1) else '' sheet1[f"R{middle_row}"] = road_value['dir_data']['flow_delay_datas'][0]['item_data']['speed'] if road_value['dir_data']['flow_delay_datas'].get(0) else '' sheet1[f"S{middle_row}"] = road_value['dir_data']['item_data']['move_speed'] sheet1[f"T{middle_row}"] = road_value['dir_data']['flow_delay_datas'][1]['item_data']['move_speed'] if road_value['dir_data']['flow_delay_datas'].get(1) else '' sheet1[f"U{middle_row}"] = road_value['dir_data']['flow_delay_datas'][0]['item_data']['move_speed'] if road_value['dir_data']['flow_delay_datas'].get(0) else '' sheet1[f"V{middle_row}"] = road_value['dir_data']['item_data']['imbalance_index'] sheet1[f"C{end_row}"] = "变化量" sheet1[f"D{end_row}"] = road_value['dir_data']['diff_data']['service_level'] sheet1[f"E{end_row}"] = road_value['dir_data']['flow_delay_datas'][1]['diff_data']['service_level'] if road_value['dir_data']['flow_delay_datas'].get(1) else '' sheet1[f"F{end_row}"] = road_value['dir_data']['flow_delay_datas'][0]['diff_data']['service_level'] if road_value['dir_data']['flow_delay_datas'].get(0) else '' sheet1[f"G{end_row}"] = road_value['dir_data']['diff_data']['stop_times'] sheet1[f"H{end_row}"] = road_value['dir_data']['flow_delay_datas'][1]['diff_data']['stop_times'] if road_value['dir_data']['flow_delay_datas'].get(1) else '' sheet1[f"I{end_row}"] = road_value['dir_data']['flow_delay_datas'][0]['diff_data']['stop_times'] if road_value['dir_data']['flow_delay_datas'].get(0) else '' sheet1[f"J{end_row}"] = road_value['dir_data']['diff_data']['high_park_percent'] sheet1[f"K{end_row}"] = road_value['dir_data']['flow_delay_datas'][1]['diff_data']['high_park_percent'] if road_value['dir_data']['flow_delay_datas'].get(1) else '' sheet1[f"L{end_row}"] = road_value['dir_data']['flow_delay_datas'][0]['diff_data']['high_park_percent'] if road_value['dir_data']['flow_delay_datas'].get(0) else '' sheet1[f"M{end_row}"] = road_value['dir_data']['diff_data']['delay_time'] sheet1[f"N{end_row}"] = road_value['dir_data']['flow_delay_datas'][1]['diff_data']['delay_time'] if road_value['dir_data']['flow_delay_datas'].get(1) else '' sheet1[f"O{end_row}"] = road_value['dir_data']['flow_delay_datas'][0]['diff_data']['delay_time'] if road_value['dir_data']['flow_delay_datas'].get(0) else '' sheet1[f"P{end_row}"] = road_value['dir_data']['diff_data']['speed'] sheet1[f"Q{end_row}"] = road_value['dir_data']['flow_delay_datas'][1]['diff_data']['speed'] if road_value['dir_data']['flow_delay_datas'].get(1) else '' sheet1[f"R{end_row}"] = road_value['dir_data']['flow_delay_datas'][0]['diff_data']['speed'] if road_value['dir_data']['flow_delay_datas'].get(0) else '' sheet1[f"S{end_row}"] = road_value['dir_data']['diff_data']['move_speed'] sheet1[f"T{end_row}"] = road_value['dir_data']['flow_delay_datas'][1]['diff_data']['move_speed'] if road_value['dir_data']['flow_delay_datas'].get(1) else '' sheet1[f"U{end_row}"] = road_value['dir_data']['flow_delay_datas'][0]['diff_data']['move_speed'] if road_value['dir_data']['flow_delay_datas'].get(0) else '' sheet1[f"V{end_row}"] = road_value['dir_data']['diff_data']['imbalance_index'] center_alignment = Alignment(horizontal='center', vertical='center') for row in sheet1.iter_rows(min_row=1, max_row=sheet1.max_row, min_col=1, max_col=sheet1.max_column): for cell in row: if cell.value is not None: cell.alignment = center_alignment file_name = f"【{cross_name}】{compare_date_list[0]}-{compare_date_list[-1]} vs {date_list[0]}-{date_list[-1]} {time_range} 进口道及转向运行指标.xlsx" file_stream = io.BytesIO() wb.save(file_stream) file_stream.seek(0) # 将指针移到文件开头 return send_file( file_stream, mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', as_attachment=True, download_name=file_name) def query_wave_cross_survey_result(userid, crossid): url = f"http://172.21.32.10:6060/api/get_cross_pngs?userid={userid}&cross_id={crossid}" headers = {"Content-Type": "application/json"} response = requests.get(url, headers=headers) if response.status_code != 200 or response.json()['status'] != 0: logging.error(f"{crossid}不存在绿波管家的实景勘察结果，查询失败") return {} wave_cross_survey_result = json.loads(response.text)['data'] return wave_cross_survey_result def get_cross_png(image_list): min_dist = 999999 res = None for png_info in image_list: if 40 < png_info['dist'] < min_dist: min_dist = png_info['dist'] res = png_info['image_url'] return res