app.py

import nltk
import sklearn_crfsuite
from sklearn_crfsuite import metrics
from nltk.stem import LancasterStemmer
import numpy as np
from sklearn.metrics import confusion_matrix
import seaborn as sns
import matplotlib.pyplot as plt
import re
import gradio as gr
lancaster = LancasterStemmer()

nltk.download('brown')
nltk.download('universal_tagset')

class CRF_POS_Tagger:
    def __init__(self, train=False):
        print("Loading Data...")
        self.corpus = nltk.corpus.brown.tagged_sents(tagset='universal')
        print("Data Loaded...")
        self.corpus = [[(word, tag) for word, tag in sentence] for sentence in self.corpus]
        self.actual_tag = []
        self.predicted_tag = []
        self.prefixes = [
            "a", "anti", "auto", "bi", "co", "dis", "en", "em", "ex", "in", "im",
            "inter", "mis", "non", "over", "pre", "re", "sub", "trans", "un", "under"
        ]

        self.suffixes = [
            "able", "ible", "al", "ance", "ence", "dom", "er", "or", "ful", "hood",
            "ic", "ing", "ion", "tion", "ity", "ty", "ive", "less", "ly", "ment",
            "ness", "ous", "ship", "y", "es", "s"
        ]

        self.prefix_pattern = f"^({'|'.join(self.prefixes)})"
        self.suffix_pattern = f"({'|'.join(self.suffixes)})$"

        self.X = [[self.word_features(sentence, i) for i in range(len(sentence))] for sentence in self.corpus]
        self.y = [[postag for _, postag in sentence] for sentence in self.corpus]

        self.split = int(0.8 * len(self.X))
        self.X_train = self.X[:self.split]
        self.y_train = self.y[:self.split]
        self.X_test = self.X[self.split:]
        self.y_test = self.y[self.split:]
        print("Data Loaded...")
        self.crf_model = sklearn_crfsuite.CRF(algorithm='lbfgs', c1=0.1, c2=0.1, max_iterations=100, all_possible_transitions=True)
        print("Model Created...")
        if train:
            self.train()

    def word_splitter(self, word):
        prefix = ""
        stem = word
        suffix = ""

        prefix_match = re.match(self.prefix_pattern, word)
        if prefix_match:
            prefix = prefix_match.group(1)
            stem = word[len(prefix):]

        suffix_match = re.search(self.suffix_pattern, stem)
        if suffix_match:
            suffix = suffix_match.group(1)
            stem = stem[: -len(suffix)]

        return prefix, stem, suffix

    # Define a function to extract features for each word in a sentence
    def word_features(self, sentence, i):
        word = sentence[i][0]
        prefix, stem, suffix = self.word_splitter(word)
        # features = {
        #     'word': word,
        #     'prefix': prefix,
        #     # 'stem': stem,
        #     'stem': lancaster.stem(word),
        #     'suffix': suffix,
        #     'position': i,
        #     'is_first': i == 0, #if the word is a first word
        #     'is_last': i == len(sentence) - 1,  #if the word is a last word
        #     # 'is_capitalized': word[0].upper() == word[0],
        #     'is_all_caps': word.isupper(),      #word is in uppercase
        #     'is_all_lower': word.islower(),      #word is in lowercase

        #     'prefix-1': word[0],
        #     'prefix-2': word[:2],
        #     'prefix-3': word[:3],
        #     'suffix-1': word[-1],
        #     'suffix-2': word[-2:],
        #     'suffix-3': word[-3:],

        #     'prefix-un': word[:2] == 'un',   #if word starts with un
        #     'prefix-re': word[:2] == 're',   #if word starts with re
        #     'prefix-over': word[:4] == 'over',  #if word starts with over
        #     'prefix-dis': word[:4] == 'dis',   #if word starts with dis
        #     'prefix-mis': word[:4] == 'mis',   #if word starts with mis
        #     'prefix-pre': word[:4] == 'pre',   #if word starts with pre
        #     'prefix-non': word[:4] == 'non',   #if word starts with non
        #     'prefix-de': word[:3] == 'de',     #if word starts with de
        #     'prefix-in': word[:3] == 'in',     #if word starts with in
        #     'prefix-en': word[:3] == 'en',     #if word starts with en

        #     'suffix-ed': word[-2:] == 'ed',   #if word ends with ed
        #     'suffix-ing': word[-3:] == 'ing',  #if word ends with ing
        #     'suffix-es': word[-2:] == 'es',    #if word ends with es
        #     'suffix-ly': word[-2:] == 'ly',    #if word ends with ly
        #     'suffix-ment': word[-4:] == 'ment',  #if word ends with ment
        #     'suffix-er': word[-2:] == 'er',     #if word ends with er
        #     'suffix-ive': word[-3:] == 'ive',
        #     'suffix-ous': word[-3:] == 'ous',
        #     'suffix-ness': word[-4:] == 'ness',
        #     'ends_with_s': word[-1] == 's',
        #     'ends_with_es': word[-2:] == 'es',

        #     'has_hyphen': '-' in word,    #if word has hypen
        #     'is_numeric': word.isdigit(),  #if word is in numeric
        #     'capitals_inside': word[1:].lower() != word[1:],
        #     'is_title_case': word.istitle(),  #if first letter is in uppercase

        # }
        # features = {
        #     'word': word,
        #     'prefix': prefix,
        #     'stem': lancaster.stem(word),
        #     'suffix': suffix,
        #     'position': i,
        #     'is_first': i == 0,
        #     'is_last': i == len(sentence) - 1,
        #     'is_all_caps': word.isupper(),
        #     'is_all_lower': word.islower(),

        #     'prev_word': sentence[i-1][0] if i > 0 else "<START>",
        #     'next_word': sentence[i+1][0] if i < len(sentence) - 1 else "<END>",
        #     'prev_is_capitalized': sentence[i-1][0].istitle() if i > 0 else False,
        #     'next_is_capitalized': sentence[i+1][0].istitle() if i < len(sentence) - 1 else False,
        #     'prev_is_numeric': sentence[i-1][0].isdigit() if i > 0 else False,
        #     'next_is_numeric': sentence[i+1][0].isdigit() if i < len(sentence) - 1 else False,
        #     'prev_suffix': self.word_splitter(sentence[i-1][0])[2] if i > 0 else "<START>",
        #     'next_suffix': self.word_splitter(sentence[i+1][0])[2] if i < len(sentence) - 1 else "<END>",
        #     'prev_prefix': self.word_splitter(sentence[i-1][0])[0] if i > 0 else "<START>",
        #     'next_prefix': self.word_splitter(sentence[i+1][0])[0] if i < len(sentence) - 1 else "<END>",
        # }

        features = {
            'word': word,
            'is_first': i == 0,
            'is_last': i == len(sentence) - 1,
            'is_capitalized': word[0].upper() == word[0],
            'is_all_caps': word.upper() == word, 
            'is_all_lower': word.lower() == word,

            'prefix-1': word[0],   
            'prefix-2': word[:2],
            'prefix-3': word[:3],
            'suffix-1': word[-1],
            'suffix-2': word[-2:],
            'suffix-3': word[-3:],
            
            'prev_word': '' if i == 0 else sentence[i-1][0],
            'next_word': '' if i == len(sentence)-1 else sentence[i+1][0],
            
            'has_hyphen': '-' in word,
            'is_numeric': word.isdigit(),
            'capitals_inside': word[1:].lower() != word[1:]
        }


        if i > 0:
            # prev_word, prev_postag = sentence[i-1]
            prev_word = sentence[i-1][0]
            prev_prefix, prev_stem, prev_suffix = self.word_splitter(prev_word)

            features.update({
                'prev_word': prev_word,
                # 'prev_postag': prev_postag,
                'prev_prefix': prev_prefix,
                'prev_stem': lancaster.stem(prev_word),
                'prev_suffix': prev_suffix,
                'prev:is_all_caps': prev_word.isupper(),
                'prev:is_all_lower': prev_word.islower(),
                'prev:is_numeric': prev_word.isdigit(),
                'prev:is_title_case': prev_word.istitle(),
            })

        if i < len(sentence)-1:
            next_word = sentence[i-1][0]
            next_prefix, next_stem, next_suffix = self.word_splitter(next_word)
            features.update({
                'next_word': next_word,
                'next_prefix': next_prefix,
                'next_stem': lancaster.stem(next_word),
                'next_suffix': next_suffix,
                'next:is_all_caps': next_word.isupper(),
                'next:is_all_lower': next_word.islower(),
                'next:is_numeric': next_word.isdigit(),
                'next:is_title_case': next_word.istitle(),
            })

        return features

    def train(self, data=None):
        if data:
            X_train, y_train = zip(*data)
        else:
            X_train, y_train = self.X_train, self.y_train

        print("Training CRF Model...", len(self.X_train), len(self.y_train))

        # Ensure X_train is a list of lists of dictionaries
        X_train = [list(map(dict, x)) for x in X_train]
        self.crf_model.fit(X_train, y_train)

    def predict(self, X_test):
        return self.crf_model.predict(X_test)

    def accuracy(self, test_data):
        X_test, y_test = zip(*test_data)
        y_pred = self.predict(X_test)
        self.actual_tag.extend([item for sublist in y_test for item in sublist])
        self.predicted_tag.extend([item for sublist in y_pred for item in sublist])
        print(len(self.actual_tag), len(self.predicted_tag))
        return metrics.flat_accuracy_score(y_test, y_pred)

    def cross_validation(self):
        validator = CRF_POS_Tagger()
        data = list(zip(self.X, self.y))
        print("Cross-Validation...")
        accuracies = []
        for i in range(5):
            n1 = int(i / 5.0 * len(data))
            n2 = int((i + 1) / 5.0 * len(data))
            test_data = data[n1:n2]
            train_data = data[:n1] + data[n2:]
            validator.train(train_data)
            acc = validator.accuracy(test_data)
            accuracies.append(acc)
        self.actual_tag = validator.actual_tag
        self.predicted_tag = validator.predicted_tag
        return accuracies, sum(accuracies) / 5.0

    def con_matrix(self):
        self.labels = np.unique(self.actual_tag)
        print(self.labels, self.actual_tag, self.predicted_tag)
        conf_matrix = confusion_matrix(self.actual_tag, self.predicted_tag, labels=self.labels)
        normalized_matrix = conf_matrix/np.sum(conf_matrix, axis=1, keepdims=True)
        plt.figure(figsize=(10, 7))
        sns.heatmap(normalized_matrix, annot=True, fmt='.2f', cmap='Blues', xticklabels=self.labels, yticklabels=self.labels)
        plt.xlabel('Predicted Tags')
        plt.ylabel('Actual Tags')
        plt.title('Confusion Matrix Heatmap')
        plt.savefig("Confusion_matrix.png")
        plt.show()

        return normalized_matrix

    def per_pos_accuracy(self, conf_matrix):
        print("Per Tag Precision, Recall, and F-Score:")
        per_tag_metrics = {}

        for i, tag in enumerate(self.labels):
            true_positives = conf_matrix[i, i]
            false_positives = np.sum(conf_matrix[:, i]) - true_positives
            false_negatives = np.sum(conf_matrix[i, :]) - true_positives

            precision = true_positives / (true_positives + false_positives) if (true_positives + false_positives) > 0 else 0
            recall = true_positives / (true_positives + false_negatives) if (true_positives + false_negatives) > 0 else 0
            f1_score = (2 * precision * recall) / (precision + recall) if (precision + recall) > 0 else 0
            beta_0_5 = 0.5
            beta_2 = 2.0

            f0_5_score = (1 + beta_0_5**2) * (precision * recall) / ((beta_0_5**2 * precision) + recall) if (precision + recall) > 0 else 0
            f2_score = (1 + beta_2**2) * (precision * recall) / ((beta_2**2 * precision) + recall) if (precision + recall) > 0 else 0

            per_tag_metrics[tag] = {
                'Precision': precision,
                'Recall': recall,
                'f1-Score': f1_score,
                'f05-Score': f0_5_score,
                'f2-Score': f2_score
            }

            print(f"{tag}: Precision = {precision:.2f}, Recall = {recall:.2f}, f1-Score = {f1_score:.2f}, "
                  f"f05-Score = {f0_5_score:.2f}, f2-Score = {f2_score:.2f}")

    def tagging(self, input):
        sentence = (re.sub(r'(\S)([.,;:!?])', r'\1 \2', input.strip())).split()
        sentence_list = [[word] for word in sentence]
        features = [self.word_features(sentence_list, i) for i in range(len(sentence_list))]

        predicted_tags = self.crf_model.predict([features])
        output = "".join(f"{sentence[i]}[{predicted_tags[0][i]}]    " for i in range(len(sentence)))
        return output


validate = CRF_POS_Tagger()
accuracies, avg_accuracy = validate.cross_validation()
print(f"Cross-Validation Accuracies: {accuracies}")
print(f"Average Accuracy: {avg_accuracy}")

conf_matrix = validate.con_matrix()
print(validate.per_pos_accuracy(conf_matrix))

tagger = CRF_POS_Tagger(True)
interface = gr.Interface(fn = tagger.tagging,
                         inputs = gr.Textbox(
                             label="Input Sentence",
                             placeholder="Enter your sentence here...",
                         ),
                         outputs = gr.Textbox(
                             label="Tagged Output",
                             placeholder="Tagged sentence appears here...",
                         ),
                         title = "Conditional Random Field POS Tagger",
                         description = "CS626 Assignment 1B (Autumn 2024)",
                         theme=gr.themes.Soft())
interface.launch(inline = False, share = True)