cross_doctor/test.py

# -*- 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()