import torch
from torch import nn
from transformers import BertModel, BertConfig
from huggingface_hub import login
import json
import math
login(token="<TRUFFLEHOG_REDACTED_HUGGINGFACE>")




class BertConfig:
    def __init__(self):
        self.vocab_size = 1000
        self.num_labels = 10
        self.hidden_size = 768
        self.intermediate_size = 3072
        self.layers = 12
        self.max_position_embeddings = 512
        self.type_token_embeddings = 2
        self.emb_drop = 0.1
        self.attention_heads = 12
        self.attention_drop = 0.0
        self.dropout_rate = 0.1
        self.classifier_dropout = 0.1
        self.hidden_act = "gelu"

    @classmethod
    def from_json_file(cls, file):

        config = BertConfig()
        with open(file, "r") as f:
            config_dict = json.load(f)

        for key in list(config_dict.keys()):
            config.__dict__[key] = config_dict[key]

        return config

    def __str__(self):
        return str(self.__dict__)


def expand_attention_mask(mask, dtype=torch.float32):
    mask = mask[:, None, None, :].to(dtype)
    mask = (1.0 - mask) * -10000.0
    return mask


class Embeddings(nn.Module):
    def __init__(self, config: BertConfig):
        super(Embeddings, self).__init__()

        self.word_embedding = nn.Embedding(config.vocab_size, config.hidden_size)
        self.position_embedding = nn.Embedding(config.max_position_embeddings, config.hidden_size)
        self.type_token_embedding = nn.Embedding(config.type_token_embeddings, config.hidden_size)

        self.norm = nn.LayerNorm(config.hidden_size)
        self.dropout = nn.Dropout(config.emb_drop)

        self.positions = torch.arange(config.max_position_embeddings, dtype=torch.int32)
        self.type_tokens = torch.zeros((1, config.max_position_embeddings), dtype=torch.int32)

    def forward(self, input_ids=None, position_ids=None, type_token_ids=None):

        baz, seq_len = input_ids.shape

        input_emb = self.word_embedding(input_ids)

        device = input_emb.device

        if type_token_ids is None:
            type_token_ids = self.type_tokens[:, :seq_len].expand(baz, -1).to(device)

        type_token_emb = self.type_token_embedding(type_token_ids)

        input_emb = input_emb + type_token_emb

        if position_ids is None:
            position_ids = self.positions[:seq_len].unsqueeze(0).to(device)

        position_emb = self.position_embedding(position_ids)

        embeddings = input_emb + position_emb

        embeddings = self.norm(embeddings)
        embeddings = self.dropout(embeddings)

        return embeddings


class SelfAttention(nn.Module):
    def __init__(self, config: BertConfig):
        super(SelfAttention, self).__init__()

        self.att_hidden = config.hidden_size // config.attention_heads
        self.num_heads = config.attention_heads

        self.q_linear = nn.Linear(config.hidden_size, self.att_hidden * self.num_heads)
        self.k_linear = nn.Linear(config.hidden_size, self.att_hidden * self.num_heads)
        self.v_linear = nn.Linear(config.hidden_size, self.att_hidden * self.num_heads)

        self.dropout = nn.Dropout(config.attention_drop)

    def trans_shape(self, x):
        return x.view(x.shape[0], -1, self.num_heads, self.att_hidden)

    def forward(self, hidden_state=None, attention_mask=None):

        q_value = self.q_linear(hidden_state) # b, s, h
        k_value = self.k_linear(hidden_state)
        v_value = self.v_linear(hidden_state)

        q_value = self.trans_shape(q_value) # b s h a
        k_value = self.trans_shape(k_value)
        v_value = self.trans_shape(v_value)

        q_value = q_value.transpose(1, 2)# b h s a
        k_value = k_value.transpose(1, 2).transpose(2, 3)# b h a s
        v_value = v_value.transpose(1, 2)

        attention_scores = torch.matmul(q_value, k_value) # b h s s
        attention_scores = attention_scores / math.sqrt(self.att_hidden)

        if attention_mask is not None:

            attention_scores = attention_scores + attention_mask

        attention_probs = torch.softmax(attention_scores, dim=-1)
        attention_probs = self.dropout(attention_probs)

        attention_out = torch.matmul(attention_probs, v_value) # b h s a

        attention_out = attention_out.transpose(1, 2).contiguous()
        attention_out = attention_out.flatten(2)

        return attention_out


class SelfOutput(nn.Module):
    def __init__(self, config: BertConfig):
        super(SelfOutput, self).__init__()

        self.dense = nn.Linear(config.hidden_size, config.hidden_size)
        self.norm = nn.LayerNorm(config.hidden_size)
        self.drop = nn.Dropout(config.dropout_rate)

    def forward(self, hidden_state, input_state):
        x = self.dense(hidden_state)
        x = self.drop(x)
        x = self.norm(x + input_state)
        return x


class Intermediate(nn.Module):
    def __init__(self, config: BertConfig):
        super(Intermediate, self).__init__()

        self.dense = nn.Linear(config.hidden_size, config.intermediate_size)
        self.ACT_fc = nn.GELU() if config.hidden_act == "gelu" else nn.ReLU()

    def forward(self, hidden_state):

        x = self.dense(hidden_state)
        x = self.ACT_fc(x)

        return x

class Output(nn.Module):
    def __init__(self, config: BertConfig):
        super(Output, self).__init__()

        self.dense = nn.Linear(config.intermediate_size, config.hidden_size)
        self.norm = nn.LayerNorm(config.hidden_size)
        self.drop = nn.Dropout(config.dropout_rate)

    def forward(self, hidden_state, input_state):
        x = self.dense(hidden_state)
        x = self.drop(x)
        x = self.norm(x + input_state)
        return x


class HeadTransform(nn.Module):
    def __init__(self, config: BertConfig):
        super(HeadTransform, self).__init__()

        self.dense = nn.Linear(config.hidden_size ,config.hidden_size)
        self.ACT_fc = nn.GELU() if config.hidden_act == "gelu" else nn.ReLU()
        self.norm = nn.LayerNorm(config.hidden_size)

    def forward(self, hidden_state):
        x = self.dense(hidden_state)
        x = self.ACT_fc(x)
        x = self.norm(x)
        return x

class Pooler(nn.Module):
    def __init__(self, config: BertConfig):
        super(Pooler).__init__()
        self.dense = nn.Linear(config.hidden_size, config.hidden_size)
        self.activation = nn.Tanh()

    def forward(self, hidden_states):

        first_token_tensor = hidden_states[:, 0]
        pooled_output = self.dense(first_token_tensor)
        pooled_output = self.activation(pooled_output)
        return pooled_output

class BertLayer(nn.Module):
    def __init__(self, config:BertConfig):
        super(BertLayer, self).__init__()

        self.self_attention = SelfAttention(config)
        self.self_out = SelfOutput(config)

        self.intermediate = Intermediate(config)
        self.output = Output(config)

    def forward(self, hidden_state=None, attention_mask=None):

        self_atc = self.self_attention(hidden_state, attention_mask=attention_mask)
        self_out = self.self_out(self_atc, hidden_state)

        x = self.intermediate(self_out)

        x = self.output(x, self_out)

        return x

class BertEncoder(nn.Module):
    def __init__(self, config: BertConfig):
        super(BertEncoder, self).__init__()

        self.layers = nn.ModuleList([BertLayer(config) for _ in range(config.layers)])

    def forward(self, hidden_state=None, attention_mask=None):

        for layer in self.layers:
            hidden_state = layer(hidden_state, attention_mask)

        return hidden_state


class BertClassificationHead(nn.Module):
    def __init__(self, config):
        super().__init__()
        self.dense = nn.Linear(config.hidden_size, config.hidden_size)
        classifier_dropout = (
            config.classifier_dropout if config.classifier_dropout is not None else config.dropout_rate
        )
        self.dropout = nn.Dropout(classifier_dropout)
        self.out_proj = nn.Linear(config.hidden_size, config.num_labels)

    def forward(self, features):
        x = features[:, 0, :]
        x = self.dropout(x)
        x = self.dense(x)
        x = torch.tanh(x)
        x = self.dropout(x)
        x = self.out_proj(x)
        return x

class Bert(nn.Module):
    def __init__(self, config: BertConfig):
        super(Bert, self).__init__()

        self.embeddings = Embeddings(config)

        self.encoder = BertEncoder(config)

        self.video_fc = nn.Conv1d(config.hidden_size, config.hidden_size, kernel_size=8, stride=8)
        self.audio_fc = nn.Conv1d(config.hidden_size, config.hidden_size, kernel_size=7, stride=4)

        self.fc_layers = BertClassificationHead(config)

    def forward(self, input_ids=None, attention_mask=None, position_ids=None, token_type_ids=None, video_values=None, audio_values=None):

        embeddings_out = self.embeddings(input_ids=input_ids, position_ids=position_ids, type_token_ids=token_type_ids)

        audio_values = audio_values.transpose(-1, -2)
        audio_values = self.audio_fc(audio_values)
        audio_values = audio_values.transpose(-1, -2)

        video_values = video_values.transpose(-1, -2)
        video_values = self.video_fc(video_values[:, :, 1:])
        video_values = video_values.transpose(-1, -2)

        # embeddings_out = audio_values # 模态融合
        # attention_mask = None #changed to audio/video only

        embeddings_out = torch.cat([embeddings_out, audio_values, video_values], dim=1)

        if attention_mask is not None:

            lens = embeddings_out.shape[1]
            batch_size, seq_len = attention_mask.shape
            mask = torch.ones((batch_size, lens), dtype=torch.int32, device=attention_mask.device)

            mask[:, :seq_len] = attention_mask

            attention_mask = expand_attention_mask(mask)

        output = self.encoder(embeddings_out, attention_mask=attention_mask)

        logits = self.fc_layers(output)

        return logits


config = BertConfig()

model = Bert(config)
state_dict = torch.load(
    r"C:\Users\Manish\Downloads\multitask_model lst.pth",
    map_location=torch.device('cpu'),
    weights_only=True,
)

state_dict = torch.load(r"C:\Users\Manish\Downloads\multitask_model lst.pth", map_location=torch.device('cpu'))


model.eval()


import os
import torch
from torch import nn
from torch.utils.data import DataLoader, Dataset
import numpy as np
from sklearn.model_selection import train_test_split
from transformers import BertTokenizer, BertModel, BertConfig
from glob import glob
from tqdm import tqdm
import torch.optim as optim

# Device configuration
device = "cuda" if torch.cuda.is_available() else "cpu"
batch_size = 16
lr = 1e-5
epochs = 10
config = BertConfig()
train_size = 0.8
tokenizer = BertTokenizer.from_pretrained("bert-base-uncased")
config.num_labels = 5
config.vocab_size = 30521

# Labels for sentiment and sarcasm
label_ls = ['Anger', 'Happiness', 'Excitement', 'Sadness', 'Neutral', 'Frustration', 'Surprise', 'Fear', 'Disgust']  # Sentiment labels
sarcasm_ls = ["Sarcastic", "Not Sarcastic"]  # Sarcasm labels

# Reading data function
def read_data():
    paths = glob(os.path.join('/content/drive/MyDrive/multi_model_sentiment/video_tensor', '*'))
    paths = paths[:400]
    train_paths, test_paths = train_test_split(paths, train_size=train_size)
    return train_paths, test_paths

# Custom dataset class
class CustomData(Dataset):
    def __init__(self, datas):
        self.datas = datas

    def __getitem__(self, item):
        video_path = self.datas[item]
        name = os.path.splitext(os.path.basename(video_path))[0]

        audio_path = os.path.join('/content/drive/MyDrive/multi_model_sentiment/audio_tensor', f"{name}.pkl")
        text_path = os.path.join('/content/drive/MyDrive/multi_model_sentiment/texts', f"{name}.txt")

        video_value = torch.load(video_path)
        audio_value = torch.load(audio_path)

        with open(text_path, encoding="utf-8") as f:
            texts = f.read()

        # Splitting text and labels
        text, sarcasm_label, sentiment_label = texts.split("\t")
        sentiment_label = label_ls.index(sentiment_label)
        sarcasm_label = 'Sarcastic' if sarcasm_label == 1 else "Not Sarcastic"
        sarcasm_label = sarcasm_ls.index(sarcasm_label)

        return text, sentiment_label, sarcasm_label, video_value, audio_value

    @staticmethod
    def call_fc(batch):
        texts = []
        sentiment_labels = []
        sarcasm_labels = []
        videos = []
        wavs = []

        for x, sentiment, sarcasm, video_value, wav_value in batch:
            texts.append(x)
            sentiment_labels.append(sentiment)
            sarcasm_labels.append(sarcasm)
            videos.append(video_value)
            wavs.append(wav_value)

        inputs = tokenizer.batch_encode_plus(texts, padding=True, truncation="only_first", max_length=1024, return_tensors="pt")

        sentiment_labels = np.array(sentiment_labels, dtype="int64")
        sarcasm_labels = np.array(sarcasm_labels, dtype="int64")

        sentiment_labels = torch.from_numpy(sentiment_labels)
        sarcasm_labels = torch.from_numpy(sarcasm_labels)

        inputs["video_values"] = torch.cat(videos, dim=0)
        inputs["audio_values"] = torch.cat(wavs, dim=0)

        return inputs, (sentiment_labels, sarcasm_labels)

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

# Model definition for multi-task learning
class Bert(nn.Module):
    def __init__(self, config):
        super(Bert, self).__init__()
        self.bert = BertModel.from_pretrained("bert-base-uncased", config=config, ignore_mismatched_sizes=True)
        self.sentiment_head = nn.Linear(config.hidden_size, len(label_ls))  # Sentiment output
        self.sarcasm_head = nn.Linear(config.hidden_size, 1)  # Sarcasm output (single value for binary classification)

    def forward(self, input_ids, attention_mask=None, token_type_ids=None, video_values=None, audio_values=None):
        outputs = self.bert(input_ids, attention_mask, token_type_ids)
        pooled_output = outputs[1]

        sentiment_output = self.sentiment_head(pooled_output)
        sarcasm_output = self.sarcasm_head(pooled_output)  # Only 1 output for sarcasm

        return sentiment_output, sarcasm_output


def preprocess_sample(video_path, audio_path, text_path):
    with open(text_path, encoding="utf-8") as f:
        text = f.read()

    # Get the video and audio tensors
    video_value = torch.load(video_path)
    audio_value = torch.load(audio_path)

    # Tokenize the text
    inputs = tokenizer.encode_plus(
        text,
        padding=True,
        truncation=True,
        max_length=1024,
        return_tensors="pt"
    )

    # Return the necessary data
    return inputs, video_value, audio_value

# Example paths (replace with actual paths)
video_path = r'C:/Users/Manish/Downloads/infrencing_model/Playing With Fire Movie Clip - Sarcasm (2019) _ Movieclips Coming Soon.mp4.pkl'
audio_path = 'C:/Users/Manish/Downloads/infrencing_model/audio.pkl'
text_path = r'C:/Users/Manish/Downloads/infrencing_model/Playing With Fire Movie Clip - Sarcasm (2019) _ Movieclips Coming Soon.txt'

inputs, video_value, audio_value = preprocess_sample(video_path, audio_path, text_path)

# Move tensors to the correct device (GPU/CPU)
inputs = {k: v.to(device) for k, v in inputs.items()}
video_value = video_value.to(device)
audio_value = audio_value.to(device)


# Make predictions
with torch.no_grad():
    sentiment_out, sarcasm_out = model(
        input_ids=inputs["input_ids"],
        attention_mask=inputs["attention_mask"],
        token_type_ids=inputs.get("token_type_ids"),
        video_values=video_value,
        audio_values=audio_value
    )

# Convert logits to predicted classes
sentiment_pred = torch.argmax(sentiment_out, dim=-1)
sarcasm_pred = torch.round(torch.sigmoid(sarcasm_out)).long()

# Map the predictions back to labels
sentiment_label = label_ls[sentiment_pred.item()]
sarcasm_label = sarcasm_ls[sarcasm_pred.item()]

print(f"Predicted Sentiment: {sentiment_label}")
print(f"Predicted Sarcasm: {sarcasm_label}")



PLease explain this