<s> import quagga <EOL> import numpy as np <EOL> from itertools import izip <EOL> from unittest import TestCase <EOL> from quagga . matrix import Matrix <EOL> from quagga . connector import Connector <EOL> from quagga . blocks import SoftmaxCeBlock <EOL> from quagga . blocks import VerticalStackBlock <EOL> class TestVerticalStackBlock ( TestCase ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> cls . rng = np . random . RandomState ( seed = <NUM_LIT> ) <EOL> cls . N = <NUM_LIT:15> <EOL> def test_fprop ( self ) : <EOL> r = [ ] <EOL> for i in xrange ( self . N ) : <EOL> matrices = [ ] <EOL> ncols = self . rng . random_integers ( <NUM_LIT:1> , <NUM_LIT> ) <EOL> for _ in xrange ( self . rng . random_integers ( <NUM_LIT:1> , <NUM_LIT:10> ) ) : <EOL> nrows = self . rng . random_integers ( <NUM_LIT:1> , <NUM_LIT> ) <EOL> matrices . append ( self . rng . rand ( nrows , ncols ) . astype ( np . float32 ) ) <EOL> output = { } <EOL> for processor_type in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> quagga . processor_type = processor_type <EOL> qmatrices = [ Connector ( Matrix . from_npa ( m ) ) for m in matrices ] <EOL> for m in qmatrices : <EOL> m . fprop ( ) <EOL> vstack_block = VerticalStackBlock ( * qmatrices ) <EOL> vstack_block . fprop ( ) <EOL> output [ processor_type ] = vstack_block . output . to_host ( ) <EOL> r . append ( np . allclose ( output [ '<STR_LIT>' ] , output [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( sum ( r ) , self . N ) <EOL> def test_numpy_fprop ( self ) : <EOL> r = [ ] <EOL> quagga . processor_type = '<STR_LIT>' <EOL> for _ in xrange ( self . N ) : <EOL> matrices = [ ] <EOL> ncols = self . rng . random_integers ( <NUM_LIT:1> , <NUM_LIT> ) <EOL> for _ in xrange ( self . rng . random_integers ( <NUM_LIT:1> , <NUM_LIT:10> ) ) : <EOL> nrows = self . rng . random_integers ( <NUM_LIT:1> , <NUM_LIT> ) <EOL> matrices . append ( self . rng . rand ( nrows , ncols ) . astype ( np . float32 ) ) <EOL> numpy_output = np . vstack ( [ m for m in matrices ] ) <EOL> matrices = [ Connector ( Matrix . from_npa ( m ) ) for m in matrices ] <EOL> vstack_block = VerticalStackBlock ( * matrices ) <EOL> for m in matrices : <EOL> m . fprop ( ) <EOL> vstack_block . fprop ( ) <EOL> quagga_output = vstack_block . output . to_host ( ) <EOL> r . append ( np . allclose ( numpy_output , quagga_output ) ) <EOL> self . assertEqual ( sum ( r ) , self . N ) <EOL> def test_bprop ( self ) : <EOL> r = [ ] <EOL> for i in xrange ( self . N ) : <EOL> matrices = [ ] <EOL> ncols = self . rng . random_integers ( <NUM_LIT:1> , <NUM_LIT> ) <EOL> nrows = [ <NUM_LIT:0> ] <EOL> row_slices = [ ] <EOL> device_ids = [ ] <EOL> for _ in xrange ( self . rng . random_integers ( <NUM_LIT:1> , <NUM_LIT:10> ) ) : <EOL> _nrows = self . rng . random_integers ( <NUM_LIT:1> , <NUM_LIT> ) <EOL> nrows . append ( nrows [ - <NUM_LIT:1> ] + _nrows ) <EOL> if self . rng . choice ( [ True , False ] ) : <EOL> device_ids . append ( <NUM_LIT:0> ) <EOL> row_slices . append ( ( nrows [ - <NUM_LIT:2> ] , nrows [ - <NUM_LIT:1> ] ) ) <EOL> else : <EOL> device_ids . append ( None ) <EOL> matrices . append ( self . rng . rand ( _nrows , ncols ) . astype ( np . float32 ) ) <EOL> true_labels = self . rng . randint ( ncols , size = ( nrows [ - <NUM_LIT:1> ] , <NUM_LIT:1> ) ) . astype ( np . int32 ) <EOL> if not row_slices : <EOL> r . append ( True ) <EOL> continue <EOL> output = { } <EOL> for processor_type in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> quagga . processor_type = processor_type <EOL> qmatrices = [ Connector ( Matrix . from_npa ( m ) , d_id ) for m , d_id in izip ( matrices , device_ids ) ] <EOL> qtrue_labels = Connector ( Matrix . from_npa ( true_labels ) ) <EOL> vstack_block = VerticalStackBlock ( * qmatrices ) <EOL> sce_block = SoftmaxCeBlock ( vstack_block . output , qtrue_labels ) <EOL> for m in qmatrices : <EOL> m . fprop ( ) <EOL> qtrue_labels . fprop ( ) <EOL> vstack_block . fprop ( ) <EOL> sce_block . fprop ( ) <EOL> sce_block . bprop ( ) <EOL> vstack_block . bprop ( ) <EOL> output [ processor_type ] = [ m . backward_matrix . to_host ( ) <EOL> for m in qmatrices if m . bpropagable ] <EOL> for dL_dm_gpu , dL_dm_cpu in izip ( output [ '<STR_LIT>' ] , output [ '<STR_LIT>' ] ) : <EOL> if not np . allclose ( dL_dm_gpu , dL_dm_cpu ) : <EOL> r . append ( False ) <EOL> break <EOL> else : <EOL> r . append ( True ) <EOL> self . assertEqual ( sum ( r ) , self . N ) </s>
