7
\$\begingroup\$

[I'm awaiting suggestions for improvement/optimization/more speed/general feedback ...]

This code takes a label and a folder path of subfolders as input that have certain labels ex: trees, cats with each folder containing a list of HD photos corresponding to the folder name, then a multithreaded image processor converts data to .h5 format and classifies photos with the given label (75% accuracy with default parameters but might get better or worse). It's a very simple algorithm using logistic regression model(implementation) not sklearn's LogisticRegression() and some visualizations using matplotlib(read the docs).

The following link has the code as well as the necessary .h5 files to make it work and demonstrate an example on classifying dog photos but feel free to test with your own image data.

Running the example in the code with the random seed set to 127 should give these results:

  1. Labeled predicted images(sample): The following figure represents n samples of the correct predicted results and displays 1 for dog and 0 for non-dog images.

Results

  1. Learning curve with the given parameters: This is the cost function/degree of error with respect to gradient descent iteration number (max_iter=3000)

Learning curve Code:

from concurrent.futures import ThreadPoolExecutor, as_completed
import matplotlib.pyplot as plt
from time import perf_counter
import pandas as pd
import numpy as np
import cv2
import os


def read_and_resize(img, new_size):
    """
    Read and resize image.
    Args:
        img: Image path.
        new_size: New image size.

    Return:
         Resized Image.
    """
    img = cv2.imread(img)
    return cv2.resize(img, new_size)


def folder_to_hdf(folder_path, label, new_size, threads=5):
    """
    Save a folder images to hdf format.
    Args:
        folder_path: Path to folder containing images.
        label: Data label.
        new_size: New image size(tuple).
        threads: Number of parallel threads.

    Return:
         None
    """
    data, resized = pd.DataFrame(), []
    with ThreadPoolExecutor(max_workers=threads) as executor:
        future_resized_images = {
            executor.submit(read_and_resize, folder_path + img, new_size): img
            for img in os.listdir(folder_path)
            if img != '.DS_Store'
        }
        for future in as_completed(future_resized_images):
            result = future.result()
            resized.append(result)
            print(f'Processing ({label})-{future_resized_images[future]} ... done.')
            del future_resized_images[future]
        data['Images'] = resized
        data['Label'] = label
        data.to_hdf(folder_path + label + '.h5', label)


def folders_to_hdf(folder_path, new_size, threads=5):
    """
    Save folders containing images to hdf format.
    Args:
        folder_path: Path to folder containing folders containing images.
        new_size: New image size(tuple).
        threads: Number of parallel threads.

    Return:
         None
    """
    for folder_name in os.listdir(folder_path):
        if folder_name != '.DS_Store':
            path = ''.join([folder_path, folder_name, '/'])
            folder_to_hdf(path, folder_name, new_size, threads)


def clear_hdf(folder_path):
    """
    Delete .h5 files in every sub-folder in folder_path.
    Args:
        folder_path: A folder containing folders of images and their .h5.

    Return:
        None
    """
    for folder_name in os.listdir(folder_path):
        if folder_name != '.DS_Store':
            file_name = ''.join([folder_path, folder_name, '/', folder_name, '.h5'])
            os.remove(file_name)
            print(f'Removed {file_name.split("/")[-1]}')


def load_hdf(folder_path, label, random_seed=None):
    """
    Load all classification data.
    Args:
        folder_path: Folder containing folders with respective images.
        label: Label to set to 1.
        random_seed: int, representing the random seed.

    Return:
         X and Y as numpy arrays and frames.
    """
    if random_seed:
        np.random.seed(random_seed)
    file_names = [
        ''.join([folder_path, folder_name, '/', folder_name, '.h5'])
        for folder_name in os.listdir(folder_path)
        if folder_name != '.DS_Store'
    ]
    frames = [pd.read_hdf(file_name) for file_name in file_names]
    frames = pd.concat(frames)
    frames['Classification'] = 0
    frames.loc[frames['Label'] == label, 'Classification'] = 1
    new_index = np.random.permutation(frames.index)
    frames.index = new_index
    frames.sort_index(inplace=True)
    image_data, labels = (
        np.array(list(frames['Images'])),
        np.array(list(frames['Classification'])),
    )
    return image_data, labels, frames


def pre_process(
    data, test_size, label, max_pixel=255, display_images=None, show_details=False
):
    """
    Split the data into train and test sets and prepare the data for further processing.
    Args:
        data: X and Y as numpy arrays and frames.
        test_size: Percentage of test set.
        label: label to classify.
        max_pixel: The maximum value of a pixel channel.
        display_images: A tuple (n_images, rows, columns)
        show_details: If True, details about the sizes will be displayed.

    Return:
        x_train, y_train, x_test, y_test, frames.
    """
    image_data, labels, frames = data
    total_images = len(image_data)
    if display_images:
        fig = plt.figure()
        plt.title(
            f'Initial(before prediction) {display_images[1]} x {display_images[2]} '
            f'data sample (Classification of {label})'
        )
        rows, columns = display_images[1], display_images[2]
        for i in range(display_images[0]):
            img = image_data[i]
            fig.add_subplot(rows, columns, i + 1)
            plt.imshow(img)
        plt.show()
    image_data = image_data.reshape(total_images, -1) / max_pixel
    labels = labels.reshape(total_images, -1)
    separation_index = int(test_size * total_images)
    x_train = image_data[separation_index:].T
    y_train = labels[separation_index:].T
    x_test = image_data[:separation_index].T
    y_test = labels[:separation_index].T
    if show_details:
        print(f'Total number of images: {total_images}')
        print(f'x_train shape: {x_train.shape}')
        print(f'y_train shape: {y_train.shape}')
        print(f'x_test shape: {x_test.shape}')
        print(f'y_test shape: {y_test.shape}')
    return x_train, y_train, x_test, y_test, frames


def sigmoid(x):
    """
    Calculate sigmoid function.
    Args:
        x: Image data in the following shape(pixels * pixels * 3, number of images).

    Return:
        sigmoid(x).
    """
    return 1 / (1 + np.exp(-x))


def compute_cost(w, b, x, y):
    """
    Compute cost function using forward and back propagation.
    Args:
        w: numpy array of weights(also called Theta).
        b: Bias(int)
        x: Image data in the following shape(pixels * pixels * 3, number of images)
        y: Label numpy array of labels(0s and 1s)

    Return:
        Cost and gradient(dw and db).
    """
    total_images = x.shape[1]
    activation = sigmoid(np.dot(w.T, x) + b)
    cost = (-1 / total_images) * (
        np.sum(y * np.log(activation) + (1 - y) * np.log(1 - activation))
    )
    cost = np.squeeze(cost)
    dw = (1 / total_images) * np.dot(x, (activation - y).T)
    db = (1 / total_images) * np.sum(activation - y)
    return cost, dw, db


def g_descent(w, b, x, y, max_iter, learning_rate, iteration_number=None):
    """
    Optimize weights and bias using gradient descent algorithm.
    Args:
        w: numpy array of weights(also called Theta).
        b: Bias(int)
        x: Image data in the following shape(pixels * pixels * 3, number of images)
        y: Label numpy array of labels(0s and 1s)
        max_iter: Maximum number of iterations
        learning_rate: The rate of learning.
        iteration_number: Display iteration and current cost every n.

    Return:
        w, b, dw, db, costs
    """
    dw, db, costs = 0, 0, []
    for iteration in range(max_iter):
        cost, dw, db = compute_cost(w, b, x, y)
        w -= dw * learning_rate
        b -= db * learning_rate
        costs.append(cost)
        if iteration_number and iteration % iteration_number == 0:
            print(f'Iteration number: {iteration} out of {max_iter} iterations')
            print(f'Current cost: {cost}\n')
    return w, b, dw, db, costs


def predict(w, b, x):
    """
    Predict labels of x.
    Args:
        w: numpy array of weights(also called Theta).
        b: Bias(int)
        x: Image data in the following shape(pixels * pixels * 3, number of images)

    Return:
        Y^ numpy array of predictions.
    """
    w = w.reshape(x.shape[0], 1)
    activation = sigmoid(np.dot(w.T, x) + b)
    activation[activation > 0.5] = 1
    activation[activation <= 0.5] = 0
    return activation


def predict_from_folder(
    folder_path,
    target_label,
    test_size=0.2,
    max_iter=3000,
    learning_rate=0.0001,
    random_seed=None,
    max_pixel=255,
    display_initial_images=None,
    show_initial_details=False,
    create_hdf=False,
    threads=5,
    new_image_sizes=(150, 150),
    display_results=False,
    iteration_number=100,
    plot_learning_curve=False,
    photo_results=None,
):
    """
    Classify a target label from different folders containing HD images in a way that each folder contains
    only images that represent the folder name/label.
    Args:
        folder_path: A folder path that contains other folders named according to what label they contain.
        target_label: One of the folder labels in folder_path.
        test_size: Test size in percentage ex: 0.7
        max_iter: Maximum number of iterations for gradient descent.
        learning_rate: Learning rate for gradient descent,
        random_seed: int, representing a random seed.
        max_pixel: The maximum value of a pixel channel.
        display_initial_images: A tuple (n_images, rows, columns)
        show_initial_details: If True, details about the initial sizes will be displayed.
        create_hdf: If True, old .h5 files found in the labeled folders will be cleared(if any)
         and new ones will be created.
        threads: int, representing the number of parallel threads for .h5 creation process.
        new_image_sizes: New dimensions to store in .h5 files.
        display_results: If True, training accuracy will be calculated and displayed.
        iteration_number: Display gradient descent iteration number and current cost.
        plot_learning_curve: If True, learning curve will be plotted.
        photo_results: A tuple (n_photos, rows, columns) to display n labeled results.
    Return:
        pandas DataFrame with the results.
    """
    start_time = perf_counter()
    if create_hdf:
        clear_hdf(folder_path)
        folders_to_hdf(folder_path, new_image_sizes, threads)
    data = load_hdf(folder_path, target_label, random_seed)
    x_train, y_train, x_test, y_test, frames = pre_process(
        data,
        test_size,
        target_label,
        max_pixel,
        display_initial_images,
        show_initial_details,
    )
    w, b = np.zeros((len(x_train), 1)), 0
    w, b, dw, db, costs = g_descent(
        w, b, x_train, y_train, max_iter, learning_rate, iteration_number
    )
    train_predictions = predict(w, b, x_train)
    test_predictions = predict(w, b, x_test)
    all_predictions = np.append(train_predictions, test_predictions)
    training_accuracy = 100 - np.mean(np.abs(train_predictions - y_train)) * 100
    test_accuracy = 100 - np.mean(np.abs(test_predictions - y_test)) * 100
    frames['Predictions'] = all_predictions
    frames['Accuracy'] = 0
    frames.loc[frames['Predictions'] == frames['Classification'], 'Accuracy'] = 1
    if display_results:
        print(f'Training accuracy: {training_accuracy}%')
        print(f'Test accuracy: {test_accuracy}%')
        print(f'Train predictions: \n{train_predictions}')
        print(f'Train actual: \n{y_train}')
        print(f'Test predictions: \n{test_predictions}')
        print(f'Test actual: \n{y_test}')
    if plot_learning_curve:
        plt.title('Learning curve')
        plt.plot(range(max_iter), costs)
        plt.xlabel('Iterations')
        plt.ylabel('Cost')
    if photo_results:
        to_display = frames[frames['Accuracy'] == 1][['Images', 'Predictions']].head(
            photo_results[0]
        )
        images = np.array(list(to_display['Images']))
        predictions = np.array(list(to_display['Predictions']))
        fig = plt.figure()
        plt.title(f'Classification of {target_label} results sample')
        rows, columns = photo_results[1], photo_results[2]
        for i in range(photo_results[0]):
            img = images[i]
            ax = fig.add_subplot(rows, columns, i + 1)
            ax.title.set_text(f'Prediction: {predictions[i]}')
            plt.imshow(img)
        plt.show()
    end_time = perf_counter()
    print(f'Time: {end_time - start_time} seconds.')
    return frames


if __name__ == '__main__':
    folder = 'test_photos/'
    target = 'Dog'
    photo_display_size = (10, 2, 5)
    predict_from_folder(
        folder,
        target,
        display_initial_images=photo_display_size,
        show_initial_details=True,
        display_results=True,
        plot_learning_curve=True,
        photo_results=photo_display_size,
        random_seed=127
    )
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.