cicv_evaluate_master-single_eval/single_case_eval.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

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)