135 lines
5.3 KiB
Python
135 lines
5.3 KiB
Python
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)
|
|
} |