179 lines
8.0 KiB
Python
179 lines
8.0 KiB
Python
import os
|
||
from loguru import logger
|
||
import threading
|
||
from midware.config.excel_config import load_csv_config
|
||
|
||
'''
|
||
3. 基础配置模块（midware/config/base_config.py）
|
||
定义全局配置常量，初始化外设、UDP、总线基础参数，预留配置动态加载接口：
|
||
'''
|
||
# 基础配置 - UDP（故障注入软件交互）
|
||
UDP_CONFIG = {
|
||
"LOCAL_IP": "10.20.48.138", # 本地UDP地址，用于回环测试,自测时候是"127.0.0.1"，科学所是"192.168.0.150"","10.20.48.138"
|
||
"LOCAL_PORT": 8888, # 本地UDP端口
|
||
"FAULT_INJECT_IP": "10.20.48.129",# 故障注入软件IP，用于回环测试,自测时候是"127.0.0.1"，科学所是"192.168.0.130"","10.20.48.129"
|
||
"FAULT_INJECT_PORT": 18889, # 故障注入软件端口
|
||
"BUFFER_SIZE": 4096 # UDP缓冲区大小
|
||
}
|
||
|
||
# 基础配置 - 外设总线（支持UART/CAN/1553B/AD/OC）
|
||
BUS_CONFIG = {
|
||
"SUPPORT_BUS": ["UART", "CAN", "1553B", "AD", "OC"],
|
||
"MAX_DEVICE_NUM": 10, # 最大支持10个单机
|
||
"WRITE_BYTE_UART": 2, # UART每次写入2字节
|
||
"WRITE_BYTE_CAN": 2, # CAN每次写入2字节
|
||
"READ_BYTE_CAN": 1, # CAN每次读取1字节
|
||
"WRITE_BYTE_1553B": 2, # 1553B每次写入2字节
|
||
"WRITE_BYTE_AD": 2, # AD每次写入2字节
|
||
"READ_BYTE_OC": 1 # OC每次读取1字节
|
||
}
|
||
|
||
# 基础配置 - UART芯片寄存器（初始写死，后续由chip_config.ini加载）
|
||
UART_CHIP_CONFIG = {
|
||
"TBR": 0x00, # 发送FIFO
|
||
"FIFO_STATUS": 0x04, # FIFO状态寄存器
|
||
"FRAME_COUNT": 0x08, # 帧计数
|
||
"FIFO_REMAIN": 0x0C, # FIFO剩余字节数
|
||
"CLOCK_CONFIG": 0x10, # 时钟配置
|
||
"SCRAMBLE": 0x14, # 加解扰使能
|
||
"RESET": 0x7C # 复位
|
||
}
|
||
|
||
# 全局外设配置字典，存储已启用的外设信息
|
||
# ==================== 全局配置（唯一数据源，仅在此文件定义）====================
|
||
# 初始化为空字典，由init_base_config()初始化
|
||
DEVICE_CONFIG_DICT = {} #{1: {}, 2: {}, 3: {}, 4: {}, 5: {}, 6: {}, 7: {}, 8: {}, 9: {}, 10: {}}
|
||
|
||
# 配置读写锁（保证多线程安全，防止读写冲突）
|
||
CONFIG_LOCK = threading.RLock()
|
||
|
||
# ==================== 日志初始化（工程全局）====================
|
||
def init_logger():
|
||
os.makedirs("logs", exist_ok=True)
|
||
# 运行日志+错误日志分离
|
||
logger.add("logs/run_{time:YYYYMMDDHHmmss}.log", level="INFO",
|
||
format="{time:YYYY-MM-DD HH:mm:ss} | {level} | {message}")
|
||
logger.add("logs/error_{time:YYYYMMDDHHmmss}.log", level="ERROR",
|
||
format="{time:YYYY-MM-DD HH:mm:ss} | {level} | {file}:{line} | {message}")
|
||
# 工程启动时自动初始化日志
|
||
init_logger()
|
||
|
||
'''
|
||
# 加载xlsx文件
|
||
def init_base_config():
|
||
"""初始化基础配置，加载默认外设配置"""
|
||
try:
|
||
# 加载默认Excel配置（后续实现）
|
||
from midware.config.excel_config import load_excel_config
|
||
default_excel = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "config_files/device_config.xlsx")
|
||
if os.path.exists(default_excel):
|
||
DEVICE_CONFIG_DICT = load_excel_config(default_excel)
|
||
logger.info(f"加载默认Excel配置，共{len(DEVICE_CONFIG_DICT)}个外设")
|
||
else:
|
||
logger.warning(f"默认配置文件不存在：{default_excel}，使用空配置")
|
||
logger.info("基础配置初始化完成")
|
||
except Exception as e:
|
||
logger.error(f"基础配置初始化失败：{str(e)}", exc_info=True)
|
||
raise e
|
||
'''
|
||
|
||
'''
|
||
# midware/config/base_config.py 中init_base_config函数
|
||
def init_base_config():
|
||
"""初始化基础配置，加载默认外设配置"""
|
||
global DEVICE_CONFIG_DICT
|
||
try:
|
||
# 替换为CSV加载方法
|
||
#from midware.config.excel_config import load_csv_config
|
||
default_csv = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "config_files/device_config.csv")
|
||
if os.path.exists(default_csv):
|
||
#global DEVICE_CONFIG_DICT
|
||
DEVICE_CONFIG_DICT = load_csv_config(default_csv)
|
||
logger.info(f"加载默认CSV配置，共{len(DEVICE_CONFIG_DICT)}个外设")
|
||
else:
|
||
logger.warning(f"默认配置文件不存在：{default_csv}，使用空配置")
|
||
DEVICE_CONFIG_DICT = {}
|
||
logger.info("基础配置初始化完成")
|
||
except Exception as e:
|
||
logger.error(f"基础配置初始化失败：{str(e)}", exc_info=True)
|
||
DEVICE_CONFIG_DICT = {}
|
||
#raise e #注释掉这行，避免程序崩溃
|
||
|
||
def update_device_config(new_config: dict):
|
||
"""
|
||
对外提供配置更新方法（如UI勾选外设、导入新配置时调用）
|
||
:param new_config: 新的外设配置字典，格式与DEVICE_CONFIG_DICT一致
|
||
"""
|
||
global DEVICE_CONFIG_DICT
|
||
if isinstance(new_config, dict):
|
||
DEVICE_CONFIG_DICT = new_config
|
||
logger.info(f"外设配置已更新，当前共{len(DEVICE_CONFIG_DICT)}个外设")
|
||
# 配置更新后，自动触发UDP套接字重新加载（关键：保证UDP端口与配置同步）
|
||
from midware.network.udp_handler import udp_server
|
||
udp_server.reload_sockets()
|
||
else:
|
||
logger.error("配置更新失败：新配置非字典类型")
|
||
# 工程启动时自动初始化日志
|
||
#init_logger()
|
||
'''
|
||
|
||
# ==================== 配置核心方法（对外暴露，唯一入口）====================
|
||
def init_base_config(csv_path: str = None):
|
||
"""
|
||
初始化外设配置：工程启动时**唯一调用**，加载CSV配置到DEVICE_CONFIG_DICT
|
||
:param csv_path: 配置文件路径，默认取工程根目录config_files/device_config.csv
|
||
"""
|
||
global DEVICE_CONFIG_DICT # 声明修改全局变量
|
||
if csv_path is None:
|
||
# 自动拼接工程根目录的配置文件路径
|
||
csv_path = os.path.join(
|
||
os.path.dirname(os.path.dirname(os.path.dirname(__file__))),
|
||
"config_files/device_config.csv"
|
||
)
|
||
# 加锁保证初始化原子性，防止多线程同时修改
|
||
with CONFIG_LOCK:
|
||
try:
|
||
if os.path.exists(csv_path):
|
||
# 加载CSV配置并覆盖全局变量
|
||
DEVICE_CONFIG_DICT = load_csv_config(csv_path)
|
||
logger.info(f"配置初始化成功｜加载{len(DEVICE_CONFIG_DICT)}个外设｜配置文件：{csv_path}")
|
||
else:
|
||
DEVICE_CONFIG_DICT = {}
|
||
logger.error(f"配置初始化失败｜配置文件不存在：{csv_path}")
|
||
except Exception as e:
|
||
DEVICE_CONFIG_DICT = {}
|
||
logger.error(f"配置初始化异常｜{str(e)}", exc_info=True)
|
||
return DEVICE_CONFIG_DICT
|
||
|
||
def update_device_config(new_config: dict):
|
||
"""
|
||
唯一配置更新入口：修改配置后自动同步UDP端口
|
||
【关键】延迟导入udp_server，避免模块顶层循环导入
|
||
"""
|
||
global DEVICE_CONFIG_DICT
|
||
with CONFIG_LOCK:
|
||
if not isinstance(new_config, dict):
|
||
logger.error("配置更新失败｜新配置非字典类型")
|
||
return False
|
||
# 覆盖全局配置
|
||
DEVICE_CONFIG_DICT = new_config.copy() # 深拷贝，防止外部修改源字典
|
||
logger.info(f"配置更新成功｜当前外设数量：{len(DEVICE_CONFIG_DICT)}")
|
||
# 延迟导入：仅在函数内导入，避免模块加载时循环依赖
|
||
try:
|
||
from midware.network.udp_handler import udp_server
|
||
udp_server.reload_sockets(new_config) # 传参同步配置，不依赖全局引用
|
||
except ImportError as e:
|
||
logger.warning(f"UDP配置同步失败｜未找到UDP模块：{str(e)}")
|
||
return True
|
||
|
||
def get_device_config():
|
||
"""
|
||
全局配置**只读接口**：所有其他模块必须通过此方法获取配置
|
||
返回副本，防止外部修改原全局变量导致数据混乱
|
||
"""
|
||
with CONFIG_LOCK:
|
||
return DEVICE_CONFIG_DICT.copy() # 返回深拷贝，彻底隔离原变量
|
||
|
||
def get_udp_config():
|
||
"""UDP基础配置只读接口"""
|
||
return UDP_CONFIG.copy() |