zhaoyuan/scripts/evaluator_optimized.py

# evaluation_engine.py
import sys
import warnings
import time
import importlib
import yaml  # 添加yaml模块导入
from pathlib import Path
import argparse
from concurrent.futures import ThreadPoolExecutor
from functools import lru_cache
from typing import Dict, Any, List, Optional
from datetime import datetime

# 强制导入所有可能动态加载的模块



# 安全设置根目录路径（动态路径管理）
# 判断是否处于编译模式
if hasattr(sys, "_MEIPASS"):
    # 编译模式下使用临时资源目录
    _ROOT_PATH = Path(sys._MEIPASS)
else:
    # 开发模式下使用原工程路径
    _ROOT_PATH = Path(__file__).resolve().parent.parent

sys.path.insert(0, str(_ROOT_PATH))
print(f"当前根目录：{_ROOT_PATH}")
print(f'当前系统路径：{sys.path}')


class EvaluationCore:
    """评估引擎核心类（单例模式）"""

    _instance = None

    def __new__(cls, logPath: str, configPath: str = None, customConfigPath: str = None, customMetricsPath: str = None):
        if not cls._instance:
            cls._instance = super().__new__(cls)
            cls._instance._init(logPath, configPath, customConfigPath, customMetricsPath)
        return cls._instance

    def _init(self, logPath: str = None, configPath: str = None, customConfigPath: str = None, customMetricsPath: str = None) -> None:
        """初始化引擎组件"""
        self.log_path = logPath
        self.config_path = configPath
        self.custom_config_path = customConfigPath
        self.custom_metrics_path = customMetricsPath
        
        # 加载配置
        self.metrics_config = {}
        self.custom_metrics_config = {}
        self.merged_config = {}  # 添加合并后的配置
        
        # 自定义指标脚本模块
        self.custom_metrics_modules = {}
        
        self._init_log_system()
        self._load_configs()  # 加载并合并配置
        self._init_metrics()
        self._load_custom_metrics()

    def _init_log_system(self) -> None:
        """集中式日志管理"""
        try:
            from modules.lib.log_manager import LogManager

            log_manager = LogManager(self.log_path)
            self.logger = log_manager.get_logger()
        except (PermissionError, IOError) as e:
            sys.stderr.write(f"日志系统初始化失败: {str(e)}\n")
            sys.exit(1)

    def _init_metrics(self) -> None:
        """初始化评估模块（策略模式）"""
        # from modules.metric import safety, comfort, traffic, efficient, function
        self.metric_modules = {
            "safety": self._load_module("modules.metric.safety", "SafeManager"),
            "comfort": self._load_module("modules.metric.comfort", "ComfortManager"),
            "traffic": self._load_module("modules.metric.traffic", "ViolationManager"),
            "efficient": self._load_module("modules.metric.efficient", "EfficientManager"),
            "function": self._load_module("modules.metric.function", "FunctionManager"),
        }

    @lru_cache(maxsize=32)
    def _load_module(self, module_path: str, class_name: str) -> Any:
        """动态加载评估模块（带缓存）"""
        try:
            __import__(module_path)
            return getattr(sys.modules[module_path], class_name)
        except (ImportError, AttributeError) as e:
            self.logger.error(f"模块加载失败: {module_path}.{class_name} - {str(e)}")
            raise

    def _load_configs(self) -> None:
        """加载并合并内置指标和自定义指标配置"""
        # 加载内置指标配置
        if self.config_path and Path(self.config_path).exists():
            try:
                with open(self.config_path, 'r', encoding='utf-8') as f:
                    self.metrics_config = yaml.safe_load(f)
                self.logger.info(f"成功加载内置指标配置: {self.config_path}")
            except Exception as e:
                self.logger.error(f"加载内置指标配置失败: {str(e)}")
                self.metrics_config = {}
        
        # 加载自定义指标配置
        if self.custom_config_path and Path(self.custom_config_path).exists():
            try:
                with open(self.custom_config_path, 'r', encoding='utf-8') as f:
                    self.custom_metrics_config = yaml.safe_load(f)
                self.logger.info(f"成功加载自定义指标配置: {self.custom_config_path}")
            except Exception as e:
                self.logger.error(f"加载自定义指标配置失败: {str(e)}")
                self.custom_metrics_config = {}
        
        # 合并配置
        self.merged_config = self._merge_configs(self.metrics_config, self.custom_metrics_config)

    def _merge_configs(self, base_config: Dict, custom_config: Dict) -> Dict:
        """
        合并内置指标和自定义指标配置
        
        策略:
        1. 如果自定义指标与内置指标有相同的一级指标，则合并其下的二级指标
        2. 如果自定义指标与内置指标有相同的二级指标，则合并其下的三级指标
        3. 如果是全新的指标，则直接添加
        """
        merged = base_config.copy()
        
        for level1_key, level1_value in custom_config.items():
            # 跳过非指标配置项(如vehicle等)
            if not isinstance(level1_value, dict) or 'name' not in level1_value:
                if level1_key not in merged:
                    merged[level1_key] = level1_value
                continue
                
            if level1_key not in merged:
                # 全新的一级指标
                merged[level1_key] = level1_value
            else:
                # 合并已存在的一级指标下的内容
                for level2_key, level2_value in level1_value.items():
                    if level2_key == 'name' or level2_key == 'priority':
                        continue
                        
                    if isinstance(level2_value, dict):
                        if level2_key not in merged[level1_key]:
                            # 新的二级指标
                            merged[level1_key][level2_key] = level2_value
                        else:
                            # 合并已存在的二级指标下的内容
                            for level3_key, level3_value in level2_value.items():
                                if level3_key == 'name' or level3_key == 'priority':
                                    continue
                                    
                                if isinstance(level3_value, dict):
                                    if level3_key not in merged[level1_key][level2_key]:
                                        # 新的三级指标
                                        merged[level1_key][level2_key][level3_key] = level3_value
        
        return merged

    def _load_custom_metrics(self) -> None:
        """加载自定义指标脚本"""
        if not self.custom_metrics_path or not Path(self.custom_metrics_path).exists():
            return
            
        custom_metrics_dir = Path(self.custom_metrics_path)
        if not custom_metrics_dir.is_dir():
            self.logger.warning(f"自定义指标路径不是目录: {custom_metrics_dir}")
            return
            
        # 遍历自定义指标脚本目录
        for file_path in custom_metrics_dir.glob("*.py"):
            if file_path.name.startswith("metric_") and file_path.name.endswith(".py"):
                try:
                    # 解析脚本名称，获取指标层级信息
                    parts = file_path.stem[7:].split('_')  # 去掉'metric_'前缀
                    if len(parts) < 3:
                        self.logger.warning(f"自定义指标脚本 {file_path.name} 命名不符合规范，应为 metric_<level1>_<level2>_<level3>.py")
                        continue
                    
                    level1, level2, level3 = parts[0], parts[1], parts[2]
                    
                    # 检查指标是否在配置中
                    if not self._check_metric_in_config(level1, level2, level3, self.custom_metrics_config):
                        self.logger.warning(f"自定义指标 {level1}.{level2}.{level3} 在配置中不存在，跳过加载")
                        continue
                    
                    # 加载脚本模块
                    module_name = f"custom_metric_{level1}_{level2}_{level3}"
                    spec = importlib.util.spec_from_file_location(module_name, file_path)
                    module = importlib.util.module_from_spec(spec)
                    spec.loader.exec_module(module)
                    
                    # 检查模块是否包含必要的函数
                    if not hasattr(module, 'evaluate'):
                        self.logger.warning(f"自定义指标脚本 {file_path.name} 缺少 evaluate 函数")
                        continue
                    
                    # 存储模块引用
                    key = f"{level1}.{level2}.{level3}"
                    self.custom_metrics_modules[key] = module
                    self.logger.info(f"成功加载自定义指标脚本: {file_path.name}")
                    
                except Exception as e:
                    self.logger.error(f"加载自定义指标脚本 {file_path.name} 失败: {str(e)}")

    def _check_metric_in_config(self, level1: str, level2: str, level3: str, config: Dict) -> bool:
        """检查指标是否在配置中存在"""
        try:
            return (level1 in config and 
                    isinstance(config[level1], dict) and
                    level2 in config[level1] and
                    isinstance(config[level1][level2], dict) and
                    level3 in config[level1][level2] and
                    isinstance(config[level1][level2][level3], dict))
        except Exception:
            return False

    def parallel_evaluate(self, data: Any) -> Dict[str, Any]:
        """并行化评估引擎（动态线程池）"""
        # 存储所有评估结果
        results = {}
        
        # 1. 先评估内置指标
        self._evaluate_built_in_metrics(data, results)
        
        # 2. 再评估自定义指标并合并结果
        self._evaluate_and_merge_custom_metrics(data, results)
        
        return results
    
    def _evaluate_built_in_metrics(self, data: Any, results: Dict[str, Any]) -> None:
        """评估内置指标"""
        # 关键修改点1：线程数=模块数
        with ThreadPoolExecutor(max_workers=len(self.metric_modules)) as executor:
            # 关键修改点2：按模块名创建future映射
            futures = {
                module_name: executor.submit(
                    self._run_module, module, data, module_name
                )
                for module_name, module in self.metric_modules.items()
            }

            # 关键修改点3：按模块顺序处理结果
            for module_name, future in futures.items():
                try:
                    from modules.lib.score import Score
                    evaluator = Score(self.merged_config, module_name)
                    result_module = future.result()
                    result = evaluator.evaluate(result_module)
                    # results.update(result[module_name])
                    results.update(result)
                except Exception as e:
                    self.logger.error(
                        f"{module_name} 评估失败: {str(e)}",
                        exc_info=True,
                        extra={"stack": True},  # 记录完整堆栈
                    )
                    # 错误信息结构化存储
                    results[module_name] = {
                        "status": "error",
                        "message": str(e),
                        "timestamp": datetime.now().isoformat(),
                    }
    
    def _evaluate_and_merge_custom_metrics(self, data: Any, results: Dict[str, Any]) -> None:
        """评估自定义指标并合并结果"""
        if not self.custom_metrics_modules:
            return
            
        # 按一级指标分组自定义指标
        grouped_metrics = {}
        for metric_key in self.custom_metrics_modules:
            level1 = metric_key.split('.')[0]
            if level1 not in grouped_metrics:
                grouped_metrics[level1] = []
            grouped_metrics[level1].append(metric_key)
        
        # 处理每个一级指标组
        for level1, metric_keys in grouped_metrics.items():
            # 检查是否为内置一级指标
            is_built_in = level1 in self.metrics_config and 'name' in self.metrics_config[level1]
            level1_name = self.merged_config[level1].get('name', level1) if level1 in self.merged_config else level1
            
            # 如果是内置一级指标，将结果合并到已有结果中
            if is_built_in and level1_name in results:
                for metric_key in metric_keys:
                    self._evaluate_and_merge_single_metric(data, results, metric_key, level1_name)
            else:
                # 如果是新的一级指标，创建新的结果结构
                if level1_name not in results:
                    results[level1_name] = {}
                
                # 评估该一级指标下的所有自定义指标
                for metric_key in metric_keys:
                    self._evaluate_and_merge_single_metric(data, results, metric_key, level1_name)
    
    def _evaluate_and_merge_single_metric(self, data: Any, results: Dict[str, Any], metric_key: str, level1_name: str) -> None:
        """评估单个自定义指标并合并结果"""
        try:
            level1, level2, level3 = metric_key.split('.')
            module = self.custom_metrics_modules[metric_key]
            
            # 获取指标配置
            metric_config = self.custom_metrics_config[level1][level2][level3]
            
            # 获取指标名称
            level2_name = self.custom_metrics_config[level1][level2].get('name', level2)
            level3_name = metric_config.get('name', level3)
            
            # 确保结果字典结构存在
            if level2_name not in results[level1_name]:
                results[level1_name][level2_name] = {}
            
            # 调用自定义指标评测函数
            metric_result = module.evaluate(data)
            from modules.lib.score import Score
            evaluator = Score(self.merged_config, level1_name)
            
            result = evaluator.evaluate(metric_result)
           
            results.update(result)
            
            
            self.logger.info(f"评测自定义指标: {level1_name}.{level2_name}.{level3_name}")
            
        except Exception as e:
            self.logger.error(f"评测自定义指标 {metric_key} 失败: {str(e)}")
            
            # 尝试添加错误信息到结果中
            try:
                level1, level2, level3 = metric_key.split('.')
                level2_name = self.custom_metrics_config[level1][level2].get('name', level2)
                level3_name = self.custom_metrics_config[level1][level2][level3].get('name', level3)
                
                if level2_name not in results[level1_name]:
                    results[level1_name][level2_name] = {}
                    
                results[level1_name][level2_name][level3_name] = {
                    "status": "error",
                    "message": str(e),
                    "timestamp": datetime.now().isoformat(),
                }
            except Exception:
                pass

    def _run_module(
        self, module_class: Any, data: Any, module_name: str
    ) -> Dict[str, Any]:
        """执行单个评估模块（带熔断机制）"""
        try:
            instance = module_class(data)
            return {module_name: instance.report_statistic()}
        except Exception as e:
            self.logger.error(f"{module_name} 执行异常: {str(e)}", stack_info=True)
            return {module_name: {"error": str(e)}}




class EvaluationPipeline:
    """评估流水线控制器"""

    def __init__(self, configPath: str, logPath: str, dataPath: str, resultPath: str, customMetricsPath: Optional[str] = None, customConfigPath: Optional[str] = None):
        self.configPath = Path(configPath)
        self.custom_config_path = Path(customConfigPath) if customConfigPath else None
        self.data_path = Path(dataPath)
        self.report_path = Path(resultPath)
        self.custom_metrics_path = Path(customMetricsPath) if customMetricsPath else None
        
        # 创建评估引擎实例，传入所有必要参数
        self.engine = EvaluationCore(
            logPath, 
            configPath=str(self.configPath), 
            customConfigPath=str(self.custom_config_path) if self.custom_config_path else None,
            customMetricsPath=str(self.custom_metrics_path) if self.custom_metrics_path else None
        )
        
        self.data_processor = self._load_data_processor()

    def _load_data_processor(self) -> Any:
        """动态加载数据预处理模块"""
        try:
            from modules.lib import data_process

            return data_process.DataPreprocessing(self.data_path, self.configPath)
        except ImportError as e:
            raise RuntimeError(f"数据处理器加载失败: {str(e)}") from e

    def execute_pipeline(self) -> Dict[str, Any]:
        """端到端执行评估流程"""
        self._validate_case()

        try:
            metric_results = self.engine.parallel_evaluate(self.data_processor)
            report = self._generate_report(
                self.data_processor.case_name, metric_results
            )
            return report
        except Exception as e:
            self.engine.logger.critical(f"流程执行失败: {str(e)}", exc_info=True)
            return {"error": str(e)}

    def _validate_case(self) -> None:
        """用例路径验证"""
        case_path = self.data_path
        if not case_path.exists():
            raise FileNotFoundError(f"用例路径不存在: {case_path}")
        if not case_path.is_dir():
            raise NotADirectoryError(f"无效的用例目录: {case_path}")

    def _generate_report(self, case_name: str, results: Dict) -> Dict:
        """生成评估报告（模板方法模式）"""
        from modules.lib.common import dict2json

        report_path = self.report_path
        report_path.mkdir(parents=True, exist_ok=True, mode=0o777)

        report_file = report_path / f"{case_name}_report.json"
        dict2json(results, report_file)
        self.engine.logger.info(f"评估报告已生成: {report_file}")
        return results


def main():
    """命令行入口（工厂模式）"""
    parser = argparse.ArgumentParser(
        description="自动驾驶评估系统 V3.0 - 支持动态指标选择和自定义指标",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    # 带帮助说明的参数定义，增加默认值
    parser.add_argument(
        "--logPath",
        type=str,
        default="d:/Kevin/zhaoyuan/zhaoyuan_new/logs/test.log",
        help="日志文件存储路径",
    )
    parser.add_argument(
        "--dataPath",
        type=str,
        default="d:/Kevin/zhaoyuan/zhaoyuan_new/data/zhaoyuan1",
        help="预处理后的输入数据目录",
    )
    parser.add_argument(
        "--configPath",
        type=str,
        default="d:/Kevin/zhaoyuan/zhaoyuan_new/config/metrics_config.yaml",
        help="评估指标配置文件路径",
    )
    parser.add_argument(
        "--reportPath",
        type=str,
        default="d:/Kevin/zhaoyuan/zhaoyuan_new/reports",
        help="评估报告输出目录",
    )
    # 新增自定义指标路径参数（可选）
    parser.add_argument(
        "--customMetricsPath",
        type=str,
        default="d:/Kevin/zhaoyuan/zhaoyuan_new/custom_metrics",
        help="自定义指标脚本目录（可选）",
    )
    # 新增自定义指标路径参数（可选）
    parser.add_argument(
        "--customConfigPath",
        type=str,
        default="d:/Kevin/zhaoyuan/zhaoyuan_new/config/custom_metrics_config.yaml",
        help="自定义指标脚本目录（可选）",
    )
    args = parser.parse_args()

    try:
        pipeline = EvaluationPipeline(
            args.configPath, args.logPath, args.dataPath, args.reportPath, args.customMetricsPath, args.customConfigPath
        )
        start_time = time.perf_counter()

        result = pipeline.execute_pipeline()

        if "error" in result:
            sys.exit(1)

        print(f"评估完成，耗时: {time.perf_counter()-start_time:.2f}s")
        print(f"报告路径: {pipeline.report_path}")
    except KeyboardInterrupt:
        print("\n用户中断操作")
        sys.exit(130)
    except Exception as e:
        print(f"程序执行异常: {str(e)}")
        sys.exit(1)


if __name__ == "__main__":
    warnings.filterwarnings("ignore")
    main()