343 lines
12 KiB
Python
343 lines
12 KiB
Python
import os
|
||
import json
|
||
import logging
|
||
|
||
class ConfigLoader:
|
||
"""配置加载器，用于加载和管理应用配置"""
|
||
|
||
_instance = None
|
||
|
||
@classmethod
|
||
def get_instance(cls):
|
||
"""获取单例实例"""
|
||
if cls._instance is None:
|
||
cls._instance = ConfigLoader()
|
||
return cls._instance
|
||
|
||
def __init__(self):
|
||
"""初始化配置加载器"""
|
||
self.config = {}
|
||
self.config_file = os.path.join('config', 'app_config.json')
|
||
|
||
# 创建配置目录（如果不存在）
|
||
os.makedirs('config', exist_ok=True)
|
||
|
||
# 加载配置
|
||
self.load_config()
|
||
|
||
def load_config(self):
|
||
"""加载配置文件"""
|
||
try:
|
||
if os.path.exists(self.config_file):
|
||
with open(self.config_file, 'r', encoding='utf-8') as f:
|
||
self.config = json.load(f)
|
||
logging.info(f"已加载配置文件: {self.config_file}")
|
||
|
||
# 检查并升级配置文件结构（兼容旧版本的配置）
|
||
self._upgrade_config_if_needed()
|
||
else:
|
||
# 创建默认配置
|
||
self.config = {
|
||
"app": {
|
||
"name": "腾智微丝产线包装系统",
|
||
"version": "1.0.0",
|
||
"features": {
|
||
"enable_serial_ports": False,
|
||
"enable_keyboard_listener": False,
|
||
"enable_camera": False
|
||
},
|
||
"mode": "standalone" # 默认为单机模式
|
||
},
|
||
"database": {
|
||
"current": "sqlite",
|
||
"sources": {
|
||
"sqlite": {
|
||
"path": "db/jtDB.db",
|
||
"description": "默认SQLite数据库"
|
||
},
|
||
"postgresql": {
|
||
"host": "localhost",
|
||
"port": "5432",
|
||
"user": "postgres",
|
||
"password": "",
|
||
"name": "jtDB",
|
||
"description": ""
|
||
},
|
||
"mysql": {
|
||
"host": "localhost",
|
||
"port": "3306",
|
||
"user": "root",
|
||
"password": "",
|
||
"name": "jtDB",
|
||
"description": ""
|
||
}
|
||
}
|
||
},
|
||
"camera": {
|
||
"enabled": False,
|
||
"default_exposure": 20000,
|
||
"default_gain": 10,
|
||
"default_framerate": 30
|
||
},
|
||
"modbus": {
|
||
"host": "localhost",
|
||
"port": "5020"
|
||
},
|
||
"serial": {
|
||
"keyboard": {
|
||
"trigger_key": "Key.page_up",
|
||
"enabled": False
|
||
}
|
||
}
|
||
}
|
||
|
||
# 保存默认配置
|
||
self.save_config()
|
||
logging.info(f"已创建默认配置文件: {self.config_file}")
|
||
except Exception as e:
|
||
logging.error(f"加载配置文件失败: {e}")
|
||
# 使用默认配置
|
||
self.config = {
|
||
"app": {
|
||
"name": "腾智微丝产线包装系统",
|
||
"version": "1.0.0",
|
||
"features": {
|
||
"enable_serial_ports": False,
|
||
"enable_keyboard_listener": False,
|
||
"enable_camera": False
|
||
},
|
||
"mode": "standalone" # 默认为单机模式
|
||
},
|
||
"database": {
|
||
"current": "sqlite",
|
||
"sources": {
|
||
"sqlite": {
|
||
"path": "db/jtDB.db",
|
||
"description": "默认SQLite数据库"
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
def _upgrade_config_if_needed(self):
|
||
"""升级配置文件结构（兼容旧版本的配置）"""
|
||
try:
|
||
# 检查数据库配置是否需要升级
|
||
if 'database' in self.config:
|
||
db_config = self.config['database']
|
||
|
||
# 旧版本配置结构检查
|
||
if 'type' in db_config and 'sources' not in db_config:
|
||
logging.info("检测到旧版本的数据库配置，正在升级...")
|
||
|
||
# 获取旧配置
|
||
db_type = db_config.get('type', 'sqlite')
|
||
|
||
# 创建新的配置结构
|
||
new_db_config = {
|
||
"default": db_type,
|
||
"sources": {}
|
||
}
|
||
|
||
# 转移旧配置到新结构
|
||
if db_type == 'sqlite':
|
||
new_db_config['sources']['sqlite'] = {
|
||
"path": db_config.get('path', 'db/jtDB.db'),
|
||
"description": "从旧版本升级的SQLite数据库"
|
||
}
|
||
else:
|
||
new_db_config['sources'][db_type] = {
|
||
"host": db_config.get('host', 'localhost'),
|
||
"port": db_config.get('port', '5432' if db_type == 'postgresql' else '3306'),
|
||
"user": db_config.get('user', ''),
|
||
"password": db_config.get('password', ''),
|
||
"name": db_config.get('name', 'jtDB'),
|
||
"description": f"从旧版本升级的{db_type}数据库"
|
||
}
|
||
|
||
# 确保至少有一个sqlite配置
|
||
if 'sqlite' not in new_db_config['sources']:
|
||
new_db_config['sources']['sqlite'] = {
|
||
"path": "db/jtDB.db",
|
||
"description": "默认SQLite数据库"
|
||
}
|
||
|
||
# 更新配置
|
||
self.config['database'] = new_db_config
|
||
self.save_config()
|
||
logging.info("数据库配置已成功升级")
|
||
|
||
# 将current改为default
|
||
if 'current' in db_config and 'default' not in db_config:
|
||
logging.info("将数据库配置中的'current'改为'default'...")
|
||
db_config['default'] = db_config.pop('current')
|
||
self.save_config()
|
||
|
||
# 将pgsql改为postgresql
|
||
if 'sources' in db_config and 'pgsql' in db_config['sources'] and 'postgresql' not in db_config['sources']:
|
||
logging.info("将数据库配置中的'pgsql'改为'postgresql'...")
|
||
db_config['sources']['postgresql'] = db_config['sources'].pop('pgsql')
|
||
# 如果默认值是pgsql，也更新
|
||
if db_config.get('default') == 'pgsql':
|
||
db_config['default'] = 'postgresql'
|
||
self.save_config()
|
||
|
||
# 检查并添加运行模式配置
|
||
if 'app' in self.config and 'mode' not in self.config['app']:
|
||
logging.info("添加运行模式配置...")
|
||
self.config['app']['mode'] = 'standalone'
|
||
self.save_config()
|
||
|
||
except Exception as e:
|
||
logging.error(f"升级配置文件结构失败: {e}")
|
||
|
||
def save_config(self):
|
||
"""保存配置到文件"""
|
||
try:
|
||
with open(self.config_file, 'w', encoding='utf-8') as f:
|
||
json.dump(self.config, f, indent=4, ensure_ascii=False)
|
||
logging.info(f"已保存配置文件: {self.config_file}")
|
||
return True
|
||
except Exception as e:
|
||
logging.error(f"保存配置文件失败: {e}")
|
||
return False
|
||
|
||
def get_value(self, key_path, default=None):
|
||
"""
|
||
获取配置值
|
||
|
||
Args:
|
||
key_path: 配置键路径，例如 "app.features.enable_serial_ports"
|
||
default: 默认值，如果配置项不存在则返回此值
|
||
|
||
Returns:
|
||
配置值或默认值
|
||
"""
|
||
keys = key_path.split('.')
|
||
value = self.config
|
||
|
||
try:
|
||
for key in keys:
|
||
value = value[key]
|
||
return value
|
||
except (KeyError, TypeError):
|
||
return default
|
||
|
||
def set_value(self, key_path, value):
|
||
"""
|
||
设置配置值
|
||
|
||
Args:
|
||
key_path: 配置键路径，例如 "app.features.enable_serial_ports"
|
||
value: 要设置的值
|
||
|
||
Returns:
|
||
bool: 是否设置成功
|
||
"""
|
||
keys = key_path.split('.')
|
||
config = self.config
|
||
|
||
# 遍历路径中的所有键，除了最后一个
|
||
for key in keys[:-1]:
|
||
if key not in config:
|
||
config[key] = {}
|
||
config = config[key]
|
||
|
||
# 设置最后一个键的值
|
||
config[keys[-1]] = value
|
||
|
||
# 保存配置
|
||
return self.save_config()
|
||
|
||
def get_app_mode(self):
|
||
"""
|
||
获取应用运行模式
|
||
|
||
Returns:
|
||
str: 'standalone'(单机模式) 或 'api'(接口模式)
|
||
"""
|
||
return self.get_value('app.mode', 'standalone')
|
||
|
||
def set_app_mode(self, mode):
|
||
"""
|
||
设置应用运行模式
|
||
|
||
Args:
|
||
mode: 'standalone'(单机模式) 或 'api'(接口模式)
|
||
|
||
Returns:
|
||
bool: 是否设置成功
|
||
"""
|
||
if mode not in ['standalone', 'api']:
|
||
logging.error(f"无效的运行模式: {mode}，必须是 'standalone' 或 'api'")
|
||
return False
|
||
|
||
return self.set_value('app.mode', mode)
|
||
|
||
def is_standalone_mode(self):
|
||
"""
|
||
检查是否为单机模式
|
||
|
||
Returns:
|
||
bool: True表示单机模式，False表示接口模式
|
||
"""
|
||
return self.get_app_mode() == 'standalone'
|
||
|
||
def is_api_mode(self):
|
||
"""
|
||
检查是否为接口模式
|
||
|
||
Returns:
|
||
bool: True表示接口模式，False表示单机模式
|
||
"""
|
||
return self.get_app_mode() == 'api'
|
||
|
||
def get_config(self, key):
|
||
"""
|
||
获取serial配置下的指定配置
|
||
|
||
Args:
|
||
key: 配置键，例如'mdz', 'cz'等
|
||
|
||
Returns:
|
||
dict: 配置数据
|
||
"""
|
||
serial_config = self.config.get('serial', {})
|
||
return serial_config.get(key, {})
|
||
|
||
def set_config(self, key, config_data):
|
||
"""
|
||
设置serial配置下的指定配置
|
||
|
||
Args:
|
||
key: 配置键，例如'mdz', 'cz'等
|
||
config_data: 配置数据
|
||
|
||
Returns:
|
||
bool: 是否设置成功
|
||
"""
|
||
if 'serial' not in self.config:
|
||
self.config['serial'] = {}
|
||
|
||
self.config['serial'][key] = config_data
|
||
return self.save_config()
|
||
|
||
def get_database_config(self, db_type=None):
|
||
"""
|
||
获取数据库配置
|
||
|
||
Args:
|
||
db_type: 数据库类型，如果为None则返回当前使用的数据库配置
|
||
|
||
Returns:
|
||
dict: 数据库配置
|
||
"""
|
||
db_config = self.config.get('database', {})
|
||
db_sources = db_config.get('sources', {})
|
||
|
||
# 如果未指定数据库类型，则使用当前配置的数据库类型
|
||
if db_type is None:
|
||
db_type = db_config.get('current', 'sqlite')
|
||
|
||
# 返回指定类型的数据库配置
|
||
return db_sources.get(db_type, {}) |