Creating Custom Datasets

When working with standard datasets like MNIST, data is prepackaged and ready for use. However, for domain-specific applications, creating custom datasets becomes essential.

The process involves mapping image paths to corresponding labels through a dedicated function that returns features and their associated labels.

To generate .npy formatted datasets for training, extract grayscale pixel values into a feature list and append labels to a separate label list. Here's an example extraction function:

import numpy as np
from PIL import Image

def create_dataset(data_dir, label_file):
    with open(label_file, 'r') as f:
        lines = f.readlines()
    
    features, labels = [], []
    for line in lines:
        parts = line.strip().split()
        img_path = data_dir + parts[0]
        img = Image.open(img_path).convert('L')
        img_array = np.array(img)
        img_array = img_array / 255.0
        features.append(img_array)
        labels.append(parts[1])
        print(f'Loading: {line}')
    
    features = np.array(features)
    labels = np.array(labels).astype(np.int64)
    return features, labels

Complete Implemantation Example

import tensorflow as tf
from PIL import Image
import numpy as np
import os

# Paths
train_path = './fashion_image_label/fashion_train_jpg_60000/'
train_txt = './fashion_image_label/fashion_train_jpg_60000.txt'
x_train_savepath = './fashion_image_label/fashion_x_train.npy'
y_train_savepath = './fashion_image_label/fahion_y_train.npy'
test_path = './fashion_image_label/fashion_test_jpg_10000/'
test_txt = './fashion_image_label/fashion_test_jpg_10000.txt'
x_test_savepath = './fashion_image_label/fashion_x_test.npy'
y_test_savepath = './fashion_image_label/fashion_y_test.npy'

# Dataset creation function

def create_dataset(data_dir, label_file):
    with open(label_file, 'r') as f:
        lines = f.readlines()
    
    features, labels = [], []
    for line in lines:
        parts = line.strip().split()
        img_path = data_dir + parts[0]
        img = Image.open(img_path).convert('L')
        img_array = np.array(img)
        img_array = img_array / 255.0
        features.append(img_array)
        labels.append(parts[1])
        print(f'Loading: {line}')
    
    features = np.array(features)
    labels = np.array(labels).astype(np.int64)
    return features, labels

# Load or generate dataset
if all(os.path.exists(p) for p in [x_train_savepath, y_train_savepath, x_test_savepath, y_test_savepath]):
    print('Loading existing datasets...')
    x_train = np.load(x_train_savepath)
    y_train = np.load(y_train_savepath)
    x_test = np.load(x_test_savepath)
    y_test = np.load(y_test_savepath)
    x_train = x_train.reshape(len(x_train), 28, 28)
    x_test = x_test.reshape(len(x_test), 28, 28)
else:
    print('Generating new datasets...')
    x_train, y_train = create_dataset(train_path, train_txt)
    x_test, y_test = create_dataset(test_path, test_txt)
    
    # Save reshaped versions
    x_train_save = x_train.reshape(len(x_train), -1)
    x_test_save = x_test.reshape(len(x_test), -1)
    np.save(x_train_savepath, x_train_save)
    np.save(y_train_savepath, y_train)
    np.save(x_test_savepath, x_test_save)
    np.save(y_test_savepath, y_test)

# Build model
model = tf.keras.Sequential([
    tf.keras.layers.Flatten(),
    tf.keras.layers.Dense(128, activation='relu'),
    tf.keras.layers.Dense(10, activation='softmax')
])

model.compile(optimizer='adam',
              loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=False),
              metrics=['sparse_categorical_accuracy'])

model.fit(x_train, y_train, batch_size=32, epochs=5, validation_data=(x_test, y_test), validation_freq=1)
model.summary()

Data Augmentation

When dataset size is limited, augmentation techniques help improve model generalization by artificially expanding training data.

This approach handles variations caused by different capture angles or lighting conditions.

import tensorflow as tf
from tensorflow.keras.preprocessing.image import ImageDataGenerator

# Load Fashion MNIST dataset
fashion = tf.keras.datasets.fashion_mnist
(x_train, y_train), (x_test, y_test) = fashion.load_data()
x_train, x_test = x_train / 255.0, x_test / 255.0

# Reshape for compatibility
x_train = x_train.reshape(x_train.shape[0], 28, 28, 1)

# Define augmentation parameters
augmenter = ImageDataGenerator(
    rescale=1./1.,
    rotation_range=45,
    width_shift_range=0.15,
    height_shift_range=0.15,
    horizontal_flip=True,
    zoom_range=0.5
)

augmenter.fit(x_train)

# Model definition
model = tf.keras.Sequential([
    tf.keras.layers.Flatten(),
    tf.keras.layers.Dense(128, activation='relu'),
    tf.keras.layers.Dense(10, activation='softmax')
])

model.compile(optimizer='adam',
              loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=False),
              metrics=['sparse_categorical_accuracy'])

# Train using augmented data
model.fit(augmenter.flow(x_train, y_train, batch_size=32), epochs=5, validation_data=(x_test, y_test), validation_freq=1)
model.summary()

Checkpointing and Model Persistence

For long-running training sessions, saving checkpoints ensures progress can be resumed if interrupted.

Use ModelCheckpoint callback to save best weights automatically:

checkpoint_save_path = "./checkpoint/fashion.ckpt"
callback = tf.keras.callbacks.ModelCheckpoint(
    filepath=checkpoint_save_path,
    save_weights_only=True,
    save_best_only=True
)

history = model.fit(x_train, y_train, batch_size=32, epochs=5, validation_data=(x_test, y_test), validation_freq=1, callbacks=[callback])

Extracting Network Parameters

Accessing trainable variables allows exporting learned parameters for deployment across platforms:

print(model.trainable_variables)
with open('./weights.txt', 'w') as f:
    for var in model.trainable_variables:
        f.write(f'{var.name}\n')
        f.write(f'{var.shape}\n')
        f.write(f'{var.numpy()}\n')

Visualizing Training Metrics

Plotting accuracy and loss curves provides insights into model performance over time:

import matplotlib.pyplot as plt

acc = history.history['sparse_categorical_accuracy']
val_acc = history.history['val_sparse_categorical_accuracy']
loss = history.history['loss']
val_loss = history.history['val_loss']

plt.figure(figsize=(12, 4))

plt.subplot(1, 2, 1)
plt.plot(acc, label='Training Accuracy')
plt.plot(val_acc, label='Validation Accuracy')
plt.title('Accuracy')
plt.legend()

plt.subplot(1, 2, 2)
plt.plot(loss, label='Training Loss')
plt.plot(val_loss, label='Validation Loss')
plt.title('Loss')
plt.legend()

plt.show()

Application Deployment

Deploying trained models for inference requires loading the architecture and saved weights:

from PIL import Image
import numpy as np
import tensorflow as tf

class_names = ['T-shirt/top', 'Trouser', 'Pullover', 'Dress', 'Coat', 'Sandal', 'Shirt', 'Sneaker', 'Bag', 'Ankle boot']

model = tf.keras.Sequential([
    tf.keras.layers.Flatten(),
    tf.keras.layers.Dense(128, activation='relu'),
    tf.keras.layers.Dense(10, activation='softmax')
])

model.load_weights('./checkpoint/fashion.ckpt')

num_images = int(input("Enter number of test images:"))
for _ in range(num_images):
    img_path = input("Enter image path:")
    img = Image.open(img_path).resize((28, 28)).convert('L')
    img_array = 255 - np.array(img)  # Invert colors
    img_array = img_array / 255.0
    prediction = model.predict(img_array[np.newaxis, ...])
    predicted_class = np.argmax(prediction)
    print(f"Predicted class: {class_names[predicted_class]}")