<s> import numpy <EOL> import theano <EOL> from fuel . datasets import IterableDataset <EOL> from numpy . testing import assert_allclose <EOL> from theano import tensor <EOL> from blocks . extensions import TrainingExtension , FinishAfter <EOL> from blocks . extensions . monitoring import ( <EOL> MonitoringExtension , <EOL> TrainingDataMonitoring ) <EOL> from blocks . monitoring import aggregation <EOL> from blocks . algorithms import GradientDescent , Scale <EOL> from blocks . utils import shared_floatx <EOL> from blocks . main_loop import MainLoop <EOL> class MeanFeaturesTimesTarget ( aggregation . MonitoredQuantity ) : <EOL> def initialize ( self ) : <EOL> self . _aggregated = <NUM_LIT:0.> <EOL> self . _num_batches = <NUM_LIT:0> <EOL> def aggregate ( self , features , targets ) : <EOL> self . _aggregated += features * targets <EOL> self . _num_batches += <NUM_LIT:1> <EOL> def get_aggregated_value ( self ) : <EOL> return self . _aggregated / self . _num_batches <EOL> def test_monitoring_extension__record_name ( ) : <EOL> test_name = "<STR_LIT>" <EOL> monitor = MonitoringExtension ( ) <EOL> assert monitor . _record_name ( test_name ) == test_name <EOL> monitor = MonitoringExtension ( prefix = "<STR_LIT:abc>" ) <EOL> assert ( monitor . _record_name ( test_name ) == <EOL> "<STR_LIT:abc>" + monitor . SEPARATOR + test_name ) <EOL> monitor = MonitoringExtension ( suffix = "<STR_LIT:abc>" ) <EOL> assert ( monitor . _record_name ( test_name ) == <EOL> test_name + monitor . SEPARATOR + "<STR_LIT:abc>" ) <EOL> monitor = MonitoringExtension ( prefix = "<STR_LIT:abc>" , suffix = "<STR_LIT>" ) <EOL> assert ( monitor . _record_name ( test_name ) == <EOL> "<STR_LIT:abc>" + monitor . SEPARATOR + test_name + monitor . SEPARATOR + "<STR_LIT>" ) <EOL> try : <EOL> monitor = MonitoringExtension ( prefix = "<STR_LIT:abc>" , suffix = "<STR_LIT>" ) <EOL> monitor . _record_name ( None ) <EOL> except ValueError as e : <EOL> assert str ( e ) == "<STR_LIT>" <EOL> def test_training_data_monitoring ( ) : <EOL> weights = numpy . array ( [ - <NUM_LIT:1> , <NUM_LIT:1> ] , dtype = theano . config . floatX ) <EOL> features = [ numpy . array ( f , dtype = theano . config . floatX ) <EOL> for f in [ [ <NUM_LIT:1> , <NUM_LIT:2> ] , [ <NUM_LIT:3> , <NUM_LIT:5> ] , [ <NUM_LIT:5> , <NUM_LIT:8> ] ] ] <EOL> targets = numpy . array ( [ ( weights * f ) . sum ( ) for f in features ] ) <EOL> n_batches = <NUM_LIT:3> <EOL> dataset = IterableDataset ( dict ( features = features , targets = targets ) ) <EOL> x = tensor . vector ( '<STR_LIT>' ) <EOL> y = tensor . scalar ( '<STR_LIT>' ) <EOL> W = shared_floatx ( [ <NUM_LIT:0> , <NUM_LIT:0> ] , name = '<STR_LIT>' ) <EOL> V = shared_floatx ( <NUM_LIT:7> , name = '<STR_LIT>' ) <EOL> W_sum = W . sum ( ) . copy ( name = '<STR_LIT>' ) <EOL> cost = ( ( x * W ) . sum ( ) - y ) ** <NUM_LIT:2> <EOL> cost . name = '<STR_LIT>' <EOL> class TrueCostExtension ( TrainingExtension ) : <EOL> def before_batch ( self , data ) : <EOL> self . main_loop . log . current_row [ '<STR_LIT>' ] = ( <EOL> ( ( W . get_value ( ) * data [ "<STR_LIT>" ] ) . sum ( ) - <EOL> data [ "<STR_LIT>" ] ) ** <NUM_LIT:2> ) <EOL> ftt1 = MeanFeaturesTimesTarget ( <EOL> requires = [ x , y ] , name = '<STR_LIT>' ) <EOL> ftt2 = MeanFeaturesTimesTarget ( <EOL> requires = [ x , y ] , name = '<STR_LIT>' ) <EOL> main_loop = MainLoop ( <EOL> model = None , data_stream = dataset . get_example_stream ( ) , <EOL> algorithm = GradientDescent ( cost = cost , parameters = [ W ] , <EOL> step_rule = Scale ( <NUM_LIT> ) ) , <EOL> extensions = [ <EOL> FinishAfter ( after_n_epochs = <NUM_LIT:1> ) , <EOL> TrainingDataMonitoring ( [ W_sum , cost , V , ftt1 ] , prefix = "<STR_LIT>" , <EOL> after_batch = True ) , <EOL> TrainingDataMonitoring ( [ aggregation . mean ( W_sum ) , cost , ftt2 ] , <EOL> prefix = "<STR_LIT>" , after_epoch = True ) , <EOL> TrueCostExtension ( ) ] ) <EOL> main_loop . run ( ) <EOL> assert_allclose ( main_loop . log . current_row [ '<STR_LIT>' ] , <NUM_LIT> ) <EOL> for i in range ( n_batches ) : <EOL> assert_allclose ( main_loop . log [ i ] [ '<STR_LIT>' ] , <EOL> main_loop . log [ i + <NUM_LIT:1> ] [ '<STR_LIT>' ] ) <EOL> assert_allclose ( <EOL> main_loop . log [ n_batches ] [ '<STR_LIT>' ] , <EOL> sum ( [ main_loop . log [ i ] [ '<STR_LIT>' ] <EOL> for i in range ( n_batches ) ] ) / n_batches ) <EOL> assert_allclose ( <EOL> main_loop . log [ n_batches ] [ '<STR_LIT>' ] , <EOL> sum ( [ main_loop . log [ i ] [ '<STR_LIT>' ] <EOL> for i in range ( <NUM_LIT:1> , n_batches + <NUM_LIT:1> ) ] ) / n_batches ) <EOL> for i in range ( n_batches ) : <EOL> assert_allclose ( main_loop . log [ i + <NUM_LIT:1> ] [ '<STR_LIT>' ] , <EOL> features [ i ] * targets [ i ] ) <EOL> assert_allclose ( main_loop . log [ n_batches ] [ '<STR_LIT>' ] , <EOL> ( features * targets [ : , None ] ) . mean ( axis = <NUM_LIT:0> ) ) </s>
