# -*- coding: utf-8 -*- # @Author: Owl # @Date: 2025/10/13 10:26 # @Description: 用于本地测试的执行文件 import configparser from google.protobuf.json_format import MessageToJson from app.common_worker import generate_date_range, convert_time from app.eva_common import gen_crossids_roads_dir_dict_by_mysql, check_outside_left from app.global_source import * import proto.xlcomm_pb2 as pb def init(): config = configparser.ConfigParser() global g_db_pool, g_db_gw config.read('cross_doctor.ini', encoding='utf-8') net_filename = 'data/xl_roadnet_130200.txt' roadinfo_filename = 'data/xl_roadinfo_130200.txt' roadlinks_filename = 'data/xl_roadlinks_130200.txt' g_roadnet.load_from_file(net_filename) g_roadinfo_manager.load_from_file(roadinfo_filename) g_roadlinks_manager.load_from_file(roadlinks_filename) # if config.has_option('cloud_db', 'host'): # g_cloud_db['host'] = config.get('cloud_db', 'host') # if config.has_option('cloud_db', 'port'): # g_cloud_db['port'] = int(config.get('cloud_db', 'port')) # g_cloud_pool.init_pool(g_cloud_db) # if config.has_option('db', 'host'): # g_dbinfo['host'] = config.get('db', 'host') # g_db_pool.init_pool(g_dbinfo) if config.has_option('cross_delay_db', 'host'): g_cross_delay_db['host'] = config.get('cross_delay_db', 'host') if config.has_option('cross_delay_db', 'port'): g_cross_delay_db['port'] = int(config.get('cross_delay_db', 'port')) if config.has_option('cross_delay_db', 'dbname'): g_cross_delay_db['db'] = config.get('cross_delay_db', 'dbname') if config.has_option('cross_delay_db', 'password'): g_cross_delay_db['password'] = config.get('cross_delay_db', 'password') print(g_cross_delay_db) if config.has_option('roadnet_db', 'host'): host = config.get('roadnet_db', 'host') g_roadnet_db['host'] = host if config.has_option('roadnet_db', 'user'): user = config.get('roadnet_db', 'user') g_roadnet_db['user'] = user if config.has_option('roadnet_db', 'password'): password = config.get('roadnet_db', 'password') g_roadnet_db['password'] = password if config.has_option('roadnet_db', 'dbname'): dbname = config.get('roadnet_db', 'dbname') g_roadnet_db['db'] = dbname g_roadnet_pool.init_pool(g_roadnet_db) g_cross_delay_pool.init_pool(g_cross_delay_db) def test_get_cross_delay_data(): row_list = db_cross.query_cross_delay_info('CR_11987179_2632645', 350100, '20251013', 't830') data = row_list[0]['data'] cross_delay = pb.xl_cross_delayinfo_t() cross_delay.ParseFromString(data) print(MessageToJson(cross_delay, indent=None, always_print_fields_with_no_presence=True)) def check_err_data(): row_list = db_cross.check_err_data() prev_date = (datetime.now() - timedelta(days=1)).strftime('%Y%m%d') for row in row_list: if row['final_state'] == 6 or row['level_color'] == 4: continue first_date = row['first_date'] end_date = row['end_date'] if end_date: date_list = generate_date_range(first_date, end_date) else: date_list = generate_date_range(first_date, prev_date) phase_types = row['phase_type'] phase_details = row['phase_detail'] phase_type_list = phase_types.split('^') phase_detail_list = phase_details.split('^') if len(date_list) < len(phase_type_list): # new_first_date = (datetime.strptime(first_date, '%Y%m%d') - timedelta(days=1)).strftime('%Y%m%d') # ret = db_cross.update_err_data(row['crossid'], row['start_hm'], row['end_hm'], row['cont_times'], row['final_state'], row['level_color'], first_date, new_first_date) # if ret != 1: # print(row) print(len(date_list), len(phase_type_list), row['cont_times']) def gen_monitor_cross_ledger_info(): all_crosses = db_tmnet.query_cross_info_by_area_id(130201, 130200) cross_info_dict = {item['crossid']: item for item in all_crosses} crossid_list = [item['crossid'] for item in all_crosses] cross_roads_dir_dict = gen_crossids_roads_dir_dict_by_mysql(crossid_list, 130200) for crossid in crossid_list: cross_info_dict[crossid]['roads_dir_dict'] = cross_roads_dir_dict[crossid] reversed_lanes_crossid = db_tmnet.query_reversed_lanes_crossid(130200) res = [] has_right_light_crosses = db_tmnet.check_right_light(130200) right_list = [item['crossid'] for item in has_right_light_crosses] for crossid in crossid_list: cross_info = cross_info_dict[crossid] cross_name = cross_info['cross_name'] cross_no = cross_info['cross_no'] if cross_info['cross_no'] else '未配置' cross_type_code = cross_info['cross_type'] right_light = '无' if crossid in right_list: right_light = '有' if cross_type_code == 1: cross_type = '丁字路口' elif cross_type_code == 2: cross_type = '十字路口' elif cross_type_code == 3: cross_type = '行人过街路口' elif cross_type_code == 4: cross_type = '丁字路口' else: cross_type = '其他' internet_code = cross_info['internet'] if internet_code == 0: internet_name = '未配置' elif internet_code == 1: internet_name = '联网' elif internet_code == 2: internet_name = '脱机' else: internet_name = '未联网' company = cross_info['company'] if cross_info['company'] else '未配置' bicycle = '是' if cross_info['bicycle_twice_cross_street'] == 1 else '否' roads_dir_dict = cross_info['roads_dir_dict'] reversed_status = '是' if crossid in reversed_lanes_crossid else '否' outside_left, hold_straight, bus_lane, reverse_turn = '否', '否', '否', '否' for dir in roads_dir_dict: lane_info_list = roads_dir_dict[dir]['lane_turn_info'].split('|') if '17' in lane_info_list: bus_lane = '是' if roads_dir_dict[dir]['reverse_turn'] == 1: reverse_turn = '是' if roads_dir_dict[dir]['hold_straight'] == 1: hold_straight = '是' if '' not in lane_info_list and check_outside_left(lane_info_list): outside_left = '是' res.append((crossid, cross_name, cross_no, cross_type, internet_name, company, bicycle, reverse_turn, hold_straight, reversed_status, bus_lane, outside_left, right_light)) for item in res: print(item) return None if __name__ == '__main__': init() gen_monitor_cross_ledger_info()