#NEWLINE# def train( self , data_iterator ) : #NEWLINE# #INDENT# '\n        Train a keras model on a worker and send asynchronous updates\n        to parameter server\n        ' #NEWLINE# ( feature_iterator , label_iterator ) = tee ( data_iterator , 2 ) #NEWLINE# x_train = np . asarray ( [ x for ( x , y ) in feature_iterator ] ) #NEWLINE# y_train = np . asarray ( [ y for ( x , y ) in label_iterator ] ) #NEWLINE# if ( x_train . size == 0 ) : #NEWLINE# #INDENT# return #NEWLINE# #UNINDENT# model = model_from_yaml ( self . yaml ) #NEWLINE# nb_epoch = self . train_config [ 'nb_epoch' ] #NEWLINE# batch_size = self . train_config . get ( 'batch_size' ) #NEWLINE# nb_train_sample = len ( x_train [ 0 ] ) #NEWLINE# nb_batch = int ( np . ceil ( ( nb_train_sample / float ( batch_size ) ) ) ) #NEWLINE# index_array = np . arange ( nb_train_sample ) #NEWLINE# batches = [ ( ( i * batch_size ) , min ( nb_train_sample , ( ( i + 1 ) * batch_size ) ) ) for i in range ( 0 , nb_batch ) ] #NEWLINE# if ( self . frequency == 'epoch' ) : #NEWLINE# #INDENT# for epoch in range ( nb_epoch ) : #NEWLINE# #INDENT# weights_before_training = get_server_weights ( self . master_url ) #NEWLINE# model . set_weights ( weights_before_training ) #NEWLINE# self . train_config [ 'nb_epoch' ] = 1 #NEWLINE# if ( x_train . shape [ 0 ] > batch_size ) : #NEWLINE# #INDENT# model . fit ( x_train , y_train , show_accuracy = True , ** self . train_config ) #NEWLINE# #UNINDENT# weights_after_training = model . get_weights ( ) #NEWLINE# deltas = subtract_params ( weights_before_training , weights_after_training ) #NEWLINE# put_deltas_to_server ( deltas , self . master_url ) #NEWLINE# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# if ( self . frequency == 'batch' ) : #NEWLINE# #INDENT# for epoch in range ( nb_epoch ) : #NEWLINE# #INDENT# if ( x_train . shape [ 0 ] > batch_size ) : #NEWLINE# #INDENT# for ( batch_start , batch_end ) in batches : #NEWLINE# #INDENT# weights_before_training = get_server_weights ( self . master_url ) #NEWLINE# model . set_weights ( weights_before_training ) #NEWLINE# batch_ids = index_array [ batch_start : batch_end ] #NEWLINE# X = slice_X ( x_train , batch_ids ) #NEWLINE# y = slice_X ( y_train , batch_ids ) #NEWLINE# model . train_on_batch ( X , y ) #NEWLINE# weights_after_training = model . get_weights ( ) #NEWLINE# deltas = subtract_params ( weights_before_training , weights_after_training ) #NEWLINE# put_deltas_to_server ( deltas , self . master_url ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# print ( 'Choose frequency to be either batch or epoch' ) #NEWLINE# #UNINDENT# #UNINDENT# ( yield [ ] )