<s> '''<STR_LIT>''' <EOL> import numpy as np <EOL> import cv2 <EOL> from multiprocessing . pool import ThreadPool <EOL> from digits import * <EOL> try : <EOL> import cloud <EOL> have_cloud = True <EOL> except ImportError : <EOL> have_cloud = False <EOL> def cross_validate ( model_class , params , samples , labels , kfold = <NUM_LIT:3> , pool = None ) : <EOL> n = len ( samples ) <EOL> folds = np . array_split ( np . arange ( n ) , kfold ) <EOL> def f ( i ) : <EOL> model = model_class ( ** params ) <EOL> test_idx = folds [ i ] <EOL> train_idx = list ( folds ) <EOL> train_idx . pop ( i ) <EOL> train_idx = np . hstack ( train_idx ) <EOL> train_samples , train_labels = samples [ train_idx ] , labels [ train_idx ] <EOL> test_samples , test_labels = samples [ test_idx ] , labels [ test_idx ] <EOL> model . train ( train_samples , train_labels ) <EOL> resp = model . predict ( test_samples ) <EOL> score = ( resp != test_labels ) . mean ( ) <EOL> print "<STR_LIT:.>" , <EOL> return score <EOL> if pool is None : <EOL> scores = map ( f , xrange ( kfold ) ) <EOL> else : <EOL> scores = pool . map ( f , xrange ( kfold ) ) <EOL> return np . mean ( scores ) <EOL> class App ( object ) : <EOL> def __init__ ( self , usecloud = False , cloud_env = '<STR_LIT>' ) : <EOL> if usecloud and not have_cloud : <EOL> print '<STR_LIT>' <EOL> usecloud = False <EOL> self . usecloud = usecloud <EOL> self . cloud_env = cloud_env <EOL> if self . usecloud : <EOL> print '<STR_LIT>' <EOL> cloud . files . put ( DIGITS_FN ) <EOL> self . preprocess_job = cloud . call ( self . preprocess , _env = self . cloud_env ) <EOL> else : <EOL> self . _samples , self . _labels = self . preprocess ( ) <EOL> def preprocess ( self ) : <EOL> if self . usecloud : <EOL> cloud . files . get ( DIGITS_FN ) <EOL> digits , labels = load_digits ( DIGITS_FN ) <EOL> shuffle = np . random . permutation ( len ( digits ) ) <EOL> digits , labels = digits [ shuffle ] , labels [ shuffle ] <EOL> digits2 = map ( deskew , digits ) <EOL> samples = preprocess_hog ( digits2 ) <EOL> return samples , labels <EOL> def get_dataset ( self ) : <EOL> if self . usecloud : <EOL> return cloud . result ( self . preprocess_job ) <EOL> else : <EOL> return self . _samples , self . _labels <EOL> def run_jobs ( self , f , jobs ) : <EOL> if self . usecloud : <EOL> jids = cloud . map ( f , jobs , _env = self . cloud_env , _profile = True , _depends_on = self . preprocess_job ) <EOL> ires = cloud . iresult ( jids ) <EOL> else : <EOL> pool = ThreadPool ( processes = cv2 . getNumberOfCPUs ( ) ) <EOL> ires = pool . imap_unordered ( f , jobs ) <EOL> return ires <EOL> def adjust_SVM ( self ) : <EOL> Cs = np . logspace ( <NUM_LIT:0> , <NUM_LIT:10> , <NUM_LIT:15> , base = <NUM_LIT:2> ) <EOL> gammas = np . logspace ( - <NUM_LIT:7> , <NUM_LIT:4> , <NUM_LIT:15> , base = <NUM_LIT:2> ) <EOL> scores = np . zeros ( ( len ( Cs ) , len ( gammas ) ) ) <EOL> scores [ : ] = np . nan <EOL> print '<STR_LIT>' <EOL> def f ( job ) : <EOL> i , j = job <EOL> samples , labels = self . get_dataset ( ) <EOL> params = dict ( C = Cs [ i ] , gamma = gammas [ j ] ) <EOL> score = cross_validate ( SVM , params , samples , labels ) <EOL> return i , j , score <EOL> ires = self . run_jobs ( f , np . ndindex ( * scores . shape ) ) <EOL> for count , ( i , j , score ) in enumerate ( ires ) : <EOL> scores [ i , j ] = score <EOL> print '<STR_LIT>' % ( count + <NUM_LIT:1> , scores . size , np . nanmin ( scores ) * <NUM_LIT:100> , score * <NUM_LIT:100> ) <EOL> print scores <EOL> print '<STR_LIT>' <EOL> np . savez ( '<STR_LIT>' , scores = scores , Cs = Cs , gammas = gammas ) <EOL> i , j = np . unravel_index ( scores . argmin ( ) , scores . shape ) <EOL> best_params = dict ( C = Cs [ i ] , gamma = gammas [ j ] ) <EOL> print '<STR_LIT>' , best_params <EOL> print '<STR_LIT>' % ( scores . min ( ) * <NUM_LIT:100> ) <EOL> return best_params <EOL> def adjust_KNearest ( self ) : <EOL> print '<STR_LIT>' <EOL> def f ( k ) : <EOL> samples , labels = self . get_dataset ( ) <EOL> err = cross_validate ( KNearest , dict ( k = k ) , samples , labels ) <EOL> return k , err <EOL> best_err , best_k = np . inf , - <NUM_LIT:1> <EOL> for k , err in self . run_jobs ( f , xrange ( <NUM_LIT:1> , <NUM_LIT:9> ) ) : <EOL> if err < best_err : <EOL> best_err , best_k = err , k <EOL> print '<STR_LIT>' % ( k , err * <NUM_LIT:100> ) <EOL> best_params = dict ( k = best_k ) <EOL> print '<STR_LIT>' , best_params , '<STR_LIT>' % ( best_err * <NUM_LIT:100> ) <EOL> return best_params <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> import getopt <EOL> import sys <EOL> print __doc__ <EOL> args , _ = getopt . getopt ( sys . argv [ <NUM_LIT:1> : ] , '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> args = dict ( args ) <EOL> args . setdefault ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> args . setdefault ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> if args [ '<STR_LIT>' ] not in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> print '<STR_LIT>' % args [ '<STR_LIT>' ] <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> t = clock ( ) <EOL> app = App ( usecloud = '<STR_LIT>' in args , cloud_env = args [ '<STR_LIT>' ] ) <EOL> if args [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> app . adjust_KNearest ( ) <EOL> else : <EOL> app . adjust_SVM ( ) <EOL> print '<STR_LIT>' % ( clock ( ) - t ) </s>
