Réseau de neurone complet. Image classifier

part_3.py

#!/usr/bin/env python3
# coding=utf-8

'''
1. Créer un fichier de test avec moins de données
2. Charger les données depuis le fichier csv
3. Séparer les colonnes selon le truc
4. Créer les outils du NN
5. Créer les différents layers
6. Entrainer
7. Tester
8. Evaluer

'''
import numpy as np
import csv
import matplotlib.pyplot as plt

TRAIN_PROP = 0.6
NN_STRUCTURE = [64, 30, 10]
DATAPATH = 'data/optdigits.tes'


def display_selected_image(given_inputs,
                           given_outputs,
                           random=True,
                           selected_item=None):
    '''
    Choisit une donnée au hasard, la dessine et affiche la valeur
    '''
    if random:
        selected_item = np.random.randint(0, len(given_inputs) - 1)
    digits_images = np.reshape(given_inputs[selected_item, :], (8, 8))
    plt.figure(1, figsize=(3, 3))
    plt.imshow(digits_images, cmap=plt.cm.gray_r, interpolation='nearest')
    print("real output : {}".format(given_outputs[selected_item]))
    plt.show()


def sigmoid(x):
    '''
    Calcule la fonction sigmoid
    '''
    return 1 / (1 + np.exp(-x))


def derivate_sigmoid(x):
    '''
    Calcule la dérivée de la fonction sigmoid
    '''
    y = sigmoid(x)
    return y * (1 - y)


def mean_std(dataset, nb_inputs):
    '''
    Calcule la moyenne et l'écart type de chaque colonne
    '''
    means_vector = np.empty((1, nb_inputs))
    stds_vector = np.empty((1, nb_inputs))
    for col in range(nb_inputs):
        means_vector[0, col] = np.mean(dataset[:, col])
        stds_vector[0, col] = np.std(dataset[:, col])
    return means_vector, stds_vector


def normalize_data(dataset):
    '''
    Normalise chaque colonne des données
    Renvoie une nouvelle matrice avec (X-u)/s en colonne
    si X est la colonne de la matrice dataset

    '''
    shape = np.shape(dataset)
    nb_inputs = shape[1]
    normalised_data = np.empty(shape)
    means_vector, stds_vector = mean_std(dataset, nb_inputs)
    for col in range(nb_inputs):
        mean = means_vector[0, col]
        std = stds_vector[0, col]
        if std == 0:
            # arrive parfois, np va qd meme calculer des NaN un peu partout...
            # pour éviter l'échec on donne une valeur standard à l'écart-type
            std = 1
        normalised_data[:, col] = (dataset[:, col] - mean) * (1 / std)
    return normalised_data, means_vector, stds_vector


def convert_output_vector(dataset):
    '''
    Converts the output vectors to an array
    ie. output = 2 --> [0, 0, 1, 0, 0, 0, 0, 0, 0, 0]
    '''
    vect = np.zeros((len(dataset), 10), dtype=int)
    for i in range(len(dataset)):
        vect[i, dataset[i]] = 1
    return vect


def setup_and_init_weights(nn_structure):
    '''
    Initialise les poids des synapses et du biais.
    '''
    W = {}
    b = {}
    for l in range(1, len(nn_structure)):
        W[l] = np.random.random_sample((
            nn_structure[l], nn_structure[l - 1]
        ))
        b[l] = np.random.random_sample((nn_structure[l], ))
    return W, b


def init_delta_values(nn_structure):
    '''
    Initialise les valeurs des Delta_W et Delta_b
    Ils ont les mêmes dimensions que W et b
    '''
    delta_W = {}
    delta_b = {}
    for l in range(1, len(nn_structure)):
        delta_W[l] = np.zeros((
            nn_structure[l], nn_structure[l - 1]
        ))
        delta_b[l] = np.zeros((nn_structure[l], ))
    return delta_W, delta_b


def feed_forward(in_data, W, b):
    '''
    Calcule les sorties de chaque neurone

    Pour le premier niveau on prend directement la donnée,
    pour les autres on prend le résultat du précédent
    '''
    node_out = {1: in_data}
    weighted_sum = {}
    for l in range(1, len(W) + 1):
        if l == 1:
            node_in = in_data
        else:
            node_in = node_out[l]
        W[l].dot(node_in)
        weighted_sum[l + 1] = W[l].dot(node_in) + b[l]
        node_out[l + 1] = sigmoid(weighted_sum[l + 1])
    return node_out, weighted_sum


def calculate_out_layer_delta(y, h_out, z_out):
    # delta^(nl) = -(y_i - h_i^(nl)) * f'(z_i^(nl))
    return -(y-h_out) * derivate_sigmoid(z_out)


def calculate_hidden_delta(delta_plus_1, w_l, z_l):
    # delta^(l) = (transpose(W^(l)) * delta^(l+1)) * f'(z^(l))
    return np.dot(np.transpose(w_l), delta_plus_1) * derivate_sigmoid(z_l)


def train_nn(nn_structure, X, y, iter_num=3000, alpha=0.25):
    W, b = setup_and_init_weights(nn_structure)
    cnt = 0
    m = len(y)
    avg_cost_func = []
    print('Starting gradient descent for {} iterations'.format(iter_num))
    while cnt < iter_num:
        if cnt % 100 == 0:
            print('Iteration {} of {}'.format(cnt, iter_num))
        delta_W, delta_b = init_delta_values(nn_structure)
        avg_cost = 0
        for i in range(len(y)):
            delta = {}
            h, z = feed_forward(X[i, :], W, b)
            for l in range(len(nn_structure), 0, -1):
                if l == len(nn_structure):
                    delta[l] = calculate_out_layer_delta(
                        y[i, :],
                        h[l],
                        z[l]
                    )
                    avg_cost += np.linalg.norm((y[i, :] - h[l]))
                else:
                    if l > 1:
                        delta[l] = calculate_hidden_delta(
                            delta[l + 1],
                            W[l],
                            z[l]
                        )
                    delta_W[l] += np.dot(
                        delta[l + 1][:, np.newaxis],
                        np.transpose(h[l][:, np.newaxis])
                    )
                    delta_b[l] += delta[l+1]
        for l in range(len(nn_structure) - 1, 0, -1):
            W[l] += -alpha * (1/m * delta_W[l])
            b[l] += -alpha * (1/m * delta_b[l])
        avg_cost = 1/m * avg_cost
        avg_cost_func.append(avg_cost)
        cnt += 1
    return W, b, avg_cost_func


def predict_y(W, b, X, n_layers):
    '''
    Réalise les prédictions de sortie
    '''
    m = X.shape[0]
    y = np.zeros((m,), dtype=int)
    for i in range(m):
        h, z = feed_forward(X[i, :], W, b)
        y[i] = np.argmax(h[n_layers])
    return y


def accuracy(y_test, y_pred, failed=False):
    '''
    Calcule la précision d'une prédiction on la comparant aux données réelles
    Renvoie un float entre 0 et 1
    '''
    if failed:
        failures = []
    count_correct_pred = 0
    m = y_test.shape[0]
    for i in range(m):
        if y_test[i] == y_pred[i]:
            count_correct_pred += 1
        elif failed:
            failures.append(i)
    if not failed:
        return count_correct_pred / m
    else:
        return count_correct_pred / m, failures


def display_failures(failures, test_input, test_output, y_pred):
    '''
    Affiche jusqu'à 10 images qui parmi celles demandées
    Présente aussi la prédiction
    '''
    for k in range(min(10, len(failures))):
        print("Predicted output : {}".format(y_pred[failures[k]]))
        display_selected_image(test_input,
                               test_output,
                               random=False,
                               selected_item=failures[k])


def classification_and_test(iter_num=3000):
    '''
    Extrait les données
    Normalise les données
    Sépare les données d'entraînement des données de test
    Réalise l'apprentissage du réseau de neurones
    Testes les résultats
    Affiche quelques informations finales :
        * fonction de coût,
        * Précision,
        * Affiche quelques entrées dont la prédiction est fausse.
    '''
    # extract the data
    with open(DATAPATH, newline='') as csvfile:
        digit_reader = csv.reader(csvfile, delimiter=',')
        data = list(digit_reader)
        data = np.array(data).astype(int)

    # print(data[0, :])

    # shuffle the data
    np.random.shuffle(data)

    # séparer les entrées des sorties avant normalisation
    output = data[:, -1]
    input = np.delete(data, -1, axis=1)
    input, _, __ = normalize_data(input)
    # print(input[0, :])

    # séparer train et test
    separation = int(TRAIN_PROP * len(data))
    train_input, test_input = input[:separation], input[separation:]
    train_output, test_output = output[:separation], output[separation:]

    # convert the outputs
    train_output_vect = convert_output_vector(train_output)
    test_output_vect = convert_output_vector(test_output)

    # entrainer
    W, b, avg_cost_func = train_nn(
        NN_STRUCTURE, train_input, train_output_vect,
        iter_num=iter_num
    )

    # évolution de la fonction de cout
    plt.plot(avg_cost_func)
    plt.ylabel('Average J')
    plt.xlabel('Itration Number')
    plt.show()

    # predictions
    y_pred = predict_y(W, b, test_input, 3)
    y_pred = np.array(y_pred)

    # precision
    print(y_pred[:50])
    print(test_output[:50])
    acc_score, failures = accuracy(test_output, y_pred, failed=True)
    print(acc_score)

    # display some failed predictions
    display_failures(failures, test_input, test_output, y_pred)


if __name__ == '__main__':
    classification_and_test(iter_num=3000)