<s> from __future__ import absolute_import <EOL> import time <EOL> import heapq <EOL> import calendar <EOL> def _merge ( sorted_iterables ) : <EOL> """<STR_LIT>""" <EOL> heap = [ ] <EOL> for iterable in sorted_iterables : <EOL> iterator = iter ( iterable ) <EOL> for value in iterator : <EOL> heapq . heappush ( heap , ( value , iterator ) ) <EOL> break <EOL> while heap : <EOL> value , iterator = heapq . heappop ( heap ) <EOL> yield value <EOL> for value in iterator : <EOL> heapq . heappush ( heap , ( value , iterator ) ) <EOL> break <EOL> def _dedup ( sorted_iterable ) : <EOL> """<STR_LIT>""" <EOL> iterator = iter ( sorted_iterable ) <EOL> for previous in iterator : <EOL> yield previous <EOL> break <EOL> for value in iterator : <EOL> if value == previous : <EOL> continue <EOL> yield value <EOL> previous = value <EOL> class _CmpMixin ( object ) : <EOL> def __eq__ ( self , other ) : <EOL> return self is other <EOL> def __ne__ ( self , other ) : <EOL> result = self . __eq__ ( other ) <EOL> if result is NotImplemented : <EOL> return NotImplemented <EOL> return not result <EOL> min_time_tuple = [ <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , - <NUM_LIT:1> ] <EOL> def normalized_time_tuple ( time_tuple ) : <EOL> timestamp = calendar . timegm ( list ( time_tuple [ : <NUM_LIT:6> ] ) + [ <NUM_LIT:0> , <NUM_LIT:0> , - <NUM_LIT:1> ] ) <EOL> return time . gmtime ( timestamp ) [ : <NUM_LIT:8> ] + ( - <NUM_LIT:1> , ) <EOL> def is_valid_time_tuple ( time_tuple ) : <EOL> normalized = normalized_time_tuple ( time_tuple ) <EOL> return tuple ( normalized [ : <NUM_LIT:6> ] ) == tuple ( time_tuple [ : <NUM_LIT:6> ] ) <EOL> class _Unit ( object ) : <EOL> def __init__ ( self , min , max , time_tuple_index , aliases = { } ) : <EOL> self . min = min <EOL> self . max = max <EOL> self . time_tuple_index = time_tuple_index <EOL> self . aliases = { } <EOL> for value , alias_list in aliases . iteritems ( ) : <EOL> for alias in alias_list : <EOL> self . aliases [ alias ] = value <EOL> def map ( self , value ) : <EOL> if not isinstance ( value , basestring ) : <EOL> return value <EOL> normalized = value . strip ( ) . lower ( ) <EOL> if normalized not in self . aliases : <EOL> raise ValueError ( "<STR_LIT>" . format ( value ) ) <EOL> return self . aliases [ normalized ] <EOL> def _resolve_one ( self , ( first , last , step ) ) : <EOL> first = self . min if first is None else self . map ( first ) <EOL> last = self . max if last is None else self . map ( last ) <EOL> if last < first : <EOL> raise ValueError ( "<STR_LIT>" . format ( first , last ) ) <EOL> if not self . min <= first <= last <= self . max : <EOL> if first == last : <EOL> raise ValueError ( "<STR_LIT>" . format ( first , self . min , self . max ) ) <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" . format ( first , last , self . min , self . max ) ) <EOL> if not <NUM_LIT:1> <= step <= self . max - self . min + <NUM_LIT:1> : <EOL> raise ValueError ( "<STR_LIT>" . format ( step , self . max - self . min + <NUM_LIT:1> ) ) <EOL> return first , last , step <EOL> def resolve ( self , ranges ) : <EOL> return _Resolved ( self , map ( self . _resolve_one , ranges ) ) <EOL> def iter_range ( self , time_tuple , ( first , last , step ) ) : <EOL> time_tuple = list ( time_tuple ) <EOL> index = self . time_tuple_index <EOL> if time_tuple [ index ] < first : <EOL> time_tuple [ index ] = first <EOL> time_tuple [ index + <NUM_LIT:1> : ] = min_time_tuple [ index + <NUM_LIT:1> : ] <EOL> else : <EOL> modulo = ( time_tuple [ index ] - first ) % step <EOL> if modulo > <NUM_LIT:0> : <EOL> time_tuple [ index ] = time_tuple [ index ] - modulo + step <EOL> time_tuple [ index + <NUM_LIT:1> : ] = min_time_tuple [ index + <NUM_LIT:1> : ] <EOL> while time_tuple [ index ] <= last and is_valid_time_tuple ( time_tuple ) : <EOL> yield normalized_time_tuple ( time_tuple ) <EOL> time_tuple [ index ] += step <EOL> time_tuple [ index + <NUM_LIT:1> : ] = min_time_tuple [ index + <NUM_LIT:1> : ] <EOL> class _MonthUnit ( _Unit ) : <EOL> def iter_range ( self , time_tuple , range ) : <EOL> time_tuple = list ( time_tuple ) <EOL> while True : <EOL> for result in _Unit . iter_range ( self , time_tuple , range ) : <EOL> yield result <EOL> time_tuple [ <NUM_LIT:0> ] += <NUM_LIT:1> <EOL> time_tuple [ <NUM_LIT:1> : ] = min_time_tuple [ <NUM_LIT:1> : ] <EOL> class _WeekdayUnit ( _Unit ) : <EOL> def iter_range ( self , time_tuple , ( first , last , step ) ) : <EOL> time_tuple = list ( time_tuple ) <EOL> while True : <EOL> if not is_valid_time_tuple ( time_tuple ) : <EOL> break <EOL> weekday = ( calendar . weekday ( * time_tuple [ : <NUM_LIT:3> ] ) + <NUM_LIT:1> ) % <NUM_LIT:7> <EOL> if weekday > last : <EOL> time_tuple [ <NUM_LIT:2> ] += <NUM_LIT:7> - weekday <EOL> time_tuple [ <NUM_LIT:3> : ] = min_time_tuple [ <NUM_LIT:3> : ] <EOL> continue <EOL> if weekday < first : <EOL> time_tuple [ <NUM_LIT:2> ] += first - weekday <EOL> time_tuple [ <NUM_LIT:3> : ] = min_time_tuple [ <NUM_LIT:3> : ] <EOL> continue <EOL> yield normalized_time_tuple ( time_tuple ) <EOL> time_tuple [ <NUM_LIT:2> ] += <NUM_LIT:1> <EOL> time_tuple [ <NUM_LIT:3> : ] = min_time_tuple [ <NUM_LIT:3> : ] <EOL> class _Ranges ( object ) : <EOL> def __init__ ( self , ranges ) : <EOL> self . _ranges = list ( ranges ) <EOL> def or_range ( self , start , end , step = <NUM_LIT:1> ) : <EOL> return _Ranges ( self . _ranges + [ ( start , end , step ) ] ) <EOL> def or_value ( self , value ) : <EOL> return _Ranges ( self . _ranges + [ ( value , value , <NUM_LIT:1> ) ] ) <EOL> def or_any ( self , step = <NUM_LIT:1> ) : <EOL> return _Ranges ( self . _ranges + [ ( None , None , step ) ] ) <EOL> def __iter__ ( self ) : <EOL> return iter ( tuple ( self . _ranges ) ) <EOL> class _Resolved ( _CmpMixin ) : <EOL> def __init__ ( self , unit , ranges ) : <EOL> self . _unit = unit <EOL> self . _ranges = tuple ( ranges ) <EOL> def iter_from ( self , value ) : <EOL> return _dedup ( _merge ( self . _unit . iter_range ( value , x ) for x in self . _ranges ) ) <EOL> def __eq__ ( self , other ) : <EOL> if not isinstance ( other , self . __class__ ) : <EOL> return NotImplemented <EOL> return set ( self . _ranges ) == set ( other . _ranges ) <EOL> class _ResolvedOr ( _CmpMixin ) : <EOL> def __init__ ( self , left , right ) : <EOL> self . _left = left <EOL> self . _right = right <EOL> def iter_from ( self , value ) : <EOL> left = self . _left . iter_from ( value ) <EOL> right = self . _right . iter_from ( value ) <EOL> return _dedup ( _merge ( [ left , right ] ) ) <EOL> def __eq__ ( self , other ) : <EOL> if not isinstance ( other , self . __class__ ) : <EOL> return NotImplemented <EOL> if self . _left == other . _left and self . _right == other . _right : <EOL> return True <EOL> if self . _left == other . _right and self . _right == other . _left : <EOL> return True <EOL> return False <EOL> def empty ( ) : <EOL> return _Ranges ( [ ] ) <EOL> def range ( start , end , step = <NUM_LIT:1> ) : <EOL> return _Ranges ( [ ( start , end , step ) ] ) <EOL> def value ( value ) : <EOL> return _Ranges ( [ ( value , value , <NUM_LIT:1> ) ] ) <EOL> def any ( step = <NUM_LIT:1> ) : <EOL> return _Ranges ( [ ( None , None , step ) ] ) <EOL> class Cron ( _CmpMixin ) : <EOL> _units = { <EOL> "<STR_LIT>" : _Unit ( <NUM_LIT:0> , <NUM_LIT> , <NUM_LIT:4> ) , <EOL> "<STR_LIT>" : _Unit ( <NUM_LIT:0> , <NUM_LIT> , <NUM_LIT:3> ) , <EOL> "<STR_LIT>" : _Unit ( <NUM_LIT:1> , <NUM_LIT> , <NUM_LIT:2> ) , <EOL> "<STR_LIT>" : _WeekdayUnit ( <NUM_LIT:0> , <NUM_LIT:6> , <NUM_LIT:2> , aliases = { <EOL> <NUM_LIT:0> : [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> <NUM_LIT:1> : [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> <NUM_LIT:2> : [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> <NUM_LIT:3> : [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> <NUM_LIT:4> : [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> <NUM_LIT:5> : [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> <NUM_LIT:6> : [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> } ) , <EOL> "<STR_LIT>" : _MonthUnit ( <NUM_LIT:1> , <NUM_LIT:12> , <NUM_LIT:1> , aliases = { <EOL> <NUM_LIT:1> : [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> <NUM_LIT:2> : [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> <NUM_LIT:3> : [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> <NUM_LIT:4> : [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> <NUM_LIT:5> : [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> <NUM_LIT:6> : [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> <NUM_LIT:7> : [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> <NUM_LIT:8> : [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> <NUM_LIT:9> : [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> <NUM_LIT:10> : [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> <NUM_LIT:11> : [ "<STR_LIT>" , "<STR_LIT>" ] , <EOL> <NUM_LIT:12> : [ "<STR_LIT>" , "<STR_LIT>" ] <EOL> } ) <EOL> } <EOL> def __init__ ( self , * args , ** keys ) : <EOL> self . _spec = { } <EOL> max_priority = min ( x . time_tuple_index for x in self . _units . values ( ) ) <EOL> for key , arg in dict ( * args , ** keys ) . iteritems ( ) : <EOL> if key not in self . _units : <EOL> raise TypeError ( "<STR_LIT>" . format ( key ) ) <EOL> unit = self . _units [ key ] <EOL> max_priority = max ( max_priority , unit . time_tuple_index ) <EOL> rangeobj = self . _coerce ( arg ) <EOL> self . _spec [ key ] = unit . resolve ( rangeobj ) <EOL> for key , unit in self . _units . iteritems ( ) : <EOL> if key in self . _spec : <EOL> continue <EOL> if max_priority >= unit . time_tuple_index : <EOL> self . _spec [ key ] = unit . resolve ( any ( ) ) <EOL> else : <EOL> self . _spec [ key ] = unit . resolve ( value ( unit . min ) ) <EOL> if self . _is_any ( "<STR_LIT>" ) : <EOL> self . _spec [ "<STR_LIT>" ] = self . _units [ "<STR_LIT>" ] . resolve ( empty ( ) ) <EOL> elif self . _is_any ( "<STR_LIT>" ) : <EOL> self . _spec [ "<STR_LIT>" ] = self . _units [ "<STR_LIT>" ] . resolve ( empty ( ) ) <EOL> self . _spec [ "<STR_LIT>" ] = _ResolvedOr ( self . _spec . pop ( "<STR_LIT>" ) , self . _spec . pop ( "<STR_LIT>" ) ) <EOL> def _is_any ( self , unit_name ) : <EOL> unit = self . _units [ unit_name ] <EOL> return unit . resolve ( any ( ) ) == self . _spec [ unit_name ] <EOL> def _coerce ( self , arg ) : <EOL> if isinstance ( arg , _Ranges ) : <EOL> return arg <EOL> return value ( arg ) <EOL> def _iter ( self , time_tuple , order ) : <EOL> if not order : <EOL> yield time_tuple <EOL> return <EOL> resolved = self . _spec [ order [ <NUM_LIT:0> ] ] <EOL> for new_tuple in resolved . iter_from ( time_tuple ) : <EOL> for result in self . _iter ( new_tuple , order [ <NUM_LIT:1> : ] ) : <EOL> yield normalized_time_tuple ( result ) <EOL> def iter ( self , time_tuple = None ) : <EOL> if time_tuple is None : <EOL> time_tuple = time . localtime ( ) <EOL> time_tuple = normalized_time_tuple ( time_tuple ) <EOL> order = sorted ( self . _spec , key = lambda x : self . _units [ x ] . time_tuple_index ) <EOL> return self . _iter ( time_tuple , order ) <EOL> def next ( self , time_tuple = None ) : <EOL> if time_tuple is None : <EOL> time_tuple = time . localtime ( ) <EOL> time_tuple = normalized_time_tuple ( time_tuple ) <EOL> for result in self . iter ( time_tuple ) : <EOL> if result [ : <NUM_LIT:6> ] != time_tuple [ : <NUM_LIT:6> ] : <EOL> return result <EOL> return None <EOL> def matches ( self , time_tuple ) : <EOL> if time_tuple is None : <EOL> time_tuple = time . localtime ( ) <EOL> time_tuple = normalized_time_tuple ( time_tuple ) <EOL> for result in self . iter ( time_tuple ) : <EOL> return result [ : <NUM_LIT:6> ] == time_tuple [ : <NUM_LIT:6> ] <EOL> return False <EOL> def __eq__ ( self , other ) : <EOL> if not isinstance ( other , self . __class__ ) : <EOL> return NotImplemented <EOL> return self . _spec == other . _spec <EOL> class CronParser ( object ) : <EOL> _units = [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> def parse ( self , string ) : <EOL> fields = string . split ( ) <EOL> if len ( fields ) != <NUM_LIT:5> : <EOL> raise ValueError ( "<STR_LIT>" . format ( string ) ) <EOL> results = { } <EOL> for unit , field in zip ( self . _units , fields ) : <EOL> ranges = empty ( ) <EOL> for part in [ x . strip ( ) for x in field . split ( "<STR_LIT:U+002C>" ) ] : <EOL> if not part : <EOL> raise ValueError ( "<STR_LIT>" + unit + "<STR_LIT>" ) <EOL> if "<STR_LIT:/>" not in part : <EOL> step = <NUM_LIT:1> <EOL> else : <EOL> part , step = part . rsplit ( "<STR_LIT:/>" , <NUM_LIT:1> ) <EOL> if not step . isdigit ( ) : <EOL> raise ValueError ( "<STR_LIT>" . format ( step ) ) <EOL> step = int ( step ) <EOL> if part . strip ( ) == "<STR_LIT:*>" : <EOL> first = None <EOL> last = None <EOL> else : <EOL> if "<STR_LIT:->" in part : <EOL> first , last = part . rsplit ( "<STR_LIT:->" , <NUM_LIT:1> ) <EOL> else : <EOL> first = part <EOL> last = part <EOL> if first . isdigit ( ) : <EOL> first = int ( first ) <EOL> if last . isdigit ( ) : <EOL> last = int ( last ) <EOL> ranges = ranges . or_range ( first , last , step ) <EOL> results [ unit ] = ranges <EOL> return Cron ( results ) <EOL> parse = CronParser ( ) . parse </s>
