#!/usr/bin/python ''' Learning Machines Taught by Patrick Hebron at NYU ITP Unrolled Multilayer Perceptron implementation. * Provides one hidden layer. * Explicitly shows ''' import numpy as np import matplotlib.pyplot as plt from matplotlib.lines import Line2D class MlpVisualizer: def __init__(self,data_xmin,data_xmax,data_ymin,data_ymax,report_freq,buffer_size = 100): self.report_freq = report_freq self.error_buffer = buffer_size # Setup plotter data: self.error_xdata = [] self.error_ydata = [] # Setup plotter: plt.ion() self.fig = plt.figure( 1 ) self.fig.subplots_adjust( hspace = 0.3 ) # Add subplots: self.datav_plot = self.fig.add_subplot( 2, 1, 1 ) self.error_plot = self.fig.add_subplot( 2, 1, 2 ) # Setup predictions subplot: self.datav_plot.set_title( 'Predictions' ) self.datav_targ_line = Line2D( [], [], color='green', marker='+', linestyle='None' ) self.datav_pred_line = Line2D( [], [], color='red', marker='x', linestyle='None' ) self.datav_plot.add_line( self.datav_targ_line ) self.datav_plot.add_line( self.datav_pred_line ) self.datav_plot.set_xlim( data_xmin, data_xmax ) self.datav_plot.set_ylim( data_ymin, data_ymax ) # Setup error rate subplot: self.error_plot.set_xlabel( 'Epoch' ) self.error_plot.set_ylabel( 'Error' ) self.error_line = Line2D( [], [], color='black' ) self.error_plot.add_line( self.error_line ) self.error_plot.set_ylim( 0.0, 1.0 ) # Show plot: plt.show() def saveImage(self,filepath): plt.savefig( filepath ) def update(self,epoch,error,input,target,output): # Update error plotter data: if len( self.error_xdata ) == self.error_buffer: self.error_xdata.pop( 0 ) self.error_ydata.pop( 0 ) self.error_xdata.append( epoch ) self.error_ydata.append( error ) # title = 'Epoch: %d, Error: %f' % ( epoch, error ) self.error_plot.set_title( title ) # Compute error plotter x-range: mlen = self.report_freq * self.error_buffer xmin = np.amin( self.error_xdata ) xmax = max( xmin + mlen, np.amax( self.error_xdata ) ) # Update error plotter: self.error_line.set_data( self.error_xdata, self.error_ydata ) self.error_plot.set_xlim( xmin, xmax ) # Update predictions plotter: self.datav_targ_line.set_data( input, target ) self.datav_pred_line.set_data( input, output ) # Draw plot: plt.draw() plt.pause( 0.01 ) def sigmoid_fn(x): return 1.0 / ( 1.0 + np.exp( -x ) ) def sigmoid_dfn(x): y = sigmoid_fn( x ) return y * ( 1.0 - y ) def tanh_fn(x): return np.sinh( x ) / np.cosh( x ) def tanh_dfn(x): return 1.0 - np.power( tanh_fn( x ), 2.0 ) activation_fn = tanh_fn activation_dfn = tanh_dfn report_freq = 10 learning_rate = 0.05 sample_size = 1 hidden_size = 15 output_size = 1 batch_size = 10 num_examples = 300 num_epochs = 10000 batch_mult = learning_rate * ( 1.0 / float( num_examples ) ) # Initialize weights and biases: layer0_weights = np.random.rand( hidden_size, sample_size ) layer0_bias = np.random.rand( hidden_size, 1 ) layer1_weights = np.random.rand( output_size, hidden_size ) layer1_bias = np.random.rand( output_size, 1 ) # # Choose a random input: sample_vec = np.random.uniform( 0.0, np.pi * 2.0, ( sample_size, num_examples ) ) # Compute expected output: output_vec = np.sin( sample_vec ) vis = MlpVisualizer( np.amin( sample_vec ), np.amax( sample_vec ), np.amin( output_vec ), np.amax( output_vec ), report_freq ) # Perform each epoch: for epoch in range( num_epochs ): # Reset error accumulator: error = 0.0 # Iterate over each example: for batch_start in range( 0, num_examples, batch_size ): # Compute batch stop index: batch_stop = min( batch_start + batch_size, num_examples ) # Perform training epoch on batch: #batch_error = self.trainEpoch( input[ :, batch_start:batch_stop ], target[ :, batch_start:batch_stop ], learn_rate ) # Add scaled batch error to total error: # error += batch_error * ( float( batch_stop - batch_start ) / float( num_examples ) ) sample_batch = sample_vec[ :, batch_start:batch_stop ] output_batch = output_vec[ :, batch_start:batch_stop ] # Feed forward (input to hidden): layer1_activations = np.dot( layer0_weights, sample_batch ) + layer0_bias layer1_outputs = activation_fn( layer1_activations ) # Feed forward (hidden to output): layer2_activations = np.dot( layer1_weights, layer1_outputs ) + layer1_bias layer2_outputs = activation_fn( layer2_activations ) # Back propagate (output to hidden): layer2_deltas = activation_dfn( layer2_activations ) * ( layer2_outputs - output_batch ) # Back propagate (hidden to input): layer1_deltas = activation_dfn( layer1_activations ) * np.dot( layer1_weights.T, layer2_deltas ) # Apply deltas (layer 0): layer0_weights -= batch_mult * np.dot( layer1_deltas, sample_batch.T ) layer0_bias -= batch_mult * np.expand_dims( np.sum( layer1_deltas, axis=1 ), axis=1 ) # Apply deltas (layer 1): layer1_weights -= batch_mult * np.dot( layer2_deltas, layer1_outputs.T ) layer1_bias -= batch_mult * np.expand_dims( np.sum( layer2_deltas, axis=1 ), axis=1 ) # Add current error to accumulator: error += np.sum( np.absolute( output_batch - layer2_outputs ) ) * ( float( batch_stop - batch_start ) / float( num_examples ) ) # Report error rate: if ( epoch + 1 ) % report_freq == 0: print "Epoch: %d\nError: %f" % ( epoch + 1, error ) # Feed forward (input to hidden): layer1_activations = np.dot( layer0_weights, sample_batch ) + layer0_bias layer1_outputs = activation_fn( layer1_activations ) # Feed forward (hidden to output): layer2_activations = np.dot( layer1_weights, layer1_outputs ) + layer1_bias layer2_outputs = activation_fn( layer2_activations ) vis.update( epoch, error, sample_vec, output_vec, layer2_outputs )