<s> """<STR_LIT>""" <EOL> from __future__ import with_statement <EOL> import Queue <EOL> import cStringIO as StringIO <EOL> import collections <EOL> import gc <EOL> import logging <EOL> import multiprocessing <EOL> import os <EOL> import os . path <EOL> import subprocess <EOL> import sys <EOL> import tempfile <EOL> import threading <EOL> import uuid <EOL> import h5py <EOL> import numpy as np <EOL> import zmq <EOL> import cellprofiler <EOL> import cellprofiler . cpimage as cpimage <EOL> import cellprofiler . measurements as cpmeas <EOL> import cellprofiler . preferences as cpprefs <EOL> import cellprofiler . workspace as cpw <EOL> from cellprofiler . utilities . zmqrequest import AnalysisRequest , Request , Reply , UpstreamExit <EOL> from cellprofiler . utilities . zmqrequest import get_announcer_address <EOL> from cellprofiler . utilities . zmqrequest import register_analysis , cancel_analysis <EOL> logger = logging . getLogger ( __name__ ) <EOL> use_analysis = True <EOL> DEBUG = '<STR_LIT>' <EOL> ANNOUNCE_DONE = "<STR_LIT>" <EOL> class Analysis ( object ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , pipeline , measurements_filename , <EOL> initial_measurements = None ) : <EOL> '''<STR_LIT>''' <EOL> self . pipeline = pipeline <EOL> initial_measurements = cpmeas . Measurements ( copy = initial_measurements ) <EOL> self . initial_measurements_buf = initial_measurements . file_contents ( ) <EOL> initial_measurements . close ( ) <EOL> self . output_path = measurements_filename <EOL> self . debug_mode = False <EOL> self . analysis_in_progress = False <EOL> self . runner = None <EOL> self . runner_lock = threading . Lock ( ) <EOL> def start ( self , analysis_event_callback , num_workers = None , overwrite = True ) : <EOL> '''<STR_LIT>''' <EOL> with self . runner_lock : <EOL> assert not self . analysis_in_progress <EOL> self . analysis_in_progress = uuid . uuid1 ( ) . hex <EOL> self . runner = AnalysisRunner ( self . analysis_in_progress , <EOL> self . pipeline , <EOL> self . initial_measurements_buf , <EOL> self . output_path , <EOL> analysis_event_callback ) <EOL> self . runner . start ( num_workers = num_workers , overwrite = overwrite ) <EOL> return self . analysis_in_progress <EOL> def pause ( self ) : <EOL> with self . runner_lock : <EOL> assert self . analysis_in_progress <EOL> self . runner . pause ( ) <EOL> def resume ( self ) : <EOL> with self . runner_lock : <EOL> assert self . analysis_in_progress <EOL> self . runner . resume ( ) <EOL> def cancel ( self ) : <EOL> with self . runner_lock : <EOL> if not self . analysis_in_progress : <EOL> return <EOL> self . analysis_in_progress = False <EOL> self . runner . cancel ( ) <EOL> self . runner = None <EOL> def check_running ( self ) : <EOL> '''<STR_LIT>''' <EOL> with self . runner_lock : <EOL> if self . analysis_in_progress : <EOL> return self . runner . check ( ) <EOL> return False <EOL> class AnalysisRunner ( object ) : <EOL> '''<STR_LIT>''' <EOL> workers = [ ] <EOL> deadman_switches = [ ] <EOL> STATUS = "<STR_LIT>" <EOL> STATUS_UNPROCESSED = "<STR_LIT>" <EOL> STATUS_IN_PROCESS = "<STR_LIT>" <EOL> STATUS_FINISHED_WAITING = "<STR_LIT>" <EOL> STATUS_DONE = "<STR_LIT>" <EOL> STATUSES = [ STATUS_UNPROCESSED , STATUS_IN_PROCESS , STATUS_FINISHED_WAITING , STATUS_DONE ] <EOL> def __init__ ( self , analysis_id , pipeline , <EOL> initial_measurements_buf , <EOL> output_path , event_listener ) : <EOL> self . initial_measurements_buf = initial_measurements_buf <EOL> self . output_path = output_path <EOL> self . analysis_id = analysis_id <EOL> self . pipeline = pipeline . copy ( ) <EOL> self . event_listener = event_listener <EOL> self . interface_work_cv = threading . Condition ( ) <EOL> self . jobserver_work_cv = threading . Condition ( ) <EOL> self . paused = False <EOL> self . cancelled = False <EOL> self . work_queue = Queue . Queue ( ) <EOL> self . in_process_queue = Queue . Queue ( ) <EOL> self . finished_queue = Queue . Queue ( ) <EOL> self . received_measurements_queue = Queue . Queue ( maxsize = <NUM_LIT:10> ) <EOL> self . shared_dicts = None <EOL> self . interface_thread = None <EOL> self . jobserver_thread = None <EOL> def start ( self , num_workers = None , overwrite = True ) : <EOL> '''<STR_LIT>''' <EOL> self . stop_workers ( ) <EOL> start_signal = threading . Semaphore ( <NUM_LIT:0> ) <EOL> self . interface_thread = start_daemon_thread ( <EOL> target = self . interface , <EOL> args = ( start_signal , ) , <EOL> kwargs = dict ( overwrite = overwrite ) , <EOL> name = '<STR_LIT>' ) <EOL> start_signal . acquire ( ) <EOL> self . jobserver_thread = start_daemon_thread ( <EOL> target = self . jobserver , <EOL> args = ( self . analysis_id , start_signal ) , <EOL> name = '<STR_LIT>' ) <EOL> start_signal . acquire ( ) <EOL> self . start_workers ( num_workers ) <EOL> def check ( self ) : <EOL> return ( ( self . interface_thread is not None ) and <EOL> ( self . jobserver_thread is not None ) and <EOL> self . interface_thread . is_alive ( ) and <EOL> self . jobserver_thread . is_alive ( ) ) <EOL> def notify_threads ( self ) : <EOL> with self . interface_work_cv : <EOL> self . interface_work_cv . notify ( ) <EOL> with self . jobserver_work_cv : <EOL> self . jobserver_work_cv . notify ( ) <EOL> def cancel ( self ) : <EOL> '''<STR_LIT>''' <EOL> logger . debug ( "<STR_LIT>" ) <EOL> self . stop_workers ( ) <EOL> logger . debug ( "<STR_LIT>" ) <EOL> self . cancelled = True <EOL> self . paused = False <EOL> self . notify_threads ( ) <EOL> logger . debug ( "<STR_LIT>" ) <EOL> self . interface_thread . join ( ) <EOL> logger . debug ( "<STR_LIT>" ) <EOL> self . jobserver_thread . join ( ) <EOL> logger . debug ( "<STR_LIT>" ) <EOL> def pause ( self ) : <EOL> '''<STR_LIT>''' <EOL> self . paused = True <EOL> self . notify_threads ( ) <EOL> def resume ( self ) : <EOL> '''<STR_LIT>''' <EOL> self . paused = False <EOL> self . notify_threads ( ) <EOL> def post_event ( self , evt ) : <EOL> self . event_listener ( evt ) <EOL> def post_run_display_handler ( self , workspace , module ) : <EOL> event = DisplayPostRunRequest ( module . module_num , <EOL> workspace . display_data ) <EOL> self . event_listener ( event ) <EOL> def interface ( self , <EOL> start_signal , <EOL> image_set_start = <NUM_LIT:1> , <EOL> image_set_end = None , <EOL> overwrite = True ) : <EOL> '''<STR_LIT>''' <EOL> from javabridge import attach , detach <EOL> posted_analysis_started = False <EOL> acknowledged_thread_start = False <EOL> measurements = None <EOL> workspace = None <EOL> attach ( ) <EOL> try : <EOL> self . pipeline . add_listener ( lambda pipe , evt : self . post_event ( evt ) ) <EOL> initial_measurements = None <EOL> if self . output_path is None : <EOL> fd , filename = tempfile . mkstemp ( <EOL> "<STR_LIT>" , dir = cpprefs . get_temporary_directory ( ) ) <EOL> try : <EOL> fd = os . fdopen ( fd , "<STR_LIT:wb>" ) <EOL> fd . write ( self . initial_measurements_buf ) <EOL> fd . close ( ) <EOL> initial_measurements = cpmeas . Measurements ( <EOL> filename = filename , mode = "<STR_LIT:r>" ) <EOL> measurements = cpmeas . Measurements ( <EOL> image_set_start = None , <EOL> copy = initial_measurements , <EOL> mode = "<STR_LIT:a>" ) <EOL> finally : <EOL> if initial_measurements is not None : <EOL> initial_measurements . close ( ) <EOL> os . unlink ( filename ) <EOL> else : <EOL> with open ( self . output_path , "<STR_LIT:wb>" ) as fd : <EOL> fd . write ( self . initial_measurements_buf ) <EOL> measurements = cpmeas . Measurements ( image_set_start = None , <EOL> filename = self . output_path , <EOL> mode = "<STR_LIT:a>" ) <EOL> self . shared_dicts = [ m . get_dictionary ( ) for m in self . pipeline . modules ( ) ] <EOL> workspace = cpw . Workspace ( self . pipeline , None , None , None , <EOL> measurements , cpimage . ImageSetList ( ) ) <EOL> if image_set_end is None : <EOL> image_set_end = measurements . get_image_numbers ( ) [ - <NUM_LIT:1> ] <EOL> image_sets_to_process = filter ( <EOL> lambda x : x >= image_set_start and x <= image_set_end , <EOL> measurements . get_image_numbers ( ) ) <EOL> self . post_event ( AnalysisStarted ( ) ) <EOL> posted_analysis_started = True <EOL> has_groups = measurements . has_groups ( ) <EOL> if self . pipeline . requires_aggregation ( ) : <EOL> overwrite = True <EOL> if has_groups and not overwrite : <EOL> if not measurements . has_feature ( cpmeas . IMAGE , self . STATUS ) : <EOL> overwrite = True <EOL> else : <EOL> group_status = { } <EOL> for image_number in measurements . get_image_numbers ( ) : <EOL> group_number = measurements [ <EOL> cpmeas . IMAGE , cpmeas . GROUP_NUMBER , image_number ] <EOL> status = measurements [ cpmeas . IMAGE , self . STATUS , <EOL> image_number ] <EOL> if status != self . STATUS_DONE : <EOL> group_status [ group_number ] = self . STATUS_UNPROCESSED <EOL> elif group_number not in group_status : <EOL> group_status [ group_number ] = self . STATUS_DONE <EOL> new_image_sets_to_process = [ ] <EOL> for image_set_number in image_sets_to_process : <EOL> needs_reset = False <EOL> if ( overwrite or <EOL> ( not measurements . has_measurements ( <EOL> cpmeas . IMAGE , self . STATUS , image_set_number ) ) or <EOL> ( measurements [ cpmeas . IMAGE , self . STATUS , image_set_number ] <EOL> != self . STATUS_DONE ) ) : <EOL> needs_reset = True <EOL> elif has_groups : <EOL> group_number = measurements [ <EOL> cpmeas . IMAGE , cpmeas . GROUP_NUMBER , image_set_number ] <EOL> if group_status [ group_number ] != self . STATUS_DONE : <EOL> needs_reset = True <EOL> if needs_reset : <EOL> measurements [ cpmeas . IMAGE , self . STATUS , image_set_number ] = self . STATUS_UNPROCESSED <EOL> new_image_sets_to_process . append ( image_set_number ) <EOL> image_sets_to_process = new_image_sets_to_process <EOL> if has_groups or self . pipeline . requires_aggregation ( ) : <EOL> worker_runs_post_group = True <EOL> job_groups = { } <EOL> for image_set_number in image_sets_to_process : <EOL> group_number = measurements [ cpmeas . IMAGE , <EOL> cpmeas . GROUP_NUMBER , <EOL> image_set_number ] <EOL> group_index = measurements [ cpmeas . IMAGE , <EOL> cpmeas . GROUP_INDEX , <EOL> image_set_number ] <EOL> job_groups [ group_number ] = job_groups . get ( group_number , [ ] ) + [ ( group_index , image_set_number ) ] <EOL> job_groups = [ [ isn for _ , isn in sorted ( job_groups [ group_number ] ) ] <EOL> for group_number in sorted ( job_groups ) ] <EOL> else : <EOL> worker_runs_post_group = False <EOL> job_groups = [ [ image_set_number ] for image_set_number in image_sets_to_process ] <EOL> if not worker_runs_post_group : <EOL> self . work_queue . put ( ( job_groups [ <NUM_LIT:0> ] , <EOL> worker_runs_post_group , <EOL> True ) ) <EOL> waiting_for_first_imageset = True <EOL> del job_groups [ <NUM_LIT:0> ] <EOL> else : <EOL> waiting_for_first_imageset = False <EOL> for job in job_groups : <EOL> self . work_queue . put ( ( job , worker_runs_post_group , False ) ) <EOL> job_groups = [ ] <EOL> start_signal . release ( ) <EOL> acknowledged_thread_start = True <EOL> while not self . cancelled : <EOL> while not self . received_measurements_queue . empty ( ) : <EOL> image_numbers , buf = self . received_measurements_queue . get ( ) <EOL> image_numbers = [ int ( i ) for i in image_numbers ] <EOL> recd_measurements = cpmeas . load_measurements_from_buffer ( buf ) <EOL> self . copy_recieved_measurements ( recd_measurements , measurements , image_numbers ) <EOL> recd_measurements . close ( ) <EOL> del recd_measurements <EOL> while not self . in_process_queue . empty ( ) : <EOL> image_set_numbers = self . in_process_queue . get ( ) <EOL> for image_set_number in image_set_numbers : <EOL> measurements [ cpmeas . IMAGE , self . STATUS , int ( image_set_number ) ] = self . STATUS_IN_PROCESS <EOL> while not self . finished_queue . empty ( ) : <EOL> finished_req = self . finished_queue . get ( ) <EOL> measurements [ <EOL> cpmeas . IMAGE , self . STATUS , int ( finished_req . image_set_number ) ] = self . STATUS_FINISHED_WAITING <EOL> if waiting_for_first_imageset : <EOL> assert isinstance ( finished_req , <EOL> ImageSetSuccessWithDictionary ) <EOL> self . shared_dicts = finished_req . shared_dicts <EOL> waiting_for_first_imageset = False <EOL> assert len ( self . shared_dicts ) == len ( self . pipeline . modules ( ) ) <EOL> for job in job_groups : <EOL> self . work_queue . put ( ( job , worker_runs_post_group , False ) ) <EOL> finished_req . reply ( Ack ( ) ) <EOL> counts = collections . Counter ( measurements [ cpmeas . IMAGE , self . STATUS , image_set_number ] <EOL> for image_set_number in image_sets_to_process ) <EOL> self . post_event ( AnalysisProgress ( counts ) ) <EOL> if counts [ self . STATUS_DONE ] == len ( image_sets_to_process ) : <EOL> last_image_number = measurements . get_image_numbers ( ) [ - <NUM_LIT:1> ] <EOL> measurements . image_set_number = last_image_number <EOL> if not worker_runs_post_group : <EOL> self . pipeline . post_group ( workspace , { } ) <EOL> workspace = cpw . Workspace ( self . pipeline , <EOL> None , None , None , <EOL> measurements , None , None ) <EOL> workspace . post_run_display_handler = self . post_run_display_handler <EOL> self . pipeline . post_run ( workspace ) <EOL> break <EOL> measurements . flush ( ) <EOL> with self . interface_work_cv : <EOL> while ( self . paused or <EOL> ( ( not self . cancelled ) and <EOL> self . in_process_queue . empty ( ) and <EOL> self . finished_queue . empty ( ) and <EOL> self . received_measurements_queue . empty ( ) ) ) : <EOL> self . interface_work_cv . wait ( ) <EOL> finally : <EOL> detach ( ) <EOL> if not acknowledged_thread_start : <EOL> start_signal . release ( ) <EOL> if posted_analysis_started : <EOL> was_cancelled = self . cancelled <EOL> self . post_event ( AnalysisFinished ( measurements , was_cancelled ) ) <EOL> self . stop_workers ( ) <EOL> self . analysis_id = False <EOL> def copy_recieved_measurements ( self , <EOL> recd_measurements , <EOL> measurements , <EOL> image_numbers ) : <EOL> '''<STR_LIT>''' <EOL> measurements . copy_relationships ( recd_measurements ) <EOL> for o in recd_measurements . get_object_names ( ) : <EOL> if o == cpmeas . EXPERIMENT : <EOL> continue <EOL> elif o == cpmeas . IMAGE : <EOL> for feature in recd_measurements . get_feature_names ( o ) : <EOL> if not measurements . has_feature ( cpmeas . IMAGE , feature ) : <EOL> f_image_numbers = image_numbers <EOL> else : <EOL> local_values = measurements [ <EOL> cpmeas . IMAGE , feature , image_numbers ] <EOL> remote_values = recd_measurements [ <EOL> cpmeas . IMAGE , feature , image_numbers ] <EOL> f_image_numbers = [ <EOL> i for i , lv , rv in zip ( <EOL> image_numbers , local_values , remote_values ) <EOL> if ( np . any ( rv != lv ) if isinstance ( lv , np . ndarray ) <EOL> else lv != rv ) ] <EOL> if len ( f_image_numbers ) > <NUM_LIT:0> : <EOL> measurements [ o , feature , f_image_numbers ] = recd_measurements [ o , feature , f_image_numbers ] <EOL> else : <EOL> for feature in recd_measurements . get_feature_names ( o ) : <EOL> measurements [ o , feature , image_numbers ] = recd_measurements [ o , feature , image_numbers ] <EOL> for image_set_number in image_numbers : <EOL> measurements [ cpmeas . IMAGE , self . STATUS , image_set_number ] = self . STATUS_DONE <EOL> def jobserver ( self , analysis_id , start_signal ) : <EOL> request_queue = Queue . Queue ( ) <EOL> boundary = register_analysis ( analysis_id , <EOL> request_queue ) <EOL> start_signal . release ( ) <EOL> i_was_paused_before = False <EOL> while not self . cancelled : <EOL> with self . jobserver_work_cv : <EOL> if self . paused and not i_was_paused_before : <EOL> self . post_event ( AnalysisPaused ( ) ) <EOL> i_was_paused_before = True <EOL> if self . paused or request_queue . empty ( ) : <EOL> self . jobserver_work_cv . wait ( <NUM_LIT:1> ) <EOL> continue <EOL> if i_was_paused_before : <EOL> self . post_event ( AnalysisResumed ( ) ) <EOL> i_was_paused_before = False <EOL> try : <EOL> req = request_queue . get ( timeout = <NUM_LIT> ) <EOL> except Queue . Empty : <EOL> continue <EOL> if isinstance ( req , PipelinePreferencesRequest ) : <EOL> logger . debug ( "<STR_LIT>" ) <EOL> req . reply ( Reply ( pipeline_blob = np . array ( self . pipeline_as_string ( ) ) , <EOL> preferences = cpprefs . preferences_as_dict ( ) ) ) <EOL> logger . debug ( "<STR_LIT>" ) <EOL> elif isinstance ( req , InitialMeasurementsRequest ) : <EOL> logger . debug ( "<STR_LIT>" ) <EOL> req . reply ( Reply ( buf = self . initial_measurements_buf ) ) <EOL> logger . debug ( "<STR_LIT>" ) <EOL> elif isinstance ( req , WorkRequest ) : <EOL> if not self . work_queue . empty ( ) : <EOL> logger . debug ( "<STR_LIT>" ) <EOL> job , worker_runs_post_group , wants_dictionary = self . work_queue . get ( ) <EOL> req . reply ( WorkReply ( <EOL> image_set_numbers = job , <EOL> worker_runs_post_group = worker_runs_post_group , <EOL> wants_dictionary = wants_dictionary ) ) <EOL> self . queue_dispatched_job ( job ) <EOL> logger . debug ( "<STR_LIT>" % <EOL> "<STR_LIT:U+002C>" . join ( [ str ( i ) for i in job ] ) ) <EOL> else : <EOL> req . reply ( NoWorkReply ( ) ) <EOL> elif isinstance ( req , ImageSetSuccess ) : <EOL> logger . debug ( "<STR_LIT>" ) <EOL> self . queue_imageset_finished ( req ) <EOL> logger . debug ( "<STR_LIT>" ) <EOL> elif isinstance ( req , SharedDictionaryRequest ) : <EOL> logger . debug ( "<STR_LIT>" ) <EOL> req . reply ( SharedDictionaryReply ( dictionaries = self . shared_dicts ) ) <EOL> logger . debug ( "<STR_LIT>" ) <EOL> elif isinstance ( req , MeasurementsReport ) : <EOL> logger . debug ( "<STR_LIT>" ) <EOL> self . queue_received_measurements ( req . image_set_numbers , <EOL> req . buf ) <EOL> req . reply ( Ack ( ) ) <EOL> logger . debug ( "<STR_LIT>" ) <EOL> elif isinstance ( req , AnalysisCancelRequest ) : <EOL> logger . debug ( "<STR_LIT>" ) <EOL> with self . interface_work_cv : <EOL> self . cancelled = True <EOL> self . interface_work_cv . notify ( ) <EOL> req . reply ( Ack ( ) ) <EOL> elif isinstance ( req , ( InteractionRequest , DisplayRequest , <EOL> DisplayPostGroupRequest , <EOL> ExceptionReport , DebugWaiting , DebugComplete , <EOL> OmeroLoginRequest ) ) : <EOL> logger . debug ( "<STR_LIT>" ) <EOL> self . post_event ( req ) <EOL> logger . debug ( "<STR_LIT>" ) <EOL> else : <EOL> msg = "<STR_LIT>" % ( req , type ( req ) ) <EOL> logger . error ( msg ) <EOL> raise ValueError ( msg ) <EOL> boundary . cancel ( analysis_id ) <EOL> def queue_job ( self , image_set_number ) : <EOL> self . work_queue . put ( image_set_number ) <EOL> def queue_dispatched_job ( self , job ) : <EOL> self . in_process_queue . put ( job ) <EOL> with self . interface_work_cv : <EOL> self . interface_work_cv . notify ( ) <EOL> def queue_imageset_finished ( self , finished_req ) : <EOL> self . finished_queue . put ( finished_req ) <EOL> with self . interface_work_cv : <EOL> self . interface_work_cv . notify ( ) <EOL> def queue_received_measurements ( self , image_set_numbers , measurements ) : <EOL> self . received_measurements_queue . put ( ( image_set_numbers , measurements ) ) <EOL> with self . interface_work_cv : <EOL> self . interface_work_cv . notify ( ) <EOL> def pipeline_as_string ( self ) : <EOL> s = StringIO . StringIO ( ) <EOL> self . pipeline . savetxt ( s ) <EOL> return s . getvalue ( ) <EOL> @ classmethod <EOL> def start_workers ( cls , num = None ) : <EOL> if cls . workers : <EOL> return <EOL> try : <EOL> num = multiprocessing . cpu_count ( ) if num is None else num <EOL> except NotImplementedError : <EOL> num = <NUM_LIT:4> <EOL> cls . work_announce_address = get_announcer_address ( ) <EOL> logger . info ( "<STR_LIT>" % cls . work_announce_address ) <EOL> if '<STR_LIT>' in os . environ : <EOL> if os . environ [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> return <EOL> from cellprofiler . analysis_worker import AnalysisWorker , NOTIFY_ADDR , NOTIFY_STOP <EOL> from cellprofiler . pipeline import CancelledException <EOL> class WorkerRunner ( threading . Thread ) : <EOL> def __init__ ( self , work_announce_address ) : <EOL> threading . Thread . __init__ ( self ) <EOL> self . work_announce_address = work_announce_address <EOL> self . notify_socket = zmq . Context . instance ( ) . socket ( zmq . PUB ) <EOL> self . notify_socket . bind ( NOTIFY_ADDR ) <EOL> def run ( self ) : <EOL> with AnalysisWorker ( self . work_announce_address ) as aw : <EOL> try : <EOL> aw . run ( ) <EOL> except CancelledException : <EOL> logger . info ( "<STR_LIT>" ) <EOL> def wait ( self ) : <EOL> self . notify_socket . send ( NOTIFY_STOP ) <EOL> self . join ( ) <EOL> thread = WorkerRunner ( cls . work_announce_address ) <EOL> thread . setDaemon ( True ) <EOL> thread . start ( ) <EOL> cls . workers . append ( thread ) <EOL> return <EOL> close_fds = False <EOL> for idx in range ( num ) : <EOL> if sys . platform == '<STR_LIT>' : <EOL> close_all_on_exec ( ) <EOL> aw_args = [ "<STR_LIT>" , cls . work_announce_address , <EOL> "<STR_LIT>" , cpprefs . get_plugin_directory ( ) , <EOL> "<STR_LIT>" , cpprefs . get_ij_plugin_directory ( ) ] <EOL> jvm_arg = "<STR_LIT>" % cpprefs . get_jvm_heap_mb ( ) <EOL> aw_args . append ( "<STR_LIT>" % jvm_arg ) <EOL> if hasattr ( sys , '<STR_LIT>' ) : <EOL> if sys . platform == '<STR_LIT>' : <EOL> executable = os . path . join ( <EOL> os . path . dirname ( sys . executable ) , "<STR_LIT>" ) <EOL> args = ( [ executable ] + aw_args ) <EOL> elif sys . platform . startswith ( '<STR_LIT>' ) : <EOL> aw_path = os . path . join ( os . path . dirname ( __file__ ) , <EOL> "<STR_LIT>" ) <EOL> args = [ sys . executable , aw_path ] + aw_args <EOL> else : <EOL> args = [ sys . executable ] + aw_args <EOL> worker = subprocess . Popen ( args , <EOL> env = find_worker_env ( idx ) , <EOL> stdin = subprocess . PIPE , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . STDOUT , <EOL> close_fds = close_fds ) <EOL> else : <EOL> worker = subprocess . Popen ( <EOL> [ find_python ( ) , <EOL> '<STR_LIT>' , <EOL> find_analysis_worker_source ( ) ] + aw_args , <EOL> env = find_worker_env ( idx ) , <EOL> stdin = subprocess . PIPE , <EOL> stdout = subprocess . PIPE , <EOL> stderr = subprocess . STDOUT , <EOL> close_fds = close_fds ) <EOL> def run_logger ( workR , widx ) : <EOL> while True : <EOL> try : <EOL> line = workR . stdout . readline ( ) <EOL> if not line : <EOL> break <EOL> logger . info ( "<STR_LIT>" , widx , line . rstrip ( ) ) <EOL> except : <EOL> break <EOL> start_daemon_thread ( target = run_logger , args = ( worker , idx , ) , name = '<STR_LIT>' ) <EOL> cls . workers += [ worker ] <EOL> cls . deadman_switches += [ worker . stdin ] <EOL> @ classmethod <EOL> def stop_workers ( cls ) : <EOL> for deadman_switch in cls . deadman_switches : <EOL> deadman_switch . close ( ) <EOL> for worker in cls . workers : <EOL> worker . wait ( ) <EOL> cls . workers = [ ] <EOL> cls . deadman_switches = [ ] <EOL> def find_python ( ) : <EOL> if hasattr ( sys , '<STR_LIT>' ) : <EOL> if sys . platform == "<STR_LIT>" : <EOL> app_python = os . path . join ( os . path . dirname ( os . environ [ '<STR_LIT>' ] ) , "<STR_LIT>" ) <EOL> return app_python <EOL> return sys . executable <EOL> def find_worker_env ( idx ) : <EOL> '''<STR_LIT>''' <EOL> newenv = os . environ . copy ( ) <EOL> root_dir = os . path . abspath ( <EOL> os . path . join ( os . path . dirname ( cellprofiler . __file__ ) , '<STR_LIT:..>' ) ) <EOL> added_paths = [ ] <EOL> if '<STR_LIT>' in newenv : <EOL> old_path = newenv [ '<STR_LIT>' ] <EOL> if not any ( [ root_dir == path <EOL> for path in old_path . split ( os . pathsep ) ] ) : <EOL> added_paths . append ( root_dir ) <EOL> else : <EOL> added_paths . append ( root_dir ) <EOL> if hasattr ( sys , '<STR_LIT>' ) : <EOL> if sys . platform == "<STR_LIT>" : <EOL> added_paths += [ p for p in sys . path if isinstance ( p , basestring ) ] <EOL> if '<STR_LIT>' in newenv : <EOL> added_paths . insert ( <NUM_LIT:0> , newenv [ '<STR_LIT>' ] ) <EOL> newenv [ '<STR_LIT>' ] = os . pathsep . join ( <EOL> [ x . encode ( '<STR_LIT:utf-8>' ) for x in added_paths ] ) <EOL> if "<STR_LIT>" in newenv : <EOL> del newenv [ "<STR_LIT>" ] <EOL> if "<STR_LIT>" in newenv : <EOL> port = str ( int ( newenv [ "<STR_LIT>" ] ) + idx ) <EOL> newenv [ "<STR_LIT>" ] = port <EOL> del newenv [ "<STR_LIT>" ] <EOL> for key in newenv : <EOL> if isinstance ( newenv [ key ] , unicode ) : <EOL> newenv [ key ] = newenv [ key ] . encode ( '<STR_LIT:utf-8>' ) <EOL> return newenv <EOL> def find_analysis_worker_source ( ) : <EOL> import cellprofiler . analysis <EOL> return os . path . join ( os . path . dirname ( cellprofiler . analysis . __file__ ) , "<STR_LIT>" ) <EOL> def start_daemon_thread ( target = None , args = ( ) , kwargs = None , name = None ) : <EOL> thread = threading . Thread ( target = target , args = args , kwargs = kwargs , name = name ) <EOL> thread . daemon = True <EOL> thread . start ( ) <EOL> return thread <EOL> class AnalysisStarted ( object ) : <EOL> pass <EOL> class AnalysisProgress ( object ) : <EOL> def __init__ ( self , counts ) : <EOL> self . counts = counts <EOL> class AnalysisPaused ( object ) : <EOL> pass <EOL> class AnalysisResumed ( object ) : <EOL> pass <EOL> class AnalysisFinished ( object ) : <EOL> def __init__ ( self , measurements , cancelled ) : <EOL> self . measurements = measurements <EOL> self . cancelled = cancelled <EOL> class PipelinePreferencesRequest ( AnalysisRequest ) : <EOL> pass <EOL> class InitialMeasurementsRequest ( AnalysisRequest ) : <EOL> pass <EOL> class WorkRequest ( AnalysisRequest ) : <EOL> pass <EOL> class ImageSetSuccess ( AnalysisRequest ) : <EOL> def __init__ ( self , analysis_id , image_set_number = None ) : <EOL> AnalysisRequest . __init__ ( self , analysis_id , <EOL> image_set_number = image_set_number ) <EOL> class ImageSetSuccessWithDictionary ( ImageSetSuccess ) : <EOL> def __init__ ( self , analysis_id , image_set_number , shared_dicts ) : <EOL> ImageSetSuccess . __init__ ( self , analysis_id , <EOL> image_set_number = image_set_number ) <EOL> self . shared_dicts = shared_dicts <EOL> class DictionaryReqRep ( Reply ) : <EOL> pass <EOL> class MeasurementsReport ( AnalysisRequest ) : <EOL> def __init__ ( self , analysis_id , buf , image_set_numbers = [ ] ) : <EOL> AnalysisRequest . __init__ ( self , analysis_id , <EOL> buf = buf , <EOL> image_set_numbers = image_set_numbers ) <EOL> class InteractionRequest ( AnalysisRequest ) : <EOL> pass <EOL> class AnalysisCancelRequest ( AnalysisRequest ) : <EOL> pass <EOL> class DisplayRequest ( AnalysisRequest ) : <EOL> pass <EOL> class DisplayPostRunRequest ( object ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , module_num , display_data ) : <EOL> self . module_num = module_num <EOL> self . display_data = display_data <EOL> class DisplayPostGroupRequest ( AnalysisRequest ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , analysis_id , module_num , display_data , image_set_number ) : <EOL> AnalysisRequest . __init__ ( <EOL> self , analysis_id , <EOL> module_num = module_num , <EOL> image_set_number = image_set_number , <EOL> display_data = display_data ) <EOL> class SharedDictionaryRequest ( AnalysisRequest ) : <EOL> def __init__ ( self , analysis_id , module_num = - <NUM_LIT:1> ) : <EOL> AnalysisRequest . __init__ ( self , analysis_id , module_num = module_num ) <EOL> class SharedDictionaryReply ( Reply ) : <EOL> def __init__ ( self , dictionaries = [ { } ] ) : <EOL> Reply . __init__ ( self , dictionaries = dictionaries ) <EOL> class ExceptionReport ( AnalysisRequest ) : <EOL> def __init__ ( self , analysis_id , <EOL> image_set_number , module_name , <EOL> exc_type , exc_message , exc_traceback , <EOL> filename , line_number ) : <EOL> AnalysisRequest . __init__ ( self , <EOL> analysis_id , <EOL> image_set_number = image_set_number , <EOL> module_name = module_name , <EOL> exc_type = exc_type , <EOL> exc_message = exc_message , <EOL> exc_traceback = exc_traceback , <EOL> filename = filename , <EOL> line_number = line_number ) <EOL> def __str__ ( self ) : <EOL> return "<STR_LIT>" % ( self . exc_type , self . exc_message ) <EOL> class ExceptionPleaseDebugReply ( Reply ) : <EOL> def __init__ ( self , disposition , verification_hash = None ) : <EOL> Reply . __init__ ( self , disposition = disposition , verification_hash = verification_hash ) <EOL> class DebugWaiting ( AnalysisRequest ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , analysis_id , port ) : <EOL> AnalysisRequest . __init__ ( self , <EOL> analysis_id = analysis_id , <EOL> port = port ) <EOL> class DebugCancel ( Reply ) : <EOL> '''<STR_LIT>''' <EOL> class DebugComplete ( AnalysisRequest ) : <EOL> pass <EOL> class InteractionReply ( Reply ) : <EOL> pass <EOL> class WorkReply ( Reply ) : <EOL> pass <EOL> class NoWorkReply ( Reply ) : <EOL> pass <EOL> class ServerExited ( UpstreamExit ) : <EOL> pass <EOL> class OmeroLoginRequest ( AnalysisRequest ) : <EOL> pass <EOL> class OmeroLoginReply ( Reply ) : <EOL> def __init__ ( self , credentials ) : <EOL> Reply . __init__ ( self , credentials = credentials ) <EOL> class Ack ( Reply ) : <EOL> def __init__ ( self , message = "<STR_LIT>" ) : <EOL> Reply . __init__ ( self , message = message ) <EOL> if sys . platform == "<STR_LIT>" : <EOL> import fcntl <EOL> def close_all_on_exec ( ) : <EOL> '''<STR_LIT>''' <EOL> try : <EOL> maxfd = os . sysconf ( '<STR_LIT>' ) <EOL> except : <EOL> maxfd = <NUM_LIT> <EOL> for fd in range ( <NUM_LIT:3> , maxfd ) : <EOL> try : <EOL> fcntl . fcntl ( fd , fcntl . FD_CLOEXEC ) <EOL> except : <EOL> pass <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> import time <EOL> import cellprofiler . pipeline <EOL> import cellprofiler . preferences <EOL> import cellprofiler . utilities . thread_excepthook <EOL> import cellprofiler . analysis <EOL> globals ( ) . update ( cellprofiler . analysis . __dict__ ) <EOL> AnalysisRunner . start_workers ( <NUM_LIT:2> ) <EOL> AnalysisRunner . stop_workers ( ) </s>
