cross_doctor/test.py

52 lines
2.1 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.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_350300.txt'
roadinfo_filename = 'data/xl_roadinfo_350300.txt'
roadlinks_filename = 'data/xl_roadlinks_350300.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)
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', '1200')
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))
if __name__ == '__main__':
init()
test_get_cross_delay_data()