234 lines
9.9 KiB
Python
234 lines
9.9 KiB
Python
# -*- 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
|
||
from app.task_worker import send_confirm_email
|
||
|
||
|
||
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('db', 'host'):
|
||
host = config.get('db', 'host')
|
||
g_dbinfo['host'] = host
|
||
if config.has_option('db', 'user'):
|
||
user = config.get('db', 'user')
|
||
g_dbinfo['user'] = user
|
||
if config.has_option('db', 'password'):
|
||
password = config.get('db', 'password')
|
||
g_dbinfo['password'] = password
|
||
if config.has_option('db', 'dbname'):
|
||
dbname = config.get('db', 'dbname')
|
||
g_dbinfo['db'] = dbname
|
||
|
||
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
|
||
|
||
if config.has_option('cloud_db', 'host'):
|
||
host = config.get('cloud_db', 'host')
|
||
g_cloud_db['host'] = host
|
||
if config.has_option('cloud_db', 'user'):
|
||
user = config.get('cloud_db', 'user')
|
||
g_cloud_db['user'] = user
|
||
if config.has_option('cloud_db', 'password'):
|
||
password = config.get('cloud_db', 'password')
|
||
g_cloud_db['password'] = password
|
||
if config.has_option('cloud_db', 'dbname'):
|
||
dbname = config.get('cloud_db', 'dbname')
|
||
g_cloud_db['db'] = dbname
|
||
if config.has_option('user_dbinfo', 'port'):
|
||
port = config.get('user_dbinfo', 'port')
|
||
g_user_db['port'] = port
|
||
if config.has_option('user_dbinfo', 'dbname'):
|
||
dbname = config.get('user_dbinfo', 'dbname')
|
||
g_user_db['db'] = dbname
|
||
if config.has_option('user_dbinfo', 'user'):
|
||
user = config.get('user_dbinfo', 'user')
|
||
g_user_db['user'] = user
|
||
if config.has_option('user_dbinfo', 'password'):
|
||
password = config.get('user_dbinfo', 'password')
|
||
g_user_db['password'] = password
|
||
if config.has_option('cross_delay_db', 'port'):
|
||
port = int(config.get('cross_delay_db', 'port'))
|
||
g_cross_delay_db['port'] = port
|
||
if config.has_option('cross_delay_db', 'dbname'):
|
||
dbname = config.get('cross_delay_db', 'dbname')
|
||
g_cross_delay_db['db'] = dbname
|
||
if config.has_option('cross_delay_db', 'user'):
|
||
user = config.get('cross_delay_db', 'user')
|
||
g_cross_delay_db['user'] = user
|
||
if config.has_option('cross_delay_db', 'password'):
|
||
password = config.get('cross_delay_db', 'password')
|
||
g_cross_delay_db['password'] = password
|
||
if config.has_option('task_db', 'host'):
|
||
host = config.get('task_db', 'host')
|
||
g_task_db['host'] = host
|
||
if config.has_option('task_db', 'port'):
|
||
port = int(config.get('task_db', 'port'))
|
||
g_task_db['port'] = port
|
||
if config.has_option('task_db', 'dbname'):
|
||
dbname = config.get('task_db', 'dbname')
|
||
g_task_db['db'] = dbname
|
||
if config.has_option('task_db', 'user'):
|
||
user = config.get('task_db', 'user')
|
||
g_task_db['user'] = user
|
||
if config.has_option('task_db', 'password'):
|
||
password = config.get('task_db', 'password')
|
||
g_task_db['password'] = password
|
||
|
||
# 新增绿波相关mysql链接信息
|
||
if config.has_option('wave_db', 'host'):
|
||
host = config.get('wave_db', 'host')
|
||
g_wave_tool_db['host'] = host
|
||
if config.has_option('wave_db', 'port'):
|
||
port = int(config.get('wave_db', 'port'))
|
||
g_wave_tool_db['port'] = port
|
||
if config.has_option('wave_db', 'dbname'):
|
||
dbname = config.get('wave_db', 'dbname')
|
||
g_wave_tool_db['db'] = dbname
|
||
if config.has_option('wave_db', 'user'):
|
||
user = config.get('wave_db', 'user')
|
||
g_wave_tool_db['user'] = user
|
||
if config.has_option('wave_db', 'password'):
|
||
password = config.get('wave_db', 'password')
|
||
g_wave_tool_db['password'] = password
|
||
g_db_pool.init_pool(g_dbinfo)
|
||
g_roadnet_pool.init_pool(g_roadnet_db)
|
||
g_cloud_pool.init_pool(g_cloud_db)
|
||
g_user_pool.init_pool(g_user_db)
|
||
g_cross_delay_pool.init_pool(g_cross_delay_db)
|
||
g_task_pool.init_pool(g_task_db)
|
||
g_wave_db.init_pool(g_wave_tool_db)
|
||
|
||
|
||
def test_get_cross_delay_data():
|
||
row_list = db_cross.query_cross_delay_info('CR_11902719_2542847', 350300, ['20260101'], 't700')
|
||
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()
|
||
send_confirm_email(['15836903493'], '当前任务到达阶段3，请及时处理', 12, '测试任务', 3, 130200, 130201, '创建人', '负责人') |