<s> from __future__ import absolute_import , print_function , division <EOL> from nose . plugins . skip import SkipTest <EOL> import numpy <EOL> import theano <EOL> from theano . gof . link import PerformLinker <EOL> from theano . gof . cc import CLinker , DualLinker , OpWiseCLinker <EOL> from theano . gof . type import Type <EOL> from theano . gof . graph import Variable , Apply , Constant <EOL> from theano . gof . op import Op <EOL> from theano . gof import fg <EOL> def as_variable ( x ) : <EOL> assert isinstance ( x , Variable ) <EOL> return x <EOL> class TDouble ( Type ) : <EOL> def filter ( self , data , strict = False , allow_downcast = False ) : <EOL> return float ( data ) <EOL> def c_declare ( self , name , sub , check_input = True ) : <EOL> return "<STR_LIT>" % locals ( ) <EOL> def c_init ( self , name , sub ) : <EOL> return """<STR_LIT>""" % locals ( ) <EOL> def c_literal ( self , data ) : <EOL> return str ( data ) <EOL> def c_extract ( self , name , sub , check_input = True ) : <EOL> return """<STR_LIT>""" % dict ( locals ( ) , ** sub ) <EOL> def c_sync ( self , name , sub ) : <EOL> return """<STR_LIT>""" % locals ( ) <EOL> def c_cleanup ( self , name , sub ) : <EOL> return """<STR_LIT>""" % locals ( ) <EOL> def c_code_cache_version ( self ) : <EOL> return ( <NUM_LIT:1> , ) <EOL> def __eq__ ( self , other ) : <EOL> return type ( self ) == type ( other ) <EOL> def __hash__ ( self ) : <EOL> return hash ( type ( self ) ) <EOL> tdouble = TDouble ( ) <EOL> def double ( name ) : <EOL> return Variable ( tdouble , None , None , name = name ) <EOL> class MyOp ( Op ) : <EOL> __props__ = ( "<STR_LIT>" , "<STR_LIT:name>" ) <EOL> def __init__ ( self , nin , name ) : <EOL> self . nin = nin <EOL> self . name = name <EOL> def make_node ( self , * inputs ) : <EOL> assert len ( inputs ) == self . nin <EOL> inputs = list ( map ( as_variable , inputs ) ) <EOL> for input in inputs : <EOL> if input . type is not tdouble : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> outputs = [ double ( self . name + "<STR_LIT>" ) ] <EOL> return Apply ( self , inputs , outputs ) <EOL> def __str__ ( self ) : <EOL> return self . name <EOL> def perform ( self , node , inputs , out_ ) : <EOL> out , = out_ <EOL> out [ <NUM_LIT:0> ] = self . impl ( * inputs ) <EOL> def c_code_cache_version ( self ) : <EOL> return ( <NUM_LIT:1> , ) <EOL> class Binary ( MyOp ) : <EOL> def __init__ ( self ) : <EOL> MyOp . __init__ ( self , <NUM_LIT:2> , self . __class__ . __name__ ) <EOL> class Add ( Binary ) : <EOL> def c_code ( self , node , name , inp , out , sub ) : <EOL> x , y = inp <EOL> z , = out <EOL> return "<STR_LIT>" % locals ( ) <EOL> def impl ( self , x , y ) : <EOL> return x + y <EOL> add = Add ( ) <EOL> class BadSub ( Binary ) : <EOL> def c_code ( self , node , name , inp , out , sub ) : <EOL> x , y = inp <EOL> z , = out <EOL> return "<STR_LIT>" % locals ( ) <EOL> def impl ( self , x , y ) : <EOL> return - <NUM_LIT:10> <EOL> bad_sub = BadSub ( ) <EOL> class Mul ( Binary ) : <EOL> def c_code ( self , node , name , inp , out , sub ) : <EOL> x , y = inp <EOL> z , = out <EOL> return "<STR_LIT>" % locals ( ) <EOL> def impl ( self , x , y ) : <EOL> return x * y <EOL> mul = Mul ( ) <EOL> class Div ( Binary ) : <EOL> def c_code ( self , node , name , inp , out , sub ) : <EOL> x , y = inp <EOL> z , = out <EOL> return "<STR_LIT>" % locals ( ) <EOL> def impl ( self , x , y ) : <EOL> return x / y <EOL> div = Div ( ) <EOL> def inputs ( ) : <EOL> x = double ( '<STR_LIT:x>' ) <EOL> y = double ( '<STR_LIT:y>' ) <EOL> z = double ( '<STR_LIT:z>' ) <EOL> return x , y , z <EOL> def Env ( inputs , outputs ) : <EOL> e = fg . FunctionGraph ( inputs , outputs ) <EOL> return e <EOL> def test_clinker_straightforward ( ) : <EOL> if not theano . config . cxx : <EOL> raise SkipTest ( "<STR_LIT>" ) <EOL> x , y , z = inputs ( ) <EOL> e = add ( mul ( add ( x , y ) , div ( x , y ) ) , bad_sub ( bad_sub ( x , y ) , z ) ) <EOL> lnk = CLinker ( ) . accept ( Env ( [ x , y , z ] , [ e ] ) ) <EOL> fn = lnk . make_function ( ) <EOL> assert fn ( <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) == <NUM_LIT> <EOL> def test_clinker_literal_inlining ( ) : <EOL> if not theano . config . cxx : <EOL> raise SkipTest ( "<STR_LIT>" ) <EOL> x , y , z = inputs ( ) <EOL> z = Constant ( tdouble , <NUM_LIT> ) <EOL> e = add ( mul ( add ( x , y ) , div ( x , y ) ) , bad_sub ( bad_sub ( x , y ) , z ) ) <EOL> lnk = CLinker ( ) . accept ( Env ( [ x , y ] , [ e ] ) ) <EOL> fn = lnk . make_function ( ) <EOL> assert abs ( fn ( <NUM_LIT> , <NUM_LIT> ) + <NUM_LIT> ) < <NUM_LIT> <EOL> code = lnk . code_gen ( ) <EOL> assert "<STR_LIT>" in code <EOL> def test_clinker_literal_cache ( ) : <EOL> if not theano . config . cxx : <EOL> raise SkipTest ( "<STR_LIT>" ) <EOL> mode = theano . Mode ( linker = '<STR_LIT:c>' ) <EOL> A = theano . tensor . matrix ( ) <EOL> input1 = theano . tensor . vector ( ) <EOL> normal_svd = numpy . array ( [ [ <NUM_LIT> , - <NUM_LIT> , - <NUM_LIT> ] , <EOL> [ - <NUM_LIT> , <NUM_LIT> , - <NUM_LIT> ] , <EOL> [ - <NUM_LIT> , - <NUM_LIT> , <NUM_LIT> ] ] , <EOL> dtype = theano . config . floatX ) <EOL> orientationi = numpy . array ( [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] , <EOL> dtype = theano . config . floatX ) <EOL> for out1 in [ A - input1 [ <NUM_LIT:0> ] * numpy . identity ( <NUM_LIT:3> ) , <EOL> input1 [ <NUM_LIT:0> ] * numpy . identity ( <NUM_LIT:3> ) ] : <EOL> benchmark = theano . function ( <EOL> inputs = [ A , input1 ] , <EOL> outputs = [ out1 ] , <EOL> on_unused_input = '<STR_LIT:ignore>' , <EOL> mode = mode ) <EOL> out1 = benchmark ( normal_svd , orientationi ) <EOL> def test_clinker_single_node ( ) : <EOL> if not theano . config . cxx : <EOL> raise SkipTest ( "<STR_LIT>" ) <EOL> x , y , z = inputs ( ) <EOL> node = add . make_node ( x , y ) <EOL> lnk = CLinker ( ) . accept ( Env ( node . inputs , node . outputs ) ) <EOL> fn = lnk . make_function ( ) <EOL> assert fn ( <NUM_LIT> , <NUM_LIT> ) == <NUM_LIT:9> <EOL> def test_clinker_dups ( ) : <EOL> if not theano . config . cxx : <EOL> raise SkipTest ( "<STR_LIT>" ) <EOL> x , y , z = inputs ( ) <EOL> e = add ( x , x ) <EOL> lnk = CLinker ( ) . accept ( Env ( [ x , x ] , [ e ] ) ) <EOL> fn = lnk . make_function ( ) <EOL> assert fn ( <NUM_LIT> , <NUM_LIT> ) == <NUM_LIT:4> <EOL> def test_clinker_not_used_inputs ( ) : <EOL> if not theano . config . cxx : <EOL> raise SkipTest ( "<STR_LIT>" ) <EOL> x , y , z = inputs ( ) <EOL> e = add ( x , y ) <EOL> lnk = CLinker ( ) . accept ( Env ( [ x , y , z ] , [ e ] ) ) <EOL> fn = lnk . make_function ( ) <EOL> assert fn ( <NUM_LIT> , <NUM_LIT> , <NUM_LIT:1.0> ) == <NUM_LIT> <EOL> def test_clinker_dups_inner ( ) : <EOL> if not theano . config . cxx : <EOL> raise SkipTest ( "<STR_LIT>" ) <EOL> x , y , z = inputs ( ) <EOL> e = add ( mul ( y , y ) , add ( x , z ) ) <EOL> lnk = CLinker ( ) . accept ( Env ( [ x , y , z ] , [ e ] ) ) <EOL> fn = lnk . make_function ( ) <EOL> assert fn ( <NUM_LIT:1.0> , <NUM_LIT> , <NUM_LIT> ) == <NUM_LIT> <EOL> def test_opwiseclinker_straightforward ( ) : <EOL> x , y , z = inputs ( ) <EOL> e = add ( mul ( add ( x , y ) , div ( x , y ) ) , bad_sub ( bad_sub ( x , y ) , z ) ) <EOL> lnk = OpWiseCLinker ( ) . accept ( Env ( [ x , y , z ] , [ e ] ) ) <EOL> fn = lnk . make_function ( ) <EOL> if theano . config . cxx : <EOL> assert fn ( <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) == <NUM_LIT> <EOL> else : <EOL> assert fn ( <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) == - <NUM_LIT:6> <EOL> def test_opwiseclinker_constant ( ) : <EOL> x , y , z = inputs ( ) <EOL> x = Constant ( tdouble , <NUM_LIT> , name = '<STR_LIT:x>' ) <EOL> e = add ( mul ( x , y ) , mul ( y , z ) ) <EOL> lnk = OpWiseCLinker ( ) . accept ( Env ( [ y , z ] , [ e ] ) ) <EOL> fn = lnk . make_function ( ) <EOL> res = fn ( <NUM_LIT> , <NUM_LIT> ) <EOL> assert res == <NUM_LIT> <EOL> class MyExc ( Exception ) : <EOL> pass <EOL> def _my_checker ( x , y ) : <EOL> if x [ <NUM_LIT:0> ] != y [ <NUM_LIT:0> ] : <EOL> raise MyExc ( "<STR_LIT>" , <EOL> { '<STR_LIT>' : x [ <NUM_LIT:0> ] , '<STR_LIT>' : y [ <NUM_LIT:0> ] } ) <EOL> def test_duallinker_straightforward ( ) : <EOL> x , y , z = inputs ( ) <EOL> e = add ( mul ( x , y ) , mul ( y , z ) ) <EOL> lnk = DualLinker ( checker = _my_checker ) . accept ( Env ( [ x , y , z ] , [ e ] ) ) <EOL> fn = lnk . make_function ( ) <EOL> res = fn ( <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) <EOL> assert res == <NUM_LIT> <EOL> def test_duallinker_mismatch ( ) : <EOL> if not theano . config . cxx : <EOL> raise SkipTest ( "<STR_LIT>" ) <EOL> x , y , z = inputs ( ) <EOL> e = bad_sub ( mul ( x , y ) , mul ( y , z ) ) <EOL> g = Env ( [ x , y , z ] , [ e ] ) <EOL> lnk = DualLinker ( checker = _my_checker ) . accept ( g ) <EOL> fn = lnk . make_function ( ) <EOL> assert CLinker ( ) . accept ( g ) . make_function ( ) ( <NUM_LIT:1.0> , <NUM_LIT> , <NUM_LIT> ) == - <NUM_LIT> <EOL> assert OpWiseCLinker ( ) . accept ( g ) . make_function ( ) ( <NUM_LIT:1.0> , <NUM_LIT> , <NUM_LIT> ) == - <NUM_LIT> <EOL> assert PerformLinker ( ) . accept ( g ) . make_function ( ) ( <NUM_LIT:1.0> , <NUM_LIT> , <NUM_LIT> ) == - <NUM_LIT> <EOL> try : <EOL> fn ( <NUM_LIT:1.0> , <NUM_LIT> , <NUM_LIT> ) <EOL> raise Exception ( "<STR_LIT>" ) <EOL> except MyExc as e : <EOL> pass <EOL> class AddFail ( Binary ) : <EOL> def c_code ( self , node , name , inp , out , sub ) : <EOL> x , y = inp <EOL> z , = out <EOL> fail = sub [ '<STR_LIT>' ] <EOL> return """<STR_LIT>""" % locals ( ) <EOL> def impl ( self , x , y ) : <EOL> return x + y <EOL> add_fail = AddFail ( ) <EOL> def test_c_fail_error ( ) : <EOL> if not theano . config . cxx : <EOL> raise SkipTest ( "<STR_LIT>" ) <EOL> x , y , z = inputs ( ) <EOL> x = Constant ( tdouble , <NUM_LIT> , name = '<STR_LIT:x>' ) <EOL> e = add_fail ( mul ( x , y ) , mul ( y , z ) ) <EOL> lnk = OpWiseCLinker ( ) . accept ( Env ( [ y , z ] , [ e ] ) ) <EOL> fn = lnk . make_function ( ) <EOL> try : <EOL> fn ( <NUM_LIT> , <NUM_LIT> ) <EOL> except RuntimeError : <EOL> print ( '<STR_LIT>' ) <EOL> return <EOL> assert <NUM_LIT:0> <EOL> def test_shared_input_output ( ) : <EOL> inc = theano . tensor . iscalar ( '<STR_LIT>' ) <EOL> state = theano . shared ( <NUM_LIT:0> ) <EOL> state . name = '<STR_LIT:state>' <EOL> linker = theano . gof . CLinker ( ) <EOL> mode = theano . Mode ( linker = linker ) <EOL> f = theano . function ( [ inc ] , state , updates = [ ( state , state + inc ) ] , <EOL> mode = mode ) <EOL> g = theano . function ( [ inc ] , state , updates = [ ( state , state + inc ) ] ) <EOL> f0 = f ( <NUM_LIT:0> ) <EOL> g0 = g ( <NUM_LIT:0> ) <EOL> assert f0 == g0 == <NUM_LIT:0> , ( f0 , g0 ) <EOL> f2 = f ( <NUM_LIT:2> ) <EOL> assert f2 == f0 , ( f2 , f0 ) <EOL> f0 = f ( <NUM_LIT:0> ) <EOL> g0 = g ( <NUM_LIT:0> ) <EOL> assert f0 == g0 == <NUM_LIT:2> , ( f0 , g0 ) <EOL> g3 = g ( <NUM_LIT:3> ) <EOL> assert g3 == g0 , ( g3 , g0 ) <EOL> f0 = f ( <NUM_LIT:0> ) <EOL> g0 = g ( <NUM_LIT:0> ) <EOL> assert f0 == g0 == <NUM_LIT:5> , ( f0 , g0 ) <EOL> vstate = theano . shared ( numpy . zeros ( <NUM_LIT:3> , dtype = '<STR_LIT>' ) ) <EOL> vstate . name = '<STR_LIT>' <EOL> fv = theano . function ( [ inc ] , vstate , updates = [ ( vstate , vstate + inc ) ] , <EOL> mode = mode ) <EOL> gv = theano . function ( [ inc ] , vstate , updates = [ ( vstate , vstate + inc ) ] ) <EOL> fv0 = fv ( <NUM_LIT:0> ) <EOL> gv0 = gv ( <NUM_LIT:0> ) <EOL> assert numpy . all ( fv0 == <NUM_LIT:0> ) , fv0 <EOL> assert numpy . all ( gv0 == <NUM_LIT:0> ) , gv0 <EOL> fv2 = fv ( <NUM_LIT:2> ) <EOL> assert numpy . all ( fv2 == fv0 ) , ( fv2 , fv0 ) <EOL> fv0 = fv ( <NUM_LIT:0> ) <EOL> gv0 = gv ( <NUM_LIT:0> ) <EOL> assert numpy . all ( fv0 == <NUM_LIT:2> ) , fv0 <EOL> assert numpy . all ( gv0 == <NUM_LIT:2> ) , gv0 <EOL> gv3 = gv ( <NUM_LIT:3> ) <EOL> assert numpy . all ( gv3 == gv0 ) , ( gv3 , gv0 ) <EOL> fv0 = fv ( <NUM_LIT:0> ) <EOL> gv0 = gv ( <NUM_LIT:0> ) <EOL> assert numpy . all ( fv0 == <NUM_LIT:5> ) , fv0 <EOL> assert numpy . all ( gv0 == <NUM_LIT:5> ) , gv0 </s>
