# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3
#     language: python
#     name: python3
# ---

# # Reconhecimento de Imagens usando TensorflowOnSpark
#
# ## Exemplo reconhecimento  de espécies de flores
#
# Este exemplo lê um diretório de imagens, cria um vetor de features e inclui um tag de identificação.
#
# Cada diretório pode comnter qualquer numero de imagens, no entanto, é obrigatório que cada diretório possua apenas uma categoria de imagem.

# +
# Load libs
import findspark
findspark.init()

from pyspark import SparkContext
from pyspark.sql import SparkSession

from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark.ml import Pipeline
from pyspark.ml.image import ImageSchema

import tensorflow as tf
import tensorflowonspark as tfos

from tensorflow.keras import layers, models, utils
from tensorflow.keras.applications import VGG19, ResNet50, InceptionV3

from tensorflowonspark import compat, TFNode, TFCluster
from pyspark.conf import SparkConf

import argparse
import IPython.display as display
import numpy as np
import matplotlib.pyplot as plt
import os
import pathlib
from time import perf_counter

AUTOTUNE = tf.data.experimental.AUTOTUNE

tf.__version__
# -

tfos.__version__

# +
# Set main images directory
data_repo = "/data/dataset/flower_photos/"

TRAIN_SAMPLE = 0.8
TEST_SAMPLE = 0.2
BATCH_SIZE = 10
IMG_HEIGHT = 224
IMG_WIDTH = 224
EPOCHS=8

# Set base directory (where are all flower's directories)
data_dir = pathlib.Path(data_repo)
print(data_dir)

# +
# Starting Spark Cluster with Docker
spark = SparkSession.builder\
        .appName("FlowerID_TF_TensorflowOnSpark") \
        .master("mesos://zk://10.129.64.20:2181,10.129.64.10:2181,10.129.64.30:2181/mesos") \
        .config("spark.mesos.executor.docker.image","lasid/spark-worker-tensorflow:2.4.6_bionic") \
        .config("spark.mesos.containerizer","docker") \
        .getOrCreate()

spark.conf.set("spark.submit.deployMode", "client")

# create the Spark Context
sc = spark.sparkContext

strategy = tf.distribute.experimental.MultiWorkerMirroredStrategy()


# +
#configurations = spark.sparkContext.getConf().getAll()

#for conf in configurations:
#    print(conf)
# -

# ## Bibliotecas úteis

# +
# Some usefull libs to read and prepare images

def get_label(file_path):
    # Get directory name,, i.e., flower specie 
    parts = tf.strings.split(file_path, '/')
    # Check file path and return label
    return parts[-2] == CLASS_NAMES

def decode_img(img):
    # Decode jpeg image
    img = tf.image.decode_jpeg(img, channels=3) #color images
    # Convert unit8 tensor to floats in the [0,1]range
    img = tf.image.convert_image_dtype(img, tf.float32) 
    # Resize the image  
    return tf.image.resize(img, [IMG_WIDTH, IMG_HEIGHT]) 

def process_path(file_path):
    # Get label
    label = get_label(file_path)
    # Get image file and convert to tensor
    img = tf.io.read_file(file_path)
    img = decode_img(img)
    return img, label

def prepare_for_training(ds, cache=True, shuffle_buffer_size=1000):
    # Test if images are on cache and use it
    if cache:
        if isinstance(cache, str):
            ds = ds.cache(cache)
        else:
            ds = ds.cache()
    ds = ds.shuffle(buffer_size=shuffle_buffer_size)
    ds = ds.repeat()
    ds = ds.batch(BATCH_SIZE)
    # `prefetch` lets the dataset fetch batches in the background
    ds = ds.prefetch(buffer_size=AUTOTUNE)
    return ds

def show_batch(image_batch, label_batch):
    plt.figure(figsize=(10,10))
    for n in range(10):
        ax = plt.subplot(5,5,n+1)
        plt.xticks([])
        plt.yticks([])
        plt.grid(False)
        plt.imshow(image_batch[n])
        plt.title(CLASS_NAMES[label_batch[n]==1][0].title())
        plt.axis('off')
    plt.show() 


# +
def main_fun(args, ctx):
#  import tensorflow_datasets as tfds
    import tensorflow as tf
    from tensorflowonspark import compat

#  tfds.disable_progress_bar()
    strategy = tf.distribute.experimental.MultiWorkerMirroredStrategy()

    BUFFER_SIZE = args.buffer_size
    BATCH_SIZE = args.batch_size
    NUM_WORKERS = args.cluster_size
    EPOCHS = args.epochs
    TRAIN_SAMPLE = args.train_sample
    TEST_SAMPLE = 1 - TRAIN_SAMPLE
    GLOBAL_BATCH_SIZE = BATCH_SIZE * NUM_WORKERS
    
  # Scaling MNIST data from (0, 255] to (0., 1.]
  #def scale(image, label):
  #  return tf.cast(image, tf.float32) / 255, label

  # workaround for https://github.com/tensorflow/datasets/issues/1405
  #datasets = tfds.load(name='mnist', split='train', as_supervised=True)
    #options = tf.data.Options()
    #options.experimental_distribute.auto_shard_policy = tf.data.experimental.AutoShardPolicy.DATA
    #train_datasets_unbatched = datasets.with_options(options).repeat().map(scale).shuffle(BUFFER_SIZE)

    def get_label(file_path):
        # Get directory name,, i.e., flower specie 
        parts = tf.strings.split(file_path, '/')
        # Check file path and return label
        return parts[-2] == CLASS_NAMES

    def decode_img(img):
        # Decode jpeg image
        img = tf.image.decode_jpeg(img, channels=3) #color images
        # Convert unit8 tensor to floats in the [0,1]range
        img = tf.image.convert_image_dtype(img, tf.float32) 
        # Resize the image  
        return tf.image.resize(img, [IMG_WIDTH, IMG_HEIGHT]) 

    def process_path(file_path):
        # Get label
        label = get_label(file_path)
        # Get image file and convert to tensor
        img = tf.io.read_file(file_path)
        img = decode_img(img)
        return img, label

    def prepare_for_training(ds, cache=True, shuffle_buffer_size=1000):
        # Test if images are on cache and use it
        if cache:
            if isinstance(cache, str):
                ds = ds.cache(cache)
            else:
                ds = ds.cache()
        ds = ds.shuffle(buffer_size=shuffle_buffer_size)
        ds = ds.repeat()
        ds = ds.batch(BATCH_SIZE)
        # `prefetch` lets the dataset fetch batches in the background
        ds = ds.prefetch(buffer_size=AUTOTUNE)
        return ds
    
    def build_and_compile_cnn_model():
        model = models.Sequential([
            layers.Conv2D(16, 3, padding='same', activation='relu', input_shape=(IMG_HEIGHT, IMG_WIDTH ,3)),
            layers.MaxPooling2D(),
            layers.Conv2D(32, 3, padding='same', activation='relu'),
            layers.MaxPooling2D(),
            layers.Conv2D(64, 3, padding='same', activation='relu'),
            layers.MaxPooling2D(),
            layers.Flatten(),
            layers.Dense(5, activation='relu'),
            layers.Dense(5)
        ])

        model.compile(optimizer='adam',
              loss=tf.keras.losses.CategoricalCrossentropy(from_logits=True),
              metrics=['accuracy'])
        return model
        


      # single node
      # single_worker_model = build_and_compile_cnn_model()
      # single_worker_model.fit(x=train_datasets, epochs=3)

      # Here the batch size scales up by number of workers since
      # `tf.data.Dataset.batch` expects the global batch size. Previously we used 64,
      # and now this becomes 128.


 #   train_datasets = train_ds.batch(GLOBAL_BATCH_SIZE)

      # this fails
      # callbacks = [tf.keras.callbacks.ModelCheckpoint(filepath=args.model_dir)]
#    tf.io.gfile.makedirs(args.model_dir)
#    filepath = args.model_dir + "/weights-{epoch:04d}"
#    callbacks = [
#        tf.keras.callbacks.ModelCheckpoint(filepath=filepath, verbose=1, save_weights_only=True),
#        tf.keras.callbacks.TensorBoard(log_dir=args.model_dir)
#    ]

    # Read all images files and set Tensor
    list_ds = tf.data.Dataset.list_files(str(args.data_dir/'*/*'))

    with strategy.scope():
        labeled_ds = list_ds.map(process_path, num_parallel_calls=AUTOTUNE)
        images_ds = prepare_for_training(labeled_ds)
        train_ds = images_ds.take(int(TRAIN_SAMPLE*SIZE_OF_DATASET))
        test_ds = images_ds.skip(int(TEST_SAMPLE*SIZE_OF_DATASET))
        train_images, train_labels = next(iter(train_ds))
        test_images, test_labels = next(iter(test_ds))
        multi_worker_model = build_and_compile_cnn_model()
        multi_worker_model.fit(train_images, train_labels, batch_size=BATCH_SIZE, epochs=EPOCHS, validation_data=(test_images, test_labels))
        
#    multi_worker_model.fit(x=train_datasets, epochs=args.epochs, steps_per_epoch=args.steps_per_epoch, callbacks=callbacks)

#    from tensorflow_estimator.python.estimator.export import export_lib
#    export_dir = export_lib.get_timestamped_export_dir(args.export_dir)
#    compat.export_saved_model(multi_worker_model, export_dir, ctx.job_name == 'chief')

# -

# ## Leitura images do repositorio

# +
# Count the number of flower images and set dataset size
SIZE_OF_DATASET = len(list(data_dir.glob('*/*.jpg')))

print("Number of images: ",SIZE_OF_DATASET)
# -

# Get the Class names reading directory names
CLASS_NAMES = np.array([item.name for item in data_dir.glob('*') if item.name != "LICENSE.txt"])
#CLASS_NAMES = list([item.name for item in data_dir.glob('*') if item.name != "LICENSE.txt"])
print(CLASS_NAMES)

# +
# Read all images files and set Tensor
#list_ds = tf.data.Dataset.list_files(str(data_dir/'*/*'))

#for f in list_ds.take(5):
#    print(f.numpy())

# +
# Process images
start_time = perf_counter()

#image_df = ImageSchema.readImages("/data/myimages")

#labeled_ds = list_ds.map(process_path, num_parallel_calls=AUTOTUNE)
#images_ds = prepare_for_training(labeled_ds)

# +
# Split train and test sample
#train_ds = images_ds.take(int(TRAIN_SAMPLE*SIZE_OF_DATASET))
#test_ds = images_ds.skip(int(TEST_SAMPLE*SIZE_OF_DATASET))

#train_images, train_labels = next(iter(train_ds))
#test_images, test_labels = next(iter(test_ds))

#print("train image: ", tf.size(train_images))
#print("train labels: ", tf.size(train_labels))

#time_ip = perf_counter() - start_time
#tf.print("Image processing time:", time_ip)

# +
#image_batch, label_batch = next(iter(train_ds))
#show_batch(image_batch.numpy(), label_batch.numpy())
# -

executors = sc._conf.get("spark.executor.instances")
num_executors = int(executors) if executors is not None else 1
num_executors = 3

# +
parser = argparse.ArgumentParser()
parser.add_argument("--batch_size", help="number of records per batch", type=int, default=BATCH_SIZE)
parser.add_argument("--buffer_size", help="size of shuffle buffer", type=int, default=10000)
parser.add_argument("--cluster_size", help="number of nodes in the cluster", type=int, default=num_executors)
parser.add_argument("--epochs", help="number of epochs", type=int, default=EPOCHS)
parser.add_argument("--train_sample", help="percentage of train sample", type=int, default=TRAIN_SAMPLE)
parser.add_argument("--model_dir", help="path to save model/checkpoint", default="./model")
#parser.add_argument("--export_dir", help="path to export saved_model", default="mnist_export")
parser.add_argument("--data_dir", help="path to data dir", default=data_dir)
parser.add_argument("--steps_per_epoch", help="number of steps per epoch", type=int, default=469)
parser.add_argument("--tensorboard", help="launch tensorboard process", action="store_true")



args, unknown = parser.parse_known_args()
#args = parser.parse_args()
print("args:", args)

# +
# cluster = TFCluster.run(sc, main_fun, args, args.cluster_size, num_ps=0, tensorboard=args.tensorboard, input_mode=TFCluster.InputMode.TENSORFLOW, master_node='chief', log_dir=args.model_dir)    

cluster = TFCluster.run(sc, main_fun, args, args.cluster_size, num_ps=0, input_mode=TFCluster.InputMode.TENSORFLOW)
# -

# Stop all
cluster.shutdown()
spark.stop()

# # Create Convolutional Neural Network (CNN) Model
#
# ## Modelo Simples 

# +
start_time = perf_counter()

model = models.Sequential([
    layers.Conv2D(16, 3, padding='same', activation='relu', input_shape=(IMG_HEIGHT, IMG_WIDTH ,3)),
    layers.MaxPooling2D(),
    layers.Conv2D(32, 3, padding='same', activation='relu'),
    layers.MaxPooling2D(),
    layers.Conv2D(64, 3, padding='same', activation='relu'),
    layers.MaxPooling2D(),
    layers.Flatten(),
    layers.Dense(5, activation='relu'),
    layers.Dense(5)
])

model.compile(optimizer='adam',
              loss=tf.keras.losses.CategoricalCrossentropy(from_logits=True),
              metrics=['accuracy'])

model.summary()
# -

history = model.fit(train_images, train_labels, batch_size=BATCH_SIZE, epochs=EPOCHS, validation_data=(test_images, test_labels))   

# ### Evaluating the Simple CNN Model Performance

plt.plot(history.history['loss'])
plt.plot(history.history['val_loss'])
plt.title('Model Loss')
plt.ylabel('Loss')
plt.xlabel('Epochs')
plt.legend(['Train', 'Test'])
plt.show()

plt.plot(history.history['accuracy'])
plt.plot(history.history['val_accuracy'])
plt.title('Model Accuracy')
plt.ylabel('Accuracy')
plt.xlabel('Epochs')
plt.legend(['Train', 'Test'])
plt.show()

train_loss_sm, train_acc_sm = model.evaluate(train_images, train_labels, verbose=2)
test_loss_sm, test_acc_sm = model.evaluate(test_images, test_labels, verbose=2)
time_sm = perf_counter() - start_time
print(time_sm)

# ## Modelo Complexo 

# +
start_time = perf_counter()

model = models.Sequential([
# layers.Conv2D(16, 3, padding='same', activation='relu', input_shape=(IMG_HEIGHT, IMG_WIDTH ,3)),
    layers.Conv2D(32, (3,3), padding='same', activation='relu', input_shape=(IMG_HEIGHT, IMG_WIDTH ,3), name='conv_1'),
    layers.Conv2D(32, (3,3), activation='relu', name='conv_2'),
    layers.MaxPooling2D(pool_size=(2,2), name='maxpool_1'),
    layers.Dropout(0.25),
    layers.Conv2D(64, (3,3), padding='same', activation='relu', name='conv_3'),
    layers.Conv2D(64, (3,3), activation='relu', name='conv_4'),
    layers.MaxPooling2D(pool_size=(2,2), name='maxpool_2'),
    layers.Dropout(0.25),
    layers.Conv2D(128, (3,3), padding='same', activation='relu', name='conv_5'),
    layers.Conv2D(128, (3,3), activation='relu', name='conv_6'),
    layers.MaxPooling2D(pool_size=(2,2), name='maxpool_3'),
    layers.Flatten(),
    layers.Dense(512, activation='relu', name='dense_1'),
    layers.Dropout(0.5),
    layers.Dense(128, activation='relu', name='dense_2'),
    layers.Dense(len(CLASS_NAMES), name='output'),
    layers.Activation('softmax')
])

model.compile(optimizer='adam',
              loss=tf.keras.losses.CategoricalCrossentropy(from_logits=True),
              metrics=['accuracy'])

model.summary()
# -

history = model.fit(train_images, train_labels, batch_size=BATCH_SIZE, epochs=EPOCHS, validation_data=(test_images, test_labels))   

# ### Evaluating the Complex CNN Model Performance

plt.plot(history.history['loss'])
plt.plot(history.history['val_loss'])
plt.title('Model Loss')
plt.ylabel('Loss')
plt.xlabel('Epochs')
plt.legend(['Train', 'Test'])
plt.show()

plt.plot(history.history['accuracy'])
plt.plot(history.history['val_accuracy'])
plt.title('Model Accuracy')
plt.ylabel('Accuracy')
plt.xlabel('Epochs')
plt.legend(['Train', 'Test'])
plt.show()

train_loss_cm, train_acc_cm = model.evaluate(train_images, train_labels, verbose=2)
test_loss_cm, test_acc_cm = model.evaluate(test_images, test_labels, verbose=2)
time_cm = perf_counter() - start_time
print(time_cm)

# ## Modelo VGG19

# +
start_time = perf_counter()

pre_trained_model = VGG19(input_shape=(IMG_HEIGHT, IMG_WIDTH ,3), include_top=False, weights="imagenet")

#pre_trained_model.trainable = False
for layer in pre_trained_model.layers[:19]:
    layer.trainable = False

# creating the final model  
model = models.Sequential([
    pre_trained_model,
    layers.MaxPool2D((2,2) , strides = 2),
    layers.Flatten(),
    layers.Dense(len(CLASS_NAMES) , activation='softmax')
])

#model.compile(optimizer = "adam" , loss = 'categorical_crossentropy' , metrics = ['accuracy'])

#model.compile(optimizer='adam',
#              loss=tf.keras.losses.CategoricalCrossentropy(from_logits=True),
#              metrics=['accuracy'])


model.compile(optimizer='adam',
              loss=tf.keras.losses.CategoricalCrossentropy(from_logits=True),
              metrics=['accuracy'])

model.summary()
# -

history = model.fit(train_images, train_labels, batch_size=BATCH_SIZE, epochs=EPOCHS, validation_data=(test_images, test_labels))   

# ### Evaluating the VGG19 CNN Model Performance

plt.plot(history.history['loss'])
plt.plot(history.history['val_loss'])
plt.title('Model Loss')
plt.ylabel('Loss')
plt.xlabel('Epochs')
plt.legend(['Train', 'Test'])
plt.show()

plt.plot(history.history['accuracy'])
plt.plot(history.history['val_accuracy'])
plt.title('Model Accuracy')
plt.ylabel('Accuracy')
plt.xlabel('Epochs')
plt.legend(['Train', 'Test'])
plt.show()

train_loss_vg, train_acc_vg = model.evaluate(train_images, train_labels, verbose=2)
test_loss_vg, test_acc_vg = model.evaluate(test_images, test_labels, verbose=2)
time_vg = perf_counter() - start_time
print(time_vg)

# ## Modelo ResNet

# +
start_time = perf_counter()

pre_trained_model = ResNet50(include_top=False, pooling='avg')

model = models.Sequential([
    pre_trained_model,
    layers.Flatten(),
    layers.BatchNormalization(),
    layers.Dense(2048, activation='relu'),
    layers.BatchNormalization(),
    layers.Dense(1024, activation='relu'),
    layers.BatchNormalization(),
    layers.Dense(len(CLASS_NAMES), activation='softmax')
])

model.layers[0].trainable = False

model.compile(optimizer='adam',
              loss=tf.keras.losses.CategoricalCrossentropy(from_logits=True),
              metrics=['accuracy'])

model.summary()
# -

history = model.fit(train_images, train_labels, batch_size=BATCH_SIZE, epochs=EPOCHS, validation_data=(test_images, test_labels))   

# ### Evaluating the Resnet CNN Model Performance

plt.plot(history.history['loss'])
plt.plot(history.history['val_loss'])
plt.title('Model Loss')
plt.ylabel('Loss')
plt.xlabel('Epochs')
plt.legend(['Train', 'Test'])
plt.show()

plt.plot(history.history['accuracy'])
plt.plot(history.history['val_accuracy'])
plt.title('Model Accuracy')
plt.ylabel('Accuracy')
plt.xlabel('Epochs')
plt.legend(['Train', 'Test'])
plt.show()

train_loss_rn, train_acc_rn = model.evaluate(train_images, train_labels, verbose=2)
test_loss_rn, test_acc_rn = model.evaluate(test_images, test_labels, verbose=2)
time_rn = perf_counter() - start_time
print(time_rn)

# ## Modelo InceptionV3

# +
start_time = perf_counter()

pre_trained_model = InceptionV3(include_top=False, weights="imagenet", pooling='avg', classifier_activation='softmax')

model = models.Sequential([
    pre_trained_model,
    layers.Flatten(),
    layers.BatchNormalization(),
    layers.Dense(2048, activation='relu'),
    layers.BatchNormalization(),
    layers.Dense(1024, activation='relu'),
    layers.BatchNormalization(),
    layers.Dense(len(CLASS_NAMES), activation='softmax')
])

model.layers[0].trainable = False

model.compile(optimizer='adam',
              loss=tf.keras.losses.CategoricalCrossentropy(from_logits=True),
              metrics=['accuracy'])

model.summary()
# -

# ### Evaluating the InceptionV3 CNN Model Performance

plt.plot(history.history['loss'])
plt.plot(history.history['val_loss'])
plt.title('Model Loss')
plt.ylabel('Loss')
plt.xlabel('Epochs')
plt.legend(['Train', 'Test'])
plt.show()

plt.plot(history.history['accuracy'])
plt.plot(history.history['val_accuracy'])
plt.title('Model Accuracy')
plt.ylabel('Accuracy')
plt.xlabel('Epochs')
plt.legend(['Train', 'Test'])
plt.show()

train_loss_iv, train_acc_iv = model.evaluate(train_images, train_labels, verbose=2)
test_loss_iv, test_acc_iv = model.evaluate(test_images, test_labels, verbose=2)
time_iv = perf_counter() - start_time
print(time_iv)

# +
cluster = TFCluster.run(sc, main_fun, args, args.cluster_size, num_ps=0, tensorboard=args.tensorboard, input_mode=TFCluster.InputMode.SPARK, master_node='chief')

# Note: need to feed extra data to ensure that each worker receives sufficient data to complete epochs
# to compensate for variability in partition sizes and spark scheduling

cluster.train(images_labels, args.epochs)
cluster.shutdown()

# -

# ## Sumary  

print("===========================================================================")
print("===============  Compare CNN Algorithms Acurancy and Time =================")
print()
print("   Dataset size = ",SIZE_OF_DATASET,"   Train sample = ",TRAIN_SAMPLE*100,"%  Test sample = ",TEST_SAMPLE*100,"%")
print("   Image size: (%d,%d)   Batch size = %d         Epochs = %d        " % (IMG_HEIGHT, IMG_WIDTH, BATCH_SIZE, EPOCHS))
print()
print("Processing images:                                          time = %3.1f s" % (time_ip))
print("Simple Model:         train_acc = %1.2f    test_acc = %1.2f   time = %3.1f s" % (train_acc_sm, test_acc_sm, time_sm))
print("Complex Model:        train_acc = %1.2f    test_acc = %1.2f   time = %3.1f s" % (train_acc_cm, test_acc_cm, time_cm))
print("VGG19:                train_acc = %1.2f    test_acc = %1.2f   time = %3.1f s" % (train_acc_vg, test_acc_vg, time_vg))
print("ResNet50:             train_acc = %1.2f    test_acc = %1.2f   time = %3.1f s" % (train_acc_rn, test_acc_rn, time_rn))
print("InceptionV3:          train_acc = %1.2f    test_acc = %1.2f   time = %3.1f s" % (train_acc_iv, test_acc_iv, time_iv))
print("===========================================================================")

spark.stop()



StarcoderdataJupyterScriptsDedupFiltered