Merge branch 'TP5' into main

TP5/.gitignore

@@ -0,0 +1,2 @@
+*.csv
+*.png

TP5/README.md

@@ -0,0 +1,15 @@
+# TP5
+
+## 72.27 - Sistemas de Inteligencia Artificial - 2º cuatrimestre 2022
+
+### Instituto Tecnológico de Buenos Aires (ITBA)
+
+## Autores
+
+- [Sicardi, Julián Nicolas](https://github.com/Jsicardi) - Legajo 60347
+- [Quintairos, Juan Ignacio](https://github.com/juaniq99) - Legajo 59715
+- [Zavalia Pángaro, Salustiano Jose](https://github.com/szavalia) - Legajo 60312
+
+## Índice
+- [Autores](#autores)
+- [Índice](#índice)

TP5/algorithms/autoencoder.py

@@ -0,0 +1,139 @@
+from models import Autoencoder, Observables, Properties
+import numpy as np
+import scipy.optimize as sco
+from algorithms.noiser import noise_font
+from io_parser import generate_output_file,generate_samples_file
+from autograd.misc.optimizers import adam 
+import numdifftools as nd 
+
+def execute(properties:Properties):
+    # Create autoencoder
+    autoencoder:Autoencoder = build_autoencoder(properties)
+
+    # Train
+    trained_weigths = train_autoencoder(autoencoder,properties)
+
+    autoencoder.weights = autoencoder.unflatten_weights(trained_weigths)
+
+    # After training: 
+    # Get outputs for inputs
+    unflattened = autoencoder.unflatten_weights(trained_weigths)
+    output = autoencoder.get_output(properties.training_set, unflattened)
+
+    # If mode was DAE, get new noised font and see how well it denoises
+    noised_font = None
+    noised_output = None
+    if properties.mode == "DAE":
+        noised_font = noise_font(properties.orig_training_set, properties.noise_prob)
+        noised_output = []
+        for letter in noised_font:
+            noised_output.append(autoencoder.get_output(letter,unflattened))
+        # print error for noised font
+        print("\nError for new noised font: " + str(autoencoder.error_given_sets(trained_weigths, noised_font)))
+
+    generate_output_file(properties.mode,properties.training_set,output,noise_font,noised_output)
+    latent_outputs = get_latent_outputs(autoencoder,properties)
+
+    if properties.mode == "DEFAULT" and properties.neurons_per_layer[autoencoder.latent_index] == 2:
+        distances = get_distances(latent_outputs)
+        pairs = [distances[0][:2],distances[int(len(distances)/2)][:2],distances[-1][:2]]
+
+        samples = []
+        char_pairs = []
+        for pair_indexes in pairs:
+            value1 = latent_outputs[pair_indexes[0]]
+            value2 = latent_outputs[pair_indexes[1]]
+            char_pairs.append([properties.font_chars[pair_indexes[0]],properties.font_chars[pair_indexes[1]]])
+            samples.append(get_samples(value1,value2,autoencoder))
+
+        generate_samples_file(char_pairs,samples)
+
+    return Observables(autoencoder.errors_per_step,latent_outputs)
+
+def build_autoencoder(properties:Properties):
+    weights = []
+
+    # Add layers
+    for layer_index, neurons_count in enumerate(properties.neurons_per_layer):
+        weights.append([])  
+        for index in range(0, neurons_count):
+            if layer_index == 0: 
+                w = np.random.randn(len(properties.training_set[0]))
+            else:
+                w = np.random.randn(len(weights[-2]))
+            weights[-1].append(w)
+
+    # Add output layer of decoder
+    # Number of neurons in output layer depends on number of camps in expected output values
+    weights.append([])
+    for i in range(len(properties.output_set[0])):
+        w = np.random.randn(len(weights[-2]))
+        weights[-1].append(w)
+
+    # Convert to ndarray
+    for (i, layer) in enumerate(weights):
+        weights[i] = np.array(layer, dtype=float)
+
+    return Autoencoder(np.array(weights, dtype=object),int((len(properties.neurons_per_layer))/2),properties.training_set,properties.output_set,properties.neurons_per_layer)
+
+def flatten_weights(weigths):
+    flattened_weights = np.array([])
+    for (i,layer) in enumerate(weigths):
+        flattened_weights =  np.append(flattened_weights,layer.flatten())
+    return flattened_weights.flatten() 
+
+def train_autoencoder(autoencoder:Autoencoder,properties:Properties):
+    flattened_weights = flatten_weights(autoencoder.weights)
+
+    # Optimize
+    if properties.minimizer == "powell":
+        trained_weights = sco.minimize(
+                autoencoder.error, flattened_weights, method='Powell', callback=autoencoder.callback, 
+                options={'maxiter': properties.epochs}
+            ).x
+    elif properties.minimizer == "adam":
+        trained_weights = adam(nd.Gradient(autoencoder.error), flattened_weights, callback=autoencoder.callback, num_iters=properties.epochs, step_size=0.1, b1=0.9, b2=0.999, eps=10**-2)    
+
+    return trained_weights
+
+def get_latent_outputs(autoencoder:Autoencoder,properties:Properties):
+    return autoencoder.get_latent_output(properties.training_set)
+
+def get_decoder_outputs(autoencoder:Autoencoder,samples):
+    return autoencoder.get_decoder_output(samples)
+
+def get_distances(points):
+    distances = []
+    for (i,value) in enumerate(points):
+        for j in range(i+1, len(points)):
+            distances.append(np.array([int(i),int(j),np.linalg.norm(value - points[j])], dtype = object))
+    distances = np.array(distances, dtype = object)
+    return distances[distances[:, 2].argsort()]
+
+def line(m,b,x):
+    return m*x - b
+
+def get_samples(point1,point2,autoencoder:Autoencoder):
+    m = (point1[1]-point2[1]) / (point1[0]-point2[0])
+    b = (point1[0]*point2[1] - point2[0]*point1[1])/(point1[0]-point2[0])
+
+    min_x = min(point1[0],point2[0])
+    max_x = max(point1[0],point2[0])
+    inputs = []
+
+    print(np.linspace(min_x,max_x,5))
+    for x in np.linspace(min_x,max_x,5):
+        y = line(m,b,x)
+        inputs.append(np.array([x,y]))
+
+    inputs = np.array(inputs)
+
+    #print(inputs)
+
+    outputs = autoencoder.get_decoder_output(inputs)
+
+    return outputs
+
+
+
+

TP5/algorithms/noiser.py

@@ -0,0 +1,19 @@
+import random
+
+# Adds noise to all elements in font, with probability prob to switch binary value
+def noise_font(font, prob):
+    new_font = []
+    for letter in font:
+        new_font.append(noise_elem(letter, prob))
+    return new_font
+
+
+# Adds noise to single character image
+def noise_elem(elem, prob):
+    new_elem = []
+    for bit in elem:
+        if random.random() < prob:
+            new_elem.append(1 - bit)
+        else:
+            new_elem.append(bit)
+    return new_elem

TP5/algorithms/vae.py

@@ -0,0 +1,130 @@
+from models import Properties, VAEObservables
+from keras.layers import Input, Dense, Lambda, Reshape
+from keras.models import Model
+from keras import backend as K
+from keras import metrics
+import tensorflow as tf
+import numpy as np
+from keras.datasets import mnist,fashion_mnist
+import matplotlib.pyplot as plt
+from scipy.stats import norm
+
+from tensorflow.python.framework.ops import disable_eager_execution
+disable_eager_execution()
+
+class VAE:
+    def __init__(self,latent_neurons,dim,intermediate_layers):
+        self.latent_neurons = latent_neurons
+        self.dim = dim
+        self.intermediate_layers = intermediate_layers
+        self.set_vae()
+
+    def train(self,training_set,epochs,batch_size):
+        self.model.fit(training_set,training_set,epochs=epochs)
+
+    def set_vae(self):
+        # input to our encoder
+        x = Input(shape=(self.dim,), name="input")
+        self.encoder = self.set_encoder(x)
+        # print out summary of what we just did
+        self.encoder.summary()
+        self.decoder = self.set_decoder()
+        self.decoder.summary()
+        # grab the output. Recall, that we need to grab the 3rd element our sampling z
+        output_combined = self.decoder(self.encoder(x)[2])
+        # link the input and the overall output
+        self.model = Model(x, output_combined)
+        # print out what the overall model looks like
+        self.model.summary()
+        self.model.compile(loss=self.vae_loss)
+
+    def set_encoder(self,x):
+        # intermediate layers
+        if(len(self.intermediate_layers) != 0):
+            aux_h = x
+            for (i,neurons) in enumerate(self.intermediate_layers):
+                h = Dense(neurons,activation='relu', name="encoding_{0}".format(i))(aux_h)
+                aux_h = h
+        # defining the mean of the latent space
+        self.z_mean = Dense(self.latent_neurons, name="mean")(h)
+        # defining the log variance of the latent space
+        self.z_log_var = Dense(self.latent_neurons, name="log-variance")(h)
+        # note that "output_shape" isn't necessary with the TensorFlow backend
+        z = Lambda(self.get_samples, output_shape=(self.latent_neurons,))([self.z_mean, self.z_log_var])
+        # defining the encoder as a keras model
+        encoder = Model(x, [self.z_mean, self.z_log_var, z], name="encoder")
+        return encoder
+
+    def set_decoder(self):
+        # Input to the decoder
+        input_decoder = Input(shape=(self.latent_neurons,), name="decoder_input")
+        # intermediate layers
+        reversed_layers = self.intermediate_layers.copy()
+        reversed_layers.reverse()
+        if(len(self.intermediate_layers) != 0):
+            aux_h = input_decoder
+            for (i,neurons) in enumerate(reversed_layers):
+                h = Dense(neurons,activation='relu',name="encoding_{0}".format(i))(aux_h)
+                aux_h = h
+        #getting the mean from the original dimension
+        x_decoded = Dense(self.dim, activation='sigmoid', name="flat_decoded")(h)
+        # defining the decoder as a keras model
+        decoder = Model(input_decoder, x_decoded, name="decoder")
+        return decoder
+
+    def get_samples(self,args: tuple):
+        # we grab the variables from the tuple
+        z_mean, z_log_var = args
+        print(z_mean)
+        print(z_log_var)
+        epsilon = K.random_normal(shape=(K.shape(z_mean)[0], self.latent_neurons), mean=0.,stddev=1.0)
+        return z_mean + K.exp(z_log_var / 2) * epsilon  # h(z)
+
+    def vae_loss(self,x: tf.Tensor, x_decoded_mean: tf.Tensor):
+        # Aca se computa la cross entropy entre los "labels" x que son los valores 0/1 de los pixeles, y lo que salió al final del Decoder.
+        xent_loss = self.dim * metrics.binary_crossentropy(x, x_decoded_mean) # x-^X
+        kl_loss = - 0.5 * K.sum(1 + self.z_log_var - K.square(self.z_mean) - K.exp(self.z_log_var), axis=-1)
+        vae_loss = K.mean(xent_loss + kl_loss)
+        return vae_loss
+
+def generate_samples(dataset,vae:VAE):
+    n = 15  # figure with 15x15 digits
+    digit_size = 28
+    figure = np.zeros((digit_size * n, digit_size * n))
+    # linearly spaced coordinates on the unit square were transformed through the inverse CDF (ppf) of the Gaussian
+    # to produce values of the latent variables z, since the prior of the latent space is Gaussian
+    grid_x = norm.ppf(np.linspace(0.05, 0.95, n))
+    grid_y = norm.ppf(np.linspace(0.05, 0.95, n))
+
+    for i, yi in enumerate(grid_x):
+        for j, xi in enumerate(grid_y):
+            z_sample = np.array([[xi, yi]])
+            x_decoded = vae.decoder.predict(z_sample)
+            digit = x_decoded[0].reshape(digit_size, digit_size)
+            figure[i * digit_size: (i + 1) * digit_size,
+               j * digit_size: (j + 1) * digit_size] = digit
+
+    plt.figure(figsize=(10, 10))
+    plt.imshow(figure, cmap='Greys_r')
+    if(dataset == "mnist"):
+        plt.savefig("mnist.png")
+    else:
+        plt.savefig("fashion_mnist.png")
+
+def execute(properties:Properties):
+    if(properties.VAE_dataset == "mnist"):
+        (x_train, y_train), (x_test, y_test) = mnist.load_data()
+    else:
+        (x_train, y_train), (x_test, y_test) = fashion_mnist.load_data()
+    x_train = x_train.astype('float32') / 255.
+    x_test = x_test.astype('float32') / 255.
+    x_train = x_train.reshape((len(x_train), np.prod(x_train.shape[1:])))
+    x_test = x_test.reshape((len(x_test), np.prod(x_test.shape[1:])))
+    vae = VAE(2,28*28,[256])
+    vae.train(x_train,properties.epochs,100)
+    latent_outputs = vae.encoder.predict(x_test, batch_size=100)[0]
+    generate_samples(properties.VAE_dataset,vae)
+    return VAEObservables(latent_outputs,y_test)
+
+def flatten_set(training_set):
+    return np.array(training_set).reshape((len(training_set), np.prod(np.array(training_set).shape[1:])))

TP5/config.json

@@ -0,0 +1,12 @@
+{
+    "mode": "DEFAULT",
+    "vae_dataset" : "fashion_mnist",
+    "minimizer": "powell",
+    "noise_probability": 0.1,
+    "neurons_per_layer": [20], 
+    "latent_layer_neurons": 2, 
+    "font_set": "font2",
+    "font_subset_size": 10,
+    "beta": 1,
+    "epochs": 10
+}