Arabic Translation System: Seq2Seq Statistical learning

SOS = 0
EOS = 1
FILES = {'ar': 'ara.txt', 'en': 'eng.txt'}


class Dictionary:
    def __init__(self, name):
        print('Building Dictionary for lang %s' %(name))
        self.name = name
        self.word_to_idx = {}
        self.idx_to_word = {0: "<S>", 1: "</S>"}
        self.words_count = {}
        self.n_words = 2  # Count SOS, EOS special tokens

    def add_word(self, word):
        if word not in self.word_to_idx:
            self.word_to_idx[word] = self.n_words
            self.words_count[word] = 1
            self.idx_to_word[self.n_words] = word
            self.n_words += 1
        else:
            self.words_count[word] += 1

    def add_sentence(self, sentence):
        for word in sentence.split():
            self.add_word(word)
            
class Reader:
    
    def __init__(self, lang, path, max_len=10, min_len=1, max_chars=40):
        self.lang = lang
        self.path = path
        self.sentences = []
        self.max_len = max_len
        self.min_len = min_len
        self.max_chars = max_chars

    def read(self):
        print('Reading Sentences for language %s ...' %(self.lang))
        with open(self.path, 'r') as reader:
            self.sentences += list(map(tools.cleaner_job, reader.readlines()))
        return self.sentences

    def get_tokenized(self):
        print('Tokenizing %s ...' %(self.lang))
        return [[w for w in s.split()] for s in self.sentences]

class SentenceReader(Reader):

    def __init__(self, input_lang_map, outpt_lang_map, max_length=10):
        super(SentenceReader, self).__init__(input_lang_map, outpt_lang_map, max_len=max_length)
        in_lang, in_path = tuple(*(input_lang_map.items()))
        out_lang, out_path = tuple(*(outpt_lang_map.items()))
        self.input = Reader(lang=in_lang, path=in_path, max_len=max_length)
        self.outpt = Reader(lang=out_lang, path=out_path, max_len=max_length)
        self.lang_map = {in_lang: self.input, out_lang: self.outpt}
        self.input.read()
        self.outpt.read()
        self.to_remove = set()
        self.to_have = set()
        self.__filter_sentences()
        
    def read_sentences(self, lang):
        to_have = self.to_have
        lang = self.lang_map.get(lang)
        if lang:
            return lang.sentences
        else:
            raise AttributeError("Invalid Language Attribute")
        
    def get_tokenized(self, lang):
        print('Tokenizing %s ...' %(lang))
        sentences = self.lang_map.get(lang).sentences
        return [[w for w in s.split()] for s in sentences]
    
    def __filter_sentences(self):
        max_len, min_len, max_chars = self.max_len, self.min_len, self.max_chars
        to_have = self.to_have
        to_remove = self.to_remove
        
        for reader in [self.input, self.outpt]:
            for i, sentence in enumerate(reader.sentences):
                if len(sentence.split()) >= max_len:
                    to_remove.add(i)
                elif len(sentence.split()) < min_len:
                    to_remove.add(i)
                if i not in to_remove:
                    to_have.add(i)
                    
        for reader in [self.input, self.outpt]:
            to_have = [i for i in to_have if i not in to_remove]
            sent_series_obj = pd.Series(reader.sentences)
            reader.sentences = sent_series_obj[[i for i in to_have]].tolist()

def prepare_dataset(in_lang_path, out_lang_path, max_length=10):
    print('Preparing Dataset')
    
    reader = SentenceReader({'en': in_lang_path}, {'ar': out_lang_path}, max_length=max_length)

    pairs = [(in_sent, out_sent) for (in_sent, out_sent) in zip(reader.read_sentences('en'), 
                                                                reader.read_sentences('ar'))]

    print('Building Language Dictionary')

    in_dictionary = Dictionary('english')
    ou_dictionary = Dictionary('arabic')
    for in_sent, out_sent in pairs:
        in_dictionary.add_sentence(in_sent)
        ou_dictionary.add_sentence(out_sent)

    print('Encoding Tokenized Sentences to unique identifiers')
    in_word_to_idx = in_dictionary.word_to_idx
    ou_word_to_idx = ou_dictionary.word_to_idx
    pairs_encoded = [([in_word_to_idx[in_w]  for in_w in in_sent], 
                      [ou_word_to_idx[out_w] for out_w in out_sent]) 
                                             for (in_sent, out_sent) 
                                             in zip(reader.get_tokenized('en'), 
                                                    reader.get_tokenized('ar'))]
    
    return {
        'pairs': pairs, 
        'pairs_encoded': pairs_encoded, 
        'in_dictionary': in_dictionary, 
        'out_dictionary': ou_dictionary,
    }
    
def pad_sequence(indices):
    max_len = 0
    for seq in indices:
        if len(seq) > max_len:
            max_len = len(seq)
            
    for seq in indices:
        seq += [EOS] * (max_len - len(seq))

def index_to_tensor(pairs):
    input_idxs = []
    output_idxs = []
    tensors = {}
    
    for pair in pairs.copy():
        input_idxs += [pair[0] + [EOS]]
        output_idxs += [[SOS] + pair[1] + [EOS]]
        
    for indices in [input_idxs, output_idxs]:
        pad_sequence(indices)
        
    print(input_idxs[:10])
    print(output_idxs[:10])
        
    tensors['input'] = torch.tensor(input_idxs)
    tensors['target'] = torch.tensor(output_idxs)
    
    return tensors

data = loader.prepare_dataset('./eng.txt', './ara.txt', max_length=16)

import random

data_text = [random.choice(data.get('pairs')) for i in range(10)]
data_encoded = [random.choice(data.get('pairs_encoded')) for i in range(10)]

for item in data_text:
  print(item)

for item in data_encoded:
  print(item)

HIDDEN_SIZE = 128
INPUT_SIZE = data.get('in_dictionary').n_words
OUTPUT_SIZE = data.get('out_dictionary').n_words

tensors = data.get('pairs_encoded')

dictionaries = {'output': data.get('out_dictionary'),
                'input' : data.get('in_dictionary')}

model = Seq2seqModel(iterations=1000000, lr=0.01, hidden_size=HIDDEN_SIZE, dictionaries=dictionaries,
                     input_size=INPUT_SIZE, output_size=OUTPUT_SIZE, max_length=16)

""" 
------ Trainging ------ 
"""

model.train(tensors=tensors) 

""" 
------ Evaluating ------ 
"""
sentences = data.get('pairs')
model.evaluate(sentences)