<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import <EOL> from google . cloud . dataflow . transforms import combiners <EOL> from google . cloud . dataflow . utils . counters import Counter <EOL> class Aggregator ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , <EOL> name , <EOL> combine_fn = sum , <EOL> input_type = None ) : <EOL> self . name = name <EOL> self . combine_fn = combine_fn <EOL> self . aggregation_kind = self . _aggregator_counter_kind ( combine_fn ) <EOL> self . input_type = input_type <EOL> def __str__ ( self ) : <EOL> return '<STR_LIT>' % self . _str_internal ( ) <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( self . _str_internal ( ) , hex ( id ( self ) ) ) <EOL> def _str_internal ( self ) : <EOL> """<STR_LIT>""" <EOL> def get_name ( thing ) : <EOL> try : <EOL> return thing . __name__ <EOL> except AttributeError : <EOL> return thing . __class__ . __name__ <EOL> combine_fn_str = get_name ( self . combine_fn ) <EOL> input_arg = '<STR_LIT>' % get_name ( self . input_type ) if self . input_type else '<STR_LIT>' <EOL> if combine_fn_str == '<STR_LIT>' and not input_arg : <EOL> combine_call = '<STR_LIT>' <EOL> else : <EOL> combine_call = '<STR_LIT>' % ( combine_fn_str , input_arg ) <EOL> return '<STR_LIT>' % ( self . name , combine_call ) <EOL> @ staticmethod <EOL> def _aggregator_counter_kind ( combine_fn ) : <EOL> """<STR_LIT>""" <EOL> combine_kind_map = { sum : Counter . SUM , max : Counter . MAX , min : Counter . MIN , <EOL> combiners . Mean : Counter . MEAN } <EOL> try : <EOL> return combine_kind_map [ combine_fn ] <EOL> except KeyError : <EOL> try : <EOL> return combine_kind_map [ combine_fn . __class__ ] <EOL> except KeyError : <EOL> raise ValueError ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> % ( combine_fn , combine_fn . __class__ ) ) </s>
