Keras_CV

import numpy as np
import pandas as pd
import tensorflow as tf
import os
from tensorflow.keras.preprocessing.image import ImageDataGenerator
from tensorflow.keras.layers import Conv2D, MaxPool2D, Dense, GlobalAveragePooling2D, Flatten
from tensorflow.keras.models import Sequential
from tensorflow.keras.optimizers import Adam, SGD, RMSprop
from tensorflow.keras.applications.resnet import ResNet50
from tensorflow.keras.losses import MeanAbsoluteError

def load_train(path):

    dataframe = os.path.join(path, 'labels.csv')

    train_datagen = ImageDataGenerator(rescale = 1/255.,
                                      horizontal_flip=True,
                                      vertical_flip=True)

    train_flow = train_datagen.flow_from_dataframe(
        dataframe = pd.read_csv(dataframe),
        directory = os.path.join(path, 'final_files'),
        x_col='file_name',
        y_col='real_age',
        target_size=(224, 224),
        batch_size=32,
        class_mode='raw',
        seed=51,
        shuffle = True)

    return train_flow


def load_test(path):

    dataframe = os.path.join(path, 'labels.csv')
    test_datagen = ImageDataGenerator(rescale = 1/255.)

    test_flow = train_datagen.flow_from_dataframe(
        dataframe = pd.read_csv(dataframe),
        directory = os.path.join(path, 'final_files'),
        x_col='file_name',
        y_col='real_age',
        target_size=(224, 224),
        batch_size=32,
        class_mode='raw',
        shuffle = False)

    return test_flow

def create_model(inp_shape=(224, 224, 3)):

    resnet_50 = ResNet50(input_shape = inp_shape,
                        weights = '/datasets/keras_models/resnet50_weights_tf_dim_ordering_tf_kernels_notop.h5',
                        include_top = False)

    for layer in resnet_50.layers:
        layer.trainable = False

    optimizer = Adam(learning_rate=0.001)

    model = Sequential()
    #model.add(inception)
    model.add(resnet_50)
    model.add(GlobalAveragePooling2D())
    model.add(Dense(128, activation = 'relu'))
    model.add(Dense(1, activation = 'relu'))
    model.compile(optimizer=optimizer, loss='mse',
                  metrics=['mae'])

    return model

def train_model(model, train_data, test_data, batch_size = 32, epochs = 20,
                steps_per_epoch=None, validation_steps=None):

    model.fit(train_data,
              validation_data=test_data,
              steps_per_epoch=steps_per_epoch,
              validation_steps=validation_steps,
              batch_size = batch_size,
              verbose=2,
              shuffle=True,
              epochs = epochs)

    return model