From 6b70ef10b81a169be94d84b6c23ee182c90af573 Mon Sep 17 00:00:00 2001 From: wangxu <1318272526@qq.com> Date: Tue, 28 Oct 2025 10:53:40 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8F=90=E4=BA=A4=E6=9C=AA=E7=BC=96=E8=BE=91?= =?UTF-8?q?=E5=AE=8C=E6=88=90=E7=9A=84=E9=97=AE=E9=A2=98=E8=AF=8A=E6=96=AD?= =?UTF-8?q?=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/common_worker.py | 22 ++ app/cross_evaluate_worker.py | 21 +- app/eva_common.py | 472 ++++++++++++++++++++++++++++++----- app/tmnet_db_func.py | 2 +- 4 files changed, 448 insertions(+), 69 deletions(-) diff --git a/app/common_worker.py b/app/common_worker.py index da36cf8..edd8fe3 100644 --- a/app/common_worker.py +++ b/app/common_worker.py @@ -404,3 +404,25 @@ def count_lsr(turn_str): # 返回格式化的结果 return f"{left_count}/{straight_count}/{right_count}" + + +def time_overlap(a: str, b: List[str]) -> bool: + def to_minutes(t: str) -> int: + h, m = map(int, t.split(':')) + return h * 60 + m + + a_start, a_end = map(to_minutes, a.split('-')) + + for interval in b: + b_start, b_end = map(to_minutes, interval.split('-')) + # 检查是否有重叠 + if not (a_end <= b_start or a_start >= b_end): + return True + return False + + +def parse_week_str(s: str): + start, end = [datetime.strptime(d, "%Y%m%d") for d in s.split('-')] + date_list = [(start + timedelta(days=i)).strftime("%Y%m%d") + for i in range((end - start).days + 1)] + return date_list diff --git a/app/cross_evaluate_worker.py b/app/cross_evaluate_worker.py index 624c1fb..c2908b1 100644 --- a/app/cross_evaluate_worker.py +++ b/app/cross_evaluate_worker.py @@ -43,12 +43,18 @@ def query_cross_list(params): "19:00-22:00", "22:00-00:00" ] + peak_tp = [ + "07:00-09:00", + "17:00-19:00" + ] else: tp_info = tp_desc[0]['tp_desc'].split(',') + peak_tp = tp_desc[0]['peak_tp'].split(',') res = make_common_res(0, 'ok') res['data'] = { 'cross_list': cross_list, 'tp_info': tp_info, + 'peak_tp': peak_tp } return json.dumps(res, ensure_ascii=False) @@ -126,6 +132,8 @@ def query_cross_delay_info_controller(params): inroad_static_info_dict = {item['roadid']: item for item in cross_inroads} # 路口静态信息及台账信息 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) # 路口指标数据概览 @@ -188,13 +196,18 @@ def query_cross_problems(params): return json.dumps(make_common_res(9, '当前所选日期范围内该评测时段无可用数据')) cross_inroads = db_tmnet.query_cross_inroads(crossid, nodeid) inroad_static_info_dict = {item['roadid']: item for item in cross_inroads} + # 验证是否为高峰时段 + is_peak = check_is_peak_tp(time_range, area_id, nodeid) # 路口静态信息及台账信息 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_cross_problems(avg_cross_delay_info, roads_dir_dict, inroad_static_info_dict) - # todo 与子鉴确认逻辑 - cross_phase, err = QueryCrossRunningPhase(nodeid, [crossid]) + 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) # 指标变化趋势接口 @@ -242,6 +255,8 @@ def query_cross_index_trend_controller(params): # 查询台账信息 获取路网渠化关系 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(9, '查询路口信息失败')) 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) diff --git a/app/eva_common.py b/app/eva_common.py index 87c06e4..26556f0 100644 --- a/app/eva_common.py +++ b/app/eva_common.py @@ -330,7 +330,7 @@ def query_cross_ledger_info(crossid, nodeid, area_id, userid): cross_ledger_info = requests.get(ledger_url, headers=headers) if cross_ledger_info.status_code != 200 or cross_ledger_info.json()['status'] != 0: logging.error(f"查询路口台账信息失败,crossid:{crossid},nodeid:{nodeid},area_id:{area_id},userid:{userid}") - return None, None + return None cross_ledger_info_dict = json.loads(cross_ledger_info.text) return cross_ledger_info_dict @@ -570,10 +570,12 @@ def calc_service_level(delay_time): return service_level -def calc_tide_index(crossid, nodeid, date_list, roads_dir_dict): +def calc_tide_index(crossid, nodeid, date_list, roads_dir_dict, date_type=None): tide_index_list = [] am_tp_start, am_tp_end = 't700', '900' pm_tp_start, pm_tp_end = 't1700', '1900' + if date_type == 'week': + date_list = parse_week_str(date_list[0]) 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) date_am_delay_info_dict = {item['day']: item['data'] for item in am_cross_delay_data_list} @@ -679,7 +681,7 @@ def parse_single_cross_delay_info(crossid, nodeid, data_list, data_type, roads_d (x for x in pb_data_list if x is not None), key=lambda x: x.delay_info.car_num, default=None - ).delay_info.car_num + ).delay_info.car_num if pb_data_list else 0 for item in data_list: tp_start = item['tp_start'] tp_end = item['tp_end'] @@ -696,7 +698,7 @@ def parse_single_cross_delay_info(crossid, nodeid, data_list, data_type, roads_d overview_data = ['-', '-', '-', '-', '-', '-', '-', '-', '-', '-', # 下方为上述指标变化率颜色展示flag 0, 0, 0, 0, 0, 0, 0, 0, 0, 0] - tide_index = calc_tide_index(crossid, nodeid, [day], roads_dir_dict) if data_type == 'day' or data_type == 'week' else '-' + tide_index = calc_tide_index(crossid, nodeid, [day], roads_dir_dict, date_type='week' if data_type == 'week' else None) if data_type != 'hour' else '-' road_data_dict = {} cross_car_num = 0 if item_cross_delay_info: @@ -907,25 +909,25 @@ def parse_data2pb(data_list): return res_list -def gen_cross_problems(avg_cross_delay_info, roads_dir_dict, inroad_static_info_dict): +def gen_cross_problems(avg_cross_delay_info, roads_dir_dict, inroad_static_info_dict, cross_phase, is_peak): # 运行效率、均衡调控、配时方案、路口渠化 - operating_efficiency_problems = gen_operating_efficiency_problems(avg_cross_delay_info, roads_dir_dict) - balanced_control_problems = gen_balanced_control_problems(avg_cross_delay_info, roads_dir_dict, inroad_static_info_dict) + 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) phase_problems = gen_phase_problems() cross_channelized_problems = gen_cross_channelized_problems() problems = [operating_efficiency_problems, balanced_control_problems, phase_problems, cross_channelized_problems] return problems -def gen_operating_efficiency_problems(avg_cross_delay_info, roads_dir_dict): +def gen_operating_efficiency_problems(avg_cross_delay_info, roads_dir_dict, cross_phase, is_peak): if not avg_cross_delay_info: return [{ 'item': '运行效率', 'values': [] }] road_delay_infos = avg_cross_delay_info.inroad_delay_infos - high_park_problems, high_park_suggestions = gen_high_park_problems(road_delay_infos, roads_dir_dict) - + high_park_problems, high_park_suggestions = gen_high_park_problems(road_delay_infos, roads_dir_dict, cross_phase) + high_stop_times_problems, high_stop_times_suggestions = gen_high_stop_time_problems(avg_cross_delay_info, is_peak) operating_efficiency_problems = { 'item': '运行效率', 'values': [ @@ -935,14 +937,24 @@ def gen_operating_efficiency_problems(avg_cross_delay_info, roads_dir_dict): 'reason': '某一进口道转向的多次停车率大于15%', 'suggestions': high_park_suggestions }, + { + 'item': '停车较多', + 'detail': high_stop_times_problems, + 'reason': '高峰时段路口停车次数大于2次，非高峰时段停车次数大于1次', + 'suggestions': high_stop_times_suggestions + } ] } - pass + return operating_efficiency_problems -def gen_high_park_problems(road_delay_infos, roads_dir_dict): +def gen_high_park_problems(road_delay_infos, roads_dir_dict, cross_phase): detail = [{ - 'text': '未见异常' + 'child_detail': [ + { + 'text': '未见异常' + } + ] }] suggestion = [] err_src_dict = {} @@ -966,12 +978,12 @@ def gen_high_park_problems(road_delay_infos, roads_dir_dict): if err_src_dict: detail = [] for src_dir in err_src_dict.keys(): - src_detail = [] + src_detail, src_suggestions, turn_type_list = [], [], [] for turn_type in err_src_dict[src_dir]: + turn_type_list.append(int(turn_type.split(':')[0])) flow_detail = { 'turn_type': int_turn_type2str[int(turn_type.split(':')[0])], - 'text': '车辆多次停车率过大（', - 'percent': turn_type.split(':')[1] + '%），' + 'text': '车辆多次停车率过大（' + turn_type.split(':')[1] + '%），' } src_detail.append(flow_detail) src_detail.append({ @@ -979,24 +991,68 @@ def gen_high_park_problems(road_delay_infos, roads_dir_dict): }) detail.append({ 'src_dir': dir_str_dict[src_dir] + '进口', - 'src_detail': src_detail + 'child_detail': src_detail }) - # todo 补充生成建议的逻辑 + turn_type_str = '、'.join([int_turn_type2str[int(item.split(':')[0])] for item in err_src_dict[src_dir]]) + src_suggestions.append( + { + 'text': '增加' + } + ) + src_suggestions.append({ + 'src_dir': dir_str_dict[src_dir] + '进口' + turn_type_str, + 'text': '所属相位的绿灯时间' + }) + phase_suggestions = get_flow_phase_detail(src_dir, turn_type_list, cross_phase) + suggestion.append({ + 'child_detail': src_suggestions + }) + suggestion.extend(phase_suggestions) return detail, suggestion -def gen_high_stop_time_problems(road_delay_infos, roads_dir_dict): +def gen_high_stop_time_problems(avg_cross_delay_info, is_peak): detail = [{ - 'text': '未见异常' + 'child_detail': [ + { + 'text': '未见异常' + } + ] }] suggestion = [] - # todo 需与产品再次确认逻辑 - pass + desc, max_stop_times = '非高峰时段', 1 + 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: + detail = [{ + 'child_detail': [ + { + 'desc': desc, + 'text': '车辆多次停车时长过长（' + str(avg_cross_delay_info.delay_info.stop_times) + '），整体运行效率不高', + } + ] + }] + suggestion = [ + { + 'child_detail': [ + { + 'text': '对路口', + 'desc': desc + }, + { + 'text': '配时方案进行优化' + } + ] + } + ] + return detail, suggestion -def gen_balanced_control_problems(avg_cross_delay_info, roads_dir_dict, inroad_static_info_dict): +def gen_balanced_control_problems(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_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 需要与产品确认路口潮汐问题诊断的逻辑 balanced_control_problems = { @@ -1020,9 +1076,13 @@ def gen_balanced_control_problems(avg_cross_delay_info, roads_dir_dict, inroad_s # 路口失衡问题诊断 -def gen_cross_imbalance_problems(road_delay_infos, roads_dir_dict): +def gen_cross_imbalance_problems(road_delay_infos, roads_dir_dict, cross_phase): detail = [{ - 'text': '未见异常' + 'child_detail': [ + { + 'text': '未见异常' + } + ] }] suggestion = [] road_src_dict = {item['in']: item for item in roads_dir_dict} @@ -1032,47 +1092,66 @@ def gen_cross_imbalance_problems(road_delay_infos, roads_dir_dict): and max_stop_times_road.delay_info.stop_times - min_stop_times_road.delay_info.stop_times > 0.5: max_roadid = max_stop_times_road.inroadid min_roadid = min_stop_times_road.inroadid - max_src = road_src_dict[max_roadid] - min_src = road_src_dict[min_roadid] + max_src = road_src_dict[max_roadid] if max_roadid in road_src_dict else None + min_src = road_src_dict[min_roadid] if min_roadid in road_src_dict else None + if max_src is None or min_src is None: + return detail, suggestion detail = [ { - 'src_dir': dir_str_dict[max_src] + '进口', - 'text': '的停车次数（' + str(round(max_stop_times_road.delay_info.stop_times, 2)) + '）与' - }, - { - 'src_dir': dir_str_dict[min_src] + '进口', - 'text': f"""的停车次数（{str(round(max_stop_times_road.delay_info.stop_times, 2))}）相差过大，两者之比为{int(round(max_stop_times_road.delay_info.stop_times, 2) / round(min_stop_times_road.delay_info.stop_times, 2) * 100)}%，分配的绿灯时长不匹配""" + 'child_detail': [ + { + 'src_dir': dir_str_dict[max_src] + '进口', + 'text': '的停车次数（' + str(round(max_stop_times_road.delay_info.stop_times, 2)) + '）与' + }, + { + 'src_dir': dir_str_dict[min_src] + '进口', + 'text': f"""的停车次数（{str(round(max_stop_times_road.delay_info.stop_times, 2))}）相差过大，两者之比为{int(round(max_stop_times_road.delay_info.stop_times, 2) / round(min_stop_times_road.delay_info.stop_times, 2) * 100)}%，分配的绿灯时长不匹配""" + } + ] } ] suggestion = [ { - 'text': '调整', - 'max_src_dir': dir_str_dict[max_src] + '进口', - }, - { - 'text': '和', - 'min_src_dir': dir_str_dict[min_src] + '进口', - }, - { - 'text': '的绿灯时长的分配情况' + 'child_detail': [ + { + 'text': '调整', + 'max_src_dir': dir_str_dict[max_src] + '进口', + }, + { + 'text': '和', + 'min_src_dir': dir_str_dict[min_src] + '进口', + }, + { + 'text': '的绿灯时长的分配情况' + } + ] } ] + is_tide = False if max_src == src_reverse[min_src]: - # todo 补充对向情况下额外建议的情况 - pass + is_tide = True + phase_suggestions = gen_src_dir_phase_detail(max_src, min_src, cross_phase, is_tide=is_tide) + suggestion.extend(phase_suggestions) return detail, suggestion # 转向失衡问题诊断 -def gen_turn_imbalance_problems(road_delay_infos, roads_dir_dict, inroad_static_info_dict): +def gen_turn_imbalance_problems(road_delay_infos, roads_dir_dict, inroad_static_info_dict, cross_phase): detail = [{ - 'text': '未见异常' + 'child_detail': [ + { + 'text': '未见异常' + } + ] }] suggestion = [] err_road_dict = {} road_src_dict = {item['in']: item for item in roads_dir_dict} for road_delay_info in road_delay_infos: inroadid = road_delay_info.inroadid + src_dir = road_src_dict[inroadid] if inroadid in road_src_dict.keys() else None + if src_dir is None: + continue flow_delay_infos = road_delay_info.flow_delay_infos turn_type_flow_delay_info_dict = {item.turn_type: item for item in flow_delay_infos} lane_num_info = count_lsr(inroad_static_info_dict[inroadid]['lane_turn_info']) if inroadid in inroad_static_info_dict.keys() else None @@ -1109,27 +1188,40 @@ def gen_turn_imbalance_problems(road_delay_infos, roads_dir_dict, inroad_static_ detail.append(item_detail) item_suggestion = [ { - 'src_dir': dir_str_dict[src_dir] + '进口', - 'text': '信号灯直左分控，调整执行和左转车辆绿灯时长分配情况' - }, - # todo 补充配时相关建议 + 'child_detail': [ + { + 'src_dir': dir_str_dict[src_dir] + '进口', + 'text': '信号灯直左分控，调整直行和左转车辆绿灯时长分配情况' + } + ] + } ] suggestion.extend(item_suggestion) + phase_suggestion = get_flow_phase_detail(src_dir, ['直行', '左转'], cross_phase) + suggestion.extend(phase_suggestion) if err_road_dict[src_dir][2]: road_num_suggestion.append( { - 'src_dir': dir_str_dict[src_dir] + '进口', - 'text': '左转/直行/右转现有车道数分别为' + err_road_dict[src_dir][2] + 'child_detail': [ + { + 'src_dir': dir_str_dict[src_dir] + '进口', + 'text': '左转/直行/右转现有车道数分别为' + err_road_dict[src_dir][2] + } + ] } ) lane_num_suggestion = [ { - 'text': '调整', - 'src_dir': '、'.join([dir_str_dict[item] + '进口' for item in err_road_dict.keys()]) - }, - { - 'text': '车道分配情况' + 'child_detail': [ + { + 'text': '调整', + 'src_dir': '、'.join([dir_str_dict[item] + '进口' for item in err_road_dict.keys()]) + }, + { + 'text': '车道分配情况' + } + ] } ] lane_num_suggestion.extend(road_num_suggestion) @@ -1142,14 +1234,264 @@ def gen_phase_problems(): pass -def gen_cross_channelized_problems(): +def gen_cross_channelized_problems(inroad_static_info_dict): + + pass -def query_cross_phase(nodeid, crossid): - cross_phases, err = QueryCrossRunningPhase(nodeid, [crossid]) - if err or not cross_phases or cross_phases.code != 0: - return None - cross_phase_info = cross_phases.data[0] +def gen_inroad_num_problems(road_delay_infos, inroad_static_info_dict, roads_dir_dict): + detail = [{ + 'child_detail': [ + { + 'text': '未见异常' + } + ] + }] + suggestions, err_src_dirs = [], {} + road_src_dict = {item['in']: item for item in roads_dir_dict} + max_flow_car_num = 0 + + for road_delay in road_delay_infos: + inroadid = road_delay.inroadid + src_dir = road_src_dict[inroadid] + if src_dir is None: + continue + lane_info = inroad_static_info_dict[inroadid]['lane_turn_info'] + flow_delay_infos = road_delay.flow_delay_infos + max_car_num = max([item.delay_info.car_num for item in flow_delay_infos if item.turn_type in [0, 1]]).delay_info.car_num + if max_car_num > max_flow_car_num: + max_flow_car_num = max_car_num + 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) + 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 + for turn_type in [0, 1]: + if turn_type not in turn_delay_dict.keys(): + continue + if turn_type == 0: + turn_type_str = '直行' + lane_num_rate = straight_lane_num_rate + else: + turn_type_str = '左转' + lane_num_rate = left_lane_num_rate + flow_car_num = turn_delay_dict[turn_type].delay_info.car_num + relative_rate = flow_car_num / max_flow_car_num + flow_rate = flow_car_num / road_car_num + if relative_rate > 0.6 and lane_num_rate < 0.5: + if src_dir in err_src_dirs.keys(): + err_src_dirs[src_dir].append({ + 'turn_type': turn_type_str, + 'flow_rate': flow_rate, + 'lane_num_rate': lane_num_rate + }) + else: + err_src_dirs[src_dir] = [{ + 'turn_type': turn_type_str, + 'flow_rate': flow_rate, + 'lane_num_rate': lane_num_rate + }] + if len(err_src_dirs.keys()) > 0: + detail = [] + for src_dir in err_src_dirs.keys(): + err_road_dict = err_src_dirs[src_dir] + src_detail = [ + { + 'src_dir': dir_str_dict[src_dir] + '进口' + } + ] + for turn_type in err_road_dict: + src_detail.append({ + 'child_detail': [ + { + 'turn_type': turn_type['turn_type'], + 'text': '分流转向比过大（' + str(int(round(turn_type['flow_rate'], 2) * 100)) + '%），但分配的转向车道占比不足（' + str(int(round(turn_type['lane_num_rate'], 2) * 100)) + '%），分配车道资源不匹配' + } + ] + }) + detail.append( + { + 'child_detail': src_detail + } + ) + + pass + pass + + +def check_is_peak_tp(time_range, area_id, nodeid): + tp_desc = db_tmnet.query_city_tp_info(nodeid, area_id) + peak_tp = ['07:00-09:00', '17:00-19:00'] + if tp_desc and len(tp_desc) > 0: + peak_tp = tp_desc[0]['peak_tp'].split(',') + return time_overlap(time_range, peak_tp) + + +def get_flow_phase_detail(src_dir, turn_type_list, cross_phase): + phase_suggestion = [] + for item in cross_phase: + # 单个元素的含义为一个周计划 + schedule_suggestion = [] + need_src_flow = {dir_str_dict[src_dir] + int_turn_type2str[item]: 0 for item in turn_type_list} + schedule_name = item.schedule_name + tp_info = item.tps[0] + planid = tp_info.planid + plan_name = tp_info.plan_name + stage_list = tp_info.stages + for stage in stage_list: + phases_name = stage.phases_name + phases_name_list = [] + if phases_name != '': + phases_name_list = phases_name.split(',') + green = stage.green + for key in need_src_flow.keys(): + if key in phases_name_list: + need_src_flow[key] += green + need_src_flow = {k: v for k, v in need_src_flow.items() if v != 0} + if len(need_src_flow.keys()) > 0: + schedule_suggestion.append( + { + 'phase': f"【{schedule_name}】{str(planid)}-{plan_name}", + 'text': '现有的' + } + ) + keys = list(need_src_flow.keys()) + for i in range(len(keys)): + if i != len(keys) - 1: + schedule_suggestion.append( + { + 'src_dir': keys[i], + 'text': '绿灯时长为' + str(keys[i]) + 's，' + } + ) + else: + schedule_suggestion.append( + { + 'src_dir': keys[i], + 'text': '绿灯时长为' + str(keys[i]) + 's' + } + ) + phase_suggestion.append({ + 'child_detail': schedule_suggestion + }) + + return phase_suggestion + + +def gen_src_dir_phase_detail(max_src_dir, min_src_dir, cross_phase, is_tide=False): + phase_suggestion = [] + for item in cross_phase: + # 单个元素的含义为一个周计划 + schedule_suggestion = [] + schedule_name = item.schedule_name + src_green = { + dir_str_dict[max_src_dir]: 0, + dir_str_dict[min_src_dir]: 0 + } + tp_info = item.tps[0] + planid = tp_info.planid + plan_name = tp_info.plan_name + stage_list = tp_info.stages + for stage in stage_list: + phases_name = stage.phases_name + src_dir_set = set() + if phases_name != '': + tmp_list = phases_name.split(',') + for phase in tmp_list: + if '行人' in phase: + continue + src_dir_set.add(phase[:-2]) + if dir_str_dict[max_src_dir] in src_dir_set: + src_green[dir_str_dict[max_src_dir]] += stage.green + if dir_str_dict[min_src_dir] in src_dir_set: + src_green[dir_str_dict[min_src_dir]] += stage.green + if src_green[dir_str_dict[max_src_dir]] != 0 and src_green[dir_str_dict[min_src_dir]] != 0: + schedule_suggestion = [ + { + 'phase': f"【{schedule_name}】{str(planid)}-{plan_name}", + 'text': '现有的' + }, + { + 'src_dir': dir_str_dict[max_src_dir] + '进口', + 'text': '绿灯时长为' + str(src_green[dir_str_dict[max_src_dir]]) + 's，' + }, + { + 'src_dir': dir_str_dict[min_src_dir] + '进口', + 'text': '绿灯时长为' + str(src_green[dir_str_dict[min_src_dir]]) + 's' + } + ] + if schedule_suggestion: + phase_suggestion.append({ + 'child_detail': schedule_suggestion + }) + need2sug = False + if is_tide: + for item in cross_phase: + stages_info = item.tps[0].stages + for stage in stages_info: + phases_name = stage.phases_name + if dir_str_dict[max_src_dir] + '直行' in phases_name\ + and dir_str_dict[min_src_dir] + '直行' in phases_name\ + and dir_str_dict[max_src_dir] + '左转' in phases_name\ + and dir_str_dict[min_src_dir] + '左转' in phases_name: + need2sug = True + if need2sug: + phase_suggestion.append({ + 'text': '相位放行方式可以考虑从对称放行调整为单路口放行' + }) + + return phase_suggestion + + +def count_lsr_half(turn_str): + """ + 统计车道转向中左转(l)、直行(s)、右转(r)的数量。 + 当一个车道有多个转向时，每个转向计0.5。 + + Args: + turn_str: 形如 "2|1|1|1|5" 的字符串，每个数字对应 g_turn2str 中的键。 + + Returns: + 形如 "1.0/3.0/1.0" 的字符串，顺序为左转、直行、右转的数量。 + """ + + # 初始化计数器 + left_count = 0.0 + straight_count = 0.0 + right_count = 0.0 + + # 分割输入字符串 + turn_keys = turn_str.split('|') + + # 遍历每个车道 + for key in turn_keys: + if key in g_turn2str: + turn_value = g_turn2str[key] + # 忽略掉头和特殊车道 + if turn_value in ['t', 'bus', 'reversible', '-']: + continue + + # 计算该车道的转向数量 + turn_count = 0 + if 'l' in turn_value: + turn_count += 1 + if 's' in turn_value: + turn_count += 1 + if 'r' in turn_value: + turn_count += 1 + + # 如果该车道有转向，则按转向数量平分权重 + if turn_count > 0: + weight = 0.5 if turn_count > 1 else 1.0 + if 'l' in turn_value: + left_count += weight + if 's' in turn_value: + straight_count += weight + if 'r' in turn_value: + right_count += weight + + # 返回格式化的结果，保留一位小数 + return round(left_count, 1), round(straight_count, 1), round(right_count, 1) - pass \ No newline at end of file diff --git a/app/tmnet_db_func.py b/app/tmnet_db_func.py index 5503215..9d04e84 100644 --- a/app/tmnet_db_func.py +++ b/app/tmnet_db_func.py @@ -91,5 +91,5 @@ class TmnetDbHelper(TableDbHelperBase): return self.do_select(sql) def query_city_tp_info(self, nodeid, area_id): - sql = f"select tp_desc from cross_doctor_config.area_tp_config where nodeid = {nodeid} and area_id = {area_id}" + sql = f"select tp_desc, peak_tp from cross_doctor_config.area_tp_config where nodeid = {nodeid} and area_id = {area_id}" return self.do_select(sql) \ No newline at end of file