<s> def keyset_reduce ( iter_ , params_ ) : <EOL> import ujson <EOL> import disco . util <EOL> def _safe_str ( value ) : <EOL> try : <EOL> return str ( value ) <EOL> except UnicodeEncodeError : <EOL> return unicode ( value ) . encode ( '<STR_LIT:utf-8>' ) <EOL> def _disco_message ( message ) : <EOL> print message <EOL> def _inferno_debug ( params , message , * args ) : <EOL> if getattr ( params , '<STR_LIT>' , False ) : <EOL> _disco_message ( message % args ) <EOL> def _inferno_error ( message , * args ) : <EOL> import traceback <EOL> trace = traceback . format_exc ( <NUM_LIT:15> ) <EOL> _disco_message ( '<STR_LIT>' % ( message , trace ) ) <EOL> def _apply_process ( params , it , func ) : <EOL> for key , val in it : <EOL> for rkey , rval in func ( key , val , params ) : <EOL> yield rkey , rval <EOL> def _post_process ( params , key , val ) : <EOL> it = iter ( list ( [ ( key , val ) ] ) ) <EOL> keysets = getattr ( params , '<STR_LIT>' , None ) <EOL> if keysets : <EOL> keyset = keysets . get ( key [ <NUM_LIT:0> ] , None ) <EOL> if keyset and keyset [ '<STR_LIT>' ] : <EOL> for func in keyset [ '<STR_LIT>' ] : <EOL> it = _apply_process ( params , it , func ) <EOL> for k , v in it : <EOL> yield k , v <EOL> def _convert_to_numeric ( val ) : <EOL> if isinstance ( val , ( float , int , long ) ) : <EOL> return val <EOL> else : <EOL> try : <EOL> return float ( val ) <EOL> except : <EOL> return <NUM_LIT:0> <EOL> def _sum_group ( key , value ) : <EOL> summed_values = [ ] <EOL> for row in value : <EOL> index = <NUM_LIT:0> <EOL> _inferno_debug ( '<STR_LIT>' , key , row ) <EOL> for item in row : <EOL> item = _convert_to_numeric ( item ) <EOL> if index >= len ( summed_values ) : <EOL> summed_values . append ( item ) <EOL> else : <EOL> summed_values [ index ] += item <EOL> index += <NUM_LIT:1> <EOL> return summed_values <EOL> init_func = getattr ( params_ , '<STR_LIT>' , None ) <EOL> if init_func : <EOL> init_func ( iter_ , params_ ) <EOL> for key , value in disco . util . kvgroup ( iter_ ) : <EOL> try : <EOL> key = ujson . loads ( key ) <EOL> sum_ = _sum_group ( key , value ) <EOL> for xkey , xval in _post_process ( params_ , key , sum_ ) : <EOL> if ( hasattr ( params_ , '<STR_LIT>' ) and <EOL> params_ . serial_out ) : <EOL> serial = '<STR_LIT:U+002C>' . join ( [ _safe_str ( y ) for y in xkey [ <NUM_LIT:1> : ] ] ) <EOL> result = serial , ujson . dumps ( xval ) <EOL> else : <EOL> result = xkey , xval <EOL> _inferno_debug ( '<STR_LIT>' , result ) <EOL> yield result <EOL> except Exception as e : <EOL> _inferno_error ( '<STR_LIT>' , e , key , value ) </s>
