# -*- coding: utf-8 -*-
"""code5-few_more_tests.ipynb

Automatically generated by Colaboratory.

Original file is located at
    https://colab.research.google.com/drive/1T6bzB0oXSKtr_hfviLkpt7hmHBWvT73Y
"""

!pip install transformers torch
!pip install accelerate -U

from google.colab import files
import zipfile
import os

uploaded = files.upload()

for fn in uploaded.keys():
    print('User uploaded file "{name}" with length {length} bytes'.format(
        name=fn, length=len(uploaded[fn])))

    if fn.endswith('.zip'):
        with zipfile.ZipFile(fn, 'r') as zip_ref:
            zip_ref.extractall()

        print(f"Files from '{fn}' unzipped in the current directory!")
    else:
        print(f"'{fn}' is not a zip file.")

os.listdir()

from transformers import (RobertaTokenizer, RobertaForSequenceClassification,
                          TrainingArguments, Trainer)
from sklearn.metrics import classification_report
from sklearn.model_selection import train_test_split
from torch.utils.data import Dataset
import json
import numpy as np
import torch

def compute_metrics(p):
    pred_labels = np.argmax(p.predictions, axis=1)
    true_labels = p.label_ids
    metrics = classification_report(true_labels, pred_labels, output_dict=True)
    accuracy = metrics['accuracy']
    weighted_avg = metrics['weighted avg']
    precision = weighted_avg['precision']
    f1 = weighted_avg['f1-score']
    return {
        'accuracy': accuracy,
        'f1': f1,
        'precision': precision
    }

# Load data
with open('surprise.data', 'r') as data_file:
    data = [json.loads(line) for line in data_file]
with open('surprise.solution', 'r') as solution_file:
    solutions = [json.loads(line) for line in solution_file]

intent_map = {item['indoml_id']: item['intent'] for item in solutions}
indoml_ids = [item['indoml_id'] for item in data]
intents = [intent_map[indoml_id] for indoml_id in indoml_ids]
utt = [item['utt'] for item in data]

label_map = {intent: i for i, intent in enumerate(set(intents))}
train_labels = [label_map[label] for label in intents]

utt_train, utt_test, labels_train, labels_test = train_test_split(utt, train_labels, test_size=0.3, random_state=42)

max_length = 64

class IntentDataset(Dataset):
    def __init__(self, encodings, labels):
        self.encodings = encodings
        self.labels = labels

    def __getitem__(self, idx):
        item = {key: torch.tensor(val[idx]) for key, val in self.encodings.items()}
        item['labels'] = torch.tensor(self.labels[idx])
        return item

    def __len__(self):
        return len(self.labels)

training_args = TrainingArguments(
    output_dir='./results',
    save_strategy="epoch",
    num_train_epochs=50,
    per_device_train_batch_size=16,
    per_device_eval_batch_size=64,
    warmup_steps=500,
    weight_decay=0.01,
    logging_dir='./logs',
    logging_steps=50,
    learning_rate=2e-5,
    evaluation_strategy="epoch",
)

# Assuming num_classes is defined earlier in the code. If not, define it:
num_classes = len(label_map)


transfer_tokenizer = RobertaTokenizer.from_pretrained('projecte-aina/roberta-base-ca-v2-massive')
transfer_model = RobertaForSequenceClassification.from_pretrained('projecte-aina/roberta-base-ca-v2-massive', num_labels=num_classes, ignore_mismatched_sizes=True)

transfer_train_encodings = transfer_tokenizer(utt_train, truncation=True, padding=True, max_length=max_length, return_tensors='pt')
transfer_test_encodings = transfer_tokenizer(utt_test, truncation=True, padding=True, max_length=max_length, return_tensors='pt')
transfer_train_dataset = IntentDataset(transfer_train_encodings, labels_train)
transfer_test_dataset = IntentDataset(transfer_test_encodings, labels_test)

for param in transfer_model.parameters():
    param.requires_grad = False
for param in transfer_model.classifier.parameters():
    param.requires_grad = True

transfer_trainer = Trainer(
    model=transfer_model,
    args=training_args,
    train_dataset=transfer_train_dataset,
    eval_dataset=transfer_test_dataset,
    compute_metrics=compute_metrics
)
transfer_trainer.train()
transfer_results = transfer_trainer.evaluate()

def display_results(name, results):
    print(f"Results for {name}:")
    print(f"Accuracy: {results['eval_accuracy']}")
    print(f"F1: {results['eval_f1']}")
    print(f"Precision: {results['eval_precision']}")

display_results("Transfer RoBERTa", transfer_results)

from transformers import (RobertaTokenizer, RobertaForSequenceClassification,
                          TrainingArguments, Trainer, get_linear_schedule_with_warmup)
from sklearn.metrics import classification_report
from sklearn.model_selection import train_test_split
from torch.utils.data import Dataset
import json
import numpy as np
import torch


def compute_metrics(p):
    pred_labels = np.argmax(p.predictions, axis=1)
    true_labels = p.label_ids
    metrics = classification_report(true_labels, pred_labels, output_dict=True)
    accuracy = metrics['accuracy']
    weighted_avg = metrics['weighted avg']
    precision = weighted_avg['precision']
    f1 = weighted_avg['f1-score']
    return {
        'accuracy': accuracy,
        'f1': f1,
        'precision': precision
    }


# Load data
with open('surprise.data', 'r') as data_file:
    data = [json.loads(line) for line in data_file]
with open('surprise.solution', 'r') as solution_file:
    solutions = [json.loads(line) for line in solution_file]

intent_map = {item['indoml_id']: item['intent'] for item in solutions}
indoml_ids = [item['indoml_id'] for item in data]
intents = [intent_map[indoml_id] for indoml_id in indoml_ids]
utt = [item['utt'] for item in data]

label_map = {intent: i for i, intent in enumerate(set(intents))}
train_labels = [label_map[label] for label in intents]

utt_train, utt_test, labels_train, labels_test = train_test_split(utt, train_labels, test_size=0.3, random_state=42)

max_length = 64


class IntentDataset(Dataset):
    def __init__(self, encodings, labels):
        self.encodings = encodings
        self.labels = labels

    def __getitem__(self, idx):
        item = {key: torch.tensor(val[idx]) for key, val in self.encodings.items()}
        item['labels'] = torch.tensor(self.labels[idx])
        return item

    def __len__(self):
        return len(self.labels)


def unfreeze_model_layers(model, num_layers):
    for child in list(model.children())[:-num_layers]:
        for param in child.parameters():
            param.requires_grad = False
    for child in list(model.children())[-num_layers:]:
        for param in child.parameters():
            param.requires_grad = True


training_args = TrainingArguments(
    output_dir='./results',
    save_strategy="steps",   # Change to "steps"
    num_train_epochs=50,
    per_device_train_batch_size=16,
    per_device_eval_batch_size=64,
    warmup_steps=500,
    weight_decay=0.01,
    logging_dir='./logs',
    logging_steps=50,
    learning_rate=2e-5,
    evaluation_strategy="steps",  # Change to "steps"
    eval_steps=50,
    save_total_limit=2,
    load_best_model_at_end=True
)


num_classes = len(label_map)

transfer_tokenizer = RobertaTokenizer.from_pretrained('projecte-aina/roberta-base-ca-v2-massive')
transfer_model = RobertaForSequenceClassification.from_pretrained('projecte-aina/roberta-base-ca-v2-massive', num_labels=num_classes, ignore_mismatched_sizes=True)

transfer_train_encodings = transfer_tokenizer(utt_train, truncation=True, padding=True, max_length=max_length, return_tensors='pt')
transfer_test_encodings = transfer_tokenizer(utt_test, truncation=True, padding=True, max_length=max_length, return_tensors='pt')
transfer_train_dataset = IntentDataset(transfer_train_encodings, labels_train)
transfer_test_dataset = IntentDataset(transfer_test_encodings, labels_test)

unfreeze_model_layers(transfer_model, 2)  # Initially unfreeze the last 2 layers for training

optimizer = torch.optim.AdamW(transfer_model.parameters(), lr=2e-5)
scheduler = get_linear_schedule_with_warmup(optimizer, num_warmup_steps=500, num_training_steps=len(transfer_train_dataset))

transfer_trainer = Trainer(
    model=transfer_model,
    args=training_args,
    train_dataset=transfer_train_dataset,
    eval_dataset=transfer_test_dataset,
    compute_metrics=compute_metrics,
    optimizers=(optimizer, scheduler)
)
transfer_trainer.train()
transfer_results = transfer_trainer.evaluate()

def display_results(name, results):
    print(f"Results for {name}:")
    print(f"Accuracy: {results['eval_accuracy']}")
    print(f"F1: {results['eval_f1']}")
    print(f"Precision: {results['eval_precision']}")

display_results("Transfer RoBERTa", transfer_results)

!pip install sentencepiece
import json
import numpy as np
import torch
from transformers import (BertTokenizer, BertForSequenceClassification,
                          RobertaTokenizer, RobertaForSequenceClassification,
                          XLNetTokenizer, XLNetForSequenceClassification,
                          DebertaTokenizer, DebertaForSequenceClassification,
                          BartTokenizer, BartForSequenceClassification,
                          TrainingArguments, Trainer)
from sklearn.model_selection import train_test_split
from torch.utils.data import Dataset

from sklearn.metrics import classification_report

def compute_metrics(p):
    pred_labels = np.argmax(p.predictions, axis=1)
    true_labels = p.label_ids
    metrics = classification_report(true_labels, pred_labels, output_dict=True)
    accuracy = metrics['accuracy']
    weighted_avg = metrics['weighted avg']
    precision = weighted_avg['precision']
    f1 = weighted_avg['f1-score']
    return {
        'accuracy': accuracy,
        'f1': f1,
        'precision': precision
    }


# Load surprise data
with open('surprise.data', 'r') as data_file:
    data = [json.loads(line) for line in data_file]
with open('surprise.solution', 'r') as solution_file:
    solutions = [json.loads(line) for line in solution_file]

intent_map = {item['indoml_id']: item['intent'] for item in solutions}
indoml_ids = [item['indoml_id'] for item in data]
intents = [intent_map[indoml_id] for indoml_id in indoml_ids]
utt = [item['utt'] for item in data]

label_map = {intent: i for i, intent in enumerate(set(intents))}
train_labels = [label_map[label] for label in intents]

utt_train, utt_test, labels_train, labels_test = train_test_split(utt, train_labels, test_size=0.3, random_state=42)

max_length = 64

class IntentDataset(Dataset):
    def __init__(self, encodings, labels):
        self.encodings = encodings
        self.labels = labels

    def __getitem__(self, idx):
        item = {key: torch.tensor(val[idx]) for key, val in self.encodings.items()}
        item['labels'] = torch.tensor(self.labels[idx])
        return item

    def __len__(self):
        return len(self.labels)

# Common training arguments
training_args = TrainingArguments(
    output_dir='./results',
    save_strategy="epoch",
    num_train_epochs=12,
    per_device_train_batch_size=16,
    per_device_eval_batch_size=64,
    warmup_steps=500,
    weight_decay=0.01,
    logging_dir='./logs',
    logging_steps=50,
    learning_rate=2e-5,
    evaluation_strategy="epoch",
)

# ... [previous imports]

# Importing XLNet and DeBERTa v2 XL components
from transformers import (XLNetTokenizer, XLNetForSequenceClassification,
                          DebertaV2Tokenizer, DebertaV2ForSequenceClassification)

# ... [previous code]

# XLNet
xlnet_tokenizer = XLNetTokenizer.from_pretrained("xlnet-large-cased")
xlnet_model = XLNetForSequenceClassification.from_pretrained("xlnet-large-cased", num_labels=len(label_map))
xlnet_train_encodings = xlnet_tokenizer(utt_train, truncation=True, padding=True, max_length=max_length, return_tensors='pt')
xlnet_test_encodings = xlnet_tokenizer(utt_test, truncation=True, padding=True, max_length=max_length, return_tensors='pt')
xlnet_train_dataset = IntentDataset(xlnet_train_encodings, labels_train)
xlnet_test_dataset = IntentDataset(xlnet_test_encodings, labels_test)

xlnet_trainer = Trainer(
    model=xlnet_model,
    args=training_args,
    train_dataset=xlnet_train_dataset,
    eval_dataset=xlnet_test_dataset,
    compute_metrics=compute_metrics
)
xlnet_trainer.train()
xlnet_results = xlnet_trainer.evaluate()

# DeBERTa v2 XL
debv2_tokenizer = DebertaV2Tokenizer.from_pretrained("microsoft/deberta-v2-xl")
debv2_model = DebertaV2ForSequenceClassification.from_pretrained("microsoft/deberta-v2-xl", num_labels=len(label_map))
debv2_train_encodings = debv2_tokenizer(utt_train, truncation=True, padding=True, max_length=max_length, return_tensors='pt')
debv2_test_encodings = debv2_tokenizer(utt_test, truncation=True, padding=True, max_length=max_length, return_tensors='pt')
debv2_train_dataset = IntentDataset(debv2_train_encodings, labels_train)
debv2_test_dataset = IntentDataset(debv2_test_encodings, labels_test)

debv2_trainer = Trainer(
    model=debv2_model,
    args=training_args,
    train_dataset=debv2_train_dataset,
    eval_dataset=debv2_test_dataset,
    compute_metrics=compute_metrics
)
debv2_trainer.train()
debv2_results = debv2_trainer.evaluate()

display_results("XLNet", xlnet_results)
display_results("DeBERTa v2 XL", debv2_results)

!pip install sentencepiece
import json
import numpy as np
import torch
from transformers import (XLNetTokenizer, XLNetForSequenceClassification,
                          DebertaV2Tokenizer, DebertaV2ForSequenceClassification,
                          TrainingArguments, Trainer)
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report
from torch.utils.data import Dataset

def compute_metrics(p):
    pred_labels = np.argmax(p.predictions, axis=1)
    true_labels = p.label_ids
    metrics = classification_report(true_labels, pred_labels, output_dict=True)
    accuracy = metrics['accuracy']
    weighted_avg = metrics['weighted avg']
    precision = weighted_avg['precision']
    f1 = weighted_avg['f1-score']
    return {
        'accuracy': accuracy,
        'f1': f1,
        'precision': precision
    }

def display_results(model_name, results):
    print(f"Results for {model_name}:")
    for key, value in results.items():
        print(f"{key}: {value:.4f}")
    print("\n")

# Load surprise data
with open('surprise.data', 'r') as data_file:
    data = [json.loads(line) for line in data_file]

with open('surprise.solution', 'r') as solution_file:
    solutions = [json.loads(line) for line in solution_file]

intent_map = {item['indoml_id']: item['intent'] for item in solutions}
indoml_ids = [item['indoml_id'] for item in data]
intents = [intent_map[indoml_id] for indoml_id in indoml_ids]
utt = [item['utt'] for item in data]

label_map = {intent: i for i, intent in enumerate(set(intents))}
train_labels = [label_map[label] for label in intents]

utt_train, utt_test, labels_train, labels_test = train_test_split(utt, train_labels, test_size=0.3, random_state=42)

max_length = 64

class IntentDataset(Dataset):
    def __init__(self, encodings, labels):
        self.encodings = encodings
        self.labels = labels

    def __getitem__(self, idx):
        item = {key: torch.tensor(val[idx]) for key, val in self.encodings.items()}
        item['labels'] = torch.tensor(self.labels[idx])
        return item

    def __len__(self):
        return len(self.labels)

training_args = TrainingArguments(
    output_dir='./results',
    save_strategy="epoch",
    num_train_epochs=12,
    per_device_train_batch_size=16,
    per_device_eval_batch_size=64,
    warmup_steps=500,
    weight_decay=0.01,
    logging_dir='./logs',
    logging_steps=50,
    learning_rate=2e-5,
    evaluation_strategy="epoch",
)


# DeBERTa v2 XL
debv2_tokenizer = DebertaV2Tokenizer.from_pretrained("microsoft/deberta-v2-xlarge")
debv2_model = DebertaV2ForSequenceClassification.from_pretrained("microsoft/deberta-v2-xlarge", num_labels=len(label_map))
debv2_train_encodings = debv2_tokenizer(utt_train, truncation=True, padding=True, max_length=max_length, return_tensors='pt')
debv2_test_encodings = debv2_tokenizer(utt_test, truncation=True, padding=True, max_length=max_length, return_tensors='pt')
debv2_train_dataset = IntentDataset(debv2_train_encodings, labels_train)
debv2_test_dataset = IntentDataset(debv2_test_encodings, labels_test)

debv2_trainer = Trainer(
    model=debv2_model,
    args=training_args,
    train_dataset=debv2_train_dataset,
    eval_dataset=debv2_test_dataset,
    compute_metrics=compute_metrics
)
debv2_trainer.train()
debv2_results = debv2_trainer.evaluate()

display_results("XLNet", xlnet_results)
display_results("DeBERTa v2 XL", debv2_results)