提交除了配时方案异常以外的问题诊断代码

app/cross_evaluate_worker.py

@@ -36,9 +36,7 @@ def query_cross_list(params):
         tp_info = [
             "00:00-07:00",
             "07:00-09:00",
-            "09:00-12:00",
+            "09:00-17:00",
-            "12:00-14:00",
-            "14:00-17:00",
             "17:00-19:00",
             "19:00-22:00",
             "22:00-00:00"
@@ -128,6 +126,11 @@ def query_cross_delay_info_controller(params):
     avg_cross_delay_info = gen_avg_cross_delay_pb(cross_delay_data_list)
     if not avg_cross_delay_info:
         return json.dumps(make_common_res(9, '当前所选日期范围内该评测时段无可用数据'))
+    # print(MessageToJson(
+    #     avg_cross_delay_info,
+    #     always_print_fields_with_no_presence=True,  # 输出所有默认值
+    #     preserving_proto_field_name=True      # 保持原始字段名（不用驼峰）
+    # ))
     cross_inroads = db_tmnet.query_cross_inroads(crossid, nodeid)
     inroad_static_info_dict = {item['roadid']: item for item in cross_inroads}
     # 路口静态信息及台账信息

app/eva_common.py

@@ -523,17 +523,9 @@ def gen_flow_turn_rate_index(avg_cross_delay_info, roads_dir_dict):
         if roadid != '-' and roadid in road_delay_dict.keys():
             car_num = road_delay_dict[roadid].delay_info.car_num
             in_flow_rate = round(car_num / cross_sum_car_num, 2) if cross_out_sum_car_num != 0 else '-'
-
+            l_rate = round(road_delay_dict[roadid].delay_info.turn_ratio_1 / car_num, 2) if car_num != 0 else '-'
-            flow_delay_infos = road_delay_dict[roadid].flow_delay_infos
+            s_rate = round(road_delay_dict[roadid].delay_info.turn_ratio_0 / car_num, 2) if car_num != 0 else '-'
-            for flow_delay_info in flow_delay_infos:
+            r_rate = round(road_delay_dict[roadid].delay_info.turn_ratio_2 / car_num, 2) if car_num != 0 else '-'
-                if flow_delay_info.turn_type not in (0, 1, 2):
-                    continue
-                if flow_delay_info.turn_type == 0:
-                    l_rate = round(flow_delay_info.delay_info.car_num / car_num, 2) if car_num != 0 else '-'
-                elif flow_delay_info.turn_type == 1:
-                    s_rate = round(flow_delay_info.delay_info.car_num / car_num, 2) if car_num != 0 else '-'
-                elif flow_delay_info.turn_type == 2:
-                    r_rate = round(flow_delay_info.delay_info.car_num / car_num, 2) if car_num != 0 else '-'
         out_road_id = roads_dir_dict[dir]['out']
         if out_road_id != '-' and out_road_id in outroad_info_dict.keys():
             out_car_num = outroad_info_dict[out_road_id].turn_info.car_num
@@ -678,11 +670,12 @@ def parse_single_cross_delay_info(crossid, nodeid, data_list, data_type, roads_d
                 for flow_delay_info in flow_delay_infos:
                     if flow_delay_info.delay_info.car_num > max_flow_car_num:
                         max_flow_car_num = flow_delay_info.delay_info.car_num
-    max_cross_car_num = max(
+    max_cross_car_num_pb = max(
                 (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 if pb_data_list else 0
+            )
+    max_cross_car_num = max_cross_car_num_pb.delay_info.car_num if max_cross_car_num_pb else 0
     for item in data_list:
         tp_start = item['tp_start']
         tp_end = item['tp_end']
@@ -914,6 +907,7 @@ def parse_data2pb(data_list):
     return res_list
 
 
+# 生成路口诊断问题
 def gen_cross_problems(crossid, nodeid, date_list, avg_cross_delay_info, roads_dir_dict, inroad_static_info_dict, cross_phase, is_peak, cross_ledger_info):
     # 运行效率、均衡调控、配时方案、路口渠化
     operating_efficiency_problems = gen_operating_efficiency_problems(avg_cross_delay_info, roads_dir_dict, cross_phase, is_peak)
@@ -924,6 +918,7 @@ def gen_cross_problems(crossid, nodeid, date_list, avg_cross_delay_info, roads_d
     return problems
 
 
+# 生成运行效率诊断问题结果
 def gen_operating_efficiency_problems(avg_cross_delay_info, roads_dir_dict, cross_phase, is_peak):
     if not avg_cross_delay_info:
         return [{
@@ -953,6 +948,7 @@ def gen_operating_efficiency_problems(avg_cross_delay_info, roads_dir_dict, cros
     return operating_efficiency_problems
 
 
+# 运行效率-多次排队
 def gen_high_park_problems(road_delay_infos, roads_dir_dict, cross_phase):
     detail = [{
         'child_detail': [
@@ -1016,6 +1012,7 @@ def gen_high_park_problems(road_delay_infos, roads_dir_dict, cross_phase):
     return detail, suggestion
 
 
+# 运行效率-停车较多
 def gen_high_stop_time_problems(avg_cross_delay_info, is_peak):
     detail = [{
         'child_detail': [
@@ -1055,6 +1052,7 @@ def gen_high_stop_time_problems(avg_cross_delay_info, is_peak):
     return detail, suggestion
 
 
+# 均衡调控
 def gen_balanced_control_problems(crossid, nodeid, date_list, avg_cross_delay_info, roads_dir_dict, inroad_static_info_dict, cross_phase):
     road_delay_infos = avg_cross_delay_info.inroad_delay_infos
     cross_imbalance_detail, cross_imbalance_suggestions = gen_cross_imbalance_problems(road_delay_infos, roads_dir_dict, cross_phase)
@@ -1086,7 +1084,7 @@ def gen_balanced_control_problems(crossid, nodeid, date_list, avg_cross_delay_in
     return balanced_control_problems
 
 
-# 路口失衡问题诊断
+# 均衡调控-路口失衡问题诊断
 def gen_cross_imbalance_problems(road_delay_infos, roads_dir_dict, cross_phase):
     detail = [{
         'child_detail': [
@@ -1146,7 +1144,7 @@ def gen_cross_imbalance_problems(road_delay_infos, roads_dir_dict, cross_phase):
     return detail, suggestion
 
 
-# 转向失衡问题诊断
+# 均衡调控-转向失衡问题诊断
 def gen_turn_imbalance_problems(road_delay_infos, roads_dir_dict, inroad_static_info_dict, cross_phase):
     detail = [{
         'child_detail': [
@@ -1241,7 +1239,7 @@ def gen_turn_imbalance_problems(road_delay_infos, roads_dir_dict, inroad_static_
     return detail, suggestion
 
 
-# 路口潮汐问题判定
+# 均衡调控-路口潮汐问题诊断
 def gen_cross_tide_problems(crossid, nodeid, date_list, roads_dir_dict):
     detail = [{
         'child_detail': [
@@ -1330,10 +1328,12 @@ def gen_cross_tide_problems(crossid, nodeid, date_list, roads_dir_dict):
     return detail, suggestions
 
 
+# 配时方案
 def gen_phase_problems():
     pass
 
 
+# 路口渠化
 def gen_cross_channelized_problems(avg_cross_delay_info, roads_dir_dict, inroad_static_info_dict, cross_ledger_info):
     road_delay_infos = avg_cross_delay_info.inroad_delay_infos
     inroad_num_detail, inroad_num_suggestion = gen_inroad_num_problems(road_delay_infos, inroad_static_info_dict, roads_dir_dict)
@@ -1358,6 +1358,7 @@ def gen_cross_channelized_problems(avg_cross_delay_info, roads_dir_dict, inroad_
     return cross_channelized_problems
 
 
+# 路口渠化-车道资源不匹配
 def gen_inroad_num_problems(road_delay_infos, inroad_static_info_dict, roads_dir_dict):
     detail = [{
         'child_detail': [
@@ -1461,6 +1462,7 @@ def gen_inroad_num_problems(road_delay_infos, inroad_static_info_dict, roads_dir
     return detail, suggestions
 
 
+# 路口渠化-进出口道数不匹配
 def gen_in_out_lane_num_gap_problems(cross_ledger_info):
     detail = [{
         'child_detail': [

app/gwperf_cloud_db_func.py

@@ -105,11 +105,11 @@ class GWPerfCloudDbHelper(TableDbHelperBase):
         except Exception as error:
             return None, error
 
-    def query_all_gw_cross_examine_problems(self, day):
+    def query_all_cross_examine_problems(self, day):
         if day == 0:
             return []
         sql = """
-            select waveid, start_hm, end_hm, weekday, coordinate_dir, index_reason, index_details, xxlight_reason, xxlight_details from gw_examine_problems where day = '%s'
+            select * from cross_examine_problems where day = '%s'
         """ % (day)
         return self.do_select(sql)
 
@@ -155,7 +155,7 @@ class GWPerfCloudDbHelper(TableDbHelperBase):
             return None, error
         
     
-    def query_cross_offset_by_light(self, waveid, crossids, day, start_hm):
+    def query_cross_offset_by_lights(self, waveid, crossids, day, start_hm):
         conn, cursor = self.connect()
         try:
             sql = '''SELECT

app/tmnet_db_func.py

@@ -93,3 +93,61 @@ class TmnetDbHelper(TableDbHelperBase):
     def query_city_tp_info(self, nodeid, 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)
+
+    def query_all_cross_examine_records(self, yes_str):
+        sql = "select * from cross_doctor_matedata.cross_phase_problems_record where date(update_time) = '%s' and end_date is null" % yes_str
+        return self.do_select(sql)
+
+    def insert_cross_examine_records(self, insert_list):
+        conn, cursor = self.connect()
+        try:
+            sql = """insert into cross_doctor_matedata.cross_phase_problems_record(start_hm,end_hm,crossid,phase_type, phase_detail,cont_times,final_state,level_color,first_date,change_red_daynums)
+                    values(%s, %s, %s, %s,%s, %s, %s, %s,%s, %s)
+            """
+            values = [
+                (
+                    d['start_hm'], d['end_hm'], d['crossid'], d['phase_type'], d['phase_detail'], d['cont_times'], d['final_state'],
+                    d['level_color'], d['first_date'], d['change_red_daynums']
+                ) for d in insert_list
+            ]
+            ret = cursor.executemany(sql, values)
+            if ret == len(insert_list):
+                conn.commit()
+                self.close(conn, cursor)
+                return ret, None
+            else:
+                conn.rollback()
+                self.close(conn, cursor)
+                return 0, "insert_cross_examine_records error"
+        except Exception as e:
+            conn.rollback()
+            self.close(conn, cursor)
+            return 0, e
+
+    def update_cross_examine_records(self, update_list):
+        conn, cursor = self.connect()
+        try:
+            sql = """
+                update cross_doctor_matedata.cross_phase_problems_record
+                set phase_type = %s, phase_detail = %s, cont_times = %s, final_state = %s, level_color = %s, change_red_daynums = %s,
+                end_date = %s where crossid = %s and start_hm = %s and first_date = %s
+            """
+            values = [
+                (
+                    d['phase_type'], d['phase_detail'], d['cont_times'], d['final_state'], d['level_color'],
+                    d['change_red_daynums'], d['end_date'], d['crossid'], d['start_hm'], d['first_date']
+                ) for d in update_list
+            ]
+            ret = cursor.executemany(sql, values)
+            if ret == len(update_list):
+                conn.commit()
+                self.close(conn, cursor)
+                return ret, None
+            else:
+                conn.rollback()
+                self.close(conn, cursor)
+                return ret, None
+        except Exception as e:
+            conn.rollback()
+            self.close(conn, cursor)
+            return 0, e

daily_cross_pahse_problems.py

@@ -0,0 +1,278 @@
+# -*- coding: utf-8 -*-
+# @Author: Owl
+# @Date: 2025/10/28 16:25
+# @Description:
+import argparse
+
+# from app.gw_db_func import GreenWaveDbHelper
+from app.gwperf_cloud_db_func import *
+from app.eva_common import *
+from tool.mysql_common_connector_pool import DatabaseManager
+
+g_cloud_db = {
+    'host': 'bj-cdb-64eqw2oe.sql.tencentcdb.com',
+    'port': 26807,
+    'user': 'root',
+    'password': 'pmenJIn7EaK40oThn',
+    'db': 'gwperf'
+}
+
+g_dbinfo = {
+    'host': '120.53.125.169',
+    'port': 3306,
+    'user': 'root',
+    'password': 'pmenJIn7EaK40oThn~~~',
+    'db': 'cross_doctor_matedata'
+}
+
+src_str2_eng = {
+    "东": "E",
+    "南": "S",
+    "西": "W",
+    "北": "N",
+    "东南": "SE",
+    "东北": "NE",
+    "西南": "SW",
+    "西北": "NW"
+}
+
+g_cloud_pool = DatabaseManager(g_cloud_db)
+g_db_cloud = GWPerfCloudDbHelper(g_cloud_pool)
+
+g_db_pool = DatabaseManager(g_dbinfo)
+g_db_cross = TmnetDbHelper(g_db_pool)
+
+
+def dev_args():
+    try:
+        parser = argparse.ArgumentParser()
+        parser.add_argument('--dev', default=0, help='开发环境参数：0否，1是')
+        parser.add_argument('--day', default=0, help='开发环境参数：0否，1是')
+        args = parser.parse_args()
+        return args
+    except SystemExit:
+        pass
+    return None
+
+
+def init(args):
+    dev, day = 0, 0
+    if args:
+        dev = int(args.dev)
+        day = args.day
+    if dev == 0:
+        g_cloud_db['host'] = '172.21.32.41'
+        g_cloud_db['port'] = 3306
+        g_dbinfo['host'] = '172.21.32.21'
+    g_db_pool.init_pool(g_dbinfo)
+    g_cloud_pool.init_pool(g_cloud_db)
+    if day == 0:
+        day = (datetime.now() - timedelta(days=1)).strftime('%Y%m%d')
+    today_data = g_db_cloud.query_all_cross_examine_problems(day)
+    yes_str = datetime.strptime(day, '%Y%m%d').strftime('%Y-%m-%d')
+    yesterday_records = g_db_cross.query_all_cross_examine_records(yes_str)
+    yesterday_data_dict = parse_records2dict(yesterday_records)
+    today_records = {}
+    for row in today_data:
+        start_hm = row['start_hm']
+        end_hm = row['end_hm']
+        crossid = row['crossid']
+        if row['xxlight_details'] and row['xxlight_details'] != '':
+            xxlight_detail = row['xxlight_details']
+            xxlight_reason = row['xxlight_reason']
+            key = '{}-{}-{}'.format(start_hm, end_hm, crossid)
+            today_records[key] = {
+                'cont_times': 1,
+                'phase_type': xxlight_detail,
+                'phase_detail': xxlight_reason,
+                'index_detail': '',
+                'index_detail_r': ''
+            }
+    insert_list, update_list = calc_final_state_color(yesterday_data_dict, today_records, day)
+    insert_ret, update_ret, update_e, insert_e = 0, 0, None, None
+    # 先清除当前天的数据]
+    if len(insert_list) > 0:
+        insert_ret, insert_e = g_db_cross.insert_cross_examine_records(insert_list)
+    if len(update_list) > 0:
+        update_ret, update_e = g_db_cross.update_cross_examine_records(update_list)
+    if insert_ret == len(insert_list) and update_ret == len(update_list) and not insert_e and not update_e:
+        print('success')
+    else:
+        print('insert_ret: {}, update_ret: {}'.format(insert_ret, update_ret))
+        print('insert_list: {}, update_list: {}'.format(len(insert_list), len(update_list)))
+        print(update_e)
+        print(insert_e)
+        print('failed')
+
+
+def parse_records2dict(records):
+    record_dict = {}
+    for row in records:
+        start_hm = row['start_hm']
+        end_hm = row['end_hm']
+        crossid = row['crossid']
+        key = '{}-{}-{}'.format(start_hm, end_hm, crossid)
+        phase_type = row['phase_type']
+        phase_detail = row['phase_detail']
+        final_state = row['final_state']
+        level_color = row['level_color']
+        first_date = row['first_date']
+        change_red_time = row['change_red_daynums']
+        cont_times = row['cont_times']
+        record_dict[key] = {
+            'final_state': final_state,
+            'level_color': level_color,
+            'first_date': first_date,
+            'change_red_daynums': change_red_time,
+            'phase_type': phase_type,
+            'phase_detail': phase_detail,
+            'cont_times': cont_times
+        }
+    return record_dict
+
+
+def calc_final_state_color(yesterday_data_dict, today_records, day):
+    insert_list, update_list = [], []
+    for key in today_records.keys():
+        start_hm = key.split('-')[0]
+        end_hm = key.split('-')[1]
+        crossid = key.split('-')[2]
+        phase_type = today_records[key]['phase_type']
+        phase_detail = today_records[key]['phase_detail']
+        if key in yesterday_data_dict.keys():
+            phase_cont_times = today_records[key]['cont_times'] + yesterday_data_dict[key]['cont_times']
+            change_red_daynums = yesterday_data_dict[key]['change_red_daynums']
+            yesterday_final_state = int(yesterday_data_dict[key]['final_state'])
+            yesterday_level_color = yesterday_data_dict[key]['level_color']
+            first_date = yesterday_data_dict[key]['first_date']
+            end_date = None
+            if yesterday_final_state not in [4, 5, 6]:
+                if yesterday_final_state == 1:
+                    final_state, level_color = 1, 1
+                elif yesterday_final_state == 2:
+                    if yesterday_level_color == 2:
+                        if today_records[key]['cont_times'] == 1:
+                            final_state, level_color, change_red_daynums = 2, 1, 0
+                        else:
+                            final_state, level_color = 2, 2
+                    else:
+                        if change_red_daynums >= 2:
+                            final_state, level_color = 1, 1
+                        else:
+                            final_state, level_color = 2, 1
+                else:
+                    if today_records[key]['cont_times'] == 1:
+                        final_state, level_color = 2, 1
+                    else:
+                        final_state, level_color = 2, 2
+                update_list.append({
+                    "start_hm": start_hm,
+                    "end_hm": end_hm,
+                    "crossid": crossid,
+                    "phase_type": str(yesterday_data_dict[key]['phase_type']) + '^' + str(phase_type),
+                    "phase_detail": yesterday_data_dict[key]['phase_detail'] + '^' + phase_detail,
+                    "cont_times": phase_cont_times,
+                    "final_state": final_state,
+                    "level_color": level_color,
+                    "first_date": first_date,
+                    "change_red_daynums": change_red_daynums,
+                    "end_date": end_date
+                })
+            else:
+                first_date = day
+                phase_cont_times = today_records[key]['cont_times']
+                final_state, level_color = 3, 3
+                insert_list.append({
+                    "start_hm": start_hm,
+                    "end_hm": end_hm,
+                    "crossid": crossid,
+                    "phase_type": str(phase_type),
+                    "phase_detail": phase_detail,
+                    "cont_times": phase_cont_times,
+                    "final_state": final_state,
+                    "level_color": level_color,
+                    "first_date": first_date,
+                    "change_red_daynums": change_red_daynums
+                })
+            yesterday_data_dict.pop(key)
+        else:
+            first_date = day
+            phase_cont_times = today_records[key]['cont_times']
+            change_red_daynums = 0
+            final_state, level_color = 3, 3
+            insert_list.append({
+                "start_hm": start_hm,
+                "end_hm": end_hm,
+                "crossid": crossid,
+                "phase_type": str(phase_type),
+                "phase_detail": phase_detail,
+                "cont_times": phase_cont_times,
+                "final_state": final_state,
+                "level_color": level_color,
+                "first_date": first_date,
+                "change_red_daynums": change_red_daynums
+            })
+    for key in yesterday_data_dict.keys():
+        if yesterday_data_dict[key]['final_state'] in [4, 5, 6]:
+            continue
+        start_hm = key.split('-')[0]
+        end_hm = key.split('-')[1]
+        crossid = key.split('-')[2]
+        phase_type = yesterday_data_dict[key]['phase_type']
+        phase_detail = yesterday_data_dict[key]['phase_detail']
+        end_date = None
+        first_date = yesterday_data_dict[key]['first_date']
+        change_red_daynums = yesterday_data_dict[key]['change_red_daynums']
+        if yesterday_data_dict[key]['final_state'] == 3:
+            end_date = day
+            final_state = 6
+            level_color = yesterday_data_dict[key]['level_color']
+            phase_cont_times = 0
+        elif yesterday_data_dict[key]['final_state'] == 2:
+            phase_cont_times = yesterday_data_dict[key]['cont_times'] + 1
+            if yesterday_data_dict[key]['level_color'] == 2:
+                final_state = 2
+                level_color = 2
+            else:
+                change_red_daynums = yesterday_data_dict[key]['change_red_daynums'] + 1
+                if yesterday_data_dict[key]['change_red_daynums'] >= 2:
+                    final_state = 1
+                    level_color = 1
+                else:
+                    final_state = 2
+                    level_color = 1
+        else:
+            phase_cont_times = yesterday_data_dict[key]['cont_times'] + 1
+            final_state = yesterday_data_dict[key]['final_state']
+            level_color = yesterday_data_dict[key]['level_color']
+        update_list.append({
+            "start_hm": start_hm,
+            "end_hm": end_hm,
+            "crossid": crossid,
+            "phase_type": str(phase_type) + '^' + '',
+            "phase_detail": phase_detail + '^' + '',
+            "cont_times": phase_cont_times,
+            "final_state": final_state,
+            "level_color": level_color,
+            "first_date": first_date,
+            "change_red_daynums": change_red_daynums,
+            "end_date": end_date
+        })
+    return insert_list, update_list
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--dev', default=0, help='开发环境参数：0否，1是')
+    parser.add_argument('--day', default=0, help='执行日期')
+    args = parser.parse_args()
+    init(args)
+
+
+
+
+
+
+
+
+