app.py

import numpy as np
import csv
from typing import Optional
from urllib.request import urlopen
import gradio as gr


class SentimentTransform():
    def __init__(
            self,
            model_name: str = "cardiffnlp/twitter-roberta-base-sentiment",
            highlight: bool = False,
            positive_sentiment_name: str = "positive",
            max_number_of_shap_documents: Optional[int] = None,
            min_abs_score: float = 0.1,
            sensitivity: float = 0,
            **kwargs,
    ):
        """
        Sentiment Ops.
        Parameters
        -------------
        model_name: str
            The name of the model
        sensitivity: float
            How confident it is about being `neutral`. If you are dealing with news sources,
            you probably want less sensitivity
        """
        self.model_name = model_name
        self.highlight = highlight
        self.positive_sentiment_name = positive_sentiment_name
        self.max_number_of_shap_documents = max_number_of_shap_documents
        self.min_abs_score = min_abs_score
        self.sensitivity = sensitivity
        for k, v in kwargs.items():
            setattr(self, k, v)

    def preprocess(self, text: str):
        new_text = []
        for t in text.split(" "):
            t = "@user" if t.startswith("@") and len(t) > 1 else t
            t = "http" if t.startswith("http") else t
            new_text.append(t)
        return " ".join(new_text)

    @property
    def classifier(self):
        if not hasattr(self, "_classifier"):
            import transformers

            self._classifier = transformers.pipeline(
                return_all_scores=True,
                model=self.model_name,
            )
        return self._classifier

    def _get_label_mapping(self, task: str):
        # Note: this is specific to the current model
        labels = []
        mapping_link = f"https://raw.githubusercontent.com/cardiffnlp/tweeteval/main/datasets/{task}/mapping.txt"
        with urlopen(mapping_link) as f:
            html = f.read().decode("utf-8").split("\n")
            csvreader = csv.reader(html, delimiter="\t")
        labels = [row[1] for row in csvreader if len(row) > 1]
        return labels

    @property
    def label_mapping(self):
        return {"LABEL_0": "negative", "LABEL_1": "neutral", "LABEL_2": "positive"}

    def analyze_sentiment(
            self,
            text,
            highlight: bool = False,
            positive_sentiment_name: str = "positive",
            max_number_of_shap_documents: Optional[int] = None,
            min_abs_score: float = 0.1,
    ):
        if text is None:
            return None
        labels = self.classifier([str(text)], truncation=True, max_length=512)
        ind_max = np.argmax([l["score"] for l in labels[0]])
        sentiment = labels[0][ind_max]["label"]
        max_score = labels[0][ind_max]["score"]
        sentiment = self.label_mapping.get(sentiment, sentiment)
        if sentiment.lower() == "neutral" and max_score > self.sensitivity:
            overall_sentiment = 1e-5
        elif sentiment.lower() == "neutral":
            # get the next highest score
            new_labels = labels[0][:ind_max] + labels[0][(ind_max + 1):]
            new_ind_max = np.argmax([l["score"] for l in new_labels])
            new_max_score = new_labels[new_ind_max]["score"]
            new_sentiment = new_labels[new_ind_max]["label"]
            new_sentiment = self.label_mapping.get(new_sentiment, new_sentiment)
            overall_sentiment = self._calculate_overall_sentiment(
                new_max_score, new_sentiment
            )

        else:
            overall_sentiment = self._calculate_overall_sentiment(max_score, sentiment)
        # Adjust to avoid bug
        if overall_sentiment == 0:
            overall_sentiment = 1e-5
        if not highlight:
            return {
                "sentiment": sentiment,
                "overall_sentiment_score": overall_sentiment,
            }
        shap_documents = self.get_shap_values(
            text,
            sentiment_ind=ind_max,
            max_number_of_shap_documents=max_number_of_shap_documents,
            min_abs_score=min_abs_score,
        )
        return {
            "sentiment": sentiment,
            "score": max_score,
            "overall_sentiment": overall_sentiment,
            "highlight_chunk_": shap_documents,
        }

    def _calculate_overall_sentiment(self, score: float, sentiment: str):
        if sentiment.lower().strip() == self.positive_sentiment_name:
            return score
        else:
            return -score

    # def explainer(self):
    #     if hasattr(self, "_explainer"):
    #         return self._explainer
    #     else:
    #         try:
    #             import shap
    #         except ModuleNotFoundError:
    #             raise MissingPackageError("shap")
    #         self._explainer = shap.Explainer(self.classifier)
    #         return self._explainer

    def get_shap_values(
            self,
            text: str,
            sentiment_ind: int = 2,
            max_number_of_shap_documents: Optional[int] = None,
            min_abs_score: float = 0.1,
    ):
        """Get SHAP values"""
        shap_values = self.explainer([text])
        cohorts = {"": shap_values}
        cohort_labels = list(cohorts.keys())
        cohort_exps = list(cohorts.values())
        features = cohort_exps[0].data
        feature_names = cohort_exps[0].feature_names
        values = np.array([cohort_exps[i].values for i in range(len(cohort_exps))])
        shap_docs = [
            {"text": v, "score": f}
            for f, v in zip(
                [x[sentiment_ind] for x in values[0][0].tolist()], feature_names[0]
            )
        ]
        if max_number_of_shap_documents is not None:
            sorted_scores = sorted(shap_docs, key=lambda x: x["score"], reverse=True)
        else:
            sorted_scores = sorted(shap_docs, key=lambda x: x["score"], reverse=True)[
                            :max_number_of_shap_documents
                            ]
        return [d for d in sorted_scores if abs(d["score"]) > min_abs_score]

    def transform(self, text):
        # # For each document, update the field
        # sentiment_docs = [{"_id": d["_id"]} for d in documents]
        # for i, t in enumerate(self.text_fields):
        #     if self.output_fields is not None:
        #         output_field = self.output_fields[i]
        #     else:
        #         output_field = self._get_output_field(t)
        sentiment = self.analyze_sentiment(
            text,
            highlight=self.highlight,
            max_number_of_shap_documents=self.max_number_of_shap_documents,
            min_abs_score=self.min_abs_score, )
        return sentiment


def sentiment_classifier(text, model_type, sensitivity):
    if model_type == 'Social Media Model':
        model_name = "cardiffnlp/twitter-roberta-base-sentiment"
    elif model_type == 'Survey Model':
        model_name = "j-hartmann/sentiment-roberta-large-english-3-classes"
    else:
        model_name = "j-hartmann/sentiment-roberta-large-english-3-classes"
    model = SentimentTransform(model_name=model_name, sensitivity=sensitivity)
    res_dict = model.transform(text)
    return res_dict['sentiment'], res_dict['overall_sentiment_score']


demo = gr.Interface(
    fn=sentiment_classifier,
    inputs=[gr.Textbox(placeholder="Put the text here and click 'submit' to predict its sentiment", label="Input Text"), gr.Dropdown(["Social Media Model", "Survey Model"], value="Survey Model", label="Select the Model that you want to use."), gr.Slider(0, 1, step = 0.01, label="Sensitivity (How confident it is about being `neutral`. If you are dealing with news sources, you probably want less sensitivity.)")],
    outputs=[gr.Textbox(label='Sentiment'), gr.Textbox(label='Sentiment Score')],
)
demo.launch(debug=True)