Delete prediction_sinhala.py

prediction_sinhala.py

@@ -1,231 +0,0 @@
-from typing import List, Optional, Tuple
-
-import torch
-from torch import Tensor
-from torch import nn
-from transformers import RobertaModel
-
-from faknow.model.layers.layer import TextCNNLayer
-from faknow.model.model import AbstractModel
-import pandas as pd
-
-
-class _MLP(nn.Module):
-    def __init__(self,
-                 input_dim: int,
-                 embed_dims: List[int],
-                 dropout_rate: float,
-                 output_layer=True):
-        super().__init__()
-        layers = list()
-        for embed_dim in embed_dims:
-            layers.append(nn.Linear(input_dim, embed_dim))
-            layers.append(nn.BatchNorm1d(embed_dim))
-            layers.append(nn.ReLU())
-            layers.append(nn.Dropout(p=dropout_rate))
-            input_dim = embed_dim
-        if output_layer:
-            layers.append(torch.nn.Linear(input_dim, 1))
-        self.mlp = torch.nn.Sequential(*layers)
-
-    def forward(self, x: Tensor) -> Tensor:
-        """
-
-        Args:
-            x (Tensor): shared feature from domain and text, shape=(batch_size, embed_dim)
-
-        """
-        return self.mlp(x)
-
-
-class _MaskAttentionLayer(torch.nn.Module):
-    """
-    Compute attention layer
-    """
-    def __init__(self, input_size: int):
-        super(_MaskAttentionLayer, self).__init__()
-        self.attention_layer = torch.nn.Linear(input_size, 1)
-
-    def forward(self,
-                inputs: Tensor,
-                mask: Optional[Tensor] = None) -> Tuple[Tensor, Tensor]:
-        weights = self.attention_layer(inputs).view(-1, inputs.size(1))
-        if mask is not None:
-            weights = weights.masked_fill(mask == 0, float("-inf"))
-        weights = torch.softmax(weights, dim=-1).unsqueeze(1)
-        outputs = torch.matmul(weights, inputs).squeeze(1)
-        return outputs, weights
-
-
-class MDFEND(AbstractModel):
-    r"""
-    MDFEND: Multi-domain Fake News Detection, CIKM 2021
-    paper: https://dl.acm.org/doi/10.1145/3459637.3482139
-    code: https://github.com/kennqiang/MDFEND-Weibo21
-    """
-    def __init__(self,
-                 pre_trained_bert_name: str,
-                 domain_num: int,
-                 mlp_dims: Optional[List[int]] = None,
-                 dropout_rate=0.2,
-                 expert_num=5):
-        """
-
-        Args:
-            pre_trained_bert_name (str): the name or local path of pre-trained bert model
-            domain_num (int): total number of all domains
-            mlp_dims (List[int]): a list of the dimensions in MLP layer, if None, [384] will be taken as default, default=384
-            dropout_rate (float): rate of Dropout layer, default=0.2
-            expert_num (int): number of experts also called TextCNNLayer, default=5
-        """
-        super(MDFEND, self).__init__()
-        self.domain_num = domain_num
-        self.expert_num = expert_num
-        self.bert = RobertaModel.from_pretrained(
-            pre_trained_bert_name).requires_grad_(False)
-        self.embedding_size = self.bert.config.hidden_size
-        self.loss_func = nn.BCELoss()
-        if mlp_dims is None:
-            mlp_dims = [384]
-
-        filter_num = 64
-        filter_sizes = [1, 2, 3, 5, 10]
-        experts = [
-            TextCNNLayer(self.embedding_size, filter_num, filter_sizes)
-            for _ in range(self.expert_num)
-        ]
-        self.experts = nn.ModuleList(experts)
-
-        self.gate = nn.Sequential(
-            nn.Linear(self.embedding_size * 2, mlp_dims[-1]), nn.ReLU(),
-            nn.Linear(mlp_dims[-1], self.expert_num), nn.Softmax(dim=1))
-
-        self.attention = _MaskAttentionLayer(self.embedding_size)
-
-        self.domain_embedder = nn.Embedding(num_embeddings=self.domain_num,
-                                            embedding_dim=self.embedding_size)
-        self.classifier = _MLP(320, mlp_dims, dropout_rate)
-
-    def forward(self, token_id: Tensor, mask: Tensor,
-                domain: Tensor) -> Tensor:
-        """
-
-        Args:
-            token_id (Tensor): token ids from bert tokenizer, shape=(batch_size, max_len)
-            mask (Tensor): mask from bert tokenizer, shape=(batch_size, max_len)
-            domain (Tensor): domain id, shape=(batch_size,)
-
-        Returns:
-            FloatTensor: the prediction of being fake, shape=(batch_size,)
-        """
-        text_embedding = self.bert(token_id,
-                                   attention_mask=mask).last_hidden_state
-        attention_feature, _ = self.attention(text_embedding, mask)
-
-        domain_embedding = self.domain_embedder(domain.view(-1, 1)).squeeze(1)
-
-        gate_input = torch.cat([domain_embedding, attention_feature], dim=-1)
-        gate_output = self.gate(gate_input)
-
-        shared_feature = 0
-        for i in range(self.expert_num):
-            expert_feature = self.experts[i](text_embedding)
-            shared_feature += (expert_feature * gate_output[:, i].unsqueeze(1))
-
-        label_pred = self.classifier(shared_feature)
-
-        return torch.sigmoid(label_pred.squeeze(1))
-
-    def calculate_loss(self, data) -> Tensor:
-        """
-        calculate loss via BCELoss
-
-        Args:
-            data (dict): batch data dict
-
-        Returns:
-            loss (Tensor): loss value
-        """
-
-        token_ids = data['text']['token_id']
-        masks = data['text']['mask']
-        domains = data['domain']
-        labels = data['label']
-        output = self.forward(token_ids, masks, domains)
-        return self.loss_func(output, labels.float())
-
-    def predict(self, data_without_label) -> Tensor:
-        """
-        predict the probability of being fake news
-
-        Args:
-            data_without_label (Dict[str, Any]): batch data dict
-
-        Returns:
-            Tensor: one-hot probability, shape=(batch_size, 2)
-        """
-
-        token_ids = data_without_label['text']['token_id']
-        masks = data_without_label['text']['mask']
-        domains = data_without_label['domain']
-
-
-        output_prob = self.forward(token_ids, masks,domains)
-
-        return output_prob
-from faknow.data.dataset.text import TextDataset
-from faknow.data.process.text_process import TokenizerFromPreTrained
-from faknow.evaluate.evaluator import Evaluator
-
-import torch
-from torch.utils.data import DataLoader
-testing_path = "/content/drive/MyDrive/sinhala-dataset/test_data.json"
-
-df = pd.read_json(testing_path)
-df.head()
-df =df[:100]
-df["label"] = int(0)
-df.head()
-print(len(df))
-path = '/content/drive/MyDrive/sinhala-dataset'
-testing_json = "/testing.json"
-df.to_json(path + testing_json, orient='records')
-
-MODEL_SAVE_PATH = "/content/drive/MyDrive/models-path-improvement/last-epoch-model-2024-03-08-15_34_03_6.pth"
-
-max_len, bert = 160 , 'sinhala-nlp/sinbert-sold-si'
-tokenizer = TokenizerFromPreTrained(max_len, bert)
-
-# dataset
-batch_size = 100
-
-
-testing_path = path + testing_json
-
-testing_set = TextDataset(testing_path, ['text'], tokenizer)
-testing_loader = DataLoader(testing_set, batch_size, shuffle=False)
-
-# prepare model
-domain_num = 3
-
-model = MDFEND(bert, domain_num , expert_num=18 , mlp_dims = [5080 ,4020, 3010, 2024 ,1012 ,606 , 400])
-model.load_state_dict(torch.load(f=MODEL_SAVE_PATH, map_location=torch.device('cpu')))
-
-
-
-outputs = []
-for batch_data in testing_loader:
-      outputs.append(model.predict(batch_data))
-outputs
-# 1 ====> offensive
-# 0 ====> not offensive
-label = []
-for output in outputs:
-  for out in output:
-    output_prob = out.item()
-    if output_prob >= 0.5:
-      label.append(1)
-    else:
-      label.append(0)
-
-label