eng_fre.py

# -*- coding: utf-8 -*-
"""ENG_FRE.ipynb

Automatically generated by Colab.

Original file is located at
    https://colab.research.google.com/github/VaradBelwalkar/NLP_Seq2Seq/blob/main/ENG_FRE.ipynb

## English to French Seq2Seq model
"""

from keras.models import Model
from keras.layers import LSTM
from keras.layers import Dense, Input, Embedding
from keras.preprocessing.sequence import pad_sequences
from collections import Counter
from keras.callbacks import ModelCheckpoint
import nltk
import numpy as np
import os
import zipfile
import sys
import urllib.request
from gensim.models import KeyedVectors
from keras.utils import plot_model
from nltk.translate.bleu_score import sentence_bleu, SmoothingFunction
from keras.optimizers import Adam

from keras.activations import softmax
from keras.layers import Dense, Activation, RepeatVector, Permute
from keras.layers import Input, Embedding, Multiply, Concatenate, Lambda
from keras.layers import TimeDistributed

"""#hyperparameters"""

BATCH_SIZE = 64
NUM_EPOCHS = 100
HIDDEN_UNITS = 256
NUM_SAMPLES = 10000
MAX_VOCAB_SIZE = 10000
EMBEDDING_SIZE = 100
DATA = 'fra.txt'

tar_count = Counter()

GLOVE_MODEL = "glove.6B.100d.txt"
WEIGHT_FILE_PATH = 'eng-to-fr-glove-weights.h5'

"""## Downloading the GloVe embeddings zip"""

!wget http://nlp.stanford.edu/data/glove.6B.zip

"""## Extracting the zip to get required embeddings"""

!unzip glove.6B.zip

"""## Sample word embeddings"""

!head glove.6B.100d.txt

"""## Loading GloVe Embeddings"""

def load_glove():

    _word2em = {}
    file = open(GLOVE_MODEL, mode='r', encoding='utf8')
    for line in file:
        words = line.strip().split()
        word = words[0]
        embeds = np.array(words[1:], dtype=np.float32)
        _word2em[word] = embeds
    file.close()
    return _word2em

word2em = load_glove()

"""## Contents of the fra.txt"""

!head fra.txt

nltk.download('punkt')

"""## Preparing Data for Sequence Translation"""

lines = open(DATA, 'r', encoding='utf8').read().split('\n')
for line in lines[: len(lines)-1]:
    input_text, target_text = line.split('\t')
    input_words = [w for w in nltk.word_tokenize(input_text.lower())]
    target_text = '\t' + target_text + '\n'
    for char in target_text:
        tar_count[char] += 1

"""## Creating Token Index Dictionaries"""

target_word2idx = dict()

for idx, word in enumerate(tar_count.most_common(MAX_VOCAB_SIZE)):
    #print(word)
    target_word2idx[word[0]] = idx

target_idx2word = dict([(idx, word) for word, idx in target_word2idx.items()])

num_decoder_tokens = len(target_idx2word)

unknown_emb = np.random.randn(EMBEDDING_SIZE)

encoder_max_seq_length = 0
decoder_max_seq_length = 0

input_texts_word2em = []

"""## Processing Input and Target Texts"""

lines = open(DATA, 'r', encoding='utf8').read().split('\n')
for line in lines[: min(NUM_SAMPLES, len(lines)-1)]:
    input_text, target_text = line.split('\t')
    target_text = '\t' + target_text + '\n'
    input_words = [w for w in nltk.word_tokenize(input_text.lower())]
    encoder_input_wids = []
    for w in input_words:
        em = unknown_emb
        if w in word2em:
            em = word2em[w]
        encoder_input_wids.append(em)
    input_texts_word2em.append(encoder_input_wids)
    encoder_max_seq_length = max(len(encoder_input_wids), encoder_max_seq_length)
    decoder_max_seq_length = max(len(target_text), decoder_max_seq_length)

encoder_input_data = pad_sequences(input_texts_word2em, encoder_max_seq_length)

"""## decoder word2index input"""

decoder_target_data = np.zeros(shape=(NUM_SAMPLES, decoder_max_seq_length, num_decoder_tokens))
decoder_input_data = np.zeros(shape=(NUM_SAMPLES, decoder_max_seq_length, num_decoder_tokens))
lines = open(DATA, 'rt', encoding='utf8').read().split('\n')
for lineIdx, line in enumerate(lines[: min(NUM_SAMPLES, len(lines)-1)]):
    _, target = line.split('\t')
    target = '\t' + target + '\n'
    for idx, char in enumerate(target):
        if char in target_word2idx:
            w2idx = target_word2idx[char]
            decoder_input_data[lineIdx, idx, w2idx] = 1
            if idx > 0:
                decoder_target_data[lineIdx, idx-1, w2idx] = 1

context = dict()
context['num_decoder_tokens'] = num_decoder_tokens
context['encoder_max_seq_length'] = encoder_max_seq_length
context['decoder_max_seq_length'] = decoder_max_seq_length

"""## defining Encoder- Decoder Model"""

encoder_inputs = Input(shape=(None, EMBEDDING_SIZE), name='encoder_inputs')
encoder_lstm = LSTM(units=HIDDEN_UNITS, return_state=True, name='encoder_lstm')
encoder_outputs, encoder_state_h, encoder_state_c = encoder_lstm(encoder_inputs)
encoder_states = [encoder_state_h, encoder_state_c]

decoder_inputs = Input(shape=(None, num_decoder_tokens), name='decoder_inputs')
decoder_lstm = LSTM(units=HIDDEN_UNITS, return_state=True, return_sequences=True, name='decoder_lstm')
decoder_outputs, decoder_state_h, decoder_state_c = decoder_lstm(decoder_inputs,
                                                                 initial_state=encoder_states)
decoder_dense = Dense(units=num_decoder_tokens, activation='softmax', name='decoder_dense')
decoder_outputs = decoder_dense(decoder_outputs)

model = Model([encoder_inputs, decoder_inputs], decoder_outputs)

model.compile(loss='categorical_crossentropy', optimizer='adam')


checkpoint = ModelCheckpoint(filepath=WEIGHT_FILE_PATH, save_best_only=True)
model.fit([encoder_input_data, decoder_input_data], decoder_target_data, batch_size=BATCH_SIZE, epochs=NUM_EPOCHS,
          verbose=1, validation_split=0.3, callbacks=[checkpoint])

model.save_weights(WEIGHT_FILE_PATH)

model.summary()

encoder_model_inf = Model(encoder_inputs, encoder_states)


decoder_state_input_h = Input(shape=(HIDDEN_UNITS,))
decoder_state_input_c = Input(shape=(HIDDEN_UNITS,))
decoder_input_states = [decoder_state_input_h, decoder_state_input_c]

decoder_out, decoder_h, decoder_c = decoder_lstm(decoder_inputs,
                                                 initial_state=decoder_input_states)

decoder_states = [decoder_h , decoder_c]

decoder_out = decoder_dense(decoder_out)

decoder_model_inf = Model(inputs=[decoder_inputs] + decoder_input_states,
                          outputs=[decoder_out] + decoder_states )

max_encoder_seq_length = context['encoder_max_seq_length']
max_decoder_seq_length = context['decoder_max_seq_length']
num_decoder_tokens = context['num_decoder_tokens']

def predict_sent(input_text):
        input_seq = []
        input_wids = []
        for word in nltk.word_tokenize(input_text.lower()):
            emb = unknown_emb
            if word in word2em:
                emb = word2em[word]
            input_wids.append(emb)
        input_seq.append(input_wids)
        input_seq = pad_sequences(input_seq, max_encoder_seq_length)
        states_value = encoder_model_inf.predict(input_seq)
        target_seq = np.zeros((1, 1,num_decoder_tokens))
        target_seq[0, 0, target_word2idx['\t']] = 1
        target_text = ''
        terminated = False
        while not terminated:
            output_tokens, h, c = decoder_model_inf.predict([target_seq] + states_value)

            sample_token_idx = np.argmax(output_tokens[0, -1, :])
            sample_word = target_idx2word[sample_token_idx]
            target_text += sample_word

            if sample_word == '\n' or len(target_text) >= max_decoder_seq_length:
                terminated = True

            target_seq = np.zeros((1, 1, num_decoder_tokens))
            target_seq[0, 0, sample_token_idx] = 1

            states_value = [h, c]
        return target_text.strip()

print(predict_sent('this is amazing.'))

def calculate_bleu_score(reference, candidate):
    smoothie = SmoothingFunction().method4
    return sentence_bleu([reference], candidate, smoothing_function=smoothie)

input_sentence = 'This is amazing.'
predicted_sentence = predict_sent(input_sentence)
print('Predicted:', predicted_sentence)

reference_sentence = 'Ceci est incroyable.'  # Reference translation
bleu_score = calculate_bleu_score(reference_sentence.split(), predicted_sentence.split())
print('BLEU score:', bleu_score)