Add the required files

app.py

@@ -0,0 +1,42 @@
+import streamlit as st
+from gcg.pipelines import predict
+import os
+
+# Define the directory to save uploaded files
+TEMP_DIR = "temp"
+os.makedirs(TEMP_DIR, exist_ok=True)  # Create the temp directory if it doesn't exist
+
+st.title("Retinal Lesion Detector")
+st.subheader("Upload retinal images and get predictions with heatmaps")
+
+# File uploader to accept multiple images
+uploaded_files = st.file_uploader(
+    "Upload Retinal Images", 
+    type=["jpg", "jpeg", "png"], 
+    accept_multiple_files=True
+)
+
+if st.button("Run Inference"):
+    if uploaded_files:
+        img_paths = []
+        for uploaded_file in uploaded_files:
+            # Save each uploaded file to the temp directory
+            file_path = os.path.join(TEMP_DIR, uploaded_file.name)
+            with open(file_path, "wb") as f:
+                f.write(uploaded_file.getbuffer())
+            img_paths.append(file_path)  # Collect the file path for inference
+
+        # Pass the file paths to the predict function
+        st.info("Running predictions...")
+        predictions = predict(img_paths)
+
+        # Display predictions and heatmaps
+        st.success("Inference completed! Here are the results:")
+        for img_path, predicted_class in zip(img_paths, predictions):
+            st.write(f"**Image**: {os.path.basename(img_path)}")
+            st.write(f"**Predicted Class**: {predicted_class}")
+            heatmap_path = os.path.join("heatmaps", f"heatmap_{os.path.basename(img_path)}")
+            if os.path.exists(heatmap_path):
+                st.image(heatmap_path, caption="Attention Map", use_container_width=True)
+    else:
+        st.error("Please upload at least one image.")

gcg/components/__init__.py

@@ -0,0 +1,3 @@
+from .model import build_model, train_model, evaluate_model
+from .data_processor import preprocess_image, load_data
+from .gradcam import grad_cam_plus, show_GradCAM

gcg/components/data_processor.py

@@ -0,0 +1,82 @@
+import os
+import cv2
+import numpy as np
+import sys
+from gcg import config
+from sklearn.preprocessing import LabelEncoder
+from tensorflow.keras.utils import to_categorical
+from sklearn.model_selection import train_test_split
+from gcg.utils import logging, CustomException, save_object
+
+
+def preprocess_image(img_path, image_size):
+    # Read the image from the specified path
+    img = cv2.imread(img_path)
+    # Convert the image from BGR to RGB
+    img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
+    img_array=cv2.resize(img, (image_size[0],image_size[1]), fx=1, fy=1,interpolation = cv2.INTER_CUBIC)  
+
+    return img_array
+
+# Data Loading
+def load_data(data_path, image_size):
+    try:
+        subfolders = config.labels
+        logging.info("Dataset Loading...")
+
+        img_data_list=[]
+        labels_list = []
+        num_images_per_class = []
+
+        for category in subfolders:
+            img_list=os.listdir(data_path +'/'+ category)
+            if("Annotations" in category):
+                continue
+            
+            logging.info(f'Loading : {len(img_list)}, images of category: {category}')
+            for img in img_list:
+                # Load an image from this path
+                img_path = data_path + '/'+ category + '/'+ img
+
+                # Preprocess image
+                img_array=preprocess_image(img_path, image_size)  
+
+                img_data_list.append(img_array) 
+                labels_list.append(category)
+            num_images_per_class.append(len(img_list))
+
+        le = LabelEncoder()
+        labels = le.fit_transform(labels_list)
+        labels = to_categorical(labels)
+
+        # Saving the label encoder object for use during inference
+        save_object(config.labelencoder_save_path, le)
+
+        data = np.array(img_data_list)
+
+        # Dataset Summary
+        logging.info(f"Total number of uploaded data: {data.shape[0]} with data shape, ({data.shape[1]},{data.shape[2]},{data.shape[3]})")
+        
+        logging.info("Initiated train_test_split")
+        X_train, X_test, y_train, y_test = initiate_train_test_split(data, labels)
+
+        return X_train, X_test, y_train, y_test
+    
+    except Exception as e:
+        raise CustomException(e, sys)
+
+# Train Test Split
+def initiate_train_test_split(data, labels):
+
+    # Split the dataset into two subsets (80%-20%). The first one will be used for training.
+    X_train, X_test, y_train, y_test = train_test_split(data, labels, test_size=0.2, random_state=195, stratify=labels)
+
+    logging.info(f"X_train has shape: {X_train.shape}")
+    logging.info(f"y_train has shape: {y_train.shape}\n")
+
+    logging.info(f"X_test has shape: {X_test.shape}")
+    logging.info(f"y_test has shape: {y_test.shape}\n")
+
+    logging.info(f"X_train + X_test = {X_train.shape[0] + X_test.shape[0]} samples in total")
+
+    return X_train, X_test, y_train, y_test

gcg/components/gradcam.py

@@ -0,0 +1,189 @@
+import numpy as np
+import cv2
+import matplotlib.pyplot as plt
+import tensorflow as tf
+from tensorflow.keras.preprocessing import image
+from tensorflow.keras import Model
+from gcg.utils import logging
+
+def grad_cam(model, img,
+             layer_name="block5_conv3", label_name=None,
+             category_id=None):
+    """Get a heatmap by Grad-CAM.
+
+    Args:
+        model: A model object, build from tf.keras 2.X.
+        img: An image ndarray.
+        layer_name: A string, layer name in model.
+        label_name: A list or None,
+            show the label name by assign this argument,
+            it should be a list of all label names.
+        category_id: An integer, index of the class.
+            Default is the category with the highest score in the prediction.
+
+    Return:
+        A heatmap ndarray(without color).
+    """
+    img_tensor = np.expand_dims(img, axis=0)
+
+    conv_layer = model.get_layer(layer_name)
+    heatmap_model = Model([model.inputs], [conv_layer.output, model.output])
+
+    with tf.GradientTape() as gtape:
+        conv_output, predictions = heatmap_model(img_tensor)
+        if category_id is None:
+            category_id = np.argmax(predictions[0])
+        if label_name is not None:
+            print(label_name[category_id])
+        output = predictions[:, category_id]
+        grads = gtape.gradient(output, conv_output)
+        pooled_grads = tf.reduce_mean(grads, axis=(0, 1, 2))
+
+    heatmap = tf.reduce_mean(tf.multiply(pooled_grads, conv_output), axis=-1)
+    heatmap = np.maximum(heatmap, 0)
+    max_heat = np.max(heatmap)
+    if max_heat == 0:
+        max_heat = 1e-10
+    heatmap /= max_heat
+
+    return np.squeeze(heatmap)
+
+def grad_cam_plus(model, img,
+                  layer_name="block5_conv3", label_name=None,
+                  category_id=None):
+    """Get a heatmap by Grad-CAM++.
+
+    Args:
+        model: A model object, build from tf.keras 2.X.
+        img: An image ndarray.
+        layer_name: A string, layer name in model.
+        label_name: A list or None,
+            show the label name by assign this argument,
+            it should be a list of all label names.
+        category_id: An integer, index of the class.
+            Default is the category with the highest score in the prediction.
+
+    Return:
+        A heatmap ndarray(without color).
+    """
+    img_tensor = np.expand_dims(img, axis=0)
+
+    conv_layer = model.get_layer(layer_name)
+    heatmap_model = Model([model.inputs], [conv_layer.output, model.output])
+
+    with tf.GradientTape() as gtape1:
+        with tf.GradientTape() as gtape2:
+            with tf.GradientTape() as gtape3:
+                conv_output, predictions = heatmap_model(img_tensor)
+                if category_id is None:
+                    category_id = np.argmax(predictions[0])
+                if label_name is not None:
+                    print(label_name[category_id])
+                output = predictions[:, category_id]
+                conv_first_grad = gtape3.gradient(output, conv_output)
+            conv_second_grad = gtape2.gradient(conv_first_grad, conv_output)
+        conv_third_grad = gtape1.gradient(conv_second_grad, conv_output)
+
+    global_sum = np.sum(conv_output, axis=(0, 1, 2))
+
+    alpha_num = conv_second_grad[0]
+    alpha_denom = conv_second_grad[0]*2.0 + conv_third_grad[0]*global_sum
+    alpha_denom = np.where(alpha_denom != 0.0, alpha_denom, 1e-10)
+
+    alphas = alpha_num/alpha_denom
+    alpha_normalization_constant = np.sum(alphas, axis=(0,1))
+    alphas /= alpha_normalization_constant
+
+    weights = np.maximum(conv_first_grad[0], 0.0)
+
+    deep_linearization_weights = np.sum(weights*alphas, axis=(0,1))
+    grad_cam_map = np.sum(deep_linearization_weights*conv_output[0], axis=2)
+
+    heatmap = np.maximum(grad_cam_map, 0)
+    max_heat = np.max(heatmap)
+    if max_heat == 0:
+        max_heat = 1e-10
+    heatmap /= max_heat
+
+    return heatmap
+
+
+
+def preprocess_image(img_path, image_size=(512, 512, 3)):
+    """Preprocess the image by reshape and normalization.
+
+    Args:
+        img_path: A string.
+        target_size: A tuple, reshape to this size.
+    Return:
+        An image array.
+    """
+    # Read the image from the specified path
+    #img = cv2.imread(img_path)
+    # Convert the image from BGR to RGB
+    #img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
+    #img_array=cv2.resize(img, (image_size[0],image_size[1]), fx=1, fy=1,interpolation = cv2.INTER_CUBIC)        
+
+    img = image.load_img(img_path, target_size=image_size)
+    img = image.img_to_array(img)
+    
+    return img
+
+def show_GradCAM(img, heatmap, alpha=0.4, save_path=None, return_array=False):
+    """Show the image with heatmap.
+
+    Args:
+        img_path: string.
+        heatmap: image array, get it by calling grad_cam().
+        alpha: float, transparency of heatmap.
+        return_array: bool, return a superimposed image array or not.
+    Return:
+        None or image array.
+    """
+
+    # Resize the heatmap to match the original image dimensions
+    heatmap = cv2.resize(heatmap, (img.shape[1], img.shape[0]))
+
+    # Apply color map to the heatmap
+    heatmap = (heatmap * 255).astype("uint8")
+    heatmap_colored = cv2.applyColorMap(heatmap, cv2.COLORMAP_JET)
+
+    # Create superimposed image
+    superimposed_img = heatmap_colored * alpha + img
+    superimposed_img = np.clip(superimposed_img, 0, 255).astype("uint8")
+
+    # Create the combined plot
+    fig, axes = plt.subplots(1, 3, figsize=(12, 4))
+
+    # Remove space around subplots
+    fig.subplots_adjust(wspace=0, hspace=0)
+
+    # Original Image
+    axes[0].imshow(img)
+    axes[0].set_title('Original Image')
+    axes[0].axis('off')
+
+    # Heatmap
+    axes[1].imshow(heatmap_colored)
+    axes[1].set_title('Heatmap')
+    axes[1].axis('off')
+
+    # Superimposed Image
+    axes[2].imshow(superimposed_img)
+    axes[2].set_title('Superimposed Image')
+    axes[2].axis('off')
+
+    # Adjust layout
+    plt.tight_layout()
+
+    # plt.show()
+
+    # Save the figure if save_path is provided
+    if save_path:
+        fig.savefig(save_path)
+        logging.info(f"Saved combined visualization to {save_path}")
+        # cv2.imwrite(save_path, superimposed_img)
+
+    # Return superimposed image if return_array is True
+    if return_array:
+        return superimposed_img

gcg/components/model.py

@@ -0,0 +1,323 @@
+from keras.layers import Conv1D, Conv2D, Conv3D
+from keras.layers import Conv2D, LayerNormalization, Layer
+from tensorflow.keras.layers import Reshape, Activation, Softmax, Permute, Add, Dot
+from keras.optimizers import RMSprop
+from tensorflow.keras import layers, models, regularizers
+from tensorflow.keras.applications import EfficientNetV2B0
+from tensorflow.keras.callbacks import ModelCheckpoint
+from keras import ops
+import tensorflow as tf
+import numpy as np
+from sklearn.metrics import accuracy_score,precision_score,recall_score,f1_score,cohen_kappa_score,roc_auc_score,classification_report
+from gcg import config
+from gcg.utils import CustomException, logging
+import sys
+
+class GlobalContextAttention(tf.keras.layers.Layer):
+    def __init__(self, reduction_ratio=8, transform_activation='linear', **kwargs):
+        """
+        Initializes the GlobalContextAttention layer.
+
+        Args:
+            reduction_ratio (int): Reduces the input filters by this factor for the
+                                  bottleneck block of the transform submodule.
+            transform_activation (str): Activation function to apply to the output
+                                        of the transform block.
+            **kwargs: Additional keyword arguments for the Layer class.
+        """
+        super(GlobalContextAttention, self).__init__(**kwargs)
+        self.reduction_ratio = reduction_ratio
+        self.transform_activation = transform_activation
+
+    def build(self, input_shape):
+        """
+        Builds the layer by initializing weights and sub-layers.
+
+        Args:
+            input_shape: Shape of the input tensor.
+        """
+        self.channel_dim = 1 if tf.keras.backend.image_data_format() == 'channels_first' else -1
+        self.rank = len(input_shape)
+
+        # Validate input rank
+        if self.rank not in [3, 4, 5]:
+            raise ValueError('Input dimension has to be either 3 (temporal), 4 (spatial), or 5 (spatio-temporal)')
+
+        # Calculate the number of channels
+        self.channels = input_shape[self.channel_dim]
+
+        # Initialize sub-layers
+        self.conv_context = self._convND_layer(1)  # Context modelling block
+        self.conv_transform_bottleneck = self._convND_layer(self.channels // self.reduction_ratio)  # Transform bottleneck
+        self.conv_transform_output = self._convND_layer(self.channels)  # Transform output block
+
+        # Softmax and Dot layers
+        self.softmax = Softmax(axis=self._get_flat_spatial_dim())
+        self.dot = Dot(axes=(1, 1))  # Dot product over the flattened spatial dimensions
+
+        # Activation layers
+        self.activation_relu = Activation('relu')
+        self.activation_transform = Activation(self.transform_activation)
+
+        # Add layer for final output
+        self.add = Add()
+
+        super(GlobalContextAttention, self).build(input_shape)
+
+    def call(self, inputs):
+        """
+        Performs the forward pass of the layer.
+
+        Args:
+            inputs: Input tensor.
+
+        Returns:
+            Output tensor with global context attention applied.
+        """
+        # Context Modelling Block
+        input_flat = self._spatial_flattenND(inputs)  # [B, spatial_dims, C]
+        context = self.conv_context(inputs)  # [B, spatial_dims, 1]
+        context = self._spatial_flattenND(context)  # [B, spatial_dims, 1]
+        context = self.softmax(context)  # Apply softmax over spatial_dims
+        context = self.dot([input_flat, context])  # [B, C, 1]
+        context = self._spatial_expandND(context)  # [B, C, 1, 1, ...]
+
+        # Transform Block
+        transform = self.conv_transform_bottleneck(context)  # [B, C // R, 1, 1, ...]
+        transform = self.activation_relu(transform)
+        transform = self.conv_transform_output(transform)  # [B, C, 1, 1, ...]
+        transform = self.activation_transform(transform)
+
+        # Apply context transform
+        out = self.add([inputs, transform])  # [B, spatial_dims, C]
+
+        return out
+
+    def _convND_layer(self, filters):
+        """
+        Creates a Conv1D, Conv2D, or Conv3D layer based on the input rank.
+
+        Args:
+            filters (int): Number of filters for the convolutional layer.
+
+        Returns:
+            A Conv1D, Conv2D, or Conv3D layer.
+        """
+        if self.rank == 3:
+            return Conv1D(filters, kernel_size=1, padding='same', use_bias=False, kernel_initializer='he_normal')
+        elif self.rank == 4:
+            return Conv2D(filters, kernel_size=1, padding='same', use_bias=False, kernel_initializer='he_normal')
+        elif self.rank == 5:
+            return Conv3D(filters, kernel_size=1, padding='same', use_bias=False, kernel_initializer='he_normal')
+
+    def _spatial_flattenND(self, ip):
+        """
+        Flattens the spatial dimensions of the input tensor.
+
+        Args:
+            ip: Input tensor.
+
+        Returns:
+            Flattened tensor.
+        """
+        if self.rank == 3:
+            return ip  # Identity op for rank 3
+        else:
+            shape = (ip.shape[self.channel_dim], -1) if self.channel_dim == 1 else (-1, ip.shape[-1])
+            return Reshape(shape)(ip)
+
+    def _spatial_expandND(self, ip):
+        """
+        Expands the spatial dimensions of the input tensor.
+
+        Args:
+            ip: Input tensor.
+
+        Returns:
+            Expanded tensor.
+        """
+        if self.rank == 3:
+            return Permute((2, 1))(ip)  # Identity op for rank 3
+        else:
+            shape = (-1, *(1 for _ in range(self.rank - 2))) if self.channel_dim == 1 else (*(1 for _ in range(self.rank - 2)), -1)
+            return Reshape(shape)(ip)
+
+    def _get_flat_spatial_dim(self):
+        """
+        Returns the axis for flattening spatial dimensions.
+
+        Returns:
+            Axis for flattening.
+        """
+        return 1 if self.channel_dim == 1 else -1
+
+    def get_config(self):
+        """
+        Returns the configuration of the layer for serialization.
+
+        Returns:
+            A dictionary containing the layer configuration.
+        """
+        config = super(GlobalContextAttention, self).get_config()
+        config.update({
+            'reduction_ratio': self.reduction_ratio,
+            'transform_activation': self.transform_activation,
+        })
+        return config
+    
+
+class AttentionGate(Layer):
+    def __init__(self, filters, **kwargs):
+        self.filters = filters
+        super(AttentionGate, self).__init__(**kwargs)
+
+    def build(self, input_shape):
+        # Create trainable parameters for attention gate
+        self.conv_xl = Conv2D(self.filters, kernel_size=(1, 1), strides=(1, 1), padding='same', activation='relu')
+        self.conv_g = Conv2D(self.filters, kernel_size=(1, 1), strides=(1, 1), padding='same', activation='relu')
+        self.psi = Conv2D(1, kernel_size=(1, 1), strides=(1, 1), padding='same', activation='linear')
+        self.layer_norm = LayerNormalization(axis=-1)
+
+        # Build the child layers
+        self.conv_xl.build(input_shape[0])  # Build conv_xl with the shape of xl
+        self.conv_g.build(input_shape[1])   # Build conv_g with the shape of g
+        self.psi.build(input_shape[0])      # Build psi with the shape of xl
+        self.layer_norm.build(input_shape[0])  # Build layer_norm with the shape of xl
+
+        # Add trainable weights
+        self.bxg = self.add_weight(name='bxg',
+                                  shape=(self.filters,),
+                                  initializer='zeros',
+                                  trainable=True)
+        self.bpsi = self.add_weight(name='bpsi',
+                                   shape=(1,),
+                                   initializer='zeros',
+                                   trainable=True)
+
+        super(AttentionGate, self).build(input_shape)
+
+    def call(self, inputs):
+        xl, g = inputs
+
+        # Apply convolutional operations
+        xl_conv = self.conv_xl(xl)
+        g_conv = self.conv_g(g)
+
+        # Compute additive attention
+        att = tf.keras.backend.relu(xl_conv + g_conv + self.bxg)
+        att = self.layer_norm(att)  # Add LayerNormalization
+        att = self.psi(att) + self.bpsi
+        att = tf.keras.backend.sigmoid(att)
+
+        # Apply attention gate
+        x_hat = att * xl
+
+        return x_hat
+    
+    def compute_output_shape(self, input_shape):
+        return input_shape[0]
+
+    def get_config(self):
+        config = super(AttentionGate, self).get_config()
+        config.update({'filters': self.filters})
+        return config
+    
+
+class GCRMSprop(RMSprop):
+    def get_gradients(self, loss, params):
+        # We here just provide a modified get_gradients() function since we are trying to just compute the centralized gradients.
+
+        grads = []
+        gradients = super().get_gradients()
+        for grad in gradients:
+            grad_len = len(grad.shape)
+            if grad_len > 1:
+                axis = list(range(grad_len - 1))
+                grad -= ops.mean(grad, axis=axis, keep_dims=True)
+            grads.append(grad)
+
+        return grads
+
+# Build the model
+def build_model(input_shape, num_classes):
+    try:
+        logging.info("Loading weights of EfficientNetV2B0...")
+        base_model = EfficientNetV2B0(weights='imagenet', include_top=False, input_shape=input_shape)
+
+        fmaps = base_model.output
+
+        logging.info("Initializing Global Context Attention...")
+        context_fmaps = GlobalContextAttention()(fmaps)
+
+        logging.info("Initializing AttentionGate...")
+        att_fmaps = AttentionGate(fmaps.shape[-1])([fmaps, context_fmaps])
+
+        x = layers.GlobalAveragePooling2D()(att_fmaps)
+
+        # First Dense Layer
+        x = layers.Dense(512, activation='relu', kernel_regularizer=regularizers.l2(0.005))(x)
+        x = layers.BatchNormalization()(x)
+        x = layers.Dropout(0.3)(x)
+
+        # Second Dense Layer
+        x = layers.Dense(256, activation='relu', kernel_regularizer=regularizers.l2(0.005))(x)
+        x = layers.BatchNormalization()(x)
+        x = layers.Dropout(0.2)(x)
+
+        output = layers.Dense(num_classes, activation='softmax')(x)
+        
+        model = models.Model(inputs=base_model.input, outputs=output)
+        model.compile(optimizer=GCRMSprop(learning_rate=1e-4), loss='categorical_crossentropy', metrics=['accuracy'])
+
+        logging.info("Model Built Successfully!")
+
+        return model
+    
+    except Exception as e:
+        raise CustomException(e, sys)
+
+# Train the model
+def train_model(model, X_train, X_test, y_train, y_test):
+
+    try:
+        # Define the necessary callbacks
+        checkpoint = ModelCheckpoint(config.model_path, monitor='val_accuracy', verbose=1, save_best_only=True, mode='max')
+
+        callbacks = [checkpoint]
+
+        logging.info(f"Training network for {config.EPOCHS} epochs...")
+        hist = model.fit(X_train, y_train, batch_size=config.batch_size,
+                        validation_data=(X_test, y_test),
+                        epochs=config.EPOCHS, callbacks=callbacks)
+        
+        return hist
+    
+    except Exception as e:
+        raise CustomException(e, sys)
+    
+# Evaluate the model
+def evaluate_model(model, X_test, y_test):
+    try:
+        
+        y_score = model.predict(X_test)
+        y_pred = np.argmax(y_score, axis=-1)
+        Y_test = np.argmax(y_test, axis=-1)
+
+        acc = accuracy_score(Y_test,y_pred)
+        mpre = precision_score(Y_test,y_pred,average='macro')
+        mrecall = recall_score(Y_test,y_pred,average='macro')
+        mf1 = f1_score(Y_test,y_pred,average='macro')
+        kappa = cohen_kappa_score(Y_test,y_pred,weights='quadratic')
+        auc = roc_auc_score(Y_test, y_score, average='macro', multi_class='ovr')
+
+        logging.info(f"Accuracy: {round(acc*100,2)}")
+        logging.info(f"Macro Precision: {round(mpre*100,2)}")
+        logging.info(f"Macro Recall: {round(mrecall*100,2)}")
+        logging.info(f"Macro F1-Score: {round(mf1*100,2)}")
+        logging.info(f"Quadratic Kappa Score: {round(kappa*100,2)}")
+        logging.info(f"ROC AUC Score: {round(auc*100,2)}")
+        logging.info(classification_report(Y_test, y_pred, digits=4))
+
+    except Exception as e:
+        raise CustomException(e, sys)
+    

gcg/config.py

@@ -0,0 +1,30 @@
+import os
+
+# Get the absolute path to the root directory (where gcg, test_images, etc. are located)
+ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+# Dataset details
+data_path = "/Users/tejacherukuri/TReNDS/MyResearch/Datasets/Zenodo-DR7"
+labels = ['No DR', 'Mild NPDR', 'Moderate NPDR', 'Severe NPDR', 'Very Severe NPDR', 'PDR', 'Advanced PDR']
+image_size = (512,512,3)
+num_classes = 7
+labelencoder_save_path = os.path.join(ROOT_DIR, 'saves', 'labelencoder.pkl')
+heatmaps_save_path = os.path.join(ROOT_DIR, 'heatmaps')
+
+# Training Parameters
+EPOCHS = 100
+batch_size = 32
+model_path = os.path.join(ROOT_DIR, 'saves', 'gcg.weights.keras')
+
+# Attention Layer to get features
+gcg_layer_name = 'attention_gate'
+
+#Inference
+test_images = [
+    os.path.join(ROOT_DIR, 'test_images', '184_No_DR.jpg'),
+    os.path.join(ROOT_DIR, 'test_images', '198_Moderate_NPDR.jpg'),
+    os.path.join(ROOT_DIR, 'test_images', '440_Severe_NPDR.jpg'),
+    os.path.join(ROOT_DIR, 'test_images', '500_Very_Severe_NPDR.jpg'),
+    os.path.join(ROOT_DIR, 'test_images', '635_PDR.jpg'),
+    os.path.join(ROOT_DIR, 'test_images', '705_Advanced_PDR.jpg')
+]

gcg/pipelines/__init__.py

@@ -0,0 +1 @@
+from .inference import predict

gcg/pipelines/evaluate.py

@@ -0,0 +1,22 @@
+'''
+This file is used only for evaluation without the need to train the model
+i.e Loads the best model and evaluates on your test dataset
+'''
+
+from gcg.components import load_data, build_model, evaluate_model
+from gcg.utils import load_from_checkpoint, logging
+from gcg import config
+
+logging.info("Initiated evaluation pipeline")
+
+logging.info(f"Loading data from {config.data_path}")
+X_train, X_test, y_train, y_test = load_data(config.data_path, config.image_size)
+
+logging.info("Building model...")
+model = build_model(input_shape=config.image_size, num_classes=7)
+
+logging.info("Loading the model checkpoint...")
+model = load_from_checkpoint(model, config.model_path)
+
+logging.info("Evaluating the model on test set...")
+evaluate_model(model, X_test, y_test)

gcg/pipelines/inference.py

@@ -0,0 +1,60 @@
+'''
+    This file is used for inference on new test samples.
+    One sample or mutiple samples can be fed to the predict() method
+'''
+import os
+import sys
+from typing import List
+import numpy as np
+from gcg import config
+from gcg.utils import logging, CustomException, load_from_checkpoint, load_object
+from gcg.components import build_model, preprocess_image, grad_cam_plus, show_GradCAM
+
+def predict(img_paths:List):
+    '''
+    Inputs: List of image paths
+    Output: Predictions List and Generates heatmaps
+    '''
+    try:
+        predictions_list = []
+        # Step 1: Building the model
+        model = build_model(input_shape=config.image_size, num_classes=config.num_classes)
+
+        # Step 2: Loading the model from checkpoint     
+        logging.info("Loading the model from checkpoint...")
+        model = load_from_checkpoint(model, config.model_path)
+
+        # Step 3: Loading the label encoder to decode indices
+        le = load_object(config.labelencoder_save_path)
+
+        for img_path in img_paths:
+            # Step 4: Read and preprocess the image
+            img_name = img_path.split("/")[-1]
+
+            resized_img = preprocess_image(img_path, config.image_size)
+            img_array = np.expand_dims(resized_img, axis=0)
+            
+            # Step 5: Inference on the model
+            logging.info("Getting your prediction...")
+            pred = model.predict(img_array)
+            predicted_class = np.argmax(pred, axis=1)[0]  # Get class index
+            
+            logging.info(f"Prediction: {le.classes_[predicted_class]}, Path: {img_name}")
+            predictions_list.append(le.classes_[predicted_class])
+            
+            # Step 6: Generating heatmap for the image
+            logging.info("Generating the heatmap using GradCAM++")
+            heatmap_plus = grad_cam_plus(model, resized_img, config.gcg_layer_name, label_name=config.labels, category_id=predicted_class)
+
+            os.makedirs(config.heatmaps_save_path, exist_ok=True) 
+            heatmap_img = config.heatmaps_save_path + f'/heatmap_{img_name}'
+            show_GradCAM(resized_img, heatmap_plus, save_path=heatmap_img)
+
+        return predictions_list
+
+    except Exception as e:
+        raise CustomException(e, sys)
+    
+if __name__=='__main__':
+    predictions_list = predict(config.test_images)
+    print(predictions_list)

gcg/pipelines/train.py

@@ -0,0 +1,22 @@
+'''
+This file is used for end to end training from scratch and then, evaluation.
+'''
+
+from gcg.components import load_data, build_model, train_model, evaluate_model
+from gcg import config
+from gcg.utils import logging
+
+logging.info("Initiated train pipeline")
+
+logging.info(f"Loading data from {config.data_path}")
+X_train, X_test, y_train, y_test = load_data(config.data_path, config.image_size)
+
+logging.info("Building model...")
+model = build_model(input_shape=config.image_size, num_classes=7)
+
+logging.info("Training the model...")
+train_model(model, X_train, X_test, y_train, y_test)
+
+logging.info("Evaluating the model on test set...")
+evaluate_model(model, X_test, y_test)
+

gcg/utils/__init__.py

@@ -0,0 +1,3 @@
+from .exception import CustomException
+from .logger import logging
+from .utils import load_from_checkpoint, load_object, save_object

gcg/utils/exception.py

@@ -0,0 +1,21 @@
+import sys
+
+# Defining Detailed Error Message
+def error_message_detail(error, error_detail:sys):
+    _, _, exc_tb = error_detail.exc_info()
+    file_name = exc_tb.tb_frame.f_code.co_filename
+    error_message = "Error occured in python script name [{0}] line number [{1}] error message[{2}]".format(
+        file_name,
+        exc_tb.tb_lineno,
+        str(error)
+    )
+    return error_message
+
+# Defining CustomException class
+class CustomException(Exception):
+    def __init__(self, error_message, error_detail:sys):
+        super().__init__(error_message)
+        self.error_message=error_message_detail(error_message, error_detail=error_detail)
+    
+    def __str__(self):
+        return self.error_message

gcg/utils/logger.py

@@ -0,0 +1,20 @@
+import logging
+import os
+from datetime import datetime
+
+LOG_FILE=f"{datetime.now().strftime('%m_%d_%Y_%H_%M_%S')}.log"
+logs_path=os.path.join(os.getcwd(),"logs")
+os.makedirs(logs_path,exist_ok=True)
+
+# Create the full log file path
+log_file_path = os.path.join(logs_path, LOG_FILE)
+
+# Set up logging configuration
+logging.basicConfig(
+    level=logging.INFO,
+    format="[ %(asctime)s ] - %(filename)s - %(lineno)d - %(name)s - %(levelname)s - %(message)s",
+    handlers=[
+        logging.FileHandler(log_file_path),  # Save logs to a file
+        logging.StreamHandler()  # Display logs in the console
+    ]
+)

gcg/utils/utils.py

@@ -0,0 +1,24 @@
+import sys
+import joblib
+from gcg.utils import logging, CustomException
+
+def load_from_checkpoint(model, model_path):
+    try:
+        model.load_weights(model_path)
+        return model
+    except Exception as e:
+        raise CustomException(e, sys)
+    
+def save_object(file_path, object):
+    try:
+        logging.info("Serializing the object as pickle file")
+        joblib.dump(object, file_path)
+    except Exception as e:
+        raise CustomException(e, sys)
+    
+def load_object(file_path):
+    try:
+        logging.info("Deserializing the pickle file as object")
+        return joblib.load(file_path)
+    except Exception as e:
+        raise CustomException(e, sys)

requirements.txt

@@ -0,0 +1,8 @@
+streamlit
+tensorflow==2.18.0
+keras==3.8.0
+opencv-python==4.11.0.86
+matplotlib==3.10.0
+numpy==2.0.2
+scikit-learn==1.6.1
+-e .

setup.py

@@ -0,0 +1,22 @@
+from setuptools import setup, find_packages
+from typing import List
+
+HYPEN_E_DOT='-e .'
+def get_requirements(filepath:str) -> List[str]:
+    with open(filepath) as file:
+        requirements = file.readlines()
+        requirements = [req.replace("\n", "") for req in requirements]
+    if HYPEN_E_DOT in requirements:
+        requirements.remove(HYPEN_E_DOT)
+
+    return requirements
+
+setup(
+    name='Guided Context Gating Attention',
+    version='0.1.0',
+    description='Diabetic Retinopathy Classification based on Attention',
+    author='Teja Cherukuri',
+    author_email='[email protected]',
+    packages=find_packages(),
+    install_requires=get_requirements('requirements.txt')
+)