<s> __author__ = ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> from scipy import random , outer , zeros , ones <EOL> from pybrain . datasets import SupervisedDataSet , UnsupervisedDataSet <EOL> from pybrain . supervised . trainers import Trainer <EOL> from pybrain . utilities import abstractMethod <EOL> class RbmGibbsTrainerConfig : <EOL> def __init__ ( self ) : <EOL> self . batchSize = <NUM_LIT:10> <EOL> self . rWeights = <NUM_LIT:0.1> <EOL> self . rHidBias = <NUM_LIT:0.1> <EOL> self . rVisBias = <NUM_LIT:0.1> <EOL> self . weightCost = <NUM_LIT> <EOL> self . iniMm = <NUM_LIT:0.5> <EOL> self . finMm = <NUM_LIT> <EOL> self . mmSwitchIter = <NUM_LIT:5> <EOL> self . maxIter = <NUM_LIT:9> <EOL> self . visibleDistribution = '<STR_LIT>' <EOL> class RbmGibbsTrainer ( Trainer ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , rbm , dataset , cfg = None ) : <EOL> self . rbm = rbm <EOL> self . invRbm = rbm . invert ( ) <EOL> self . dataset = dataset <EOL> self . cfg = RbmGibbsTrainerConfig ( ) if cfg is None else cfg <EOL> if isinstance ( self . dataset , SupervisedDataSet ) : <EOL> self . datasetField = '<STR_LIT:input>' <EOL> elif isinstance ( self . dataset , UnsupervisedDataSet ) : <EOL> self . datasetField = '<STR_LIT>' <EOL> def train ( self ) : <EOL> self . trainOnDataset ( self . dataset ) <EOL> def trainOnDataset ( self , dataset ) : <EOL> """<STR_LIT>""" <EOL> cfg = self . cfg <EOL> for rows in dataset . randomBatches ( self . datasetField , cfg . batchSize ) : <EOL> olduw , olduhb , olduvb = zeros ( ( self . rbm . visibleDim , self . rbm . hiddenDim ) ) , zeros ( self . rbm . hiddenDim ) , zeros ( self . rbm . visibleDim ) <EOL> for t in range ( cfg . maxIter ) : <EOL> params = self . rbm . params <EOL> params = params . reshape ( ( self . rbm . visibleDim , self . rbm . hiddenDim ) ) <EOL> biasParams = self . rbm . biasParams <EOL> mm = cfg . iniMm if t < cfg . mmSwitchIter else cfg . finMm <EOL> w , hb , vb = self . calcUpdateByRows ( rows ) <EOL> olduw = uw = olduw * mm + cfg . rWeights * ( w - cfg . weightCost * params ) <EOL> olduhb = uhb = olduhb * mm + cfg . rHidBias * hb <EOL> olduvb = uvb = olduvb * mm + cfg . rVisBias * vb <EOL> params += uw <EOL> biasParams += uhb <EOL> invBiasParams = self . invRbm . biasParams <EOL> invBiasParams += uvb <EOL> self . invRbm = self . rbm . invert ( ) <EOL> self . invRbm . biasParams [ : ] = invBiasParams <EOL> def calcUpdateByRow ( self , row ) : <EOL> """<STR_LIT>""" <EOL> poshp = self . rbm . activate ( row ) <EOL> pos = outer ( row , poshp ) <EOL> poshb = poshp <EOL> posvb = row <EOL> sampled = self . sampler ( poshp ) <EOL> recon = self . invRbm . activate ( sampled ) <EOL> neghp = self . rbm . activate ( recon ) <EOL> neg = outer ( recon , neghp ) <EOL> neghb = neghp <EOL> negvb = recon <EOL> return self . updater ( pos , neg , poshb , neghb , posvb , negvb ) <EOL> def sampler ( self , probabilities ) : <EOL> abstractMethod ( ) <EOL> def updater ( self , pos , neg , poshb , neghb , posvb , negvb ) : <EOL> abstractMethod ( ) <EOL> def calcUpdateByRows ( self , rows ) : <EOL> """<STR_LIT>""" <EOL> delta_w , delta_hb , delta_vb = zeros ( ( self . rbm . visibleDim , self . rbm . hiddenDim ) ) , zeros ( self . rbm . hiddenDim ) , zeros ( self . rbm . visibleDim ) <EOL> for row in rows : <EOL> dw , dhb , dvb = self . calcUpdateByRow ( row ) <EOL> delta_w += dw <EOL> delta_hb += dhb <EOL> delta_vb += dvb <EOL> delta_w /= len ( rows ) <EOL> delta_hb /= len ( rows ) <EOL> delta_vb /= len ( rows ) <EOL> return delta_w , delta_hb , delta_vb <EOL> class RbmBernoulliTrainer ( RbmGibbsTrainer ) : <EOL> def sampler ( self , probabilities ) : <EOL> result = probabilities > random . rand ( self . rbm . hiddenDim ) <EOL> return result . astype ( '<STR_LIT>' ) <EOL> def updater ( self , pos , neg , poshb , neghb , posvb , negvb ) : <EOL> return pos - neg , poshb - neghb , posvb - negvb <EOL> class RbmGaussTrainer ( RbmGibbsTrainer ) : <EOL> def __init__ ( self , rbm , dataset , cfg = None ) : <EOL> super ( RbmGaussTrainer , self ) . __init__ ( rbm , dataset , cfg ) <EOL> self . visibleVariances = ones ( rbm . net . outdim ) <EOL> def sampler ( self , probabilities ) : <EOL> return random . normal ( probabilities , self . visibleVariances ) <EOL> def updater ( self , pos , neg , poshb , neghb , posvb , negvb ) : <EOL> pos = pos / self . visibleVariances <EOL> return pos - neg , poshb - neghb , posvb - negvb </s>
