6
\$\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\$
0

1 Answer 1

2
\$\begingroup\$

I'm not especially fond of 1st identifier here:

def read_and_resize(img, new_size):

Identifiers in a public API have greater documentation burden than private methods or local temp vars. Better to spell it out, as img_path or image_path.

Also, optional type hinting in the signature wouldn't hurt. That said, the docstring is pretty clear on the details.

It did take me a while to navigate up to predict_from_folder and learn that new_image_sizes=(150, 150) is the recommended default. We pass this parameter through many function calls. Consider turning some functions into methods, and making that image size a self. object attribute. BTW, some of the docstrings explain it's a tuple, while others seem to suggest it could be a scalar integer.


I don't find that this tuple unpack helps aid human comprehension:

    data, resized = pd.DataFrame(), []

Prefer to init data down below the loop, just prior to assigning its columns. You might choose to pass in resized and label in the creation call.

I don't understand why this is valuable:

            del future_resized_images[future]

The entire list will go out-of-scope after just a few source lines. Do we truly not have room for 2x storage for a moment?

If that's true, consider pre-populating the dataframe with a bunch of None values, and overwrite them as you iterate.

Plus, I'm skeptical we even see 2x RAM consumption. Isn't it just a pair of pointers to same underlying image data? Premature optimization is the root of all evil (p.268).


In folders_to_hdf you chose not to use os.walk(). Ok, that's fine. But I don't understand this test:

        if folder_name != '.DS_Store':

Based on the identifier name, surely the relevant test would be is_dir?

    for folder_name in Path(folder_path).glob("*"):
        if folder_name.is_dir():

No?

(I observe that folder_name.is_file() will be True for ".DS_Store".)


In clear_hdf I counsel DRY. It seems like this and the previous method might call a common helper.


We see a call to pre_process(data, test_size, ... ). The data identifier is sometimes warranted, but it is super vague. Here, I would rather see a call of (*data, ... and a signature of:

def pre_process(image_data, labels, frames, test_size, ... ):

Rather than

        rows, columns = display_images[1], display_images[2]

prefer

        _, rows, columns = display_images

Oh, wait, we have a range() in the next line! It's not an ignored parameter. So prefer

        n_images, rows, columns = display_images

Consider Extract Helper for that whole if display_images: clause. Or maybe even evict it entirely -- caller can call the helper separately if desired. It is unclear to me how the stuff within that clause supports this method's Single Responsibility.

Kudos on throwing max_pixel=255, into the signature; I really like that usage. No magic numbers here, good. The rest of the code is extremely clear. Similarly, g_descent is very clear. Nice decomposition into helpers.


I am reading this:

    photo_display_size = (10, 2, 5)

It seems a bit redundant. Consider making it a named tuple. Consider deriving the initial 10 from 2 * 5. Maybe "layout" instead of "size"?


LGTM. Ship it!

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.