core.py

import keras
from keras.layers import Input, Add, Dense, Activation, ZeroPadding2D, BatchNormalization, Flatten, Conv2D, \
    AveragePooling2D, MaxPooling2D
from keras.models import Model
from keras.initializers import glorot_uniform
from keras.preprocessing import image
from matplotlib.pyplot import imshow
import numpy as np
from load_images import load_images_and_labels


def identity_block(X, f, filters, stage, block):
    """
    Implementation of the identity block

    Arguments:
    X -- input tensor of shape (m, n_H_prev, n_W_prev, n_C_prev)
    f -- integer, specifying the shape of the middle CONV's window for the main path
    filters -- python list of integers, defining the number of filters in the CONV layers of the main path
    stage -- integer, used to name the layers, depending on their position in the network
    block -- string/character, used to name the layers, depending on their position in the network

    Returns:
    X -- output of the identity block, tensor of shape (n_H, n_W, n_C)
    """

    # Defining name basis
    conv_name_base = 'res' + str(stage) + block + '_branch'
    bn_name_base = 'bn' + str(stage) + block + '_branch'

    # Retrieve Filters
    F1, F2, F3 = filters

    # Save the input value
    X_shortcut = X

    # First component of main path
    X = Conv2D(filters=F1, kernel_size=(1, 1), strides=(1, 1), padding='valid',
               name=conv_name_base + '2a', kernel_initializer=glorot_uniform(seed=0))(X)
    X = BatchNormalization(axis=3, name=bn_name_base + '2a')(X)
    X = Activation('relu')(X)

    # Second component of main path
    X = Conv2D(filters=F2, kernel_size=(f, f), strides=(1, 1), padding='same',
               name=conv_name_base + '2b', kernel_initializer=glorot_uniform(seed=0))(X)
    X = BatchNormalization(axis=3, name=bn_name_base + '2b')(X)
    X = Activation('relu')(X)

    # Third component of main path
    X = Conv2D(filters=F3, kernel_size=(1, 1), strides=(1, 1), padding='valid',
               name=conv_name_base + '2c', kernel_initializer=glorot_uniform(seed=0))(X)
    X = BatchNormalization(axis=3, name=bn_name_base + '2c')(X)

    # Final step: Add shortcut value to main path, and pass it through a RELU activation
    X = Add()([X, X_shortcut])
    X = Activation('relu')(X)

    return X


def convolutional_block(X, f, filters, stage, block, s=2):
    """
    Implementation of the convolutional block

    Arguments:
    X -- input tensor of shape (m, n_H_prev, n_W_prev, n_C_prev)
    f -- integer, specifying the shape of the middle CONV's window for the main path
    filters -- python list of integers, defining the number of filters in the CONV layers of the main path
    stage -- integer, used to name the layers, depending on their position in the network
    block -- string/character, used to name the layers, depending on their position in the network
    s -- Integer, specifying the stride to be used

    Returns:
    X -- output of the convolutional block, tensor of shape (n_H, n_W, n_C)
    """

    # Defining name basis
    conv_name_base = 'res' + str(stage) + block + '_branch'
    bn_name_base = 'bn' + str(stage) + block + '_branch'

    # Retrieve Filters
    F1, F2, F3 = filters

    # Save the input value
    X_shortcut = X

    # MAIN PATH
    # First component of main path
    X = Conv2D(filters=F1, kernel_size=(1, 1), strides=(s, s), padding='valid',
               name=conv_name_base + '2a', kernel_initializer=glorot_uniform(seed=0))(X)
    X = BatchNormalization(axis=3, name=bn_name_base + '2a')(X)
    X = Activation('relu')(X)

    # Second component of main path
    X = Conv2D(filters=F2, kernel_size=(f, f), strides=(1, 1), padding='same',
               name=conv_name_base + '2b', kernel_initializer=glorot_uniform(seed=0))(X)
    X = BatchNormalization(axis=3, name=bn_name_base + '2b')(X)
    X = Activation('relu')(X)

    # Third component of main path
    X = Conv2D(filters=F3, kernel_size=(1, 1), strides=(1, 1), padding='valid',
               name=conv_name_base + '2c', kernel_initializer=glorot_uniform(seed=0))(X)
    X = BatchNormalization(axis=3, name=bn_name_base + '2c')(X)

    # SHORTCUT PATH
    X_shortcut = Conv2D(filters=F3, kernel_size=(1, 1), strides=(s, s), padding='valid',
                        name=conv_name_base + '1', kernel_initializer=glorot_uniform(seed=0))(X_shortcut)
    X_shortcut = BatchNormalization(axis=3, name=bn_name_base + '1')(X_shortcut)

    X = Add()([X, X_shortcut])
    X = Activation('relu')(X)

    return X


def ResNet50(input_shape=(64, 64, 3), classes=6):

    X_input = Input(input_shape)

    X = ZeroPadding2D((3, 3))(X_input)

    # Stage 1
    X = Conv2D(64, (7, 7), strides=(2, 2), name='conv1', kernel_initializer=glorot_uniform(seed=0))(X)
    X = BatchNormalization(axis=3, name='bn_conv1')(X)
    X = Activation('relu')(X)
    X = MaxPooling2D((3, 3), strides=(2, 2))(X)

    # Stage 2
    X = convolutional_block(X, f=3, filters=[64, 64, 256], stage=2, block='a', s=1)
    X = identity_block(X, 3, [64, 64, 256], stage=2, block='b')
    X = identity_block(X, 3, [64, 64, 256], stage=2, block='c')

    # For output shape tracking - Stage 2 output
    stage2_output = Conv2D(64, (1, 1), name='stage2_output')(X)
    print(f"Stage 2 output shape: {stage2_output}")

    # Stage 3
    X = convolutional_block(X, f=3, filters=[128, 128, 512], stage=3, block='a', s=2)
    X = identity_block(X, 3, [128, 128, 512], stage=3, block='b')
    X = identity_block(X, 3, [128, 128, 512], stage=3, block='c')
    X = identity_block(X, 3, [128, 128, 512], stage=3, block='d')

    # For output shape tracking - Stage 3 output
    stage3_output = Conv2D(128, (1, 1), name='stage3_output')(X)
    print(f"Stage 3 output shape: {stage3_output}")

    # Stage 4
    X = convolutional_block(X, f=3, filters=[256, 256, 1024], stage=4, block='a', s=2)
    X = identity_block(X, 3, [256, 256, 1024], stage=4, block='b')
    X = identity_block(X, 3, [256, 256, 1024], stage=4, block='c')
    X = identity_block(X, 3, [256, 256, 1024], stage=4, block='d')
    X = identity_block(X, 3, [256, 256, 1024], stage=4, block='e')
    X = identity_block(X, 3, [256, 256, 1024], stage=4, block='f')

    # For output shape tracking - Stage 4 output
    stage4_output = Conv2D(256, (1, 1), name='stage4_output')(X)
    print(f"Stage 4 output shape: {stage4_output}")

    # Stage 5
    X = convolutional_block(X, f=3, filters=[512, 512, 2048], stage=5, block='a', s=2)
    X = identity_block(X, 3, [512, 512, 2048], stage=5, block='b')
    X = identity_block(X, 3, [512, 512, 2048], stage=5, block='c')

    # For output shape tracking - Stage 5 output
    stage5_output = Conv2D(512, (1, 1), name='stage5_output')(X)
    print(f"Stage 5 output shape: {stage5_output}")

    # Average Pooling
    X = AveragePooling2D(pool_size=(2, 2), padding='same')(X)

    # Output layer
    X = Flatten()(X)
    X = Dense(classes, activation='softmax', name='fc' + str(classes), kernel_initializer=glorot_uniform(seed=0))(X)

    # Create model
    model = Model(inputs=X_input, outputs=X, name='ResNet50')

    return model


def convert_to_one_hot(Y, C):
    """
    Converts a class vector (integers) to binary class matrix.

    Arguments:
    Y -- numpy array of shape (number of examples)
    C -- number of classes

    Returns:
    Y -- numpy array of shape (number of examples, number of classes)
    """
    Y = np.eye(C)[Y.reshape(-1)]
    return Y


def load_dataset(path_to_images, do_print = False):
    """
    Placeholder function for loading dataset.
    Replace this with your actual dataset loading code.

    Returns:
    X_train -- training set images
    Y_train -- training set labels
    X_test -- test set images
    Y_test -- test set labels
    classes -- list of class names
    """
    Xs, Ys = load_images_and_labels(path_to_images)
    limit = int(Xs.shape[0]*0.7)
    X_train = Xs[:limit]
    Y_train = Ys[:limit]
    X_test = Xs[limit:]
    Y_test = Ys[limit:]
    classes = ['0', '1', '2', '3', '4', '5']

    X_train = X_train / 255.
    X_test = X_test / 255.

    Y_train =  convert_to_one_hot(Y_train, 6)
    Y_test = convert_to_one_hot(Y_test, 6)

    if do_print:
        print("number of training examples = " + str(X_train.shape[0]))
        print("number of test examples = " + str(X_test.shape[0]))
        print("X_train shape: " + str(X_train.shape))
        print("Y_train shape: " + str(Y_train.shape))
        print("X_test shape: " + str(X_test.shape))
        print("Y_test shape: " + str(Y_test.shape))

    return X_train, Y_train, X_test, Y_test, classes

def classify_image(model, path_to_image):
    img = image.load_img(path_to_image, target_size=(64, 64))
    x = image.img_to_array(img)
    x = np.expand_dims(x, axis=0)
    x = x / 255.0
    x2 = x
    print('Input image shape:', x.shape)
    imshow(img)
    prediction = model.predict(x2)
    print("Class prediction vector [p(0), p(1), p(2), p(3), p(4), p(5)] = ", prediction)
    print("Class:", np.argmax(prediction))

def new_model(path_to_model, print_summary = False):
    model = ResNet50(input_shape=(64, 64, 3), classes=6)
    if print_summary:
        model.summary()
    opt = keras.optimizers.Adam(learning_rate=0.00015)
    model.compile(optimizer=opt, loss='categorical_crossentropy', metrics=['accuracy'])
    model.save(path_to_model)

    print("----> model created <----")

def train_model(path_to_model, path_to_images):
    loaded_model = keras.models.load_model(path_to_model)
    X_train, Y_train, X_test, Y_test, classes = load_dataset(path_to_images)

    loaded_model.fit(X_train, Y_train, epochs=10, batch_size=32)
    eval_ = loaded_model.evaluate(X_test, Y_test)
    print("Loss = " + str(eval_[0]))
    print("Test Accuracy = " + str(eval_[1]))
    loaded_model.save(path_to_model)
    return eval_

def train_():
    acc = 0
    while acc < 0.85:
        res = train_model('resnet50_trained.keras', "images_with_labels")
        acc = res[1]

def predict(path_to_model, path_to_image):
    loaded_model = keras.models.load_model(path_to_model)
    return classify_image(loaded_model, path_to_image)

if __name__ == "__main__":
    predict("resnet50_trained.keras", "./images/image11.png")