<s> from __future__ import absolute_import , print_function , division <EOL> from copy import copy , deepcopy <EOL> from functools import wraps <EOL> import logging <EOL> import sys <EOL> import unittest <EOL> from nose_parameterized import parameterized <EOL> from six import integer_types <EOL> from six . moves import StringIO <EOL> try : <EOL> from nose . plugins . attrib import attr <EOL> except ImportError : <EOL> def attr ( tag ) : <EOL> def func ( f ) : <EOL> return f <EOL> return func <EOL> import numpy <EOL> import theano <EOL> import theano . tensor as T <EOL> from theano . configparser import config <EOL> try : <EOL> from nose . plugins . skip import SkipTest <EOL> except ImportError : <EOL> class SkipTest ( Exception ) : <EOL> """<STR_LIT>""" <EOL> _logger = logging . getLogger ( "<STR_LIT>" ) <EOL> def custom_name_func ( testcase_func , param_num , param ) : <EOL> return "<STR_LIT>" % ( <EOL> testcase_func . __name__ , <EOL> parameterized . to_safe_name ( "<STR_LIT:_>" . join ( str ( x ) for x in param . args ) ) , <EOL> ) <EOL> def fetch_seed ( pseed = None ) : <EOL> """<STR_LIT>""" <EOL> seed = pseed or config . unittests . rseed <EOL> if seed == '<STR_LIT>' : <EOL> seed = None <EOL> try : <EOL> if seed : <EOL> seed = int ( seed ) <EOL> else : <EOL> seed = None <EOL> except ValueError : <EOL> print ( ( '<STR_LIT>' '<STR_LIT>' ) , file = sys . stderr ) <EOL> seed = None <EOL> return seed <EOL> def seed_rng ( pseed = None ) : <EOL> """<STR_LIT>""" <EOL> seed = fetch_seed ( pseed ) <EOL> if pseed and pseed != seed : <EOL> print ( '<STR_LIT>' '<STR_LIT>' % ( seed , pseed ) , file = sys . stderr ) <EOL> numpy . random . seed ( seed ) <EOL> return seed <EOL> def verify_grad ( op , pt , n_tests = <NUM_LIT:2> , rng = None , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if rng is None : <EOL> seed_rng ( ) <EOL> rng = numpy . random <EOL> T . verify_grad ( op , pt , n_tests , rng , * args , ** kwargs ) <EOL> verify_grad . E_grad = T . verify_grad . E_grad <EOL> class TestOptimizationMixin ( object ) : <EOL> def assertFunctionContains ( self , f , op , min = <NUM_LIT:1> , max = sys . maxsize ) : <EOL> toposort = f . maker . fgraph . toposort ( ) <EOL> matches = [ node for node in toposort if node . op == op ] <EOL> assert ( min <= len ( matches ) <= max ) , ( toposort , matches , <EOL> str ( op ) , len ( matches ) , min , max ) <EOL> def assertFunctionContains0 ( self , f , op ) : <EOL> return self . assertFunctionContains ( f , op , min = <NUM_LIT:0> , max = <NUM_LIT:0> ) <EOL> def assertFunctionContains1 ( self , f , op ) : <EOL> return self . assertFunctionContains ( f , op , min = <NUM_LIT:1> , max = <NUM_LIT:1> ) <EOL> def assertFunctionContainsN ( self , f , op , N ) : <EOL> return self . assertFunctionContains ( f , op , min = N , max = N ) <EOL> def assertFunctionContainsClass ( self , f , op , min = <NUM_LIT:1> , max = sys . maxsize ) : <EOL> toposort = f . maker . fgraph . toposort ( ) <EOL> matches = [ node for node in toposort if isinstance ( node . op , op ) ] <EOL> assert ( min <= len ( matches ) <= max ) , ( toposort , matches , <EOL> str ( op ) , len ( matches ) , min , max ) <EOL> def assertFunctionContainsClassN ( self , f , op , N ) : <EOL> return self . assertFunctionContainsClass ( f , op , min = N , max = N ) <EOL> def SkipTest ( self , msg = '<STR_LIT>' ) : <EOL> raise SkipTest ( msg ) <EOL> class T_OpContractMixin ( object ) : <EOL> other_op = T . add <EOL> def copy ( self , x ) : <EOL> return copy ( x ) <EOL> def deepcopy ( self , x ) : <EOL> return deepcopy ( x ) <EOL> def clone ( self , op ) : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> def test_eq ( self ) : <EOL> for i , op_i in enumerate ( self . ops ) : <EOL> assert op_i == op_i <EOL> assert op_i == self . copy ( op_i ) <EOL> assert op_i == self . deepcopy ( op_i ) <EOL> assert op_i == self . clone ( op_i ) <EOL> assert op_i != self . other_op <EOL> for j , op_j in enumerate ( self . ops ) : <EOL> if i == j : <EOL> continue <EOL> assert op_i != op_j <EOL> def test_hash ( self ) : <EOL> for i , op_i in enumerate ( self . ops ) : <EOL> h_i = hash ( op_i ) <EOL> assert h_i == hash ( op_i ) <EOL> assert h_i == hash ( self . copy ( op_i ) ) <EOL> assert h_i == hash ( self . deepcopy ( op_i ) ) <EOL> assert h_i == hash ( self . clone ( op_i ) ) <EOL> assert h_i != hash ( self . other_op ) <EOL> for j , op_j in enumerate ( self . ops ) : <EOL> if i == j : <EOL> continue <EOL> assert op_i != hash ( op_j ) <EOL> def test_name ( self ) : <EOL> for op in self . ops : <EOL> s = str ( op ) <EOL> assert s <EOL> class InferShapeTester ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> seed_rng ( ) <EOL> mode = getattr ( self , '<STR_LIT>' , None ) <EOL> if mode is None : <EOL> mode = theano . compile . get_default_mode ( ) <EOL> self . mode = mode . including ( "<STR_LIT>" ) <EOL> def _compile_and_check ( self , inputs , outputs , numeric_inputs , cls , <EOL> excluding = None , warn = True , check_topo = True ) : <EOL> """<STR_LIT>""" <EOL> mode = self . mode <EOL> if excluding : <EOL> mode = mode . excluding ( * excluding ) <EOL> if warn : <EOL> for var , inp in zip ( inputs , numeric_inputs ) : <EOL> if isinstance ( inp , ( integer_types , float , list , tuple ) ) : <EOL> inp = var . type . filter ( inp ) <EOL> if not hasattr ( inp , "<STR_LIT>" ) : <EOL> continue <EOL> if hasattr ( var . type , "<STR_LIT>" ) : <EOL> shp = [ inp . shape [ i ] for i in range ( inp . ndim ) <EOL> if not var . type . broadcastable [ i ] ] <EOL> else : <EOL> shp = inp . shape <EOL> if len ( set ( shp ) ) != len ( shp ) : <EOL> _logger . warn ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> cls , inp . shape ) <EOL> break <EOL> outputs_function = theano . function ( inputs , outputs , mode = mode ) <EOL> shapes_function = theano . function ( inputs , [ o . shape for o in outputs ] , <EOL> mode = mode ) <EOL> if check_topo : <EOL> topo_shape = shapes_function . maker . fgraph . toposort ( ) <EOL> assert not any ( isinstance ( t . op , cls ) for t in topo_shape ) <EOL> topo_out = outputs_function . maker . fgraph . toposort ( ) <EOL> assert any ( isinstance ( t . op , cls ) for t in topo_out ) <EOL> numeric_outputs = outputs_function ( * numeric_inputs ) <EOL> numeric_shapes = shapes_function ( * numeric_inputs ) <EOL> for out , shape in zip ( numeric_outputs , numeric_shapes ) : <EOL> assert numpy . all ( out . shape == shape ) , ( out . shape , shape ) <EOL> def str_diagnostic ( expected , value , rtol , atol ) : <EOL> """<STR_LIT>""" <EOL> sio = StringIO ( ) <EOL> try : <EOL> ssio = StringIO ( ) <EOL> print ( "<STR_LIT>" , file = ssio ) <EOL> print ( "<STR_LIT>" , end = '<STR_LIT:U+0020>' , file = ssio ) <EOL> print ( expected . shape , end = '<STR_LIT:U+0020>' , file = ssio ) <EOL> print ( expected . dtype , end = '<STR_LIT:U+0020>' , file = ssio ) <EOL> print ( expected . strides , end = '<STR_LIT:U+0020>' , file = ssio ) <EOL> print ( expected . min ( ) , end = '<STR_LIT:U+0020>' , file = ssio ) <EOL> print ( expected . max ( ) , end = '<STR_LIT:U+0020>' , file = ssio ) <EOL> print ( numpy . isinf ( expected ) . sum ( ) , end = '<STR_LIT:U+0020>' , file = ssio ) <EOL> print ( numpy . isnan ( expected ) . sum ( ) , end = '<STR_LIT:U+0020>' , file = ssio ) <EOL> print ( ssio . getvalue ( ) , file = sio ) <EOL> except Exception : <EOL> pass <EOL> try : <EOL> ssio = StringIO ( ) <EOL> print ( "<STR_LIT>" , end = '<STR_LIT:U+0020>' , file = ssio ) <EOL> print ( value . shape , end = '<STR_LIT:U+0020>' , file = ssio ) <EOL> print ( value . dtype , end = '<STR_LIT:U+0020>' , file = ssio ) <EOL> print ( value . strides , end = '<STR_LIT:U+0020>' , file = ssio ) <EOL> print ( value . min ( ) , end = '<STR_LIT:U+0020>' , file = ssio ) <EOL> print ( value . max ( ) , end = '<STR_LIT:U+0020>' , file = ssio ) <EOL> print ( numpy . isinf ( value ) . sum ( ) , end = '<STR_LIT:U+0020>' , file = ssio ) <EOL> print ( numpy . isnan ( value ) . sum ( ) , end = '<STR_LIT:U+0020>' , file = ssio ) <EOL> print ( ssio . getvalue ( ) , file = sio ) <EOL> except Exception : <EOL> pass <EOL> print ( "<STR_LIT>" , expected , file = sio ) <EOL> print ( "<STR_LIT>" , value , file = sio ) <EOL> try : <EOL> ov = numpy . asarray ( expected ) <EOL> nv = numpy . asarray ( value ) <EOL> ssio = StringIO ( ) <EOL> absdiff = numpy . absolute ( nv - ov ) <EOL> print ( "<STR_LIT>" , numpy . max ( absdiff ) , file = ssio ) <EOL> print ( "<STR_LIT>" , numpy . mean ( absdiff ) , file = ssio ) <EOL> print ( "<STR_LIT>" , numpy . median ( absdiff ) , file = ssio ) <EOL> print ( "<STR_LIT>" , numpy . std ( absdiff ) , file = ssio ) <EOL> reldiff = numpy . absolute ( nv - ov ) / ( numpy . absolute ( nv ) + <EOL> numpy . absolute ( ov ) ) <EOL> print ( "<STR_LIT>" , numpy . max ( reldiff ) , file = ssio ) <EOL> print ( "<STR_LIT>" , numpy . mean ( reldiff ) , file = ssio ) <EOL> print ( "<STR_LIT>" , numpy . median ( reldiff ) , file = ssio ) <EOL> print ( "<STR_LIT>" , numpy . std ( reldiff ) , file = ssio ) <EOL> print ( ssio . getvalue ( ) , file = sio ) <EOL> except Exception : <EOL> pass <EOL> atol_ , rtol_ = T . basic . _get_atol_rtol ( expected , value ) <EOL> if rtol is not None : <EOL> rtol_ = rtol <EOL> if atol is not None : <EOL> atol_ = atol <EOL> print ( "<STR_LIT>" , rtol_ , atol_ , file = sio ) <EOL> return sio . getvalue ( ) <EOL> class WrongValue ( Exception ) : <EOL> def __init__ ( self , expected_val , val , rtol , atol ) : <EOL> Exception . __init__ ( self ) <EOL> self . val1 = expected_val <EOL> self . val2 = val <EOL> self . rtol = rtol <EOL> self . atol = atol <EOL> def __str__ ( self ) : <EOL> s = "<STR_LIT>" <EOL> return s + str_diagnostic ( self . val1 , self . val2 , self . rtol , self . atol ) <EOL> def assert_allclose ( val1 , val2 , rtol = None , atol = None ) : <EOL> if not T . basic . _allclose ( val1 , val2 , rtol , atol ) : <EOL> raise WrongValue ( val1 , val2 , rtol , atol ) <EOL> class AttemptManyTimes : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , n_attempts , n_req_successes = <NUM_LIT:1> ) : <EOL> assert n_attempts >= n_req_successes <EOL> self . n_attempts = n_attempts <EOL> self . n_req_successes = n_req_successes <EOL> def __call__ ( self , fct ) : <EOL> @ wraps ( fct ) <EOL> def attempt_multiple_times ( * args , ** kwargs ) : <EOL> original_seed = config . unittests . rseed <EOL> current_seed = original_seed <EOL> if ( len ( args ) == <NUM_LIT:1> and hasattr ( args [ <NUM_LIT:0> ] , "<STR_LIT>" ) ) : <EOL> test_in_class = True <EOL> class_instance = args [ <NUM_LIT:0> ] <EOL> else : <EOL> test_in_class = False <EOL> n_fail = <NUM_LIT:0> <EOL> n_success = <NUM_LIT:0> <EOL> for i in range ( self . n_attempts ) : <EOL> try : <EOL> config . unittests . rseed = current_seed <EOL> if test_in_class and hasattr ( class_instance , "<STR_LIT>" ) : <EOL> class_instance . setUp ( ) <EOL> fct ( * args , ** kwargs ) <EOL> n_success += <NUM_LIT:1> <EOL> if n_success == self . n_req_successes : <EOL> break <EOL> except Exception : <EOL> n_fail += <NUM_LIT:1> <EOL> if n_fail + self . n_req_successes > self . n_attempts : <EOL> raise <EOL> finally : <EOL> config . unittests . rseed = original_seed <EOL> if test_in_class and hasattr ( class_instance , "<STR_LIT>" ) : <EOL> class_instance . tearDown ( ) <EOL> if current_seed not in [ None , "<STR_LIT>" ] : <EOL> current_seed = str ( int ( current_seed ) + <NUM_LIT:1> ) <EOL> return attempt_multiple_times </s>
