Updated data.py

src/pipes/const.py

@@ -1,3 +1,3 @@
 data_dir: str = "E:/bn_multi_tribe_mt/data/"
-example_count: int = 1270
-langs: list[str] = ['bn', 'en', 'gr']
+langs: list[str] = ['bn', 'en', 'gr']
+MAX_SEQ_LEN = 30

src/pipes/data.py

@@ -1,78 +1,137 @@
 import random
-from pipes import utils, const
+import const
+import utils
+import string
 
-
-class Sentence:
+class SequenceLoader:
     def __init__(self):
-        self.sentence_dict = None
-        self.shuffled_sentences = None
+        self.sequence_dict = None
+        self.shuffled_sequences = None
         self.shuffled_indices = None
-        self.sentences = None
+        self.sequences = None
         self.max_seq_length = None
         self.vocab = None
+        self.lang = None
 
-    def pack(self, lang):
-        self.sentences = utils.read_file("{}/raw/{}.txt".format(const.data_dir, lang))
+    def pack(self):
+        self.sequences = utils.read_file("{}/raw/{}.txt".format(const.data_dir, self.lang))
 
-        example_count = len(self.sentences)
-        split_index = int(example_count * 0.80)
+        examples_count = len(self.sequences)
+        split_index = int(examples_count * 0.80)
 
         if self.shuffled_indices is None:
-            self.shuffled_indices = list(range(example_count))
+            self.shuffled_indices = list(range(examples_count))
             random.shuffle(self.shuffled_indices)
 
-        self.shuffled_sentences = [self.sentences[i] for i in self.shuffled_indices]
-
-        for i in range(example_count):
-            self.shuffled_sentences[i] = utils.remove_punctuation(self.shuffled_sentences[i])
-            self.sentences[i] = utils.add_start_end_tags(self.sentences[i])
-
-        self.vocab = utils.build_vocab(self.shuffled_sentences)
-        self.max_seq_length = max(len(sentence.split()) for sentence in self.shuffled_sentences)
-        self.sentence_dict = dict(
-            max_seq_len=self.max_seq_length,
-            vocab_size=len(self.vocab),
-            vocab=self.vocab,
-            train=self.shuffled_sentences[:split_index],
-            val=self.shuffled_sentences[split_index:],
-            count=example_count
+        self.shuffled_sequences = [self.sequences[i] for i in self.shuffled_indices]
+
+        self.sequence_dict = dict(
+            train=self.shuffled_sequences[:split_index],
+            val=self.shuffled_sequences[split_index:],
+            count=examples_count,
         )
 
+    def get_dict(self):
+        return self.sequence_dict
+
+    def set_lang(self, lang):
+        self.lang = lang
+
+
+def remove_punctuation_from_seq(seq):
+    english_punctuations = string.punctuation
+    bangla_punctuations = "৷-–—’‘৳…।"
+    all_punctuations = english_punctuations + bangla_punctuations
+    cleaned_seq = ''.join([char for char in seq if char not in all_punctuations])
+    cleaned_seq = cleaned_seq.strip()
+    cleaned_seq = ' '.join(cleaned_seq.split())
+    return cleaned_seq
+
+
+def add_start_end_tags_seq(sequence):
+    return '<SOS> ' + sequence + ' <EOS>'
+
+
+def pad_sequence(sequence, max_seq_len, padding_token=0):
+    padded_sequence = sequence[:max_seq_len] + [padding_token] * (max_seq_len - len(sequence))
+    return padded_sequence
+
+
+class SequenceProcessor:
+    def __init__(self, _dataset_dict):
+        self.max_seq_len = 0
+        self.lang = None
+        self.dataset_dict = _dataset_dict
+        self.vocab = None
+
+    def remove_punctuation(self):
+        for i in range(len(self.dataset_dict[self.lang]["train"])):
+            self.dataset_dict[self.lang]["train"][i] = remove_punctuation_from_seq(
+                self.dataset_dict[self.lang]["train"][i])
+
+        for i in range(len(self.dataset_dict[self.lang]["val"])):
+            self.dataset_dict[self.lang]["val"][i] = remove_punctuation_from_seq(
+                self.dataset_dict[self.lang]["val"][i])
+
+    def build_vocab(self):
+        vocab = set()
+
+        for i in range(len(self.dataset_dict[self.lang]["train"])):
+            seq = self.dataset_dict[self.lang]["train"][i]
+            vocab.update(seq.split())
+
+        for i in range(len(self.dataset_dict[self.lang]["val"])):
+            seq = self.dataset_dict[self.lang]["val"][i]
+            vocab.update(seq.split())
+
+        self.vocab = sorted(list(vocab))
+        self.dataset_dict[self.lang]["vocab"] = self.vocab
+        self.dataset_dict[self.lang]["vocab_size"] = len(self.vocab)
+
+    def add_start_end_tags(self):
+        for i in range(len(self.dataset_dict[self.lang]["train"])):
+            self.dataset_dict[self.lang]["train"][i] = add_start_end_tags_seq(
+                self.dataset_dict[self.lang]["train"][i])
+            self.max_seq_len = max(len(self.dataset_dict[self.lang]["train"][i].split()), self.max_seq_len)
+
+        for i in range(len(self.dataset_dict[self.lang]["val"])):
+            self.dataset_dict[self.lang]["val"][i] = add_start_end_tags_seq(
+                self.dataset_dict[self.lang]["val"][i])
+            self.max_seq_len = max(len(self.dataset_dict[self.lang]["val"][i].split()), self.max_seq_len)
+
+        self.dataset_dict[self.lang]["max_seq_len"] = self.max_seq_len
+
     def tokenize(self):
-        tokenized_train_sentences = []
-        for sentence in self.sentence_dict["train"]:
+        for i in range(len(self.dataset_dict[self.lang]["train"])):
+            seq = self.dataset_dict[self.lang]["train"][i]
             tokens = []
-            for word in sentence.split():
+            for word in seq.split():
                 tokens.append(self.vocab.index(word))
-            tokenized_train_sentences.append(tokens)
+            self.dataset_dict[self.lang]["train"][i] = tokens
 
-        tokenized_val_sentences = []
-        for sentence in self.sentence_dict["train"]:
+        for i in range(len(self.dataset_dict[self.lang]["val"])):
+            seq = self.dataset_dict[self.lang]["val"][i]
             tokens = []
-            for word in sentence.split():
+            for word in seq.split():
                 tokens.append(self.vocab.index(word))
-            tokenized_val_sentences.append(tokens)
-
-        self.sentence_dict["train"] = tokenized_train_sentences
-        self.sentence_dict["val"] = tokenized_val_sentences
+            self.dataset_dict[self.lang]["val"][i] = tokens
 
-    def pad(self, max_seq_len=None):
-        if max_seq_len is None:
-            max_seq_len = self.sentence_dict["max_seq_len"]
+    def pad(self, max_seq_len=const.MAX_SEQ_LEN):
 
-        padded_train_sentences = []
-        for sentence in self.sentence_dict["train"]:
-            padded_train_sentences.append(utils.pad_sequence(sentence, max_seq_len))
+        for i in range(len(self.dataset_dict[self.lang]["train"])):
+            self.dataset_dict[self.lang]["train"][i] = pad_sequence(
+                sequence=self.dataset_dict[self.lang]["train"][i], max_seq_len=max_seq_len)
 
-        padded_val_sentences = []
-        for sentence in self.sentence_dict["val"]:
-            padded_val_sentences.append(utils.pad_sequence(sentence, max_seq_len))
+        for i in range(len(self.dataset_dict[self.lang]["val"])):
+            self.dataset_dict[self.lang]["val"][i] = pad_sequence(sequence=self.dataset_dict[self.lang]["val"][i],
+                                                                  max_seq_len=self.max_seq_len)
 
-        self.sentence_dict["train"] = padded_train_sentences
-        self.sentence_dict["val"] = padded_val_sentences
+    def set_lang(self, lang):
+        self.lang = lang
+        self.max_seq_len = 0
 
     def get_dict(self):
-        return self.sentence_dict
+        return self.dataset_dict
 
 
 class Dataset:
@@ -81,18 +140,22 @@ class Dataset:
         self.dataset_dict = {}
 
     def pack(self):
-        sentence_object = Sentence()
+        seq_loader = SequenceLoader()
         for lang in self.langs:
-            sentence_object.pack(lang)
-            self.dataset_dict[lang] = sentence_object.get_dict()
+            seq_loader.set_lang(lang)
+            seq_loader.pack()
+            self.dataset_dict[lang] = seq_loader.get_dict()
 
     def process(self):
-        sentence_object = Sentence()
+        seq_processor = SequenceProcessor(self.dataset_dict)
         for lang in self.langs:
-            sentence_object.pack(lang)
-            sentence_object.tokenize()
-            sentence_object.pad()
-            self.dataset_dict[lang] = sentence_object.get_dict()
+            seq_processor.set_lang(lang)
+            seq_processor.remove_punctuation()
+            seq_processor.add_start_end_tags()
+            seq_processor.build_vocab()
+            seq_processor.tokenize()
+            seq_processor.pad()
+        self.dataset_dict = seq_processor.get_dict()
 
     def get_dict(self):
         return self.dataset_dict
@@ -101,6 +164,7 @@ class Dataset:
 if __name__ == "__main__":
     dataset_object = Dataset(const.langs)
     dataset_object.pack()
-    dataset_object.process()
     dataset_dict = dataset_object.get_dict()
     utils.save_dict("{}/dataset.txt".format(const.data_dir), dataset_dict)
+    dataset_object.process()
+    print(utils.load_dict("{}/dataset.txt".format(const.data_dir)))

src/pipes/utils.py

@@ -2,15 +2,6 @@ import json
 import string
 
 
-def pad_sequence(sequence, max_length, padding_token=0):
-    padded_sequence = sequence[:max_length] + [padding_token] * (max_length - len(sequence))
-    return padded_sequence
-
-
-def add_start_end_tags(sentence):
-    return '<START> ' + sentence + ' <END>'
-
-
 def save_dict(file_path, data_dict, encoding='utf-8'):
     with open(file_path, "w", encoding=encoding) as f:
         json.dump(data_dict, f, ensure_ascii=False)
@@ -26,21 +17,3 @@ def read_file(file_path):
     with open(file_path, "r", encoding="utf-8") as f:
         sentences = f.readlines()
     return sentences
-
-
-def build_vocab(sentences):
-    vocab = set()
-    for sentence in sentences:
-        vocab.update(sentence.split())
-    return sorted(list(vocab))
-
-
-def remove_punctuation(sentence):
-    english_punctuations = string.punctuation
-    bangla_punctuations = "৷-–—’‘৳…।"
-    all_punctuations = english_punctuations + bangla_punctuations
-    cleaned_sentence = ''.join([char for char in sentence if char not in all_punctuations])
-    cleaned_sentence = cleaned_sentence.strip()
-    cleaned_sentence = ' '.join(cleaned_sentence.split())
-    return cleaned_sentence
-