<s> from __future__ import print_function , division , absolute_import <EOL> from timeit import Timer <EOL> import numpy as np <EOL> from numba import njit , vectorize <EOL> from numba import unittest_support as unittest <EOL> from numba import compiler , typing , typeof , ir <EOL> from numba . compiler import Pipeline , _PipelineManager , Flags <EOL> from numba . targets import cpu <EOL> from . support import MemoryLeakMixin , TestCase <EOL> class Namespace ( dict ) : <EOL> def __getattr__ ( s , k ) : <EOL> return s [ k ] if k in s else super ( Namespace , s ) . __getattr__ ( k ) <EOL> def axy ( a , x , y ) : <EOL> return a * x + y <EOL> def ax2 ( a , x , y ) : <EOL> return a * x + y <EOL> def pos_root ( As , Bs , Cs ) : <EOL> return ( - Bs + ( ( ( Bs ** <NUM_LIT> ) - ( <NUM_LIT> * As * Cs ) ) ** <NUM_LIT:0.5> ) ) / ( <NUM_LIT> * As ) <EOL> def neg_root_common_subexpr ( As , Bs , Cs ) : <EOL> _2As = <NUM_LIT> * As <EOL> _4AsCs = <NUM_LIT> * _2As * Cs <EOL> _Bs2_4AsCs = ( Bs ** <NUM_LIT> - _4AsCs ) <EOL> return ( - Bs - ( _Bs2_4AsCs ** <NUM_LIT:0.5> ) ) / _2As <EOL> def neg_root_complex_subexpr ( As , Bs , Cs ) : <EOL> _2As = <NUM_LIT> * As <EOL> _4AsCs = <NUM_LIT> * _2As * Cs <EOL> _Bs2_4AsCs = ( Bs ** <NUM_LIT> - _4AsCs ) + <NUM_LIT> <EOL> return ( - Bs - ( _Bs2_4AsCs ** <NUM_LIT:0.5> ) ) / _2As <EOL> vaxy = vectorize ( axy ) <EOL> def call_stuff ( a0 , a1 ) : <EOL> return np . cos ( vaxy ( a0 , np . sin ( a1 ) - <NUM_LIT:1.> , <NUM_LIT:1.> ) ) <EOL> def are_roots_imaginary ( As , Bs , Cs ) : <EOL> return ( Bs ** <NUM_LIT:2> - <NUM_LIT:4> * As * Cs ) < <NUM_LIT:0> <EOL> def div_add ( As , Bs , Cs ) : <EOL> return As / Bs + Cs <EOL> def distance_matrix ( vectors ) : <EOL> n_vectors = vectors . shape [ <NUM_LIT:0> ] <EOL> result = np . empty ( ( n_vectors , n_vectors ) , dtype = np . float64 ) <EOL> for i in range ( n_vectors ) : <EOL> for j in range ( i , n_vectors ) : <EOL> result [ i , j ] = result [ j , i ] = np . sum ( <EOL> ( vectors [ i ] - vectors [ j ] ) ** <NUM_LIT:2> ) ** <NUM_LIT:0.5> <EOL> return result <EOL> class RewritesTester ( Pipeline ) : <EOL> @ classmethod <EOL> def mk_pipeline ( cls , args , return_type = None , flags = None , locals = { } , <EOL> library = None , typing_context = None , target_context = None ) : <EOL> if not flags : <EOL> flags = Flags ( ) <EOL> flags . nrt = True <EOL> if typing_context is None : <EOL> typing_context = typing . Context ( ) <EOL> if target_context is None : <EOL> target_context = cpu . CPUContext ( typing_context ) <EOL> return cls ( typing_context , target_context , library , args , return_type , <EOL> flags , locals ) <EOL> @ classmethod <EOL> def mk_no_rw_pipeline ( cls , args , return_type = None , flags = None , locals = { } , <EOL> library = None , ** kws ) : <EOL> if not flags : <EOL> flags = Flags ( ) <EOL> flags . no_rewrites = True <EOL> return cls . mk_pipeline ( args , return_type , flags , locals , library , ** kws ) <EOL> class TestArrayExpressions ( MemoryLeakMixin , TestCase ) : <EOL> def test_simple_expr ( self ) : <EOL> '''<STR_LIT>''' <EOL> A = np . linspace ( <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:10> ) <EOL> X = np . linspace ( <NUM_LIT:2> , <NUM_LIT:1> , <NUM_LIT:10> ) <EOL> Y = np . linspace ( <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:10> ) <EOL> arg_tys = [ typeof ( arg ) for arg in ( A , X , Y ) ] <EOL> control_pipeline = RewritesTester . mk_no_rw_pipeline ( arg_tys ) <EOL> cres_0 = control_pipeline . compile_extra ( axy ) <EOL> nb_axy_0 = cres_0 . entry_point <EOL> test_pipeline = RewritesTester . mk_pipeline ( arg_tys ) <EOL> cres_1 = test_pipeline . compile_extra ( axy ) <EOL> nb_axy_1 = cres_1 . entry_point <EOL> control_pipeline2 = RewritesTester . mk_no_rw_pipeline ( arg_tys ) <EOL> cres_2 = control_pipeline2 . compile_extra ( ax2 ) <EOL> nb_ctl = cres_2 . entry_point <EOL> expected = nb_axy_0 ( A , X , Y ) <EOL> actual = nb_axy_1 ( A , X , Y ) <EOL> control = nb_ctl ( A , X , Y ) <EOL> np . testing . assert_array_equal ( expected , actual ) <EOL> np . testing . assert_array_equal ( control , actual ) <EOL> ir0 = control_pipeline . interp . blocks <EOL> ir1 = test_pipeline . interp . blocks <EOL> ir2 = control_pipeline2 . interp . blocks <EOL> self . assertEqual ( len ( ir0 ) , len ( ir1 ) ) <EOL> self . assertEqual ( len ( ir0 ) , len ( ir2 ) ) <EOL> self . assertGreater ( len ( ir0 [ <NUM_LIT:0> ] . body ) , len ( ir1 [ <NUM_LIT:0> ] . body ) ) <EOL> self . assertEqual ( len ( ir0 [ <NUM_LIT:0> ] . body ) , len ( ir2 [ <NUM_LIT:0> ] . body ) ) <EOL> def _get_array_exprs ( self , block ) : <EOL> for instr in block : <EOL> if isinstance ( instr , ir . Assign ) : <EOL> if isinstance ( instr . value , ir . Expr ) : <EOL> if instr . value . op == '<STR_LIT>' : <EOL> yield instr <EOL> def _array_expr_to_set ( self , expr , out = None ) : <EOL> '''<STR_LIT>''' <EOL> if out is None : <EOL> out = set ( ) <EOL> if not isinstance ( expr , tuple ) : <EOL> raise ValueError ( "<STR_LIT>" . format ( expr ) ) <EOL> operation , operands = expr <EOL> processed_operands = [ ] <EOL> for operand in operands : <EOL> if isinstance ( operand , tuple ) : <EOL> operand , _ = self . _array_expr_to_set ( operand , out ) <EOL> processed_operands . append ( operand ) <EOL> processed_expr = operation , tuple ( processed_operands ) <EOL> out . add ( processed_expr ) <EOL> return processed_expr , out <EOL> def _test_root_function ( self , fn = pos_root ) : <EOL> A = np . random . random ( <NUM_LIT:10> ) <EOL> B = np . random . random ( <NUM_LIT:10> ) + <NUM_LIT:1.> <EOL> C = np . random . random ( <NUM_LIT:10> ) <EOL> arg_tys = [ typeof ( arg ) for arg in ( A , B , C ) ] <EOL> control_pipeline = RewritesTester . mk_no_rw_pipeline ( arg_tys ) <EOL> control_cres = control_pipeline . compile_extra ( fn ) <EOL> nb_fn_0 = control_cres . entry_point <EOL> test_pipeline = RewritesTester . mk_pipeline ( arg_tys ) <EOL> test_cres = test_pipeline . compile_extra ( fn ) <EOL> nb_fn_1 = test_cres . entry_point <EOL> np_result = fn ( A , B , C ) <EOL> nb_result_0 = nb_fn_0 ( A , B , C ) <EOL> nb_result_1 = nb_fn_1 ( A , B , C ) <EOL> np . testing . assert_array_almost_equal ( np_result , nb_result_0 ) <EOL> np . testing . assert_array_almost_equal ( nb_result_0 , nb_result_1 ) <EOL> return Namespace ( locals ( ) ) <EOL> def _assert_total_rewrite ( self , ir0 , ir1 ) : <EOL> '''<STR_LIT>''' <EOL> self . assertEqual ( len ( ir0 ) , len ( ir1 ) ) <EOL> self . assertGreater ( len ( ir0 [ <NUM_LIT:0> ] . body ) , len ( ir1 [ <NUM_LIT:0> ] . body ) ) <EOL> self . assertEqual ( len ( list ( self . _get_array_exprs ( ir0 [ <NUM_LIT:0> ] . body ) ) ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( len ( list ( self . _get_array_exprs ( ir1 [ <NUM_LIT:0> ] . body ) ) ) , <NUM_LIT:1> ) <EOL> def test_complicated_expr ( self ) : <EOL> '''<STR_LIT>''' <EOL> ns = self . _test_root_function ( ) <EOL> self . _assert_total_rewrite ( ns . control_pipeline . interp . blocks , <EOL> ns . test_pipeline . interp . blocks ) <EOL> def test_common_subexpressions ( self , fn = neg_root_common_subexpr ) : <EOL> '''<STR_LIT>''' <EOL> ns = self . _test_root_function ( fn ) <EOL> ir0 = ns . control_pipeline . interp . blocks <EOL> ir1 = ns . test_pipeline . interp . blocks <EOL> self . assertEqual ( len ( ir0 ) , len ( ir1 ) ) <EOL> self . assertGreater ( len ( ir0 [ <NUM_LIT:0> ] . body ) , len ( ir1 [ <NUM_LIT:0> ] . body ) ) <EOL> self . assertEqual ( len ( list ( self . _get_array_exprs ( ir0 [ <NUM_LIT:0> ] . body ) ) ) , <NUM_LIT:0> ) <EOL> array_expr_instrs = list ( self . _get_array_exprs ( ir1 [ <NUM_LIT:0> ] . body ) ) <EOL> self . assertGreater ( len ( array_expr_instrs ) , <NUM_LIT:1> ) <EOL> array_sets = list ( self . _array_expr_to_set ( instr . value . expr ) [ <NUM_LIT:1> ] <EOL> for instr in array_expr_instrs ) <EOL> for expr_set_0 , expr_set_1 in zip ( array_sets [ : - <NUM_LIT:1> ] , array_sets [ <NUM_LIT:1> : ] ) : <EOL> intersections = expr_set_0 . intersection ( expr_set_1 ) <EOL> if intersections : <EOL> self . fail ( "<STR_LIT>" <EOL> "<STR_LIT>" . format ( intersections ) ) <EOL> def test_complex_subexpression ( self ) : <EOL> return self . test_common_subexpressions ( neg_root_complex_subexpr ) <EOL> def test_ufunc_and_dufunc_calls ( self ) : <EOL> '''<STR_LIT>''' <EOL> A = np . random . random ( <NUM_LIT:10> ) <EOL> B = np . random . random ( <NUM_LIT:10> ) <EOL> arg_tys = [ typeof ( arg ) for arg in ( A , B ) ] <EOL> vaxy_descr = vaxy . _dispatcher . targetdescr <EOL> control_pipeline = RewritesTester . mk_no_rw_pipeline ( <EOL> arg_tys , <EOL> typing_context = vaxy_descr . typing_context , <EOL> target_context = vaxy_descr . target_context ) <EOL> cres_0 = control_pipeline . compile_extra ( call_stuff ) <EOL> nb_call_stuff_0 = cres_0 . entry_point <EOL> test_pipeline = RewritesTester . mk_pipeline ( <EOL> arg_tys , <EOL> typing_context = vaxy_descr . typing_context , <EOL> target_context = vaxy_descr . target_context ) <EOL> cres_1 = test_pipeline . compile_extra ( call_stuff ) <EOL> nb_call_stuff_1 = cres_1 . entry_point <EOL> expected = call_stuff ( A , B ) <EOL> control = nb_call_stuff_0 ( A , B ) <EOL> actual = nb_call_stuff_1 ( A , B ) <EOL> np . testing . assert_array_almost_equal ( expected , control ) <EOL> np . testing . assert_array_almost_equal ( expected , actual ) <EOL> self . _assert_total_rewrite ( control_pipeline . interp . blocks , <EOL> test_pipeline . interp . blocks ) <EOL> def test_cmp_op ( self ) : <EOL> '''<STR_LIT>''' <EOL> ns = self . _test_root_function ( are_roots_imaginary ) <EOL> self . _assert_total_rewrite ( ns . control_pipeline . interp . blocks , <EOL> ns . test_pipeline . interp . blocks ) <EOL> class TestRewriteIssues ( MemoryLeakMixin , TestCase ) : <EOL> def test_issue_1184 ( self ) : <EOL> from numba import jit <EOL> import numpy as np <EOL> @ jit ( nopython = True ) <EOL> def foo ( arr ) : <EOL> return arr <EOL> @ jit ( nopython = True ) <EOL> def bar ( arr ) : <EOL> c = foo ( arr ) <EOL> d = foo ( arr ) <EOL> return c , d <EOL> arr = np . arange ( <NUM_LIT:10> ) <EOL> out_c , out_d = bar ( arr ) <EOL> self . assertIs ( out_c , out_d ) <EOL> self . assertIs ( out_c , arr ) <EOL> def test_issue_1264 ( self ) : <EOL> n = <NUM_LIT:100> <EOL> x = np . random . uniform ( size = n * <NUM_LIT:3> ) . reshape ( ( n , <NUM_LIT:3> ) ) <EOL> expected = distance_matrix ( x ) <EOL> actual = njit ( distance_matrix ) ( x ) <EOL> np . testing . assert_array_almost_equal ( expected , actual ) <EOL> def test_issue_1372 ( self ) : <EOL> """<STR_LIT>""" <EOL> from numba import njit <EOL> @ njit <EOL> def foo ( a , b ) : <EOL> b = np . sin ( b ) <EOL> return b + b + a <EOL> a = np . random . uniform ( <NUM_LIT:10> ) <EOL> b = np . random . uniform ( <NUM_LIT:10> ) <EOL> expect = foo . py_func ( a , b ) <EOL> got = foo ( a , b ) <EOL> np . testing . assert_allclose ( got , expect ) <EOL> def test_unary_arrayexpr ( self ) : <EOL> """<STR_LIT>""" <EOL> @ njit <EOL> def foo ( a , b ) : <EOL> return b - a + - a <EOL> b = <NUM_LIT> <EOL> a = np . arange ( <NUM_LIT:10> , dtype = np . int32 ) <EOL> expect = foo . py_func ( a , b ) <EOL> got = foo ( a , b ) <EOL> self . assertPreciseEqual ( got , expect ) <EOL> def test_bitwise_arrayexpr ( self ) : <EOL> """<STR_LIT>""" <EOL> @ njit <EOL> def foo ( a , b ) : <EOL> return ~ ( a & ( ~ b ) ) <EOL> a = np . array ( [ True , True , False , False ] ) <EOL> b = np . array ( [ False , True , False , True ] ) <EOL> expect = foo . py_func ( a , b ) <EOL> got = foo ( a , b ) <EOL> self . assertPreciseEqual ( got , expect ) <EOL> class TestSemantics ( MemoryLeakMixin , unittest . TestCase ) : <EOL> def test_division_by_zero ( self ) : <EOL> pyfunc = div_add <EOL> cfunc = njit ( pyfunc ) <EOL> a = np . float64 ( [ <NUM_LIT:0.0> , <NUM_LIT:1.0> , float ( '<STR_LIT>' ) ] ) <EOL> b = np . float64 ( [ <NUM_LIT:0.0> , <NUM_LIT:0.0> , <NUM_LIT:1.0> ] ) <EOL> c = np . ones_like ( a ) <EOL> expect = pyfunc ( a , b , c ) <EOL> got = cfunc ( a , b , c ) <EOL> np . testing . assert_array_equal ( expect , got ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> unittest . main ( ) </s>
