cicv_evaluate_master-single_eval/single_run_test.py

#!/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
import time
import json
import pandas as pd
from single_case_eval import single_case_eval
import log


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)


if __name__ == "__main__":

    # import warnings
    # warnings.filterwarnings("ignore")

    from warnings import simplefilter

    # simplefilter(action="ignore", category=FutureWarning)
    simplefilter(action="ignore")

    # if len(sys.argv) >= 6:
    #     configPath = sys.argv[1]
    #     dataPath = sys.argv[2]
    #     reportPath = sys.argv[3]
    #     csvPath = sys.argv[4]
    #     logPath = sys.argv[5]
    #     customMetricPath = sys.argv[6]
    #     customScorePath = sys.argv[7]
    #     playbackPath = sys.argc[8]
    #     case_name = sys.argc[9]

    # # case_name = os.path.basename(os.path.dirname(dataPath))
    #     log.setup_logger(logPath)
    #     logger = log.get_logger()

    #     if not os.path.exists(configPath):
    #         print('Invalid configPath!')
    #         logger.error(f"[case:{case_name}] SINGLE_CASE_EVAL: Invalid configPath!")
    #         sys.exit(-1)
    #     elif not os.path.exists(dataPath):
    #         print('Invalid dataPath!')
    #         logger.error(f"[case:{case_name}] SINGLE_CASE_EVAL: Invalid dataPath!")
    #         sys.exit(-1)
    #     elif not os.path.exists(customMetricPath):
    #         print('Invalid customMetricPath!')
    #         logger.error(f"[case:{case_name}] SINGLE_CASE_EVAL: Invalid customMetricPath!")
    #         sys.exit(-1)
    #     elif not os.path.exists(customScorePath):
    #         print('Invalid customScorePath!')
    #         logger.error(f"[case:{case_name}] SINGLE_CASE_EVAL: Invalid customScorePath!")
    #         sys.exit(-1)
    #     else:
    #         try:
    #             logger.info(f"[case:{case_name}] SINGLE_CASE_EVAL: Start evaluating:")
    #             print("SINGLE_CASE_EVAL程序开始运行：")
    #             print(f"   configPath: {configPath},\n   dataPath: {dataPath},\n   reportPath: {reportPath},\n   "
    #                   f"csvPath: {csvPath},\n   logPath: {logPath},\n   customMetricPath: {customMetricPath}\n   customScorePath: {customScorePath}")
    #             t1 = time.time()
    #             single_case_eval(configPath, dataPath, reportPath, csvPath, playbackPath, customMetricPath, customScorePath)
    #             t2 = time.time()
    #             print(f"程序结束，执行时间：{int(t2 - t1)} s")
    #             logger.info(f"[case:{case_name}] SINGLE_CASE_EVAL: End.")
    #             sys.exit(0)
    #         except Exception as e:
    #             print("异常，退出...")
    #             print(repr(e))
    #             sys.exit(-1)
    # else:
    #     logPath = sys.argv[5]
    #     log.setup_logger(logPath)
    #     logger = log.get_logger()
    #     print('SINGLE_CASE_EVAL: No enough arguments!')
    #     logger.error(f"SINGLE_CASE_EVAL: No enough arguments!")
    #     sys.exit(-1)

    # configPath = rf"./config/config.json"
    # configPath = rf"./config/config0531.json"
    # configPath = rf"./config/config0612_ica+lka.json"
    # configPath = rf"./config/config_ica_0730.json"
    # configPath = rf"./config/config_lka_0730.json"
    # configPath = rf"./config/config_lka_0730-2.json"
    # configPath = rf"./config/config_ica_0731-2.json"
    configPath = rf"./config/config_nofunction.json"

    # dataPath = r"./task_20240401_replace/alc_zicheyouganraobiandao_30/data/"
    # dataPath = r"./task_0430/zuozhuan_cheliangchongtutongxiang/data"
    # dataPath = r"./task_0430/case0415/data"
    # dataPath = r"./task_0515/case0528/data"
    # dataPath = r"./task_0515/data0529-2/data"
    # dataPath = r"./task_0515/case0529-1"
    # dataPath = r"./task_0515/VOYAH-H53-AEB-01/data"
    # dataPath = r"./task_0515/case0529-LKA/data"
    # dataPath = r"./task_0515/case0530-ica-post/data"
    # dataPath = r"./task_0515/0805-7/data"
    # dataPath = r"./task_0515/0806a-2/data"
    dataPath = r"./task_0515/accident_0716_6/data"

    case_name = os.path.basename(os.path.dirname(os.path.normpath(dataPath)))
    casePath = os.path.join("./result", case_name)

    if not os.path.exists(casePath):
        os.makedirs(casePath)

    customMetricPath = './custom'
    # customMetricPath = './custom/cicv_LKA'
    # customMetricPath = './custom/cicv_ICA_lateral_control'
    # customMetricPath = './demoICA'
    # customMetricPath = './custom/cicv_ICA'
    customScorePath = './customScore'
    logPath = './log.log'

    reportPath = rf"{casePath}/report.json"
    csvPath = rf"{casePath}/evalData.csv"
    playbackPath = f"{casePath}/playback.csv"

    log.setup_logger(logPath)
    logger = log.get_logger()

    if not os.path.exists(configPath):
        print('Invalid configPath!')
        logger.error(f"[case:{case_name}] SINGLE_CASE_EVAL: Invalid configPath!")
    elif not os.path.exists(dataPath):
        print('Invalid dataPath!')
        logger.error(f"[case:{case_name}] SINGLE_CASE_EVAL: Invalid dataPath!")
    elif not os.path.exists(customMetricPath):
        print('Invalid dataPath!')
        logger.error(f"[case:{case_name}] SINGLE_CASE_EVAL: Invalid dataPath!")
    else:
        try:
            logger.info(f"[case:{case_name}] SINGLE_CASE_EVAL: Start evaluating:")
            print("SINGLE_CASE_EVAL程序开始运行：")
            print(f"  configPath: {configPath}, \n  dataPath: {dataPath}, \n  reportPath: {reportPath}")
            t1 = time.time()
            single_case_eval(configPath, dataPath, reportPath, csvPath, playbackPath, customMetricPath, customScorePath,
                             case_name)
            t2 = time.time()
            print(f"程序结束，执行时间：{int(t2 - t1)} s")
            logger.info(f"[case:{case_name}] SINGLE_CASE_EVAL: End.")

        except Exception as e:
            guarantee_result(reportPath, csvPath, playbackPath)
            print("异常如下：")
            print(repr(e))
            # traceback.print_exc()