Flask API for Multi-class classification using Keras

Flask API for Multi-class classification using Keras

How to create a production-ready API using Flask for a deep learning model

In this article, you will learn how to create a modularized production-ready library for a deep learning model using the Fashion MNIST dataset built using Keras and Flask.

What are the key features to have a production-ready code for a deep learning model?

Features of production-ready deep learning code

• Exception Handling to monitor errors and to understand the flow of code in case of errors.
• Logging can be set to different levels like debug, info, warning, error, or critical . In production, we should be set the logging level to log warnings, errors, and critical information only.
• Version control for code using GitLab
• Code comments are very vital to understand the code
• Code optimization for efficient memory usage and computation
• Containerize the deep learning model code and all its dependent libraries.

We will learn about Docker and how to containerize the deep learning model in a different article.

Create Config class

We need to parametrize some common attributes used for fine-tuning deep learning models and can be done using a Config class. We also add parameters subjected to change based on input data so that we can keep the model genric.

• Input dataset related parameters : image dimensions — height and width, dataset size, and labels of the classes to be identified
• Model fine-tuning parameters : optimizers, learning rate for the optimizer, no. of epochs and batch_size. We can also include drop out rate, no. of input nodes, no. of hidden layers, etc.
• Files to store information : Weight file for saving trained weights and log filename for logging

You can add/remove config parameters based on your requirements. These config parameters are used across all different operations in a deep learning model.

class Config(object):

IMAGE_HEIGHT=28
 IMAGE_WIDTH=28
 DATA_SIZE=1000
 CLASS_NAME='T-shirt/top', 'Trouser', 'Pullover', 'Dress', 'Coat','Sandal', 'Shirt', 'Sneaker', 'Bag', 'Ankle boot'
WEIGHT_FILENAME='FashionMNIST.h5'
 LOG_FILENAME='LOG_FASHION.TXT'
 EPOCHS=100
 OPTIMIZER='RMSProp'
 LEARNING_RATE=0.001
 BATCH_SIZE=64

def __init__(self):
 self.IMAGE_DIM = (self.IMAGE_HEIGHT, self.IMAGE_WIDTH)

Build a Classification Model

Create a class classFashionMNIST for handling different aspects of the deep learning model. The class will have methods to

• Normalize the data
• Build a deep learning model
• Train the model
• Make predictions using the model
• Display the image from the dataset
• Find the actual class of the image from the dataset

Logging and Exception handling is an integral part of the code to make it robust.

The init() is for setting parameters that we need for different operations of the class, like the height, width for the image, dataset size, and the different classes we want to predict. The init() is also used for reading the Fashion MNIST dataset for training as well as validation,

def __init__(self, height, width, data_size, class_name):
 try:
 self.height= height
 self.width=width
 self.data_size=data_size
 self.class_names =list(class_name)
 (self.train_images, self.train_labels), (self.test_images, self.test_labels) = tf.keras.datasets.fashion_mnist.load_data()
 self.test_data= self.test_images
 except:
 logging.error("Error in init %s", sys.exc_info())

Normalize the dataset

As the pixel intensity in the images are between 1–255, normalize the images by scaling the values between 0 and 1

def normalize_data(self):
try:
logging.info("Normalizing data")

# load train and test images and labels based on data size
 self.train_labels = self.train_labels[:self.data_size]
 self.test_labels = self.test_labels[:self.data_size]

#Normalize the data
 self.train_images = self.train_images[:self.data_size].astype('float32') / 255
 self.test_images = self.test_images[:self.data_size].astype('float32') / 255
 logging.info("Rshaping data")
# Reshape the data
 self.train_images = self.train_images.reshape((self.train_images.shape[0], self.width, self.height,1))
 self.test_images = self.test_images.reshape((self.test_images.shape[0], self.width, self.height,1))
except:
 logging.error("Error", sys.exc_info())

Create a deep learning classification model

We pass the optimizer and the learning rate set in the configuration file for compiling the model. As the deep learning model is a multi-class classification, the loss function used is sparse_categorical_crossentropy. If you are doing a binary classification model, then use binary_crossentropy as the loss function.

def create_model(self, optimizer, learning_rate):
try:
logging.info("Creatig model")
model = tf.keras.Sequential()
 # Must define the input shape in the first layer of the neural network
 model.add(tf.keras.layers.Conv2D(filters=64, kernel_size=2, padding='same', activation='relu', input_shape=(28,28,1))) 
 model.add(tf.keras.layers.MaxPooling2D(pool_size=2))
 model.add(tf.keras.layers.Dropout(0.3))
 model.add(tf.keras.layers.Conv2D(filters=32, kernel_size=2, padding='same', activation='relu'))
 model.add(tf.keras.layers.MaxPooling2D(pool_size=2))
 model.add(tf.keras.layers.Dropout(0.3))
model.add(tf.keras.layers.Flatten())
 model.add(tf.keras.layers.Dense(256, activation='relu'))
 model.add(tf.keras.layers.Dropout(0.5))
 model.add(tf.keras.layers.Dense(10, activation='softmax'))
logging.info("Model Created")
# creating optimizer based on the config
opt= self.get_optimizer(optimizer, learning_rate)

#Compiling the model
model.compile(loss='sparse_categorical_crossentropy',
 optimizer=opt,
 metrics=['accuracy'])
logging.info(" Model Compiled")
 except:
 logging.error(" Error during Model Creation - %s", sys.exc_info())
 finally:
 return model

Setting the optimizer

Handled three popular optimizers: Adam, SGD, and RMSProp . RMSProp is the default optimizer, and the default learning rate is set to 0.001. We can other optimizers like Momentum, Nesterov, Adagrad, Adadelta.

def get_optimizer(self,optimizer_name='RMSProp', learning_rate=0.001):
try:
 if optimizer_name=='Adam': 
 optimizer = optimizers.Adam(learning_rate=learning_rate, beta_1=0.9, beta_2=0.999, amsgrad=False)
 elif optimizer_name=='SGD':
 optimizer = optimizers.SGD(lr=learning_rate, momentum=0.9)
 elif optimizer_name=='RMSProp':
 optimizer = optimizers.RMSprop()
 logging.info("Optimizer created %s", optimizer_name)
 return optimizer
except:
 logging.error(" Error during visualization - %s", sys.exc_info())

Training the Model

Create the model, normalize the data, and finally train the data on train images and train labels.

If the accuracy is less than 0.8 or validation accuracy is less than 0.7, then we are raising a warning in the log file for notifying the team that the model may need retraining.

def train_model(self,filename, epochs, optimizer, learning_rate, batch_size):
 try:
 model = self.create_model(optimizer, learning_rate)
logging.info("Model created ")
 logging.info("Normalizing the data")
 self.normalize_data()
logging.info(self.train_images.shape)
 logging.info("Training started")
 history=model.fit(self.train_images, 
 self.train_labels, 
 batch_size=batch_size,
 epochs=epochs,
 validation_data=(self.test_images,self.test_labels))
logging.info(" Training finished")
 acc= np.average(history.history['acc'])
 val_acc=np.average(history.history['val_acc'])
 logging.info(" Model accurcay on train images : {:5.2f}".format(acc))
 logging.info("Accurcay too low for val {:5.2f}".format(val_acc))
 model.save(filename)
logging.info("Model saved %s", filename)
 if acc <.8 or val_acc<0.7:
logging.warn("Accurcay too low {:5.2f}".format(acc) )
 logging.warn("Accurcay too low for val {:5.2f}".format(val_acc))
 return history, model
 except:
 logging.error(" Error during Model Creation - %s", sys.exc_info())

Predicting the data

To predict the data, we pass the index of the test image and the file that contains the trained weights.

def predict_data(self, test_image_num, filename):
try:
 logging.info("Predciting the data for %d", test_image_num)
 test_img = self.test_images[test_image_num].reshape((1, self.width, self.height,1))
 test_img=test_img.astype('float32') / 255
 model = tf.keras.models.load_model(filename)
logging.info("Loaded the trained weights from %s", filename)
 pred= model.predict(test_img)
 pred= np.argmax(pred)
 logging.info("Predicted class %s",self.class_names[pred] )
 return self.class_names[pred]
 except:
 logging.error(" Error during Model predcition - %s", sys.exc_info())

Actual class of the test image

As we have used FashoinMNIST, we know the class of the test images that we can use for comparing the output

def actual_data(self,test_image_num):
 return self.class_names[self.test_labels[test_image_num]]

Full code for class classFashionMNIST

#Importing required libraries
import tensorflow as tf
import numpy as np
import matplotlib.pyplot as plt
import sys
import logging
from tensorflow.keras import optimizers# setting the random seed
np.random.seed(1)
tf.compat.v1.set_random_seed(1)class classFashionMNIST:

'''
 Method Name: init
 Functionality: initializes the class
 Parameters: sets the height, width of the image, data size and class labels
 '''
def __init__(self, height, width, data_size, class_name):
try:
 self.height= height
 self.width=width
 self.data_size=data_size
 self.class_names =list(class_name)
 (self.train_images, self.train_labels), (self.test_images, self.test_labels) = tf.keras.datasets.fashion_mnist.load_data()
 self.test_data= self.test_images
except:
 logging.error("Error in init %s", sys.exc_info())

'''
 Method Name: normalize data
 Functionality: Normalizes the images pixel intensity values by
 scaling pixel values to the range 0-1 to centering and 
 even standardizing the values.
 Parameters: None
 '''
def normalize_data(self):
try:
 logging.info("Normalizing data")

 # load train and test images and labels based on data size
 self.train_labels = self.train_labels[:self.data_size]
 self.test_labels = self.test_labels[:self.data_size]

 #Normalize the data
 self.train_images = self.train_images[:self.data_size].astype('float32') / 255
 self.test_images = self.test_images[:self.data_size].astype('float32') / 255
 logging.info("Rshaping data")
 # Reshape the data
 self.train_images = self.train_images.reshape((self.train_images.shape[0], self.width, self.height,1))
 self.test_images = self.test_images.reshape((self.test_images.shape[0], self.width, self.height,1))
except:
 logging.error("Error", sys.exc_info())'''
 Method Name: create_mode
 Functionality: Creates the deep learning model for multiclass classification
 Parameters: optimizer - optimizers can be Adam, SGD or RMSProp
 Learning_rate- learning rate of the optimizer
 '''def create_model(self, optimizer, learning_rate):
try:
logging.info("Creatig model")
 model = tf.keras.Sequential()
 # Must define the input shape in the first layer of the neural network
 model.add(tf.keras.layers.Conv2D(filters=64, kernel_size=2, padding='same', activation='relu', input_shape=(28,28,1))) 
 model.add(tf.keras.layers.MaxPooling2D(pool_size=2))
 model.add(tf.keras.layers.Dropout(0.3))
 model.add(tf.keras.layers.Conv2D(filters=32, kernel_size=2, padding='same', activation='relu'))
 model.add(tf.keras.layers.MaxPooling2D(pool_size=2))
 model.add(tf.keras.layers.Dropout(0.3))
 model.add(tf.keras.layers.Flatten())
 model.add(tf.keras.layers.Dense(256, activation='relu'))
 model.add(tf.keras.layers.Dropout(0.5))
 model.add(tf.keras.layers.Dense(10, activation='softmax'))
logging.info("Model Created")
 # creating optimizer based on the config
 opt= self.get_optimizer(optimizer, learning_rate)

 #Compiling the model
 model.compile(loss='sparse_categorical_crossentropy',
 optimizer=opt,
 metrics=['accuracy'])
 logging.info(" Model Compiled")
 except:
logging.error(" Error during Model Creation - %s", sys.exc_info())
 finally:
 return model'''
 Method Name: train_model
 Functionality: Trains the deep learning multiclass classification model
 Parameters: filename : File o save the trained weights
 epochs : No. of epcohs to train the model
 optimizer - optimizers can be Adam, SGD or RMSProp
 Learning_rate- learning rate of the optimizer
 Batch_size - batch_size of the dataset to train the model
 ''' 
def train_model(self,filename, epochs, optimizer, learning_rate, batch_size):
try:
 model = self.create_model(optimizer, learning_rate)
logging.info("Model created ")
 logging.info("Normalizing the data")
 self.normalize_data()
logging.info(self.train_images.shape)
 logging.info("Training started")
 history=model.fit(self.train_images, 
 self.train_labels, 
 batch_size=batch_size,
 epochs=epochs,
 validation_data=(self.test_images,self.test_labels))
logging.info(" Training finished")
 acc= np.average(history.history['acc'])
 val_acc=np.average(history.history['val_acc'])
logging.info(" Model accurcay on train images : {:5.2f}".format(acc))
 logging.info("Accurcay too low for val {:5.2f}".format(val_acc))
 model.save(filename)
logging.info("Model saved %s", filename)
 if acc <.8 or val_acc<0.7:
logging.warn("Accurcay too low {:5.2f}".format(acc) )
 logging.warn("Accurcay too low for val {:5.2f}".format(val_acc))
 return history, model
except:
 logging.error(" Error during Model Creation - %s", sys.exc_info())'''
 Method Name: predict_data
 Functionality: predicts the data for multiclass classification model
 Parameters: test_image_num - index of the test image that we want to predcit 
 filename : File containing the trained weights

 ''' 
def predict_data(self, test_image_num, filename):
try:
 logging.info("Predciting the data for %d", test_image_num)
 test_img = self.test_images[test_image_num].reshape((1, self.width, self.height,1))
 test_img=test_img.astype('float32') / 255
 model = tf.keras.models.load_model(filename)
logging.info("Loaded the trained weights from %s", filename)
 pred= model.predict(test_img)
 pred= np.argmax(pred)
 logging.info("Predicted class %s",self.class_names[pred] )
 return self.class_names[pred]
 except:
 logging.error(" Error during Model predcition - %s", sys.exc_info())'''
 Method Name: actual_data
 Functionality: Retrives the actual class for the test image based on the index passed
 Parameters: test_image_num - index of the test image that we want to predcit 
 '''
 def actual_data(self,test_image_num):
 return self.class_names[self.test_labels[test_image_num]]



 '''
 Method Name: get_optimizer
 Functionality: Creates the optimizers based on passed parameter and learning rate
 Parameters: Optimizer_name - optimizers can be Adam, SGD or RMSProp
 Learning_rate- learning rate of the optimizer
 '''  
def get_optimizer(self,optimizer_name='RMSProp', learning_rate=0.001):
try:
 if optimizer_name=='Adam': 
 optimizer = optimizers.Adam(learning_rate=learning_rate, beta_1=0.9, beta_2=0.999, amsgrad=False)
 elif optimizer_name=='SGD':
 optimizer = optimizers.SGD(lr=learning_rate, momentum=0.9)
 elif optimizer_name=='RMSProp':
 optimizer = optimizers.RMSprop()
logging.info("Optimizer created %s", optimizer_name)
 return optimizer
 except:
 logging.error(" Error during visualization - %s", sys.exc_info())

Create the API using Flask

Read here for a basic understanding of how to create a basic API using Flask

Importing required libraries for using Flask API, classFashionMNIST, Fashion MNIST Config class, and logging class.

Create an application object by creating an instance of Flask to which we pass a predefined variable “__name__”, which is the name of our module.

from flask import Flask, jsonify, request
from flask_restful import Api, Resource
import numpy as np
from Fashion_MNIST import classFashionMNIST
from FashionConfig import Config as cfg
import logging
import absl.loggingapp=Flask(__name__)

We write two GET methods, one for making the prediction and one for retrieving the actual class

@app.route("/predict", methods=["GET"])
def predict():
 pred =""
 posted_data = request.get_json()
 test_image_num=posted_data['test_image_num']
 logging.info("In Predict")
 model_filename=cfg.WEIGHT_FILENAME
 pred= fashionMNISTclass.predict_data(test_image_num, model_filename)
 return jsonify(pred)@app.route("/real", methods=["GET"])
def real():
 data =""
 posted_data = request.get_json()
 test_image_num=posted_data['test_image_num']
 data = fashionMNISTclass.actual_data(test_image_num)
 return jsonify(data)

I have used TF 1.14, and hence we need to disable abseil-py so that the logs are not redirected to stderr. We set the log file based on the log filename specified in the Config parameters

Train the dataset as we load the API, hence wait for the predictions till the model is fully loaded and compiled

if __name__ == '__main__':
logging.root.removeHandler(absl.logging._absl_handler)
absl.logging._warn_preinit_stderr = False
logging.basicConfig(filename=cfg.LOG_FILENAME, filemode='a', format='%(filename)s-%(asctime)s %(msecs)d- %(process)d-%(levelname)s - %(message)s', 
 datefmt='%d-%b-%y %H:%M:%S %p' ,
 level=logging.DEBUG)
 fashionMNISTclass= classFashionMNIST(cfg.IMAGE_HEIGHT, cfg.IMAGE_WIDTH, cfg.DATA_SIZE, cfg.CLASS_NAME)
 # noramlize the data
fashionMNISTclass.normalize_data()
# train the model
 history, model = history, model = fashionMNISTclass.train_model(cfg.WEIGHT_FILENAME, 
 cfg.EPOCHS,
 cfg.OPTIMIZER,
 cfg.LEARNING_RATE,
 cfg.BATCH_SIZE) 
 app.run(debug=True)

Testing the predict method using postman

Checking the actual class name

Full code available here

Conclusion:

You are now all set to write a production-ready code using Keras for binary or multi-class classification models. There are a lot of tweaks that can be done to the code, and some of them are mentioned in the article.