<s> from collections import deque <EOL> import logging <EOL> from operator import itemgetter <EOL> from twisted . internet import protocol <EOL> from twisted . python import failure <EOL> from twimp import vecbuf <EOL> LOG_CATEGORY = '<STR_LIT>' <EOL> import twimp . log <EOL> log = twimp . log . get_logger ( LOG_CATEGORY ) <EOL> def ms_time ( t ) : <EOL> "<STR_LIT>" <EOL> return int ( t * <NUM_LIT:1000> ) <EOL> def ms_time_wrapped ( t ) : <EOL> """<STR_LIT>""" <EOL> return ms_time ( t ) % <NUM_LIT> <EOL> class GeneratorWrapperProtocol ( protocol . Protocol ) : <EOL> def __init__ ( self , proto = None ) : <EOL> self . _buf = vecbuf . VecBuf ( ) <EOL> self . _toread = <NUM_LIT:0> <EOL> self . _handler_changed = False <EOL> self . _handler = proto <EOL> if proto : <EOL> self . _toread = self . _handler . send ( None ) <EOL> def init_handler ( self , proto = None , do_init = True ) : <EOL> if proto : <EOL> if self . _handler : <EOL> self . _handler_changed = True <EOL> self . _handler = proto <EOL> if do_init : <EOL> self . _toread = self . _handler . send ( None ) <EOL> def connectionLost ( self , reason = protocol . connectionDone ) : <EOL> self . _buf . read ( len ( self . _buf ) ) <EOL> self . _handler = None <EOL> self . _toread = None <EOL> def dataReceived ( self , data ) : <EOL> self . _buf . write ( data ) <EOL> while <NUM_LIT:1> : <EOL> try : <EOL> while len ( self . _buf ) >= self . _toread : <EOL> self . _toread = self . _handler . send ( self . _buf ) <EOL> break <EOL> except StopIteration : <EOL> if self . _handler_changed : <EOL> self . _handler_changed = False <EOL> else : <EOL> self . _toread = None <EOL> self . _handler = None <EOL> break <EOL> except : <EOL> f = failure . Failure ( ) <EOL> log . info ( f . value , exc_info = log . isEnabledFor ( logging . DEBUG ) ) <EOL> return f <EOL> class FrameSorter ( object ) : <EOL> def __init__ ( self , callback , keys ) : <EOL> self . _max = len ( keys ) <EOL> assert self . _max > <NUM_LIT:1> , '<STR_LIT>' <EOL> self . _queues = dict ( ( k , [ None , deque ( ) , k ] ) for k in keys ) <EOL> self . _active = <NUM_LIT:0> <EOL> self . _callback = callback <EOL> self . _key_getter = itemgetter ( <NUM_LIT:0> ) <EOL> self . _grpos_getter = itemgetter ( <NUM_LIT:0> ) <EOL> def add ( self , key , grpos , data ) : <EOL> sq = self . _queues [ key ] <EOL> if not sq [ <NUM_LIT:1> ] : <EOL> self . _active += <NUM_LIT:1> <EOL> sq [ <NUM_LIT:0> ] = grpos <EOL> sq [ <NUM_LIT:1> ] . append ( ( grpos , data ) ) <EOL> self . _check_send ( ) <EOL> def _check_send ( self ) : <EOL> if self . _active < self . _max : <EOL> return <EOL> while self . _active >= self . _max : <EOL> first_q = min ( self . _queues . itervalues ( ) , key = self . _key_getter ) <EOL> key = first_q [ <NUM_LIT:2> ] <EOL> second_q = min ( ( q for ( k , q ) in self . _queues . iteritems ( ) <EOL> if k != key ) , key = self . _key_getter ) <EOL> switch_grpos = second_q [ <NUM_LIT:0> ] <EOL> frame_queue = first_q [ <NUM_LIT:1> ] <EOL> while frame_queue : <EOL> if frame_queue [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] > switch_grpos : <EOL> break <EOL> ( grpos , frame ) = frame_queue . popleft ( ) <EOL> self . _callback ( grpos , key , frame ) <EOL> if not frame_queue : <EOL> self . _active -= <NUM_LIT:1> <EOL> else : <EOL> first_q [ <NUM_LIT:0> ] = frame_queue [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> def flush ( self ) : <EOL> sorted_frames = sorted ( ( ( grpos , key , frame ) <EOL> for ( min_grpos , frame_queue , key ) <EOL> in self . _queues . values ( ) <EOL> for ( grpos , frame ) in frame_queue ) , <EOL> key = self . _grpos_getter ) <EOL> self . _active = <NUM_LIT:0> <EOL> self . _queues = dict ( ( k , [ None , deque ( ) , k ] ) for k in self . _queues ) <EOL> for f in sorted_frames : <EOL> self . _callback ( * f ) </s>
