common/metric.py

'''
Author: Qiguang Chen
Date: 2023-01-11 10:39:26
LastEditors: Qiguang Chen
LastEditTime: 2023-02-17 19:39:22
Description: Metric calculation class

'''
from collections import Counter
from typing import List, Dict

import numpy as np
from sklearn.metrics import f1_score

from common.utils import InputData, OutputData


class Evaluator(object):
    """Evaluation metric funtions library class
        supported metric:
        - slot_f1
        - intent_acc
        - exactly_match_accuracy
        - intent_f1 (defult "macro_intent_f1")
            - macro_intent_f1
            - micro_intent_f1=
    """
    @staticmethod
    def exactly_match_accuracy(pred_slot: List[List[str or int]],
                               real_slot: List[List[str or int]],
                               pred_intent: List[List[str or int] or str or int],
                               real_intent: List[List[str or int] or str or int]) -> float:
        """Compute the accuracy based on the whole predictions of given sentence, including slot and intent.
            (both support str or int index as the representation of slot and intent)
        Args:
            pred_slot (List[List[str or int]]): predicted sequence of slot list
            real_slot (List[List[str or int]]): golden sequence of slot list.
            pred_intent (List[List[str or int] or str or int]): golden intent list / golden multi intent list.
            real_intent (List[List[str or int] or str or int]): predicted intent list / predicted multi intent list.

        Returns:
            float: exactly match accuracy score
        """
        total_count, correct_count = 0.0, 0.0
        for p_slot, r_slot, p_intent, r_intent in zip(pred_slot, real_slot, pred_intent, real_intent):
            if isinstance(p_intent, list):
                p_intent, r_intent = set(p_intent), set(r_intent)
            if p_slot == r_slot and p_intent == r_intent:
                correct_count += 1.0
            total_count += 1.0

        return 1.0 * correct_count / total_count


    @staticmethod
    def intent_accuracy(pred_list: List, real_list: List) -> float:
        """Get  intent accuracy measured by predictions and ground-trues. Support both multi intent and single intent.

        Args:
            pred_list (List): predicted intent list
            real_list (List): golden intent list

        Returns:
            float: intent accuracy score
        """
        total_count, correct_count = 0.0, 0.0
        for p_intent, r_intent in zip(pred_list, real_list):
            if isinstance(p_intent, list):
                p_intent, r_intent = set(p_intent), set(r_intent)
            if p_intent == r_intent:
                correct_count += 1.0
            total_count += 1.0

        return 1.0 * correct_count / total_count

    @staticmethod
    def intent_f1(pred_list: List[List[int]], real_list: List[List[int]], num_intent: int, average='macro') -> float:
        """Get  intent accuracy measured by predictions and ground-trues. Support both multi intent and single intent.
        (Only support multi intent now, but you can use [[intent1], [intent2], ...] to compute intent f1 in single intent)
        Args:
            pred_list (List[List[int]]): predicted multi intent list.
            real_list (List[List[int]]): golden multi intent list.
            num_intent (int)
            average (str): support "micro" and "macro"

        Returns:
            float: intent accuracy score
        """
        return f1_score(Evaluator.__instance2onehot(num_intent, real_list),
                        Evaluator.__instance2onehot(num_intent, pred_list),
                        average=average,
                        zero_division=0)

    @staticmethod
    def __multilabel2one_hot(labels, nums):
        res = [0.] * nums
        if len(labels) == 0:
            return res
        if isinstance(labels[0], list):
            for label in labels[0]:
                res[label] = 1.
            return res
        for label in labels:
            res[label] = 1.
        return res

    @staticmethod
    def __instance2onehot(num_intent, data):
        res = []
        for intents in data:
            res.append(Evaluator.__multilabel2one_hot(intents, num_intent))
        return np.array(res)

    @staticmethod
    def __startOfChunk(prevTag, tag, prevTagType, tagType, chunkStart=False):
        if prevTag == 'B' and tag == 'B':
            chunkStart = True
        if prevTag == 'I' and tag == 'B':
            chunkStart = True
        if prevTag == 'O' and tag == 'B':
            chunkStart = True
        if prevTag == 'O' and tag == 'I':
            chunkStart = True

        if prevTag == 'E' and tag == 'E':
            chunkStart = True
        if prevTag == 'E' and tag == 'I':
            chunkStart = True
        if prevTag == 'O' and tag == 'E':
            chunkStart = True
        if prevTag == 'O' and tag == 'I':
            chunkStart = True

        if tag != 'O' and tag != '.' and prevTagType != tagType:
            chunkStart = True
        return chunkStart

    @staticmethod
    def __endOfChunk(prevTag, tag, prevTagType, tagType, chunkEnd=False):
        if prevTag == 'B' and tag == 'B':
            chunkEnd = True
        if prevTag == 'B' and tag == 'O':
            chunkEnd = True
        if prevTag == 'I' and tag == 'B':
            chunkEnd = True
        if prevTag == 'I' and tag == 'O':
            chunkEnd = True

        if prevTag == 'E' and tag == 'E':
            chunkEnd = True
        if prevTag == 'E' and tag == 'I':
            chunkEnd = True
        if prevTag == 'E' and tag == 'O':
            chunkEnd = True
        if prevTag == 'I' and tag == 'O':
            chunkEnd = True

        if prevTag != 'O' and prevTag != '.' and prevTagType != tagType:
            chunkEnd = True
        return chunkEnd

    @staticmethod
    def __splitTagType(tag):
        s = tag.split('-')
        if len(s) > 2 or len(s) == 0:
            raise ValueError('tag format wrong. it must be B-xxx.xxx')
        if len(s) == 1:
            tag = s[0]
            tagType = ""
        else:
            tag = s[0]
            tagType = s[1]
        return tag, tagType

    @staticmethod
    def computeF1Score(correct_slots: List[List[str]], pred_slots: List[List[str]]) -> float:
        """compute f1 score is modified from conlleval.pl

        Args:
            correct_slots (List[List[str]]): golden slot string list
            pred_slots (List[List[str]]): predicted slot string list

        Returns:
            float: slot f1 score
        """
        correctChunk = {}
        correctChunkCnt = 0.0
        foundCorrect = {}
        foundCorrectCnt = 0.0
        foundPred = {}
        foundPredCnt = 0.0
        correctTags = 0.0
        tokenCount = 0.0
        for correct_slot, pred_slot in zip(correct_slots, pred_slots):
            inCorrect = False
            lastCorrectTag = 'O'
            lastCorrectType = ''
            lastPredTag = 'O'
            lastPredType = ''
            for c, p in zip(correct_slot, pred_slot):
                c = str(c)
                p = str(p)
                correctTag, correctType = Evaluator.__splitTagType(c)
                predTag, predType = Evaluator.__splitTagType(p)

                if inCorrect == True:
                    if Evaluator.__endOfChunk(lastCorrectTag, correctTag, lastCorrectType, correctType) == True and \
                            Evaluator.__endOfChunk(lastPredTag, predTag, lastPredType, predType) == True and \
                            (lastCorrectType == lastPredType):
                        inCorrect = False
                        correctChunkCnt += 1.0
                        if lastCorrectType in correctChunk:
                            correctChunk[lastCorrectType] += 1.0
                        else:
                            correctChunk[lastCorrectType] = 1.0
                    elif Evaluator.__endOfChunk(lastCorrectTag, correctTag, lastCorrectType, correctType) != \
                            Evaluator.__endOfChunk(lastPredTag, predTag, lastPredType, predType) or \
                            (correctType != predType):
                        inCorrect = False

                if Evaluator.__startOfChunk(lastCorrectTag, correctTag, lastCorrectType, correctType) == True and \
                        Evaluator.__startOfChunk(lastPredTag, predTag, lastPredType, predType) == True and \
                        (correctType == predType):
                    inCorrect = True

                if Evaluator.__startOfChunk(lastCorrectTag, correctTag, lastCorrectType, correctType) == True:
                    foundCorrectCnt += 1
                    if correctType in foundCorrect:
                        foundCorrect[correctType] += 1.0
                    else:
                        foundCorrect[correctType] = 1.0

                if Evaluator.__startOfChunk(lastPredTag, predTag, lastPredType, predType) == True:
                    foundPredCnt += 1.0
                    if predType in foundPred:
                        foundPred[predType] += 1.0
                    else:
                        foundPred[predType] = 1.0

                if correctTag == predTag and correctType == predType:
                    correctTags += 1.0

                tokenCount += 1.0

                lastCorrectTag = correctTag
                lastCorrectType = correctType
                lastPredTag = predTag
                lastPredType = predType

            if inCorrect == True:
                correctChunkCnt += 1.0
                if lastCorrectType in correctChunk:
                    correctChunk[lastCorrectType] += 1.0
                else:
                    correctChunk[lastCorrectType] = 1.0

        if foundPredCnt > 0:
            precision = 1.0 * correctChunkCnt / foundPredCnt
        else:
            precision = 0

        if foundCorrectCnt > 0:
            recall = 1.0 * correctChunkCnt / foundCorrectCnt
        else:
            recall = 0

        if (precision + recall) > 0:
            f1 = (2.0 * precision * recall) / (precision + recall)
        else:
            f1 = 0

        return f1

    @staticmethod
    def max_freq_predict(sample):
        """Max frequency prediction.
        """
        predict = []
        for items in sample:
            predict.append(Counter(items).most_common(1)[0][0])
        return predict

    @staticmethod
    def __token_map(indexes, token_label_map):
        return [[token_label_map[idx] if idx in token_label_map else -1 for idx in index] for index in indexes]

    @staticmethod
    def compute_all_metric(inps: InputData,
                           output: OutputData,
                           intent_label_map: dict = None,
                           metric_list: List=None)-> Dict:
        """Auto compute all metric mentioned in 'metric_list'
        
        Args:
            inps (InputData): input golden slot and intent labels
            output (OutputData): output predicted slot and intent labels
            intent_label_map (dict, Optional): dict like {"intent1": 0, "intent2": 1, ...},which aims to map intent string to index
            metric_list (List): support metrics in ["slot_f1", "intent_acc", "intent_f1", "macro_intent_f1", "micro_intent_f1", "EMA"]

        Returns:
            Dict: all metric mentioned in 'metric_list', like {'EMA': 0.7, ...}
            
        
        Example:
            if compute slot metric:
            
                inps.slot = [["slot1", "slot2", ...], ...]; output.slot_ids=[["slot1", "slot2", ...], ...];
                
            if compute intent metric:
            
                [Multi Intent] inps.intent = [["intent1", "intent2", ...], ...]; output.intent_ids = [["intent1", "intent2", ...], ...] 
                
                [Single Intent] inps.intent = ["intent1", ...]; [Single Intent] output.intent_ids = ["intent1", ...]
        """
        if not metric_list:
            metric_list = ["slot_f1", "intent_acc", "EMA"]
        res_dict = {}
        use_slot = output.slot_ids is not None and len(output.slot_ids) > 0
        use_intent = output.intent_ids is not None and len(
            output.intent_ids) > 0
        if use_slot and "slot_f1" in metric_list:
            
            res_dict["slot_f1"] = Evaluator.computeF1Score(
                output.slot_ids, inps.slot)
        if use_intent and "intent_acc" in metric_list:
            res_dict["intent_acc"] = Evaluator.intent_accuracy(
                output.intent_ids, inps.intent)
            if isinstance(output.intent_ids[0], list):
                if "intent_f1" in metric_list:
                    res_dict["intent_f1"] = Evaluator.intent_f1(Evaluator.__token_map(output.intent_ids, intent_label_map),
                                                                Evaluator.__token_map(
                                                                    inps.intent, intent_label_map),
                                                                len(intent_label_map.keys()))
                elif "macro_intent_f1" in metric_list:
                    res_dict["macro_intent_f1"] = Evaluator.intent_f1(Evaluator.__token_map(output.intent_ids, intent_label_map),
                                                                      Evaluator.__token_map(inps.intent, intent_label_map),
                                                                      len(intent_label_map.keys()), average="macro")
                if "micro_intent_f1" in metric_list:
                    res_dict["micro_intent_f1"] = Evaluator.intent_f1(Evaluator.__token_map(output.intent_ids, intent_label_map),
                                                                      Evaluator.__token_map(inps.intent, intent_label_map),
                                                                      len(intent_label_map.keys()), average="micro")

        if use_slot and use_intent and "EMA" in metric_list:
            res_dict["EMA"] = Evaluator.exactly_match_accuracy(output.slot_ids, inps.slot, output.intent_ids,
                                                               inps.intent)
        return res_dict