Untitled

import matplotlib.pyplot as plt
import numpy as np
import tensorflow as tf
import pandas as pd
import scipy as sp

from tensorflow import keras
from tensorflow.keras import layers
from tensorflow.keras.models import Sequential
from tensorflow.keras.callbacks import EarlyStopping, ModelCheckpoint
from tensorflow.keras.preprocessing.image import ImageDataGenerator

train_df = pd.read_csv('train.csv')
val_df = pd.read_csv('val.csv')
test_df = pd.read_csv('test.csv')
submission_df = pd.read_csv('sample_submission.csv')

print(train_df)
batch_size = 32
img_height = 64
img_width = 64
trainimg_dir = 'train_images'
valimg_dir = 'val_images'
testimg_dir = 'test_images'

checkpoint_filepath = 'best_model.h5'

early_stopping_callback = tf.keras.callbacks.EarlyStopping(
    patience=5,  # Number of epochs with no improvement after which training will be stopped
    restore_best_weights=True  # Restores the weights of the best-performing epoch
)
checkpoint_callback = tf.keras.callbacks.ModelCheckpoint(
    'best_model.h5',  # Filepath to save the best model
    monitor='val_accuracy',  # Metric to monitor for saving the best model
    save_best_only=True,  # Save only the best model
    mode='max'  # Mode for monitoring ('max' for accuracy)
)
def preprocess_image(file_path):
    img = tf.io.read_file(file_path)
    img = tf.image.decode_png(img, channels=3)
    img = tf.image.resize(img, [img_height, img_width])
    return img
train_images = []
train_labels = []

for index, row in train_df.iterrows():
    image_path = trainimg_dir + '/' + row['Image']
    train_images.append(preprocess_image(image_path))
    train_labels.append(row['Class'])

train_images = np.array(train_images)
train_labels = np.array(train_labels)

class_names = np.unique(train_labels)

train_ds = tf.data.Dataset.from_tensor_slices((train_images, train_labels))
train_ds = train_ds.batch(batch_size)

# Verify the number of images and labels
num_images = len(train_images)
num_labels = len(train_labels)
print("Number of images: ", num_images)
print("Number of labels: ", num_labels)
plt.figure(figsize=(10,10))
for images, labels in train_ds.take(1):
    for i in range(9):
        ax = plt.subplot(3,3,i+1)
        plt.imshow(images[i].numpy().astype("uint8"))
        plt.title(class_names[labels[i]])
        plt.axis("off")
        
plt.show()
for image_batch, labels_batch in train_ds:
    print(image_batch.shape)
    print(labels_batch.shape)
    break
validation_images = []
validation_labels = []

for index, row in val_df.iterrows():
    image_path = valimg_dir + '/' + row['Image']
    validation_images.append(preprocess_image(image_path))
    validation_labels.append(row['Class'])
    
validation_images = np.array(validation_images)
validation_labels = np.array(validation_labels)

val_ds = tf.data.Dataset.from_tensor_slices((validation_images, validation_labels))
val_ds = val_ds.batch(batch_size)
plt.figure(figsize=(10,10))
for images, labels in val_ds.take(1):
    for i in range(9):
        ax = plt.subplot(3,3,i+1)
        plt.imshow(images[i].numpy().astype("uint8"))
        plt.title(class_names[labels[i]])
        plt.axis("off")
        
plt.show()
AUTOTUNE = tf.data.experimental.AUTOTUNE

train_ds = train_ds.cache().shuffle(1000).prefetch(buffer_size=AUTOTUNE)
val_ds = val_ds.cache().prefetch(buffer_size=AUTOTUNE)
normalization_layer = layers.Rescaling(1./255)
normalized_train_ds = train_ds.map(lambda x, y: (normalization_layer(x), y))
normalized_val_ds = val_ds.map(lambda x, y: (normalization_layer(x), y))

image_batch_train, labels_batch_train = next(iter(normalized_train_ds))
first_image_train = image_batch_train[0]

print("Train - Min", np.min(first_image_train), "Max: ", np.max(first_image_train))

image_batch_val, labels_batch_val = next(iter(normalized_val_ds))
first_image_val = image_batch_val[0]

print("Val - Min", np.min(first_image_val), "Max: ", np.max(first_image_val))
num_classes = len(class_names)
print(num_classes)

model = Sequential([
    layers.Rescaling(1./255, input_shape=(img_height, img_width, 3)),
    layers.Conv2D(64, 3, padding='same', activation='relu'),
    layers.BatchNormalization(),
    layers.Conv2D(64, 3, padding='same', activation='relu'),
    layers.BatchNormalization(),
    layers.MaxPooling2D(),

    layers.Conv2D(128, 3, padding='same', activation='relu'),
    layers.BatchNormalization(),
    layers.Conv2D(128, 3, padding='same', activation='relu'),
    layers.BatchNormalization(),
    layers.MaxPooling2D(),

    layers.Conv2D(256, 3, padding='same', activation='relu'),
    layers.BatchNormalization(),
    layers.Conv2D(256, 3, padding='same', activation='relu'),
    layers.BatchNormalization(),
    layers.MaxPooling2D(),

    layers.Conv2D(512, 3, padding='same', activation='relu'),
    layers.BatchNormalization(),
    layers.Conv2D(512, 3, padding='same', activation='relu'),
    layers.BatchNormalization(),
    layers.MaxPooling2D(),

    layers.Flatten(),
    layers.Dropout(0.5),
    layers.Dense(1024, activation='relu'),
    layers.BatchNormalization(),
    layers.Dense(num_classes)
])

model.compile(
    optimizer=tf.keras.optimizers.Adam(learning_rate=0.0001),
    loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),
    metrics=['accuracy']
)

model.summary()
def learning_rate_schedule(epoch):
    if epoch < 10:
        return 0.001
    else:
        return 0.001 * tf.math.exp(0.1 * (10 - epoch))
lr_schedule_callback = tf.keras.callbacks.LearningRateScheduler(learning_rate_schedule)
epochs=100
history = model.fit(
    train_ds,
    validation_data=val_ds,
    epochs=epochs,
    callbacks=[early_stopping_callback, checkpoint_callback, lr_schedule_callback]
)
acc = history.history['accuracy']
val_acc = history.history['val_accuracy']

loss = history.history['loss']
val_loss = history.history['val_loss']

epochs_range = range(len(acc))

plt.figure(figsize=(20, 10))
plt.subplot(1, 2, 1)
plt.plot(epochs_range, acc, label='Training Accuracy')
plt.plot(epochs_range, val_acc, label='Validation Accuracy')
plt.legend(loc='lower right')
plt.title('Training and Validation Accuracy')

plt.subplot(1, 2, 2)
plt.plot(epochs_range, loss, label='Training Loss')
plt.plot(epochs_range, val_loss, label='Validation Loss')
plt.legend(loc='upper right')
plt.title('Training and Validation Loss')

plt.show()
# Custom function for Random Erasing
def random_erasing(img):
    p = 0.5  # Probability of applying Random Erasing
    s_l = 0.02  # Minimum erasing area
    s_h = 0.4  # Maximum erasing area
    r_1 = 0.3  # Minimum aspect ratio of erasing area
    mean = np.mean(img)

    if np.random.rand() > p:
        return img

    h, w, _ = img.shape
    s = h * w

    while True:
        se = np.random.uniform(s_l, s_h) * s
        re = np.random.uniform(r_1, 1 / r_1)

        he = int(np.sqrt(se * re))
        we = int(np.sqrt(se / re))

        xe = np.random.randint(0, w)
        ye = np.random.randint(0, h)

        if xe + we <= w and ye + he <= h:
            img[ye:ye + he, xe:xe + we, :] = mean
            return img
data_generator = ImageDataGenerator(
    rotation_range=20,
    width_shift_range=0.2,
    height_shift_range=0.2,
    brightness_range=[0.8, 1.2],
    shear_range=0.2,
    zoom_range=0.2,
    horizontal_flip=True,
    vertical_flip=True,
    preprocessing_function=random_erasing
)

plt.figure(figsize=(10, 10))
for images, _ in train_ds.take(1):
    for i in range(9):
        augmented_images = next(data_generator.flow(images, shuffle=False))
        ax = plt.subplot(3, 3, i + 1)
        plt.imshow(augmented_images[0].astype("uint8"))
        plt.axis("off")
        
plt.show()

augmented_train_ds = data_generator.flow(train_images, train_labels, batch_size=batch_size, shuffle=False)

augmented_train_images = np.concatenate([train_images, augmented_train_ds[0][0]])
augmented_train_labels = np.concatenate([train_labels, augmented_train_ds[0][1]])

print(augmented_train_images.shape)
print(augmented_train_labels.shape)

augmented_train_ds = tf.data.Dataset.from_tensor_slices((augmented_train_images, augmented_train_labels))
augmented_train_ds = augmented_train_ds.shuffle(buffer_size=10000).batch(batch_size)
num_classes = len(class_names)
print(num_classes)

model = Sequential([
    layers.Rescaling(1./255, input_shape=(img_height, img_width, 3)),
    
    layers.Conv2D(128, 3, padding='same', activation='relu'),
    layers.BatchNormalization(),
    layers.Conv2D(128, 3, padding='same', activation='relu'),
    layers.BatchNormalization(),
    layers.MaxPooling2D(),
    
    layers.Conv2D(256, 3, padding='same', activation='relu'),
    layers.BatchNormalization(),
    layers.Conv2D(256, 3, padding='same', activation='relu'),
    layers.BatchNormalization(),
    layers.MaxPooling2D(),
    
    layers.Conv2D(512, 3, padding='same', activation='relu'),
    layers.BatchNormalization(),
    layers.Conv2D(512, 3, padding='same', activation='relu'),
    layers.BatchNormalization(),
    layers.MaxPooling2D(),
    
    layers.Conv2D(1024, 3, padding='same', activation='relu'),
    layers.BatchNormalization(),
    layers.Conv2D(1024, 3, padding='same', activation='relu'),
    layers.BatchNormalization(),
    layers.MaxPooling2D(),
    
    layers.GlobalAveragePooling2D(),  # Change to Global Average Pooling
    
    layers.Dropout(0.5),  # Reduce the dropout rate to 0.5
    
    layers.Dense(128, activation='relu'),
    layers.BatchNormalization(),  # Add Batch Normalization after Dense layer
    layers.Dropout(0.5),  # Add dropout after Dense layer
    
    layers.Dense(num_classes, activation='softmax')
])

# Use a higher learning rate
optimizer = tf.keras.optimizers.Adam(learning_rate=0.001)

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

model.summary()
epochs = 200

# Define a custom learning rate schedule
def learning_rate_schedule(epoch):
    initial_learning_rate = 0.1
    decay_per_epoch = 0.99
    lrate = initial_learning_rate * (decay_per_epoch ** epoch)
    return lrate

lr_schedule_callback = tf.keras.callbacks.LearningRateScheduler(learning_rate_schedule)

history = model.fit(
    augmented_train_ds,
    validation_data=val_ds,
    epochs=epochs,
    callbacks=[early_stopping_callback, checkpoint_callback, lr_schedule_callback]
)
acc = history.history['accuracy']
val_acc = history.history['val_accuracy']

loss = history.history['loss']
val_loss = history.history['val_loss']

epochs_range = range(len(acc))

plt.figure(figsize=(15, 15))
plt.subplot(2, 2, 1)
plt.plot(epochs_range, acc, label='Training Accuracy')
plt.plot(epochs_range, val_acc, label='Validation Accuracy')
plt.legend(loc='lower right')
plt.title('Training and Validation Accuracy')

plt.subplot(2, 2, 2)
plt.plot(epochs_range, loss, label='Training Loss')
plt.plot(epochs_range, val_loss, label='Validation Loss')
plt.legend(loc='upper right')
plt.title('Training and Validation Loss')

plt.show()
test_labels = []

for img in test_df['Image']:
    img = preprocess_image(testimg_dir + '/' + img)
    img = tf.keras.preprocessing.image.img_to_array(img)
    img = tf.expand_dims(img, 0)
    
    prediction = model.predict(img)
    score = tf.nn.softmax(prediction[0])
    
    print("This image most likely belongs to {} with a {:.2f} percent confidence."
            .format(class_names[np.argmax(score)], 100 * np.max(score)))
    
    test_labels.append(class_names[np.argmax(score)])

submission_df['Class'] = test_labels

submission_df.to_csv('gorkalarrucea.csv', index=False)