style: format code with Autopep8, isort and Ruff Formatter

src/everyai/classifier/classify.py

@@ -50,23 +50,19 @@ def split_data(
     """
     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(
+        x_test,
+        y_test,
+        indices_test,
+        test_size=test_size / (test_size + valid_size),
+        random_state=42,
     )
     return (
         x_train,
@@ -157,8 +153,7 @@ def __init__(
             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"
         )
         if self.split_size is not None:
             self.train_size = self.split_size.get("train_size", 0.8)
@@ -175,9 +170,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}"

src/everyai/classifier/fusion_classifer.py

@@ -6,11 +6,8 @@
 
 from everyai.classifier.classify import TextClassifer, label_encode, split_data
 from everyai.classifier.multi_feature_model.fusionBert import (
-    CrossAttentionFeatureFusion,
-    FeatureFusionBertClassfier,
-    FeatureFusionDataModule,
-    HFeatureFusion,
-)
+    CrossAttentionFeatureFusion, FeatureFusionBertClassfier,
+    FeatureFusionDataModule, HFeatureFusion)
 
 
 class PLClassifer(TextClassifer):

src/everyai/classifier/multi_feature_model/fusionBert.py

@@ -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(
@@ -58,9 +54,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)
@@ -75,12 +71,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):
@@ -112,9 +104,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
@@ -149,9 +139,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
@@ -220,9 +208,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])
@@ -241,9 +227,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}
@@ -255,9 +239,7 @@ def test_epoch_end(self, outputs):
         self.log("avg_test_f1", avg_f1, prog_bar=True)
 
     def configure_optimizers(self):
-        optimizer = torch.optim.Adam(
-            self.parameters(), lr=self.lr, momentum=0.9
-        )
+        optimizer = torch.optim.Adam(self.parameters(), lr=self.lr, momentum=0.9)
         scheduler = get_scheduler(
             name="linear",
             optimizer=optimizer,
@@ -318,8 +300,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(

src/everyai/classifier/multi_feature_model/fusionBert_backup.py

@@ -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(
@@ -58,9 +54,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)
@@ -75,12 +71,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):
@@ -111,9 +103,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
@@ -151,9 +141,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
@@ -255,19 +243,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()}"
+                )

src/everyai/classifier/pytorch_classier.py

@@ -28,13 +28,13 @@ def __init__(
             language=language,
             **classfiy_config,
         )
-        tokenzier_dict  = {
-            "fusion_bert": FeatureFusionBertTokenizer(AutoTokenizer("bert-base-uncased"))
+        tokenzier_dict = {
+            "fusion_bert": FeatureFusionBertTokenizer(
+                AutoTokenizer("bert-base-uncased")
+            )
         }
         self.tokenizer = tokenzier_dict[self.tokenizer_name]
-        model_dict = {
-            "fusion_bert": FeatureFusionBertClassfier(feature_num=3)
-        }
+        model_dict = {"fusion_bert": FeatureFusionBertClassfier(feature_num=3)}
         self.model = model_dict[self.model_name]
         self.label_encoder = None
         self.train_dataset, self.valid_dataset, self.test_dataset = (
@@ -47,9 +47,7 @@ def __init__(
     def _tokenize(self, texts: list[str], labels: list[str]):
         self.label_encoder, tokenzied_labels = label_encode(labels)
         tokenzied_labels = torch.tensor(tokenzied_labels)
-        dataset = datasets.Dataset.from_dict(
-            {"text": texts, "label": tokenzied_labels}
-        )
+        dataset = datasets.Dataset.from_dict({"text": texts, "label": tokenzied_labels})
 
         def _tokenizer_fn(example):
             return self.tokenizer(example["text"], **self.tokenizer_config)
@@ -73,12 +71,8 @@ def train(self):
             self.data.valid_indices,
             self.data.test_indices,
         ) = split_data(self.texts, self.labels)
-        self.train_dataset = self._tokenize(
-            self.data.x_train, self.data.y_train
-        )
-        self.valid_dataset = self._tokenize(
-            self.data.x_valid, self.data.y_valid
-        )
+        self.train_dataset = self._tokenize(self.data.x_train, self.data.y_train)
+        self.valid_dataset = self._tokenize(self.data.x_valid, self.data.y_valid)
         self.test_dataset = self._tokenize(self.data.x_test, self.data.y_test)
         train_args = TrainingArguments(**self.train_args)
         data_collator = DataCollatorWithPadding(tokenizer=self.tokenizer)
@@ -94,15 +88,11 @@ def train(self):
     def test(self):
         trainer = Trainer(model=self.model)
         predictions = trainer.predict(self.test_dataset)
-        self.data.y_pred = torch.argmax(
-            torch.tensor(predictions.predictions), axis=1
-        )
+        self.data.y_pred = torch.argmax(torch.tensor(predictions.predictions), axis=1)
 
         self.data.y_test = self.label_encoder.transform(self.data.y_test)
 
     def show_score(self):
         metric = evaluate.load("accuracy")
-        metric.compute(
-            predictions=self.data.y_pred, references=self.data.y_test
-        )
+        metric.compute(predictions=self.data.y_pred, references=self.data.y_test)
         logging.info("Accuracy: %s", metric)

src/everyai/data_loader/data_load.py

@@ -65,9 +65,7 @@ def load_data(
         if max_count is not None and loaded_data is not None:
             loaded_data = loaded_data.head(max_count)
         else:
-            logging.info(
-                "Max count is None and all the records will be loaded"
-            )
+            logging.info("Max count is None and all the records will be loaded")
         loaded_data.rename(
             columns={
                 self.question_column: "question",
@@ -78,9 +76,7 @@ def load_data(
         if return_type == "pandas":
             result = loaded_data
         elif return_type == "list":
-            result = loaded_data[["question", "answer"]].to_dict(
-                orient="records"
-            )
+            result = loaded_data[["question", "answer"]].to_dict(orient="records")
         else:
             logging.error("Invalid return type")
         return result
@@ -90,4 +86,4 @@ def apply_filter(self, orginal_data: pd.DataFrame) -> pd.DataFrame:
             orginal_data
             if self.data_filter is None
             else orginal_data[orginal_data.apply(self.data_filter, axis=1)]
-        )
+        )