Upload 6 files

evaluation/evaluate_utils/evaluate_dicts.py

@@ -0,0 +1,68 @@
+from typing import Dict, List
+import numpy as np
+
+from evaluation.evaluate_utils.utils import _align_bags
+
+
+def calculate_f1_score(precision, recall):
+    if precision + recall == 0:
+        return 0  # Handle the case to avoid division by zero
+    return 2 * (precision * recall) / (precision + recall)
+
+
+def calc_recall(pred: Dict, gold: Dict, use_gold_for_eval: bool):
+    from evaluation.evaluate_utils.evaluate_factory import get_evaluator_from_gold_answer
+
+    recall = []
+    for gold_key, gold_value in gold.items():
+        pred_value = pred.get(gold_key)
+        gold_value = fix_number(gold_value)
+        pred_value = fix_number(pred_value)
+        if gold_key not in pred:
+            recall.append(0)
+        else:
+            evaluator = (
+                get_evaluator_from_gold_answer(type(gold_value))
+                if use_gold_for_eval
+                else get_evaluator_from_gold_answer(type(pred_value))
+            )
+            if type(pred_value) != type(gold_value):
+                recall.append(0)
+                continue
+            recall.append(evaluator(pred_value, gold_value))
+    avg_recall = np.average(recall)
+    return avg_recall
+
+
+def fix_number(number):
+
+    if type(number) == str:
+        copy_ans = number
+        copy_ans = ' '.join(' '.join(' '.join(copy_ans.split('$')).split('%')).split('sqft')).strip()
+        copy_ans = copy_ans.strip()
+        copy_ans = copy_ans.replace(',', '.')
+        try:
+            return float(copy_ans)
+        except:
+            return number
+    elif type(number) == int:
+        return float(number)
+    else:
+        return number
+
+def evaluate_pair_of_dicts(pred: Dict, gold: Dict):
+    recall = calc_recall(pred, gold, True)
+    precision = calc_recall(gold, pred, False)
+    f1 = calculate_f1_score(precision, recall)
+    return f1
+
+
+def evaluate_dicts(pred: List[Dict], gold: List[Dict]):
+    if not (
+        type(pred) == dict
+        or len(pred) == 0
+        or (type(pred) == list and type(pred[0]) == dict)
+    ):
+        return 0
+    max_alignment_scores = _align_bags(pred, gold, evaluate_pair_of_dicts)
+    return np.average(max_alignment_scores)

evaluation/evaluate_utils/evaluate_factory.py

@@ -0,0 +1,28 @@
+from typing import Union, Dict
+
+from evaluation.evaluate_utils.evaluate_dicts import evaluate_dicts
+from evaluation.evaluate_utils.evaluate_numbers import evaluate_numbers
+from evaluation.evaluate_utils.evaluate_strings import evaluate_strings
+
+EvaluatorFactory = {
+    "string": evaluate_strings,
+    "number": evaluate_numbers,
+    "json": evaluate_dicts,
+    "string list": evaluate_strings,
+}
+
+EvaluatorFactoryFromType = {
+    str: evaluate_strings,
+    int: evaluate_numbers,
+    float: evaluate_numbers,
+    bool: evaluate_strings,
+    list: evaluate_strings
+}
+
+
+def get_evaluator(evaluator: str):
+    return EvaluatorFactory[evaluator]
+
+
+def get_evaluator_from_gold_answer(gold_answer: Union[str, int, float]):
+    return EvaluatorFactoryFromType[gold_answer]

evaluation/evaluate_utils/evaluate_numbers.py

@@ -0,0 +1,33 @@
+from typing import Union
+import numpy as np
+
+
+# Renamed calc_z function to distance_function_log
+def distance_function_log(pred: float, gold: float):
+    if pred == gold == 0:
+        return 1
+    if pred == 0:
+        pred = 1e-4
+    if gold == 0:
+        gold = 1e-4
+    if pred > gold:
+        return max(0, 1 - np.log(pred / gold))
+    else:
+        return max(0, 1 - np.log(gold / pred))
+
+
+def evaluate_numbers(pred: Union[float, str], gold: float):
+    res = None
+    if type(pred) != float and type(pred) != int:
+        try:
+            pred = float(pred)
+        except ValueError:
+            res = 0
+    if type(gold) != float and type(gold) != int:
+        try:
+            gold = float(gold)
+        except ValueError:
+            res = 0
+    if res is None:
+        res = distance_function_log(pred, gold)
+    return res

evaluation/evaluate_utils/evaluate_strings.py

@@ -0,0 +1,179 @@
+"""
+Evaluation for two strings or list of strings.
+
+Code taken from the DROP benchmark - https://github.com/allenai/allennlp-reading-comprehension/blob/master/allennlp_rc/eval/drop_eval.py
+"""
+
+from collections import defaultdict
+from typing import List, Set, Tuple, Union
+import string
+import re
+import numpy as np
+from scipy.optimize import linear_sum_assignment
+
+
+# From here through _normalize_answer was originally copied from:
+# https://worksheets.codalab.org/rest/bundles/0x6b567e1cf2e041ec80d7098f031c5c9e/contents/blob/
+# Then cleaned up and modified a bit.
+def _remove_articles(text: str) -> str:
+    regex = re.compile(r"\b(a|an|the)\b", re.UNICODE)
+    return re.sub(regex, " ", text)
+
+
+def _white_space_fix(text: str) -> str:
+    return " ".join(text.split())
+
+
+EXCLUDE = set(string.punctuation)
+
+
+def _remove_punc(text: str) -> str:
+    if not _is_number(text):
+        return "".join(ch for ch in text if ch not in EXCLUDE)
+    else:
+        return text
+
+
+def _lower(text: str) -> str:
+    return text.lower()
+
+
+def _tokenize(text: str) -> List[str]:
+    return re.split(" |-", text)
+
+
+def _normalize_answer(text: str) -> str:
+    """Lower text and remove punctuation, articles and extra whitespace."""
+
+    parts = [
+        _white_space_fix(
+            _remove_articles(_normalize_number(_remove_punc(_lower(token))))
+        )
+        for token in _tokenize(text)
+    ]
+    parts = [part for part in parts if part.strip()]
+    normalized = " ".join(parts).strip()
+    return normalized
+
+
+def _is_number(text: str) -> bool:
+    try:
+        float(text)
+        return True
+    except ValueError:
+        return False
+
+
+def _normalize_number(text: str) -> str:
+    if _is_number(text):
+        return str(float(text))
+    else:
+        return text
+
+
+def _answer_to_bags(
+    answer: Union[str, List[str], Tuple[str, ...]]
+) -> Tuple[List[str], List[Set[str]]]:
+    if isinstance(answer, (list, tuple)):
+        raw_spans = answer
+    else:
+        raw_spans = [answer]
+    normalized_spans: List[str] = []
+    token_bags = []
+    for raw_span in raw_spans:
+        normalized_span = _normalize_answer(raw_span)
+        normalized_spans.append(normalized_span)
+        token_bags.append(set(normalized_span.split()))
+    return normalized_spans, token_bags
+
+
+def _align_bags(predicted: List[Set[str]], gold: List[Set[str]]) -> List[float]:
+    """
+    Takes gold and predicted answer sets and first finds the optimal 1-1 alignment
+    between them and gets maximum metric values over all the answers.
+    """
+    scores = np.zeros([len(gold), len(predicted)])
+    for gold_index, gold_item in enumerate(gold):
+        for pred_index, pred_item in enumerate(predicted):
+            if _match_numbers_if_present(gold_item, pred_item):
+                scores[gold_index, pred_index] = _compute_f1(pred_item, gold_item)
+    row_ind, col_ind = linear_sum_assignment(-scores)
+
+    max_scores = np.zeros([max(len(gold), len(predicted))])
+    for row, column in zip(row_ind, col_ind):
+        max_scores[row] = max(max_scores[row], scores[row, column])
+    return max_scores
+
+
+def _compute_f1(predicted_bag: Set[str], gold_bag: Set[str]) -> float:
+    intersection = len(gold_bag.intersection(predicted_bag))
+    if not predicted_bag:
+        precision = 1.0
+    else:
+        precision = intersection / float(len(predicted_bag))
+    if not gold_bag:
+        recall = 1.0
+    else:
+        recall = intersection / float(len(gold_bag))
+    f1 = (
+        (2 * precision * recall) / (precision + recall)
+        if not (precision == 0.0 and recall == 0.0)
+        else 0.0
+    )
+    return f1
+
+
+def _match_numbers_if_present(gold_bag: Set[str], predicted_bag: Set[str]) -> bool:
+    gold_numbers = set()
+    predicted_numbers = set()
+    for word in gold_bag:
+        if _is_number(word):
+            gold_numbers.add(word)
+    for word in predicted_bag:
+        if _is_number(word):
+            predicted_numbers.add(word)
+    if (not gold_numbers) or gold_numbers.intersection(predicted_numbers):
+        return True
+    return False
+
+
+def get_metrics(
+    predicted: Union[str, List[str], Tuple[str, ...]],
+    gold: Union[str, List[str], Tuple[str, ...]],
+) -> Tuple[float, float]:
+    """
+    Takes a predicted answer and a gold answer (that are both either a string or a list of
+    strings), and returns exact match and the DROP F1 metric for the prediction.  If you are
+    writing a script for evaluating objects in memory (say, the output of predictions during
+    validation, or while training), this is the function you want to call, after using
+    :func:`answer_json_to_strings` when reading the gold answer from the released data file.
+    """
+    predicted_bags = _answer_to_bags(predicted)
+    gold_bags = _answer_to_bags(gold)
+
+    if set(predicted_bags[0]) == set(gold_bags[0]) and len(predicted_bags[0]) == len(
+        gold_bags[0]
+    ):
+        exact_match = 1.0
+    else:
+        exact_match = 0.0
+
+    f1_per_bag = _align_bags(predicted_bags[1], gold_bags[1])
+    f1 = np.mean(f1_per_bag)
+    f1 = round(f1, 2)
+    return exact_match, f1
+
+
+def evaluate_strings(prediction, gold):
+    if type(prediction) != list and type(prediction) != str:
+        prediction = str(prediction)
+    if type(gold) != list and type(gold) != str:
+        gold = str(gold)
+    try:
+        predicted_bags = _answer_to_bags(prediction)
+        gold_bags = _answer_to_bags(gold)
+        f1_per_bag = _align_bags(predicted_bags[1], gold_bags[1])
+        f1 = np.mean(f1_per_bag)
+    except Exception:
+        f1 = 0.0
+    return f1

evaluation/evaluate_utils/utils.py

@@ -0,0 +1,24 @@
+from typing import List, Set, Tuple, Union, Callable
+import numpy as np
+from scipy.optimize import linear_sum_assignment
+
+
+def _align_bags(
+    predicted: List[Set[str]],
+    gold: List[Set[str]],
+    method: Callable[[object, object], float],
+) -> List[float]:
+    """
+    Takes gold and predicted answer sets and first finds the optimal 1-1 alignment
+    between them and gets maximum metric values over all the answers.
+    """
+    scores = np.zeros([len(gold), len(predicted)])
+    for gold_index, gold_item in enumerate(gold):
+        for pred_index, pred_item in enumerate(predicted):
+            scores[gold_index, pred_index] = method(pred_item, gold_item)
+    row_ind, col_ind = linear_sum_assignment(-scores)
+
+    max_scores = np.zeros([max(len(gold), len(predicted))])
+    for row, column in zip(row_ind, col_ind):
+        max_scores[row] = max(max_scores[row], scores[row, column])
+    return max_scores

evaluation/evaluator.py

@@ -0,0 +1,89 @@
+import json
+from evaluation.evaluate_utils.evaluate_factory import get_evaluator
+
+
+def fix_ans(answer):
+
+    try:
+        answer = answer.replace("{'", '{"').replace("', '", '", "').replace("': '", '": "').replace("'}", '"}')
+        answer = answer.replace("': ", '": ')
+        return answer
+    except:
+        return answer
+
+
+def parse_answer(answer):
+
+    if len(answer) == 1:
+        if answer[0].isnumeric():
+            ans, is_num = fix_number(answer[0])
+            if is_num:
+                return ans, 'number'
+        try:
+            ans = json.loads(fix_ans(answer[0]))
+            return [ans], 'json'
+        except:
+            ans, is_num = fix_number(answer[0])
+            if is_num:
+                return ans, 'number'
+            else:
+                return answer[0], 'string'
+    else:
+        try:
+            ans = [json.loads(fix_ans(ex)) for ex in answer]
+            return ans, 'json'
+        except:
+            return answer, "string list"
+
+
+def fix_number(number):
+
+    if type(number) == str:
+        copy_ans = number
+        copy_ans = ' '.join(' '.join(' '.join(copy_ans.split('$')).split('%')).split('sqft')).strip()
+        copy_ans = copy_ans.strip()
+        copy_ans = copy_ans.replace(',', '.').replace(' square kilometers', '')
+        try:
+            return float(copy_ans), True
+        except:
+            return number, False
+    elif type(number) == int:
+        return float(number), True
+    else:
+        return number, True
+
+
+def fix_prediction(prediction, gold_answer, evaluator):
+
+    if type(prediction) == list and len(prediction) == 1 and (type(prediction[0]) == int or ((type(prediction[0]) == str) and prediction[0].isnumeric())):
+        prediction = fix_number(prediction[0])
+
+    if type(prediction) != list:
+        prediction, is_num = fix_number(prediction)
+        if evaluator == 'json':
+            try:
+                prediction = [json.loads(pred) for pred in prediction.split('\n')]
+            except:
+                prediction = [prediction]
+
+    if (hasattr(type(prediction), '__len__')) and (len(prediction) == 0):
+        return prediction, False
+
+    if (type(prediction) == list and len(prediction) > 1) and type(gold_answer) == float:
+        return prediction, False
+
+    return prediction, True
+
+
+def question_scorer(prediction, gold_answer):
+
+    answer_list = [x for x in gold_answer.split("\n") if len(x.strip()) > 0] if type(gold_answer) != list else gold_answer
+    gold_answer, evaluator = parse_answer(answer_list)
+    prediction, run_eval = fix_prediction(prediction, gold_answer, evaluator)
+
+    if not run_eval:
+        return 0.
+
+    metric_eval = get_evaluator(evaluator)
+    accuracy = metric_eval(prediction, gold_answer)
+    return accuracy