Upload 21 files

.gitattributes

@@ -33,3 +33,10 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+calm.wav filter=lfs diff=lfs merge=lfs -text
+data/chew1.wav filter=lfs diff=lfs merge=lfs -text
+data/tapping1.wav filter=lfs diff=lfs merge=lfs -text
+data/theeStallion1.wav filter=lfs diff=lfs merge=lfs -text
+data/trump1.wav filter=lfs diff=lfs merge=lfs -text
+data/trump2.wav filter=lfs diff=lfs merge=lfs -text
+hrv-breathing.gif filter=lfs diff=lfs merge=lfs -text

app.py

@@ -0,0 +1,173 @@
+import os
+import whisper
+import evaluate
+from evaluate.utils import launch_gradio_widget
+import gradio as gr
+import torch
+import pandas as pd
+import random
+import classify
+import replace_explitives
+from whisper.model import Whisper
+from whisper.tokenizer import get_tokenizer
+from speechbrain.pretrained.interfaces import foreign_class
+from transformers import AutoModelForSequenceClassification, pipeline, WhisperTokenizer, RobertaForSequenceClassification, RobertaTokenizer, AutoTokenizer
+
+
+# pull in emotion detection
+# --- Add element for specification
+# pull in text classification
+# --- Add custom labels
+# --- Associate labels with radio elements
+# add logic to initiate mock notificaiton when detected
+# pull in misophonia-specific model
+
+model_cache = {}
+
+# Building prediction function for gradio
+emo_dict = {
+    'sad': 'Sad', 
+    'hap': 'Happy',
+    'ang': 'Anger',
+    'neu': 'Neutral'
+}
+
+# static classes for now, but it would be best ot have the user select from multiple, and to enter their own
+class_options = {
+    "Racism": ["racism", "hate speech", "bigotry", "racially targeted", "racial slur", "ethnic slur", "ethnic hate", "pro-white nationalism"],
+    "LGBTQ+ Hate": ["gay slur", "trans slur", "homophobic slur", "transphobia", "anti-LBGTQ+"],
+    "Sexually Explicit": ["sexually explicit", "sexually coercive", "sexual exploitation", "vulgar", "raunchy", "sexist", "sexually demeaning", "sexual violence", "victim blaming"],
+    "Pregnancy Complications": ["miscarriage", "child loss", "child death", "abortion", "pregnancy", "childbirth", "baby shower", "postpartum"],
+}
+
+pipe = pipeline("automatic-speech-recognition", model="openai/whisper-large")
+
+toxicity_module = evaluate.load("toxicity",  "facebook/roberta-hate-speech-dynabench-r4-target")
+emotion_classifier = foreign_class(source="speechbrain/emotion-recognition-wav2vec2-IEMOCAP", pymodule_file="custom_interface.py", classname="CustomEncoderWav2vec2Classifier")
+device = torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu")
+text_classifier = pipeline("zero-shot-classification", model="facebook/bart-large-mnli")
+
+def classify_emotion(audio):
+    #### Emotion classification ####
+    # EMO MODEL LINE emotion_classifier = foreign_class(source="speechbrain/emotion-recognition-wav2vec2-IEMOCAP", pymodule_file="custom_interface.py", classname="CustomEncoderWav2vec2Classifier")
+    out_prob, score, index, text_lab = emotion_classifier.classify_file(audio)
+    return  emo_dict[text_lab[0]]
+
+def slider_logic(slider):
+    threshold = 0
+    if slider == 1:
+        threshold = .90
+    elif slider == 2:
+        threshold = .80
+    elif slider == 3:
+        threshold = .60
+    elif slider == 4:
+        threshold = .50
+    elif slider == 5:
+        threshold = .40
+    else:
+        threshold = []
+    return threshold
+
+# Create a Gradio interface with audio file and text inputs
+def classify_toxicity(audio_file, classify_anxiety, emo_class, explitive_selection, slider):
+    
+    # Transcribe the audio file using Whisper ASR
+    transcribed_text = pipe(audio_file)["text"]
+    
+    ## SLIDER ##
+    threshold = slider_logic(slider)
+    
+    #------- explitive call ---------------
+    
+    if replace_explitives != None and emo_class == None:
+        transcribed_text = replace_explitives.sub_explitives(transcribed_text, explitive_selection)
+    
+    #### Toxicity Classifier ####
+        
+    # TOX MODEL LINE toxicity_module = evaluate.load("toxicity",  "facebook/roberta-hate-speech-dynabench-r4-target")
+    #toxicity_module = evaluate.load("toxicity", 'DaNLP/da-electra-hatespeech-detection', module_type="measurement")
+
+    toxicity_results = toxicity_module.compute(predictions=[transcribed_text])
+ 
+    toxicity_score = toxicity_results["toxicity"][0]
+    print(toxicity_score)
+    
+    # emo call
+    if emo_class != None:
+        classify_emotion(audio_file)
+
+    #### Text classification #####
+    if classify_anxiety != None: 
+        # DEVICE LINE device = torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu")
+    
+        # CLASSIFICATION LINE text_classifier = pipeline("zero-shot-classification", model="facebook/bart-large-mnli")
+    
+        sequence_to_classify = transcribed_text
+        print(classify_anxiety, class_options)
+        candidate_labels = class_options.get(classify_anxiety, [])
+        # classification_output = classifier(sequence_to_classify, candidate_labels, multi_label=False)
+        classification_output = text_classifier(sequence_to_classify, candidate_labels, multi_label=True)
+        print("class output ", type(classification_output))
+        # classification_df = pd.DataFrame.from_dict(classification_output)
+        print("keys ", classification_output.keys())
+        
+        # formatted_classification_output = "\n".join([f"{key}: {value}" for key, value in classification_output.items()])
+        # label_score_pairs = [(label, score) for label, score in zip(classification_output['labels'], classification_output['scores'])]
+        label_score_dict = {label: score for label, score in zip(classification_output['labels'], classification_output['scores'])}
+        k = max(label_score_dict, key=label_score_dict.get)
+        print("k keys: ", k)
+        maxval = label_score_dict[k]
+        print("max value: ", maxval)
+        topScore = ""
+        affirm = ""
+        if maxval > threshold:
+            print("Toxic")
+            affirm = positive_affirmations()
+            topScore = maxval
+        else:
+            print("Not Toxic")
+            affirm = ""
+            topScore = maxval
+    else:
+        topScore = ""
+        affirm = ""
+        if toxicity_score > threshold:
+            affirm = positive_affirmations()
+            topScore = toxicity_score
+        else:
+            affirm = ""
+            topScore = toxicity_score
+        label_score_dict = {"toxicity" : toxicity_score}
+
+    return transcribed_text, topScore, label_score_dict, affirm
+    # return f"Toxicity Score ({available_models[selected_model]}): {toxicity_score:.4f}"
+    
+def positive_affirmations():
+    affirmations = [
+        "I have survived my anxiety before and I will survive again now",
+        "I am not in danger; I am just uncomfortable; this too will pass",
+        "I forgive and release the past and look forward to the future",
+        "I can't control what other people say but I can control my breathing and my response"
+    ]
+    selected_affirm = random.choice(affirmations)
+    return selected_affirm
+    
+with gr.Blocks() as iface:
+    show_state = gr.State([])
+    with gr.Column():
+        anxiety_class = gr.Radio(label="Specify Subclass", choices=["Racism", "LGBTQ+ Hate", "Sexually Explicit", "Pregnancy Complications"])
+        explit_preference = gr.Radio(choices=["N-Word", "B-Word", "All Explitives"], label="Words to omit from general anxiety classes", info="certain words may be acceptible within certain contects for given groups of people, and some people may be unbothered by explitives broadly speaking.")
+        emo_class = gr.Radio(choices=["negaitve emotionality"], label="Negative Emotionality", info="Select if you would like explitives to be considered anxiety-indiucing in the case of anger/ negative emotionality.")
+        sense_slider = gr.Slider(minimum=1, maximum=5, step=1.0, label="How readily do you want the tool to intervene? 1 = in extreme cases and 5 = at every opportunity")
+    with gr.Column():
+        aud_input = gr.Audio(source="upload", type="filepath", label="Upload Audio File")
+        submit_btn = gr.Button(label="Run")
+    with gr.Column():
+        out_text = gr.Textbox(label="Transcribed Audio")
+        out_val = gr.Textbox(label="Overall Toxicity")
+        out_affirm = gr.Textbox(label="Intervention")
+        out_class = gr.Label(label="Toxicity Class Breakdown")
+    submit_btn.click(fn=classify_toxicity, inputs=[aud_input, anxiety_class, emo_class, explit_preference, sense_slider], outputs=[out_text, out_val, out_class, out_affirm])
+
+iface.launch()

b_word.txt

@@ -0,0 +1,2 @@
+bitch
+bitches

calm.wav

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:1bafc436a822d1e3670b457660087b3caf518e9d0d83d8c999bc642a5166f4b1
+size 15916220

classify.py

@@ -0,0 +1,66 @@
+from typing import List, Optional
+
+import torch
+import torch.nn.functional as F
+from whisper.audio import N_FRAMES, N_MELS, log_mel_spectrogram, pad_or_trim
+from whisper.model import Whisper
+from whisper.tokenizer import Tokenizer
+
+
+@torch.no_grad()
+def calculate_audio_features(audio_path: Optional[str], model: Whisper) -> torch.Tensor:
+    if audio_path is None:
+        segment = torch.zeros((N_MELS, N_FRAMES), dtype=torch.float32).to(model.device)
+    else:
+        mel = log_mel_spectrogram(audio_path)
+        segment = pad_or_trim(mel, N_FRAMES).to(model.device)
+    return model.embed_audio(segment.unsqueeze(0))
+
+
+@torch.no_grad()
+def calculate_average_logprobs(
+    model: Whisper,
+    audio_features: torch.Tensor,
+    class_names: List[str],
+    tokenizer: Tokenizer,
+) -> torch.Tensor:
+    initial_tokens = (
+        torch.tensor(tokenizer.sot_sequence_including_notimestamps).unsqueeze(0).to(model.device)
+    )
+    eot_token = torch.tensor([tokenizer.eot]).unsqueeze(0).to(model.device)
+
+    average_logprobs = torch.zeros(len(class_names))
+    for i, class_name in enumerate(class_names):
+        class_name_tokens = (
+            torch.tensor(tokenizer.encode(" " + class_name)).unsqueeze(0).to(model.device)
+        )
+        input_tokens = torch.cat([initial_tokens, class_name_tokens, eot_token], dim=1)
+
+        logits = model.logits(input_tokens, audio_features)  # (1, T, V)
+        logprobs = F.log_softmax(logits, dim=-1).squeeze(0)  # (T, V)
+        logprobs = logprobs[len(tokenizer.sot_sequence_including_notimestamps) - 1 : -1]  # (T', V)
+        logprobs = torch.gather(logprobs, dim=-1, index=class_name_tokens.view(-1, 1))  # (T', 1)
+        average_logprob = logprobs.mean().item()
+        average_logprobs[i] = average_logprob
+
+    return average_logprobs
+
+
+def calculate_internal_lm_average_logprobs(
+    model: Whisper,
+    class_names: List[str],
+    tokenizer: Tokenizer,
+    verbose: bool = False,
+) -> torch.Tensor:
+    audio_features_from_empty_input = calculate_audio_features(None, model)
+    average_logprobs = calculate_average_logprobs(
+        model=model,
+        audio_features=audio_features_from_empty_input,
+        class_names=class_names,
+        tokenizer=tokenizer,
+    )
+    if verbose:
+        print("Internal LM average log probabilities for each class:")
+        for i, class_name in enumerate(class_names):
+            print(f"  {class_name}: {average_logprobs[i]:.3f}")
+    return average_logprobs

data/chew1.wav

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:91ded13633e05d45c791366de676d09c9eb2f4b51979e211112e734d97cfdf08
+size 5120104

data/tapping1.wav

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:9506ae17e7176ef99a36a3f556f9f45837f804e77a6b0013a38470e73f8ed5e4
+size 2958378

data/theeStallion1.wav

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:d347f0917b4989ebe491bb96955d31afcf3b31e6480d43136a7ff3bffd8dd9da
+size 2736184

data/trump1.wav

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:3952dd160ff5031da8fab133d6d685bbcdc190bb34fa0ee7c75b7c7e5ff9a8ea
+size 10700952

data/trump2.wav

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:c2ee9ec1a06544428bf8c0f66eda962533e7e70467a0b30492dfbfd1d30c0981
+size 7329432

expletives.txt

@@ -0,0 +1,15 @@
+shit
+fuck
+fucked
+damn
+damned
+goddamn 
+goddmaned
+crap
+crapped
+ass
+asshole
+bastard
+bastards
+piss
+pissed

n_word.txt

@@ -0,0 +1,9 @@
+nigga
+niggas
+nigg
+nig
+niggs
+nigs
+nigger
+niggers
+niggaz

packages.txt

@@ -0,0 +1 @@
+ffmpeg

replace_explitives.py

@@ -0,0 +1,44 @@
+import regex as re
+import nltk
+
+def load_words_from_file(file_path):
+    with open(file_path, "r", encoding="utf-8") as f:
+        words = [line.strip() for line in f.readlines()]
+    return words
+
+def sub_explitives(textfile, selection):
+
+    replacetext = "person"
+    
+    # Load target words from text files
+    b_word_list = load_words_from_file("b_word.txt")
+    n_word_list = load_words_from_file("n_word.txt")
+    expletives_list = load_words_from_file("expletives.txt")
+    
+    # text = word_tokenize(textfile)
+    # print(text)
+    # sentences = sent_tokenize(textfile)
+
+    if selection == "B-Word":
+        target_word = b_word_list
+    elif selection == "N-Word":
+        target_word = n_word_list
+    elif selection == "All Explitives":
+        target_word = expletives_list
+    else:
+        target_word = []
+        
+    print("selection:", selection, "target_word:", target_word)
+    lines = textfile.split('\n')
+
+    if target_word:
+        print("target word was found, ", target_word)
+        print(textfile)
+        for i, line in enumerate(lines):
+            for word in target_word:
+                pattern = r"\b" + re.escape(word) + r"\b"  
+                # textfile = re.sub(target_word, replacetext, textfile, flags=re.IGNORECASE)
+                lines[i] = re.sub(pattern, replacetext, lines[i], flags=re.IGNORECASE)
+
+    textfile = '\n'.join(lines)
+    return textfile

requirements.txt

@@ -0,0 +1,11 @@
+git+https://github.com/huggingface/evaluate@775555d80af30d83dc6e9f42051840d29a34f31b
+git+https://github.com/openai/whisper.git
+transformers
+torch
+speechbrain
+torchaudio
+git+https://github.com/openai/whisper.git
+tqdm
+gradio==3.14.0
+regex
+nltk

toxicity.py

@@ -0,0 +1,141 @@
+# Copyright 2020 The HuggingFace Evaluate Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+""" Toxicity detection measurement. """
+
+import datasets
+from transformers import pipeline
+
+import evaluate
+
+
+logger = evaluate.logging.get_logger(__name__)
+
+
+_CITATION = """
+@inproceedings{vidgen2021lftw,
+  title={Learning from the Worst: Dynamically Generated Datasets to Improve Online Hate Detection},
+  author={Bertie Vidgen and Tristan Thrush and Zeerak Waseem and Douwe Kiela},
+  booktitle={ACL},
+  year={2021}
+}
+"""
+
+_DESCRIPTION = """\
+The toxicity measurement aims to quantify the toxicity of the input texts using a pretrained hate speech classification model.
+"""
+
+_KWARGS_DESCRIPTION = """
+Compute the toxicity of the input sentences.
+
+Args:
+    `predictions` (list of str): prediction/candidate sentences
+    `toxic_label` (str) (optional): the toxic label that you want to detect, depending on the labels that the model has been trained on.
+        This can be found using the `id2label` function, e.g.:
+        model = AutoModelForSequenceClassification.from_pretrained("DaNLP/da-electra-hatespeech-detection")
+        print(model.config.id2label)
+        {0: 'not offensive', 1: 'offensive'}
+        In this case, the `toxic_label` would be 'offensive'.
+    `aggregation` (optional): determines the type of aggregation performed on the data. If set to `None`, the scores for each prediction are returned.
+    Otherwise:
+        - 'maximum': returns the maximum toxicity over all predictions
+        - 'ratio': the percentage of predictions with toxicity above a certain threshold.
+    `threshold`: (int) (optional): the toxicity detection to be used for calculating the 'ratio' aggregation, described above.
+    The default threshold is 0.5, based on the one established by [RealToxicityPrompts](https://arxiv.org/abs/2009.11462).
+
+Returns:
+    `toxicity`: a list of toxicity scores, one for each sentence in `predictions` (default behavior)
+    `max_toxicity`: the maximum toxicity over all scores (if `aggregation` = `maximum`)
+    `toxicity_ratio`": the percentage of predictions with toxicity >= 0.5 (if `aggregation` = `ratio`)
+
+Examples:
+
+    Example 1 (default behavior):
+        >>> toxicity = evaluate.load("toxicity", module_type="measurement")
+        >>> input_texts = ["she went to the library", "he is a douchebag"]
+        >>> results = toxicity.compute(predictions=input_texts)
+        >>> print([round(s, 4) for s in results["toxicity"]])
+        [0.0002, 0.8564]
+
+    Example 2 (returns ratio of toxic sentences):
+        >>> toxicity = evaluate.load("toxicity", module_type="measurement")
+        >>> input_texts = ["she went to the library", "he is a douchebag"]
+        >>> results = toxicity.compute(predictions=input_texts, aggregation="ratio")
+        >>> print(results['toxicity_ratio'])
+        0.5
+
+    Example 3 (returns the maximum toxicity score):
+
+        >>> toxicity = evaluate.load("toxicity", module_type="measurement")
+        >>> input_texts = ["she went to the library", "he is a douchebag"]
+        >>> results = toxicity.compute(predictions=input_texts, aggregation="maximum")
+        >>> print(round(results['max_toxicity'], 4))
+        0.8564
+
+    Example 4 (uses a custom model):
+
+        >>> toxicity = evaluate.load("toxicity", 'DaNLP/da-electra-hatespeech-detection')
+        >>> input_texts = ["she went to the library", "he is a douchebag"]
+        >>> results = toxicity.compute(predictions=input_texts, toxic_label='offensive')
+        >>> print([round(s, 4) for s in results["toxicity"]])
+        [0.0176, 0.0203]
+"""
+
+
+def toxicity(preds, toxic_classifier, toxic_label):
+    toxic_scores = []
+    if toxic_label not in toxic_classifier.model.config.id2label.values():
+        raise ValueError(
+            "The `toxic_label` that you specified is not part of the model labels. Run `model.config.id2label` to see what labels your model outputs."
+        )
+
+    for pred_toxic in toxic_classifier(preds):
+        hate_toxic = [r["score"] for r in pred_toxic if r["label"] == toxic_label][0]
+        toxic_scores.append(hate_toxic)
+    return toxic_scores
+
+
+@evaluate.utils.file_utils.add_start_docstrings(_DESCRIPTION, _KWARGS_DESCRIPTION)
+class Toxicity(evaluate.Measurement):
+    def _info(self):
+        return evaluate.MeasurementInfo(
+            module_type="measurement",
+            description=_DESCRIPTION,
+            citation=_CITATION,
+            inputs_description=_KWARGS_DESCRIPTION,
+            features=datasets.Features(
+                {
+                    "predictions": datasets.Value("string", id="sequence"),
+                }
+            ),
+            codebase_urls=[],
+            reference_urls=[],
+        )
+
+    def _download_and_prepare(self, dl_manager):
+        if self.config_name == "default":
+            logger.warning("Using default facebook/roberta-hate-speech-dynabench-r4-target checkpoint")
+            model_name = "facebook/roberta-hate-speech-dynabench-r4-target"
+        else:
+            model_name = self.config_name
+        self.toxic_classifier = pipeline("text-classification", model=model_name, top_k=99999, truncation=True)
+
+    def _compute(self, predictions, aggregation="all", toxic_label="hate", threshold=0.5):
+        scores = toxicity(predictions, self.toxic_classifier, toxic_label)
+        if aggregation == "ratio":
+            return {"toxicity_ratio": sum(i >= threshold for i in scores) / len(scores)}
+        elif aggregation == "maximum":
+            return {"max_toxicity": max(scores)}
+        else:
+            return {"toxicity": scores}