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, {})