Create app.py

app.py

@@ -0,0 +1,317 @@
+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)