app.py

import csv
import statistics
import string

import gensim.downloader as api
import gradio as gr
import nltk
import pandas as pd
import readability
import seaborn as sns
import torch
from fuzzywuzzy import fuzz
from nltk.corpus import stopwords
from nltk.tokenize import word_tokenize
from sklearn.metrics.pairwise import cosine_similarity
from transformers import DistilBertTokenizer
from transformers import pipeline

nltk.download('cmudict')

nltk.download('stopwords')

nltk.download('punkt')

glove_vectors = api.load('glove-wiki-gigaword-100')

tokenizer = DistilBertTokenizer.from_pretrained('distilbert-base-uncased')
device = torch.device('cuda' if torch.cuda.is_available else 'cpu')

# loading model
PATH = 'pytorchBERTmodel'
model = torch.load(PATH, map_location=torch.device('cpu'))
model.eval()

model.to('cpu')

p = pipeline("automatic-speech-recognition")

w2v = dict({})
for idx, key in enumerate(glove_vectors.key_to_index.keys()):
    w2v[key] = glove_vectors.get_vector(key)


def calculate_diversity(text):

  stop_words = set(stopwords.words('english'))
  for i in string.punctuation:
    stop_words.add(i)

  tokenized_text = word_tokenize(text)
  tokenized_text = list(map(lambda word: word.lower(), tokenized_text))
  sim_words = {}
  if len(tokenized_text) <= 1:
    return 1,"More Text Required"



  
  for idx, anc_word in enumerate(tokenized_text):
    if anc_word in stop_words:
      continue

    vocab = [anc_word]

    for pos, comp_word in enumerate(tokenized_text):
      if anc_word in sim_words.get(pos, []):
        if anc_word == sim_words[pos][0]:
          sim_words[idx] = sim_words[pos]
          continue
      try:
        if not comp_word in stop_words and cosine_similarity(w2v[anc_word].reshape(1, -1), w2v[comp_word].reshape(1, -1)) > .75:     
          vocab.append(comp_word)
        sim_words[idx] = vocab

      except KeyError:
        continue
  

  scores = {}
  for key, value in sim_words.items():
    if len(value) == 1:
      scores[key] = 1
      continue
    if len(value) == 2:
      scores[key] = -1
      continue
    t_sim = len(value) - 1 
    t_rep = (len(value) - 1) - (len(set(value[1:]))) 
    
    score = ((t_sim - t_rep)/t_sim)**2

    scores[key] = score

  mean_score = 0
  total = 0

  for value in scores.values():
    if value == -1:
      continue
    mean_score += value
    total += 1
  
  return scores, mean_score/total


def dict_to_list(dictionary, max_size=10):
    outer_list = []
    inner_list = []

    for value in dictionary.values():
        inner_list.append(value)
        if len(inner_list) == max_size:
            outer_list.append(inner_list)
            inner_list = []
    if len(inner_list) > 0:
        outer_list.append(inner_list)
    return outer_list


def heatmap(scores, df):
    total = 0
    loops = 0

    for ratio in scores.values():
        # conditional to visualize the difference between no ratio and a 0 ratio score
        if ratio != -.3:
            total += ratio
            loops += 1

    diversity_average = total / loops

    return sns.heatmap(df, cmap='gist_gray_r', vmin=-.3).set(
        title='Word Diversity Score Heatmap (Average Score: ' + str(diversity_average) + ')')


def stats(text):
    results = readability.getmeasures(text, lang='en')
    return results


def predict(text, tokenizer=tokenizer):
    model.eval()
    model.to('cpu')

    def prepare_data(text, tokenizer):
        input_ids = []
        attention_masks = []

        encoded_text = tokenizer.encode_plus(
            text,
            truncation=True,
            add_special_tokens=True,
            max_length=315,
            pad_to_max_length=True,
            return_attention_mask=True,
            return_tensors='pt'
        )

        input_ids.append(encoded_text['input_ids'])
        attention_masks.append(encoded_text['attention_mask'])

        input_ids = torch.cat(input_ids, dim=0)
        attention_masks = torch.cat(attention_masks, dim=0)
        return {'input_ids': input_ids, 'attention_masks': attention_masks}

    tokenized_example_text = prepare_data(text, tokenizer)
    with torch.no_grad():
        result = model(
            tokenized_example_text['input_ids'].to('cpu'),
            attention_mask=tokenized_example_text['attention_masks'].to('cpu'),
            return_dict=True
        ).logits

    return result


def reading_difficulty(excerpt):
    if len(excerpt) == 0:
        return "No Text Provided"
    windows = []
    words = tokenizer.tokenize(excerpt)

    if len(words) > 301:
        for idx, text in enumerate(words):
            if idx % 300 == 0:
                if idx <= len(words) - 301:
                    x = ' '.join(words[idx: idx + 299])
                    windows.append(x)

        win_preds = []
        for text in windows:
            win_preds.append(predict(text, tokenizer).item())
        result = statistics.mean(win_preds)
        score = -(result * 1.786 + 6.4) + 10
        return score

    else:
        result = predict(excerpt).item()
        score = -(result * 1.786 + 6.4) + 10
        return score


def calculate_stats(file_name, data_index):
    # unicode escape only for essays
    with open(file_name, encoding='unicode_escape') as f:
        information = {'lines': 0, 'words_per_sentence': 0, 'words': 0, 'syll_per_word': 0, 'characters_per_word': 0,
                       'reading_difficulty': 0}
        reader = csv.reader(f)

        for line in reader:

            if len(line[data_index]) < 100:
                continue

            # if detect(line[data_index][len(line[data_index]) -400: len(line[data_index])-1]) == 'en':

            try:
                stat = stats(line[data_index])

            except ValueError:
                continue

            information['lines'] += 1
            information['words_per_sentence'] += stat['sentence info']['words_per_sentence']
            information['words'] += stat['sentence info']['words']
            information['syll_per_word'] += stat['sentence info']['syll_per_word']
            information['characters_per_word'] += stat['sentence info']['characters_per_word']
            information['reading_difficulty'] += reading_difficulty(line[data_index])

    for i in information:
        if i != 'lines' and i != 'words':
            information[i] /= information['lines']

    return information


def transcribe(audio):
    # speech to text using pipeline
    text = p(audio)["text"]
    transcription.append(text)
    return text


def compute_score(target, actual):
    target = target.lower()
    actual = actual.lower()
    return fuzz.ratio(target, actual)


def phon(text):
    alph = nltk.corpus.cmudict.dict()
    text = word_tokenize(text)
    pronun = []
    for word in text:
        try:
            pronun.append(alph[word][0])
        except Exception as e:
            pronun.append(word)
    return pronun


def gradio_fn(text, audio, target, actual_audio):
    if text is None and audio is None and target is None and actual_audio is None:
        return "No Inputs", "No Inputs", "No Inputs", "No Inputs"
    speech_score = 0

    if actual_audio is not None:
        actual = p(actual_audio)["text"]
        speech_score = compute_score(target, actual)

        return "Difficulty Score: " + str(reading_difficulty(actual)), "Transcript: " + str(
            actual.lower()), "Diversity Score: " + str(calculate_diversity(target)[1]), "Speech Score: " + str(speech_score)
    div = calculate_diversity(text)
    transcription = []
    if audio is not None:
        text = p(audio)["text"]
        transcription.append(text)
        state = div[0]
        return "Difficulty Score: " + str(reading_difficulty(text)), "Transcript: " + str(
            transcription[-1].lower()), "Diversity Score: " + str(div[1]), "No Inputs"

    return "Difficulty Score: " + str(reading_difficulty(text)), "Diversity Score: " + str(
        div[1]), "No Audio Provided", "No Audio Provided"


def plot():
    text = state
    diversity = calculate_diversity(text)[0]
    df = pd.DataFrame(dict_to_list(diversity))
    return heatmap(diversity, df)



example_data = []
x = 0
with open('train.csv') as f:
    reader = csv.reader(f)
    next(reader)
    for line in reader:
        example_data.append([line[3]])
        x += 1
        if x > 100:
            break

state = {}
interface = gr.Interface(
    fn=gradio_fn,
    inputs=[gr.components.Textbox(
        label="Text Difficulty Scoring",
        lines = 6),
        gr.components.Audio(
            label="Speech Translation",
            source="microphone",
            type="filepath"),
        gr.components.Textbox(
            label="Type Your Target Text to Recite",
            placeholder="How much wood would a woodchuck chuck if a woodchuck could chuck wood?"
        ),
        gr.components.Audio(
            label="Read Text Typed Above for Pronunciation Score",
            source="microphone",
            type="filepath")
    ],

    outputs=["text", "text", "text", "text"],
    theme="huggingface",
    description="Enter text or speak into your microphone to have your text analyzed!",
    rounded=True,
    container=True,
    article="""
    Text Difficulty Score- Using a fine-tuned Distil-Bert model, we automatically determine how difficult something is to read while incorporating underlying semantics. 
    To efficiently compute text difficulty, a Distil-Bert pre-trained model is fine-tuned for regression using The CommonLit Ease of Readability (CLEAR) 
    Corpus. https://educationaldatamining.org/EDM2021/virtual/static/pdf/EDM21_paper_35.pdf This dataset contains over 110,000 pairwise comparisons of 
    ~1100 teachers responded to the question, "Which text is easier for students to understand?". This model is trained end-end (regression layer down to 
    the first attention layer) to ensure the best performance- Merchant et al. 2020
    
    Speech Pronunciaion Scoring: The Wave2Vec 2.0 model is utilized to convert audio into text in real-time. The model predicts words or phonemes (smallest 
    unit of speech distinguishing one word (or word element) from another) from the user input audio. Due to the nature of the model, users with poor 
    pronunciation receive inaccurate translations. This project attempts to score pronunciation by asking a user to read a target excerpt into the microphone. We then
    pass this audio through Wave2Vec 2.0 to get the inferred intended words. We measure the loss as the Levenshtein distance between the target and actual transcripts- 
    the Levenshtein distance between two words is the minimum number of single-character edits required to change one word into the other.

    Lexical Diversity Score: The lexical diversity score is computed by taking the ratio of unique similar words to total similar words squared. The similarity is computed 
    as if the cosine similarity of the word2vec embeddings is greater than .75. It is bad writing/speech practice to repeat the same words when it's possible not to. 
    Vocabulary diversity is generally computed by taking the ratio of unique strings/ total strings. This does not give an indication if the person has a large vocabulary 
    or if the topic does not require a diverse vocabulary to express it. Words that are not in the Word2Vec vocabulary will not be incorporated into the score.
    """

).launch()