BodyBalanceEvaluation/backend/devices/utils/config_manager.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
配置管理工具
提供设备配置的统一管理和读取
"""

import os
import configparser
import json
import logging
from typing import Dict, Any, Optional, Union
from pathlib import Path
import sys

class ConfigManager:
    """配置管理器"""
    
    def __init__(self, config_path: Optional[str] = None):
        """
        初始化配置管理器
        
        Args:
            config_path: 配置文件路径，默认为backend/config.ini
        """
        self.logger = logging.getLogger(f"{__name__}.ConfigManager")
        
        # 确定配置文件路径
        if config_path:
            self.config_path = config_path
        else:
            # 自动查找配置文件
            self.config_path = self._find_config_file()
            
        self.config = configparser.ConfigParser()
        self._device_configs = {}
        self._load_config()
        
    def _find_config_file(self) -> str:
        """
        查找配置文件
        
        Returns:
            str: 配置文件路径
        """
        import sys
        
        # 可能的配置文件路径
        possible_paths = []
        
        # 如果是打包后的exe文件，从exe文件同级目录获取
        if getattr(sys, 'frozen', False):
            # 打包后的exe文件路径
            exe_dir = os.path.dirname(sys.executable)
            possible_paths.append(os.path.join(exe_dir, 'config.ini'))
        
        # 开发环境下的路径
        possible_paths.extend([
            os.path.join(os.path.dirname(__file__), 'config.ini'),
            os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), 'config.ini'),  # backend/config.ini
            os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(__file__)))), 'config.ini')  # 项目根目录/config.ini
        ])
        
        for path in possible_paths:
            abs_path = os.path.abspath(path)
            if os.path.exists(abs_path):
                self.logger.info(f"找到配置文件: {abs_path}")
                return abs_path
        
        # 如果都找不到，返回默认路径（exe同级目录或backend目录）
        if getattr(sys, 'frozen', False):
            default_path = os.path.join(os.path.dirname(sys.executable), 'config.ini')
        else:
            default_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), 'config.ini')
        
        self.logger.warning(f"未找到配置文件，使用默认路径: {default_path}")
        return default_path
    def _load_config(self):
        """
        加载配置文件
        """
        try:
            if os.path.exists(self.config_path):
                self.config.read(self.config_path, encoding='utf-8')
                self.logger.info(f"成功加载配置文件: {self.config_path}")
            else:
                self.logger.warning(f"配置文件不存在: {self.config_path}")
                self._create_default_config()
        except Exception as e:
            self.logger.error(f"加载配置文件失败: {e}")
            self._create_default_config()
            
    def _create_default_config(self):
        """
        创建默认配置
        """
        self.config.clear()
        
        # 默认设备配置
        self.config['DEVICES'] = {
            'imu_port': 'COM7',
            'imu_baudrate': '9600',
            'pressure_port': 'COM8',
            'pressure_baudrate': '115200'
        }
        
        # 默认相机配置
        self.config['CAMERA'] = {
            'device_index': '0',
            'width': '1280',
            'height': '720',
            'fps': '30'
        }
        
        # 默认FemtoBolt配置
        self.config['FEMTOBOLT'] = {
            'color_resolution': '1080P',
            'depth_mode': 'NFOV_UNBINNED',
            'fps': '15',
            'depth_range_min': '500',
            'depth_range_max': '4500'
        }
        
        # 默认系统配置
        self.config['SYSTEM'] = {
            'log_level': 'INFO',
            'max_cache_size': '10',
            'cache_timeout': '5.0'
        }
        
        self.logger.info("创建默认配置")
        
    def get_device_config(self, device_name: str) -> Dict[str, Any]:
        """
        获取设备配置
        
        Args:
            device_name: 设备名称 (camera, femtobolt, imu, pressure)
            
        Returns:
            Dict[str, Any]: 设备配置字典
        """
        if device_name in self._device_configs:
            return self._device_configs[device_name].copy()
            
        config = {}
        
        if device_name == 'camera':
            config = self._get_camera_config()
        elif device_name == 'femtobolt':
            config = self._get_femtobolt_config()
        elif device_name == 'imu':
            config = self._get_imu_config()
        elif device_name == 'pressure':
            config = self._get_pressure_config()
        else:
            self.logger.warning(f"未知设备类型: {device_name}")
            
        # 缓存配置
        self._device_configs[device_name] = config
        return config.copy()
        
    def _get_camera_config(self) -> Dict[str, Any]:
        """
        获取相机配置
        
        Returns:
            Dict[str, Any]: 相机配置
        """
        return {
            'device_index': self.config.getint('CAMERA', 'device_index', fallback=0),
            'width': self.config.getint('CAMERA', 'width', fallback=1280),
            'height': self.config.getint('CAMERA', 'height', fallback=720),
            'fps': self.config.getint('CAMERA', 'fps', fallback=30),
            'buffer_size': self.config.getint('CAMERA', 'buffer_size', fallback=1),
            'fourcc': self.config.get('CAMERA', 'fourcc', fallback='MJPG')
        }
        
    def _get_femtobolt_config(self) -> Dict[str, Any]:
        """
        获取FemtoBolt配置
        
        Returns:
            Dict[str, Any]: FemtoBolt配置
        """
        return {
            'color_resolution': self.config.get('FEMTOBOLT', 'color_resolution', fallback='1080P'),
            'depth_mode': self.config.get('FEMTOBOLT', 'depth_mode', fallback='NFOV_UNBINNED'),
            'fps': self.config.getint('FEMTOBOLT', 'fps', fallback=15),
            'depth_range_min': self.config.getint('FEMTOBOLT', 'depth_range_min', fallback=500),
            'depth_range_max': self.config.getint('FEMTOBOLT', 'depth_range_max', fallback=4500),
            'synchronized_images_only': self.config.getboolean('FEMTOBOLT', 'synchronized_images_only', fallback=False)
        }
        
    def _get_imu_config(self) -> Dict[str, Any]:
        """
        获取IMU配置
        
        Returns:
            Dict[str, Any]: IMU配置
        """
        return {
            'device_type': self.config.get('DEVICES', 'imu_device_type', fallback='mock'),
            'port': self.config.get('DEVICES', 'imu_port', fallback='COM7'),
            'baudrate': self.config.getint('DEVICES', 'imu_baudrate', fallback=9600),
            'timeout': self.config.getfloat('DEVICES', 'imu_timeout', fallback=1.0),
            'calibration_samples': self.config.getint('DEVICES', 'imu_calibration_samples', fallback=100),
        }
        
    def _get_pressure_config(self) -> Dict[str, Any]:
        """
        获取压力传感器配置
        
        Returns:
            Dict[str, Any]: 压力传感器配置
        """
        return {
            'device_type': self.config.get('DEVICES', 'pressure_device_type', fallback='mock'),
            'port': self.config.get('DEVICES', 'pressure_port', fallback='COM8'),
            'baudrate': self.config.getint('DEVICES', 'pressure_baudrate', fallback=115200),
            'timeout': self.config.getfloat('DEVICES', 'pressure_timeout', fallback=1.0),
            'calibration_samples': self.config.getint('DEVICES', 'pressure_calibration_samples', fallback=50)
        }
        
    def get_system_config(self) -> Dict[str, Any]:
        """
        获取系统配置
        
        Returns:
            Dict[str, Any]: 系统配置
        """
        return {
            'log_level': self.config.get('SYSTEM', 'log_level', fallback='INFO'),
            'max_cache_size': self.config.getint('SYSTEM', 'max_cache_size', fallback=10),
            'cache_timeout': self.config.getfloat('SYSTEM', 'cache_timeout', fallback=5.0),
            'heartbeat_interval': self.config.getfloat('SYSTEM', 'heartbeat_interval', fallback=30.0)
        }
        
    def get_config_value(self, section: str, key: str, fallback: Any = None) -> Any:
        """
        获取配置值
        
        Args:
            section: 配置段
            key: 配置键
            fallback: 默认值
            
        Returns:
            Any: 配置值
        """
        try:
            if isinstance(fallback, bool):
                return self.config.getboolean(section, key, fallback=fallback)
            elif isinstance(fallback, int):
                return self.config.getint(section, key, fallback=fallback)
            elif isinstance(fallback, float):
                return self.config.getfloat(section, key, fallback=fallback)
            else:
                return self.config.get(section, key, fallback=fallback)
        except Exception as e:
            self.logger.warning(f"获取配置值失败 [{section}][{key}]: {e}")
            return fallback
            
    def set_config_value(self, section: str, key: str, value: Any):
        """
        设置配置值
        
        Args:
            section: 配置段
            key: 配置键
            value: 配置值
        """
        if not self.config.has_section(section):
            self.config.add_section(section)
            
        self.config.set(section, key, str(value))
        
        # 清除缓存
        self._device_configs.clear()
        
    def save_config(self):
        """
        保存配置到文件
        """
        try:
            # 确保目录存在
            os.makedirs(os.path.dirname(self.config_path), exist_ok=True)
            
            with open(self.config_path, 'w', encoding='utf-8') as f:
                self.config.write(f)
                
            self.logger.info(f"配置已保存到: {self.config_path}")
        except Exception as e:
            self.logger.error(f"保存配置失败: {e}")
            
    def reload_config(self):
        """
        重新加载配置
        """
        self._device_configs.clear()
        self._load_config()
        self.logger.info("配置已重新加载")
        
    def get_all_sections(self) -> Dict[str, Dict[str, str]]:
        """
        获取所有配置段
        
        Returns:
            Dict[str, Dict[str, str]]: 所有配置
        """
        result = {}
        for section_name in self.config.sections():
            result[section_name] = dict(self.config[section_name])
        return result
        
    def validate_config(self) -> Dict[str, list]:
        """
        验证配置有效性
        
        Returns:
            Dict[str, list]: 验证结果，包含错误和警告
        """
        errors = []
        warnings = []
        
        # 验证必需的配置段
        required_sections = ['DEVICES', 'CAMERA', 'FEMTOBOLT', 'SYSTEM']
        for section in required_sections:
            if not self.config.has_section(section):
                errors.append(f"缺少必需的配置段: {section}")
                
        # 验证设备配置
        try:
            imu_config = self.get_device_config('imu')
            if not imu_config.get('port'):
                warnings.append("IMU串口未配置")
        except Exception as e:
            errors.append(f"IMU配置验证失败: {e}")
            
        return {
            'errors': errors,
            'warnings': warnings,
            'valid': len(errors) == 0
        }
        
    # HTTP接口设备参数设置方法
    def set_imu_config(self, config_data: Dict[str, Any]) -> Dict[str, Any]:
        """
        设置IMU设备配置
        
        Args:
            config_data: IMU配置数据
            {
                'device_type': 'real' | 'mock',
                'port': 'COM6',
                'baudrate': 9600
            }
            
        Returns:
            Dict[str, Any]: 设置结果
        """
        try:
            # 验证必需参数
            if 'device_type' in config_data:
                self.set_config_value('DEVICES', 'imu_device_type', config_data['device_type'])
            if 'port' in config_data:
                self.set_config_value('DEVICES', 'imu_port', config_data['port'])
            if 'baudrate' in config_data:
                self.set_config_value('DEVICES', 'imu_baudrate', str(config_data['baudrate']))
                
            # 保存配置
            self.save_config()
            
            self.logger.info(f"IMU配置已更新: {config_data}")
            return {
                'success': True,
                'message': 'IMU配置更新成功',
                'config': self.get_device_config('imu')
            }
        except Exception as e:
            self.logger.error(f"设置IMU配置失败: {e}")
            return {
                'success': False,
                'message': f'设置IMU配置失败: {str(e)}'
            }
            
    def set_pressure_config(self, config_data: Dict[str, Any]) -> Dict[str, Any]:
        """
        设置压力板设备配置
        
        Args:
            config_data: 压力板配置数据
            {
                'device_type': 'real' | 'mock',
                'use_mock': False,
                'port': 'COM5',
                'baudrate': 115200
            }
            
        Returns:
            Dict[str, Any]: 设置结果
        """
        try:
            # 验证必需参数
            if 'device_type' in config_data:
                self.set_config_value('DEVICES', 'pressure_device_type', config_data['device_type'])
            if 'use_mock' in config_data:
                self.set_config_value('DEVICES', 'pressure_use_mock', str(config_data['use_mock']))
            if 'port' in config_data:
                self.set_config_value('DEVICES', 'pressure_port', config_data['port'])
            if 'baudrate' in config_data:
                self.set_config_value('DEVICES', 'pressure_baudrate', str(config_data['baudrate']))
                
            # 保存配置
            self.save_config()
            
            self.logger.info(f"压力板配置已更新: {config_data}")
            return {
                'success': True,
                'message': '压力板配置更新成功',
                'config': self.get_device_config('pressure')
            }
        except Exception as e:
            self.logger.error(f"设置压力板配置失败: {e}")
            return {
                'success': False,
                'message': f'设置压力板配置失败: {str(e)}'
            }
            
    def set_camera_config(self, config_data: Dict[str, Any]) -> Dict[str, Any]:
        """
        设置相机设备配置
        
        Args:
            config_data: 相机配置数据
            {
                'device_index': 1,
                'width': 1280,
                'height': 720,
                'fps': 30
            }
            
        Returns:
            Dict[str, Any]: 设置结果
        """
        try:
            # 验证必需参数
            if 'device_index' in config_data:
                self.set_config_value('CAMERA', 'device_index', str(config_data['device_index']))
            if 'width' in config_data:
                self.set_config_value('CAMERA', 'width', str(config_data['width']))
            if 'height' in config_data:
                self.set_config_value('CAMERA', 'height', str(config_data['height']))
            if 'fps' in config_data:
                self.set_config_value('CAMERA', 'fps', str(config_data['fps']))
                
            # 保存配置
            self.save_config()
            
            self.logger.info(f"相机配置已更新: {config_data}")
            return {
                'success': True,
                'message': '相机配置更新成功',
                'config': self.get_device_config('camera')
            }
        except Exception as e:
            self.logger.error(f"设置相机配置失败: {e}")
            return {
                'success': False,
                'message': f'设置相机配置失败: {str(e)}'
            }
            
    def set_femtobolt_config(self, config_data: Dict[str, Any]) -> Dict[str, Any]:
        """
        设置FemtoBolt设备配置
        
        Args:
            config_data: FemtoBolt配置数据
            {
                'color_resolution': '1080P',
                'depth_mode': 'NFOV_UNBINNED',
                'fps': 30,
                'depth_range_min': 1200,
                'depth_range_max': 1500
            }
            
        Returns:
            Dict[str, Any]: 设置结果
        """
        try:
            # 验证必需参数
            if 'color_resolution' in config_data:
                self.set_config_value('FEMTOBOLT', 'color_resolution', config_data['color_resolution'])
            if 'depth_mode' in config_data:
                self.set_config_value('FEMTOBOLT', 'depth_mode', config_data['depth_mode'])
            if 'fps' in config_data:
                self.set_config_value('FEMTOBOLT', 'fps', str(config_data['fps']))
            if 'depth_range_min' in config_data:
                self.set_config_value('FEMTOBOLT', 'depth_range_min', str(config_data['depth_range_min']))
            if 'depth_range_max' in config_data:
                self.set_config_value('FEMTOBOLT', 'depth_range_max', str(config_data['depth_range_max']))
                
            # 保存配置
            self.save_config()
            
            self.logger.info(f"FemtoBolt配置已更新: {config_data}")
            return {
                'success': True,
                'message': 'FemtoBolt配置更新成功',
                'config': self.get_device_config('femtobolt')
            }
        except Exception as e:
            self.logger.error(f"设置FemtoBolt配置失败: {e}")
            return {
                'success': False,
                'message': f'设置FemtoBolt配置失败: {str(e)}'
            }
            
    def get_all_device_configs(self) -> Dict[str, Any]:
        """
        获取所有设备配置
        
        Returns:
            Dict[str, Any]: 所有设备配置
        """
        return {
            'imu': self.get_device_config('imu'),
            'pressure': self.get_device_config('pressure'),
            'camera': self.get_device_config('camera'),
            'femtobolt': self.get_device_config('femtobolt')
        }