#!/usr/bin/env python # -*- coding: utf-8 -*- ################################################################## # # Copyright (c) 2023 CICV, Inc. All Rights Reserved # ################################################################## """ @Authors: yangzihao(yangzihao@china-icv.cn) @Data: 2024/01/30 @Last Modified: 2024/01/30 @Summary: Evaluateion functions """ import os import sys sys.path.append('../config') sys.path.append('../common') sys.path.append('../modules') sys.path.append('../results') import json import traceback import pandas as pd import log from config_parser import ConfigParse from single_case_evaluate import single_case_evaluate, single_case_statistic from common import df2csv, dict2json from data_quality import frame_loss_statistic def single_case_eval(configPath, dataPath, reportPath, csvPath, playbackPath, customMetricPath, customScorePath, case_name): logger = log.get_logger() # case_name = os.path.basename(os.path.dirname(dataPath)) # 判断文件夹是否为空 if len(os.listdir(dataPath)) == 0: print("No files in data_path!") # 路径异常 logger.error(f"[case:{case_name}] SINGLE_CASE_EVAL: No files in data_path!") sys.exit(-1) # 加载配置文件 try: # json_file = os.path.join(configPath, 'config.json') config = ConfigParse(configPath) except Exception as e: print('Config file parsing ERROR!', e) traceback.print_exc() logger.error(f"[case:{case_name}] SINGLE_CASE_EVAL: Config file parsing ERROR: {repr(e)}!", exc_info=True) sys.exit(-1) # data quality detect is_bad_quality = data_quality_detect(dataPath, case_name) if is_bad_quality: print("Frame loss > 10%, system exit.") # sys.exit(-1) # data complement """ TODO: data complement """ # 单用例评价，并生成报告 try: case_dict = single_case_evaluate(dataPath, config, customMetricPath, customScorePath, case_name) # 评估单用例 single_case_dict = single_case_statistic(case_dict) # 对单用例结果增加内容，并生成报告 # 回放 df2csv(case_dict['playbackData'], playbackPath) # 将DataFrame保存为csv文件 df2csv(case_dict['evalData'], csvPath) # 将单用例测试结果保存为json文件 single_case_dict.pop('playbackData') single_case_dict.pop('evalData') dict2json(single_case_dict, reportPath) except Exception as e: traceback.print_exc() logger.error(f"[case:{case_name}] SINGLE_CASE_EVAL: Evaluate single case ERROR: {repr(e)}!", exc_info=True) # guarantee_result(reportPath, csvPath, playbackPath) # 保底生成文件 sys.exit(-1) def data_quality_detect(dataPath, case_name): logger = log.get_logger() # FIRST_ORDER_LOSS = 0.01 # first : little loss 100 SECOND_ORDER_LOSS = 0.05 # second : could use 95 THIRD_ORDER_LOSS = 0.10 # third : could not use 90 is_bad_quality = False try: frame_loss_dict = frame_loss_statistic(dataPath) except Exception as e: traceback.print_exc() logger.error(f"[case:{case_name}] SINGLE_CASE_EVAL: frame loss statistic ERROR: {repr(e)}!", exc_info=True) is_bad_quality = True return is_bad_quality for key, value in frame_loss_dict.items(): # sensor data sensor_data = ["RoadMark", "RoadPos", "TrafficLight", "TrafficSign"] if any(file in key for file in sensor_data): logger.info(f"[case:{case_name}] SINGLE_CASE_EVAL: [{key}] : {value['result']}") continue if value['frame_loss_rate'] > THIRD_ORDER_LOSS: is_bad_quality = True logger.error( f"[case:{case_name}] SINGLE_CASE_EVAL: [{key}] frame loss rate > {THIRD_ORDER_LOSS * 100}%: {value['result']}") elif value['frame_loss_rate'] > SECOND_ORDER_LOSS: logger.info( f"[case:{case_name}] SINGLE_CASE_EVAL: [{key}] frame loss rate > {SECOND_ORDER_LOSS * 100}%: {value['result']}") else: logger.info( f"[case:{case_name}] SINGLE_CASE_EVAL: [{key}] frame loss rate < {SECOND_ORDER_LOSS * 100}%: {value['result']}") return is_bad_quality def guarantee_result(reportPath, csvPath, playbackPath): result = { "details": { "safe": { "name": "安全性", "weight": "100.00%", "collisionRisk": 0, "noObjectCar": "false", "score": 100, "level": "优秀", "weightDistribution": { "name": "安全性", "safeTime": { "weight": "时间类型(27.71%)", "indexes": { "TTC": "TTC(18.34%)", "MTTC": "MTTC(50.01%)", "THW": "THW(31.65%)" } }, "safeDistance": { "weight": "距离类型(44.37%)", "indexes": { "LonSD": "LonSD(76.80%)", "LatSD": "LatSD(23.20%)" } }, "safeAcceleration": { "weight": "加速度类型(2.60%)", "indexes": { "DRAC": "DRAC(66.67%)", "BTN": "BTN(6.22%)", "STN": "STN(27.11%)" } }, "safeProbability": { "weight": "概率类型(25.32%)", "indexes": { "collisionRisk": "碰撞风险概率(50.00%)", "collisionSeverity": "碰撞严重程度(50.00%)" } } }, "details": { "safeTime": { "name": "时间类型", "score": 100, "level": "较差", "description1": "TTC和THW指标表现良好，MTTC指标表现不佳，MTTC极值超过合理范围73.59%", "description2": "TTC和THW指标均在合理范围内，表现良好，MTTC指标共有0.32秒超出合理范围，算法应加强在该时间段对跟车距离的控制", "indexes": { "TTC": { "name": "TTC", "meaning": "TTC", "score": 100, "extremum": "3.76", "range": "[2.86, inf)", "rate": "100%" }, "MTTC": { "name": "MTTC", "meaning": "MTTC", "score": 100, "extremum": "0.32", "range": "[1.2, inf)", "rate": "98.9%" }, "THW": { "name": "THW", "meaning": "THW", "score": 100, "extremum": "2.98", "range": "[0.4, inf)", "rate": "100%" } }, "builtin": { "MTTC": { "name": "MTTC", "data": [], "range": "[1.2, inf)" }, "THW": { "name": "THW", "data": [], "range": "[0.4, inf)" } }, "custom": {} }, }, "description1": "未满足设计指标要求。算法在本轮测试中有碰撞风险，需要提高算法在时间类型和距离类型上的表现。在时间类型和距离类型中，MTTC指标共有0.32秒超出合理范围；LonSD指标共有3.93秒超出合理范围；STN指标共有0.07秒超出合理范围；", "description2": "安全性在时间类型和距离类型上存在严重风险，需要重点优化。", }, "commonData": { "per": { "name": "脚刹/油门踏板开度（百分比）", "legend": [ "刹车踏板开度", "油门踏板开度" ], "data": [] }, "ang": { "name": "方向盘转角（角度°）", "data": [] }, "spe": { "name": "速度（km/h）", "legend": [ "自车速度", "目标车速度", "自车与目标车相对速度" ], "data": [] }, "acc": { "name": "加速度（m/s²)", "legend": [ "横向加速度", "纵向加速度" ], "data": [] }, "dis": { "name": "前车距离（m）", "data": [] }, "ttc": { "name": "TTC(m)", "data": [] } }, "commonMarkLine": [] }, "algorithmComprehensiveScore": 100, "algorithmLevel": "优秀", "testMileage": "0.00米", "testDuration": "0.00秒", "algorithmResultDescription1": "车辆在本轮测试中，未出现违反交通规则行为、跟停行为和不舒适行为。", "algorithmResultDescription2": "综上所述，算法表现良好。" } with open(f'{reportPath}', 'w', encoding='utf-8') as f: f.write(json.dumps(result, ensure_ascii=False)) eval_data_columns = ['simTime', 'simFrame', 'playerId', 'type', 'posX', 'posY', 'posZ', 'posH', 'speedX', 'speedY', 'speedZ', 'accelX', 'accelY', 'accelZ', 'dimX', 'dimY', 'dimZ', 'offX', 'speedH', 'accelH', 'travelDist'] eval_data_df = pd.DataFrame(columns=eval_data_columns) eval_data_df.to_csv(f'{csvPath}', index=False) playback_data_columns = ['simTime', 'simFrame', 'speed', 'curvHor', 'rollRel', 'pitchRel', 'latSpeedRel', 'lonSpeedRel', 'latDistanceRel', 'lonDistanceRel'] playback_data_df = pd.DataFrame(columns=playback_data_columns) playback_data_df.to_csv(f'{playbackPath}', index=False)