<s> """<STR_LIT>""" <EOL> import itertools <EOL> import logging <EOL> import warnings <EOL> import operator <EOL> import re <EOL> import time <EOL> from datetime import datetime , timedelta <EOL> from luigi import six <EOL> import luigi <EOL> from luigi . parameter import ParameterException <EOL> from luigi . target import FileSystemTarget <EOL> from luigi . task import Register , flatten_output <EOL> logger = logging . getLogger ( '<STR_LIT>' ) <EOL> class RangeEvent ( luigi . Event ) : <EOL> """<STR_LIT>""" <EOL> COMPLETE_COUNT = "<STR_LIT>" <EOL> COMPLETE_FRACTION = "<STR_LIT>" <EOL> DELAY = "<STR_LIT>" <EOL> class RangeBase ( luigi . WrapperTask ) : <EOL> """<STR_LIT>""" <EOL> of = luigi . TaskParameter ( <EOL> description = "<STR_LIT>" ) <EOL> start = luigi . Parameter ( ) <EOL> stop = luigi . Parameter ( ) <EOL> reverse = luigi . BoolParameter ( <EOL> default = False , <EOL> description = "<STR_LIT>" ) <EOL> task_limit = luigi . IntParameter ( <EOL> default = <NUM_LIT:50> , <EOL> description = "<STR_LIT>" ) <EOL> now = luigi . IntParameter ( <EOL> default = None , <EOL> description = "<STR_LIT>" ) <EOL> param_name = luigi . Parameter ( <EOL> default = None , <EOL> description = "<STR_LIT>" , <EOL> positional = False ) <EOL> @ property <EOL> def of_cls ( self ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( self . of , six . string_types ) : <EOL> warnings . warn ( '<STR_LIT>' ) <EOL> return Register . get_task_cls ( self . of ) <EOL> return self . of <EOL> def datetime_to_parameter ( self , dt ) : <EOL> raise NotImplementedError <EOL> def parameter_to_datetime ( self , p ) : <EOL> raise NotImplementedError <EOL> def moving_start ( self , now ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def moving_stop ( self , now ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def finite_datetimes ( self , finite_start , finite_stop ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def _emit_metrics ( self , missing_datetimes , finite_start , finite_stop ) : <EOL> """<STR_LIT>""" <EOL> datetimes = self . finite_datetimes ( <EOL> finite_start if self . start is None else min ( finite_start , self . parameter_to_datetime ( self . start ) ) , <EOL> finite_stop if self . stop is None else max ( finite_stop , self . parameter_to_datetime ( self . stop ) ) ) <EOL> delay_in_jobs = len ( datetimes ) - datetimes . index ( missing_datetimes [ <NUM_LIT:0> ] ) if datetimes and missing_datetimes else <NUM_LIT:0> <EOL> self . trigger_event ( RangeEvent . DELAY , self . of_cls . task_family , delay_in_jobs ) <EOL> expected_count = len ( datetimes ) <EOL> complete_count = expected_count - len ( missing_datetimes ) <EOL> self . trigger_event ( RangeEvent . COMPLETE_COUNT , self . of_cls . task_family , complete_count ) <EOL> self . trigger_event ( RangeEvent . COMPLETE_FRACTION , self . of_cls . task_family , float ( complete_count ) / expected_count if expected_count else <NUM_LIT:1> ) <EOL> def _format_datetime ( self , dt ) : <EOL> return self . datetime_to_parameter ( dt ) <EOL> def _format_range ( self , datetimes ) : <EOL> param_first = self . _format_datetime ( datetimes [ <NUM_LIT:0> ] ) <EOL> param_last = self . _format_datetime ( datetimes [ - <NUM_LIT:1> ] ) <EOL> return '<STR_LIT>' % ( param_first , param_last ) <EOL> def _instantiate_task_cls ( self , param ) : <EOL> if self . param_name is None : <EOL> return self . of_cls ( param ) <EOL> else : <EOL> kwargs = { self . param_name : param } <EOL> return self . of_cls ( ** kwargs ) <EOL> def requires ( self ) : <EOL> if hasattr ( self , '<STR_LIT>' ) : <EOL> return self . _cached_requires <EOL> if not self . start and not self . stop : <EOL> raise ParameterException ( "<STR_LIT>" ) <EOL> if not self . start and not self . reverse : <EOL> raise ParameterException ( "<STR_LIT>" ) <EOL> if self . start and self . stop and self . start > self . stop : <EOL> raise ParameterException ( "<STR_LIT>" ) <EOL> now = datetime . utcfromtimestamp ( time . time ( ) if self . now is None else self . now ) <EOL> moving_start = self . moving_start ( now ) <EOL> finite_start = moving_start if self . start is None else max ( self . parameter_to_datetime ( self . start ) , moving_start ) <EOL> moving_stop = self . moving_stop ( now ) <EOL> finite_stop = moving_stop if self . stop is None else min ( self . parameter_to_datetime ( self . stop ) , moving_stop ) <EOL> datetimes = self . finite_datetimes ( finite_start , finite_stop ) if finite_start <= finite_stop else [ ] <EOL> if datetimes : <EOL> logger . debug ( '<STR_LIT>' , <EOL> self . _format_range ( datetimes ) , self . of_cls . task_family ) <EOL> missing_datetimes = sorted ( self . _missing_datetimes ( datetimes ) ) <EOL> logger . debug ( '<STR_LIT>' , <EOL> self . _format_range ( datetimes ) , len ( missing_datetimes ) , len ( datetimes ) , self . of_cls . task_family ) <EOL> else : <EOL> missing_datetimes = [ ] <EOL> logger . debug ( '<STR_LIT>' , self . of_cls . task_family ) <EOL> self . _emit_metrics ( missing_datetimes , finite_start , finite_stop ) <EOL> if self . reverse : <EOL> required_datetimes = missing_datetimes [ - self . task_limit : ] <EOL> else : <EOL> required_datetimes = missing_datetimes [ : self . task_limit ] <EOL> if required_datetimes : <EOL> logger . debug ( '<STR_LIT>' , <EOL> len ( required_datetimes ) , self . of_cls . task_family , self . _format_range ( required_datetimes ) ) <EOL> if self . reverse : <EOL> required_datetimes . reverse ( ) <EOL> self . _cached_requires = [ self . _instantiate_task_cls ( self . datetime_to_parameter ( d ) ) for d in required_datetimes ] <EOL> return self . _cached_requires <EOL> def missing_datetimes ( self , finite_datetimes ) : <EOL> """<STR_LIT>""" <EOL> return [ d for d in finite_datetimes if not self . _instantiate_task_cls ( self . datetime_to_parameter ( d ) ) . complete ( ) ] <EOL> def _missing_datetimes ( self , finite_datetimes ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return self . missing_datetimes ( finite_datetimes ) <EOL> except TypeError as ex : <EOL> if '<STR_LIT>' in repr ( ex ) : <EOL> warnings . warn ( '<STR_LIT>' ) <EOL> return self . missing_datetimes ( self . of_cls , finite_datetimes ) <EOL> else : <EOL> raise <EOL> class RangeDailyBase ( RangeBase ) : <EOL> """<STR_LIT>""" <EOL> start = luigi . DateParameter ( <EOL> default = None , <EOL> description = "<STR_LIT>" ) <EOL> stop = luigi . DateParameter ( <EOL> default = None , <EOL> description = "<STR_LIT>" ) <EOL> days_back = luigi . IntParameter ( <EOL> default = <NUM_LIT:100> , <EOL> description = ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> days_forward = luigi . IntParameter ( <EOL> default = <NUM_LIT:0> , <EOL> description = "<STR_LIT>" ) <EOL> def datetime_to_parameter ( self , dt ) : <EOL> return dt . date ( ) <EOL> def parameter_to_datetime ( self , p ) : <EOL> return datetime ( p . year , p . month , p . day ) <EOL> def moving_start ( self , now ) : <EOL> return now - timedelta ( days = self . days_back ) <EOL> def moving_stop ( self , now ) : <EOL> return now + timedelta ( days = self . days_forward ) <EOL> def finite_datetimes ( self , finite_start , finite_stop ) : <EOL> """<STR_LIT>""" <EOL> date_start = datetime ( finite_start . year , finite_start . month , finite_start . day ) <EOL> dates = [ ] <EOL> for i in itertools . count ( ) : <EOL> t = date_start + timedelta ( days = i ) <EOL> if t >= finite_stop : <EOL> return dates <EOL> if t >= finite_start : <EOL> dates . append ( t ) <EOL> class RangeHourlyBase ( RangeBase ) : <EOL> """<STR_LIT>""" <EOL> start = luigi . DateHourParameter ( <EOL> default = None , <EOL> description = "<STR_LIT>" ) <EOL> stop = luigi . DateHourParameter ( <EOL> default = None , <EOL> description = "<STR_LIT>" ) <EOL> hours_back = luigi . IntParameter ( <EOL> default = <NUM_LIT:100> * <NUM_LIT> , <EOL> description = ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> hours_forward = luigi . IntParameter ( <EOL> default = <NUM_LIT:0> , <EOL> description = "<STR_LIT>" ) <EOL> def datetime_to_parameter ( self , dt ) : <EOL> return dt <EOL> def parameter_to_datetime ( self , p ) : <EOL> return p <EOL> def moving_start ( self , now ) : <EOL> return now - timedelta ( hours = self . hours_back ) <EOL> def moving_stop ( self , now ) : <EOL> return now + timedelta ( hours = self . hours_forward ) <EOL> def finite_datetimes ( self , finite_start , finite_stop ) : <EOL> """<STR_LIT>""" <EOL> datehour_start = datetime ( finite_start . year , finite_start . month , finite_start . day , finite_start . hour ) <EOL> datehours = [ ] <EOL> for i in itertools . count ( ) : <EOL> t = datehour_start + timedelta ( hours = i ) <EOL> if t >= finite_stop : <EOL> return datehours <EOL> if t >= finite_start : <EOL> datehours . append ( t ) <EOL> def _format_datetime ( self , dt ) : <EOL> return luigi . DateHourParameter ( ) . serialize ( dt ) <EOL> def _constrain_glob ( glob , paths , limit = <NUM_LIT:5> ) : <EOL> """<STR_LIT>""" <EOL> def digit_set_wildcard ( chars ) : <EOL> """<STR_LIT>""" <EOL> chars = sorted ( chars ) <EOL> if len ( chars ) > <NUM_LIT:1> and ord ( chars [ - <NUM_LIT:1> ] ) - ord ( chars [ <NUM_LIT:0> ] ) == len ( chars ) - <NUM_LIT:1> : <EOL> return '<STR_LIT>' % ( chars [ <NUM_LIT:0> ] , chars [ - <NUM_LIT:1> ] ) <EOL> else : <EOL> return '<STR_LIT>' % '<STR_LIT>' . join ( chars ) <EOL> current = { glob : paths } <EOL> while True : <EOL> pos = list ( current . keys ( ) ) [ <NUM_LIT:0> ] . find ( '<STR_LIT>' ) <EOL> if pos == - <NUM_LIT:1> : <EOL> return list ( current . keys ( ) ) <EOL> char_sets = { } <EOL> for g , p in six . iteritems ( current ) : <EOL> char_sets [ g ] = sorted ( set ( path [ pos ] for path in p ) ) <EOL> if sum ( len ( s ) for s in char_sets . values ( ) ) > limit : <EOL> return [ g . replace ( '<STR_LIT>' , digit_set_wildcard ( char_sets [ g ] ) , <NUM_LIT:1> ) for g in current ] <EOL> for g , s in six . iteritems ( char_sets ) : <EOL> for c in s : <EOL> new_glob = g . replace ( '<STR_LIT>' , c , <NUM_LIT:1> ) <EOL> new_paths = list ( filter ( lambda p : p [ pos ] == c , current [ g ] ) ) <EOL> current [ new_glob ] = new_paths <EOL> del current [ g ] <EOL> def most_common ( items ) : <EOL> """<STR_LIT>""" <EOL> counts = { } <EOL> for i in items : <EOL> counts . setdefault ( i , <NUM_LIT:0> ) <EOL> counts [ i ] += <NUM_LIT:1> <EOL> return max ( six . iteritems ( counts ) , key = operator . itemgetter ( <NUM_LIT:1> ) ) <EOL> def _get_per_location_glob ( tasks , outputs , regexes ) : <EOL> """<STR_LIT>""" <EOL> paths = [ o . path for o in outputs ] <EOL> matches = [ r . search ( p ) for r , p in zip ( regexes , paths ) ] <EOL> for m , p , t in zip ( matches , paths , tasks ) : <EOL> if m is None : <EOL> raise NotImplementedError ( "<STR_LIT>" % ( p , t ) ) <EOL> n_groups = len ( matches [ <NUM_LIT:0> ] . groups ( ) ) <EOL> positions = [ most_common ( ( m . start ( i ) , m . end ( i ) ) for m in matches ) [ <NUM_LIT:0> ] for i in range ( <NUM_LIT:1> , n_groups + <NUM_LIT:1> ) ] <EOL> glob = list ( paths [ <NUM_LIT:0> ] ) <EOL> for start , end in positions : <EOL> glob = glob [ : start ] + [ '<STR_LIT>' ] * ( end - start ) + glob [ end : ] <EOL> return '<STR_LIT>' . join ( glob ) . rsplit ( '<STR_LIT:/>' , <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> def _get_filesystems_and_globs ( datetime_to_task , datetime_to_re ) : <EOL> """<STR_LIT>""" <EOL> sample_datetimes = [ datetime ( y , m , d , h ) for y in range ( <NUM_LIT> , <NUM_LIT> , <NUM_LIT:10> ) for m in range ( <NUM_LIT:1> , <NUM_LIT:4> ) for d in range ( <NUM_LIT:5> , <NUM_LIT:8> ) for h in range ( <NUM_LIT> , <NUM_LIT> ) ] <EOL> regexes = [ re . compile ( datetime_to_re ( d ) ) for d in sample_datetimes ] <EOL> sample_tasks = [ datetime_to_task ( d ) for d in sample_datetimes ] <EOL> sample_outputs = [ flatten_output ( t ) for t in sample_tasks ] <EOL> for o , t in zip ( sample_outputs , sample_tasks ) : <EOL> if len ( o ) != len ( sample_outputs [ <NUM_LIT:0> ] ) : <EOL> raise NotImplementedError ( "<STR_LIT>" % ( o , t , sample_outputs [ <NUM_LIT:0> ] , sample_tasks [ <NUM_LIT:0> ] ) ) <EOL> for target in o : <EOL> if not isinstance ( target , FileSystemTarget ) : <EOL> raise NotImplementedError ( "<STR_LIT>" % ( target , t ) ) <EOL> for o in zip ( * sample_outputs ) : <EOL> glob = _get_per_location_glob ( sample_tasks , o , regexes ) <EOL> yield o [ <NUM_LIT:0> ] . fs , glob <EOL> def _list_existing ( filesystem , glob , paths ) : <EOL> """<STR_LIT>""" <EOL> globs = _constrain_glob ( glob , paths ) <EOL> time_start = time . time ( ) <EOL> listing = [ ] <EOL> for g in sorted ( globs ) : <EOL> logger . debug ( '<STR_LIT>' , g ) <EOL> if filesystem . exists ( g ) : <EOL> listing . extend ( filesystem . listdir ( g ) ) <EOL> logger . debug ( '<STR_LIT>' , <EOL> len ( globs ) , filesystem . __class__ . __name__ , time . time ( ) - time_start , len ( listing ) ) <EOL> return set ( listing ) <EOL> def infer_bulk_complete_from_fs ( datetimes , datetime_to_task , datetime_to_re ) : <EOL> """<STR_LIT>""" <EOL> filesystems_and_globs_by_location = _get_filesystems_and_globs ( datetime_to_task , datetime_to_re ) <EOL> paths_by_datetime = [ [ o . path for o in flatten_output ( datetime_to_task ( d ) ) ] for d in datetimes ] <EOL> listing = set ( ) <EOL> for ( f , g ) , p in zip ( filesystems_and_globs_by_location , zip ( * paths_by_datetime ) ) : <EOL> listing |= _list_existing ( f , g , p ) <EOL> missing_datetimes = [ ] <EOL> for d , p in zip ( datetimes , paths_by_datetime ) : <EOL> if not set ( p ) <= listing : <EOL> missing_datetimes . append ( d ) <EOL> return missing_datetimes <EOL> class RangeDaily ( RangeDailyBase ) : <EOL> """<STR_LIT>""" <EOL> def missing_datetimes ( self , finite_datetimes ) : <EOL> try : <EOL> complete_parameters = self . of_cls . bulk_complete ( map ( self . datetime_to_parameter , finite_datetimes ) ) <EOL> return set ( finite_datetimes ) - set ( map ( self . parameter_to_datetime , complete_parameters ) ) <EOL> except NotImplementedError : <EOL> return infer_bulk_complete_from_fs ( <EOL> finite_datetimes , <EOL> lambda d : self . _instantiate_task_cls ( self . datetime_to_parameter ( d ) ) , <EOL> lambda d : d . strftime ( '<STR_LIT>' ) ) <EOL> class RangeHourly ( RangeHourlyBase ) : <EOL> """<STR_LIT>""" <EOL> def missing_datetimes ( self , finite_datetimes ) : <EOL> try : <EOL> complete_parameters = self . of_cls . bulk_complete ( list ( map ( self . datetime_to_parameter , finite_datetimes ) ) ) <EOL> return set ( finite_datetimes ) - set ( map ( self . parameter_to_datetime , complete_parameters ) ) <EOL> except NotImplementedError : <EOL> return infer_bulk_complete_from_fs ( <EOL> finite_datetimes , <EOL> lambda d : self . _instantiate_task_cls ( self . datetime_to_parameter ( d ) ) , <EOL> lambda d : d . strftime ( '<STR_LIT>' ) ) </s>
