import json from modules.lib.log_manager import LogManager class Score: def __init__(self, yaml_config, module_name: str ): self.logger = LogManager().get_logger() # 获取全局日志实例 self.calculated_metrics = None self.config = yaml_config self.module_config = None self.module_name = module_name self.t_threshold = None self.process_config(self.config) self.level_3_merics = self._extract_level_3_metrics(self.module_config) self.result = {} def process_config(self, config_dict): t_threshold = config_dict.get("T_threshold") if t_threshold is None: raise ValueError("配置中缺少 T_threshold 键") module_keys = [key for key in config_dict if key != "T_threshold"] # if len(module_keys) != 1: # raise ValueError("配置字典应包含且仅包含一个模块配置键") # module_name = module_keys[0] module_config = config_dict[self.module_name] # print(f'模块名称：{module_name}') # print(f'模块配置：{module_config}') # print(f'T_threshold：{t_threshold}') # 实际业务逻辑（示例：存储到对象属性） # self.module_name = module_name self.module_config = module_config self.t_threshold = t_threshold self.logger.info(f'模块名称：{self.module_name}') self.logger.info(f'模块配置：{self.module_config}') self.logger.info(f'T_threshold: {t_threshold}') def _extract_level_3_metrics(self, d): name = [] for key, value in d.items(): if isinstance(value, dict): # 如果值是字典，继续遍历 self._extract_level_3_metrics(value) elif key == 'name': # 找到name键时，将值添加到列表 name.append(value) return name def is_within_range(self, value, min_val, max_val): return min_val <= value <= max_val def evaluate_level_3(self, metrics): result3 = {} name = metrics.get('name') priority = metrics.get('priority') max_val = metrics.get('max') min_val = metrics.get('min') self.level_3_merics.append(name) print(f'name: {name}') print(f'self.calculated_metrics: {self.calculated_metrics}') metric_value = self.calculated_metrics.get(name) print(f'metric_value: {metric_value}') result3[name] = { 'result': True, 'priority': priority } if metric_value is None: return result3 if not self.is_within_range(metric_value, min_val, max_val) and priority == 0: result3[name]['result'] = False elif not self.is_within_range(metric_value, min_val, max_val) and priority == 1: result3[name]['priority_1_count'] += 1 # Count priority 1 failures and override result if more than 3 priority_1_metrics = [v for v in result3.values() if v['priority'] == 1 and not v['result']] if len([m for m in priority_1_metrics if not m['result']]) > 3: result3[name]['result'] = False return result3 def evaluate_level_2(self, metrics): result2 = {} name = metrics.get('name') priority = metrics.get('priority') result2[name] = {} for metric, sub_metrics in metrics.items(): if metric not in ['name', 'priority']: result2[name].update(self.evaluate_level_3(sub_metrics)) # Aggregate results for level 2 config.T0 config.T1 config.T2 priority_0_count = sum(1 for v in result2[name].values() if v['priority'] == 0 and not v['result']) priority_1_count = sum(1 for v in result2[name].values() if v['priority'] == 1 and not v['result']) priority_2_count = sum(1 for v in result2[name].values() if v['priority'] == 2 and not v['result']) if priority_0_count > self.t_threshold['T0_threshold']: result2[name]['result'] = False elif priority_1_count > self.t_threshold['T1_threshold']: for metric in result2[name].values(): metric['result'] = False elif priority_2_count > self.t_threshold['T2_threshold']: for metric in result2[name].values(): metric['result'] = False else: result2[name]['result'] = True # Default to True unless overridden result2[name]['priority'] = priority result2[name]['priority_0_count'] = priority_0_count result2[name]['priority_1_count'] = priority_1_count result2[name]['priority_2_count'] = priority_2_count return result2 def evaluate_level_1(self): name = self.module_config.get('name') priority = self.module_config.get('priority') result1 = {} result1[name] = {} for metric, metrics in self.module_config.items(): if metric not in ['name', 'priority']: result1[name].update(self.evaluate_level_2(metrics)) # Aggregate results for level 2 config.T0 config.T1 config.T2 priority_0_count = sum(1 for v in result1[name].values() if v['priority'] == 0 and not v['result']) priority_1_count = sum(1 for v in result1[name].values() if v['priority'] == 1 and not v['result']) priority_2_count = sum(1 for v in result1[name].values() if v['priority'] == 2 and not v['result']) if priority_0_count > self.t_threshold['T0_threshold']: result1[name]['result'] = False elif priority_1_count > self.t_threshold['T1_threshold']: for metric in result1[name].values(): metric['result'] = False elif priority_2_count > self.t_threshold['T2_threshold']: for metric in result1[name].values(): metric['result'] = False else: result1[name]['result'] = True # Default to True unless overridden result1[name]['priority'] = priority result1[name]['priority_0_count'] = priority_0_count result1[name]['priority_1_count'] = priority_1_count result1[name]['priority_2_count'] = priority_2_count return result1 def evaluate(self, calculated_metrics): self.calculated_metrics = calculated_metrics self.result = self.evaluate_level_1() return self.result def evaluate_single_case(self, case_name, priority, json_dict): name = case_name result = {} result[name] = {} # print(json_dict) # Aggregate results for level 2 config.T0 config.T1 config.T2 priority_0_count = sum(1 for v in json_dict.values() if v['priority'] == 0 and not v['result']) priority_1_count = sum(1 for v in json_dict.values() if v['priority'] == 1 and not v['result']) priority_2_count = sum(1 for v in json_dict.values() if v['priority'] == 2 and not v['result']) if priority_0_count > config.T0: result[name]['result'] = False elif priority_1_count > config.T1: for metric in result[name].values(): metric['result'] = False elif priority_2_count > config.T2: for metric in result[name].values(): metric['result'] = False else: result[name]['result'] = True # Default to True unless overridden result[name]['priority'] = priority result[name]['priority_0_count'] = priority_0_count result[name]['priority_1_count'] = priority_1_count result[name]['priority_2_count'] = priority_2_count result[case_name].update(json_dict) return result def evaluate_single_case_back(case_name, priority, json_dict): """对单个案例进行评估""" result = {case_name: {}} priority_counts = {priority: sum(1 for v in json_dict.values() if v['priority'] == priority and not v['result']) for priority in [0, 1, 2]} if priority_counts[0] > config.T0: result[case_name]['result'] = False elif priority_counts[1] > config.T1: for metric in result[case_name].values(): metric['result'] = False elif priority_counts[2] > config.T2: for metric in result[case_name].values(): metric['result'] = False else: result[case_name]['result'] = True result[case_name].update(priority_counts) result[case_name].update(json_dict) # 合并原始数据 return result def main(): # config_path = r'/home/kevin/kevin/zhaoyuan/evaluate_zhaoyuan/models/safety/config.yaml' # config_path1 = r'/home/kevin/kevin/zhaoyuan/evaluate_zhaoyuan/models/safety/config.json' # calculated_metrics = { # 'TTC': 1.0, # 'MTTC': 1.0, # 'THW': 1.0, # 'LonSD': 50.0, # 'LatSD': 3.0, # 'DRAC': 3.0, # 'BTN': -1000.0, # 'STN': 0.5, # 'collisionRisk': 5.0, # 'collisionSeverity': 2.0, # } # # evaluator = Score(config_path, calculated_metrics) # evaluator = Score(config_path) # result = evaluator.evaluate(calculated_metrics) # with open(config_path1, 'w') as json_file: # json.dump(result, json_file, indent=4) # `indent` 参数用于美化输出 config_path = r'/home/kevin/kevin/zhaoyuan/zhaoyuan/models/caseMetric/single_config.yaml' config_path1 = r'/home/kevin/kevin/zhaoyuan/zhaoyuan/result/data_zhaoyuan/data_zhaoyuan_single_report.json' # evaluator = Score(config_path, calculated_metrics) with open(config_path1, 'r') as file: data = json.load(file) result = evaluate_single_case("case1", 0, data) print(result) if __name__ == '__main__': main()