#NEWLINE# def main( gen , X , y , X_og , BATCH_SIZE = 32 , axis = 0 , show_aug_test = True ) : #NEWLINE# #INDENT# "Runs a test for each core functionality of `gen`.\n\n    Parameters\n    ----------\n    gen: datumio.BaseGenerator\n        Generator to test core functionality\n\n    X: iterable\n        X input to generator. For BatchGenerator, X = loaded data with shape\n        of (dataset_length, height, width, channel). For DataGenerator,\n        X = path to data files with shape of (dataset_length,).\n\n    y: iterable\n        y input to generator.\n\n    X_og: ndarray\n        Loaded data with shape of (dataset_length, height, width, channel)\n        corresponding to `X`'s order. This is use for testing generators.\n\n    BATCH_SIZE: int, optional\n        Size of minibatches per call\n\n    axis: None or int, optional\n        Axis to test zmuv operations on.\n        See numpy.mean for more options for axis\n\n    show_aug_test: bool, optional\n        If true, will plot the on-the-fly augmentations (static & random)\n        with comparison to the original. This is for manual, visual\n        test to see if augmentations are correct (or at least what we expect)\n\n    Returns\n    ----------\n    None:\n        If no exceptions were raised, all tests were ran successfully. However,\n        you must check manually check the augmentation tests visually. See\n        `show_aug_test`.\n\n    Raises\n    ----------\n    TestGenError: Exception\n        If any functionality fails to do what it's suppose to.\n    " #NEWLINE# Batcher = gen ( X , y = y ) #NEWLINE# batchgen = Batcher . get_batch ( batch_size = BATCH_SIZE , shuffle = False , chw_order = False , dtype = np . uint8 ) #NEWLINE# for ( idx , ( mb_x , mb_y ) ) in enumerate ( batchgen ) : #NEWLINE# #INDENT# if ( not np . all ( ( mb_x == X_og [ ( idx * BATCH_SIZE ) : ( ( idx + 1 ) * BATCH_SIZE ) ] ) ) ) : #NEWLINE# #INDENT# raise TestGenError ( 'Not all batch in X match correctly from data' ) #NEWLINE# #UNINDENT# if ( not np . all ( ( mb_y == y [ ( idx * BATCH_SIZE ) : ( ( idx + 1 ) * BATCH_SIZE ) ] ) ) ) : #NEWLINE# #INDENT# raise TestGenError ( 'Not all batch in y match correctly from data' ) #NEWLINE# #UNINDENT# #UNINDENT# idxs = range ( len ( X ) ) #NEWLINE# np . random . RandomState ( 16 ) . shuffle ( idxs ) #NEWLINE# Batcher = gen ( X , y = y ) #NEWLINE# batchgen = Batcher . get_batch ( batch_size = BATCH_SIZE , shuffle = True , rng_seed = 16 , chw_order = False , dtype = np . uint8 ) #NEWLINE# for ( idx , ( mb_x , mb_y ) ) in enumerate ( batchgen ) : #NEWLINE# #INDENT# if ( not np . all ( ( mb_x == X_og [ idxs ] [ ( idx * BATCH_SIZE ) : ( ( idx + 1 ) * BATCH_SIZE ) ] ) ) ) : #NEWLINE# #INDENT# raise TestGenError ( 'Batches in X are not correctly shuffled' ) #NEWLINE# #UNINDENT# if ( not np . all ( ( mb_y == y [ idxs ] [ ( idx * BATCH_SIZE ) : ( ( idx + 1 ) * BATCH_SIZE ) ] ) ) ) : #NEWLINE# #INDENT# raise TestGenError ( 'Batches in y are not correctly shuffled' ) #NEWLINE# #UNINDENT# #UNINDENT# Batcher = gen ( X ) #NEWLINE# mb_x = batchgen = Batcher . get_batch ( batch_size = BATCH_SIZE , shuffle = False , chw_order = False , dtype = np . uint8 ) . next ( ) #NEWLINE# if ( type ( mb_x ) == tuple ) : #NEWLINE# #INDENT# raise TestGenError ( "Getting batch is returning y when it shouldn't" ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if ( mb_x . shape [ 0 ] != BATCH_SIZE ) : #NEWLINE# #INDENT# raise TestGenError ( 'W/out y, but returning incorrect len of batch' ) #NEWLINE# #UNINDENT# #UNINDENT# class_weights = { #NEWLINE# #INDENT# 0 : 0.25 , #NEWLINE# 1 : 0.0 , #NEWLINE# 2 : 0.0 , #NEWLINE# 3 : 0.15 , #NEWLINE# 4 : 0.1 , #NEWLINE# 5 : 0.25 , #NEWLINE# 6 : 0.0 , #NEWLINE# 7 : 0.0 , #NEWLINE# 8 : 0.15 , #NEWLINE# 9 : 0.1 , #NEWLINE# #UNINDENT# } #NEWLINE# Batcher = gen ( X , y ) #NEWLINE# Batcher . resample_dataset ( y , class_weights , sample_fraction = 1.0 ) #NEWLINE# resampled_y = Batcher . y #NEWLINE# if np . all ( np . equal ( y , resampled_y ) ) : #NEWLINE# #INDENT# raise TestGenError ( 'Resampled y should not equal to y' ) #NEWLINE# #UNINDENT# hist = np . histogram ( resampled_y , bins = np . arange ( 11 ) ) #NEWLINE# sample_weights = ( hist [ 0 ] / float ( np . sum ( hist [ 0 ] ) ) ) #NEWLINE# mae = np . mean ( np . abs ( ( class_weights . values ( ) - sample_weights ) ) ) #NEWLINE# if ( mae > ( 0.01 * len ( sample_weights ) ) ) : #NEWLINE# #INDENT# raise TestGenError ( 'Resampled y has incorrect distribution' ) #NEWLINE# #UNINDENT# mb_x_og = X_og [ : BATCH_SIZE ] #NEWLINE# mb_y_og = y [ : BATCH_SIZE ] #NEWLINE# tmp_X = np . array ( X_og . copy ( ) , dtype = np . float32 ) #NEWLINE# mean = tmp_X . mean ( axis = axis ) #NEWLINE# tmp_X = ( tmp_X - mean ) #NEWLINE# std = tmp_X . std ( axis = axis ) #NEWLINE# zmuv_mb_x = ( mb_x_og - mean ) #NEWLINE# zmuv_mb_x /= std #NEWLINE# if ( fig . __name__ == 'DataGenerator' ) : #NEWLINE# #INDENT# Batcher = gen ( X , y = y , dataset_zmuv = True , dataset_axis = axis , dataset_zmuv_bsize = 500 ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# Batcher = gen ( X , y = y , dataset_zmuv = True , dataset_axis = axis ) #NEWLINE# #UNINDENT# ( mb_x , mb_y ) = Batcher . get_batch ( batch_size = BATCH_SIZE , shuffle = False , chw_order = False , dtype = np . float32 ) . next ( ) #NEWLINE# if ( not np . all ( ( mb_x == zmuv_mb_x ) ) ) : #NEWLINE# #INDENT# raise TestGenError ( 'Error correctly generating batch w/ dataset-zmuv' ) #NEWLINE# #UNINDENT# mb_x_manual = np . array ( mb_x_og . copy ( ) , dtype = np . float32 ) #NEWLINE# mb_x_manual = ( mb_x_manual - mb_x_manual . mean ( axis = axis ) ) #NEWLINE# mb_x_manual = ( mb_x_manual / ( mb_x_manual . std ( axis = axis ) + 1e-12 ) ) #NEWLINE# Batcher = gen ( X , y = y , batch_zmuv = True , batch_axis = axis ) #NEWLINE# ( mb_x , mb_y ) = Batcher . get_batch ( batch_size = BATCH_SIZE , shuffle = False , chw_order = False , dtype = np . float32 ) . next ( ) #NEWLINE# if ( not np . all ( ( mb_x == mb_x_manual ) ) ) : #NEWLINE# #INDENT# raise TestGenError ( 'Error correctly generating batch w/ batch-zmuv' ) #NEWLINE# #UNINDENT# mb_x_manual = [ ] #NEWLINE# for x in mb_x_og : #NEWLINE# #INDENT# ex = np . array ( x . copy ( ) , dtype = np . float32 ) #NEWLINE# ex = ( ex - ex . mean ( axis = axis ) ) #NEWLINE# ex = ( ex / ( ex . std ( axis = axis ) + 1e-12 ) ) #NEWLINE# mb_x_manual . append ( ex ) #NEWLINE# #UNINDENT# mb_x_manual = np . array ( mb_x_manual , dtype = np . float32 ) #NEWLINE# Batcher = gen ( X , y = y , sample_zmuv = True , sample_axis = axis ) #NEWLINE# ( mb_x , mb_y ) = Batcher . get_batch ( batch_size = BATCH_SIZE , shuffle = False , chw_order = False , dtype = np . float32 ) . next ( ) #NEWLINE# if ( not np . all ( ( mb_x == mb_x_manual ) ) ) : #NEWLINE# #INDENT# raise TestGenError ( 'Error correctly generating batch with sample zmuv' ) #NEWLINE# #UNINDENT# if ( not show_aug_test ) : #NEWLINE# #INDENT# return #NEWLINE# #UNINDENT# aug_params = dict ( rotation = 15 , zoom = ( 1.5 , 1.5 ) , shear = 7 , translation = ( 5 , ( - 5 ) ) , flip_lr = True ) #NEWLINE# Batcher = gen ( X , y , aug_params = aug_params ) #NEWLINE# ( mb_x_aug , mb_y_aug ) = Batcher . get_batch ( batch_size = BATCH_SIZE , shuffle = False , chw_order = False , dtype = np . uint8 ) . next ( ) #NEWLINE# if ( not np . all ( ( mb_y_og == mb_y_aug ) ) ) : #NEWLINE# #INDENT# raise TestGenError ( 'Static aug created error in generation of truth' ) #NEWLINE# #UNINDENT# rng_aug_params = dict ( zoom_range = ( ( 1 / 1.5 ) , 1.5 ) , rotation_range = ( ( - 15 ) , 15 ) , shear_range = ( ( - 7 ) , 7 ) , translation_range = ( ( - 5 ) , 5 ) , do_flip_lr = True , allow_stretch = False ) #NEWLINE# Batcher = gen ( X , y , rng_aug_params = rng_aug_params ) #NEWLINE# ( mb_x_rng , mb_y_rng ) = Batcher . get_batch ( batch_size = BATCH_SIZE , shuffle = False , chw_order = False , dtype = np . uint8 ) . next ( ) #NEWLINE# if ( not np . all ( ( mb_y_og == mb_y_rng ) ) ) : #NEWLINE# #INDENT# raise TestGenError ( 'Rng augs created error in generation of truth' ) #NEWLINE# #UNINDENT# if ( np . all ( ( mb_x_rng == mb_x_aug ) ) or np . all ( ( mb_x_rng == mb_x_og ) ) ) : #NEWLINE# #INDENT# raise TestGenError ( 'Static or Random augs not correctly done' ) #NEWLINE# #UNINDENT# rng_idxs = np . arange ( BATCH_SIZE ) #NEWLINE# rng_idxs = np . random . choice ( rng_idxs , size = 3 , replace = False ) #NEWLINE# ( fig , axes ) = plt . subplots ( 3 , 3 , figsize = ( 14 , 10 ) ) #NEWLINE# fig . suptitle ( ( 'Generator: %s' % gen . __name__ ) , fontsize = 26 ) #NEWLINE# for ( it , ax ) in enumerate ( axes [ : , 0 ] ) : #NEWLINE# #INDENT# img = mb_x_og [ rng_idxs [ it ] ] #NEWLINE# ax . imshow ( img ) #NEWLINE# ax . set_xticks ( [ ] ) #NEWLINE# ax . set_yticks ( [ ] ) #NEWLINE# if ( it == 0 ) : #NEWLINE# #INDENT# ax . set_title ( 'Original Images' ) #NEWLINE# #UNINDENT# #UNINDENT# for ( it , ax ) in enumerate ( axes [ : , 1 ] ) : #NEWLINE# #INDENT# img = mb_x_aug [ rng_idxs [ it ] ] #NEWLINE# ax . imshow ( img ) #NEWLINE# ax . set_xticks ( [ ] ) #NEWLINE# ax . set_yticks ( [ ] ) #NEWLINE# if ( it == 0 ) : #NEWLINE# #INDENT# ax . set_title ( 'Static Augmentations' ) #NEWLINE# #UNINDENT# #UNINDENT# for ( it , ax ) in enumerate ( axes [ : , 2 ] ) : #NEWLINE# #INDENT# img = mb_x_rng [ rng_idxs [ it ] ] #NEWLINE# ax . imshow ( img ) #NEWLINE# ax . set_xticks ( [ ] ) #NEWLINE# ax . set_yticks ( [ ] ) #NEWLINE# if ( it == 0 ) : #NEWLINE# #INDENT# ax . set_title ( 'Random Augmentations' )