Skip to content

style: format code with Autopep8, isort and Ruff Formatter #25

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
deepsource-autofix wants to merge 1 commit into
base: main
Choose a base branch
from
Open

style: format code with Autopep8, isort and Ruff Formatter #25

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
73 changes: 40 additions & 33 deletions src/ai_detection/classifier/classify.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,35 +44,42 @@ def label_encode(labels: list[str]):

def split_data(
x: list, y: list, train_size=0.8, valid_size=0.1, test_size=0.1
) ->tuple[list,list,list,list,list,list,list,list,list]: # -> tuple:
) -> tuple[list, list, list, list, list, list, list, list, list]: # -> tuple:
"""
split_data is a function to split the data into train, valid and test sets
tips: split_data is design for getting the index for train, valid and test set
"""
assert len(x) == len(y), "Length of x and y should be same"
original_indices = pd.DataFrame(x).index
x_train, x_test, y_train, y_test, indices_train, indices_test = (
train_test_split(
x,
y,
original_indices,
train_size=train_size,
random_state=42,
)
x_train, x_test, y_train, y_test, indices_train, indices_test = train_test_split(
x,
y,
original_indices,
train_size=train_size,
random_state=42,
)
x_test, x_valid, y_test, y_valid, indices_train, indices_valid = train_test_split(
x_test,
y_test,
indices_test,
test_size=test_size / (test_size + valid_size),
random_state=42,
)
x_test, x_valid, y_test, y_valid, indices_train, indices_valid = (
train_test_split(
if not all(
isinstance(i, list)
for i in [
x_train,
x_valid,
x_test,
y_train,
y_valid,
y_test,
indices_train,
indices_valid,
indices_test,
test_size=test_size / (test_size + valid_size),
random_state=42,
)
)
if not all(isinstance(i, list) for i in [x_train, x_valid, x_test, y_train, y_valid, y_test, indices_train, indices_valid, indices_test]):
raise ValueError(
"x_train, x_valid, x_test should be list or np.array"
)
]
):
raise ValueError("x_train, x_valid, x_test should be list or np.array")
return (
x_train,
x_valid,
Expand All @@ -93,18 +100,18 @@ class TextClassifer:

def __init__(
self,
texts: list[str]|None = None,
labels: list[str]|None = None,
texts: list[str] | None = None,
labels: list[str] | None = None,
data_name: str = "",
language: str = "English",
model_name: str = "bert-base-uncased",
tokenizer_name: str = "bert-base-uncased",
classifier_type: str = "transformers",
split_size: dict|None = None,
train_args: dict|None = None,
tokenizer_config: dict|None = None,
model_config: dict|None = None,
pipeline: list[str]|None = None,
split_size: dict | None = None,
train_args: dict | None = None,
tokenizer_config: dict | None = None,
model_config: dict | None = None,
pipeline: list[str] | None = None,
):
"""
Allowed and default keys for classify_config are:
Expand Down Expand Up @@ -161,16 +168,18 @@ def __init__(
self.tokenizer_config = tokenizer_config if tokenizer_config is not None else {}
self.model_config = model_config if model_config is not None else {}
self.pipeline = pipeline if pipeline is not None else []
self.split_size = split_size if split_size is not None else {"train_size": 0.8, "test_size": 0.1, "valid_size": 0.1}
self.split_size = (
split_size
if split_size is not None
else {"train_size": 0.8, "test_size": 0.1, "valid_size": 0.1}
)
self.classifier_name = (
f"{self.model_name}_{self.tokenizer_name}_{self.data_name}"
)
self.model_path = (
MODEL_PATH
/ f"{self.model_name}_{self.tokenizer_name}_{self.data_name}.pkl"
MODEL_PATH / f"{self.model_name}_{self.tokenizer_name}_{self.data_name}.pkl"
)


def load_data(self, texts, labels, data_name):
"""
load_data is a function to load the data into the classifier
Expand All @@ -181,9 +190,7 @@ def load_data(self, texts, labels, data_name):
raise ValueError("Length of texts and labels should be same")
self.texts = texts
self.labels = labels
logging.info(
"Loading data: %s to classifier %s", data_name, self.model_name
)
logging.info("Loading data: %s to classifier %s", data_name, self.model_name)
self.data_name = data_name
self.classifier_name = (
f"{self.model_name}_{self.tokenizer_name}_{self.data_name}"
Expand Down
50 changes: 18 additions & 32 deletions src/ai_detection/classifier/multi_feature_model/fusionBert.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,12 +10,14 @@
from sklearn.metrics import f1_score
from torch.utils.data import DataLoader, Dataset
from tqdm import tqdm
from transformers import BertForSequenceClassification, BertTokenizer, PreTrainedTokenizerBase
from transformers import (
BertForSequenceClassification,
BertTokenizer,
PreTrainedTokenizerBase,
)
from transformers.optimization import get_scheduler
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer

import spacy


def truncate_and_pad_single_sequence(seq, max_length):
truncated_seq = seq[:max_length]
Expand All @@ -26,18 +28,14 @@ def truncate_and_pad_single_sequence(seq, max_length):
class FeatureFusionBertTokenizer(PreTrainedTokenizerBase):
def __init__(self, feature_len, **kwargs):
self.feature_len = feature_len
self.sentiment_max_length = kwargs.get(
"sentiment_max_length", feature_len
)
self.sentiment_max_length = kwargs.get("sentiment_max_length", feature_len)
self.nlp = spacy.load("en_core_web_sm")
self.all_tags = self.nlp.get_pipe("tagger").labels
self.sentiment_analyzer = SentimentIntensityAnalyzer()
self.bert_model = BertForSequenceClassification.from_pretrained(
"bert-base-uncased"
)
self.bert_tokenzier = BertTokenizer.from_pretrained(
"bert-base-uncased"
)
self.bert_tokenzier = BertTokenizer.from_pretrained("bert-base-uncased")

def semantic(self, text: str, **kwargs):
tokenzied = self.bert_tokenzier(
Expand All @@ -60,9 +58,9 @@ def analyze_word_level_sentiment(self, text: str) -> torch.tensor:
if token.is_stop or token.is_punct:
sentiment = 0.0
else:
sentiment = self.sentiment_analyzer.polarity_scores(
token.text
)["compound"]
sentiment = self.sentiment_analyzer.polarity_scores(token.text)[
"compound"
]
word_sentiment.append(sentiment)
sentiment = torch.tensor(word_sentiment, dtype=torch.float)
return self._padding(sentiment)
Expand All @@ -77,12 +75,8 @@ def pos_feature(self, text: str) -> torch.tensor:
return self._padding(pos)

def _padding(self, input_tensor):
input_tensor = truncate_and_pad_single_sequence(
input_tensor, self.feature_len
)
attention_mask = torch.tensor(
[1] * len(input_tensor), dtype=torch.float
)
input_tensor = truncate_and_pad_single_sequence(input_tensor, self.feature_len)
attention_mask = torch.tensor([1] * len(input_tensor), dtype=torch.float)
return input_tensor.unsqueeze(0), attention_mask.unsqueeze(0)

def __call__(self, text: str):
Expand Down Expand Up @@ -114,9 +108,7 @@ def batch_encode_plus(self, batch_text: list[str]): # -> list:


class HFeatureFusion(nn.Module):
def __init__(
self, feature_num, feature_len, output_dim, num_heads=4, dropout=0.1
):
def __init__(self, feature_num, feature_len, output_dim, num_heads=4, dropout=0.1):
super().__init__()
self.feature_num = feature_num
self.feature_len = feature_len
Expand Down Expand Up @@ -151,9 +143,7 @@ def forward(self, *features: torch.tensor):


class CrossAttentionFeatureFusion(nn.Module):
def __init__(
self, feature_num, feature_len, output_dim, num_heads=4, dropout=0.1
):
def __init__(self, feature_num, feature_len, output_dim, num_heads=4, dropout=0.1):
super().__init__()
self.feature_len = feature_len
self.num_heads = num_heads
Expand Down Expand Up @@ -222,9 +212,7 @@ def __init__(
def forward(self, *features):
self.train() # Set all modules to train
for feature in features:
feature["input_ids"] = F.normalize(
feature["input_ids"], p=2, dim=-1
)
feature["input_ids"] = F.normalize(feature["input_ids"], p=2, dim=-1)
fused_features = self.fusion_module(*features)
encoder_output = self.encoder(fused_features)
pooler_output = self.pooler(encoder_output[0])
Expand All @@ -243,9 +231,7 @@ def test_step(self, batch, batch_idx):
y_pred = torch.argmax(outputs, dim=-1)
y_test = batch["labels"]
acc = (y_pred == y_test).sum().float() / len(y_test)
f1 = f1_score(
y_test.cpu().numpy(), y_pred.cpu().numpy(), average="weighted"
)
f1 = f1_score(y_test.cpu().numpy(), y_pred.cpu().numpy(), average="weighted")
self.log("test_acc", acc)
self.log("test_f1", f1)
return {"test_acc": acc, "test_f1": f1}
Expand Down Expand Up @@ -321,8 +307,8 @@ def prepare_data(self):
pass

def setup(self, stage=None):
self.tokenizer: FeatureFusionBertTokenizer = (
FeatureFusionBertTokenizer(feature_len=768)
self.tokenizer: FeatureFusionBertTokenizer = FeatureFusionBertTokenizer(
feature_len=768
)
if not isinstance(self.tokenizer, FeatureFusionBertTokenizer):
raise TypeError(
Expand Down
53 changes: 19 additions & 34 deletions src/ai_detection/classifier/multi_feature_model/fusionBert_backup.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
import torch
import torch.nn as nn
import torch.nn.functional as F
from datasets import load_dataset,Dataset
from datasets import Dataset, load_dataset
from torch.optim import AdamW
from torch.utils.data import DataLoader
from tqdm import tqdm
Expand All @@ -24,18 +24,14 @@ def truncate_and_pad_single_sequence(seq, max_length):
class FeatureFusionBertTokenizer:
def __init__(self, feature_len, **kwargs):
self.feature_len = feature_len
self.sentiment_max_length = kwargs.get(
"sentiment_max_length", feature_len
)
self.sentiment_max_length = kwargs.get("sentiment_max_length", feature_len)
self.nlp = spacy.load("en_core_web_sm")
self.all_tags = self.nlp.get_pipe("tagger").labels
self.sentiment_analyzer = SentimentIntensityAnalyzer()
self.bert_model = BertForSequenceClassification.from_pretrained(
"bert-base-uncased"
)
self.bert_tokenzier = BertTokenizer.from_pretrained(
"bert-base-uncased"
)
self.bert_tokenzier = BertTokenizer.from_pretrained("bert-base-uncased")

def semantic(self, text: str, **kwargs):
tokenzied = self.bert_tokenzier(
Expand All @@ -51,16 +47,16 @@ def semantic(self, text: str, **kwargs):
embedding = torch.mean(embedding, dim=1, keepdim=False)
return embedding, torch.ones_like(embedding)

def analyze_word_level_sentiment(self, text: str) ->tuple:
def analyze_word_level_sentiment(self, text: str) -> tuple:
doc = self.nlp(text)
word_sentiment = []
for token in doc:
if token.is_stop or token.is_punct:
sentiment = 0.0
else:
sentiment = self.sentiment_analyzer.polarity_scores(
token.text
)["compound"]
sentiment = self.sentiment_analyzer.polarity_scores(token.text)[
"compound"
]
word_sentiment.append(sentiment)
sentiment = torch.tensor(word_sentiment, dtype=torch.float)
return self._padding(sentiment)
Expand All @@ -75,14 +71,9 @@ def pos_feature(self, text: str) -> tuple:
posed = self._padding(pos)
return posed if isinstance(posed, tuple) else posed


def _padding(self, input_tensor):
input_tensor = truncate_and_pad_single_sequence(
input_tensor, self.feature_len
)
attention_mask = torch.tensor(
[1] * len(input_tensor), dtype=torch.float
)
input_tensor = truncate_and_pad_single_sequence(input_tensor, self.feature_len)
attention_mask = torch.tensor([1] * len(input_tensor), dtype=torch.float)
return input_tensor.unsqueeze(0), attention_mask.unsqueeze(0)

def __call__(self, text: str):
Expand Down Expand Up @@ -113,9 +104,7 @@ def batch_encode_plus(self, batch_text: list[str], **kwargs):


class HFeatureFusion(nn.Module):
def __init__(
self, feature_num, feature_len, output_dim, num_heads=4, dropout=0.1
):
def __init__(self, feature_num, feature_len, output_dim, num_heads=4, dropout=0.1):
super().__init__()
self.feature_num = feature_num
self.feature_len = feature_len
Expand All @@ -129,7 +118,7 @@ def __init__(
self.feature_fusion_dim = (feature_num - 1) * feature_len
self.fc = nn.Linear(self.feature_fusion_dim, output_dim)

def forward(self, features:list[dict[str, torch.Tensor]]):
def forward(self, features: list[dict[str, torch.Tensor]]):
if len(features) != self.feature_num:
raise ValueError(
f"Expected {self.feature_num} features, but got {len(features)}"
Expand Down Expand Up @@ -157,9 +146,7 @@ def forward(self, features:list[dict[str, torch.Tensor]]):


class CrossAttentionFeatureFusion(nn.Module):
def __init__(
self, feature_num, feature_len, output_dim, num_heads=4, dropout=0.1
):
def __init__(self, feature_num, feature_len, output_dim, num_heads=4, dropout=0.1):
super().__init__()
self.feature_len = feature_len
self.num_heads = num_heads
Expand Down Expand Up @@ -209,7 +196,7 @@ def __init__(
)
self.encoder = bert_classifier.bert.encoder
self.pooler = bert_classifier.bert.pooler

self.fusion_module = CrossAttentionFeatureFusion(
feature_num=feature_num,
feature_len=feature_len,
Expand Down Expand Up @@ -266,19 +253,17 @@ def __len__(self):
for epoch in range(num_epochs):
for step, batch in enumerate(train_dataloader):
for feature in ["semantic", "pos", "sentiment"]:
batch[feature]["input_ids"] = batch[feature]["input_ids"].to(
batch[feature]["input_ids"] = batch[feature]["input_ids"].to(device)
batch[feature]["attention_mask"] = batch[feature]["attention_mask"].to(
device
)
batch[feature]["attention_mask"] = batch[feature][
"attention_mask"
].to(device)
optimizer.zero_grad()
outputs = model(
batch["semantic"], batch["pos"], batch["sentiment"]
)
outputs = model(batch["semantic"], batch["pos"], batch["sentiment"])
loss = F.cross_entropy(outputs, batch["labels"].to(device))
loss.backward()
optimizer.step()
lr_scheduler.step()
if step % 100 == 0:
print(f"Epoch {epoch + 1}, step {step + 1} completed with loss: {loss.item()}")
print(
f"Epoch {epoch + 1}, step {step + 1} completed with loss: {loss.item()}"
)
Loading