#NETWORK ARCHITECTURES
import numpy as np
import os
import math
import tensorflow as tf
import matplotlib.pyplot as plt
from datetime import datetime
from architectures.utils.NN_building_blocks import *
def lrelu(x, alpha=0.1):
return tf.maximum(alpha*x,x)
# some dummy constants
LEARNING_RATE = None
LEARNING_RATE_D = None
LEARNING_RATE_G = None
BETA1 = None
BATCH_SIZE = None
EPOCHS = None
SAVE_SAMPLE_PERIOD = None
PATH = None
SEED = None
LAMBD = None
#seed=1
class classifier(object):
def __init__(self, X, sizes, c_name):
mi = X.get_shape().as_list()[1]
n_classes = sizes['n_classes']
self.dense_layers = []
with tf.variable_scope('classificator_'+c_name) as scope:
for count, (mo, apply_batch_norm, keep_prob, act_f, w_init) in enumerate(sizes['dense_layers'], start=0):
name = 'dense_layer_{0}'.format(count)
layer = DenseLayer(name,mi, mo,
apply_batch_norm, keep_prob,
act_f, w_init)
mi = mo
self.dense_layers.append(layer)
#readout layer
readout_layer = DenseLayer('readout_layer',
mi, n_classes,
'bn', 1., lambda x: x,
tf.random_normal_initializer())
self.dense_layers.append(readout_layer)
#return self.propagate(X)
def propagate(self, X, reuse=None, is_training=None):
print('Input for propagation', X.get_shape())
output = X
for layer in self.dense_layers:
output = layer.forward(output, reuse, is_training)
print('Logits shape', output.get_shape())
return output
#untested yet
class DNN(object):
"""
Builds densely connected deep neural network. Regularization implemented
with dropout, no regularization parameter implemented yet. Minimization through
AdamOptimizer (adaptive learning rate)
Constructor inputs:
-Positional arguments:
- dim: (int) input features
- sizes: (dict) python dictionary containing the size of the
dense layers and the number of classes for classification
sizes = {'dense_layer_n' :[mo, apply_batch_norm, keep_probability, act_f, w_init]
'n_classes': n_classes
}
mo: (int) size of layer n
apply_batch_norm: (bool) apply batch norm at layer n
keep_probability: (float32) probability of activation of output
act_f: (function) activation function for layer n
w_init: (tf initializer) random initializer for weights at layer n
n_classes: number of classes
-Keyword arguments
-lr: (float32) learning rate arg for the AdamOptimizer
-beta1: (float32) beta1 arg for the AdamOptimizer
-batch_size: (int) size of each batch
-epochs: (int) number of times the training has to be repeated over all the batches
-save_sample: (int) after how many iterations of the training algorithm performs the evaluations in fit function
-path: (str) path for saving the session checkpoint
Class attributes:
- X: (tf placeholder) input tensor of shape (batch_size, input features)
- Y: (tf placeholder) label tensor of shape (batch_size, n_classes) (one_hot encoding)
- Y_hat: (tf tensor) shape=(batch_size, n_classes) predicted class (one_hot)
- loss: (tf scalar) reduced mean of cost computed with softmax cross entropy with logits
- train_op: gradient descent algorithm with AdamOptimizer
"""
def __init__(self,
dim, sizes,
lr=LEARNING_RATE, beta1=BETA1, lambd=LAMBD,
batch_size=BATCH_SIZE, epochs=EPOCHS,
save_sample=SAVE_SAMPLE_PERIOD, path=PATH, seed=SEED):
self.seed=seed
self.n_classes = sizes['n_classes']
self.dim = dim
self.sizes=sizes
self.lambd=lambd
self.X_input = tf.placeholder(
tf.float32,
shape=(None, dim),
name = 'X_input'
)
self.X_test = tf.placeholder(
tf.float32,
shape=(None, dim),
name = 'X_test'
)
self.batch_sz=tf.placeholder(
tf.int32,
shape=(),
name='batch_sz',
)
self.Y_input = tf.placeholder(
tf.float32,
shape=(None, self.n_classes),
name='Y_input'
)
self.Y_test = tf.placeholder(
tf.float32,
shape=(None, self.n_classes),
name='Y_test'
)
NN = classifier(self.X_input, sizes, 'sig_bkg')
with tf.variable_scope('classificator_sig_bkg') as scope:
self.Y_hat = NN.propagate(self.X_input, reuse=None, is_training=True)
predicted_NN = tf.nn.softmax(self.Y_hat)
self.train_accuracy = evaluation(predicted_NN, self.Y_input)
cost = tf.nn.softmax_cross_entropy_with_logits(
logits= self.Y_hat,
labels= self.Y_input,
)
#cost = tf.square(predicted_NN-self.Y_input)
tot_params = [t for t in tf.trainable_variables() if 'classificator' in t.name]
temp =[t for t in tf.trainable_variables() if 'classificator' and not 'beta' in t.name]
weight_params =[t for t in temp if not 'gamma' in t.name]
self.regularizator=tf.reduce_sum([tf.nn.l2_loss(weight_params[i]) for i in range(len(weight_params))])
self.loss = tf.reduce_mean(cost) + self.lambd*self.regularizator
self.train_op = tf.train.AdamOptimizer(
learning_rate=lr,
beta1=beta1
).minimize(self.loss,
#var_list=tot_params
)
with tf.variable_scope('classificator_sig_bkg') as scope:
scope.reuse_variables()
self.Y_hat_from_test = NN.propagate(
self.X_test, reuse=True, is_training=False,
)
predicted_NN_test=tf.nn.softmax(self.Y_hat_from_test)
self.a=tf.reduce_sum(tf.argmax(predicted_NN_test, 1))/batch_size
self.test_accuracy = evaluation(self.Y_hat_from_test, self.Y_test)
#saving for later
self.lr = lr
self.batch_size=batch_size
self.epochs = epochs
self.path = path
self.save_sample = save_sample
self.NN = NN
def set_session(self, session):
self.session=session
for layer in self.NN.dense_layers:
layer.set_session(session)
def fit(self, X_train, Y_train, X_test, Y_test):
"""
Function is called if the flag is_training is set on TRAIN. If a model already is present
continues training the one already present, otherwise initialises all params from scratch.
Performs the training over all the epochs, at when the number of epochs of training
is a multiple of save_sample prints out training cost, train and test accuracies
Plots a plot of the cost versus epoch.
Positional arguments:
- X_train: (ndarray) size=(train set size, input features) training sample set
- X_test: (ndarray) size=(test set size, input features) test sample set
- Y_train: (ndarray) size=(train set size, input features) training labels set
- Y_test: (ndarray) size=(test set size, input features) test labels set
"""
seed = self.seed
N = X_train.shape[0]
test_size = X_test.shape[0]
n_batches = N // self.batch_size
print('\n ****** \n')
print('Training DNN for '+str(self.epochs)+' epochs with a total of ' +str(N)+ ' samples\ndistributed in ' +str(n_batches)+ ' batches of size '+str(self.batch_size)+'\n')
print('The learning rate set is '+str(self.lr))
print('\n ****** \n')
costs = []
for epoch in range(self.epochs):
# a=0
# train_acc = 0
# test_acc = 0
train_accuracies=[]
test_accuracies=[]
a_list=[]
seed += 1
train_batches = supervised_random_mini_batches(X_train, Y_train, self.batch_size, seed)
test_batches = supervised_random_mini_batches(X_test, Y_test, self.batch_size, seed)
for train_batch in train_batches:
(X_train_batch, Y_train_batch) = train_batch
# print(X_train, Y_train)
feed_dict = {
self.X_input: X_train_batch,
self.Y_input: Y_train_batch,
self.batch_sz: self.batch_size,
}
_, c = self.session.run(
(self.train_op, self.loss),
feed_dict=feed_dict
)
train_acc = self.session.run(
(self.train_accuracy),
feed_dict=feed_dict
)
costs.append(c)
train_accuracies.append(train_acc)
c = np.array(costs).mean()
train_acc = np.array(train_accuracies).mean()
if epoch % self.save_sample ==0:
for test_batch in test_batches:
(X_test_batch, Y_test_batch) = test_batch
feed_dict={
self.X_test: X_test_batch,
self.Y_test: Y_test_batch,
}
test_acc, a = self.session.run(
(self.test_accuracy, self.a),
feed_dict=feed_dict
)
a_list.append(a)
test_accuracies.append(test_acc)
print_a = np.array(a_list).mean()
test_acc = np.array(test_accuracies).mean()
print('Evaluating performance on validation/train sets')
print('At iteration {0}, train cost: {1:.4g}, train accuracy {2:.4g}, test accuracy {3:.4g}'.format(epoch, c, train_acc, test_acc))
print('Average of signal as predicted by NN {0:.4g}'.format(print_a))
plt.plot(costs)
plt.ylabel('cost')
plt.xlabel('iteration')
plt.title('learning rate=' + str(self.lr))
plt.show()
print('Parameters trained')
def test(self, X_test, Y_test):
test_accuracies=[]
test_batches = supervised_random_mini_batches(X_test, Y_test, self.batch_size, self.seed)
for test_batch in test_batches:
(X_test_batch, Y_test_batch) = test_batch
feed_dict={
self.X_test: X_test_batch,
self.Y_test: Y_test_batch,
}
test_acc = self.session.run(
self.test_accuracy,
feed_dict=feed_dict
)
test_accuracies.append(test_acc)
test_acc = np.array(test_accuracies).mean()
print('Test accuracy: {0:.4g}'.format(test_acc))
def predict(self, X):
pred = tf.nn.softmax(self.Y_hat_from_test)
output = self.session.run(
pred,
feed_dict={self.X_test:X}
)
return output
Davide Lancierini