pji_indoor_robot_evaluate_sim/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 pandas as pd
import numpy as np
import requests
import json
import pathlib
import time
import traceback

import log
from common import score_grade, mileage_format, duration_format, string_concatenate, replace_key_with_value, \
    import_class
from data_process import DataProcess
# from customDimension import CustomDimension
# from custom_run import custom_run
# from safe import Safe
# from function import Function
# from compliance import Compliance
from comfort import Comfort
from efficient import Efficient


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

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

    Returns:
        case_dict: A dict containing evaluation results. Basic infos, scores and descriptions of 5 dimensions.
    """
    logger = log.get_logger()

    try:
        data_processed = DataProcess(path, config)
    except Exception as e:
        traceback.print_exc()
        logger.error(f"[case:{case_name}] SINGLE_CASE_EVAL: Data processed ERROR: {repr(e)}!", exc_info=True)

    # custom metrics evaluate
    # customMetricPath = '../custom'
    # custom_metric_list = [metric for metric in config.metric_list if metric not in config.builtinMetricList]

    custom_data = dict()

    # if customMetricPath if empty
    # try:
    #     if custom_metric_list:
    #         custom_data = custom_run(customMetricPath, data_processed, custom_metric_list, case_name)
    #     else:
    #         custom_data = dict()
    #
    # except Exception as e:
    #     traceback.print_exc()
    #     logger.error(f"[case:{case_name}] SINGLE_CASE_EVAL: Custom metric evaluate ERROR: {repr(e)}!", exc_info=True)

    # scoreFile = 'linear_score'

    # custom score
    # if config.scoreModel == "builtin":
    #     scorePath = r"./"
    #     scoreFile = "score_weight"
    #     scoreModel = import_class(scorePath, scoreFile)
    # else:
    #     # customScorePath = r"C:\Users\cicv\Desktop\ads_evaluate_V4.1.0\customScore"
    #     scoreFile = config.scoreModel
    #     scoreModel = import_class(customScorePath, scoreFile)

    scorePath = r"./"
    scoreFile = "score_weight"
    scoreModel = import_class(scorePath, scoreFile)

    builtin_dimension_list = config.builtinDimensionList
    dimension_name_dict = config.dimension_name

    # initialization
    case_dict = {}
    case_dict["name"] = case_name

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

    # statistic the index by good or bad score
    good_indicator_list = []
    bad_indicator_list = []
    # bad_indicator_count_list = []

    dimension_dict = {}
    # eval_data_list = []

    try:
        logger.info(f"[case:{case_name}] SINGLE_CASE_EVAL: Dimension evaluate start: ")
        for dimension in dimension_list:
            dimension_instance = globals()[dimension.capitalize()](data_processed, custom_data, scoreModel, resultPath)

            # if dimension in builtin_dimension_list:
            #     dimension_instance = globals()[dimension.capitalize()](data_processed, custom_data, scoreModel)
            # else:
            #     dimension_instance = CustomDimension(dimension, data_processed, custom_data, scoreModel)

            dimension_report_dict = dimension_instance.report_statistic()
            dimension_dict[dimension] = dimension_report_dict
            score_dimension_dict[dimension] = dimension_report_dict['score']

            # dimension_eval_data = dimension_instance.get_eval_data()
            # if not dimension_eval_data.empty:
            #     eval_data_list.append(dimension_eval_data)
            # eval_data_dict[dimension] = dimension_eval_data

            if score_dimension_dict[dimension] < 80:
                bad_dimension_list.append(dimension)

            # if dimension == "function":
            #     followStopCount = dimension_report_dict['followStopCount']
            #     bad_indicator_list.append("跟停行为") if followStopCount != 0 else good_indicator_list.append("跟停行为")
            # elif dimension == "compliance":
            #     illegalCount = dimension_report_dict['illegalCount']
            #     bad_indicator_list.append("违反交通规则行为") if illegalCount != 0 else good_indicator_list.append("违反交通规则行为")
            # elif 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_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_dict['algorithmComprehensiveScore'] = score_case
    case_dict['algorithmLevel'] = grade_case

    case_dict['testMileage'] = data_processed.report_info['mileage']
    case_dict['testDuration'] = data_processed.report_info['duration']

    # generate algorithmResultDescription1 according to the actual case
    # if bad_indicator_list:
    #     for type in bad_indicator_list:
    #         if type == '不舒适行为':
    #             bad_indicator_count_list.append(f'{discomfortCount}次不舒适行为')
    #         if type == '跟停行为':
    #             bad_indicator_count_list.append(f'{followStopCount}次跟停行为')
    #         if type == '违反交通规则行为':
    #             bad_indicator_count_list.append(f'违反交通规则{illegalCount}次')
    # str_bad_indicator_count = string_concatenate(bad_indicator_count_list) if bad_indicator_count_list else ""
    # str_good_indicator = string_concatenate(good_indicator_list) if good_indicator_list else ""

    # algorithmResultDescription1 = "车辆在本轮测试中，"
    # if str_bad_indicator_count:
    #     algorithmResultDescription1 += f"共出现{str_bad_indicator_count}，"
    # if str_good_indicator:
    #     algorithmResultDescription1 += f"未出现{str_good_indicator}。"
    #
    # if not str_good_indicator and not str_bad_indicator_count:
    #     algorithmResultDescription1 += "【未进行舒适性、跟车及合规性评测】。"

    # 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}指标上的表现。'

    # description = f"算法总体表现{grade_case}，高效性表现{dimension_dict['efficient']['level']}，平顺性表现{dimension_dict['comfort']['level']}。"
    description = f"算法总体表现{grade_case}。"

    # case_dict['algorithmResultDescription1'] = algorithmResultDescription1
    # case_dict['algorithmResultDescription2'] = algorithmResultDescription2
    # case_dict['algorithmResultDescription1'] = replace_key_with_value(algorithmResultDescription1, dimension_name_dict)
    # case_dict['algorithmResultDescription'] = replace_key_with_value(algorithmResultDescription2, dimension_name_dict)
    case_dict['algorithmResultDescription'] = description

    graph_dict = {
        "linearAccelerate": os.path.join(resultPath, "LinearAccelerate.png"),
        "angularAccelerate": os.path.join(resultPath, "AngularAccelerate.png"),
        "speed": os.path.join(resultPath, "Speed.png")
    }

    case_dict['graphPath'] = graph_dict

    # for df in eval_data_list:
    #     eval_data_df = pd.merge(eval_data_df, df, on=['simTime', 'simFrame', 'playerId'], how="left")
    # eval_data_df = eval_data_df[eval_data_df['simFrame'] > 0]
    #
    # case_dict['evalData'] = eval_data_df
    # case_dict['playbackData'] = playback(eval_data_df)

    # report.json
    # case_dict = {
    #     'testMileage': mileage,
    #     'testDuration': duration,
    #     'algorithmResultDescription1': algorithmResultDescription1,
    #     'algorithmResultDescription2': algorithmResultDescription2,
    #     'algorithmComprehensiveScore': score_case,
    #     'algorithmLevel': grade_case,
    #
    #     'safe': safe_report_dict,
    #     'function': func_report_dict,
    #     'compliance': comp_report_dict,
    #     'comfort': comf_report_dict,
    #     'efficient': effi_report_dict,
    #
    #     'commonData': common_report_dict
    #
    # }

    return case_dict


def playback(df):
    """

    Args:
        df:

    Returns:

    """

    target_id = 2
    # choose_df = df[["simTime", "simFrame", "v", "lat_v_rel", "lon_v_rel", "lat_d", "lon_d", "curvHor", "rollRel", "pitchRel"]].copy()
    ego_df = df[df['playerId'] == 1][["simTime", "simFrame", "v", "curvHor", "rollRel", "pitchRel"]].copy()
    if "lat_v_rel" in df.columns:
        rel_df = df[df['playerId'] == target_id][
            ["simTime", "simFrame", "lat_v_rel", "lon_v_rel", "lat_d", "lon_d"]].copy()
        result = pd.merge(ego_df, rel_df, on=['simTime', 'simFrame'], how='left')
    else:
        result = ego_df.copy()
        result["lat_v_rel"] = ["-"] * len(result)
        result["lon_v_rel"] = ["-"] * len(result)
        result["lat_d"] = ["-"] * len(result)
        result["lon_d"] = ["-"] * len(result)

    result.rename(
        columns={"v": "speed", "lat_v_rel": "latSpeedRel", "lon_v_rel": "lonSpeedRel", "lat_d": "latDistanceRel",
                 "lon_d": "lonDistanceRel"}, inplace=True)

    result = result[result['simFrame'] > 0]
    # simTime, simFrame, speed, latSpeedRel, lonSpeedRel,
    # latDistanceRel, lonDistanceRel, curvHor, rollRel, pitchRel

    return result


def single_case_statistic(case_dict):
    """
    This function add the basic infos to the case_dict.

    Arguments:
        case_dict: A dict of single case scores and descriptions.
        config: A dict of config infos, contains basic infos.

    Returns:
        single_case_dict: A full dict of single case's performance.
    """
    single_case_dict = case_dict.copy()
    single_case_dict['testMileage'] = mileage_format(single_case_dict['testMileage'])
    single_case_dict['testDuration'] = duration_format(single_case_dict['testDuration'])

    # single_case_dict = {
    #     'testMileage': mileage,
    #     'testDuration': duration,
    #     'algorithmResultDescription1': algorithmResultDescription1,
    #     'algorithmResultDescription2': algorithmResultDescription2,
    #     'algorithmComprehensiveScore': score_case,
    #     'algorithmLevel': grade_case,
    #     'safe': safe_report_dict,
    #     'function': func_report_dict,
    #     'compliance': comp_report_dict,
    #     'comfort': comf_report_dict,
    #     'efficient': effi_report_dict
    # }

    return single_case_dict


def single_report_post(single_case_dict, case_path):
    """
    This function generate the single case report based on single_case_dict.

    Arguments:
        single_case_dict: A dict of single case scores and descriptions.
        case_path: A str of path of files, which says where the generated report is to be stored.

    Returns:
        None
    """
    print('准备发送请求：')
    report_name = case_path.split('\\')[-2]  # get case name for report name
    url_json = 'http://36.110.106.156:18080/report/generate'
    data_json = json.dumps(single_case_dict)
    response = requests.post(url_json, data_json, headers={'Content-Type': 'application/json; charset=utf-8'})
    print("返回报告结果：", response)
    runtime = time.strftime('%Y%m%d%H%M%S', time.localtime())
    p = pathlib.Path(rf'{case_path}\{report_name}_{runtime}.pdf')
    p.write_bytes(response.content)