<s> import numpy as np <EOL> import theano <EOL> import theano . tensor as T <EOL> import collections as C <EOL> import anglepy as ap <EOL> import anglepy . ndict as ndict <EOL> from anglepy . misc import lazytheanofunc <EOL> import math , inspect <EOL> def shared32 ( x , name = None , borrow = False ) : <EOL> return theano . shared ( np . asarray ( x , dtype = '<STR_LIT>' ) , name = name , borrow = borrow ) <EOL> '''<STR_LIT>''' <EOL> class GPUVAE_YZ_X ( ap . GPUVAEModel ) : <EOL> def __init__ ( self , get_optimizer , n_x , n_y , n_hidden_q , n_z , n_hidden_p , nonlinear_q = '<STR_LIT>' , nonlinear_p = '<STR_LIT>' , type_px = '<STR_LIT>' , type_qz = '<STR_LIT>' , type_pz = '<STR_LIT>' , prior_sd = <NUM_LIT:1> , init_sd = <NUM_LIT> , var_smoothing = <NUM_LIT:0> , n_mixture = <NUM_LIT:0> , uniform_y = False ) : <EOL> self . constr = ( __name__ , inspect . stack ( ) [ <NUM_LIT:0> ] [ <NUM_LIT:3> ] , locals ( ) ) <EOL> self . n_x = n_x <EOL> self . n_y = n_y <EOL> self . n_hidden_q = n_hidden_q <EOL> self . n_z = n_z <EOL> self . n_hidden_p = n_hidden_p <EOL> self . dropout = False <EOL> self . nonlinear_q = nonlinear_q <EOL> self . nonlinear_p = nonlinear_p <EOL> self . type_px = type_px <EOL> self . type_qz = type_qz <EOL> self . type_pz = type_pz <EOL> self . prior_sd = prior_sd <EOL> self . var_smoothing = var_smoothing <EOL> self . n_mixture = n_mixture <EOL> self . uniform_y = uniform_y <EOL> v , w = self . init_w ( <NUM_LIT> ) <EOL> for i in v : v [ i ] = shared32 ( v [ i ] ) <EOL> for i in w : w [ i ] = shared32 ( w [ i ] ) <EOL> self . v = v <EOL> self . w = w <EOL> super ( GPUVAE_YZ_X , self ) . __init__ ( get_optimizer ) <EOL> def factors ( self , x , z , A ) : <EOL> v = self . v <EOL> w = self . w <EOL> '''<STR_LIT>''' <EOL> def f_softplus ( x ) : return T . log ( T . exp ( x ) + <NUM_LIT:1> ) <EOL> def f_rectlin ( x ) : return x * ( x > <NUM_LIT:0> ) <EOL> def f_rectlin2 ( x ) : return x * ( x > <NUM_LIT:0> ) + <NUM_LIT> * x <EOL> nonlinear = { '<STR_LIT>' : T . tanh , '<STR_LIT>' : T . nnet . sigmoid , '<STR_LIT>' : f_softplus , '<STR_LIT>' : f_rectlin , '<STR_LIT>' : f_rectlin2 } <EOL> nonlinear_q = nonlinear [ self . nonlinear_q ] <EOL> nonlinear_p = nonlinear [ self . nonlinear_p ] <EOL> import theano . tensor . shared_randomstreams <EOL> rng = theano . tensor . shared_randomstreams . RandomStreams ( <NUM_LIT:0> ) <EOL> hidden_q = [ nonlinear_q ( T . dot ( v [ '<STR_LIT>' ] , x [ '<STR_LIT:x>' ] ) + T . dot ( v [ '<STR_LIT>' ] , x [ '<STR_LIT:y>' ] ) + T . dot ( v [ '<STR_LIT>' ] , A ) ) ] <EOL> for i in range ( <NUM_LIT:1> , len ( self . n_hidden_q ) ) : <EOL> hidden_q . append ( nonlinear_q ( T . dot ( v [ '<STR_LIT:w>' + str ( i ) ] , hidden_q [ - <NUM_LIT:1> ] ) + T . dot ( v [ '<STR_LIT:b>' + str ( i ) ] , A ) ) ) <EOL> q_mean = T . dot ( v [ '<STR_LIT>' ] , hidden_q [ - <NUM_LIT:1> ] ) + T . dot ( v [ '<STR_LIT>' ] , A ) <EOL> if self . type_qz == '<STR_LIT>' or self . type_qz == '<STR_LIT>' : <EOL> q_logvar = T . dot ( v [ '<STR_LIT>' ] , hidden_q [ - <NUM_LIT:1> ] ) + T . dot ( v [ '<STR_LIT>' ] , A ) <EOL> else : raise Exception ( ) <EOL> theanofunc = lazytheanofunc ( '<STR_LIT>' , mode = '<STR_LIT>' ) <EOL> self . dist_qz [ '<STR_LIT:z>' ] = theanofunc ( [ x [ '<STR_LIT:x>' ] , x [ '<STR_LIT:y>' ] ] + [ A ] , [ q_mean , q_logvar ] ) <EOL> eps = rng . normal ( size = q_mean . shape , dtype = '<STR_LIT>' ) <EOL> _z = q_mean + T . exp ( <NUM_LIT:0.5> * q_logvar ) * eps <EOL> hidden_p = [ nonlinear_p ( T . dot ( w [ '<STR_LIT>' ] , x [ '<STR_LIT:y>' ] ) + T . dot ( w [ '<STR_LIT>' ] , _z ) + T . dot ( w [ '<STR_LIT>' ] , A ) ) ] <EOL> for i in range ( <NUM_LIT:1> , len ( self . n_hidden_p ) ) : <EOL> hidden_p . append ( nonlinear_p ( T . dot ( w [ '<STR_LIT:w>' + str ( i ) ] , hidden_p [ - <NUM_LIT:1> ] ) + T . dot ( w [ '<STR_LIT:b>' + str ( i ) ] , A ) ) ) <EOL> if self . dropout : <EOL> hidden_p [ - <NUM_LIT:1> ] *= <NUM_LIT> * ( rng . uniform ( size = hidden_p [ - <NUM_LIT:1> ] . shape , dtype = '<STR_LIT>' ) > <NUM_LIT> ) <EOL> if self . type_px == '<STR_LIT>' : <EOL> p = T . nnet . sigmoid ( T . dot ( w [ '<STR_LIT>' ] , hidden_p [ - <NUM_LIT:1> ] ) + T . dot ( w [ '<STR_LIT>' ] , A ) ) <EOL> _logpx = - T . nnet . binary_crossentropy ( p , x [ '<STR_LIT:x>' ] ) <EOL> self . dist_px [ '<STR_LIT:x>' ] = theanofunc ( [ x [ '<STR_LIT:y>' ] , _z ] + [ A ] , p ) <EOL> elif self . type_px == '<STR_LIT>' : <EOL> x_mean = T . dot ( w [ '<STR_LIT>' ] , hidden_p [ - <NUM_LIT:1> ] ) + T . dot ( w [ '<STR_LIT>' ] , A ) <EOL> x_logvar = T . dot ( w [ '<STR_LIT>' ] , hidden_p [ - <NUM_LIT:1> ] ) + T . dot ( w [ '<STR_LIT>' ] , A ) <EOL> _logpx = ap . logpdfs . normal2 ( x [ '<STR_LIT:x>' ] , x_mean , x_logvar ) <EOL> self . dist_px [ '<STR_LIT:x>' ] = theanofunc ( [ x [ '<STR_LIT:y>' ] , _z ] + [ A ] , [ x_mean , x_logvar ] ) <EOL> elif self . type_px == '<STR_LIT>' : <EOL> x_mean = T . dot ( w [ '<STR_LIT>' ] , hidden_p [ - <NUM_LIT:1> ] ) + T . dot ( w [ '<STR_LIT>' ] , A ) <EOL> x_logvar = T . dot ( w [ '<STR_LIT>' ] , hidden_p [ - <NUM_LIT:1> ] ) + T . dot ( w [ '<STR_LIT>' ] , A ) <EOL> _logpx = ap . logpdfs . laplace ( x [ '<STR_LIT:x>' ] , x_mean , x_logvar ) <EOL> self . dist_px [ '<STR_LIT:x>' ] = theanofunc ( [ x [ '<STR_LIT:y>' ] , _z ] + [ A ] , [ x_mean , x_logvar ] ) <EOL> else : raise Exception ( "<STR_LIT>" ) <EOL> logpx = T . dot ( shared32 ( np . ones ( ( <NUM_LIT:1> , self . n_x ) ) ) , _logpx ) <EOL> if self . type_pz == '<STR_LIT>' : <EOL> logpz = - <NUM_LIT:0.5> * ( np . log ( <NUM_LIT:2> * np . pi ) + ( q_mean ** <NUM_LIT:2> + T . exp ( q_logvar ) ) ) . sum ( axis = <NUM_LIT:0> , keepdims = True ) <EOL> elif self . type_pz == '<STR_LIT>' : <EOL> logpz = ap . logpdfs . standard_normal ( _z ) . sum ( axis = <NUM_LIT:0> , keepdims = True ) <EOL> elif self . type_pz == '<STR_LIT>' : <EOL> pz = <NUM_LIT:0> <EOL> for i in range ( self . n_mixture ) : <EOL> pz += T . exp ( ap . logpdfs . normal2 ( _z , T . dot ( w [ '<STR_LIT>' + str ( i ) ] , A ) , T . dot ( w [ '<STR_LIT>' + str ( i ) ] , A ) ) ) <EOL> logpz = T . log ( pz ) . sum ( axis = <NUM_LIT:0> , keepdims = True ) - self . n_z * np . log ( float ( self . n_mixture ) ) <EOL> elif self . type_pz == '<STR_LIT>' : <EOL> logpz = ap . logpdfs . standard_laplace ( _z ) . sum ( axis = <NUM_LIT:0> , keepdims = True ) <EOL> elif self . type_pz == '<STR_LIT>' : <EOL> logpz = ap . logpdfs . studentt ( _z , T . dot ( T . exp ( w [ '<STR_LIT>' ] ) , A ) ) . sum ( axis = <NUM_LIT:0> , keepdims = True ) <EOL> else : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> if self . type_qz == '<STR_LIT>' : <EOL> logqz = - <NUM_LIT:0.5> * ( np . log ( <NUM_LIT:2> * np . pi ) + <NUM_LIT:1> + q_logvar ) . sum ( axis = <NUM_LIT:0> , keepdims = True ) <EOL> elif self . type_qz == '<STR_LIT>' : <EOL> logqz = ap . logpdfs . normal2 ( _z , q_mean , q_logvar ) . sum ( axis = <NUM_LIT:0> , keepdims = True ) <EOL> else : raise Exception ( ) <EOL> def f_prior ( _w , prior_sd = self . prior_sd ) : <EOL> return ap . logpdfs . normal ( _w , <NUM_LIT:0> , prior_sd ) . sum ( ) <EOL> logpv = <NUM_LIT:0> <EOL> logpv += f_prior ( v [ '<STR_LIT>' ] ) <EOL> logpv += f_prior ( v [ '<STR_LIT>' ] ) <EOL> for i in range ( <NUM_LIT:1> , len ( self . n_hidden_q ) ) : <EOL> logpv += f_prior ( v [ '<STR_LIT:w>' + str ( i ) ] ) <EOL> logpv += f_prior ( v [ '<STR_LIT>' ] ) <EOL> if self . type_qz in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> logpv += f_prior ( v [ '<STR_LIT>' ] ) <EOL> logpw = <NUM_LIT:0> <EOL> logpw += f_prior ( w [ '<STR_LIT>' ] ) <EOL> logpw += f_prior ( w [ '<STR_LIT>' ] ) <EOL> for i in range ( <NUM_LIT:1> , len ( self . n_hidden_p ) ) : <EOL> logpw += f_prior ( w [ '<STR_LIT:w>' + str ( i ) ] ) <EOL> logpw += f_prior ( w [ '<STR_LIT>' ] ) <EOL> if self . type_px in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> logpw += f_prior ( w [ '<STR_LIT>' ] ) <EOL> if self . type_pz == '<STR_LIT>' : <EOL> logpw += f_prior ( w [ '<STR_LIT>' ] ) <EOL> return logpx , logpz , logqz <EOL> def gen_eps ( self , n_batch ) : <EOL> z = { '<STR_LIT>' : np . random . standard_normal ( size = ( self . n_z , n_batch ) ) . astype ( '<STR_LIT>' ) } <EOL> return z <EOL> def gen_xz ( self , x , z , n_batch ) : <EOL> x , z = ndict . ordereddicts ( ( x , z ) ) <EOL> A = np . ones ( ( <NUM_LIT:1> , n_batch ) ) . astype ( np . float32 ) <EOL> for i in z : z [ i ] = z [ i ] . astype ( np . float32 ) <EOL> for i in x : x [ i ] = x [ i ] . astype ( np . float32 ) <EOL> _z = { } <EOL> if x . has_key ( '<STR_LIT:x>' ) and x . has_key ( '<STR_LIT:y>' ) and not z . has_key ( '<STR_LIT:z>' ) : <EOL> q_mean , q_logvar = self . dist_qz [ '<STR_LIT:z>' ] ( * ( [ x [ '<STR_LIT:x>' ] , x [ '<STR_LIT:y>' ] ] + [ A ] ) ) <EOL> _z [ '<STR_LIT>' ] = q_mean <EOL> _z [ '<STR_LIT>' ] = q_logvar <EOL> if not z . has_key ( '<STR_LIT>' ) : <EOL> eps = self . gen_eps ( n_batch ) [ '<STR_LIT>' ] <EOL> z [ '<STR_LIT:z>' ] = q_mean + np . exp ( <NUM_LIT:0.5> * q_logvar ) * eps <EOL> else : <EOL> if not z . has_key ( '<STR_LIT:z>' ) : <EOL> if self . type_pz in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> z [ '<STR_LIT:z>' ] = np . random . standard_normal ( size = ( self . n_z , n_batch ) ) . astype ( np . float32 ) <EOL> elif self . type_pz == '<STR_LIT>' : <EOL> z [ '<STR_LIT:z>' ] = np . random . laplace ( size = ( self . n_z , n_batch ) ) . astype ( np . float32 ) <EOL> elif self . type_pz == '<STR_LIT>' : <EOL> z [ '<STR_LIT:z>' ] = np . random . standard_t ( np . dot ( np . exp ( self . w [ '<STR_LIT>' ] . get_value ( ) ) , A ) ) . astype ( np . float32 ) <EOL> elif self . type_pz == '<STR_LIT>' : <EOL> i = np . random . randint ( self . n_mixture ) <EOL> loc = np . dot ( self . w [ '<STR_LIT>' + str ( i ) ] . get_value ( ) , A ) <EOL> scale = np . dot ( np . exp ( <NUM_LIT> * self . w [ '<STR_LIT>' + str ( i ) ] . get_value ( ) ) , A ) <EOL> z [ '<STR_LIT:z>' ] = np . random . normal ( loc = loc , scale = scale ) . astype ( np . float32 ) <EOL> else : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> if not x . has_key ( '<STR_LIT:y>' ) : <EOL> py = self . dist_px [ '<STR_LIT:y>' ] ( * ( [ A ] ) ) <EOL> _z [ '<STR_LIT:y>' ] = py <EOL> x [ '<STR_LIT:y>' ] = np . zeros ( py . shape ) <EOL> for i in range ( py . shape [ <NUM_LIT:1> ] ) : <EOL> x [ '<STR_LIT:y>' ] [ : , i ] = np . random . multinomial ( n = <NUM_LIT:1> , pvals = py [ : , i ] ) <EOL> if self . type_px == '<STR_LIT>' : <EOL> p = self . dist_px [ '<STR_LIT:x>' ] ( * ( [ x [ '<STR_LIT:y>' ] , z [ '<STR_LIT:z>' ] ] + [ A ] ) ) <EOL> _z [ '<STR_LIT:x>' ] = p <EOL> if not x . has_key ( '<STR_LIT:x>' ) : <EOL> x [ '<STR_LIT:x>' ] = np . random . binomial ( n = <NUM_LIT:1> , p = p ) <EOL> elif self . type_px == '<STR_LIT>' : <EOL> x_mean , x_logvar = self . dist_px [ '<STR_LIT:x>' ] ( * ( [ x [ '<STR_LIT:y>' ] , z [ '<STR_LIT:z>' ] ] + [ A ] ) ) <EOL> _z [ '<STR_LIT:x>' ] = x_mean <EOL> if not x . has_key ( '<STR_LIT:x>' ) : <EOL> x [ '<STR_LIT:x>' ] = np . random . laplace ( x_mean , np . exp ( <NUM_LIT:0.5> * x_logvar ) ) <EOL> elif self . type_px == '<STR_LIT>' or self . type_px == '<STR_LIT>' : <EOL> x_mean , x_logvar = self . dist_px [ '<STR_LIT:x>' ] ( * ( [ x [ '<STR_LIT:y>' ] , z [ '<STR_LIT:z>' ] ] + [ A ] ) ) <EOL> _z [ '<STR_LIT:x>' ] = x_mean <EOL> if not x . has_key ( '<STR_LIT:x>' ) : <EOL> x [ '<STR_LIT:x>' ] = np . random . normal ( x_mean , np . exp ( x_logvar / <NUM_LIT:2> ) ) <EOL> if self . type_px == '<STR_LIT>' : <EOL> x [ '<STR_LIT:x>' ] = np . maximum ( np . zeros ( x [ '<STR_LIT:x>' ] . shape ) , x [ '<STR_LIT:x>' ] ) <EOL> x [ '<STR_LIT:x>' ] = np . minimum ( np . ones ( x [ '<STR_LIT:x>' ] . shape ) , x [ '<STR_LIT:x>' ] ) <EOL> else : raise Exception ( "<STR_LIT>" ) <EOL> return x , z , _z <EOL> def variables ( self ) : <EOL> z = { } <EOL> x = { '<STR_LIT:x>' : T . fmatrix ( '<STR_LIT:x>' ) , '<STR_LIT:y>' : T . fmatrix ( '<STR_LIT:y>' ) } <EOL> return x , z <EOL> def init_w ( self , std = <NUM_LIT> ) : <EOL> def rand ( size ) : <EOL> if len ( size ) == <NUM_LIT:2> and size [ <NUM_LIT:1> ] > <NUM_LIT:1> : <EOL> return np . random . normal ( <NUM_LIT:0> , <NUM_LIT:1> , size = size ) / np . sqrt ( size [ <NUM_LIT:1> ] ) <EOL> return np . random . normal ( <NUM_LIT:0> , std , size = size ) <EOL> v = { } <EOL> v [ '<STR_LIT>' ] = rand ( ( self . n_hidden_q [ <NUM_LIT:0> ] , self . n_x ) ) <EOL> v [ '<STR_LIT>' ] = rand ( ( self . n_hidden_q [ <NUM_LIT:0> ] , self . n_y ) ) <EOL> v [ '<STR_LIT>' ] = rand ( ( self . n_hidden_q [ <NUM_LIT:0> ] , <NUM_LIT:1> ) ) <EOL> for i in range ( <NUM_LIT:1> , len ( self . n_hidden_q ) ) : <EOL> v [ '<STR_LIT:w>' + str ( i ) ] = rand ( ( self . n_hidden_q [ i ] , self . n_hidden_q [ i - <NUM_LIT:1> ] ) ) <EOL> v [ '<STR_LIT:b>' + str ( i ) ] = rand ( ( self . n_hidden_q [ i ] , <NUM_LIT:1> ) ) <EOL> v [ '<STR_LIT>' ] = rand ( ( self . n_z , self . n_hidden_q [ - <NUM_LIT:1> ] ) ) <EOL> v [ '<STR_LIT>' ] = rand ( ( self . n_z , <NUM_LIT:1> ) ) <EOL> if self . type_qz in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> v [ '<STR_LIT>' ] = np . zeros ( ( self . n_z , self . n_hidden_q [ - <NUM_LIT:1> ] ) ) <EOL> v [ '<STR_LIT>' ] = np . zeros ( ( self . n_z , <NUM_LIT:1> ) ) <EOL> w = { } <EOL> if len ( self . n_hidden_p ) > <NUM_LIT:0> : <EOL> w [ '<STR_LIT>' ] = rand ( ( self . n_hidden_p [ <NUM_LIT:0> ] , self . n_y ) ) <EOL> w [ '<STR_LIT>' ] = rand ( ( self . n_hidden_p [ <NUM_LIT:0> ] , self . n_z ) ) <EOL> w [ '<STR_LIT>' ] = rand ( ( self . n_hidden_p [ <NUM_LIT:0> ] , <NUM_LIT:1> ) ) <EOL> for i in range ( <NUM_LIT:1> , len ( self . n_hidden_p ) ) : <EOL> w [ '<STR_LIT:w>' + str ( i ) ] = rand ( ( self . n_hidden_p [ i ] , self . n_hidden_p [ i - <NUM_LIT:1> ] ) ) <EOL> w [ '<STR_LIT:b>' + str ( i ) ] = rand ( ( self . n_hidden_p [ i ] , <NUM_LIT:1> ) ) <EOL> w [ '<STR_LIT>' ] = rand ( ( self . n_x , self . n_hidden_p [ - <NUM_LIT:1> ] ) ) <EOL> w [ '<STR_LIT>' ] = np . zeros ( ( self . n_x , <NUM_LIT:1> ) ) <EOL> if self . type_px in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> w [ '<STR_LIT>' ] = rand ( ( self . n_x , self . n_hidden_p [ - <NUM_LIT:1> ] ) ) <EOL> w [ '<STR_LIT>' ] = np . zeros ( ( self . n_x , <NUM_LIT:1> ) ) <EOL> if self . type_px == '<STR_LIT>' : <EOL> w [ '<STR_LIT>' ] = np . zeros ( ( self . n_x , <NUM_LIT:1> ) ) <EOL> w [ '<STR_LIT>' ] = np . zeros ( ( self . n_x , <NUM_LIT:1> ) ) <EOL> else : <EOL> w [ '<STR_LIT>' ] = rand ( ( self . n_x , self . n_z ) ) <EOL> w [ '<STR_LIT>' ] = np . zeros ( ( self . n_x , <NUM_LIT:1> ) ) <EOL> if self . type_px == '<STR_LIT>' : <EOL> w [ '<STR_LIT>' ] = rand ( ( self . n_x , self . n_z ) ) <EOL> w [ '<STR_LIT>' ] = np . zeros ( ( self . n_x , <NUM_LIT:1> ) ) <EOL> if self . type_px == '<STR_LIT>' : <EOL> w [ '<STR_LIT>' ] = np . zeros ( ( self . n_x , <NUM_LIT:1> ) ) <EOL> w [ '<STR_LIT>' ] = np . zeros ( ( self . n_x , <NUM_LIT:1> ) ) <EOL> if self . type_pz == '<STR_LIT>' : <EOL> w [ '<STR_LIT>' ] = np . zeros ( ( self . n_z , <NUM_LIT:1> ) ) <EOL> return v , w </s>
