Image_Recognition_using_Machine_Learning.py

 #!/usr/bin/env python
# coding: utf-8

# # Image Recognition Using Tensorflow

import os
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3'
import numpy as np
import time
import queue
import threading
import tensorflow as tf
tf.compat.v1.logging.set_verbosity(tf.compat.v1.logging.ERROR)
import sys
tf.compat.v1.disable_eager_execution()

class image_recognition_using_label_data(object):
    
    def __init__(self):
        # Initialize
        self.label = None
        self.graph = None
        self.load_label("/tmp/retrain_tmp/output_labels.txt")
        
        self.graph_def = None
        self.image_reader = None
        self.caster = None
        self.dimention = None
        self.recognition = None
        self.image_height = 299
        self.image_width = 299
        self.image_mean = 0
        self.image_standard_deviation = 255
        self.detection = None
    
    def image_graph(self, graph):
        self.graph = graph
        
    @property
    def training_graph(self):
        return self.graph
    
    @property
    def training_label(self):
        return self.label
    
    # Loading the Training Machine Learning Model
    def load_label(self, file_label):
        self.label = []
        ascii_lines = tf.io.gfile.GFile(file_label).readlines()
        for size in ascii_lines:
            self.label.append(size.rstrip())
            
    def load_graph(self, file_model):
        self.graph = tf.Graph()
        self.graph_def = tf.compat.v1.GraphDef()
        
        with open(file_model, "rb") as opened:
            self.graph_def.ParseFromString(opened.read())
        with self.graph.as_default():
            tf.import_graph_def(self.graph_def)
        return self.graph
    
    # Image detection
    def image_detection(self, queue, session, bytes_image, image_path, image_input, image_output):
        self.detect_image_bytes(bytes_image)
        result = session.run(image_output.outputs[0], {image_input.outputs[0]: self.detection})
        result = np.squeeze(result)
        
        prediction = result.argsort()[-5:][::-1][0]
        queue.put( {'image_path':image_path, 'prediction':self.label[prediction].title(), 'percent':result[prediction]} )
        
        
    def detect_image_bytes(self, bytes_image):
        image_reader = tf.image.decode_png(bytes_image, channels=3, name="png_reader")
        caster = tf.cast(image_reader, tf.float32)
        dimention = tf.expand_dims(caster, 0)
        
        image_resize = tf.image.resize(dimention,[self.image_height, self.image_width])
        normalize = tf.divide(tf.subtract(image_resize, [self.image_mean]), [self.image_standard_deviation])
        session = tf.compat.v1.Session()
        self.detection = session.run(normalize)
    

def detecting_images():
    image_recognition = image_recognition_using_label_data()
    graph = image_recognition.load_graph('/tmp/retrain_tmp/output_graph.pb')
    image_recognition.image_graph(graph)
    
    image_input = image_recognition.training_graph.get_operation_by_name("import/Placeholder")
    image_output = image_recognition.training_graph.get_operation_by_name("import/final_result")
    session = tf.compat.v1.Session(graph=graph)
    
    # detection_images is the directory
    detect_image = os.listdir('detection_images')
    queue_image = queue.Queue()
    
    for image in detect_image:
        image_path = '{}/{}'.format('detection_images', image)
        print('Image Processing {}'.format(image_path))
        
        # Time laps before processing another image
        while len(threading.enumerate()) > 10:
            time.sleep(0.0001)
        
        # Reading images as byte objects as it is expecting png file
        bytes_image = open(image_path, "rb").read()
        threading.Thread(target = image_recognition.image_detection, args = (queue_image, session, bytes_image, image_path, image_input, image_output)).start()
        
    print('Waiting For Threads to Finish...')
    while queue_image.qsize() < len(detect_image):
        time.sleep(0.001)
        
    prediction_accuracy = [queue_image.get() for i in range(queue_image.qsize())]
        
    for prediction in prediction_accuracy:
        print("Predicted {image_path} is a {prediction} with {percent:.2%} Accuracy".format(**prediction))

        
if __name__ == "__main__":
    detecting_images()