<s> """<STR_LIT>""" <EOL> def simple_init ( interface , params ) : <EOL> return params <EOL> def map_fit ( interface , state , label , inp ) : <EOL> """<STR_LIT>""" <EOL> import numpy as np <EOL> combiner = { } <EOL> out = interface . output ( <NUM_LIT:0> ) <EOL> for row in inp : <EOL> row = row . strip ( ) . split ( state [ "<STR_LIT>" ] ) <EOL> if len ( row ) > <NUM_LIT:1> : <EOL> for i , j in enumerate ( state [ "<STR_LIT>" ] ) : <EOL> if row [ j ] not in state [ "<STR_LIT>" ] : <EOL> pair = row [ state [ "<STR_LIT>" ] ] + state [ "<STR_LIT>" ] + str ( j ) <EOL> if state [ "<STR_LIT>" ] [ i ] == "<STR_LIT:c>" : <EOL> if pair in combiner : <EOL> combiner [ pair ] . append ( np . float32 ( row [ j ] ) ) <EOL> else : <EOL> combiner [ pair ] = [ np . float32 ( row [ j ] ) ] <EOL> else : <EOL> pair += state [ "<STR_LIT>" ] + row [ j ] <EOL> combiner [ pair ] = combiner . get ( pair , <NUM_LIT:0> ) + <NUM_LIT:1> <EOL> combiner [ row [ state [ "<STR_LIT>" ] ] ] = combiner . get ( row [ state [ "<STR_LIT>" ] ] , <NUM_LIT:0> ) + <NUM_LIT:1> <EOL> for k , v in combiner . iteritems ( ) : <EOL> if len ( k . split ( state [ "<STR_LIT>" ] ) ) == <NUM_LIT:2> : <EOL> out . add ( k , ( np . size ( v ) , np . mean ( v , dtype = np . float32 ) , np . var ( v , dtype = np . float32 ) ) ) <EOL> else : <EOL> out . add ( k , v ) <EOL> def reduce_fit ( interface , state , label , inp ) : <EOL> """<STR_LIT>""" <EOL> from disco . util import kvgroup <EOL> import numpy as np <EOL> out = interface . output ( <NUM_LIT:0> ) <EOL> fit_model = { "<STR_LIT>" : [ ] , "<STR_LIT>" : <NUM_LIT:0> , "<STR_LIT>" : set ( ) } <EOL> combiner = { } <EOL> means , variances = [ ] , [ ] <EOL> k_prev = "<STR_LIT>" <EOL> for key , value in kvgroup ( inp ) : <EOL> k_split = key . split ( state [ "<STR_LIT>" ] ) <EOL> if len ( k_split ) == <NUM_LIT:3> : <EOL> fit_model [ "<STR_LIT>" ] . add ( tuple ( k_split [ <NUM_LIT:1> : ] ) ) <EOL> out . add ( tuple ( k_split ) , sum ( value ) ) <EOL> elif len ( k_split ) == <NUM_LIT:2> : <EOL> if k_split [ <NUM_LIT:0> ] != k_prev and k_prev != "<STR_LIT>" : <EOL> mean , var = zip ( * [ combiner [ key ] for key in sorted ( combiner . keys ( ) ) ] ) <EOL> means . append ( mean ) <EOL> variances . append ( var ) <EOL> n_a = mean_a = var_a = <NUM_LIT:0> <EOL> for n_b , mean_b , var_b in value : <EOL> n_ab = n_a + n_b <EOL> var_a = ( ( n_a * var_a + n_b * var_b ) / float ( n_ab ) ) + ( <EOL> n_a * n_b * ( ( mean_b - mean_a ) / float ( n_ab ) ) ** <NUM_LIT:2> ) <EOL> mean_a = ( n_a * mean_a + n_b * mean_b ) / float ( n_ab ) <EOL> n_a = n_ab <EOL> combiner [ int ( k_split [ <NUM_LIT:1> ] ) ] = ( mean_a , var_a + <NUM_LIT> ) <EOL> k_prev = k_split [ <NUM_LIT:0> ] <EOL> else : <EOL> fit_model [ key ] = np . sum ( value ) <EOL> fit_model [ "<STR_LIT>" ] += fit_model [ key ] <EOL> fit_model [ "<STR_LIT>" ] . append ( key ) <EOL> if len ( means ) > <NUM_LIT:0> : <EOL> mean , var = zip ( * [ combiner [ key ] for key in sorted ( combiner . keys ( ) ) ] ) <EOL> out . add ( "<STR_LIT>" , np . array ( means + [ mean ] , dtype = np . float32 ) ) <EOL> variances = np . array ( variances + [ var ] , dtype = np . float32 ) <EOL> out . add ( "<STR_LIT>" , variances ) <EOL> out . add ( "<STR_LIT>" , np . log ( np . pi * variances ) ) <EOL> prior = [ fit_model [ y_label ] / float ( fit_model [ "<STR_LIT>" ] ) for y_label in fit_model [ "<STR_LIT>" ] ] <EOL> out . add ( "<STR_LIT>" , np . array ( prior , dtype = np . float32 ) ) <EOL> out . add ( "<STR_LIT>" , np . log ( prior ) ) <EOL> out . add ( "<STR_LIT>" , list ( fit_model [ "<STR_LIT>" ] ) ) <EOL> out . add ( "<STR_LIT>" , fit_model [ "<STR_LIT>" ] ) <EOL> def map_predict ( interface , state , label , inp ) : <EOL> """<STR_LIT>""" <EOL> import numpy as np <EOL> out = interface . output ( <NUM_LIT:0> ) <EOL> continuous = [ j for i , j in enumerate ( state [ "<STR_LIT>" ] ) if <EOL> state [ "<STR_LIT>" ] [ i ] == "<STR_LIT:c>" ] <EOL> discrete = [ j for i , j in enumerate ( state [ "<STR_LIT>" ] ) if <EOL> state [ "<STR_LIT>" ] [ i ] == "<STR_LIT:d>" ] <EOL> cont = True if len ( continuous ) > <NUM_LIT:0> else False <EOL> disc = True if len ( discrete ) > <NUM_LIT:0> else False <EOL> for row in inp : <EOL> row = row . strip ( ) . split ( state [ "<STR_LIT>" ] ) <EOL> if len ( row ) > <NUM_LIT:1> : <EOL> x_id = "<STR_LIT>" if state [ "<STR_LIT>" ] == - <NUM_LIT:1> else row [ state [ "<STR_LIT>" ] ] <EOL> probs = state [ "<STR_LIT>" ] [ "<STR_LIT>" ] <EOL> if cont : <EOL> x = np . array ( [ ( <NUM_LIT:0> if row [ j ] in state [ "<STR_LIT>" ] else float ( row [ j ] ) ) for j in <EOL> continuous ] ) <EOL> probs = probs - <NUM_LIT:0.5> * np . sum ( <EOL> np . true_divide ( ( x - state [ "<STR_LIT>" ] [ "<STR_LIT>" ] ) ** <NUM_LIT:2> , state [ "<STR_LIT>" ] [ "<STR_LIT>" ] ) + <EOL> state [ "<STR_LIT>" ] [ "<STR_LIT>" ] , axis = <NUM_LIT:1> ) <EOL> if disc : <EOL> probs = probs + np . sum ( <EOL> [ ( <NUM_LIT:0> if row [ i ] in state [ "<STR_LIT>" ] else state [ "<STR_LIT>" ] . get ( ( str ( i ) , row [ i ] ) , np . zeros ( <NUM_LIT:1> ) ) ) <EOL> for i in discrete ] , axis = <NUM_LIT:0> ) <EOL> log_prob_x = np . log ( np . sum ( np . exp ( probs ) ) ) <EOL> probs = np . exp ( np . array ( probs ) - log_prob_x ) <EOL> y_predicted = max ( zip ( probs , state [ "<STR_LIT>" ] [ "<STR_LIT>" ] ) ) [ <NUM_LIT:1> ] <EOL> out . add ( x_id , ( y_predicted , probs . tolist ( ) ) ) <EOL> def fit ( dataset , save_results = True , show = False ) : <EOL> """<STR_LIT>""" <EOL> from disco . worker . pipeline . worker import Worker , Stage <EOL> from disco . core import Job <EOL> job = Job ( worker = Worker ( save_results = save_results ) ) <EOL> job . pipeline = [ <EOL> ( "<STR_LIT>" , Stage ( "<STR_LIT>" , input_chain = dataset . params [ "<STR_LIT>" ] , init = simple_init , process = map_fit ) ) , <EOL> ( '<STR_LIT>' , Stage ( "<STR_LIT>" , init = simple_init , process = reduce_fit , sort = True , combine = True ) ) ] <EOL> job . params = dataset . params <EOL> job . run ( name = "<STR_LIT>" , input = dataset . params [ "<STR_LIT>" ] ) <EOL> fitmodel_url = job . wait ( show = show ) <EOL> return { "<STR_LIT>" : fitmodel_url } <EOL> def predict ( dataset , fitmodel_url , m = <NUM_LIT:1> , save_results = True , show = False ) : <EOL> """<STR_LIT>""" <EOL> from disco . worker . pipeline . worker import Worker , Stage <EOL> from disco . core import Job , result_iterator <EOL> import numpy as np <EOL> try : <EOL> m = float ( m ) <EOL> except ValueError : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> if "<STR_LIT>" in fitmodel_url : <EOL> fit_model = dict ( ( k , v ) for k , v in result_iterator ( fitmodel_url [ "<STR_LIT>" ] ) ) <EOL> if len ( fit_model [ "<STR_LIT>" ] ) < <NUM_LIT:2> : <EOL> print "<STR_LIT>" <EOL> return [ ] <EOL> else : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> if dataset . params [ "<STR_LIT>" ] . count ( "<STR_LIT:d>" ) > <NUM_LIT:0> : <EOL> np . seterr ( divide = '<STR_LIT:ignore>' ) <EOL> for iv in fit_model [ "<STR_LIT>" ] : <EOL> dist = [ fit_model . pop ( ( y , ) + iv , <NUM_LIT:0> ) for y in fit_model [ "<STR_LIT>" ] ] <EOL> fit_model [ iv ] = np . nan_to_num ( <EOL> np . log ( np . true_divide ( np . array ( dist ) + m * fit_model [ "<STR_LIT>" ] , np . sum ( dist ) + m ) ) ) - fit_model [ <EOL> "<STR_LIT>" ] <EOL> del ( fit_model [ "<STR_LIT>" ] ) <EOL> job = Job ( worker = Worker ( save_results = save_results ) ) <EOL> job . pipeline = [ <EOL> ( "<STR_LIT>" , Stage ( "<STR_LIT>" , input_chain = dataset . params [ "<STR_LIT>" ] , init = simple_init , process = map_predict ) ) ] <EOL> job . params = dataset . params <EOL> job . params [ "<STR_LIT>" ] = fit_model <EOL> job . run ( name = "<STR_LIT>" , input = dataset . params [ "<STR_LIT>" ] ) <EOL> results = job . wait ( show = show ) <EOL> return results </s>
