test

absa_evaluator.py

@@ -0,0 +1,166 @@
+from typing import Dict, List
+
+import evaluate
+from datasets import Features, Sequence, Value
+from sklearn.metrics import accuracy_score
+
+from preprocessing import absa_term_preprocess
+
+_CITATION = """
+"""
+
+_DESCRIPTION = """
+Evaluation metrics for Aspect-Based Sentiment Analysis (ABSA) including precision, recall, and F1 score for aspect terms and polarities.
+"""
+
+_KWARGS_DESCRIPTION = """
+Computes precision, recall, and F1 score for aspect terms and polarities in Aspect-Based Sentiment Analysis (ABSA).
+
+Args:
+    predictions: List of ABSA predictions with the following structure:
+        - 'aspects': Sequence of aspect annotations, each with the following keys:
+            - 'term': Aspect term
+            - 'polarity': Polarity of the aspect term
+    references: List of ABSA references with the same structure as predictions.
+Returns:
+    aspect_precision: Precision score for aspect terms
+    aspect_recall: Recall score for aspect terms
+    aspect_f1: F1 score for aspect terms
+    polarity_precision: Precision score for aspect polarities
+    polarity_recall: Recall score for aspect polarities
+    polarity_f1: F1 score for aspect polarities
+"""
+
+
+class AbsaEvaluatorTest(evaluate.Metric):
+    def _info(self):
+        return evaluate.MetricInfo(
+            description=_DESCRIPTION,
+            citation=_CITATION,
+            inputs_description=_KWARGS_DESCRIPTION,
+            features=Features(
+                {
+                    "predictions": Features(
+                        {
+                            "aspects": Features(
+                                {
+                                    "term": Sequence(Value("string")),
+                                    "polarity": Sequence(Value("string")),
+                                }
+                            ),
+                            "category": Features(
+                                {
+                                    "category": Sequence(Value("string")),
+                                    "polarity": Sequence(Value("string")),
+                                }
+                            ),
+                        }
+                    ),
+                    "references": Features(
+                        {
+                            "aspects": Features(
+                                {
+                                    "term": Sequence(Value("string")),
+                                    "polarity": Sequence(Value("string")),
+                                }
+                            ),
+                            "category": Features(
+                                {
+                                    "category": Sequence(Value("string")),
+                                    "polarity": Sequence(Value("string")),
+                                }
+                            ),
+                        }
+                    ),
+                }
+            ),
+        )
+
+    def _compute(self, predictions, references):
+        # preprocess aspect term
+        (
+            truth_aspect_terms,
+            pred_aspect_terms,
+            truth_term_polarities,
+            pred_term_polarities,
+        ) = absa_term_preprocess(
+            references=references,
+            predictions=predictions,
+            subtask_key="aspects",
+            subtask_value="term",
+        )
+        # evaluate
+        term_results = self.semeval_metric(
+            truth_aspect_terms, pred_aspect_terms
+        )
+        term_polarity_acc = accuracy_score(
+            truth_term_polarities, pred_term_polarities
+        )
+
+        # preprocess category detection
+        (
+            truth_categories,
+            pred_categories,
+            truth_cat_polarities,
+            pred_cat_polarities,
+        ) = absa_term_preprocess(
+            references=references,
+            predictions=predictions,
+            subtask_key="category",
+            subtask_value="category",
+        )
+
+        # evaluate
+        category_results = self.semeval_metric(
+            truth_categories, pred_categories
+        )
+        cat_polarity_acc = accuracy_score(
+            truth_cat_polarities, pred_cat_polarities
+        )
+
+        return {
+            "term_extraction_results": term_results,
+            "term_polarity_results_accuracy": term_polarity_acc,
+            "category_detection_results": category_results,
+            "category_polarity_results_accuracy": cat_polarity_acc,
+        }
+
+    def semeval_metric(
+        self, truths: List[List[str]], preds: List[List[str]]
+    ) -> Dict[str, float]:
+        """
+        Implements evaluation for extraction tasks using precision, recall, and F1 score.
+
+        Parameters:
+        - truths: List of lists, where each list contains the ground truth labels for a sample.
+        - preds: List of lists, where each list contains the predicted labels for a sample.
+
+        Returns:
+        - A dictionary containing the precision, recall, F1 score, and counts of common, retrieved, and relevant.
+
+        link for code: link for this code: https://github.com/davidsbatista/Aspect-Based-Sentiment-Analysis/blob/1d9c8ec1131993d924e96676fa212db6b53cb870/libraries/baselines.py#L387
+        """
+        b = 1
+        common, relevant, retrieved = 0.0, 0.0, 0.0
+        for truth, pred in zip(truths, preds):
+            common += len([a for a in pred if a in truth])
+            retrieved += len(pred)
+            relevant += len(truth)
+        precision = common / retrieved if retrieved > 0 else 0.0
+        recall = common / relevant if relevant > 0 else 0.0
+        f1 = (
+            (1 + (b**2))
+            * precision
+            * recall
+            / ((precision * b**2) + recall)
+            if precision > 0 and recall > 0
+            else 0.0
+        )
+        return {
+            "precision": precision,
+            "recall": recall,
+            "f1_score": f1,
+            "common": common,
+            "retrieved": retrieved,
+            "relevant": relevant,
+        }

app.py

@@ -0,0 +1,5 @@
+import evaluate
+from gradio_tst import launch_gradio_widget2
+
+module = evaluate.load("absa_evaluator.py")
+launch_gradio_widget2(module)

gradio_tst.py

@@ -0,0 +1,130 @@
+import json
+import os
+import re
+import sys
+from pathlib import Path
+
+import numpy as np
+from datasets import Value
+
+import logging
+
+
+
+REGEX_YAML_BLOCK = re.compile(r"---[\n\r]+([\S\s]*?)[\n\r]+---[\n\r]")
+
+
+def infer_gradio_input_types(feature_types):
+    """
+    Maps metric feature types to input types for gradio Dataframes:
+        - float/int -> numbers
+        - string -> strings
+        - any other -> json
+    Note that json is not a native gradio type but will be treated as string that
+    is then parsed as a json.
+    """
+    input_types = []
+    for feature_type in feature_types:
+        input_type = "json"
+        if isinstance(feature_type, Value):
+            if feature_type.dtype.startswith("int") or feature_type.dtype.startswith("float"):
+                input_type = "number"
+            elif feature_type.dtype == "string":
+                input_type = "str"
+        input_types.append(input_type)
+    return input_types
+
+
+def json_to_string_type(input_types):
+    """Maps json input type to str."""
+    return ["str" if i == "json" else i for i in input_types]
+
+
+def parse_readme(filepath):
+    """Parses a repositories README and removes"""
+    if not os.path.exists(filepath):
+        return "No README.md found."
+    with open(filepath, "r") as f:
+        text = f.read()
+        match = REGEX_YAML_BLOCK.search(text)
+        if match:
+            text = text[match.end() :]
+    return text
+
+
+def parse_gradio_data(data, input_types):
+    """Parses data from gradio Dataframe for use in metric."""
+    metric_inputs = {}
+    data.replace("", np.nan, inplace=True)
+    data.dropna(inplace=True)
+    for feature_name, input_type in zip(data, input_types):
+        if input_type == "json":
+            metric_inputs[feature_name] = [json.loads(d) for d in data[feature_name].to_list()]
+        elif input_type == "str":
+            metric_inputs[feature_name] = [d.strip('"') for d in data[feature_name].to_list()]
+        else:
+            metric_inputs[feature_name] = data[feature_name]
+    return metric_inputs
+
+
+def parse_test_cases(test_cases, feature_names, input_types):
+    """
+    Parses test cases to be used in gradio Dataframe. Note that an apostrophe is added
+    to strings to follow the format in json.
+    """
+    if len(test_cases) == 0:
+        return None
+    examples = []
+    for test_case in test_cases:
+        parsed_cases = []
+        for feat, input_type in zip(feature_names, input_types):
+            if input_type == "json":
+                parsed_cases.append([str(element) for element in test_case[feat]])
+            elif input_type == "str":
+                parsed_cases.append(['"' + element + '"' for element in test_case[feat]])
+            else:
+                parsed_cases.append(test_case[feat])
+        examples.append([list(i) for i in zip(*parsed_cases)])
+    return examples
+
+
+def launch_gradio_widget2(metric):
+    """Launches `metric` widget with Gradio."""
+
+    try:
+        import gradio as gr
+    except ImportError as error:
+        logging.error("To create a metric widget with Gradio make sure gradio is installed.")
+        raise error
+
+    local_path = Path(sys.path[0])
+    # if there are several input types, use first as default.
+    if isinstance(metric.features, list):
+        (feature_names, feature_types) = zip(*metric.features[0].items())
+    else:
+        (feature_names, feature_types) = zip(*metric.features.items())
+    gradio_input_types = infer_gradio_input_types(feature_types)
+
+    def compute(data):
+        return metric.compute(**parse_gradio_data(data, gradio_input_types))
+
+    iface = gr.Interface(
+        fn=compute,
+        inputs=gr.Dataframe(
+            headers=feature_names,
+            col_count=len(feature_names),
+            row_count=1,
+            datatype=json_to_string_type(gradio_input_types),
+        ),
+        outputs=gr.Textbox(label=metric.name),
+        description=(
+            metric.info.description + "\nIf this is a text-based metric, make sure to wrap you input in double quotes."
+            " Alternatively you can use a JSON-formatted list as input."
+        ),
+        title=f"Metric: {metric.name}",
+        article=parse_readme(local_path / "README.md"),
+        # TODO: load test cases and use them to populate examples
+        # examples=[parse_test_cases(test_cases, feature_names, gradio_input_types)]
+    )
+
+    iface.launch()

preprocessing.py

@@ -0,0 +1,115 @@
+from itertools import chain
+from random import choice
+from typing import Any, Dict, List, Optional, Tuple
+
+from datasets import Dataset
+
+
+def adjust_predictions(refs, preds, choices):
+    """Adjust predictions to match the length of references with either a special token or random choice."""
+    adjusted_preds = []
+    for ref, pred in zip(refs, preds):
+        if len(pred) < len(ref):
+            missing_count = len(ref) - len(pred)
+            pred.extend([choice(choices) for _ in range(missing_count)])
+        adjusted_preds.append(pred)
+    return adjusted_preds
+
+
+def extract_aspects(data, specific_key, specific_val):
+    """Extracts and returns a list of specified aspect details from the nested 'aspects' data."""
+    return [item[specific_key][specific_val] for item in data]
+
+
+def absa_term_preprocess(references, predictions, subtask_key, subtask_value):
+    """
+    Preprocess the terms and polarities for aspect-based sentiment analysis.
+
+    Args:
+        references (List[Dict]): A list of dictionaries containing the actual terms and polarities under 'aspects'.
+        predictions (List[Dict]): A list of dictionaries containing predicted aspect categories to terms and their sentiments.
+
+    Returns:
+        Tuple[List[str], List[str], List[str], List[str]]: A tuple containing lists of true aspect terms,
+        adjusted predicted aspect terms, true polarities, and adjusted predicted polarities.
+    """
+
+    # Extract aspect terms and polarities
+    truth_aspect_terms = extract_aspects(references, subtask_key, subtask_value)
+    pred_aspect_terms = extract_aspects(predictions, subtask_key, subtask_value)
+    truth_polarities = extract_aspects(references, subtask_key, "polarity")
+    pred_polarities = extract_aspects(predictions, subtask_key, "polarity")
+
+    # Define adjustment parameters
+    special_token = "NONE"  # For missing aspect terms
+    sentiment_choices = [
+        "positive",
+        "negative",
+        "neutral",
+        "conflict",
+    ]  # For missing polarities
+
+    # Adjust the predictions to match the length of references
+    adjusted_pred_terms = adjust_predictions(
+        truth_aspect_terms, pred_aspect_terms, [special_token]
+    )
+    adjusted_pred_polarities = adjust_predictions(
+        truth_polarities, pred_polarities, sentiment_choices
+    )
+
+    return (
+        flatten_list(truth_aspect_terms),
+        flatten_list(adjusted_pred_terms),
+        flatten_list(truth_polarities),
+        flatten_list(adjusted_pred_polarities),
+    )
+
+
+def flatten_list(nested_list):
+    """Flatten a nested list into a single-level list."""
+    return list(chain.from_iterable(nested_list))
+
+
+def extract_pred_terms(
+    all_predictions: List[Dict[str, Dict[str, str]]]
+) -> List[List]:
+    """Extract and organize predicted terms from the sentiment analysis results."""
+    pred_aspect_terms = []
+    for pred in all_predictions:
+        terms = [term for cat in pred.values() for term in cat.keys()]
+        pred_aspect_terms.append(terms)
+    return pred_aspect_terms
+
+
+def merge_aspects_and_categories(aspects, categories):
+    result = []
+
+    # Assuming both lists are of the same length and corresponding indices match
+    for aspect, category in zip(aspects, categories):
+        combined_entry = {
+            "aspects": {"term": [], "polarity": []},
+            "category": {"category": [], "polarity": []},
+        }
+
+        # Process aspect entries
+        for cat_key, terms_dict in aspect.items():
+            for term, polarity in terms_dict.items():
+                combined_entry["aspects"]["term"].append(term)
+                combined_entry["aspects"]["polarity"].append(polarity)
+
+                # Add category details based on the aspect's key if available in categories
+                if cat_key in category:
+                    combined_entry["category"]["category"].append(cat_key)
+                    combined_entry["category"]["polarity"].append(
+                        category[cat_key]
+                    )
+
+        # Ensure all keys in category are accounted for
+        for cat_key, polarity in category.items():
+            if cat_key not in combined_entry["category"]["category"]:
+                combined_entry["category"]["category"].append(cat_key)
+                combined_entry["category"]["polarity"].append(polarity)
+
+        result.append(combined_entry)
+
+    return result

requirements.txt

@@ -0,0 +1,4 @@
+evaluate
+datasets
+scikit-learn
+gradio