pujin_outdoor/single_case_evaluate.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
##################################################################
#
# Copyright (c) 2023 CICV, Inc. All Rights Reserved
#
##################################################################
"""
@Authors:           yangzihao(yangzihao@china-icv.cn)
@Data:              2023/11/28
@Last Modified:     2023/11/28
@Summary:           Evaluate single case.
"""

import os
import sys

import json
import traceback

import log
from common import score_grade, mileage_format, duration_format, string_concatenate, replace_key_with_value, \
    import_class
from config_parser import ConfigParse
from data_process import DataProcess
from report_generate import report_generate

from safe import Safe
from comfort import Comfort
from accurate import Accurate
from efficient import Efficient


def single_case_eval(configPath, dataPath, resultPath, trackPath, case_name):
    """

    :param configPath: 配置文件路径
    :param dataPath: 评价数据路径
    :param resultPath: 评价结果路径
    :param trackPath: 轨迹图片路径
    :param case_name: 用例名称

    :return: None
    """
    # 获取日志记录器
    logger = log.get_logger()

    # 判断文件夹是否为空
    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:
        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)

    # 单用例评价，并生成报告
    try:
        reportDict = single_case_evaluate(config, dataPath, resultPath, case_name)  # 评估单用例
        # reportDict = single_case_statistic(case_report_dict)  # 对单用例结果增加内容

        # 生成过程文件report.json
        with open(f'{resultPath}/report.json', 'w', encoding='utf-8') as f:
            f.write(json.dumps(reportDict, ensure_ascii=False))

        # 通过report.json生成报告
        reportPdf = os.path.join(resultPath, 'report.pdf')
        report_generate(reportDict, reportPdf, trackPath)

    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)
        sys.exit(-1)


def single_case_evaluate(config, dataPath, resultPath, case_name):
    """
    This function takes a str of single case' path and a dict of config info.
    The function uses the scripts of 5 dimensions to analyze and evaluate the single case.

    Arguments:
        dataPath: A str of a single case's csv files path.
        config: A dict of config info, which contains the algorithm info, index info and so on.
        resultPath: A str of report path.
        case_name: A str of case name.

    Returns:
        case_report_dict: A dict containing evaluation results. Basic info, scores and descriptions of 5 dimensions.
    """
    # 获取日志记录器
    logger = log.get_logger()

    # 数据处理
    try:
        data_processed = DataProcess(dataPath, config)  # ego_df; obj_df; trajectory_df
    except Exception as e:
        traceback.print_exc()
        logger.error(f"[case:{case_name}] SINGLE_CASE_EVAL: Data processed ERROR: {repr(e)}!", exc_info=True)
        sys.exit(-1)

    # 选择评分模型（默认内置模型）
    scorePath = r"./"
    scoreFile = "score_weight"
    scoreModel = import_class(scorePath, scoreFile)
    # initialization
    case_report_dict = {
        "name": case_name
    }

    # score info
    dimension_list = config.dimension_list
    score_dimension_dict = {}
    bad_dimension_list = []
    dimension_dict = {}

    # 分维度进行评价
    try:
        logger.info(f"[case:{case_name}] SINGLE_CASE_EVAL: Dimension evaluate start: ")
        for dimension in dimension_list:
            # print("dimension is", dimension)
            # 根据维度列表动态调用类
            dimension_instance = globals()[dimension.capitalize()](data_processed, scoreModel, resultPath)
            # print("dimension_instance is", dimension_instance)
            # 运行各维度评价函数，进行分析评价及报告内容统计
            dimension_report_dict = dimension_instance.report_statistic()
            dimension_dict[dimension] = dimension_report_dict
            score_dimension_dict[dimension] = dimension_report_dict['score']
            # 对各维度分数进行判断，分类表现差的维度
            if score_dimension_dict[dimension] < 80:
                bad_dimension_list.append(dimension)

            # if dimension == "comfort":
            #     discomfortCount = dimension_report_dict['discomfortCount']
            #     bad_indicator_list.append("不舒适行为") if discomfortCount != 0 else good_indicator_list.append("不舒适行为")

    # 异常打印，并退出
    except Exception as e:
        traceback.print_exc()
        logger.error(f"[case:{case_name}] SINGLE_CASE_EVAL: Dimension evaluate ERROR: {repr(e)}!", exc_info=True)
        sys.exit(-1)

    # 将评价结果写入报告字典内
    case_report_dict["details"] = dimension_dict
    # calculate the score and grade of the case
    weight_dimension_dict = config.dimension_weight
    score_dimension = {key: score_dimension_dict[key] * weight_dimension_dict[key] for key in score_dimension_dict}

    # 计算用例分数、等级，写入报告字典
    score_case = round(sum(score_dimension.values()), 2)
    grade_case = score_grade(score_case)
    case_report_dict['algorithmComprehensiveScore'] = score_case
    case_report_dict['algorithmLevel'] = grade_case

    # 计算用例里程和持续时长
    # case_report_dict['testMileage'] = data_processed.report_info['mileage']
    # case_report_dict['testDuration'] = data_processed.report_info['duration']
    case_report_dict['testMileage'] = mileage_format(data_processed.report_info['mileage'])
    case_report_dict['testDuration'] = duration_format(data_processed.report_info['duration'])

    # generate algorithmResultDescription2
    if not bad_dimension_list:
        algorithmResultDescription2 = '算法在各个维度的表现俱佳。'
    else:
        str_bad_dimension = string_concatenate(bad_dimension_list)
        # replace English to Chinese dimension name
        algorithmResultDescription2 = f'建议算法优化在{str_bad_dimension}方面的表现。'

    algorithmResultDescription2 = replace_key_with_value(algorithmResultDescription2, config.dimension_name)

    algorithmResultDescription = f"算法得分{score_case}分，总体表现{grade_case}。{algorithmResultDescription2}"
    case_report_dict['algorithmResultDescription'] = algorithmResultDescription

    # 生成图片字典，将过程生成图片路径记录，用于报告生成
    graph_dict = {
        # "speed": os.path.join(resultPath, "Speed.png"),
        # "commandSpeed": os.path.join(resultPath, "CommandSpeed.png"),
        # "linearAccelerate": os.path.join(resultPath, "LinearAccelerate.png"),
        # "angularAccelerate": os.path.join(resultPath, "AngularAccelerate.png"),
        # "trajectory": os.path.join(resultPath, "track.png")
        "速度": os.path.join(resultPath, "Speed.png"),
        "控制指令速度": os.path.join(resultPath, "CommandSpeed.png"),
        "线加速度": os.path.join(resultPath, "LinearAccelerate.png"),
        "角加速度": os.path.join(resultPath, "AngularAccelerate.png"),
        "行驶轨迹": os.path.join(resultPath, "track.png")
    }

    case_report_dict['graphPath'] = graph_dict
    return case_report_dict