<s> '''<STR_LIT>''' <EOL> import numpy as np <EOL> import scipy . linalg <EOL> import os <EOL> import copy <EOL> from ObsModel import ObsModel <EOL> from bnpy . distr import ZMGaussDistr <EOL> from bnpy . distr import WishartDistr <EOL> from bnpy . util import np2flatstr , dotATA , dotATB , dotABT <EOL> from bnpy . util import LOGPI , LOGTWOPI , LOGTWO , EPS <EOL> class ZMGaussObsModel ( ObsModel ) : <EOL> def __init__ ( self , inferType , D = None , obsPrior = None , min_covar = None ) : <EOL> self . inferType = inferType <EOL> self . D = D <EOL> self . obsPrior = obsPrior <EOL> self . comp = list ( ) <EOL> self . K = <NUM_LIT:0> <EOL> if min_covar is not None : <EOL> self . min_covar = min_covar <EOL> @ classmethod <EOL> def CreateWithPrior ( cls , inferType , priorArgDict , Data ) : <EOL> D = Data . dim <EOL> if inferType == '<STR_LIT>' : <EOL> obsPrior = None <EOL> return cls ( inferType , D , obsPrior , min_covar = priorArgDict [ '<STR_LIT>' ] ) <EOL> else : <EOL> obsPrior = WishartDistr . CreateAsPrior ( priorArgDict , Data ) <EOL> return cls ( inferType , D , obsPrior ) <EOL> @ classmethod <EOL> def CreateWithAllComps ( cls , oDict , obsPrior , compDictList ) : <EOL> if '<STR_LIT>' in oDict : <EOL> mc = oDict [ '<STR_LIT>' ] <EOL> self = cls ( oDict [ '<STR_LIT>' ] , obsPrior = obsPrior , min_covar = mc ) <EOL> else : <EOL> self = cls ( oDict [ '<STR_LIT>' ] , obsPrior = obsPrior ) <EOL> self . K = len ( compDictList ) <EOL> self . comp = [ None for k in range ( self . K ) ] <EOL> if self . inferType == '<STR_LIT>' : <EOL> for k in xrange ( self . K ) : <EOL> self . comp [ k ] = ZMGaussDistr ( ** compDictList [ k ] ) <EOL> self . D = self . comp [ k ] . D <EOL> elif self . inferType . count ( '<STR_LIT>' ) > <NUM_LIT:0> : <EOL> for k in xrange ( self . K ) : <EOL> self . comp [ k ] = WishartDistr ( ** compDictList [ k ] ) <EOL> self . D = self . comp [ k ] . D <EOL> return self <EOL> def get_mean_for_comp ( self , k ) : <EOL> return np . zeros ( self . D ) <EOL> def get_covar_mat_for_comp ( self , k ) : <EOL> return self . comp [ k ] . ECovMat ( ) <EOL> '''<STR_LIT>''' <EOL> def get_global_suff_stats ( self , Data , SS , LP , ** kwargs ) : <EOL> '''<STR_LIT>''' <EOL> sqrtResp = np . sqrt ( LP [ '<STR_LIT>' ] ) <EOL> K = sqrtResp . shape [ <NUM_LIT:1> ] <EOL> xxT = np . zeros ( ( K , self . D , self . D ) ) <EOL> for k in xrange ( K ) : <EOL> xxT [ k ] = dotATA ( sqrtResp [ : , k ] [ : , np . newaxis ] * Data . X ) <EOL> SS . setField ( '<STR_LIT>' , xxT , dims = ( '<STR_LIT>' , '<STR_LIT:D>' , '<STR_LIT:D>' ) ) <EOL> return SS <EOL> def update_obs_params_EM ( self , SS , ** kwargs ) : <EOL> for k in xrange ( self . K ) : <EOL> covMat = SS . xxT [ k ] / SS . N [ k ] <EOL> covMat += self . min_covar * np . eye ( self . D ) <EOL> self . comp [ k ] = ZMGaussDistr ( Sigma = covMat ) <EOL> def update_obs_params_VB ( self , SS , mergeCompA = None , ** kwargs ) : <EOL> if mergeCompA is None : <EOL> for k in xrange ( self . K ) : <EOL> self . comp [ k ] = self . obsPrior . get_post_distr ( SS , k ) <EOL> else : <EOL> self . comp [ mergeCompA ] = self . obsPrior . get_post_distr ( SS , mergeCompA ) <EOL> def update_obs_params_soVB ( self , SS , rho , ** kwargs ) : <EOL> for k in xrange ( self . K ) : <EOL> Dstar = self . obsPrior . get_post_distr ( SS , k ) <EOL> self . comp [ k ] . post_update_soVB ( rho , Dstar ) <EOL> def set_global_params ( self , hmodel = None , Sigma = None , ** kwargs ) : <EOL> '''<STR_LIT>''' <EOL> if hmodel is not None : <EOL> self . comp = [ copy . deepcopy ( c ) for c in hmodel . obsModel . comp ] <EOL> self . K = hmodel . obsModel . K <EOL> if Sigma is not None : <EOL> self . K = Sigma . shape [ <NUM_LIT:0> ] <EOL> self . comp = [ None for k in range ( self . K ) ] <EOL> for k in range ( self . K ) : <EOL> self . comp [ k ] = ZMGaussDistr ( Sigma = Sigma [ k ] ) <EOL> def calc_evidence ( self , Data , SS , LP = None ) : <EOL> if self . inferType == '<STR_LIT>' : <EOL> return <NUM_LIT:0> <EOL> else : <EOL> return self . E_logpX ( SS ) + self . E_logpPhi ( ) - self . E_logqPhi ( ) <EOL> def E_logpX ( self , SS ) : <EOL> lpX = np . zeros ( self . K ) <EOL> for k in range ( self . K ) : <EOL> if SS . N [ k ] == <NUM_LIT:0> : <EOL> continue <EOL> lpX [ k ] = SS . N [ k ] * self . comp [ k ] . ElogdetLam ( ) - self . comp [ k ] . E_traceLam ( SS . xxT [ k ] ) <EOL> return <NUM_LIT:0.5> * np . sum ( lpX ) - <NUM_LIT:0.5> * np . sum ( SS . N ) * self . D * LOGTWOPI <EOL> def E_logpPhi ( self ) : <EOL> return self . E_logpLam ( ) <EOL> def E_logqPhi ( self ) : <EOL> return self . E_logqLam ( ) <EOL> def E_logpLam ( self ) : <EOL> lp = np . zeros ( self . K ) <EOL> for k in xrange ( self . K ) : <EOL> lp [ k ] = <NUM_LIT:0.5> * ( self . obsPrior . v - self . D - <NUM_LIT:1> ) * self . comp [ k ] . ElogdetLam ( ) <EOL> lp [ k ] -= <NUM_LIT:0.5> * self . comp [ k ] . E_traceLam ( cholS = self . obsPrior . cholinvW ( ) ) <EOL> return lp . sum ( ) - self . K * self . obsPrior . get_log_norm_const ( ) <EOL> def E_logqLam ( self ) : <EOL> lp = np . zeros ( self . K ) <EOL> for k in xrange ( self . K ) : <EOL> lp [ k ] = - <NUM_LIT:1> * self . comp [ k ] . get_entropy ( ) <EOL> return lp . sum ( ) <EOL> def get_name ( self ) : <EOL> return '<STR_LIT>' <EOL> def get_info_string ( self ) : <EOL> return '<STR_LIT>' <EOL> def get_info_string_prior ( self ) : <EOL> if self . obsPrior is None : <EOL> return '<STR_LIT:None>' <EOL> else : <EOL> return '<STR_LIT>' + self . obsPrior . to_string ( ) <EOL> def get_prior_dict ( self ) : <EOL> if self . obsPrior is None : <EOL> PDict = dict ( min_covar = self . min_covar , name = "<STR_LIT>" ) <EOL> else : <EOL> PDict = self . obsPrior . to_dict ( ) <EOL> return PDict </s>
