<jupyter_start><jupyter_text>Sarcasm Model Definition<jupyter_code>import torch
from torchtext.legacy import data
from torchtext.legacy import datasets
import random
import numpy as np

SEED = 1234

random.seed(SEED)
np.random.seed(SEED)
torch.manual_seed(SEED)
torch.backends.cudnn.deterministic = True

# create fields in emotion dataset
TEXT = data.Field(tokenize = 'spacy', 
                  tokenizer_language = 'en_core_web_sm',
                  batch_first = True
                  )
LABEL = data.LabelField(dtype=torch.float)

# tuples representative of tabular format
fields = [
          ('is_sarcastic', LABEL),
          ('headline', TEXT)
]

# LOAD SARCASM DATA
sarc_train, = data.TabularDataset.splits(
    path='/content',
    train='Sarcasm_Headlines_Dataset_v2.csv',
    format='csv',
    fields=fields,
    skip_header=True
)
MAX_VOCAB_SIZE = 25_000

sarc_train, sarc_valid, sarc_test = sarc_train.split(random_state = random.seed(SEED), split_ratio=[0.4,0.3,0.3])

TEXT.build_vocab(sarc_train, 
                 max_size = MAX_VOCAB_SIZE, 
                 vectors = "glove.6B.100d", 
                 unk_init = torch.Tensor.normal_)

LABEL.build_vocab(sarc_train)
BATCH_SIZE = 64

# device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
device = torch.device('cpu')

sarc_train_iterator, sarc_valid_iterator, sarc_test_iterator = data.BucketIterator.splits(
    (sarc_train, sarc_valid, sarc_test), 
    batch_size = BATCH_SIZE, 
    device = device,
    sort=False)
INPUT_DIM = len(TEXT.vocab)
EMBEDDING_DIM = 100
N_FILTERS = 100
FILTER_SIZES = [3,4,5]
OUTPUT_DIM = 1
DROPOUT = 0.5
PAD_IDX = TEXT.vocab.stoi[TEXT.pad_token]

model = SUPER_CNN(INPUT_DIM, EMBEDDING_DIM, N_FILTERS, FILTER_SIZES, OUTPUT_DIM, DROPOUT, PAD_IDX)
def count_parameters(model):
    return sum(p.numel() for p in model.parameters() if p.requires_grad)

print(f'The model has {count_parameters(model):,} trainable parameters')
pretrained_embeddings = TEXT.vocab.vectors
model.embedding.weight.data.copy_(pretrained_embeddings)

# zero initial weights of <unk> and <pad>
UNK_IDX = TEXT.vocab.stoi[TEXT.unk_token]

model.embedding.weight.data[UNK_IDX] = torch.zeros(EMBEDDING_DIM)
model.embedding.weight.data[PAD_IDX] = torch.zeros(EMBEDDING_DIM)
import torch.optim as optim

optimizer = optim.Adam(model.parameters())
criterion = nn.BCEWithLogitsLoss()
model = model.to(device)
criterion = criterion.to(device)
N_EPOCHS = 10
best_valid_loss = float('inf')

for epoch in range(N_EPOCHS):
    start_time = time.time()
    
    train_loss, train_acc = train(model, sarc_train_iterator, optimizer, criterion)
    valid_loss, valid_acc = evaluate(model, sarc_valid_iterator, criterion)
    
    end_time = time.time()

    epoch_mins, epoch_secs = epoch_time(start_time, end_time)
    
    if valid_loss < best_valid_loss:
        best_valid_loss = valid_loss
        torch.save(model.state_dict(), 'CNN-sarcasm-extreme.pt')
    
    print(f'Epoch: {epoch+1:02} | Epoch Time: {epoch_mins}m {epoch_secs}s')
    print(f'\tTrain Loss: {train_loss:.3f} | Train Acc: {train_acc*100:.2f}%')
    print(f'\t Val. Loss: {valid_loss:.3f} |  Val. Acc: {valid_acc*100:.2f}%')
# TESTING
model.load_state_dict(torch.load('CNN-sarcasm-extreme.pt'))

test_loss, test_acc = evaluate(model, sarc_test_iterator, criterion)

print(f'Test Loss: {test_loss:.3f} | Test Acc: {test_acc*100:.2f}%')
def binary_accuracy(preds, y):
    """
    Returns accuracy per batch, i.e. if you get 8/10 right, this returns 0.8, NOT 8
    """

    #round predictions to the closest integer
    rounded_preds = torch.round(torch.sigmoid(preds))
    correct = (rounded_preds == y).float() #convert into float for division 
    acc = correct.sum() / len(correct)
    return acc

def train(model, iterator, optimizer, criterion):
    
    epoch_loss = 0
    epoch_acc = 0
    
    model.train()
    
    for batch in iterator:
        # print(batch.__dict__)
        optimizer.zero_grad()
        predictions = model(batch.headline).squeeze(1)
        loss = criterion(predictions, batch.is_sarcastic)
        acc = binary_accuracy(predictions, batch.is_sarcastic)
        loss.backward()
        optimizer.step()
        
        epoch_loss += loss.item()
        epoch_acc += acc.item()
        
    return epoch_loss / len(iterator), epoch_acc / len(iterator)

def evaluate(model, iterator, criterion):
    
    epoch_loss = 0
    epoch_acc = 0
    
    model.eval()
    
    with torch.no_grad():
        for batch in iterator:
            predictions = model(batch.headline).squeeze(1)
            loss = criterion(predictions, batch.is_sarcastic)
            acc = binary_accuracy(predictions, batch.is_sarcastic)

            epoch_loss += loss.item()
            epoch_acc += acc.item()
        
    return epoch_loss / len(iterator), epoch_acc / len(iterator)

import time

def epoch_time(start_time, end_time):
    elapsed_time = end_time - start_time
    elapsed_mins = int(elapsed_time / 60)
    elapsed_secs = int(elapsed_time - (elapsed_mins * 60))
    return elapsed_mins, elapsed_secs
from torchtext.legacy import data
from torchtext.legacy import datasets
import torch.optim as optim
import random
import numpy as np
import time

SEED = 1234


class ModelTrainer():

    def __init__(self, model_name, model_type=None, text_field_name=None,
                 label_field_name=None, IMDB=False, train_file_name=None,
                 test_file_name=None, embedding_dim=100, criterion_str="CE",
                 max_vocab_size=25000, num_filters=100, filter_sizes=[3, 4, 5],
                 dropout=0.5, output_dim=1, batch_size=64, dtype=torch.long):
        self.model_name = model_name
        self.model_type = model_type
        self.text_field_name = text_field_name
        self.label_field_name = label_field_name
        self.IMDB = IMDB
        self.train_file_name = train_file_name
        self.test_file_name = test_file_name
        self.criterion_str = criterion_str
        self.embedding_dim = embedding_dim
        self.max_vocab_size = max_vocab_size
        self.num_filters = num_filters
        self.filter_sizes = filter_sizes
        self.dropout = dropout
        self.batch_size = batch_size
        self.dtype = dtype

    def load_dataset(self):
        """ Loads the dataset using torchtext
        """
        r_seed = random.seed(SEED)
        np.random.seed(SEED)
        torch.manual_seed(SEED)
        torch.backends.cudnn.deterministic = True

        self.TEXT = data.Field(
            tokenize='spacy',
            tokenizer_language='en_core_web_sm',
            batch_first=True)
        self.LABEL = data.LabelField(dtype=self.dtype) # dtype=torch.float)
        # LOAD DATASET
        if self.IMDB:
            self.dataset, test_data = datasets.IMDB.splits(self.TEXT, self.LABEL)
            train_data, valid_data = self.dataset.split(random_state=r_seed)
        else:
            if self.text_field_name == "Sentence":
                # Formality and Informativeness don't work with dict fields
                # tuples representative of tabular format
                fields = [
                          (self.label_field_name, self.LABEL),
                          (self.text_field_name, self.TEXT)
                ]
                skip_header = True
            else:
                fields = {
                    self.label_field_name: (self.label_field_name, self.LABEL),
                    self.text_field_name: (self.text_field_name, self.TEXT)
                }
                skip_header = False
            format = self.train_file_name.split('.')[-1]
            if self.test_file_name:
                
                self.dataset, test_data = data.TabularDataset.splits(
                    path='/content',
                    train=self.train_file_name,
                    test=self.test_file_name,
                    format=format,
                    fields=fields,
                    skip_header=skip_header
                )
                train_data, valid_data = self.dataset.split(random_state=r_seed)
            else:
                self.dataset = data.TabularDataset.splits(
                    path='/content',
                    train=self.train_file_name,
                    format=format,
                    fields=fields,
                    skip_header=skip_header
                )[0]
                train_data, valid_data, test_data = self.dataset.split(
                    split_ratio=[0.4, 0.3, 0.3], random_state=r_seed)

        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
        print(torch.cuda.get_device_name(0))

        self.train_iter, self.val_iter, self.test_iter = data.BucketIterator.splits(
            (train_data, valid_data, test_data),
            batch_size=self.batch_size,
            device=self.device,
            sort=False
            )

    def build_vocab(self):
        """ Builds the vocabulary of the dataset using torchtext
        """
        self.TEXT.build_vocab(
            self.dataset,
            max_size=self.max_vocab_size,
            vectors="glove.6B.100d",
            unk_init=torch.Tensor.normal_
        )
        self.LABEL.build_vocab(self.dataset)

    def init_model(self):
        """ Create CNN model with the supplied/derived parameters.
            Loads "glove.6B.100d" weights into the model's embedding layer
        """
        # Get the size of the vocabulary
        vocab_size = len(self.TEXT.vocab)
        if self.criterion_str == "CE":
          output_dim = len(self.LABEL.vocab)
        else:
          output_dim = 1
        pad_idx = self.TEXT.vocab.stoi[self.TEXT.pad_token]
        # Initialize model
        if self.model_type == 'deep':
          self.model = VDCNN(output_dim, vocab_size, pad_idx,
                             self.embedding_dim, shortcut=True)
        else:
          self.model = CNN(vocab_size, self.embedding_dim, self.num_filters,
                           self.filter_sizes, output_dim, self.dropout,
                           pad_idx)
        # Get pretrained weights from vocab
        pretrained_embeddings = self.TEXT.vocab.vectors
        self.model.embedding.weight.data.copy_(pretrained_embeddings)

        # zero initial weights of <unk> and <pad>
        unk_index = self.TEXT.vocab.stoi[self.TEXT.unk_token]

        self.model.embedding.weight.data[unk_index] = torch.zeros(self.embedding_dim)
        self.model.embedding.weight.data[pad_idx] = torch.zeros(self.embedding_dim)

    def init_optimizer(self):
        """ Initializes the optimizor (Adam) and the criterion (BCEWithLogitsLoss)
        """
        self.optimizer = optim.Adam(self.model.parameters())
        self.model = self.model.to(self.device)
        if self.criterion_str == "CE":
            self.criterion = nn.CrossEntropyLoss()
        elif self.criterion_str == "BCE":
            self.criterion = nn.BCEWithLogitsLoss()
        self.criterion = self.criterion.to(self.device)

    def train_model(self, num_epochs):
        """ Trains and validates the model (and prints the results) for the
            given number of epochs. Saves the model from the epoch which
            yeilded the lowest validation loss.

        Args:
            num_epochs (int): number of epochs to train the model
        """
        print(self.model_name)
        best_valid_loss = float('inf')

        for epoch in range(num_epochs):
            start_time = time.time()

            train_loss, train_acc = self.train_epoch()
            valid_loss, valid_acc = self.evaluate_epoch(self.val_iter)

            end_time = time.time()

            epoch_mins, epoch_secs = epoch_time(start_time, end_time)

            if valid_loss < best_valid_loss:
                best_valid_loss = valid_loss
                # torch.save(self.model.state_dict(), f'{self.model_name}.pt')
                torch.save(self.model, f'{self.model_name}.pt')

            print(f'Epoch: {epoch+1:02} | Epoch Time: {epoch_mins}m {epoch_secs}s')
            print(f'\tTrain Loss: {train_loss:.3f} | Train Acc: {train_acc*100:.2f}%')
            print(f'\t Val. Loss: {valid_loss:.3f} |  Val. Acc: {valid_acc*100:.2f}%')

    def test_model(self):
        """ Tests the model with the highest validation loss on the test iterator
        """
        # self.model.load_state_dict(torch.load(f'{self.model_name}.pt'))
        torch.load(f'{self.model_name}.pt')
        test_loss, test_acc = self.evaluate_epoch(self.test_iter)
        print(f'Test Loss: {test_loss:.3f} | Test Acc: {test_acc*100:.2f}%')
        with open('model_results.txt', "a") as f:
            f.write(f'{self.model_name}\nTest Loss: {test_loss:.3f} | Test Acc: {test_acc*100:.2f}%\n')

    def count_parameters(self):
        num_params = sum(p.numel() for p in self.model.parameters() if p.requires_grad)
        print(f'The model has {num_params:,} trainable parameters')

    def train_epoch(self):
        """ Trains the model for 1 epoch of the train iterator

        Returns:
            tuple: (loss, accuracy)
        """
        epoch_loss = 0
        epoch_acc = 0
        self.model.train()
        for batch in self.train_iter:
            # print(batch.__dict__)
            self.optimizer.zero_grad()
            predictions = self.model(getattr(batch, batch.input_fields[0]))
            # Reduce dimentionality if using binary cross entropy loss
            if self.criterion_str == "BCE":
                predictions = predictions.squeeze(1)
                acc_formula = binary_accuracy
            elif self.criterion_str == "CE":
                acc_formula = categorical_accuracy
            loss = self.criterion(predictions, getattr(batch, batch.target_fields[0]))
            acc = acc_formula(predictions, getattr(batch, batch.target_fields[0]))
            loss.backward()
            self.optimizer.step()

            epoch_loss += loss.item()
            epoch_acc += acc.item()

        return epoch_loss / len(self.train_iter), epoch_acc / len(self.train_iter)

    def evaluate_epoch(self, iterator):
        epoch_loss = 0
        epoch_acc = 0
        self.model.eval()
        with torch.no_grad():
            for batch in iterator:
                predictions = self.model(getattr(batch, batch.input_fields[0]))
                # Reduce dimentionality if using binary cross entropy loss
                if self.criterion_str == "BCE":
                    predictions = predictions.squeeze(1)
                    acc_formula = binary_accuracy
                elif self.criterion_str == "CE":
                    acc_formula = categorical_accuracy
                loss = self.criterion(predictions, getattr(batch, batch.target_fields[0]))
                acc = acc_formula(predictions, getattr(batch, batch.target_fields[0]))

                epoch_loss += loss.item()
                epoch_acc += acc.item()
        return epoch_loss / len(iterator), epoch_acc / len(iterator)
      
    def do_what_we_want(self, num_epochs=10):
        self.load_dataset()
        self.build_vocab()
        self.init_model()
        self.init_optimizer()
        self.train_model(num_epochs)
        self.test_model()


def binary_accuracy(preds, y):
    """
    Returns accuracy per batch, i.e. if you get 8/10 right, this returns 0.8, NOT 8
    """
    # round predictions to the closest integer
    rounded_preds = torch.round(torch.sigmoid(preds))
    correct = (rounded_preds == y).float()  # convert into float for division
    acc = correct.sum() / len(correct)
    return acc


def categorical_accuracy(preds, y):
    """
    Returns accuracy per batch, i.e. if you get 8/10 right, this returns 0.8, NOT 8
    """
    top_pred = preds.argmax(1, keepdim = True)
    correct = top_pred.eq(y.view_as(top_pred)).sum()
    acc = correct.float() / y.shape[0]
    return acc
  

def epoch_time(start_time, end_time):
    elapsed_time = end_time - start_time
    elapsed_mins = int(elapsed_time / 60)
    elapsed_secs = int(elapsed_time - (elapsed_mins * 60))
    return elapsed_mins, elapsed_secs

import gc

if __name__ == "__main__":
    # clear contents of file
    with open("model_results.txt", "w") as f:
        pass
    

    # ======================== FORMALITY BINARY MODEL ========================= #
    trainer = ModelTrainer(
        model_name="CNN-formality-binary", 
        text_field_name='Sentence', 
        label_field_name='Formality', 
        train_file_name='mturk_news_formality_BINARY.csv', 
        criterion_str='BCE',
        dtype=torch.float)
    trainer.do_what_we_want(num_epochs=8)

    
    # ======================== INFORMATIVENESS BINARY MODEL ========================= #
    trainer = ModelTrainer(
        model_name="CNN-informativeness-binary", 
        text_field_name='Sentence', 
        label_field_name='Informativeness', 
        train_file_name='mturk_news_informativeness_BINARY.csv', 
        criterion_str='BCE',
        dtype=torch.float)
    trainer.do_what_we_want(num_epochs=8)


    # ======================== EMOTION MODEL ========================= #
    trainer = ModelTrainer(
        model_name="CNN-emotion", 
        text_field_name='Text', 
        label_field_name='Emotion', 
        train_file_name='emotion_train.csv', 
        test_file_name='emotion_test.csv',
        criterion_str='CE'
    )
    trainer.do_what_we_want(num_epochs=8)
        # --------------- deep archetecture -------------------#
    # trainer = ModelTrainer(
    #     model_name="deep-CNN-emotion", 
    #     model_type='deep', 
    #     text_field_name='Text', 
    #     label_field_name='Emotion', 
    #     train_file_name='/content/emotion_train.csv', 
    #     test_file_name='/content/emotion_test.csv'
    # )
    # trainer.do_what_we_want(num_epochs=15)


    # ======================== SENTIMENT MODEL ========================= #
    trainer = ModelTrainer("CNN-sentiment", IMDB=True, criterion_str="BCE", dtype=torch.float)
    trainer.do_what_we_want(num_epochs=8)


    # ======================== SARCASM BASE MODEL ========================= #
    # trainer = ModelTrainer(
    #     model_name="CNN-sarcasm-base", 
    #     text_field_name='headline', 
    #     label_field_name='is_sarcastic', 
    #     train_file_name='Sarcasm_Headlines_Dataset_v2.csv', 
    #     criterion_str='BCE',
    #     dtype=torch.float)
    # trainer.do_what_we_want(num_epochs=8)



    # emotion_trainer = ModelTrainer(
    #     model_name="CNN-emotion", 
    #     text_field_name='Text', 
    #     label_field_name='Emotion', 
    #     train_file_name='emotion_train.csv', 
    #     test_file_name='emotion_test.csv'
    # )
    # emotion_trainer.do_what_we_want(num_epochs=8)
    # # Clear GPU memory
    # emotion_trainer = None
    # gc.collect()
    # torch.cuda.empty_cache()

    # deep_sentament_trainer = ModelTrainer("deep-CNN-sentiment", model_type='deep', IMDB=True, criterion_str="BCE")
    # deep_sentament_trainer.do_what_we_want(num_epochs=8)
    # # Clear GPU memory
    # deep_sentament_trainer = None
    # gc.collect()
    # torch.cuda.empty_cache()<jupyter_output>.vector_cache/glove.6B.zip: 0.00B [00:00, ?B/s]
{'fraction_non_alphanumeric': 0.11821553570495742, 'fraction_numerical': 0.009664106984276237, 'mean_word_length': 2.4737797133006714, 'pattern_counts': {'":': 9, '<': 10, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '19932', 'license': 'MIT', 'n_tokens_mistral': 6417, 'n_tokens_neox': 6145, 'n_words': 1130, 'path': 'notebooks/8_CNN_Master.ipynb', 'repo_name': 'gavinmcclelland/sarcasm-detection', 'size': '19143'}