<s> from __future__ import print_function , unicode_literals , division , absolute_import <EOL> import os , sys , json , subprocess , pipes <EOL> import collections , datetime <EOL> import dxpy <EOL> from . describe import ( get_field_from_jbor , get_job_from_jbor , get_index_from_jbor , <EOL> is_job_ref , job_output_to_str , JOB_STATES ) <EOL> from . printing import ( GREEN , BLUE , BOLD , ENDC , fill ) <EOL> from . resolver import is_localjob_id <EOL> from . . compat import open , str , environ , USING_PYTHON2 , basestring <EOL> from . import file_load_utils <EOL> def exit_with_error ( msg ) : <EOL> '''<STR_LIT>''' <EOL> msg += '<STR_LIT:\n>' <EOL> msg += '<STR_LIT>' + str ( environ . get ( '<STR_LIT>' ) ) <EOL> sys . exit ( msg ) <EOL> def has_local_job_refs ( io_hash ) : <EOL> '''<STR_LIT>''' <EOL> q = [ ] <EOL> for field in io_hash : <EOL> if is_job_ref ( io_hash [ field ] ) : <EOL> if get_job_from_jbor ( io_hash [ field ] ) . startswith ( '<STR_LIT>' ) : <EOL> return True <EOL> elif isinstance ( io_hash [ field ] , list ) or isinstance ( io_hash [ field ] , dict ) : <EOL> q . append ( io_hash [ field ] ) <EOL> while len ( q ) > <NUM_LIT:0> : <EOL> thing = q . pop ( ) <EOL> if isinstance ( thing , list ) : <EOL> for i in range ( len ( thing ) ) : <EOL> if is_job_ref ( thing [ i ] ) : <EOL> if get_job_from_jbor ( thing [ i ] ) . startswith ( '<STR_LIT>' ) : <EOL> return True <EOL> elif isinstance ( thing [ i ] , list ) or isinstance ( thing [ i ] , dict ) : <EOL> q . append ( thing [ i ] ) <EOL> else : <EOL> for field in thing : <EOL> if is_job_ref ( thing [ field ] ) : <EOL> if get_job_from_jbor ( thing [ field ] ) . startswith ( '<STR_LIT>' ) : <EOL> return True <EOL> elif isinstance ( thing [ field ] , list ) or isinstance ( thing [ field ] , dict ) : <EOL> q . append ( thing [ field ] ) <EOL> return False <EOL> def resolve_job_ref ( jbor , job_outputs = { } , should_resolve = True ) : <EOL> '''<STR_LIT>''' <EOL> ref_job_id = get_job_from_jbor ( jbor ) <EOL> ref_job_field = get_field_from_jbor ( jbor ) <EOL> ref_job_index = get_index_from_jbor ( jbor ) <EOL> def resolve_from_hash ( output_hash ) : <EOL> if ref_job_index is None : <EOL> return output_hash [ ref_job_field ] <EOL> else : <EOL> return output_hash [ ref_job_field ] [ ref_job_index ] <EOL> if is_localjob_id ( ref_job_id ) : <EOL> if job_outputs . get ( ref_job_id ) is None : <EOL> if should_resolve : <EOL> raise Exception ( '<STR_LIT>' + ref_job_id + '<STR_LIT>' ) <EOL> else : <EOL> return jbor <EOL> if ref_job_field not in job_outputs [ ref_job_id ] : <EOL> raise Exception ( '<STR_LIT>' + ref_job_id + '<STR_LIT>' + ref_job_field + '<STR_LIT>' ) <EOL> return resolve_from_hash ( job_outputs [ ref_job_id ] ) <EOL> else : <EOL> dxjob = dxpy . DXJob ( ref_job_id ) <EOL> try : <EOL> dxjob . wait_on_done ( ) <EOL> except Exception as e : <EOL> raise Exception ( '<STR_LIT>' + ref_job_id + '<STR_LIT>' + str ( e ) ) <EOL> job_desc = dxjob . describe ( ) <EOL> if ref_job_field not in job_desc [ '<STR_LIT>' ] : <EOL> raise Exception ( '<STR_LIT>' + ref_job_id + '<STR_LIT>' + ref_job_field + '<STR_LIT>' ) <EOL> return resolve_from_hash ( job_desc [ '<STR_LIT>' ] ) <EOL> def resolve_job_references ( io_hash , job_outputs , should_resolve = True ) : <EOL> '''<STR_LIT>''' <EOL> q = [ ] <EOL> for field in io_hash : <EOL> if is_job_ref ( io_hash [ field ] ) : <EOL> io_hash [ field ] = resolve_job_ref ( io_hash [ field ] , job_outputs , should_resolve ) <EOL> elif isinstance ( io_hash [ field ] , list ) or isinstance ( io_hash [ field ] , dict ) : <EOL> q . append ( io_hash [ field ] ) <EOL> while len ( q ) > <NUM_LIT:0> : <EOL> thing = q . pop ( ) <EOL> if isinstance ( thing , list ) : <EOL> for i in range ( len ( thing ) ) : <EOL> if is_job_ref ( thing [ i ] ) : <EOL> thing [ i ] = resolve_job_ref ( thing [ i ] , job_outputs , should_resolve ) <EOL> elif isinstance ( thing [ i ] , list ) or isinstance ( thing [ i ] , dict ) : <EOL> q . append ( thing [ i ] ) <EOL> else : <EOL> for field in thing : <EOL> if is_job_ref ( thing [ field ] ) : <EOL> thing [ field ] = resolve_job_ref ( thing [ field ] , job_outputs , should_resolve ) <EOL> elif isinstance ( thing [ field ] , list ) or isinstance ( thing [ field ] , dict ) : <EOL> q . append ( thing [ field ] ) <EOL> def get_nonclosed_data_obj_link ( thing ) : <EOL> obj_id = None <EOL> if isinstance ( thing , dict ) and '<STR_LIT>' in thing : <EOL> if isinstance ( thing [ '<STR_LIT>' ] , basestring ) : <EOL> obj_id = thing [ '<STR_LIT>' ] <EOL> elif isinstance ( thing [ '<STR_LIT>' ] , dict ) : <EOL> obj_id = thing [ '<STR_LIT>' ] . get ( '<STR_LIT:id>' ) <EOL> if obj_id is None : <EOL> return None <EOL> obj_desc = dxpy . describe ( obj_id ) <EOL> if obj_desc . get ( '<STR_LIT:state>' ) != '<STR_LIT>' : <EOL> return obj_id <EOL> def get_implicit_depends_on ( input_hash , depends_on ) : <EOL> '''<STR_LIT>''' <EOL> q = [ ] <EOL> for field in input_hash : <EOL> possible_dep = get_nonclosed_data_obj_link ( input_hash [ field ] ) <EOL> if possible_dep is not None : <EOL> depends_on . append ( possible_dep ) <EOL> elif isinstance ( input_hash [ field ] , list ) or isinstance ( input_hash [ field ] , dict ) : <EOL> q . append ( input_hash [ field ] ) <EOL> while len ( q ) > <NUM_LIT:0> : <EOL> thing = q . pop ( ) <EOL> if isinstance ( thing , list ) : <EOL> for i in range ( len ( thing ) ) : <EOL> possible_dep = get_nonclosed_data_obj_link ( thing [ i ] ) <EOL> if possible_dep is not None : <EOL> depends_on . append ( possible_dep ) <EOL> elif isinstance ( thing [ i ] , list ) or isinstance ( thing [ i ] , dict ) : <EOL> q . append ( thing [ i ] ) <EOL> else : <EOL> for field in thing : <EOL> possible_dep = get_nonclosed_data_obj_link ( thing [ field ] ) <EOL> if possible_dep is not None : <EOL> depends_on . append ( possible_dep ) <EOL> elif isinstance ( thing [ field ] , list ) or isinstance ( thing [ field ] , dict ) : <EOL> q . append ( thing [ field ] ) <EOL> def wait_for_depends_on ( depends_on , all_job_outputs ) : <EOL> if len ( depends_on ) > <NUM_LIT:0> : <EOL> print ( fill ( '<STR_LIT>' ) ) <EOL> for an_id in depends_on : <EOL> try : <EOL> print ( '<STR_LIT>' + an_id + '<STR_LIT>' ) <EOL> if an_id . startswith ( '<STR_LIT>' ) : <EOL> if all_job_outputs . get ( an_id ) is None : <EOL> raise Exception ( '<STR_LIT>' + an_id + '<STR_LIT>' ) <EOL> elif an_id . startswith ( '<STR_LIT>' ) : <EOL> dxjob = dxpy . DXJob ( an_id ) <EOL> dxjob . wait_on_done ( ) <EOL> else : <EOL> handler = dxpy . get_handler ( an_id ) <EOL> desc = handler . describe ( ) <EOL> handler . _wait_on_close ( ) <EOL> except Exception as e : <EOL> raise Exception ( '<STR_LIT>' + an_id + '<STR_LIT>' + str ( e ) ) <EOL> def ensure_env_vars ( ) : <EOL> for var in [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] : <EOL> if var not in environ : <EOL> sys . exit ( '<STR_LIT>' + var + '<STR_LIT>' ) <EOL> def queue_entry_point ( function , input_hash , depends_on = [ ] , name = None ) : <EOL> '''<STR_LIT>''' <EOL> ensure_env_vars ( ) <EOL> all_job_outputs_path = os . path . join ( environ [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> with open ( all_job_outputs_path , '<STR_LIT:r>' ) as fd : <EOL> all_job_outputs = json . load ( fd , object_pairs_hook = collections . OrderedDict ) <EOL> job_id = '<STR_LIT>' + str ( len ( all_job_outputs ) ) <EOL> with open ( all_job_outputs_path , '<STR_LIT:wb>' ) as fd : <EOL> all_job_outputs [ job_id ] = None <EOL> json . dump ( all_job_outputs , fd , indent = <NUM_LIT:4> ) <EOL> fd . write ( b'<STR_LIT:\n>' ) <EOL> job_homedir = os . path . join ( environ [ '<STR_LIT>' ] , job_id ) <EOL> os . mkdir ( job_homedir ) <EOL> job_queue_path = os . path . join ( environ [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> with open ( job_queue_path , '<STR_LIT:r>' ) as fd : <EOL> job_queue = json . load ( fd ) <EOL> job_entry = { "<STR_LIT:id>" : job_id , <EOL> "<STR_LIT>" : function , <EOL> "<STR_LIT>" : input_hash , <EOL> "<STR_LIT>" : depends_on } <EOL> if name is not None : <EOL> job_entry [ '<STR_LIT:name>' ] = name <EOL> job_queue . append ( job_entry ) <EOL> with open ( job_queue_path , '<STR_LIT:wb>' ) as fd : <EOL> json . dump ( job_queue , fd , indent = <NUM_LIT:4> ) <EOL> fd . write ( b'<STR_LIT:\n>' ) <EOL> return job_id <EOL> def run_one_entry_point ( job_id , function , input_hash , run_spec , depends_on , name = None ) : <EOL> '''<STR_LIT>''' <EOL> print ( '<STR_LIT>' ) <EOL> job_homedir = os . path . join ( environ [ '<STR_LIT>' ] , job_id ) <EOL> job_env = environ . copy ( ) <EOL> job_env [ '<STR_LIT>' ] = os . path . join ( environ [ '<STR_LIT>' ] , job_id ) <EOL> all_job_outputs_path = os . path . join ( environ [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> with open ( all_job_outputs_path , '<STR_LIT:r>' ) as fd : <EOL> all_job_outputs = json . load ( fd , object_pairs_hook = collections . OrderedDict ) <EOL> if isinstance ( name , basestring ) : <EOL> name += '<STR_LIT>' + job_id + '<STR_LIT::>' + function + '<STR_LIT:)>' <EOL> else : <EOL> name = job_id + '<STR_LIT::>' + function <EOL> job_name = BLUE ( ) + BOLD ( ) + name + ENDC ( ) <EOL> print ( job_name ) <EOL> try : <EOL> resolve_job_references ( input_hash , all_job_outputs ) <EOL> except Exception as e : <EOL> exit_with_error ( job_name + '<STR_LIT:U+0020>' + JOB_STATES ( '<STR_LIT>' ) + '<STR_LIT>' + fill ( str ( e ) ) ) <EOL> if depends_on is None : <EOL> depends_on = [ ] <EOL> get_implicit_depends_on ( input_hash , depends_on ) <EOL> try : <EOL> wait_for_depends_on ( depends_on , all_job_outputs ) <EOL> except Exception as e : <EOL> exit_with_error ( job_name + '<STR_LIT:U+0020>' + JOB_STATES ( '<STR_LIT>' ) + '<STR_LIT>' + fill ( str ( e ) ) ) <EOL> with open ( os . path . join ( job_homedir , '<STR_LIT>' ) , '<STR_LIT:wb>' ) as fd : <EOL> json . dump ( input_hash , fd , indent = <NUM_LIT:4> ) <EOL> fd . write ( b'<STR_LIT:\n>' ) <EOL> print ( job_output_to_str ( input_hash , title = ( BOLD ( ) + '<STR_LIT>' + ENDC ( ) ) , <EOL> title_len = len ( "<STR_LIT>" ) ) . lstrip ( ) ) <EOL> if run_spec [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> env_path = os . path . join ( job_homedir , '<STR_LIT>' ) <EOL> with open ( env_path , '<STR_LIT:w>' ) as fd : <EOL> job_input_file = os . path . join ( job_homedir , '<STR_LIT>' ) <EOL> var_defs_hash = file_load_utils . gen_bash_vars ( job_input_file , job_homedir = job_homedir ) <EOL> for key , val in var_defs_hash . iteritems ( ) : <EOL> fd . write ( "<STR_LIT>" . format ( key , val ) ) <EOL> print ( BOLD ( ) + '<STR_LIT>' + ENDC ( ) ) <EOL> start_time = datetime . datetime . now ( ) <EOL> if run_spec [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> script = '''<STR_LIT>''' . format ( homedir = pipes . quote ( job_homedir ) , <EOL> env_path = pipes . quote ( os . path . join ( job_env [ '<STR_LIT>' ] , '<STR_LIT>' ) ) , <EOL> code_path = pipes . quote ( environ [ '<STR_LIT>' ] ) , <EOL> function = function ) <EOL> invocation_args = [ '<STR_LIT>' , '<STR_LIT:-c>' , '<STR_LIT>' ] + ( [ '<STR_LIT>' ] if environ . get ( '<STR_LIT>' ) else [ ] ) + [ script ] <EOL> elif run_spec [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> script = '''<STR_LIT>''' . format ( homedir = repr ( job_homedir ) , <EOL> code = run_spec [ '<STR_LIT:code>' ] ) <EOL> job_env [ '<STR_LIT>' ] = function <EOL> invocation_args = [ '<STR_LIT>' , '<STR_LIT:-c>' , script ] <EOL> if USING_PYTHON2 : <EOL> invocation_args = [ arg . encode ( sys . stdout . encoding ) for arg in invocation_args ] <EOL> env = { k : v . encode ( sys . stdout . encoding ) for k , v in job_env . items ( ) } <EOL> else : <EOL> env = job_env <EOL> fn_process = subprocess . Popen ( invocation_args , env = env ) <EOL> fn_process . communicate ( ) <EOL> end_time = datetime . datetime . now ( ) <EOL> if fn_process . returncode != <NUM_LIT:0> : <EOL> exit_with_error ( job_name + '<STR_LIT:U+0020>' + JOB_STATES ( '<STR_LIT>' ) + '<STR_LIT>' + str ( fn_process . returncode ) + '<STR_LIT>' + str ( end_time - start_time ) ) <EOL> job_output_path = os . path . join ( job_env [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> if os . path . exists ( job_output_path ) : <EOL> try : <EOL> with open ( job_output_path , '<STR_LIT:r>' ) as fd : <EOL> job_output = json . load ( fd , object_pairs_hook = collections . OrderedDict ) <EOL> except Exception as e : <EOL> exit_with_error ( '<STR_LIT>' + job_name + '<STR_LIT>' + fill ( str ( e . __class__ ) + '<STR_LIT>' + str ( e ) ) ) <EOL> else : <EOL> job_output = { } <EOL> print ( job_name + '<STR_LIT>' + GREEN ( ) + '<STR_LIT>' + ENDC ( ) + '<STR_LIT>' + str ( end_time - start_time ) ) <EOL> print ( job_output_to_str ( job_output , title = ( BOLD ( ) + "<STR_LIT>" + ENDC ( ) ) , <EOL> title_len = len ( "<STR_LIT>" ) ) . lstrip ( ) ) <EOL> with open ( os . path . join ( environ [ '<STR_LIT>' ] , '<STR_LIT>' ) , '<STR_LIT:r>' ) as fd : <EOL> all_job_outputs = json . load ( fd , object_pairs_hook = collections . OrderedDict ) <EOL> all_job_outputs [ job_id ] = job_output <EOL> for other_job_id in all_job_outputs : <EOL> if all_job_outputs [ other_job_id ] is None : <EOL> continue <EOL> resolve_job_references ( all_job_outputs [ other_job_id ] , all_job_outputs , should_resolve = False ) <EOL> with open ( os . path . join ( environ [ '<STR_LIT>' ] , '<STR_LIT>' ) , '<STR_LIT:wb>' ) as fd : <EOL> json . dump ( all_job_outputs , fd , indent = <NUM_LIT:4> ) <EOL> fd . write ( b'<STR_LIT:\n>' ) <EOL> def run_entry_points ( run_spec ) : <EOL> '''<STR_LIT>''' <EOL> job_queue_path = os . path . join ( environ [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> all_job_outputs_path = os . path . join ( environ [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> while True : <EOL> with open ( job_queue_path , '<STR_LIT:r>' ) as fd : <EOL> job_queue = json . load ( fd ) <EOL> if len ( job_queue ) == <NUM_LIT:0> : <EOL> return <EOL> with open ( all_job_outputs_path , '<STR_LIT:r>' ) as fd : <EOL> all_job_outputs = json . load ( fd ) <EOL> entry_point_to_run = None <EOL> for i , entry_point in enumerate ( job_queue ) : <EOL> runnable = True <EOL> while has_local_job_refs ( entry_point [ '<STR_LIT>' ] ) : <EOL> try : <EOL> resolve_job_references ( entry_point [ '<STR_LIT>' ] , all_job_outputs ) <EOL> except : <EOL> runnable = False <EOL> break <EOL> if runnable : <EOL> entry_point_to_run = job_queue . pop ( i ) <EOL> break <EOL> if entry_point_to_run is None : <EOL> entry_point_to_run = job_queue . pop ( <NUM_LIT:0> ) <EOL> with open ( job_queue_path , '<STR_LIT:wb>' ) as fd : <EOL> json . dump ( job_queue , fd ) <EOL> fd . write ( b'<STR_LIT:\n>' ) <EOL> run_one_entry_point ( job_id = entry_point_to_run [ '<STR_LIT:id>' ] , <EOL> function = entry_point_to_run [ '<STR_LIT>' ] , <EOL> input_hash = entry_point_to_run [ '<STR_LIT>' ] , <EOL> run_spec = run_spec , <EOL> depends_on = entry_point_to_run . get ( '<STR_LIT>' , [ ] ) , <EOL> name = entry_point_to_run . get ( '<STR_LIT:name>' ) ) </s>
