Merge branch 'master' of https://github.com/PPPLDeepLearning/plasma-python

Author: ASvyatkovskiy

examples/hyper_learn.py

@@ -0,0 +1,42 @@
+from __future__ import print_function
+
+import numpy as np
+from hyperopt import Trials, tpe
+
+from plasma.conf import conf
+from pprint import pprint
+pprint(conf)
+#from plasma.primitives.shots import Shot, ShotList
+from plasma.preprocessor.normalize import Normalizer
+from plasma.models.loader import Loader
+#from plasma.models.runner import train, make_predictions,make_predictions_gpu
+
+if conf['data']['normalizer'] == 'minmax':
+    from plasma.preprocessor.normalize import MinMaxNormalizer as Normalizer
+elif conf['data']['normalizer'] == 'meanvar':
+    from plasma.preprocessor.normalize import MeanVarNormalizer as Normalizer
+elif conf['data']['normalizer'] == 'var':
+    from plasma.preprocessor.normalize import VarNormalizer as Normalizer #performs !much better than minmaxnormalizer
+elif conf['data']['normalizer'] == 'averagevar':
+    from plasma.preprocessor.normalize import AveragingVarNormalizer as Normalizer #performs !much better than minmaxnormalizer
+else:
+    print('unkown normalizer. exiting')
+    exit(1)
+
+np.random.seed(1)
+
+print("normalization",end='')
+nn = Normalizer(conf)
+nn.train()
+loader = Loader(conf,nn)
+shot_list_train,shot_list_validate,shot_list_test = loader.load_shotlists(conf)
+print("...done")
+
+print('Training on {} shots, testing on {} shots'.format(len(shot_list_train),len(shot_list_test)))
+from plasma.models import runner
+
+specific_runner = runner.HyperRunner(conf,loader,shot_list_train)
+
+best_run, best_model = specific_runner.frnn_minimize(algo=tpe.suggest,max_evals=2,trials=Trials())
+print (best_run)
+print (best_model)

plasma/conf.py

@@ -8,5 +8,7 @@
     conf = parameters('./conf.yaml')
 elif os.path.exists('./examples/conf.yaml'):
     conf = parameters('./examples/conf.yaml')
+elif os.path.exists('../examples/conf.yaml'):
+    conf = parameters('../examples/conf.yaml')    
 else:
     raise FileNotFoundError(errno.ENOENT, os.strerror(errno.ENOENT), 'conf.yaml')

plasma/models/builder.py

@@ -20,7 +20,7 @@ def on_batch_end(self, batch, logs=None):
         self.losses.append(logs.get('loss'))
 
 
-class ModelBuilder():
+class ModelBuilder(object):
 	def __init__(self,conf):
 		self.conf = conf
 
@@ -32,7 +32,6 @@ def get_unique_id(self):
 		unique_id =  hash(dill.dumps(this_conf))
 		return unique_id
 
-
 	def build_model(self,predict,custom_batch_size=None):
 		conf = self.conf
 		model_conf = conf['model']
@@ -158,3 +157,81 @@ def get_all_saved_files(self):
 			if curr_id == unique_id:
 				epochs.append(epoch)
 		return epochs
+
+
+	#FIXME this is essentially the ModelBuilder.build_model
+        #in the long run we want to replace the space dictionary with the 
+        #regular conf file - I am sure there is a way to accomodate 
+	def hyper_build_model(self,space,predict,custom_batch_size=None):
+		conf = self.conf
+		model_conf = conf['model']
+		rnn_size = model_conf['rnn_size']
+		rnn_type = model_conf['rnn_type']
+		optimizer = model_conf['optimizer']
+		lr = model_conf['lr']
+		clipnorm = model_conf['clipnorm']
+		regularization = model_conf['regularization']
+
+		if optimizer == 'sgd':
+			optimizer_class = SGD
+		elif optimizer == 'adam':
+			optimizer_class = Adam
+		elif optimizer == 'rmsprop':
+			optimizer_class = RMSprop
+		elif optimizer == 'nadam':
+			optimizer_class = Nadam
+		else:
+			optimizer = optimizer
+
+		if lr is not None or clipnorm is not None:
+			optimizer = optimizer_class(lr = lr,clipnorm=clipnorm)
+
+		loss_fn = conf['data']['target'].loss#model_conf['loss']
+		dropout_prob = model_conf['dropout_prob']
+		length = model_conf['length']
+		pred_length = model_conf['pred_length']
+		skip = model_conf['skip']
+		stateful = model_conf['stateful']
+		return_sequences = model_conf['return_sequences']
+		output_activation = conf['data']['target'].activation#model_conf['output_activation']
+		num_signals = conf['data']['num_signals']
+
+
+		batch_size = self.conf['training']['batch_size']
+		if predict:
+			batch_size = self.conf['model']['pred_batch_size']
+			#so we can predict with one time point at a time!
+			if return_sequences:
+				length =pred_length
+			else:
+				length = 1
+
+		if custom_batch_size is not None:
+			batch_size = custom_batch_size
+
+		if rnn_type == 'LSTM':
+			rnn_model = LSTM
+		elif rnn_type == 'SimpleRNN':
+			rnn_model =SimpleRNN
+		else:
+			print('Unkown Model Type, exiting.')
+			exit(1)
+    	
+		batch_input_shape=(batch_size,length, num_signals)
+		model = Sequential()
+
+		for _ in range(model_conf['rnn_layers']):
+			model.add(rnn_model(rnn_size, return_sequences=return_sequences,batch_input_shape=batch_input_shape,
+				stateful=stateful,W_regularizer=l2(regularization),U_regularizer=l2(regularization),
+				b_regularizer=l2(regularization),dropout_W=dropout_prob,dropout_U=dropout_prob))
+			model.add(Dropout(space['Dropout']))
+		if return_sequences:
+			model.add(TimeDistributed(Dense(1,activation=output_activation)))
+		else:
+			model.add(Dense(1,activation=output_activation))
+		model.compile(loss=loss_fn, optimizer=optimizer)
+		model.reset_states()
+		#model.compile(loss='mean_squared_error', optimizer='sgd') #for numerical output
+
+		return model
+

plasma/models/mpi_runner.py

@@ -139,7 +139,7 @@ def get_val(self):
 
 
 class MPIModel():
-  def __init__(self,model,optimizer,comm,batch_iterator,batch_size,num_replicas=None,warmup_steps=1000,lr=0.01):
+  def __init__(self,model,optimizer,comm,batch_iterator,batch_size,num_replicas=None,warmup_steps=1000,lr=0.01,custom_num_workers=0):
     # random.seed(task_index)
     self.epoch = 0
     self.model = model
@@ -151,11 +151,16 @@ def __init__(self,model,optimizer,comm,batch_iterator,batch_size,num_replicas=No
     self.batch_size = batch_size
     self.batch_iterator = batch_iterator
     self.warmup_steps=warmup_steps
-    self.num_workers = comm.Get_size()
+    if custom_num_workers:
+        if custom_num_workers < comm.Get_size():
+            self.num_workers = custom_num_workers
+        else: self.num_workers = comm.Get_size()
+    else:
+        self.num_workers = comm.Get_size()
     self.task_index = comm.Get_rank()
     self.history = cbks.History()
     if num_replicas is None or num_replicas < 1 or num_replicas > self.num_workers:
-      self.num_replicas = num_workers
+      self.num_replicas = self.num_workers
     else:
       self.num_replicas = num_replicas
 

plasma/models/runner.py

@@ -6,6 +6,9 @@
 import numpy as np
 from itertools import imap
 
+from hyperopt import hp, STATUS_OK
+from hyperas.distributions import conditional
+
 import time
 import sys
 import os
@@ -17,7 +20,6 @@
 from plasma.utils.performance import PerformanceAnalyzer
 from plasma.utils.evaluation import *
 
-
 def train(conf,shot_list_train,loader):
 
     np.random.seed(1)
@@ -99,8 +101,102 @@ def train(conf,shot_list_train,loader):
         plot_losses(conf,[training_losses,validation_losses,validation_roc],specific_builder,name='training_validation_roc')
     print('...done')
 
-
-
+class HyperRunner(object):
+    def __init__(self,conf,loader,shot_list):
+        self.loader = loader
+        self.shot_list = shot_list
+        self.conf = conf
+
+    #FIXME setup for hyperas search
+    def keras_fmin_fnct(self,space):
+        from plasma.models import builder
+
+        specific_builder = builder.ModelBuilder(self.conf)
+
+        train_model, test_model = specific_builder.hyper_build_model(space,False), specific_builder.hyper_build_model(space,True)
+
+        np.random.seed(1)
+        validation_losses = []
+        validation_roc = []
+        training_losses = []
+        shot_list_train,shot_list_validate = self.shot_list.split_direct(1.0-conf['training']['validation_frac'],do_shuffle=True)
+        os.environ['THEANO_FLAGS'] = 'device=gpu,floatX=float32'
+	import theano
+	from keras.utils.generic_utils import Progbar
+	from keras import backend as K
+
+
+        num_epochs = self.conf['training']['num_epochs']
+        num_at_once = self.conf['training']['num_shots_at_once']
+        lr_decay = self.conf['model']['lr_decay']
+        lr = self.conf['model']['lr']
+
+        resulting_dict = {'loss':None,'status':STATUS_OK,'model':None}
+
+        e = -1
+        #print("Current num_epochs {}".format(e))
+        while e < num_epochs-1:
+            e += 1
+            pbar =  Progbar(len(shot_list_train))
+
+            shot_list_train.shuffle()
+            shot_sublists = shot_list_train.sublists(num_at_once)[:1]
+            training_losses_tmp = []
+
+            K.set_value(train_model.optimizer.lr, lr*lr_decay**(e))
+            for (i,shot_sublist) in enumerate(shot_sublists):
+                X_list,y_list = self.loader.load_as_X_y_list(shot_sublist)
+                for j,(X,y) in enumerate(zip(X_list,y_list)):
+                    history = builder.LossHistory()
+                    train_model.fit(X,y,
+                        batch_size=Loader.get_batch_size(self.conf['training']['batch_size'],prediction_mode=False),
+                        nb_epoch=1,shuffle=False,verbose=0,
+                        validation_split=0.0,callbacks=[history])
+                    train_model.reset_states()
+                    train_loss = np.mean(history.losses)
+                    training_losses_tmp.append(train_loss)
+
+                    pbar.add(1.0*len(shot_sublist)/len(X_list), values=[("train loss", train_loss)])
+                    self.loader.verbose=False
+            sys.stdout.flush()
+            training_losses.append(np.mean(training_losses_tmp))
+            specific_builder.save_model_weights(train_model,e)
+
+            roc_area,loss = make_predictions_and_evaluate_gpu(self.conf,shot_list_validate,self.loader)
+            print("Epoch: {}, loss: {}, validation_losses_size: {}".format(e,loss,len(validation_losses)))
+            validation_losses.append(loss)
+            validation_roc.append(roc_area)
+            resulting_dict['loss'] = loss
+            resulting_dict['model'] = train_model
+            #print("Results {}, before {}".format(resulting_dict,id(resulting_dict)))
+
+        #print("Results {}, after {}".format(resulting_dict,id(resulting_dict)))
+        return resulting_dict
+
+    def get_space(self):
+        return {
+            'Dropout': hp.uniform('Dropout', 0, 1),
+        }
+
+    def frnn_minimize(self, algo, max_evals, trials, rseed=1337):
+	from hyperopt import fmin
+
+        best_run = fmin(self.keras_fmin_fnct,
+                    space=self.get_space(),
+                    algo=algo,
+                    max_evals=max_evals,
+                    trials=trials,
+                    rstate=np.random.RandomState(rseed))
+
+        best_model = None
+        for trial in trials:
+            vals = trial.get('misc').get('vals')
+            for key in vals.keys():
+                vals[key] = vals[key][0]
+            if trial.get('misc').get('vals') == best_run and 'model' in trial.get('result').keys():
+                best_model = trial.get('result').get('model')
+
+        return best_run, best_model
 
 def plot_losses(conf,losses_list,specific_builder,name=''):
     unique_id = specific_builder.get_unique_id()
@@ -140,7 +236,8 @@ def make_predictions(conf,shot_list,loader):
     model_save_path = specific_builder.get_latest_save_path()
 
     start_time = time.time()
-    pool = mp.Pool()
+    use_cores = max(1,mp.cpu_count()-2)
+    pool = mp.Pool(use_cores)
     fn = partial(make_single_prediction,builder=specific_builder,loader=loader,model_save_path=model_save_path)
 
     print('running in parallel on {} processes'.format(pool._processes))