<s> from __future__ import print_function <EOL> __author__ = "<STR_LIT>" <EOL> try : <EOL> from svm import svm_model , svm_parameter , svm_problem , cross_validation <EOL> from svm import C_SVC , NU_SVC , ONE_CLASS , EPSILON_SVR , NU_SVR <EOL> from svm import LINEAR , POLY , RBF , SIGMOID , PRECOMPUTED <EOL> except ImportError : <EOL> raise ImportError ( "<STR_LIT>" ) <EOL> from numpy import * <EOL> import logging <EOL> class SVMTrainer ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , svmunit , dataset , modelfile = None , plot = False ) : <EOL> """<STR_LIT>""" <EOL> self . svm = svmunit <EOL> self . ds = dataset <EOL> self . svmtarget = dataset [ '<STR_LIT:target>' ] . flatten ( ) <EOL> self . plot = plot <EOL> self . searchlog = '<STR_LIT>' <EOL> self . params = { <EOL> '<STR_LIT>' : RBF <EOL> } <EOL> if modelfile is not None : self . load ( modelfile ) <EOL> def train ( self , search = False , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> self . setParams ( ** kwargs ) <EOL> problem = svm_problem ( self . ds [ '<STR_LIT:target>' ] . flatten ( ) , self . ds [ '<STR_LIT:input>' ] . tolist ( ) ) <EOL> if search : <EOL> model = eval ( search + "<STR_LIT>" ) <EOL> else : <EOL> param = svm_parameter ( ** self . params ) <EOL> model = svm_model ( problem , param ) <EOL> logging . info ( "<STR_LIT>" ) <EOL> logging . info ( repr ( param ) ) <EOL> self . svm . setModel ( model ) <EOL> def save ( self , filename ) : <EOL> """<STR_LIT>""" <EOL> self . svm . saveModel ( filename ) <EOL> def load ( self , filename ) : <EOL> """<STR_LIT>""" <EOL> self . svm . loadModel ( filename ) <EOL> def setParams ( self , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if '<STR_LIT>' in kwargs : <EOL> self . params [ '<STR_LIT>' ] = len ( kwargs [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT>' in kwargs : <EOL> self . params [ '<STR_LIT:C>' ] = <NUM_LIT:2> ** kwargs [ '<STR_LIT>' ] <EOL> kwargs . pop ( '<STR_LIT>' ) <EOL> if '<STR_LIT>' in kwargs : <EOL> self . params [ '<STR_LIT>' ] = <NUM_LIT:2> ** kwargs [ '<STR_LIT>' ] <EOL> kwargs . pop ( '<STR_LIT>' ) <EOL> if '<STR_LIT>' in kwargs : <EOL> self . searchlog = kwargs [ '<STR_LIT>' ] <EOL> kwargs . pop ( '<STR_LIT>' ) <EOL> self . params . update ( kwargs ) <EOL> class GridSearch ( svm_model ) : <EOL> """<STR_LIT>""" <EOL> allPts = [ ] <EOL> allScores = [ ] <EOL> def __init__ ( self , problem , targets , cmin , cmax , cstep = None , crossval = <NUM_LIT:5> , <EOL> plotflag = False , maxdepth = <NUM_LIT:8> , searchlog = '<STR_LIT>' , ** params ) : <EOL> """<STR_LIT>""" <EOL> self . nPars = len ( cmin ) <EOL> self . usermin = cmin <EOL> self . usermax = cmax <EOL> self . userstep = cstep <EOL> self . crossval = crossval <EOL> self . plotflag = plotflag <EOL> self . maxdepth = maxdepth <EOL> self . params = params <EOL> if self . plotflag : <EOL> import pylab as p <EOL> p . ion ( ) <EOL> p . figure ( figsize = [ <NUM_LIT:12> , <NUM_LIT:8> ] ) <EOL> assert isinstance ( problem , svm_problem ) <EOL> self . problem = problem <EOL> self . targets = targets <EOL> self . resfile = open ( searchlog , '<STR_LIT:w>' ) <EOL> param = self . search ( ) <EOL> if self . plotflag : <EOL> p . ioff ( ) <EOL> p . show ( ) <EOL> self . resfile . close ( ) <EOL> svm_model . __init__ ( self , problem , param ) <EOL> def setParams ( self , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if '<STR_LIT>' in kwargs : <EOL> self . params [ '<STR_LIT>' ] = len ( kwargs [ '<STR_LIT>' ] ) <EOL> self . params . update ( kwargs ) <EOL> def search ( self ) : <EOL> """<STR_LIT>""" <EOL> jobs = self . calculate_jobs ( ) <EOL> scores = [ ] <EOL> for line in jobs : <EOL> for ( c , g ) in line : <EOL> self . setParams ( C = <NUM_LIT:2> ** c , gamma = <NUM_LIT:2> ** g ) <EOL> param = svm_parameter ( ** self . params ) <EOL> cvresult = array ( cross_validation ( self . problem , param , self . crossval ) ) <EOL> corr , = where ( cvresult == self . targets ) <EOL> res = ( c , g , float ( corr . size ) / self . targets . size ) <EOL> scores . append ( res ) <EOL> self . _save_points ( res ) <EOL> self . _redraw ( scores ) <EOL> scores = array ( scores ) <EOL> best = scores [ scores [ : , <NUM_LIT:0> ] . argmax ( ) , <NUM_LIT:1> : ] <EOL> self . setParams ( C = <NUM_LIT:2> ** best [ <NUM_LIT:0> ] , gamma = <NUM_LIT:2> ** best [ <NUM_LIT:1> ] ) <EOL> logging . info ( "<STR_LIT>" % ( best [ <NUM_LIT:0> ] , best [ <NUM_LIT:1> ] ) ) <EOL> param = svm_parameter ( ** self . params ) <EOL> return param <EOL> def _permute_sequence ( self , seq ) : <EOL> """<STR_LIT>""" <EOL> n = len ( seq ) <EOL> if n <= <NUM_LIT:1> : return seq <EOL> mid = int ( n / <NUM_LIT:2> ) <EOL> left = self . _permute_sequence ( seq [ : mid ] ) <EOL> right = self . _permute_sequence ( seq [ mid + <NUM_LIT:1> : ] ) <EOL> ret = [ seq [ mid ] ] <EOL> while left or right : <EOL> if left : ret . append ( left . pop ( <NUM_LIT:0> ) ) <EOL> if right : ret . append ( right . pop ( <NUM_LIT:0> ) ) <EOL> return ret <EOL> def _range_f ( self , begin , end , step ) : <EOL> """<STR_LIT>""" <EOL> seq = [ ] <EOL> while <NUM_LIT:1> : <EOL> if step > <NUM_LIT:0> and begin > end : break <EOL> if step < <NUM_LIT:0> and begin < end : break <EOL> seq . append ( begin ) <EOL> begin = begin + step <EOL> return seq <EOL> def calculate_jobs ( self ) : <EOL> """<STR_LIT>""" <EOL> c_seq = self . _permute_sequence ( self . _range_f ( self . usermin [ <NUM_LIT:0> ] , self . usermax [ <NUM_LIT:0> ] , self . userstep [ <NUM_LIT:0> ] ) ) <EOL> g_seq = self . _permute_sequence ( self . _range_f ( self . usermin [ <NUM_LIT:1> ] , self . usermax [ <NUM_LIT:1> ] , self . userstep [ <NUM_LIT:1> ] ) ) <EOL> nr_c = float ( len ( c_seq ) ) <EOL> nr_g = float ( len ( g_seq ) ) <EOL> global total_points <EOL> total_points = ( nr_g + <NUM_LIT:1> ) * ( nr_g ) <EOL> i = <NUM_LIT:0> <EOL> j = <NUM_LIT:0> <EOL> jobs = [ ] <EOL> while i < nr_c or j < nr_g : <EOL> if i / nr_c < j / nr_g : <EOL> line = [ ] <EOL> for k in range ( <NUM_LIT:0> , j ) : <EOL> line . append ( ( c_seq [ i ] , g_seq [ k ] ) ) <EOL> i = i + <NUM_LIT:1> <EOL> jobs . append ( line ) <EOL> else : <EOL> line = [ ] <EOL> for k in range ( <NUM_LIT:0> , i ) : <EOL> line . append ( ( c_seq [ k ] , g_seq [ j ] ) ) <EOL> j = j + <NUM_LIT:1> <EOL> jobs . append ( line ) <EOL> return jobs <EOL> def _save_points ( self , res ) : <EOL> """<STR_LIT>""" <EOL> self . resfile . write ( "<STR_LIT>" % res ) <EOL> logging . info ( "<STR_LIT>" % res ) <EOL> self . resfile . flush ( ) <EOL> def _redraw ( self , db , tofile = <NUM_LIT:0> , eta = None ) : <EOL> """<STR_LIT>""" <EOL> import pylab as p <EOL> if len ( db ) <= <NUM_LIT:3> or not self . plotflag : return <EOL> nContours = <NUM_LIT> <EOL> def cmp ( x , y ) : <EOL> if x [ <NUM_LIT:0> ] < y [ <NUM_LIT:0> ] : return - <NUM_LIT:1> <EOL> if x [ <NUM_LIT:0> ] > y [ <NUM_LIT:0> ] : return <NUM_LIT:1> <EOL> if x [ <NUM_LIT:1> ] > y [ <NUM_LIT:1> ] : return - <NUM_LIT:1> <EOL> if x [ <NUM_LIT:1> ] < y [ <NUM_LIT:1> ] : return <NUM_LIT:1> <EOL> return <NUM_LIT:0> <EOL> db . sort ( cmp ) <EOL> dbarr = p . asarray ( db ) <EOL> x = dbarr [ : , <NUM_LIT:0> ] <EOL> dimy = len ( x [ x == x [ <NUM_LIT:0> ] ] ) <EOL> dimx = x . size / dimy <EOL> print ( ( '<STR_LIT>' , dimx , dimy ) ) <EOL> x = x . reshape ( dimx , dimy ) <EOL> y = dbarr [ : , <NUM_LIT:1> ] <EOL> y = y . reshape ( dimx , dimy ) <EOL> z = dbarr [ : , <NUM_LIT:2> ] . reshape ( dimx , dimy ) <EOL> p . ioff ( ) <EOL> p . clf ( ) <EOL> p . contourf ( x , y , z , nContours ) <EOL> p . hsv ( ) <EOL> p . colorbar ( ) <EOL> p . xlim ( self . usermin [ <NUM_LIT:0> ] , self . usermax [ <NUM_LIT:0> ] ) <EOL> p . ylim ( self . usermin [ <NUM_LIT:1> ] , self . usermax [ <NUM_LIT:1> ] ) <EOL> p . xlabel ( r'<STR_LIT>' ) <EOL> p . ylabel ( r'<STR_LIT>' ) <EOL> p . ion ( ) <EOL> p . draw_if_interactive ( ) <EOL> class GridSearchDOE ( GridSearch ) : <EOL> """<STR_LIT>""" <EOL> doepat = array ( [ [ <NUM_LIT:0.5> , <NUM_LIT:1> ] , [ <NUM_LIT> , <NUM_LIT> ] , [ <NUM_LIT> , <NUM_LIT> ] , [ <NUM_LIT:0> , <NUM_LIT:0.5> ] , [ <NUM_LIT:1> , <NUM_LIT:0.5> ] , [ <NUM_LIT> , <NUM_LIT> ] , [ <NUM_LIT> , <NUM_LIT> ] , [ <NUM_LIT:0.5> , <NUM_LIT:0> ] , [ <NUM_LIT:0> , <NUM_LIT:1> ] , [ <NUM_LIT:1> , <NUM_LIT:1> ] , [ <NUM_LIT:0.5> , <NUM_LIT:0.5> ] , [ <NUM_LIT:0> , <NUM_LIT:0> ] , [ <NUM_LIT:1> , <NUM_LIT:0> ] ] ) <EOL> nPts = <NUM_LIT> <EOL> depth = <NUM_LIT:0> <EOL> def search ( self , cmin = None , cmax = None ) : <EOL> """<STR_LIT>""" <EOL> if self . depth > self . maxdepth : <EOL> best = self . allPts [ self . allScores . argmax ( ) , : ] <EOL> logging . info ( "<STR_LIT>" % ( best [ <NUM_LIT:0> ] , best [ <NUM_LIT:1> ] ) ) <EOL> self . setParams ( C = <NUM_LIT:2> ** best [ <NUM_LIT:0> ] , gamma = <NUM_LIT:2> ** best [ <NUM_LIT:1> ] ) <EOL> param = svm_parameter ( ** self . params ) <EOL> logging . info ( "<STR_LIT>" ) <EOL> logging . info ( repr ( param ) ) <EOL> return param <EOL> if cmin is None : <EOL> cmin = array ( self . usermin ) <EOL> cmax = array ( self . usermax ) <EOL> points = self . refineGrid ( cmin , cmax ) <EOL> scores = [ ] <EOL> isnew = array ( [ True ] * self . nPts ) <EOL> for i in range ( self . nPts ) : <EOL> idx = self . _findIndex ( points [ i , : ] ) <EOL> if idx >= <NUM_LIT:0> : <EOL> isnew [ i ] = False <EOL> scores . append ( self . allScores [ idx ] ) <EOL> else : <EOL> self . setParams ( C = <NUM_LIT:2> ** points [ i , <NUM_LIT:0> ] , gamma = <NUM_LIT:2> ** points [ i , <NUM_LIT:1> ] ) <EOL> param = svm_parameter ( ** self . params ) <EOL> cvresult = array ( cross_validation ( self . problem , param , self . crossval ) ) <EOL> corr , = where ( cvresult == self . targets ) <EOL> corr = float ( corr . size ) / self . targets . size <EOL> scores . append ( corr ) <EOL> self . _save_points ( ( points [ i , <NUM_LIT:0> ] , points [ i , <NUM_LIT:1> ] , corr ) ) <EOL> scores = array ( scores ) <EOL> newctr = points [ scores . argmax ( ) , : ] . copy ( ) <EOL> newdiff = ( cmax - cmin ) / <NUM_LIT> <EOL> for i in range ( self . nPars ) : <EOL> newctr [ i ] = min ( [ max ( [ newctr [ i ] , self . usermin [ i ] + newdiff [ i ] ] ) , self . usermax [ i ] - newdiff [ i ] ] ) <EOL> cmin = newctr - newdiff <EOL> cmax = newctr + newdiff <EOL> logging . info ( "<STR_LIT>" % ( self . depth , cmax [ <NUM_LIT:0> ] - cmin [ <NUM_LIT:0> ] , scores . max ( ) ) ) <EOL> if self . depth == <NUM_LIT:0> : <EOL> self . allPts = points [ isnew , : ] . copy ( ) <EOL> self . allScores = scores [ isnew ] . copy ( ) <EOL> else : <EOL> self . allPts = append ( self . allPts , points [ isnew , : ] , axis = <NUM_LIT:0> ) <EOL> self . allScores = append ( self . allScores , scores [ isnew ] , axis = <NUM_LIT:0> ) <EOL> if self . plotflag : <EOL> import pylab as p <EOL> if self . depth == <NUM_LIT:0> : <EOL> self . oPlot = p . plot ( self . allPts [ : , <NUM_LIT:0> ] , self . allPts [ : , <NUM_LIT:1> ] , '<STR_LIT:o>' ) [ <NUM_LIT:0> ] <EOL> self . oPlot . set_data ( self . allPts [ : , <NUM_LIT:0> ] , self . allPts [ : , <NUM_LIT:1> ] ) <EOL> p . draw ( ) <EOL> self . depth += <NUM_LIT:1> <EOL> return self . search ( cmin , cmax ) <EOL> def refineGrid ( self , cmin , cmax ) : <EOL> """<STR_LIT>""" <EOL> diff = array ( ( cmax - cmin ) . tolist ( ) * self . nPts ) . reshape ( self . nPts , self . nPars ) <EOL> return self . doepat * diff + array ( cmin . tolist ( ) * self . nPts ) . reshape ( self . nPts , self . nPars ) <EOL> def _findIndex ( self , point ) : <EOL> """<STR_LIT>""" <EOL> if self . depth == <NUM_LIT:0> : return - <NUM_LIT:1> <EOL> check = self . allPts [ : , <NUM_LIT:0> ] == point [ <NUM_LIT:0> ] <EOL> for i in range ( <NUM_LIT:1> , point . size ) : <EOL> check = check & ( self . allPts [ : , i ] == point [ i ] ) <EOL> idx , = where ( check ) <EOL> if idx . size == <NUM_LIT:0> : <EOL> return - <NUM_LIT:1> <EOL> elif idx . size > <NUM_LIT:1> : <EOL> logging . error ( "<STR_LIT>" ) <EOL> logging . error ( str ( point ) ) <EOL> logging . error ( str ( self . allPts ) ) <EOL> raise <EOL> else : <EOL> return idx [ <NUM_LIT:0> ] </s>
