import yaml from typing import Dict, List import os import logging from pathlib import Path class MethodReader: """方法配置读取器""" def __init__(self): """初始化方法读取器""" self.logger = logging.getLogger(__name__) self.method_config = self._load_method_config() self.parameter_config = self._load_parameter_config() def _load_method_config(self) -> Dict: """加载方法配置文件""" try: config_path = Path('date_preprocessing/method.yaml') if not config_path.exists(): raise FileNotFoundError(f"Method config file not found at {config_path}") with open(config_path, 'r', encoding='utf-8') as f: config = yaml.safe_load(f) self.logger.info("Successfully loaded method config") return config except Exception as e: self.logger.error(f"Error loading method config: {str(e)}") raise def _load_parameter_config(self) -> Dict: """加载参数配置文件""" try: config_path = Path('date_preprocessing/parameter.yaml') if not config_path.exists(): raise FileNotFoundError(f"Parameter config file not found at {config_path}") with open(config_path, 'r', encoding='utf-8') as f: config = yaml.safe_load(f) self.logger.info("Successfully loaded parameter config") return config except Exception as e: self.logger.error(f"Error loading parameter config: {str(e)}") raise def get_preprocessing_methods(self) -> Dict: """获取预处理方法列表""" try: methods = [] # 数据缩放方法 scaler_methods = list(self.method_config.get('data_scaler_methods', {}).keys()) if scaler_methods: methods.append({ "name": "data_scaler", "description": "数据缩放处理", "method": scaler_methods }) # 缺失值处理方法 missing_methods = list(self.method_config.get('missing_value_handling_methods', {}).keys()) if missing_methods: methods.append({ "name": "missing_value_handler", "description": "缺失值处理", "method": missing_methods }) # 异常值检测方法 outlier_methods = list(self.method_config.get('outlier_detection_methods', {}).keys()) if outlier_methods: methods.append({ "name": "outlier_detector", "description": "异常值检测", "method": outlier_methods }) return { "status": "success", "methods": methods } except Exception as e: self.logger.error(f"Error getting preprocessing methods: {str(e)}") return { "status": "error", "error": str(e) } def get_method_details(self, method_name: str) -> Dict: """获取指定方法的详细信息""" try: # 在各个方法类别中查找方法原理和优缺点 method_info = None for category in ['data_scaler_methods', 'missing_value_handling_methods', 'outlier_detection_methods']: if method_name in self.method_config.get(category, {}): method_info = self.method_config[category][method_name] break if method_info is None: raise ValueError(f"Method {method_name} not found in method config") # 查找方法参数信息 parameter_info = None for category in ['data_scaler_methods', 'missing_value_handling_methods', 'outlier_detection_methods']: if method_name in self.parameter_config.get(category, {}): parameter_info = self.parameter_config[category][method_name] break if parameter_info is None: raise ValueError(f"Method {method_name} not found in parameter config") # 组合返回信息 return { "status": "success", "method": { "name": method_name, "description": parameter_info.get('description', ''), "principle": method_info.get('principle', ''), "advantages": method_info.get('advantages', []), "disadvantages": method_info.get('disadvantages', []), "applicable_scenarios": method_info.get('applicable_scenarios', []), "parameters": parameter_info.get('parameters', []) } } except Exception as e: self.logger.error(f"Error getting method details: {str(e)}") return { "status": "error", "error": str(e) }