cicv_evaluate_master-multiple_eval/config_parser.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/02
@Last Modified:     2023/11/02
@Summary:           This module provides the function to parse the config json file.
"""

import pandas as pd

from common import json2dict


class ConfigParse(object):
    """

    """

    def __init__(self, json_file):
        # weight info
        self.scoreModel = ""
        self.dimension_weight = {}
        self.dimension_list = []
        self.dimension_name = {}
        self.type_weight = {}
        self.type_list = []
        self.type_name = {}
        self.metric_list = []
        self.metric_dict = {}
        self.name_dict = {}
        self.unit_dict = {}

        self.builtinDimensionList = ["safe", "function", "compliance", "comfort", "efficient"]
        self.builtinTypeList = []
        self.builtinMetricList = ["TTC", "MTTC", "THW", "LatSD", "LonSD", "DRAC", "BTN", "STN", "collisionRisk",
                                  "collisionSeverity", "followSpeedDeviation", "followDistanceDeviation",
                                  "followStopDistance", "followResponseTime", 'laneDistance',
                                  'centerDistanceExpectation', 'centerDistanceStandardDeviation',
                                  'centerDistanceMax', 'centerDistanceMin', 'centerDistanceFrequency',
                                  'centerDistanceRange', "zigzag", "shake", "cadence", "slamBrake", "slamAccelerate",
                                  "averageSpeed", "stopDuration", "stopCount", "overspeed10", "overspeed10_20",
                                  "pressSolidLine", "runRedLight", "overspeed20_50", "overspeed50"]

        # score info
        self.config = {}

        # initialization
        self.config_dict = json2dict(json_file)
        self.parse_dict = self._config_parse(self.config_dict)

    def _config_parse(self, config_dict):
        # score
        if 'scoreModel' in list(config_dict.keys()):
            self.scoreModel = config_dict['scoreModel']
        else:
            self.scoreModel = "builtin"

        # dimension info
        dimensionWeight = config_dict['dimensionWeight']
        self.dimension_weight = dimensionWeight
        self.dimension_name = config_dict['dimensionName']
        self.dimension_list = list(self.dimension_weight.keys())

        # type info
        typeWeight = config_dict['typeWeight']
        self.type_weight = typeWeight
        self.type_name = config_dict["typeName"]

        for dimension in self.dimension_list:
            if dimension == 'compliance':
                self.config[dimension] = self._compliance_config_parse(config_dict, typeWeight, dimensionWeight)
            else:
                self.config[dimension] = self._dimension_config_parse(dimension, config_dict, typeWeight,
                                                                      dimensionWeight)

            # self.name_dict[dimension] = self.config[dimension]['name']
            # self.unit_dict[dimension] = self.config[dimension]['unit']
            self.name_dict.update(self.config[dimension]['name'])
            self.unit_dict.update(self.config[dimension]['unit'])

            self.metric_dict[dimension] = self.config[dimension]['typeMetricDict']
            self.metric_list.extend(self.config[dimension]['metric'])
            self.type_list.extend(self.config[dimension]['type'])

        print()

    def _dimension_config_parse(self, dimension, config_dict, typeWeight, dimensionWeight):

        # get weight type
        typeWeightDimension = typeWeight[dimension]
        typeDimension = list(typeWeightDimension.keys())  # get type list
        typeWeightDimensionList = list(typeWeightDimension.values())
        flagCustomDimension = not all(x is None for x in typeWeightDimensionList)

        # get type name
        typeNameDimension = self.type_name[dimension]

        # Dimension
        dimension_dict = config_dict[dimension]
        dimension_value_dict = {}
        typeMetricDict = {}

        for type in typeDimension:
            dimension_value_dict.update(dimension_dict[type])
            typeMetricDict[type] = list(dimension_dict[type].keys())

        df_dimension_value = pd.DataFrame(dimension_value_dict).T

        # get metric list
        metricDimension = df_dimension_value.index.tolist()

        # get name list
        nameDimension = df_dimension_value['name'].to_dict()
        # nameDimensionList = list(nameDimension.values())

        # get unit list
        unitDimension = df_dimension_value['unit'].to_dict()
        # unitDimensionList = list(unitDimension.values())

        # get weight list
        weightDimension = df_dimension_value['weight'].to_dict()
        weightDimensionList = list(weightDimension.values())

        # get priority list
        priorityDimension = df_dimension_value['priority'].to_dict()
        priorityDimensionList = list(priorityDimension.values())

        # get paramList
        paramDimension = df_dimension_value['paramList'].to_dict()

        bulitin_first_key = next((key for key in paramDimension if key in self.builtinMetricList), None)
        first_key = next(iter(paramDimension)) if not bulitin_first_key else bulitin_first_key
        paramNum = len(paramDimension[first_key])

        kindDimension = [{} for _ in range(paramNum)]
        optimalDimension = [{} for _ in range(paramNum)]
        multipleDimension = [{} for _ in range(paramNum)]
        spareDimension = [{} for _ in range(paramNum)]
        # spare1Dimension = [{} for _ in range(paramNum)]
        # spare2Dimension = [{} for _ in range(paramNum)]

        customMetricParam = {}  # custiom metric paramList

        for key, value_list in paramDimension.items():
            if key in self.builtinMetricList:
                for i in range(len(value_list)):
                    kindDimension[i][key] = int(value_list[i]['kind'])
                    optimalDimension[i][key] = (value_list[i]['optimal'])
                    multipleDimension[i][key] = [float(x) for x in value_list[i]['multiple']]
                    spareDimension[i][key] = [item["param"] for item in value_list[i]["spare"]]
                    # spareDimension[i][key] = [float(item["param"]) for item in value_list[i]["spare"]]
                    # spare1Dimension[i][key] = (value_list[i]['spare1'])
                    # spare2Dimension[i][key] = (value_list[i]['spare2'])
            else:
                customMetricParam[key] = value_list

        kindDimensionList = [value for dict_val in kindDimension for value in dict_val.values()]
        optimalDimensionList = [value for dict_val in optimalDimension for value in dict_val.values()]
        multipleDimensionList = [value for dict_val in multipleDimension for value in dict_val.values()]
        spareDimensionList = [value for dict_val in spareDimension for value in dict_val.values()]
        # spare1DimensionList = [value for dict_val in spare1Dimension for value in dict_val.values()]
        # spare2DimensionList = [value for dict_val in spare2Dimension for value in dict_val.values()]

        if paramNum == 1:
            kindDimension = kindDimension[0]
            optimalDimension = optimalDimension[0]
            multipleDimension = multipleDimension[0]
            spareDimension = spareDimension[0]
            # spare1Dimension = spare1Dimension[0]
            # spare2Dimension = spare2Dimension[0]

        result = {
            "weightDimension": float(dimensionWeight[dimension]),
            "weightCustom": flagCustomDimension,
            "type": typeDimension,
            "typeWeight": typeWeightDimension,
            "typeWeightList": typeWeightDimensionList,
            "typeName": typeNameDimension,
            "customMetricParam": customMetricParam,
            "metric": metricDimension,
            "typeMetricDict": typeMetricDict,
            "name": nameDimension,
            # "nameList": nameDimensionList,
            "unit": unitDimension,
            # "unitList": unitDimensionList,
            "weight": weightDimension,
            "weightList": weightDimensionList,
            "priority": priorityDimension,
            "priorityList": priorityDimensionList,
            "kind": kindDimension,
            "kindList": kindDimensionList,
            "optimal": optimalDimension,
            "optimalList": optimalDimensionList,
            "multiple": multipleDimension,
            "multipleList": multipleDimensionList,
            "spare": spareDimension,
            "spareList": spareDimensionList,
            # "spare1": spare1Dimension,
            # "spare1List": spare1DimensionList,
            # "spare2": spare2Dimension,
            # "spare2List": spare2DimensionList
        }
        return result

    def _compliance_config_parse(self, config_dict, typeWeight, dimensionWeight):
        # get weight type
        typeWeightCompliance = typeWeight['compliance']
        typeCompliance = list(typeWeightCompliance.keys())
        typeWeightComplianceList = list(typeWeightCompliance.values())
        flagCustomCompliance = not all(x is None for x in typeWeightComplianceList)

        # get type name
        typeNameCompliance = self.type_name["compliance"]

        # compliance
        compliance_dict = config_dict['compliance']
        compliance_value_dict = {}
        typeMetricDict = {}

        for type in typeCompliance:
            compliance_value_dict.update(compliance_dict[type])
            typeMetricDict[type] = list(compliance_dict[type].keys())

        df_compliance_value = pd.DataFrame(compliance_value_dict).T

        # get metric list
        metricCompliance = df_compliance_value.index.tolist()

        # get weight type
        typeWeightCompliance = typeWeight['compliance']
        typeWeightComplianceList = list(typeWeightCompliance.values())

        # get name list
        nameCompliance = df_compliance_value['name'].to_dict()

        # get unit list
        unitCompliance = df_compliance_value['unit'].to_dict()

        # get weight list
        weightCompliance = df_compliance_value['weight'].to_dict()
        weightComplianceList = list(weightCompliance.values())

        compliance = {
            "weightDimension": float(dimensionWeight['compliance']),
            "weightCustom": flagCustomCompliance,
            "type": typeCompliance,
            "typeWeight": typeWeightCompliance,
            "typeWeightList": typeWeightComplianceList,
            "typeName": typeNameCompliance,
            "typeMetricDict": typeMetricDict,
            "metric": metricCompliance,
            "name": nameCompliance,
            "unit": unitCompliance,
            "weight": weightCompliance,
            "weightList": weightComplianceList
        }
        return compliance