<s> import copy <EOL> import six . moves . cPickle as pickle <EOL> import numpy <EOL> import unittest <EOL> from theano import config , gof <EOL> from six import iteritems <EOL> from theano . compile . io import In , Out <EOL> from theano . compile import function <EOL> from theano . compile import UnusedInputError <EOL> from theano . gof import MissingInputError <EOL> from theano . compat import exc_message <EOL> from theano . tests . unittest_tools import SkipTest <EOL> from theano import tensor <EOL> from theano import tensor as T <EOL> import theano <EOL> import numpy as N <EOL> PatternOptimizer = lambda p1 , p2 , ign = True : gof . OpKeyOptimizer ( gof . PatternSub ( p1 , p2 ) , ignore_newtrees = ign ) <EOL> def checkfor ( testcase , fn , E ) : <EOL> try : <EOL> fn ( ) <EOL> except Exception as e : <EOL> if isinstance ( e , E ) : <EOL> return <EOL> else : <EOL> raise <EOL> testcase . fail ( ) <EOL> class T_function ( unittest . TestCase ) : <EOL> def test_none ( self ) : <EOL> fn = function ( [ ] , None ) <EOL> rval = fn ( ) <EOL> if rval == [ ] : <EOL> raise SkipTest ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> else : <EOL> assert rval is None <EOL> def test_empty ( self ) : <EOL> fn = function ( [ ] , [ ] ) <EOL> self . assertTrue ( fn ( ) == [ ] ) <EOL> def test_extra_inputs ( self ) : <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> fn = function ( [ x ] , [ x ] ) <EOL> self . assertRaises ( TypeError , fn , <NUM_LIT:1> , <NUM_LIT:2> ) <EOL> def test_missing_inputs ( self ) : <EOL> def fn ( ) : <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> fn = function ( [ ] , [ x ] ) <EOL> checkfor ( self , fn , MissingInputError ) <EOL> def fn ( ) : <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> fn = function ( [ s ] , [ x ] , on_unused_input = '<STR_LIT:ignore>' ) <EOL> checkfor ( self , fn , MissingInputError ) <EOL> def fn ( ) : <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> fn = function ( [ s ] , [ x ] ) <EOL> checkfor ( self , fn , UnusedInputError ) <EOL> def fn ( ) : <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> fn = function ( [ s ] , x , on_unused_input = '<STR_LIT:ignore>' ) <EOL> checkfor ( self , fn , MissingInputError ) <EOL> def fn ( ) : <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> fn = function ( [ s ] , x ) <EOL> checkfor ( self , fn , UnusedInputError ) <EOL> def fn ( ) : <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> fn = function ( [ s ] , Out ( x ) , on_unused_input = '<STR_LIT:ignore>' ) <EOL> checkfor ( self , fn , MissingInputError ) <EOL> def fn ( ) : <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> fn = function ( [ s ] , Out ( x ) ) <EOL> checkfor ( self , fn , UnusedInputError ) <EOL> def fn ( ) : <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> fn = function ( [ In ( x , update = s + x ) ] , x ) <EOL> checkfor ( self , fn , MissingInputError ) <EOL> def fn ( ) : <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> fn = function ( [ In ( x , update = ( ( s * s ) + x ) ) ] , x ) <EOL> checkfor ( self , fn , MissingInputError ) <EOL> def test_input_anon_singleton ( self ) : <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> fn = function ( [ s , x ] , [ x + s ] ) <EOL> self . assertTrue ( fn ( <NUM_LIT:2> , <NUM_LIT:3> ) == [ <NUM_LIT:5> ] ) <EOL> self . assertTrue ( fn ( <NUM_LIT:2> , <NUM_LIT:3> ) == [ <NUM_LIT:5> ] ) <EOL> def test_input_anon_unpack ( self ) : <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> fn = function ( [ s , x ] , x + s ) <EOL> self . assertTrue ( fn ( <NUM_LIT:2> , <NUM_LIT:3> ) == <NUM_LIT:5> ) <EOL> def test_naming_rule0 ( self ) : <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> f = function ( [ x , s ] , x / s ) <EOL> self . assertTrue ( f ( <NUM_LIT:1> , <NUM_LIT:2> ) == <NUM_LIT:0.5> ) <EOL> self . assertTrue ( f ( <NUM_LIT:2> , <NUM_LIT:1> ) == <NUM_LIT> ) <EOL> self . assertTrue ( f ( s = <NUM_LIT:2> , x = <NUM_LIT:1> ) == <NUM_LIT:0.5> ) <EOL> self . assertTrue ( f ( x = <NUM_LIT:2> , s = <NUM_LIT:1> ) == <NUM_LIT> ) <EOL> self . assertTrue ( f ( <NUM_LIT:2> , s = <NUM_LIT:1> ) == <NUM_LIT> ) <EOL> checkfor ( self , lambda : f ( <NUM_LIT:2> , x = <NUM_LIT> ) , TypeError ) <EOL> checkfor ( self , lambda : f ( x = <NUM_LIT:1> ) , TypeError ) <EOL> checkfor ( self , lambda : f ( s = <NUM_LIT:1> ) , TypeError ) <EOL> def test_naming_rule1 ( self ) : <EOL> a = T . scalar ( ) <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> f = function ( [ a , s ] , a / s ) <EOL> self . assertTrue ( f ( <NUM_LIT:1> , <NUM_LIT:2> ) == <NUM_LIT:0.5> ) <EOL> self . assertTrue ( f ( <NUM_LIT:2> , <NUM_LIT:1> ) == <NUM_LIT> ) <EOL> self . assertTrue ( f ( <NUM_LIT:2> , s = <NUM_LIT:1> ) == <NUM_LIT> ) <EOL> checkfor ( self , lambda : f ( q = <NUM_LIT:2> , s = <NUM_LIT:1> ) , TypeError ) <EOL> checkfor ( self , lambda : f ( a = <NUM_LIT:2> , s = <NUM_LIT:1> ) , TypeError ) <EOL> def test_naming_rule2 ( self ) : <EOL> a = T . scalar ( ) <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> f = function ( [ x , a , s ] , a / s , on_unused_input = '<STR_LIT:ignore>' ) <EOL> self . assertTrue ( f ( <NUM_LIT:9> , <NUM_LIT:1> , <NUM_LIT:2> ) == <NUM_LIT:0.5> ) <EOL> self . assertTrue ( f ( <NUM_LIT:9> , <NUM_LIT:2> , <NUM_LIT:1> ) == <NUM_LIT> ) <EOL> self . assertTrue ( f ( <NUM_LIT:9> , <NUM_LIT:2> , s = <NUM_LIT:1> ) == <NUM_LIT> ) <EOL> checkfor ( self , lambda : f ( x = <NUM_LIT:9> , a = <NUM_LIT:2> , s = <NUM_LIT:1> ) , TypeError ) <EOL> checkfor ( self , lambda : f ( <NUM_LIT> , x = <NUM_LIT:9> ) , TypeError ) <EOL> def test_naming_rule3 ( self ) : <EOL> a = T . scalar ( ) <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> f = function ( [ x , In ( a , value = <NUM_LIT:1.0> ) , s ] , a / s + x ) <EOL> self . assertTrue ( f ( <NUM_LIT:9> , <NUM_LIT:2> , <NUM_LIT:4> ) == <NUM_LIT> ) <EOL> self . assertTrue ( f ( <NUM_LIT:9> , <NUM_LIT:2> , s = <NUM_LIT:4> ) == <NUM_LIT> ) <EOL> self . assertTrue ( f ( <NUM_LIT:9> , s = <NUM_LIT:4> ) == <NUM_LIT> ) <EOL> self . assertTrue ( f ( x = <NUM_LIT:9> , s = <NUM_LIT:4> ) == <NUM_LIT> ) <EOL> checkfor ( self , lambda : f ( x = <NUM_LIT:9> , a = <NUM_LIT:2> , s = <NUM_LIT:4> ) , TypeError ) <EOL> checkfor ( self , lambda : f ( ) , TypeError ) <EOL> checkfor ( self , lambda : f ( x = <NUM_LIT:9> ) , TypeError ) <EOL> def test_naming_rule4 ( self ) : <EOL> a = T . scalar ( ) <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> f = function ( [ x , In ( a , value = <NUM_LIT:1.0> , name = '<STR_LIT:a>' ) , s ] , a / s + x ) <EOL> self . assertTrue ( f ( <NUM_LIT:9> , <NUM_LIT:2> , <NUM_LIT:4> ) == <NUM_LIT> ) <EOL> self . assertTrue ( f ( <NUM_LIT:9> , <NUM_LIT:2> , s = <NUM_LIT:4> ) == <NUM_LIT> ) <EOL> self . assertTrue ( f ( <NUM_LIT:9> , s = <NUM_LIT:4> ) == <NUM_LIT> ) <EOL> self . assertTrue ( f ( <NUM_LIT:9> , a = <NUM_LIT:2> , s = <NUM_LIT:4> ) == <NUM_LIT> ) <EOL> self . assertTrue ( f ( x = <NUM_LIT:9> , a = <NUM_LIT:2> , s = <NUM_LIT:4> ) == <NUM_LIT> ) <EOL> self . assertTrue ( f ( x = <NUM_LIT:9> , s = <NUM_LIT:4> ) == <NUM_LIT> ) <EOL> checkfor ( self , lambda : f ( ) , TypeError ) <EOL> checkfor ( self , lambda : f ( <NUM_LIT> , x = <NUM_LIT:9> ) , TypeError ) <EOL> def test_state_access ( self ) : <EOL> a = T . scalar ( ) <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> f = function ( [ x , In ( a , value = <NUM_LIT:1.0> , name = '<STR_LIT:a>' ) , In ( s , value = <NUM_LIT:0.0> , update = s + a * x ) ] , s + a * x ) <EOL> self . assertTrue ( f [ a ] == <NUM_LIT:1.0> ) <EOL> self . assertTrue ( f [ s ] == <NUM_LIT:0.0> ) <EOL> self . assertTrue ( f ( <NUM_LIT> ) == <NUM_LIT> ) <EOL> self . assertTrue ( f ( <NUM_LIT> , a = <NUM_LIT> ) == <NUM_LIT> ) <EOL> self . assertTrue ( f [ a ] == <NUM_LIT:1.0> ) <EOL> self . assertTrue ( f [ s ] == <NUM_LIT> ) <EOL> f [ a ] = <NUM_LIT> <EOL> self . assertTrue ( f [ a ] == <NUM_LIT> ) <EOL> self . assertTrue ( f ( <NUM_LIT> ) == <NUM_LIT> ) <EOL> self . assertTrue ( f [ s ] == <NUM_LIT> ) <EOL> def test_same_names ( self ) : <EOL> a , x , s = T . scalars ( '<STR_LIT>' ) <EOL> f = function ( [ a , x , s ] , a + x + s ) <EOL> self . assertTrue ( f ( <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ) == <NUM_LIT:6> ) <EOL> checkfor ( self , lambda : f ( <NUM_LIT:1> , <NUM_LIT:2> , x = <NUM_LIT:3> ) , TypeError ) <EOL> def test_weird_names ( self ) : <EOL> a , x , s = T . scalars ( '<STR_LIT>' ) <EOL> checkfor ( self , lambda : function ( [ In ( a , name = [ ] ) ] , [ ] ) , TypeError ) <EOL> def t ( ) : <EOL> f = function ( [ In ( a , name = set ( [ '<STR_LIT>' , ( ) ] ) , value = <NUM_LIT:1.0> ) , <EOL> In ( x , name = ( ) , value = <NUM_LIT> ) , <EOL> In ( s , name = T . scalar ( ) , value = <NUM_LIT> ) ] , a + x + s ) <EOL> checkfor ( self , t , TypeError ) <EOL> def test_copy ( self ) : <EOL> a = T . scalar ( ) <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> f = function ( [ x , In ( a , value = <NUM_LIT:1.0> , name = '<STR_LIT:a>' ) , In ( s , value = <NUM_LIT:0.0> , update = s + a * x , mutable = True ) ] , s + a * x ) <EOL> g = copy . copy ( f ) <EOL> self . assertFalse ( g . container [ x ] . storage is f . container [ x ] . storage ) <EOL> self . assertFalse ( g . container [ a ] . storage is f . container [ a ] . storage ) <EOL> self . assertFalse ( g . container [ s ] . storage is f . container [ s ] . storage ) <EOL> self . assertFalse ( g . value [ a ] is not f . value [ a ] ) <EOL> self . assertFalse ( g . value [ s ] is f . value [ s ] ) <EOL> self . assertFalse ( ( g . value [ s ] != f . value [ s ] ) . any ( ) ) <EOL> self . assertTrue ( f ( <NUM_LIT:2> , <NUM_LIT:1> ) == g ( <NUM_LIT:2> ) ) <EOL> self . assertTrue ( f ( <NUM_LIT:2> , <NUM_LIT:1> ) == g ( <NUM_LIT:2> ) ) <EOL> f ( <NUM_LIT:1> , <NUM_LIT:2> ) <EOL> self . assertFalse ( f ( <NUM_LIT:1> , <NUM_LIT:2> ) == g ( <NUM_LIT:1> , <NUM_LIT:2> ) ) <EOL> def test_copy_share_memory ( self ) : <EOL> x = T . fscalar ( '<STR_LIT:x>' ) <EOL> y = theano . shared ( value = <NUM_LIT:1> ) <EOL> z = theano . shared ( value = <NUM_LIT:2> ) <EOL> out = T . tanh ( ( x + y + <NUM_LIT:2> ) / ( x + z - <NUM_LIT> ) ** <NUM_LIT:2> ) <EOL> for mode in [ "<STR_LIT>" , "<STR_LIT>" ] : <EOL> ori = theano . function ( [ x ] , [ out ] , mode = mode , updates = { z : z + <NUM_LIT:1> } ) <EOL> cpy = ori . copy ( share_memory = True ) <EOL> storage_map_ori = ori . fn . storage_map <EOL> storage_map_cpy = cpy . fn . storage_map <EOL> fgraph_ori = ori . maker . fgraph <EOL> fgraph_cpy = cpy . maker . fgraph <EOL> i_o_variables = fgraph_cpy . inputs + fgraph_cpy . outputs <EOL> ori_storages = storage_map_ori . values ( ) <EOL> for key in storage_map_cpy . keys ( ) : <EOL> storage = storage_map_cpy [ key ] <EOL> if key not in i_o_variables or isinstance ( key , theano . tensor . Constant ) : <EOL> self . assertTrue ( any ( [ storage is s for s in ori_storages ] ) ) <EOL> for ( input , _1 , _2 ) , here , there in zip ( ori . indices , <EOL> ori . input_storage , <EOL> cpy . input_storage ) : <EOL> self . assertTrue ( here . data is there . data ) <EOL> def test_swap_SharedVariable ( self ) : <EOL> i = T . iscalar ( ) <EOL> x_list = theano . shared ( value = numpy . random . rand ( <NUM_LIT:10> ) . astype ( config . floatX ) ) <EOL> x = T . scalar ( '<STR_LIT:x>' ) <EOL> y = theano . shared ( value = <NUM_LIT:1> , name = '<STR_LIT:y>' ) <EOL> z = theano . shared ( value = <NUM_LIT:2> , name = '<STR_LIT:z>' ) <EOL> m = theano . shared ( value = <NUM_LIT:0> , name = '<STR_LIT:m>' ) <EOL> y_rpl = theano . shared ( value = <NUM_LIT:3> , name = '<STR_LIT>' ) <EOL> z_rpl = theano . shared ( value = <NUM_LIT:4> , name = '<STR_LIT>' ) <EOL> swap = { y : y_rpl , z : z_rpl } <EOL> map_SV = { '<STR_LIT>' : y_rpl , '<STR_LIT>' : z_rpl } <EOL> out = x + y + z + m <EOL> second_time = False <EOL> for mode in [ "<STR_LIT>" , "<STR_LIT>" ] : <EOL> ori = theano . function ( [ i ] , [ out ] , mode = mode , <EOL> updates = [ ( z , z + <NUM_LIT:1> ) , ( m , m + <NUM_LIT:2> ) ] , <EOL> givens = { x : x_list [ i ] } ) <EOL> cpy = ori . copy ( swap = swap ) <EOL> ori ( <NUM_LIT:1> ) , cpy ( <NUM_LIT:1> ) , cpy ( <NUM_LIT:2> ) <EOL> if not second_time : <EOL> assert m . get_value ( ) == <NUM_LIT:6> <EOL> assert z . get_value ( ) == <NUM_LIT:3> <EOL> assert z_rpl . get_value ( ) == <NUM_LIT:6> <EOL> assert y_rpl . get_value ( ) == <NUM_LIT:3> <EOL> assert y . get_value ( ) == <NUM_LIT:1> <EOL> elif second_time : <EOL> assert m . get_value ( ) == <NUM_LIT:12> <EOL> assert z . get_value ( ) == <NUM_LIT:4> <EOL> assert z_rpl . get_value ( ) == <NUM_LIT:8> <EOL> assert y_rpl . get_value ( ) == <NUM_LIT:3> <EOL> names = map_SV . keys ( ) <EOL> for key in cpy . fn . storage_map : <EOL> if key . name in names : <EOL> assert map_SV [ key . name ] . container . storage [ <NUM_LIT:0> ] == cpy . fn . storage_map [ key ] [ <NUM_LIT:0> ] <EOL> second_time = True <EOL> def test_swap_SharedVaraile_with_given ( self ) : <EOL> """<STR_LIT>""" <EOL> train_x = theano . shared ( value = numpy . random . rand ( <NUM_LIT:10> , <NUM_LIT:10> ) . astype ( config . floatX ) ) <EOL> test_x = theano . shared ( value = numpy . random . rand ( <NUM_LIT:10> , <NUM_LIT:10> ) . astype ( config . floatX ) ) <EOL> train_y = theano . shared ( value = numpy . random . rand ( <NUM_LIT:10> , <NUM_LIT:1> ) . astype ( config . floatX ) ) <EOL> test_y = theano . shared ( value = numpy . random . rand ( <NUM_LIT:10> , <NUM_LIT:1> ) . astype ( config . floatX ) ) <EOL> i = T . iscalar ( '<STR_LIT:index>' ) <EOL> x = T . vector ( '<STR_LIT:x>' ) <EOL> y = T . vector ( '<STR_LIT:y>' ) <EOL> out = ( T . sum ( x ) - y ) ** <NUM_LIT:2> <EOL> train = theano . function ( [ i ] , out , <EOL> givens = { x : train_x [ i ] , y : train_y [ i ] } , <EOL> updates = { train_x : train_x + <NUM_LIT:0.1> } ) <EOL> test_def = theano . function ( [ i ] , out , givens = { x : test_x [ i ] , y : test_y [ i ] } ) <EOL> test_cpy = train . copy ( swap = { train_x : test_x , train_y : test_y } , <EOL> delete_updates = True ) <EOL> for in1 , in2 in zip ( test_def . maker . inputs , test_def . maker . inputs ) : <EOL> assert in1 . value is in2 . value <EOL> def test_copy_delete_updates ( self ) : <EOL> x = T . fscalar ( '<STR_LIT:x>' ) <EOL> y = theano . shared ( value = <NUM_LIT:1> , name = '<STR_LIT:y>' ) <EOL> z = theano . shared ( value = <NUM_LIT:2> , name = '<STR_LIT:z>' ) <EOL> out = x + y + z <EOL> second_time = False <EOL> for mode in [ "<STR_LIT>" , "<STR_LIT>" ] : <EOL> ori = theano . function ( [ x ] , out , mode = mode , updates = { z : z * <NUM_LIT:2> } ) <EOL> cpy = ori . copy ( delete_updates = True ) <EOL> assert cpy ( <NUM_LIT:1> ) [ <NUM_LIT:0> ] == <NUM_LIT:4> <EOL> assert cpy ( <NUM_LIT:1> ) [ <NUM_LIT:0> ] == <NUM_LIT:4> <EOL> assert cpy ( <NUM_LIT:1> ) [ <NUM_LIT:0> ] == <NUM_LIT:4> <EOL> def test_shared_state0 ( self ) : <EOL> a = T . scalar ( ) <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> f = function ( [ x , In ( a , value = <NUM_LIT:1.0> , name = '<STR_LIT:a>' ) , In ( s , value = <NUM_LIT:0.0> , update = s + a * x , mutable = True ) ] , s + a * x ) <EOL> g = function ( [ x , In ( a , value = <NUM_LIT:1.0> , name = '<STR_LIT:a>' ) , In ( s , value = f . container [ s ] , update = s - a * x , mutable = True ) ] , s + a * x ) <EOL> f ( <NUM_LIT:1> , <NUM_LIT:2> ) <EOL> self . assertTrue ( f [ s ] == <NUM_LIT:2> ) <EOL> self . assertTrue ( g [ s ] == <NUM_LIT:2> ) <EOL> g ( <NUM_LIT:1> , <NUM_LIT:2> ) <EOL> self . assertTrue ( f [ s ] == <NUM_LIT:0> ) <EOL> self . assertTrue ( g [ s ] == <NUM_LIT:0> ) <EOL> def test_shared_state1 ( self ) : <EOL> a = T . scalar ( ) <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> f = function ( [ x , In ( a , value = <NUM_LIT:1.0> , name = '<STR_LIT:a>' ) , In ( s , value = <NUM_LIT:0.0> , update = s + a * x , mutable = True ) ] , s + a * x ) <EOL> g = function ( [ x , In ( a , value = <NUM_LIT:1.0> , name = '<STR_LIT:a>' ) , In ( s , value = f . container [ s ] ) ] , s + a * x ) <EOL> f ( <NUM_LIT:1> , <NUM_LIT:2> ) <EOL> self . assertTrue ( f [ s ] == <NUM_LIT:2> ) <EOL> self . assertTrue ( g [ s ] == <NUM_LIT:2> ) <EOL> f ( <NUM_LIT:1> , <NUM_LIT:2> ) <EOL> g ( <NUM_LIT:1> , <NUM_LIT:2> ) <EOL> self . assertTrue ( f [ s ] == <NUM_LIT:4> ) <EOL> self . assertTrue ( g [ s ] == <NUM_LIT:4> ) <EOL> def test_shared_state2 ( self ) : <EOL> a = T . scalar ( ) <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> f = function ( [ x , In ( a , value = <NUM_LIT:1.0> , name = '<STR_LIT:a>' ) , In ( s , value = <NUM_LIT:0.0> , update = s + a * x , <EOL> mutable = False ) ] , s + a * x ) <EOL> g = function ( [ x , In ( a , value = <NUM_LIT:1.0> , name = '<STR_LIT:a>' ) , In ( s , value = f . container [ s ] ) ] , s + a * x ) <EOL> f ( <NUM_LIT:1> , <NUM_LIT:2> ) <EOL> self . assertTrue ( f [ s ] == <NUM_LIT:2> ) <EOL> self . assertTrue ( g [ s ] == <NUM_LIT:2> ) <EOL> f ( <NUM_LIT:1> , <NUM_LIT:2> ) <EOL> self . assertTrue ( f [ s ] == <NUM_LIT:4> ) <EOL> self . assertTrue ( g [ s ] == <NUM_LIT:4> ) <EOL> g ( <NUM_LIT:1> , <NUM_LIT:2> ) <EOL> self . assertTrue ( f [ s ] == <NUM_LIT:4> ) <EOL> self . assertTrue ( g [ s ] == <NUM_LIT:4> ) <EOL> def test_shared_state_not_implicit ( self ) : <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> inc = function ( [ x , In ( s , update = ( s + x ) , value = <NUM_LIT> ) ] , [ ] ) <EOL> dec = function ( [ x , In ( s , update = ( s - x ) , value = inc . container [ s ] , <EOL> implicit = False ) ] , [ ] ) <EOL> self . assertTrue ( dec [ s ] is inc [ s ] ) <EOL> inc [ s ] = <NUM_LIT:2> <EOL> self . assertTrue ( dec [ s ] == <NUM_LIT:2> ) <EOL> dec ( <NUM_LIT:1> ) <EOL> self . assertTrue ( inc [ s ] == <NUM_LIT:1> ) <EOL> dec ( <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> self . assertTrue ( inc [ s ] == - <NUM_LIT:1> ) <EOL> self . assertTrue ( dec [ s ] == - <NUM_LIT:1> ) <EOL> def test_constant_output ( self ) : <EOL> f = theano . function ( [ ] , theano . tensor . constant ( [ <NUM_LIT:4> ] ) ) <EOL> out = f ( ) <EOL> assert ( out == <NUM_LIT:4> ) . all ( ) <EOL> out [ <NUM_LIT:0> ] = <NUM_LIT:3> <EOL> out2 = f ( ) <EOL> assert out2 is not out <EOL> assert ( out2 == <NUM_LIT:4> ) . all ( ) <EOL> f = theano . function ( [ ] , Out ( theano . tensor . constant ( [ <NUM_LIT:4> ] ) , borrow = True ) ) <EOL> out = f ( ) <EOL> assert ( out == <NUM_LIT:4> ) . all ( ) <EOL> out [ <NUM_LIT:0> ] = <NUM_LIT:3> <EOL> out2 = f ( ) <EOL> if isinstance ( theano . compile . mode . get_default_mode ( ) , <EOL> theano . compile . DebugMode ) : <EOL> assert ( out2 == <NUM_LIT:4> ) . all ( ) <EOL> else : <EOL> assert out2 is out <EOL> assert ( out2 == <NUM_LIT:3> ) . all ( ) <EOL> def test_borrow_input ( self ) : <EOL> """<STR_LIT>""" <EOL> a = T . dmatrix ( ) <EOL> aval = numpy . random . rand ( <NUM_LIT:3> , <NUM_LIT:3> ) <EOL> f = theano . function ( [ In ( a , borrow = False ) ] , Out ( a + <NUM_LIT:1> , borrow = True ) ) <EOL> assert numpy . all ( f ( aval ) == aval + <NUM_LIT:1> ) <EOL> assert not numpy . may_share_memory ( aval , f ( aval ) ) <EOL> f = theano . function ( [ In ( a , borrow = False ) ] , Out ( a [ <NUM_LIT:0> , : ] , borrow = True ) ) <EOL> assert numpy . all ( f ( aval ) == aval [ <NUM_LIT:0> , : ] ) <EOL> assert not numpy . may_share_memory ( aval , f ( aval ) ) <EOL> def test_borrow_output ( self ) : <EOL> a = T . dmatrix ( ) <EOL> f = function ( [ a ] , Out ( a , borrow = False ) ) <EOL> o = N . ones ( ( <NUM_LIT:3> , <NUM_LIT:3> ) ) <EOL> assert o is not f ( o ) <EOL> f = function ( [ a ] , Out ( a * <NUM_LIT:4> , borrow = False ) ) <EOL> o = N . ones ( ( <NUM_LIT:3> , <NUM_LIT:3> ) ) <EOL> four = f ( o ) <EOL> assert numpy . all ( four == <NUM_LIT:4> ) <EOL> f ( o + <NUM_LIT> ) <EOL> assert numpy . all ( four == <NUM_LIT:4> ) <EOL> f = function ( [ a ] , Out ( a * <NUM_LIT:4> , borrow = True ) , mode = theano . Mode ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> o = N . ones ( ( <NUM_LIT:3> , <NUM_LIT:3> ) ) <EOL> four = f ( o ) <EOL> assert numpy . all ( four == <NUM_LIT:4> ) <EOL> f ( o + <NUM_LIT> ) <EOL> if theano . config . cxx : <EOL> assert not numpy . all ( four == <NUM_LIT:4> ) <EOL> else : <EOL> assert numpy . all ( four == <NUM_LIT:4> ) <EOL> def test_disconnected_input ( self ) : <EOL> a = T . scalar ( '<STR_LIT:a>' ) <EOL> v = T . vector ( '<STR_LIT:v>' ) <EOL> self . assertRaises ( UnusedInputError , function , [ a , v ] , v * <NUM_LIT:2> ) <EOL> f = function ( [ a , v ] , v * <NUM_LIT:2> , on_unused_input = '<STR_LIT:ignore>' ) <EOL> def test_masked_input ( self ) : <EOL> m = T . matrix ( '<STR_LIT:m>' ) <EOL> mt = m . T <EOL> mt . name = '<STR_LIT>' <EOL> self . assertRaises ( UnusedInputError , function , [ m , mt ] , mt * <NUM_LIT:2> ) <EOL> f = function ( [ m , mt ] , mt * <NUM_LIT:2> , on_unused_input = '<STR_LIT:ignore>' ) <EOL> def test_givens_input_var ( self ) : <EOL> """<STR_LIT>""" <EOL> x = T . scalar ( '<STR_LIT:x>' ) <EOL> y = x * <NUM_LIT:2> <EOL> self . assertRaises ( RuntimeError , function , [ x ] , y , givens = { x : x + <NUM_LIT:1> } ) <EOL> def test_free ( self ) : <EOL> """<STR_LIT>""" <EOL> x = T . vector ( '<STR_LIT:x>' ) <EOL> func = function ( [ x ] , x + <NUM_LIT:1> ) <EOL> func . fn . allow_gc = False <EOL> func ( [ <NUM_LIT:1> ] ) <EOL> check_list = [ ] <EOL> for key , val in iteritems ( func . fn . storage_map ) : <EOL> if not isinstance ( key , theano . gof . Constant ) : <EOL> check_list . append ( val ) <EOL> assert any ( [ val [ <NUM_LIT:0> ] for val in check_list ] ) <EOL> func . free ( ) <EOL> for key , val in iteritems ( func . fn . storage_map ) : <EOL> if not isinstance ( key , theano . gof . Constant ) : <EOL> assert ( val [ <NUM_LIT:0> ] is None ) <EOL> class T_picklefunction ( unittest . TestCase ) : <EOL> def test_deepcopy ( self ) : <EOL> a = T . scalar ( ) <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> f = function ( [ x , In ( a , value = <NUM_LIT:1.0> , name = '<STR_LIT:a>' ) , In ( s , value = <NUM_LIT:0.0> , update = s + a * x , mutable = True ) ] , s + a * x ) <EOL> try : <EOL> g = copy . deepcopy ( f ) <EOL> except NotImplementedError as e : <EOL> if e [ <NUM_LIT:0> ] . startswith ( '<STR_LIT>' ) : <EOL> return <EOL> else : <EOL> raise <EOL> self . assertFalse ( g . container [ <NUM_LIT:0> ] . storage is f . container [ <NUM_LIT:0> ] . storage ) <EOL> self . assertFalse ( g . container [ <NUM_LIT:1> ] . storage is f . container [ <NUM_LIT:1> ] . storage ) <EOL> self . assertFalse ( g . container [ <NUM_LIT:2> ] . storage is f . container [ <NUM_LIT:2> ] . storage ) <EOL> self . assertFalse ( x in g . container ) <EOL> self . assertFalse ( x in g . value ) <EOL> self . assertTrue ( len ( f . defaults ) == len ( g . defaults ) ) <EOL> self . assertTrue ( all ( [ f_req == g_req and f_feed == g_feed and <EOL> f_val == g_val <EOL> for ( ( f_req , f_feed , f_val ) , ( g_req , g_feed , g_val ) ) in zip ( <EOL> f . defaults , g . defaults ) ] ) ) <EOL> self . assertFalse ( g . value [ <NUM_LIT:1> ] is f . value [ <NUM_LIT:1> ] ) <EOL> self . assertFalse ( g . value [ <NUM_LIT:2> ] is f . value [ <NUM_LIT:2> ] ) <EOL> self . assertFalse ( ( g . value [ <NUM_LIT:2> ] != f . value [ <NUM_LIT:2> ] ) . any ( ) ) <EOL> self . assertTrue ( f ( <NUM_LIT:2> , <NUM_LIT:1> ) == g ( <NUM_LIT:2> ) ) <EOL> self . assertTrue ( f ( <NUM_LIT:2> , <NUM_LIT:1> ) == g ( <NUM_LIT:2> ) ) <EOL> f ( <NUM_LIT:1> , <NUM_LIT:2> ) <EOL> self . assertFalse ( f ( <NUM_LIT:1> , <NUM_LIT:2> ) == g ( <NUM_LIT:1> , <NUM_LIT:2> ) ) <EOL> g ( <NUM_LIT:1> , <NUM_LIT:2> ) <EOL> self . assertTrue ( f ( <NUM_LIT:3> ) == g ( <NUM_LIT:3> ) ) <EOL> def test_deepcopy_shared_container ( self ) : <EOL> a , x = T . scalars ( '<STR_LIT>' ) <EOL> h = function ( [ In ( a , value = <NUM_LIT:0.0> ) ] , a ) <EOL> f = function ( [ x , In ( a , value = h . container [ a ] , implicit = True ) ] , x + a ) <EOL> try : <EOL> memo = { } <EOL> ac = copy . deepcopy ( a ) <EOL> memo . update ( { id ( a ) : ac } ) <EOL> hc = copy . deepcopy ( h , memo = memo ) <EOL> memo . update ( { id ( h ) : hc } ) <EOL> fc = copy . deepcopy ( f , memo = memo ) <EOL> except NotImplementedError as e : <EOL> if e [ <NUM_LIT:0> ] . startswith ( '<STR_LIT>' ) : <EOL> return <EOL> else : <EOL> raise <EOL> h [ a ] = <NUM_LIT:1> <EOL> hc [ ac ] = <NUM_LIT:2> <EOL> self . assertTrue ( f [ a ] == <NUM_LIT:1> ) <EOL> self . assertTrue ( fc [ ac ] == <NUM_LIT:2> ) <EOL> def test_pickle ( self ) : <EOL> a = T . scalar ( ) <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> f = function ( [ x , In ( a , value = <NUM_LIT:1.0> , name = '<STR_LIT:a>' ) , In ( s , value = <NUM_LIT:0.0> , update = s + a * x , mutable = True ) ] , s + a * x ) <EOL> try : <EOL> g = pickle . loads ( pickle . dumps ( f , protocol = <NUM_LIT:0> ) ) <EOL> g = pickle . loads ( pickle . dumps ( f , protocol = - <NUM_LIT:1> ) ) <EOL> except NotImplementedError as e : <EOL> if e [ <NUM_LIT:0> ] . startswith ( '<STR_LIT>' ) : <EOL> return <EOL> else : <EOL> raise <EOL> self . assertFalse ( g . container [ <NUM_LIT:0> ] . storage is f . container [ <NUM_LIT:0> ] . storage ) <EOL> self . assertFalse ( g . container [ <NUM_LIT:1> ] . storage is f . container [ <NUM_LIT:1> ] . storage ) <EOL> self . assertFalse ( g . container [ <NUM_LIT:2> ] . storage is f . container [ <NUM_LIT:2> ] . storage ) <EOL> self . assertFalse ( x in g . container ) <EOL> self . assertFalse ( x in g . value ) <EOL> self . assertFalse ( g . value [ <NUM_LIT:1> ] is f . value [ <NUM_LIT:1> ] ) <EOL> self . assertFalse ( g . value [ <NUM_LIT:2> ] is f . value [ <NUM_LIT:2> ] ) <EOL> self . assertFalse ( ( g . value [ <NUM_LIT:2> ] != f . value [ <NUM_LIT:2> ] ) . any ( ) ) <EOL> self . assertTrue ( f ( <NUM_LIT:2> , <NUM_LIT:1> ) == g ( <NUM_LIT:2> ) ) <EOL> self . assertTrue ( f ( <NUM_LIT:2> , <NUM_LIT:1> ) == g ( <NUM_LIT:2> ) ) <EOL> f ( <NUM_LIT:1> , <NUM_LIT:2> ) <EOL> self . assertFalse ( f ( <NUM_LIT:1> , <NUM_LIT:2> ) == g ( <NUM_LIT:1> , <NUM_LIT:2> ) ) <EOL> def test_optimizations_preserved ( self ) : <EOL> a = T . dvector ( ) <EOL> x = T . dvector ( '<STR_LIT:x>' ) <EOL> s = T . dvector ( '<STR_LIT:s>' ) <EOL> xm = T . dmatrix ( '<STR_LIT:x>' ) <EOL> sm = T . dmatrix ( '<STR_LIT:s>' ) <EOL> f = function ( [ a , x , s , xm , sm ] , ( ( a . T . T ) * ( tensor . dot ( xm , ( sm . T . T . T ) ) + x ) . T * ( x / x ) + s ) ) <EOL> old_default_mode = config . mode <EOL> old_default_opt = config . optimizer <EOL> old_default_link = config . linker <EOL> try : <EOL> try : <EOL> str_f = pickle . dumps ( f , protocol = - <NUM_LIT:1> ) <EOL> config . mode = '<STR_LIT>' <EOL> config . linker = '<STR_LIT>' <EOL> config . optimizer = '<STR_LIT:None>' <EOL> g = pickle . loads ( str_f ) <EOL> except NotImplementedError as e : <EOL> if e [ <NUM_LIT:0> ] . startswith ( '<STR_LIT>' ) : <EOL> g = '<STR_LIT>' <EOL> finally : <EOL> config . mode = old_default_mode <EOL> config . optimizer = old_default_opt <EOL> config . linker = old_default_link <EOL> if g == '<STR_LIT>' : <EOL> return <EOL> assert f . maker is not g . maker <EOL> assert f . maker . fgraph is not g . maker . fgraph <EOL> tf = f . maker . fgraph . toposort ( ) <EOL> tg = f . maker . fgraph . toposort ( ) <EOL> assert len ( tf ) == len ( tg ) <EOL> for nf , ng in zip ( tf , tg ) : <EOL> assert nf . op == ng . op <EOL> assert len ( nf . inputs ) == len ( ng . inputs ) <EOL> assert len ( nf . outputs ) == len ( ng . outputs ) <EOL> assert [ i . type for i in nf . inputs ] == [ i . type for i in ng . inputs ] <EOL> assert [ i . type for i in nf . outputs ] == [ i . type for i in ng . outputs ] <EOL> def test_multiple_functions ( self ) : <EOL> a = T . scalar ( ) <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> v = T . vector ( '<STR_LIT:v>' ) <EOL> list_of_things = [ s , x , v ] <EOL> list_of_things . append ( a * x + s ) <EOL> f1 = function ( [ x , In ( a , value = <NUM_LIT:1.0> , name = '<STR_LIT:a>' ) , In ( s , value = <NUM_LIT:0.0> , update = s + a * x , mutable = True ) ] , s + a * x ) <EOL> list_of_things . append ( f1 ) <EOL> f2 = function ( [ x , In ( a , value = <NUM_LIT:1.0> , name = '<STR_LIT:a>' ) , In ( s , value = f1 . container [ s ] , update = s + a * x , mutable = True ) ] , s + a * x ) <EOL> list_of_things . append ( f2 ) <EOL> assert isinstance ( f2 . container [ s ] . storage , list ) <EOL> assert f2 . container [ s ] . storage is f1 . container [ s ] . storage <EOL> v_value = numpy . asarray ( [ <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT> ] , dtype = config . floatX ) <EOL> f3 = function ( [ x , In ( v , value = v_value ) ] , x + v ) <EOL> list_of_things . append ( f3 ) <EOL> try : <EOL> saved_format = pickle . dumps ( list_of_things , protocol = - <NUM_LIT:1> ) <EOL> new_list_of_things = pickle . loads ( saved_format ) <EOL> except NotImplementedError as e : <EOL> if e [ <NUM_LIT:0> ] . startswith ( '<STR_LIT>' ) : <EOL> return <EOL> else : <EOL> raise <EOL> ol = list_of_things <EOL> nl = new_list_of_things <EOL> for i in range ( <NUM_LIT:4> ) : <EOL> assert nl [ i ] != ol [ i ] <EOL> assert nl [ i ] . type == ol [ i ] . type <EOL> assert nl [ i ] . type is not ol [ i ] . type <EOL> assert ol [ <NUM_LIT:3> ] . owner . inputs [ <NUM_LIT:1> ] is s <EOL> assert nl [ <NUM_LIT:3> ] . owner . inputs [ <NUM_LIT:1> ] is not s <EOL> assert nl [ <NUM_LIT:3> ] . owner . inputs [ <NUM_LIT:1> ] . type == s . type <EOL> for i in range ( <NUM_LIT:4> , <NUM_LIT:7> ) : <EOL> assert nl [ i ] != ol [ i ] <EOL> assert nl [ <NUM_LIT:4> ] [ nl [ <NUM_LIT:0> ] ] is not ol [ <NUM_LIT:4> ] [ ol [ <NUM_LIT:0> ] ] <EOL> assert nl [ <NUM_LIT:4> ] [ nl [ <NUM_LIT:0> ] ] == ol [ <NUM_LIT:4> ] [ ol [ <NUM_LIT:0> ] ] <EOL> assert nl [ <NUM_LIT:4> ] ( <NUM_LIT:3> ) == ol [ <NUM_LIT:4> ] ( <NUM_LIT:3> ) <EOL> assert ol [ <NUM_LIT:4> ] . container [ ol [ <NUM_LIT:0> ] ] . storage is ol [ <NUM_LIT:5> ] . container [ ol [ <NUM_LIT:0> ] ] . storage <EOL> assert nl [ <NUM_LIT:4> ] . container [ nl [ <NUM_LIT:0> ] ] . storage is nl [ <NUM_LIT:5> ] . container [ nl [ <NUM_LIT:0> ] ] . storage <EOL> assert nl [ <NUM_LIT:5> ] ( <NUM_LIT:3> ) == ol [ <NUM_LIT:5> ] ( <NUM_LIT:3> ) <EOL> assert nl [ <NUM_LIT:4> ] . value [ nl [ <NUM_LIT:0> ] ] == <NUM_LIT:6> <EOL> assert numpy . all ( nl [ <NUM_LIT:6> ] [ nl [ <NUM_LIT:2> ] ] == numpy . asarray ( [ <NUM_LIT:2> , <NUM_LIT> , <NUM_LIT:4> ] ) ) <EOL> def test_broken_pickle_with_shared ( self ) : <EOL> saves = [ ] <EOL> def pers_save ( obj ) : <EOL> if isinstance ( obj , numpy . ndarray ) : <EOL> saves . append ( obj ) <EOL> return len ( saves ) - <NUM_LIT:1> <EOL> else : <EOL> return None <EOL> def pers_load ( id ) : <EOL> return saves [ id ] <EOL> a = numpy . random . rand ( <NUM_LIT:4> , <NUM_LIT:5> ) <EOL> b = numpy . random . rand ( <NUM_LIT:5> , <NUM_LIT:4> ) <EOL> x = theano . tensor . matrix ( ) <EOL> y = theano . shared ( b ) <EOL> f = theano . function ( [ x ] , theano . tensor . dot ( x , y ) ) <EOL> from theano . compat import BytesIO <EOL> fp = BytesIO ( ) <EOL> p = pickle . Pickler ( fp , <NUM_LIT:2> ) <EOL> p . persistent_id = pers_save <EOL> try : <EOL> p . dump ( f ) <EOL> except NotImplementedError as e : <EOL> if exc_message ( e ) . startswith ( '<STR_LIT>' ) : <EOL> return <EOL> else : <EOL> raise <EOL> fp2 = BytesIO ( fp . getvalue ( ) ) <EOL> fp . close ( ) <EOL> p = pickle . Unpickler ( fp2 ) <EOL> p . persistent_load = pers_load <EOL> f2 = p . load ( ) <EOL> fp2 . close ( ) <EOL> def test_pickle_class_with_functions ( self ) : <EOL> blah = SomethingToPickle ( ) <EOL> assert blah . f2 . container [ blah . s ] . storage is blah . f1 . container [ blah . s ] . storage <EOL> try : <EOL> blah2 = copy . deepcopy ( blah ) <EOL> except NotImplementedError as e : <EOL> if e [ <NUM_LIT:0> ] . startswith ( '<STR_LIT>' ) : <EOL> return <EOL> else : <EOL> raise <EOL> assert blah2 . f2 . container [ blah2 . s ] . storage is blah2 . f1 . container [ blah2 . s ] . storage <EOL> assert blah . f1 [ blah . s ] == blah2 . f1 [ blah2 . s ] <EOL> blah . f2 ( <NUM_LIT:5> ) <EOL> assert blah . f1 [ blah . s ] != blah2 . f1 [ blah2 . s ] <EOL> class SomethingToPickle ( object ) : <EOL> def __init__ ( self ) : <EOL> a = T . scalar ( ) <EOL> x , s = T . scalars ( '<STR_LIT>' ) <EOL> v = T . vector ( '<STR_LIT:v>' ) <EOL> self . s = s <EOL> self . x = x <EOL> self . v = v <EOL> self . e = a * x + s <EOL> self . f1 = function ( [ x , In ( a , value = <NUM_LIT:1.0> , name = '<STR_LIT:a>' ) , In ( s , value = <NUM_LIT:0.0> , update = s + a * x , mutable = True ) ] , s + a * x ) <EOL> self . f2 = function ( [ x , In ( a , value = <NUM_LIT:1.0> , name = '<STR_LIT:a>' ) , In ( s , value = self . f1 . container [ s ] , update = s + a * x , mutable = True ) ] , s + a * x ) <EOL> def test_empty_givens_updates ( ) : <EOL> """<STR_LIT>""" <EOL> x = T . scalar ( ) <EOL> y = x * <NUM_LIT:2> <EOL> function ( [ theano . In ( x ) ] , y , givens = { } ) <EOL> function ( [ theano . In ( x ) ] , y , updates = { } ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> if <NUM_LIT:1> : <EOL> unittest . main ( ) <EOL> elif <NUM_LIT:0> : <EOL> testcases = [ ] <EOL> testcases . append ( T_function ) <EOL> testloader = unittest . TestLoader ( ) <EOL> suite = unittest . TestSuite ( ) <EOL> for testcase in testcases : <EOL> suite . addTest ( testloader . loadTestsFromTestCase ( testcase ) ) <EOL> unittest . TextTestRunner ( verbosity = <NUM_LIT:2> ) . run ( suite ) <EOL> elif <NUM_LIT:0> : <EOL> theano . config . mode = '<STR_LIT>' <EOL> t = T_picklefunction ( ) <EOL> def fu ( b ) : <EOL> assert b <EOL> t . assertTrue = fu <EOL> t . test_deepcopy_shared_container ( ) </s>
