<s> from os . path import abspath , basename , join , exists <EOL> from os . path import isfile <EOL> from os . path import dirname <EOL> from os . path import relpath <EOL> from os import listdir , sep <EOL> from re import findall <EOL> from io import open <EOL> from . . staging import COMMAND_VERSION_FILENAME <EOL> from . . action_mapper import FileActionMapper <EOL> from . . action_mapper import path_type <EOL> from . . action_mapper import MessageAction <EOL> from . . util import PathHelper <EOL> from . . util import directory_files <EOL> from logging import getLogger <EOL> log = getLogger ( __name__ ) <EOL> def submit_job ( client , client_job_description , job_config = None ) : <EOL> """<STR_LIT:U+0020>""" <EOL> file_stager = FileStager ( client , client_job_description , job_config ) <EOL> rebuilt_command_line = file_stager . get_command_line ( ) <EOL> job_id = file_stager . job_id <EOL> launch_kwds = dict ( <EOL> command_line = rebuilt_command_line , <EOL> dependencies_description = client_job_description . dependencies_description , <EOL> env = client_job_description . env , <EOL> ) <EOL> if file_stager . job_config : <EOL> launch_kwds [ "<STR_LIT>" ] = file_stager . job_config <EOL> remote_staging = { } <EOL> remote_staging_actions = file_stager . transfer_tracker . remote_staging_actions <EOL> if remote_staging_actions : <EOL> remote_staging [ "<STR_LIT>" ] = remote_staging_actions <EOL> remote_staging [ "<STR_LIT>" ] = file_stager . action_mapper . to_dict ( ) <EOL> remote_staging [ "<STR_LIT>" ] = client_job_description . client_outputs . to_dict ( ) <EOL> if remote_staging : <EOL> launch_kwds [ "<STR_LIT>" ] = remote_staging <EOL> client . launch ( ** launch_kwds ) <EOL> return job_id <EOL> class FileStager ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , client , client_job_description , job_config ) : <EOL> """<STR_LIT:U+0020>""" <EOL> self . client = client <EOL> self . command_line = client_job_description . command_line <EOL> self . config_files = client_job_description . config_files <EOL> self . input_files = client_job_description . input_files <EOL> self . output_files = client_job_description . output_files <EOL> if client_job_description . tool is not None : <EOL> self . tool_id = client_job_description . tool . id <EOL> self . tool_version = client_job_description . tool . version <EOL> self . tool_dir = abspath ( client_job_description . tool . tool_dir ) <EOL> else : <EOL> self . tool_id = None <EOL> self . tool_version = None <EOL> self . tool_dir = None <EOL> self . working_directory = client_job_description . working_directory <EOL> self . metadata_directory = client_job_description . metadata_directory <EOL> self . version_file = client_job_description . version_file <EOL> self . arbitrary_files = client_job_description . arbitrary_files <EOL> self . rewrite_paths = client_job_description . rewrite_paths <EOL> self . job_inputs = JobInputs ( self . command_line , self . config_files ) <EOL> self . action_mapper = FileActionMapper ( client ) <EOL> self . __handle_setup ( job_config ) <EOL> self . transfer_tracker = TransferTracker ( client , self . path_helper , self . action_mapper , self . job_inputs , rewrite_paths = self . rewrite_paths ) <EOL> self . __initialize_referenced_tool_files ( ) <EOL> if self . rewrite_paths : <EOL> self . __initialize_referenced_arbitrary_files ( ) <EOL> self . __upload_tool_files ( ) <EOL> self . __upload_input_files ( ) <EOL> self . __upload_working_directory_files ( ) <EOL> self . __upload_metadata_directory_files ( ) <EOL> self . __upload_arbitrary_files ( ) <EOL> if self . rewrite_paths : <EOL> self . __initialize_output_file_renames ( ) <EOL> self . __initialize_task_output_file_renames ( ) <EOL> self . __initialize_config_file_renames ( ) <EOL> self . __initialize_version_file_rename ( ) <EOL> self . __handle_rewrites ( ) <EOL> self . __upload_rewritten_config_files ( ) <EOL> def __handle_setup ( self , job_config ) : <EOL> if not job_config : <EOL> job_config = self . client . setup ( self . tool_id , self . tool_version ) <EOL> self . new_working_directory = job_config [ '<STR_LIT>' ] <EOL> self . new_outputs_directory = job_config [ '<STR_LIT>' ] <EOL> self . new_configs_directory = job_config . get ( '<STR_LIT>' , self . new_working_directory ) <EOL> self . remote_separator = self . __parse_remote_separator ( job_config ) <EOL> self . path_helper = PathHelper ( self . remote_separator ) <EOL> galaxy_job_id = self . client . job_id <EOL> self . job_id = job_config . get ( '<STR_LIT>' , galaxy_job_id ) <EOL> if self . job_id != galaxy_job_id : <EOL> self . client . job_id = self . job_id <EOL> self . job_config = job_config <EOL> def __parse_remote_separator ( self , job_config ) : <EOL> separator = job_config . get ( "<STR_LIT>" , { } ) . get ( "<STR_LIT>" , None ) <EOL> if not separator : <EOL> separator = job_config [ "<STR_LIT>" ] <EOL> return separator <EOL> def __initialize_referenced_tool_files ( self ) : <EOL> self . referenced_tool_files = self . job_inputs . find_referenced_subfiles ( self . tool_dir ) <EOL> def __initialize_referenced_arbitrary_files ( self ) : <EOL> referenced_arbitrary_path_mappers = dict ( ) <EOL> for mapper in self . action_mapper . unstructured_mappers ( ) : <EOL> mapper_pattern = mapper . to_pattern ( ) <EOL> paths = self . job_inputs . find_pattern_references ( mapper_pattern ) <EOL> for path in paths : <EOL> if path not in referenced_arbitrary_path_mappers : <EOL> referenced_arbitrary_path_mappers [ path ] = mapper <EOL> for path , mapper in referenced_arbitrary_path_mappers . items ( ) : <EOL> action = self . action_mapper . action ( path , path_type . UNSTRUCTURED , mapper ) <EOL> unstructured_map = action . unstructured_map ( self . path_helper ) <EOL> self . arbitrary_files . update ( unstructured_map ) <EOL> def __upload_tool_files ( self ) : <EOL> for referenced_tool_file in self . referenced_tool_files : <EOL> self . transfer_tracker . handle_transfer ( referenced_tool_file , path_type . TOOL ) <EOL> def __upload_arbitrary_files ( self ) : <EOL> for path , name in self . arbitrary_files . items ( ) : <EOL> self . transfer_tracker . handle_transfer ( path , path_type . UNSTRUCTURED , name = name ) <EOL> def __upload_input_files ( self ) : <EOL> for input_file in self . input_files : <EOL> self . __upload_input_file ( input_file ) <EOL> self . __upload_input_extra_files ( input_file ) <EOL> def __upload_input_file ( self , input_file ) : <EOL> if self . __stage_input ( input_file ) : <EOL> if exists ( input_file ) : <EOL> self . transfer_tracker . handle_transfer ( input_file , path_type . INPUT ) <EOL> else : <EOL> message = "<STR_LIT>" + "<STR_LIT>" % input_file <EOL> log . debug ( message ) <EOL> def __upload_input_extra_files ( self , input_file ) : <EOL> files_path = "<STR_LIT>" % input_file [ <NUM_LIT:0> : - len ( "<STR_LIT>" ) ] <EOL> if exists ( files_path ) and self . __stage_input ( files_path ) : <EOL> for extra_file_name in directory_files ( files_path ) : <EOL> extra_file_path = join ( files_path , extra_file_name ) <EOL> remote_name = self . path_helper . remote_name ( relpath ( extra_file_path , dirname ( files_path ) ) ) <EOL> self . transfer_tracker . handle_transfer ( extra_file_path , path_type . INPUT , name = remote_name ) <EOL> def __upload_working_directory_files ( self ) : <EOL> working_directory_files = self . __working_directory_files ( ) <EOL> for working_directory_file in working_directory_files : <EOL> path = join ( self . working_directory , working_directory_file ) <EOL> self . transfer_tracker . handle_transfer ( path , path_type . WORKDIR ) <EOL> def __upload_metadata_directory_files ( self ) : <EOL> metadata_directory_files = self . __metadata_directory_files ( ) <EOL> for metadata_directory_file in metadata_directory_files : <EOL> path = join ( self . metadata_directory , metadata_directory_file ) <EOL> self . transfer_tracker . handle_transfer ( path , path_type . METADATA ) <EOL> def __working_directory_files ( self ) : <EOL> return self . __list_files ( self . working_directory ) <EOL> def __metadata_directory_files ( self ) : <EOL> return self . __list_files ( self . metadata_directory ) <EOL> def __list_files ( self , directory ) : <EOL> if directory and exists ( directory ) : <EOL> return [ f for f in listdir ( directory ) if isfile ( join ( directory , f ) ) ] <EOL> else : <EOL> return [ ] <EOL> def __initialize_version_file_rename ( self ) : <EOL> version_file = self . version_file <EOL> if version_file : <EOL> remote_path = self . path_helper . remote_join ( self . new_outputs_directory , COMMAND_VERSION_FILENAME ) <EOL> self . transfer_tracker . register_rewrite ( version_file , remote_path , path_type . OUTPUT ) <EOL> def __initialize_output_file_renames ( self ) : <EOL> for output_file in self . output_files : <EOL> remote_path = self . path_helper . remote_join ( self . new_outputs_directory , basename ( output_file ) ) <EOL> self . transfer_tracker . register_rewrite ( output_file , remote_path , path_type . OUTPUT ) <EOL> def __initialize_task_output_file_renames ( self ) : <EOL> for output_file in self . output_files : <EOL> name = basename ( output_file ) <EOL> task_file = join ( self . working_directory , name ) <EOL> remote_path = self . path_helper . remote_join ( self . new_working_directory , name ) <EOL> self . transfer_tracker . register_rewrite ( task_file , remote_path , path_type . OUTPUT_WORKDIR ) <EOL> def __initialize_config_file_renames ( self ) : <EOL> for config_file in self . config_files : <EOL> remote_path = self . path_helper . remote_join ( self . new_configs_directory , basename ( config_file ) ) <EOL> self . transfer_tracker . register_rewrite ( config_file , remote_path , path_type . CONFIG ) <EOL> def __handle_rewrites ( self ) : <EOL> """<STR_LIT>""" <EOL> self . transfer_tracker . rewrite_input_paths ( ) <EOL> def __upload_rewritten_config_files ( self ) : <EOL> for config_file , new_config_contents in self . job_inputs . config_files . items ( ) : <EOL> self . transfer_tracker . handle_transfer ( config_file , type = path_type . CONFIG , contents = new_config_contents ) <EOL> def get_command_line ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . job_inputs . command_line <EOL> def __stage_input ( self , file_path ) : <EOL> return ( not self . rewrite_paths ) or self . job_inputs . path_referenced ( file_path ) <EOL> class JobInputs ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , command_line , config_files ) : <EOL> self . command_line = command_line <EOL> self . config_files = { } <EOL> for config_file in config_files or [ ] : <EOL> config_contents = _read ( config_file ) <EOL> self . config_files [ config_file ] = config_contents <EOL> def find_pattern_references ( self , pattern ) : <EOL> referenced_files = set ( ) <EOL> for input_contents in self . __items ( ) : <EOL> referenced_files . update ( findall ( pattern , input_contents ) ) <EOL> return list ( referenced_files ) <EOL> def find_referenced_subfiles ( self , directory ) : <EOL> """<STR_LIT>""" <EOL> if directory is None : <EOL> return [ ] <EOL> pattern = r"<STR_LIT>" % ( directory , sep ) <EOL> return self . find_pattern_references ( pattern ) <EOL> def path_referenced ( self , path ) : <EOL> pattern = r"<STR_LIT:%s>" % path <EOL> found = False <EOL> for input_contents in self . __items ( ) : <EOL> if findall ( pattern , input_contents ) : <EOL> found = True <EOL> break <EOL> return found <EOL> def rewrite_paths ( self , local_path , remote_path ) : <EOL> """<STR_LIT>""" <EOL> self . __rewrite_command_line ( local_path , remote_path ) <EOL> self . __rewrite_config_files ( local_path , remote_path ) <EOL> def __rewrite_command_line ( self , local_path , remote_path ) : <EOL> self . command_line = self . command_line . replace ( local_path , remote_path ) <EOL> def __rewrite_config_files ( self , local_path , remote_path ) : <EOL> for config_file , contents in self . config_files . items ( ) : <EOL> self . config_files [ config_file ] = contents . replace ( local_path , remote_path ) <EOL> def __items ( self ) : <EOL> items = [ self . command_line ] <EOL> items . extend ( self . config_files . values ( ) ) <EOL> return items <EOL> class TransferTracker ( object ) : <EOL> def __init__ ( self , client , path_helper , action_mapper , job_inputs , rewrite_paths ) : <EOL> self . client = client <EOL> self . path_helper = path_helper <EOL> self . action_mapper = action_mapper <EOL> self . job_inputs = job_inputs <EOL> self . rewrite_paths = rewrite_paths <EOL> self . file_renames = { } <EOL> self . remote_staging_actions = [ ] <EOL> def handle_transfer ( self , path , type , name = None , contents = None ) : <EOL> action = self . __action_for_transfer ( path , type , contents ) <EOL> if action . staging_needed : <EOL> local_action = action . staging_action_local <EOL> if local_action : <EOL> response = self . client . put_file ( path , type , name = name , contents = contents , action_type = action . action_type ) <EOL> def get_path ( ) : <EOL> return response [ '<STR_LIT:path>' ] <EOL> else : <EOL> job_directory = self . client . job_directory <EOL> assert job_directory , "<STR_LIT>" % action <EOL> if not name : <EOL> name = basename ( path ) <EOL> self . __add_remote_staging_input ( action , name , type ) <EOL> def get_path ( ) : <EOL> return job_directory . calculate_path ( name , type ) <EOL> register = self . rewrite_paths or type == '<STR_LIT>' <EOL> if register : <EOL> self . register_rewrite ( path , get_path ( ) , type , force = True ) <EOL> elif self . rewrite_paths : <EOL> path_rewrite = action . path_rewrite ( self . path_helper ) <EOL> if path_rewrite : <EOL> self . register_rewrite ( path , path_rewrite , type , force = True ) <EOL> def __add_remote_staging_input ( self , action , name , type ) : <EOL> input_dict = dict ( <EOL> name = name , <EOL> type = type , <EOL> action = action . to_dict ( ) , <EOL> ) <EOL> self . remote_staging_actions . append ( input_dict ) <EOL> def __action_for_transfer ( self , path , type , contents ) : <EOL> if contents : <EOL> action = MessageAction ( contents = contents , client = self . client ) <EOL> else : <EOL> if not exists ( path ) : <EOL> message = "<STR_LIT>" % path <EOL> log . warn ( message ) <EOL> raise Exception ( message ) <EOL> action = self . __action ( path , type ) <EOL> return action <EOL> def register_rewrite ( self , local_path , remote_path , type , force = False ) : <EOL> action = self . __action ( local_path , type ) <EOL> if action . staging_needed or force : <EOL> self . file_renames [ local_path ] = remote_path <EOL> def rewrite_input_paths ( self ) : <EOL> """<STR_LIT>""" <EOL> for local_path , remote_path in self . file_renames . items ( ) : <EOL> self . job_inputs . rewrite_paths ( local_path , remote_path ) <EOL> def __action ( self , path , type ) : <EOL> return self . action_mapper . action ( path , type ) <EOL> def _read ( path ) : <EOL> """<STR_LIT>""" <EOL> input = open ( path , "<STR_LIT:r>" , encoding = "<STR_LIT:utf-8>" ) <EOL> try : <EOL> return input . read ( ) <EOL> finally : <EOL> input . close ( ) <EOL> __all__ = [ '<STR_LIT>' ] </s>
