In [1]:
# import libraries
import numpy as np
import tensorflow as tf
import keras_tuner
import PIL
import matplotlib.pyplot as plt
import pandas as pd
# Seed the tf, numpy and python with a fix random number
seed = 93
import random # from the python library
random.seed(seed)
np.random.seed(seed)
tf.random.set_seed(seed)
# Set the difficulty level to train
DIFF_LEVEL = "binary" # binary, multi, easy, mid, hard
# Train all the images with binary label
NUM_CLASSES = 1 # number of classes = 5 if multilabel is used
In [2]:
# List the gpu
devices = tf.config.list_physical_devices("GPU")
print(devices)
[PhysicalDevice(name='/physical_device:GPU:0', device_type='GPU')]
In [3]:
# path of the dataset saved from previous step
import os
# path: dataset\train\ds_binary.
# path: dataset\train\ds_multi.
diff_level = f"ds_{DIFF_LEVEL}"
dataset_file = os.path.join("dataset", "train", diff_level)
print(f"Dataset file: {dataset_file}")
if os.path.exists(dataset_file) is False:
print(f"File not found: {dataset_file}")
exit(0)
# Load the dataset
ds = tf.data.Dataset.load(dataset_file)
Dataset file: dataset\train\ds_binary
In [4]:
# print the loaded dataset info
def print_dataset_info(dataset):
print(f"Image arrays: {dataset.element_spec[0].shape}, {dataset.element_spec[0].dtype}")
print(f"Label arrays: {dataset.element_spec[1].shape}, {dataset.element_spec[1].dtype}")
print(f"cardinality: {dataset.cardinality()}")
In [5]:
print_dataset_info(ds)
Image arrays: (224, 224, 3), <dtype: 'float32'> Label arrays: (), <dtype: 'int32'> cardinality: 2028
Prepare dataset¶
In [6]:
# split the dataset into train and validation set
ds_train, ds_val = tf.keras.utils.split_dataset(
ds, left_size=0.9, right_size=0.1, shuffle=True, seed=seed
)
In [7]:
print(f"train set: {ds_train.cardinality()}, \nvalidation set: {ds_val.cardinality()}")
train set: 1825, validation set: 203
In [8]:
# preprocess input
# data augmentation and the pretrained model's preprocess function
img_augmentation_layers = [
tf.keras.layers.RandomRotation(factor=0.15),
tf.keras.layers.RandomTranslation(height_factor=0.1, width_factor=0.1),
]
def img_augmentation(images):
for layer in img_augmentation_layers:
images = layer(images)
return images
# the preprocess function of the pretrained model
pretrained_model_preprocess_fn = tf.keras.applications.efficientnet_v2.preprocess_input
def input_preprocess_train(image, label):
#image = img_augmentation(image)
image = pretrained_model_preprocess_fn(image)
return image, label
# do not augment the data in the validation or test set
def input_preprocess_test(image, label):
image = pretrained_model_preprocess_fn(image)
return image, label
In [9]:
BATCH_SIZE = 16
ds_train = ds_train.shuffle(buffer_size=1000, seed=seed)
ds_train = ds_train.map(lambda x, y: input_preprocess_train(x, y),
num_parallel_calls=tf.data.AUTOTUNE,
deterministic=True)
ds_train = ds_train.cache()
ds_train = ds_train.batch(batch_size=BATCH_SIZE,
num_parallel_calls=tf.data.AUTOTUNE,
deterministic=True,
drop_remainder=True)
ds_train = ds_train.prefetch(tf.data.AUTOTUNE)
In [10]:
ds_val = ds_val.shuffle(buffer_size=1000, seed=seed)
ds_val = ds_val.map(lambda x, y: input_preprocess_test(x, y),
num_parallel_calls=tf.data.AUTOTUNE,
deterministic=True)
ds_val = ds_val.cache()
ds_val = ds_val.batch(batch_size=BATCH_SIZE,
num_parallel_calls=tf.data.AUTOTUNE,
deterministic=True,
drop_remainder=True)
ds_val = ds_val.prefetch(tf.data.AUTOTUNE)
In [11]:
ds_t = ds_train.take(1)
ds_v = ds_val.take(1)
In [12]:
ds_t
Out[12]:
<TakeDataset element_spec=(TensorSpec(shape=(16, 224, 224, 3), dtype=tf.float32, name=None), TensorSpec(shape=(16,), dtype=tf.int32, name=None))>
In [13]:
ds_v
Out[13]:
<TakeDataset element_spec=(TensorSpec(shape=(16, 224, 224, 3), dtype=tf.float32, name=None), TensorSpec(shape=(16,), dtype=tf.int32, name=None))>
In [14]:
print(f"train set: {ds_t.cardinality()}, \nvalidation set: {ds_v.cardinality()}")
train set: 1, validation set: 1
Callbacks layers and optimizer learning rate scheduler¶
In [15]:
# The ModelCheckpoint callback can be used to implement fault-tolerance: the
# ability to restart training from the last saved state of the model in case
# training gets randomly interrupted.
# setup the callbacks
# tensorboard log directory
logdir = "logdir"
checkpoint_dir = "checkpoint"
# check for the directory
if not os.path.exists(logdir):
os.mkdir(logdir)
if not os.path.exists(checkpoint_dir):
os.mkdir(checkpoint_dir)
import time # to format time
model_name = f"best_{DIFF_LEVEL}_{time.strftime('%d_%m_%H%M')}.keras"
checkpoint_filepath = os.path.join(checkpoint_dir, model_name)
print(f"Checkpoint filepath: {checkpoint_filepath}")
callbacks = [
# early stopping
tf.keras.callbacks.EarlyStopping(patience=150),
# Save the best model
tf.keras.callbacks.ModelCheckpoint(
filepath=checkpoint_filepath,
save_weights_only=True,
monitor='val_binary_accuracy',
mode='max',
save_best_only=True
),
# tensorboard
tf.keras.callbacks.TensorBoard(
log_dir=logdir,
update_freq="epoch"
)
]
Checkpoint filepath: checkpoint\best_binary_10_02_1124.keras
Learning rate scheduluer¶
In [16]:
# Not using
# Gradually reduce learning rate
# Use default value
initial_learning_rate = 1e-3
lr_schedule = tf.keras.optimizers.schedules.ExponentialDecay(
initial_learning_rate,
decay_steps=100000,
decay_rate=0.96,
staircase=True
)
lr_schedule
Out[16]:
<keras.optimizers.schedules.learning_rate_schedule.ExponentialDecay at 0x21ab1367340>
In [ ]:
Efficient net¶
In [17]:
# prepare the dataset
def plot_history(history):
plt.plot(history.history["binary_accuracy"])
plt.plot(history.history["val_binary_accuracy"])
plt.title("Model accuracy")
plt.ylabel("accuracy")
plt.xlabel("epoch")
plt.grid(visible=True, axis="both")
plt.legend(["train", "validation"], loc="upper left")
plt.show()
In [18]:
# EfficientNetB7, image resolution: 600 x 600
# Get the image size
IMG_SIZE = ds.element_spec[0].shape[1]
print(f"Image size: {IMG_SIZE}")
Image size: 224
In [19]:
# EfficientNetV2 models expect their inputs to be float tensors of pixels with values in the [0-255] range.
# Use EfficientNetB7
# build_model: Set the range of hyperparameters.
# Then, it will call the exisiting model building code.
def build_model(hp):
pooling = hp.Choice("pooling", ["average", "global"])
norm = hp.Boolean("norm")
units = hp.Int("units", min_value=64, max_value=512, step=32)
dropout = hp.Float("dropout", min_value=0.1, max_value=0.4, step=0.1)
# call the actual model building code
model = call_existing_code(pooling, norm, units, dropout)
return model
# This is called by the keras tuner.
# This is the existing model building code that is working.
def call_existing_code(pooling, norm, units, dropout):
inputs = tf.keras.layers.Input(shape=(IMG_SIZE, IMG_SIZE, 3))
# drop_connect_rate which controls the dropout rate responsible for stochastic depth
# use this for stronger regularization
global base_model
base_model = tf.keras.applications.EfficientNetV2B0(
include_top=False,
input_tensor=inputs,
weights="imagenet",
)
# Freeze the pretrained weights
base_model.trainable = False
# Rebuild top
# pooling=None means that the output of the model will be the
# 4D tensor output of the last convolutional layer.
if pooling == "average":
x = tf.keras.layers.GlobalAveragePooling2D(name="avg_pool")(base_model.output)
elif pooling == "global":
x = tf.keras.layers.GlobalMaxPooling2D(name="max_pool")(base_model.output)
# if normalization is used
if norm is True:
x = tf.keras.layers.BatchNormalization()(x)
# Add a dense layer
# The units are chosen from the hyperparameters
x = tf.keras.layers.Dense(units, activation='relu')(x)
# Drop out
# dropout rate from the hyperparameters
top_dropout_rate = dropout
x = tf.keras.layers.Dropout(
top_dropout_rate,
name="top_dropout",
seed=seed
)(x)
# set the number of classes
num_classes = NUM_CLASSES
outputs = tf.keras.layers.Dense(
num_classes,
activation="sigmoid",
name="predict",
kernel_regularizer="l2",
bias_regularizer="l2",
activity_regularizer="l2",
)(x)
model = tf.keras.Model(inputs, outputs, name="MyEfficient-V2B0")
model = compile_model(model)
return model
def compile_model(model):
# Compile
optimizer = tf.keras.optimizers.Adam(learning_rate=lr_schedule)
metrics = [
tf.keras.metrics.BinaryAccuracy(),
tf.keras.metrics.Precision(name="precision"),
tf.keras.metrics.Recall(name="recall"),
]
model.compile(
optimizer=optimizer,
loss="binary_crossentropy",
metrics=metrics,
)
return model
Keras tuner searche_summary()¶
In [20]:
tuner = keras_tuner.RandomSearch(
hypermodel=build_model,
objective="val_binary_accuracy",
max_trials=10,
executions_per_trial=2,
seed=seed,
overwrite=True,
directory="hp_search",
project_name="all_faces",
)
tuner.search_space_summary()
Search space summary
Default search space size: 4
pooling (Choice)
{'default': 'average', 'conditions': [], 'values': ['average', 'global'], 'ordered': False}
norm (Boolean)
{'default': False, 'conditions': []}
units (Int)
{'default': None, 'conditions': [], 'min_value': 64, 'max_value': 512, 'step': 32, 'sampling': 'linear'}
dropout (Float)
{'default': 0.1, 'conditions': [], 'min_value': 0.1, 'max_value': 0.4, 'step': 0.1, 'sampling': 'linear'}
In [21]:
model = build_model(keras_tuner.HyperParameters())
#model = compile_model(model) # this will be called in the build_model
epochs = 1000
In [22]:
# This will be called by the tuner search
tuner.search(ds_train, epochs=epochs, validation_data=ds_val, callbacks=callbacks)
Trial 10 Complete [01h 26m 10s] val_binary_accuracy: 0.6197916865348816 Best val_binary_accuracy So Far: 0.6770833730697632 Total elapsed time: 18h 22m 19s
In [23]:
# The model weights (that are considered the best) can be loaded as -
#model.load_weights(checkpoint_filepath)
model_list = tuner.get_best_models(num_models=1)
model = model_list[0] # the best model from the tuner
# print a summary of the best trials
tuner.results_summary(num_trials=1)
Results summary Results in hp_search\all_faces Showing 1 best trials Objective(name="val_binary_accuracy", direction="max") Trial 02 summary Hyperparameters: pooling: global norm: False units: 512 dropout: 0.4 Score: 0.6770833730697632
In [35]:
# Save the best model
best_model = model_list[0] # the best model from the tuner
In [34]:
# print a summary of a few moretrials
tuner.results_summary(num_trials=5)
Results summary Results in hp_search\all_faces Showing 5 best trials Objective(name="val_binary_accuracy", direction="max") Trial 02 summary Hyperparameters: pooling: global norm: False units: 512 dropout: 0.4 Score: 0.6770833730697632 Trial 08 summary Hyperparameters: pooling: global norm: True units: 64 dropout: 0.4 Score: 0.6588541865348816 Trial 07 summary Hyperparameters: pooling: average norm: True units: 64 dropout: 0.4 Score: 0.65625 Trial 05 summary Hyperparameters: pooling: average norm: False units: 480 dropout: 0.4 Score: 0.6536458432674408 Trial 01 summary Hyperparameters: pooling: average norm: False units: 256 dropout: 0.1 Score: 0.6484375298023224
In [25]:
# Reference: https://keras.io/api/applications/
# Layers in the base models and models
print(f"Number of layers in base model {base_model.name}: {len(base_model.layers)}")
print(f"Number of layers in feature extraction model {model.name}: {len(model.layers)}")
Number of layers in base model efficientnetv2-b0: 270 Number of layers in feature extraction model MyEfficient-V2B0: 274
Further layer unfreezing is not workable on my device.
In [26]:
# we chose to train the top 1 we will freeze
# the first 249 layers and unfreeze the rest:
#
"""
for layer in model.layers[:268]:
layer.trainable = False
for layer in model.layers[268:]:
layer.trainable = True
"""
Out[26]:
'\nfor layer in model.layers[:268]:\n layer.trainable = False\nfor layer in model.layers[268:]:\n layer.trainable = True\n'
In [27]:
"""
model = compile_model(model)
model.fit(ds_train,
epochs=epochs,
validation_data=ds_val,
callbacks=callbacks)
"""
Out[27]:
'\nmodel = compile_model(model)\nmodel.fit(ds_train,\n epochs=epochs, \n validation_data=ds_val,\n callbacks=callbacks)\n'
Save the model¶
In [37]:
# Save in tf SavedModel format
model_filepath = os.path.join("model", "all_binary_tf_6771")
tf.saved_model.save(best_model, model_filepath)
WARNING:absl:Found untraced functions such as _jit_compiled_convolution_op, _jit_compiled_convolution_op, _jit_compiled_convolution_op, _jit_compiled_convolution_op, _jit_compiled_convolution_op while saving (showing 5 of 91). These functions will not be directly callable after loading.
INFO:tensorflow:Assets written to: model\all_binary_tf_6771\assets
INFO:tensorflow:Assets written to: model\all_binary_tf_6771\assets
In [38]:
# save in keras format
model_filepath = os.path.join("model", "all_binary_6771.keras")
tf.keras.models.save_model(best_model, model_filepath, overwrite=True)
What would happened if the best hyperparameters are used to retrained the model?¶
Retrained the model with the best hyperparameters. However the val_binary_accuracy was not the same as the one from the best model. The reason could be that one epoch is not enough. The retraining may need many epochs to reach the val_binary_accuracy like the best model.
In [24]:
# Get the top hyperparameters.
best_hps = tuner.get_best_hyperparameters(1)
# Build the model with the best hp.
model = build_model(best_hps[0])
# Fit with the entire dataset.
# Only need to train 1 time with the best hyperparameters.
history = model.fit(ds_train, epochs=1, validation_data=ds_val, callbacks=callbacks)
114/114 [==============================] - 21s 152ms/step - loss: 1.7248 - binary_accuracy: 0.5132 - precision: 0.4819 - recall: 0.4337 - val_loss: 0.7009 - val_binary_accuracy: 0.5833 - val_precision: 0.5263 - val_recall: 0.3614
In [30]:
history.history["val_binary_accuracy"][0]
Out[30]:
0.5833333730697632
In [31]:
# Quick summary of the best metrics according to binary_accuracy
val = np.argmax(np.array(history.history["val_binary_accuracy"]))
print(f"Binary accuracy: {history.history['val_binary_accuracy'][val]}")
print(f"Precision: {history.history['val_precision'][val]}")
print(f"Recall: {history.history['val_recall'][val]}")
Binary accuracy: 0.5833333730697632 Precision: 0.5263158082962036 Recall: 0.3614457845687866
In [32]:
plot_history(history)
In [ ]:
In [ ]: