<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import <EOL> from google . cloud . dataflow . transforms import timeutil <EOL> from google . cloud . dataflow . transforms . timeutil import Duration <EOL> from google . cloud . dataflow . transforms . timeutil import MAX_TIMESTAMP <EOL> from google . cloud . dataflow . transforms . timeutil import MIN_TIMESTAMP <EOL> from google . cloud . dataflow . transforms . timeutil import Timestamp <EOL> class OutputTimeFn ( object ) : <EOL> """<STR_LIT>""" <EOL> OUTPUT_AT_EOW = '<STR_LIT>' <EOL> OUTPUT_AT_EARLIEST = '<STR_LIT>' <EOL> OUTPUT_AT_LATEST = '<STR_LIT>' <EOL> OUTPUT_AT_EARLIEST_TRANSFORMED = '<STR_LIT>' <EOL> @ staticmethod <EOL> def get_impl ( output_time_fn , window_fn ) : <EOL> if output_time_fn == OutputTimeFn . OUTPUT_AT_EOW : <EOL> return timeutil . OutputAtEndOfWindowImpl ( ) <EOL> elif output_time_fn == OutputTimeFn . OUTPUT_AT_EARLIEST : <EOL> return timeutil . OutputAtEarliestInputTimestampImpl ( ) <EOL> elif output_time_fn == OutputTimeFn . OUTPUT_AT_LATEST : <EOL> return timeutil . OutputAtLatestInputTimestampImpl ( ) <EOL> elif output_time_fn == OutputTimeFn . OUTPUT_AT_EARLIEST_TRANSFORMED : <EOL> return timeutil . OutputAtEarliestTransformedInputTimestampImpl ( window_fn ) <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' % output_time_fn ) <EOL> class WindowFn ( object ) : <EOL> """<STR_LIT>""" <EOL> class AssignContext ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , timestamp , element = None , existing_windows = None ) : <EOL> self . timestamp = Timestamp . of ( timestamp ) <EOL> self . element = element <EOL> self . existing_windows = existing_windows <EOL> def assign ( self , assign_context ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> class MergeContext ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , windows ) : <EOL> self . windows = list ( windows ) <EOL> def merge ( self , to_be_merged , merge_result ) : <EOL> raise NotImplementedError <EOL> def merge ( self , merge_context ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def get_transformed_output_time ( self , window , input_timestamp ) : <EOL> """<STR_LIT>""" <EOL> return input_timestamp <EOL> class BoundedWindow ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , end ) : <EOL> self . end = Timestamp . of ( end ) <EOL> def __cmp__ ( self , other ) : <EOL> return cmp ( self . end , other . end ) or cmp ( hash ( self ) , hash ( other ) ) <EOL> def __eq__ ( self , other ) : <EOL> raise NotImplementedError <EOL> def __hash__ ( self ) : <EOL> return hash ( self . end ) <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % float ( self . end ) <EOL> class IntervalWindow ( BoundedWindow ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , start , end ) : <EOL> super ( IntervalWindow , self ) . __init__ ( end ) <EOL> self . start = Timestamp . of ( start ) <EOL> def __hash__ ( self ) : <EOL> return hash ( ( self . start , self . end ) ) <EOL> def __eq__ ( self , other ) : <EOL> return self . start == other . start and self . end == other . end <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( float ( self . start ) , float ( self . end ) ) <EOL> def intersects ( self , other ) : <EOL> return other . start < self . end or self . start < other . end <EOL> def union ( self , other ) : <EOL> return IntervalWindow ( <EOL> min ( self . start , other . start ) , max ( self . end , other . end ) ) <EOL> class WindowedValue ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , value , timestamp , windows ) : <EOL> self . value = value <EOL> self . timestamp = Timestamp . of ( timestamp ) <EOL> self . windows = windows <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( <EOL> repr ( self . value ) , <EOL> '<STR_LIT>' if self . timestamp == MIN_TIMESTAMP else <EOL> '<STR_LIT>' if self . timestamp == MAX_TIMESTAMP else <EOL> float ( self . timestamp ) , <EOL> self . windows ) <EOL> def __hash__ ( self ) : <EOL> return hash ( ( self . value , self . timestamp , self . windows ) ) <EOL> def __eq__ ( self , other ) : <EOL> return ( type ( self ) == type ( other ) <EOL> and self . value == other . value <EOL> and self . timestamp == other . timestamp <EOL> and self . windows == other . windows ) <EOL> def with_value ( self , new_value ) : <EOL> return WindowedValue ( new_value , self . timestamp , self . windows ) <EOL> class TimestampedValue ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , value , timestamp ) : <EOL> self . value = value <EOL> self . timestamp = Timestamp . of ( timestamp ) <EOL> class GlobalWindow ( BoundedWindow ) : <EOL> """<STR_LIT>""" <EOL> _instance = None <EOL> def __new__ ( cls ) : <EOL> if cls . _instance is None : <EOL> cls . _instance = super ( GlobalWindow , cls ) . __new__ ( cls ) <EOL> return cls . _instance <EOL> def __init__ ( self ) : <EOL> super ( GlobalWindow , self ) . __init__ ( MAX_TIMESTAMP ) <EOL> self . start = MIN_TIMESTAMP <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' <EOL> def __hash__ ( self ) : <EOL> return hash ( type ( self ) ) <EOL> def __eq__ ( self , other ) : <EOL> return self is other or type ( self ) is type ( other ) <EOL> class GlobalWindows ( WindowFn ) : <EOL> """<STR_LIT>""" <EOL> @ classmethod <EOL> def WindowedValue ( cls , value , timestamp = MIN_TIMESTAMP ) : <EOL> return WindowedValue ( value , timestamp , [ GlobalWindow ( ) ] ) <EOL> def assign ( self , assign_context ) : <EOL> return [ GlobalWindow ( ) ] <EOL> def merge ( self , merge_context ) : <EOL> pass <EOL> def __hash__ ( self ) : <EOL> return hash ( type ( self ) ) <EOL> def __eq__ ( self , other ) : <EOL> return self is other or type ( self ) is type ( other ) <EOL> def __ne__ ( self , other ) : <EOL> return not self == other <EOL> class FixedWindows ( WindowFn ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , size , offset = <NUM_LIT:0> ) : <EOL> if size <= <NUM_LIT:0> : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> self . size = Duration . of ( size ) <EOL> self . offset = Timestamp . of ( offset ) % self . size <EOL> def assign ( self , context ) : <EOL> timestamp = context . timestamp <EOL> start = timestamp - ( timestamp - self . offset ) % self . size <EOL> return [ IntervalWindow ( start , start + self . size ) ] <EOL> def merge ( self , merge_context ) : <EOL> pass <EOL> class SlidingWindows ( WindowFn ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , size , period , offset = <NUM_LIT:0> ) : <EOL> if size <= <NUM_LIT:0> : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> self . size = Duration . of ( size ) <EOL> self . period = Duration . of ( period ) <EOL> self . offset = Timestamp . of ( offset ) % size <EOL> def assign ( self , context ) : <EOL> timestamp = context . timestamp <EOL> start = timestamp - ( timestamp - self . offset ) % self . period <EOL> return [ IntervalWindow ( Timestamp . of ( s ) , Timestamp . of ( s ) + self . size ) <EOL> for s in range ( start , start - self . size , - self . period ) ] <EOL> def merge ( self , merge_context ) : <EOL> pass <EOL> class Sessions ( WindowFn ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , gap_size ) : <EOL> if gap_size <= <NUM_LIT:0> : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> self . gap_size = Duration . of ( gap_size ) <EOL> def assign ( self , context ) : <EOL> timestamp = context . timestamp <EOL> return [ IntervalWindow ( timestamp , timestamp + self . gap_size ) ] <EOL> def merge ( self , merge_context ) : <EOL> to_merge = [ ] <EOL> for w in sorted ( merge_context . windows , key = lambda w : w . start ) : <EOL> if to_merge : <EOL> if end > w . start : <EOL> to_merge . append ( w ) <EOL> if w . end > end : <EOL> end = w . end <EOL> else : <EOL> if len ( to_merge ) > <NUM_LIT:1> : <EOL> merge_context . merge ( to_merge , <EOL> IntervalWindow ( to_merge [ <NUM_LIT:0> ] . start , end ) ) <EOL> to_merge = [ w ] <EOL> end = w . end <EOL> else : <EOL> to_merge = [ w ] <EOL> end = w . end <EOL> if len ( to_merge ) > <NUM_LIT:1> : <EOL> merge_context . merge ( to_merge , IntervalWindow ( to_merge [ <NUM_LIT:0> ] . start , end ) ) </s>
