#NEWLINE# def dispatchToGenerate( self ) : #NEWLINE# #INDENT# processorIndex = self . config [ 'processor' ] [ 'index' ] #NEWLINE# phraseProcessorType = ( self . config [ 'processor' ] [ 'type' ] + '__phrase' ) #NEWLINE# nextPhraseIndex = 0 #NEWLINE# if ( self . config [ 'processingStartIndex' ] != None ) : #NEWLINE# #INDENT# nextPhraseIndex = self . config [ 'processingStartIndex' ] #NEWLINE# #UNINDENT# endPhraseIndex = ( - 1 ) #NEWLINE# if ( self . config [ 'processingEndIndex' ] != None ) : #NEWLINE# #INDENT# endPhraseIndex = self . config [ 'processingEndIndex' ] #NEWLINE# #UNINDENT# if ( ( endPhraseIndex != ( - 1 ) ) and ( self . processingPageSize > ( endPhraseIndex - nextPhraseIndex ) ) ) : #NEWLINE# #INDENT# self . processingPageSize = ( ( endPhraseIndex - nextPhraseIndex ) + 1 ) #NEWLINE# #UNINDENT# while True : #NEWLINE# #INDENT# phrases = self . esClient . search ( index = processorIndex , doc_type = phraseProcessorType , body = { #NEWLINE# #INDENT# 'from' : nextPhraseIndex , #NEWLINE# 'size' : self . processingPageSize , #NEWLINE# 'query' : { #NEWLINE# #INDENT# 'match_all' : { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } , #NEWLINE# #UNINDENT# } , #NEWLINE# 'sort' : [ { #NEWLINE# #INDENT# 'phrase__not_analyzed' : { #NEWLINE# #INDENT# 'order' : 'asc' , #NEWLINE# #UNINDENT# } , #NEWLINE# #UNINDENT# } ] , #NEWLINE# #UNINDENT# } , fields = [ '_id' ] ) #NEWLINE# if ( len ( phrases [ 'hits' ] [ 'hits' ] ) == 0 ) : #NEWLINE# #INDENT# break #NEWLINE# #UNINDENT# self . totalPhrasesDispatched += len ( phrases [ 'hits' ] [ 'hits' ] ) #NEWLINE# floatPrecision = ( ( '{0:.' + str ( self . config [ 'generator' ] [ 'floatPrecision' ] ) ) + 'f}' ) #NEWLINE# self . logger . info ( ( ( ( ( 'Generating features from ' + str ( nextPhraseIndex ) ) + ' to ' ) + str ( ( nextPhraseIndex + len ( phrases [ 'hits' ] [ 'hits' ] ) ) ) ) + ' phrases...' ) ) #NEWLINE# for phraseData in phrases [ 'hits' ] [ 'hits' ] : #NEWLINE# #INDENT# self . logger . info ( ( 'Dispatching phrase ' + phraseData [ '_id' ] ) ) #NEWLINE# content = { #NEWLINE# #INDENT# 'phraseId' : phraseData [ '_id' ] , #NEWLINE# 'type' : 'generate' , #NEWLINE# 'count' : 1 , #NEWLINE# 'from' : self . dispatcherName , #NEWLINE# #UNINDENT# } #NEWLINE# self . generationDispatcher . send ( content , self . workerName , self . timeout ) #NEWLINE# #UNINDENT# nextPhraseIndex += len ( phrases [ 'hits' ] [ 'hits' ] ) #NEWLINE# if ( ( endPhraseIndex != ( - 1 ) ) and ( nextPhraseIndex >= endPhraseIndex ) ) : #NEWLINE# #INDENT# break #NEWLINE# #UNINDENT# #UNINDENT# while True : #NEWLINE# #INDENT# message = self . generationDispatcher . receive ( ) #NEWLINE# if ( ( 'phraseId' in message [ 'content' ] ) and ( message [ 'content' ] [ 'phraseId' ] > 0 ) ) : #NEWLINE# #INDENT# self . phrasesGenerated += 1 #NEWLINE# self . generationDispatcher . close ( message ) #NEWLINE# self . logger . info ( ( ( ( ( 'Generated for ' + message [ 'content' ] [ 'phraseId' ] ) + str ( self . phrasesGenerated ) ) + '/' ) + str ( self . totalPhrasesDispatched ) ) ) #NEWLINE# #UNINDENT# if ( ( self . phrasesGenerated + self . phrasesNotGenerated ) >= self . totalPhrasesDispatched ) : #NEWLINE# #INDENT# self . controlChannel . send ( 'dying' ) #NEWLINE# break #NEWLINE# #UNINDENT# #UNINDENT# self . __terminate ( )