flask1.py

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
import re
import os
import sklearn
import tqdm
from tqdm import tqdm 
# import nltk
import warnings
warnings.filterwarnings("ignore") 
import cv2
from sklearn.model_selection import train_test_split
import PIL
from PIL import Image
import time

import tensorflow as tf
import keras
from keras.layers import Input,Dense,Conv2D,concatenate,Dropout,LSTM
from keras import Model
# from tensorflow.keras import activations
import warnings
warnings.filterwarnings("ignore")
import nltk.translate.bleu_score as bleu

# load the dataset 
data = pd.read_csv('data.csv')

# Load train features and test featurs
train_features=np.load("train_image_features.npy")
test_features=np.load("test_image_features.npy")


'''
First we need to load the chexnet nodel (DenseNet121),
'''
from tensorflow.keras.applications import DenseNet121

image_shape= (224,224,3)
image_input= Input(shape=(224,224,3))

base=DenseNet121(include_top=False,input_tensor=image_input,input_shape=image_shape)
pred=Dense(14,"sigmoid")(base.output)

chexnet_model=Model(inputs=base.input,outputs=pred)
chexnet_model.load_weights("chexnet.h5")

# chexnet_model.summary()
final_chexnet_model=Model(inputs=chexnet_model.inputs,outputs=chexnet_model.layers[-2].output,name="Chexnet_model")    


# Image Feature Extraction N

def image_feature_extraction_n(image1, image2):
    image_1 = Image.open(image1)
    image_1 = np.asarray(image_1.convert("RGB"))

    image_2 = Image.open(image2)
    image_2 = np.asarray(image_2.convert("RGB"))

    # normalize the values of the image
    image_1 = image_1 / 255
    image_2 = image_2 / 255

    # resize all image into (224, 224)
    image_1 = cv2.resize(image_1, (224, 224))
    image_2 = cv2.resize(image_2, (224, 224))

    image_1 = np.expand_dims(image_1, axis=0)
    image_2 = np.expand_dims(image_2, axis=0)

    # now we have read two images per patient. These are given to the chexnet model for feature extraction
    image_1_out = final_chexnet_model(image_1)
    image_2_out = final_chexnet_model(image_2)

    # concatenate along the width
    conc = np.concatenate((image_1_out, image_2_out), axis=2)

    # reshape into (no. of images passed, length * breadth * depth)
    image_feature = tf.reshape(conc, (conc.shape[0], -1, conc.shape[-1]))

    # reshape for attention mechanism
    
    # image_feature_reshaped = tf.reshape(image_feature, (image_feature.shape[0], -1, image_feature.shape[-1]))
    image_feature_reshaped = tf.reshape(image_feature, (image_feature.shape[0], -1, 2048))

    return image_feature_reshaped


# Image Feature Extraction

def image_feature_extraction(image1, image2):
    image_1 = Image.open(image1)
    image_1 = np.asarray(image_1.convert("RGB"))

    image_2 = Image.open(image2)
    image_2 = np.asarray(image_2.convert("RGB"))

    # normalize the values of the image
    image_1 = image_1 / 255
    image_2 = image_2 / 255

    # resize all image into (224, 224)
    image_1 = cv2.resize(image_1, (224, 224))
    image_2 = cv2.resize(image_2, (224, 224))

    image_1 = np.expand_dims(image_1, axis=0)
    image_2 = np.expand_dims(image_2, axis=0)

    # now we have read two images per patient. These are given to the chexnet model for feature extraction
    image_1_out = final_chexnet_model(image_1)
    image_2_out = final_chexnet_model(image_2)

    # concatenate along the width
    conc = np.concatenate((image_1_out, image_2_out), axis=2)

    # reshape into (no. of images passed, length * breadth * depth)
    image_feature = tf.reshape(conc, (conc.shape[0], -1, conc.shape[-1]))

    # reshape for attention mechanism
    
    image_feature_reshaped = tf.reshape(image_feature, (image_feature.shape[0], -1, image_feature.shape[-1]))
    # image_feature_reshaped = tf.reshape(image_feature, (image_feature.shape[0], -1, 2048))

    return image_feature_reshaped

'''
Modify the reports as <sos> report text <eos>. This format is useful for the decoder while predicting the next word
'''
train = pd.read_csv('train.csv')
test = pd.read_csv('test.csv')
train_report=["<sos> "+text+" <eos>" for text in train["report"].values]
train_report_in=["<sos> "+text for text in train["report"].values]
train_report_out=[text+" <eos>" for text in train["report"].values]

test_report=["<sos> " +text+" <eos>" for text in test["report"].values]
test_report_in=["<sos> " +text for text in test["report"].values]
test_report_out=[text+" <eos>" for text in test["report"].values]  

bs=10
max_len=80

#Obtaining the text embeddings of the report
# we use the tensorflow tokenizer to convert the text into tokens
#we also pad the sequences to a length 300 which is around the 90th percentile of the lengths of the report

token=tf.keras.preprocessing.text.Tokenizer(filters='' )

token.fit_on_texts(train_report)
vocab_size=len(token.word_index)+1 

seq=token.texts_to_sequences(train_report_in)
train_padded_inp=tf.keras.preprocessing.sequence.pad_sequences(seq,maxlen=max_len,padding="post")


seq=token.texts_to_sequences(train_report_out)
train_padded_out=tf.keras.preprocessing.sequence.pad_sequences(seq,maxlen=max_len,padding="post")

seq=token.texts_to_sequences(test_report_in)
test_padded_inp=tf.keras.preprocessing.sequence.pad_sequences(seq,maxlen=max_len,padding="post")


seq=token.texts_to_sequences(test_report_out)
test_padded_out=tf.keras.preprocessing.sequence.pad_sequences(seq,maxlen=max_len,padding="post")


#now we prepare the data set with the image fetaures and the reports
train_dataset = tf.data.Dataset.from_tensor_slices((train_features,train_padded_inp,train_padded_out)).shuffle(500)
train_dataset = train_dataset.batch(bs, drop_remainder=True)

test_dataset = tf.data.Dataset.from_tensor_slices((test_features,test_padded_inp,test_padded_out)).shuffle(500)
test_dataset = test_dataset.batch(bs,drop_remainder=True) 


#https://machinelearningmastery.com/use-word-embedding-layers-deep-learning-keras/
embeddings_index=dict()
f = open('glove.6B.300d.txt', encoding='utf-8')
for line in f:
	values = line.split()
	word = values[0]
	coefs = np.asarray(values[1:], dtype='float32')
	embeddings_index[word] = coefs
f.close()
print("Done")
# create a weight matrix for words in training docs

embedding_matrix = np.zeros((vocab_size, 300))
for word, i in tqdm(token.word_index.items()):
	embedding_vector = embeddings_index.get(word)
	if embedding_vector is not None:
		embedding_matrix[i] = embedding_vector 


enc_units=64
embedding_dim=300
dec_units=64 
att_units=64

input_img=Input(shape=(98,1024),name="image_fetaures")
# input_img=Input(shape=(None,1,1024),name="image_fetaures")
input_txt=Input(shape=(max_len),name="text_input")

#encoder model
en_out=Dense(enc_units,activation="relu",name="encoder_dense")(input_img)
enc_out=tf.keras.layers.Dropout(0.5)(en_out)

state1= Input(shape=(bs,enc_units),name="state1")
state2= Input(shape=(bs,enc_units),name="state2")
state_h=tf.keras.layers.Add()([state1,state2])
#decoder model with attention

emb_out=tf.keras.layers.Embedding(vocab_size,output_dim=300,input_length=max_len,mask_zero=True,trainable=False,weights=[embedding_matrix])(input_txt)
weights=tf.keras.layers.AdditiveAttention()([state_h,en_out])
context_vector=tf.matmul(en_out,weights,transpose_b=True)[:,:,0]
context_vector=Dense(embedding_dim)(context_vector)
result=tf.concat([tf.expand_dims(context_vector, axis=1),emb_out],axis=1)
gru_out,state_1,state_2=tf.keras.layers.Bidirectional(tf.keras.layers.GRU(dec_units,return_sequences=True, return_state=True,name="Bidirectional_GRU"))(result)
out=tf.keras.layers.Dense(vocab_size,name="decoder_final_dense")(gru_out)
en_de=Model(inputs=[input_txt,input_img,state1,state2],outputs=out)

#  Encoder_n
#encoder model

class Encoder_n(tf.keras.Model):
  def __init__(self,units):
    super().__init__()
    self.units=units
    
  
  def build(self,input_shape):
    # self.dense1=Dense(self.units,activation="relu",name="encoder_dense")
    self.dense1=Dense(2048,activation="relu",name="encoder_dense")
      
       
    self.maxpool=tf.keras.layers.Dropout(0.5)

  def call(self, input_):
        try:
            enc_out = self.maxpool(input_)
            enc_out = self.dense1(enc_out)
            return enc_out
        except Exception as e:
            raise ValueError(f"Error in Encoder layer with input shape {input_.shape}: {str(e)}")
    
  def initialize_states(self,batch_size):
      '''
      Given a batch size it will return intial hidden state
      If batch size is 32- Hidden state shape is [32,units]
      '''
      forward_h=tf.zeros((batch_size,self.units))
      back_h=tf.zeros((batch_size,self.units))
      return forward_h,back_h

#encoder model
'''
here the input will be image features with size (96,1024). We can consider this tensor as the encoder output.
But here we add another dense layer that will reduce the depth of this feature from 1024 to a low value
'''

class Encoder(tf.keras.Model):
  def __init__(self,units):
    super().__init__()
    self.units=units
    
  
  def build(self,input_shape):
    self.dense1=Dense(self.units,activation="relu",name="encoder_dense")
    # self.dense1=Dense(2048,activation="relu",name="encoder_dense")
      
       
    self.maxpool=tf.keras.layers.Dropout(0.5)

  def call(self, input_):
        try:
            enc_out = self.maxpool(input_)
            enc_out = self.dense1(enc_out)
            return enc_out
        except Exception as e:
            raise ValueError(f"Error in Encoder layer with input shape {input_.shape}: {str(e)}")
    
  def initialize_states(self,batch_size):
      '''
      Given a batch size it will return intial hidden state
      If batch size is 32- Hidden state shape is [32,units]
      '''
      forward_h=tf.zeros((batch_size,self.units))
      back_h=tf.zeros((batch_size,self.units))
      return forward_h,back_h

'''
this is the attention class. 
Here the input to the decoder and the gru hidden state at the pevious time step are given, and the context vector is calculated

This context vector is calculated uisng the attention weights. This context vector is then passed to the decoder model

Here conact function is used for calaculating the attention weights

'''

class Attention(tf.keras.layers.Layer):

  def __init__(self,att_units):

    super().__init__()
    
    self.att_units=att_units

  def build(self,input_shape):
    self.wa=tf.keras.layers.Dense(self.att_units)
    self.wb=tf.keras.layers.Dense(self.att_units)
    self.v=tf.keras.layers.Dense(1)
  
    
  def call(self,decoder_hidden_state,encoder_output):
   
    x=tf.expand_dims(decoder_hidden_state,1)
    
    # print(x.shape)
    # print(encoder_output.shape)
      
    alpha_dash=self.v(tf.nn.tanh(self.wa(encoder_output)+self.wb(x)))
    
    alphas=tf.nn.softmax(alpha_dash,1)

    # print("en",encoder_output.shape)
    # print("al",alphas.shape)
    
    context_vector=tf.matmul(encoder_output,alphas,transpose_a=True)[:,:,0]
    # context_vector = alphas*encoder_output
    # print("c",context_vector.shape)


    return (context_vector,alphas)
  

'''
This class will perform the decoder task.
The main decoder will call this onestep decoder at every time step. This one step decoder in turn class the atention model and return the ouptput 
at time step t.
This output is passed through the final softmax layer with output size =vocab size, and pass this result to the main decoder model

'''

class One_Step_Decoder(tf.keras.Model):
  def __init__(self,vocab_size, embedding_dim, input_length, dec_units ,att_units):

      # Initialize decoder embedding layer, LSTM and any other objects needed
    super().__init__()
    
    self.att_units=att_units
    self.vocab_size=vocab_size
    self.embedding_dim=embedding_dim
    self.input_length=input_length
    
    self.dec_units=dec_units
    self.attention=Attention(self.att_units)
  #def build(self,inp_shape):
    self.embedding=tf.keras.layers.Embedding(self.vocab_size,output_dim=self.embedding_dim,
                                             input_length=self.input_length,mask_zero=True,trainable=False,weights=[embedding_matrix])

    self.gru= tf.keras.layers.Bidirectional(tf.keras.layers.GRU(self.dec_units,return_sequences=True, return_state=True))
    self.dense=tf.keras.layers.Dense(self.vocab_size,name="decoder_final_dense") 
    self.dense_2=tf.keras.layers.Dense(self.embedding_dim,name="decoder_dense2")

  def call(self,input_to_decoder, encoder_output, for_h,bac_h):
    
    embed=self.embedding(input_to_decoder)
    state_h=tf.keras.layers.Add()([for_h,bac_h])
    

    context_vector,alpha=self.attention(state_h,encoder_output)
    context_vector=self.dense_2(context_vector)

    result=tf.concat([tf.expand_dims(context_vector, axis=1),embed],axis=-1)
    
   
    output,forward_h,back_h=self.gru(result,initial_state=[for_h,bac_h])
    out=tf.reshape(output,(-1,output.shape[-1]))

    out=tf.keras.layers.Dropout(0.5)(out)
    
    dense_op=self.dense(out)
    
    return dense_op,forward_h,back_h,alpha
  

'''
For every input sentence, each output word is generated using one step decoder. Each output word is stored using the final decoder model and the
final output sentence is returned

'''

class Decoder(tf.keras.Model):
    def __init__(self, vocab_size, embedding_dim, output_length, dec_units,att_units):
      super().__init__()
      #Intialize necessary variables and create an object from the class onestepdecoder
      self.onestep=One_Step_Decoder(vocab_size, embedding_dim, output_length, dec_units,att_units)


        
    def call(self, input_to_decoder,encoder_output,state_1,state_2):

        #Initialize an empty Tensor array, that will store the outputs at each and every time step
        #Create a tensor array as shown in the reference notebook
        
        #Iterate till the length of the decoder input
            # Call onestepdecoder for each token in decoder_input
            # Store the output in tensorarray
        # Return the tensor array
        
        all_outputs=tf.TensorArray(tf.float32,input_to_decoder.shape[1],name="output_array")
        for step in range(input_to_decoder.shape[1]):
          output,state_1,state_2,alpha=self.onestep(input_to_decoder[:,step:step+1],encoder_output,state_1,state_2)

          all_outputs=all_outputs.write(step,output)
        all_outputs=tf.transpose(all_outputs.stack(),[1,0,2])
        
        return all_outputs
    
import warnings
warnings.filterwarnings("ignore")

class encoder_decoder(tf.keras.Model):
  def __init__(self,enc_units,embedding_dim,vocab_size,output_length,dec_units,att_units,batch_size):
        super().__init__()

        
        self.batch_size=batch_size
        self.encoder =Encoder(enc_units)
        self.decoder=Decoder(vocab_size,embedding_dim,output_length,dec_units,att_units)
        
  
    #Coompute the image features using feature extraction model and pass it to the encoder
    # This will give encoder ouput
   # Pass the decoder sequence,encoder_output,initial states to Decoder
    # return the decoder output

  
  def call(self, data):
        features,report  = data[0], data[1]
        
        encoder_output= self.encoder(features)
        state_h,back_h=self.encoder.initialize_states(self.batch_size)
        
        output= self.decoder(report, encoder_output,state_h,back_h)
      
        return output      
  
model  = encoder_decoder(enc_units,embedding_dim,vocab_size,max_len,dec_units,att_units,bs)

optimizer = tf.keras.optimizers.Adam()

loss_function = tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True, reduction='auto')

def custom_lossfunction(y_true, y_pred):
    #getting mask value
    mask = tf.math.logical_not(tf.math.equal(y_true, 0))
    
    #calculating the loss
    loss_ = loss_function(y_true, y_pred)
    
    #converting mask dtype to loss_ dtype
    mask = tf.cast(mask, dtype=loss_.dtype)
    
    #applying the mask to loss
    loss_ = loss_*mask
    
    #getting mean over all the values
    loss_ = tf.reduce_mean(loss_)
    return loss_ 


# model.compile(optimizer=optimizer,loss=custom_lossfunction)
model.compile(optimizer=optimizer,loss=custom_lossfunction)


from tensorflow.keras.models import load_model
# Load the model with custom_objects parameter
model = load_model("model2wts/ckpt", custom_objects={"custom_lossfunction": custom_lossfunction,"Encoder": Encoder, "Decoder": Decoder})


def take_second(elem):
    return elem[1]



def beam_search(image1,image2, beam_index):

    hidden_state =  tf.zeros((1, enc_units))
    hidden_state2 =  tf.zeros((1, enc_units))
    image_features=image_feature_extraction_n(image1,image2)

    encoder_out = model.layers[0](image_features)
    # encoder_out = model.layers

    start_token = [token.word_index["<sos>"]]
    dec_word = [[start_token, 0.0]]
    while len(dec_word[0][0]) < max_len:
        temp = []
        for word in dec_word:
            
            predict, hidden_state,hidden_state2,alpha = model.layers[1].onestep(tf.expand_dims([word[0][-1]],1), encoder_out, hidden_state,hidden_state2)
           
           
            word_predict = np.argsort(predict[0])[-beam_index:]
            for i in word_predict:

                next_word, probab = word[0][:], word[1]
                next_word.append(i)
                probab += predict[0][i] 
                temp.append([next_word, probab.numpy()])
        dec_word = temp
        # Sorting according to the probabilities scores
        
        
        dec_word = sorted(dec_word, key=take_second)
       
        # Getting the top words
        dec_word = dec_word[-beam_index:] 
        
     
    final = dec_word[-1]
    
    report =final[0]
    score = final[1]
    temp = []
    
    for word in report:
      if word!=0:
        if word != token.word_index['<eos>']:
            temp.append(token.index_word[word])
        else:
            break 

    rep = ' '.join(e for e in temp)        
    
    return rep, score

# Input Images here from backend
img1 = 1
img2 = 2
result,score=beam_search(img1,img2,3) 
print("GENERATED REPORT: ",result)


# data = pd.read_csv('data.csv')
# import random 
# start=time.time()
# i=random.sample(range(test.shape[0]),1)[0]
# img1=data.iloc[i]["image1"]
# img2=data.iloc[i]["image2"]
#   #show th corresponding x-ray images
# i1=cv2.imread(img1)
# i2=cv2.imread(img2)
# plt.figure(figsize=(10,6))
# plt.subplot(131)
# plt.title("image1")
# # plt.imshow(i1)
# plt.subplot(132)
# plt.title("image2")
# plt.imshow(i2)
# # plt.show()
#   #printing the actual and generated results
  
# result,score=beam_search(img1,img2,3) 
# actual=test_report[i]
  
# print("ACTUAL REPORT: ",actual)
# print("GENERATED REPORT: ",result)
# end=time.time() 
# print("BLEU SCORE IS: ",bleu.sentence_bleu(actual,result))

# print("time required for the evaluation is ",end-start)