- galaxy-commits - lists.galaxyproject.org

commit/galaxy-central: davebgx: Remove deprecated code that was importing removed deprecated code.
by commits-noreply＠bitbucket.org 08 Jul '14

08 Jul '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/558e521c3934/ Changeset: 558e521c3934 User: davebgx Date: 2014-07-08 21:22:47 Summary: Remove deprecated code that was importing removed deprecated code. Affected #: 1 file diff -r c6e4969dfcbb0a5ecf8ff7f8cdb8b8c727c4063a -r 558e521c3934f3b02c0fa31a8b93803d08aabf22 lib/galaxy/webapps/galaxy/api/genomes.py --- a/lib/galaxy/webapps/galaxy/api/genomes.py +++ b/lib/galaxy/webapps/galaxy/api/genomes.py @@ -1,7 +1,6 @@ from galaxy import web, util from galaxy.web.base.controller import BaseAPIController from galaxy.web.framework.helpers import is_true -from galaxy.webapps.galaxy.controllers.data_admin import build_param_dict as massage def get_id( base, format ): if format: @@ -41,77 +40,3 @@ else: rval = self.app.genomes.chroms( trans, dbkey=id, num=num, chrom=chrom, low=low ) return rval - - @web.expose_api - def create( self, trans, payload, **kwd ): - """ - POST /api/genomes - Download and/or index a genome. - - Parameters:: - liftover None or array of liftover url partial paths - dbkey DB key of the build to download - indexers POST array of indexers to run after downloading (indexers[] = first, indexers[] = second, ...) - longname a more descriptive name for the genome - func Allowed values: - 'download' Download and index - 'index' Index only - Returns:: - If no error: - dict( status: 'ok', job: <job ID> ) - - If error: - dict( status: 'error', error: <error message> ) - - """ - #??: Planned? - #Parameters:: - # dbkey DB key of the build to download, ignored unless 'UCSC' is specified as the source - # ncbi_name NCBI's genome identifier, ignored unless NCBI is specified as the source - # ensembl_dbkey Ensembl's genome identifier, ignored unless Ensembl is specified as the source - # url_dbkey DB key to use for this build, ignored unless URL is specified as the source - # source Data source for this build. Can be: UCSC, Ensembl, NCBI, URL - # indexers POST array of indexers to run after downloading (indexers[] = first, indexers[] = second, ...) - # func Allowed values: - # 'download' Download and index - # 'index' Index only - params = util.Params( payload ) - paramdict = massage( params, trans ) - func = params.get( 'func', 'download' ) - if func == 'download': - url = paramdict[ 'url' ] - liftover = paramdict[ 'liftover' ] - dbkey = paramdict[ 'dbkey' ] - indexers = paramdict[ 'indexers' ] - longname = paramdict[ 'longname' ] - jobid = trans.app.job_manager.deferred_job_queue.plugins['GenomeTransferPlugin'].create_job( trans, url, dbkey, longname, indexers ) - chainjob = [] - if liftover is not None: - for chain in liftover: - liftover_url = u'ftp://hgdownload.cse.ucsc.edu%s' % chain[0] - from_genome = chain[1] - to_genome = chain[2] - destfile = liftover_url.split('/')[-1].replace('.gz', '') - lochain = trans.app.job_manager.deferred_job_queue.plugins['LiftOverTransferPlugin'].create_job( trans, liftover_url, dbkey, from_genome, to_genome, destfile, jobid ) - chainjob.append( lochain ) - job = trans.app.job_manager.deferred_job_queue.plugins['GenomeTransferPlugin'].get_job_status( jobid ) - job.params['liftover'] = chainjob - trans.app.model.context.current.flush() - return dict( status='ok', job=jobid ) - elif func == 'index': - dbkey = paramdict[ 'dbkey' ] - indexer = [ params.get( 'indexer', None ) ] - longname = None - path = None - for build in trans.app.tool_data_tables.data_tables[ 'all_fasta' ].data: - if build[0] == dbkey: - longname = build[2] - path = build[3] - break - if longname is not None and indexer is not None and path is not None: - jobid = trans.app.job_manager.deferred_job_queue.plugins['GenomeIndexPlugin'].create_job( trans, path, indexer, dbkey, longname ) - return dict( status='ok', job=jobid ) - else: - return dict( status='error', error='Build not %s found in tool data table.' % dbkey ) - else: - return dict( status='error', error='Unkown function selected.' ) Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: dannon: Close branch edit_workflow
by commits-noreply＠bitbucket.org 08 Jul '14

08 Jul '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/df24fe1b4c97/ Changeset: df24fe1b4c97 Branch: edit_workflow User: dannon Date: 2014-07-08 20:00:24 Summary: Close branch edit_workflow Affected #: 0 files Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: dannon: Merged in saketkc/galaxy-central-310/edit_workflow (pull request #433)
by commits-noreply＠bitbucket.org 08 Jul '14

08 Jul '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/c6e4969dfcbb/ Changeset: c6e4969dfcbb User: dannon Date: 2014-07-08 20:00:24 Summary: Merged in saketkc/galaxy-central-310/edit_workflow (pull request #433) Fix: Allow editing workflows on the fly for radio buttons too Affected #: 1 file diff -r 158302d1405755306533ad44b2284acad31e421a -r c6e4969dfcbb0a5ecf8ff7f8cdb8b8c727c4063a templates/webapps/galaxy/workflow/run.mako --- a/templates/webapps/galaxy/workflow/run.mako +++ b/templates/webapps/galaxy/workflow/run.mako @@ -154,18 +154,27 @@ $(".workflow-edit-button").on("click",function(){ var state = $(this).attr("name"); - var stepToolBox = $(this).parent().find('input:not([class]), select:not([class])'); - var split_name=stepToolBox.attr("name").split("|"); + var stepToolBox = $(this).parent().find('input:not([class]):not([type="hidden"]), select:not([class])'); + var labels = $(this).parent().find('label'); + var split_name = stepToolBox.attr("name").split("|"); var step_id = split_name[0]; - var step_name = split_name[split_name.length-1]; - hidden_html = "<input type='hidden' name='"+step_id+"|__runtime__"+step_name+"' value='true' />"; + var step_name = split_name.slice(1, split_name.length).join("|"); + var hidden_html = "<input type='hidden' name='"+step_id+"|__runtime__"+step_name+"' value='true' />"; + var html = ""; if (state === "edit"){ stepToolBoxClone = stepToolBox.clone(); - stepToolBoxClone.attr({"name":step_id+"|"+step_name}); - stepToolBoxClone.show() + stepToolBoxClone.each(function(index){$(stepToolBoxClone[index]).attr({"name":step_id+"|"+step_name})}); + stepToolBoxClone.show(); + if (labels.length > 0){ + stepToolBoxClone.each(function(index){ + html += stepToolBoxClone[index].outerHTML + labels[index].outerHTML + "<br/>"}); + } + else{ + html = stepToolBoxClone.outerHTML(); + } $(this).parent().find(".editable").show(); $(this).parent().parent().find(".uneditable_field").hide(); - $(this).parent().find(".editable").html(stepToolBoxClone.outerHTML()+hidden_html); + $(this).parent().find(".editable").html(html+hidden_html); editingParameter($(this)); } else{ @@ -439,7 +448,7 @@ </span><span class="editable_field"><span class="editable"> - ${param.get_html_field( t, value, other_values).get_html( str(step.id) + "|" + "editable" + "|"+ prefix )} + ${param.get_html_field( t, value, other_values).get_html( str(step.id) + "|"+ prefix )} </span><i class="fa workflow-edit-button"></i> Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: 4 new changesets
by commits-noreply＠bitbucket.org 08 Jul '14

08 Jul '14

4 new commits in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/757412c63654/ Changeset: 757412c63654 Branch: edit_workflow User: saketkc Date: 2014-07-04 13:29:24 Summary: Fix: Allow editing workflows on the fly for radio buttons too. See https://bitbucket.org/galaxy/galaxy-central/pull-request/301/add-ability-to… Affected #: 1 file diff -r 4bcf8ca400aee4f72900de8a375707256f0ae369 -r 757412c63654ff16f6cd6a0b6ff776b25e0b5b03 templates/webapps/galaxy/workflow/run.mako --- a/templates/webapps/galaxy/workflow/run.mako +++ b/templates/webapps/galaxy/workflow/run.mako @@ -154,18 +154,28 @@ $(".workflow-edit-button").on("click",function(){ var state = $(this).attr("name"); - var stepToolBox = $(this).parent().find('input:not([class]), select:not([class])'); + var stepToolBox = $(this).parent().find('input:not([class]):not([type="hidden"]), select:not([class])'); + var labels = $(this).parent().find('label'); var split_name=stepToolBox.attr("name").split("|"); var step_id = split_name[0]; - var step_name = split_name[split_name.length-1]; + var step_name = split_name.slice(1, split_name.length).join("|"); hidden_html = "<input type='hidden' name='"+step_id+"|__runtime__"+step_name+"' value='true' />"; + var html = ""; if (state === "edit"){ stepToolBoxClone = stepToolBox.clone(); - stepToolBoxClone.attr({"name":step_id+"|"+step_name}); - stepToolBoxClone.show() + stepToolBoxClone.each(function(index){$(stepToolBoxClone[index]).attr({"name":step_id+"|"+step_name})}); + stepToolBoxClone.show(); + if (labels.length > 0){ + stepToolBoxClone.each(function(index){ + html += stepToolBoxClone[index].outerHTML + labels[index].outerHTML + "<br/>"}); + } + else{ + html = stepToolBoxClone.outerHTML(); + } $(this).parent().find(".editable").show(); $(this).parent().parent().find(".uneditable_field").hide(); - $(this).parent().find(".editable").html(stepToolBoxClone.outerHTML()+hidden_html); + console.log(hidden_html) + $(this).parent().find(".editable").html(html+hidden_html); editingParameter($(this)); } else{ @@ -439,7 +449,7 @@ </span><span class="editable_field"><span class="editable"> - ${param.get_html_field( t, value, other_values).get_html( str(step.id) + "|" + "editable" + "|"+ prefix )} + ${param.get_html_field( t, value, other_values).get_html( str(step.id) + "|"+ prefix )} </span><i class="fa workflow-edit-button"></i> https://bitbucket.org/galaxy/galaxy-central/commits/3a52ec2a1015/ Changeset: 3a52ec2a1015 Branch: edit_workflow User: saketkc Date: 2014-07-04 15:41:09 Summary: Make hidden_html local Affected #: 1 file diff -r 757412c63654ff16f6cd6a0b6ff776b25e0b5b03 -r 3a52ec2a1015de51b55ec109de711d9f3adaeb12 templates/webapps/galaxy/workflow/run.mako --- a/templates/webapps/galaxy/workflow/run.mako +++ b/templates/webapps/galaxy/workflow/run.mako @@ -156,10 +156,10 @@ var state = $(this).attr("name"); var stepToolBox = $(this).parent().find('input:not([class]):not([type="hidden"]), select:not([class])'); var labels = $(this).parent().find('label'); - var split_name=stepToolBox.attr("name").split("|"); + var split_name = stepToolBox.attr("name").split("|"); var step_id = split_name[0]; var step_name = split_name.slice(1, split_name.length).join("|"); - hidden_html = "<input type='hidden' name='"+step_id+"|__runtime__"+step_name+"' value='true' />"; + var hidden_html = "<input type='hidden' name='"+step_id+"|__runtime__"+step_name+"' value='true' />"; var html = ""; if (state === "edit"){ stepToolBoxClone = stepToolBox.clone(); https://bitbucket.org/galaxy/galaxy-central/commits/b609ee31197c/ Changeset: b609ee31197c Branch: edit_workflow User: saketkc Date: 2014-07-04 15:45:54 Summary: Remove console log Affected #: 1 file diff -r 3a52ec2a1015de51b55ec109de711d9f3adaeb12 -r b609ee31197cc9f1abe2fe8e2c355d14bfa9984a templates/webapps/galaxy/workflow/run.mako --- a/templates/webapps/galaxy/workflow/run.mako +++ b/templates/webapps/galaxy/workflow/run.mako @@ -174,7 +174,6 @@ } $(this).parent().find(".editable").show(); $(this).parent().parent().find(".uneditable_field").hide(); - console.log(hidden_html) $(this).parent().find(".editable").html(html+hidden_html); editingParameter($(this)); } https://bitbucket.org/galaxy/galaxy-central/commits/c6e4969dfcbb/ Changeset: c6e4969dfcbb User: dannon Date: 2014-07-08 20:00:24 Summary: Merged in saketkc/galaxy-central-310/edit_workflow (pull request #433) Fix: Allow editing workflows on the fly for radio buttons too Affected #: 1 file diff -r 158302d1405755306533ad44b2284acad31e421a -r c6e4969dfcbb0a5ecf8ff7f8cdb8b8c727c4063a templates/webapps/galaxy/workflow/run.mako --- a/templates/webapps/galaxy/workflow/run.mako +++ b/templates/webapps/galaxy/workflow/run.mako @@ -154,18 +154,27 @@ $(".workflow-edit-button").on("click",function(){ var state = $(this).attr("name"); - var stepToolBox = $(this).parent().find('input:not([class]), select:not([class])'); - var split_name=stepToolBox.attr("name").split("|"); + var stepToolBox = $(this).parent().find('input:not([class]):not([type="hidden"]), select:not([class])'); + var labels = $(this).parent().find('label'); + var split_name = stepToolBox.attr("name").split("|"); var step_id = split_name[0]; - var step_name = split_name[split_name.length-1]; - hidden_html = "<input type='hidden' name='"+step_id+"|__runtime__"+step_name+"' value='true' />"; + var step_name = split_name.slice(1, split_name.length).join("|"); + var hidden_html = "<input type='hidden' name='"+step_id+"|__runtime__"+step_name+"' value='true' />"; + var html = ""; if (state === "edit"){ stepToolBoxClone = stepToolBox.clone(); - stepToolBoxClone.attr({"name":step_id+"|"+step_name}); - stepToolBoxClone.show() + stepToolBoxClone.each(function(index){$(stepToolBoxClone[index]).attr({"name":step_id+"|"+step_name})}); + stepToolBoxClone.show(); + if (labels.length > 0){ + stepToolBoxClone.each(function(index){ + html += stepToolBoxClone[index].outerHTML + labels[index].outerHTML + "<br/>"}); + } + else{ + html = stepToolBoxClone.outerHTML(); + } $(this).parent().find(".editable").show(); $(this).parent().parent().find(".uneditable_field").hide(); - $(this).parent().find(".editable").html(stepToolBoxClone.outerHTML()+hidden_html); + $(this).parent().find(".editable").html(html+hidden_html); editingParameter($(this)); } else{ @@ -439,7 +448,7 @@ </span><span class="editable_field"><span class="editable"> - ${param.get_html_field( t, value, other_values).get_html( str(step.id) + "|" + "editable" + "|"+ prefix )} + ${param.get_html_field( t, value, other_values).get_html( str(step.id) + "|"+ prefix )} </span><i class="fa workflow-edit-button"></i> Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: 7 new changesets
by commits-noreply＠bitbucket.org 08 Jul '14

08 Jul '14

7 new commits in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/90a490a788c1/ Changeset: 90a490a788c1 User: dannon Date: 2014-06-20 16:16:40 Summary: Prevent possible XSS in reset_password. Affected #: 1 file diff -r c972ce1e3e8840309dceaa40ac6858c3156fdff3 -r 90a490a788c1718cd2abe9f5b65bdf8c6a081d82 lib/galaxy/webapps/galaxy/controllers/user.py --- a/lib/galaxy/webapps/galaxy/controllers/user.py +++ b/lib/galaxy/webapps/galaxy/controllers/user.py @@ -1092,7 +1092,7 @@ def reset_password( self, trans, email=None, **kwd ): if trans.app.config.smtp_server is None: return trans.show_error_message( "Mail is not configured for this Galaxy instance. Please contact your local Galaxy administrator." ) - message = util.restore_text( kwd.get( 'message', '' ) ) + message = util.sanitize_text(util.restore_text( kwd.get( 'message', '' ) )) status = 'done' if kwd.get( 'reset_password_button', False ): reset_user = trans.sa_session.query( trans.app.model.User ).filter( trans.app.model.User.table.c.email == email ).first() https://bitbucket.org/galaxy/galaxy-central/commits/7b4946192456/ Changeset: 7b4946192456 User: dannon Date: 2014-06-24 23:02:28 Summary: Merge. Affected #: 140 files diff -r 90a490a788c1718cd2abe9f5b65bdf8c6a081d82 -r 7b4946192456c1619829b6836a95716d8656817d job_conf.xml.sample_advanced --- a/job_conf.xml.sample_advanced +++ b/job_conf.xml.sample_advanced @@ -19,27 +19,30 @@ <param id="drmaa_library_path">/sge/lib/libdrmaa.so</param></plugin> - <plugin id="lwr" type="runner" load="galaxy.jobs.runners.lwr:LwrJobRunner"> -  -  + <plugin id="cli" type="runner" load="galaxy.jobs.runners.cli:ShellJobRunner" /> + <plugin id="condor" type="runner" load="galaxy.jobs.runners.condor:CondorJobRunner" /> + <plugin id="slurm" type="runner" load="galaxy.jobs.runners.slurm:SlurmJobRunner" /> +  + <plugin id="pulsar_rest" type="runner" load="galaxy.jobs.runners.pulsar:PulsarRESTJobRunner"> +  -  - </plugin> - <plugin id="amqp_lwr" type="runner" load="galaxy.jobs.runners.lwr:LwrJobRunner"> - <param id="url">amqp://guest:guest@localhost:5672//</param> -  + <plugin id="pulsar_mq" type="runner" load="galaxy.jobs.runners.pulsar:PulsarMQJobRunner"> +  + <param id="amqp_url">amqp://guest:guest@localhost:5672//</param> + <param id="galaxy_url">http://localhost:8080</param> -  + </plugin> - <plugin id="cli" type="runner" load="galaxy.jobs.runners.cli:ShellJobRunner" /> - <plugin id="condor" type="runner" load="galaxy.jobs.runners.condor:CondorJobRunner" /> - <plugin id="slurm" type="runner" load="galaxy.jobs.runners.slurm:SlurmJobRunner" /> + <plugin id="pulsar_legacy" type="runner" load="galaxy.jobs.runners.pulsar:PulsarLegacyJobRunner"> +  + </plugin></plugins><handlers default="handlers"> @@ -178,7 +189,7 @@ does trust tool's specified container - but also wants tool's not configured to run in a container the following option can provide a fallback. --> -  + </destination><destination id="pbs" runner="pbs" tags="mycluster"/> @@ -213,87 +224,71 @@ <param id="function">foo</param></destination> - <destination id="secure_lwr" runner="lwr"> - <param id="url">https://windowshost.examle.com:8913/</param> -  + <destination id="secure_pulsar_rest_dest" runner="pulsar_rest"> + <param id="url">https://examle.com:8913/</param> + <param id="private_token">123456789changeme</param> -  -  +  -  -  -  +  -  -  -  +  +  + </destination> - <destination id="amqp_lwr_dest" runner="amqp_lwr" > -  -  - <param id="jobs_directory">/path/to/remote/lwr/lwr_staging/</param> -  - <param id="default_file_action">remote_transfer</param></destination><destination id="ssh_torque" runner="cli"><param id="shell_plugin">SecureShell</param> diff -r 90a490a788c1718cd2abe9f5b65bdf8c6a081d82 -r 7b4946192456c1619829b6836a95716d8656817d lib/galaxy/jobs/handler.py --- a/lib/galaxy/jobs/handler.py +++ b/lib/galaxy/jobs/handler.py @@ -13,6 +13,7 @@ from galaxy import model from galaxy.util.sleeper import Sleeper from galaxy.jobs import JobWrapper, TaskWrapper, JobDestination +from galaxy.jobs.mapper import JobNotReadyException log = logging.getLogger( __name__ ) @@ -263,7 +264,7 @@ try: # Check the job's dependencies, requeue if they're not done. # Some of these states will only happen when using the in-memory job queue - job_state = self.__check_if_ready_to_run( job ) + job_state = self.__check_job_state( job ) if job_state == JOB_WAIT: new_waiting_jobs.append( job.id ) elif job_state == JOB_INPUT_ERROR: @@ -304,7 +305,7 @@ # Done with the session self.sa_session.remove() - def __check_if_ready_to_run( self, job ): + def __check_job_state( self, job ): """ Check if a job is ready to run by verifying that each of its input datasets is ready (specifically in the OK state). If any input dataset @@ -314,62 +315,97 @@ job can be dispatched. Otherwise, return JOB_WAIT indicating that input datasets are still being prepared. """ - # If tracking in the database, job.state is guaranteed to be NEW and the inputs are guaranteed to be OK if not self.track_jobs_in_database: - if job.state == model.Job.states.DELETED: - return JOB_DELETED - elif job.state == model.Job.states.ERROR: - return JOB_ADMIN_DELETED - for dataset_assoc in job.input_datasets + job.input_library_datasets: - idata = dataset_assoc.dataset - if not idata: - continue - # don't run jobs for which the input dataset was deleted - if idata.deleted: - self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s (file: %s) was deleted before the job started" % ( idata.hid, idata.file_name ) ) - return JOB_INPUT_DELETED - # an error in the input data causes us to bail immediately - elif idata.state == idata.states.ERROR: - self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s is in error state" % ( idata.hid ) ) - return JOB_INPUT_ERROR - elif idata.state == idata.states.FAILED_METADATA: - self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s failed to properly set metadata" % ( idata.hid ) ) - return JOB_INPUT_ERROR - elif idata.state != idata.states.OK and not ( idata.state == idata.states.SETTING_METADATA and job.tool_id is not None and job.tool_id == self.app.datatypes_registry.set_external_metadata_tool.id ): - # need to requeue - return JOB_WAIT + in_memory_not_ready_state = self.__verify_in_memory_job_inputs( job ) + if in_memory_not_ready_state: + return in_memory_not_ready_state + + # Else, if tracking in the database, job.state is guaranteed to be NEW and + # the inputs are guaranteed to be OK. + # Create the job wrapper so that the destination can be set - if job.id not in self.job_wrappers: - self.job_wrappers[job.id] = self.job_wrapper( job ) - # Cause the job_destination to be set and cached by the mapper + job_id = job.id + job_wrapper = self.job_wrappers.get( job_id, None ) + if not job_wrapper: + job_wrapper = self.job_wrapper( job ) + self.job_wrappers[ job_id ] = job_wrapper + + # If state == JOB_READY, assume job_destination also set - otherwise + # in case of various error or cancelled states do not assume + # destination has been set. + state, job_destination = self.__verify_job_ready( job, job_wrapper ) + + if state == JOB_READY: + # PASS. increase usage by one job (if caching) so that multiple jobs aren't dispatched on this queue iteration + self.increase_running_job_count(job.user_id, job_destination.id ) + return state + + def __verify_job_ready( self, job, job_wrapper ): + """ Compute job destination and verify job is ready at that + destination by checking job limits and quota. If this method + return a job state of JOB_READY - it MUST also return a job + destination. + """ + job_destination = None try: - self.job_wrappers[job.id].job_destination + # Cause the job_destination to be set and cached by the mapper + job_destination = job_wrapper.job_destination + except JobNotReadyException as e: + job_state = e.job_state or JOB_WAIT + return job_state, None except Exception, e: failure_message = getattr( e, 'failure_message', DEFAULT_JOB_PUT_FAILURE_MESSAGE ) if failure_message == DEFAULT_JOB_PUT_FAILURE_MESSAGE: log.exception( 'Failed to generate job destination' ) else: log.debug( "Intentionally failing job with message (%s)" % failure_message ) - self.job_wrappers[job.id].fail( failure_message ) - return JOB_ERROR + job_wrapper.fail( failure_message ) + return JOB_ERROR, job_destination # job is ready to run, check limits # TODO: these checks should be refactored to minimize duplication and made more modular/pluggable - state = self.__check_destination_jobs( job, self.job_wrappers[job.id] ) + state = self.__check_destination_jobs( job, job_wrapper ) if state == JOB_READY: - state = self.__check_user_jobs( job, self.job_wrappers[job.id] ) + state = self.__check_user_jobs( job, job_wrapper ) if state == JOB_READY and self.app.config.enable_quotas: quota = self.app.quota_agent.get_quota( job.user ) if quota is not None: try: usage = self.app.quota_agent.get_usage( user=job.user, history=job.history ) if usage > quota: - return JOB_USER_OVER_QUOTA + return JOB_USER_OVER_QUOTA, job_destination except AssertionError, e: pass # No history, should not happen with an anon user - if state == JOB_READY: - # PASS. increase usage by one job (if caching) so that multiple jobs aren't dispatched on this queue iteration - self.increase_running_job_count(job.user_id, self.job_wrappers[job.id].job_destination.id) - return state + return state, job_destination + + def __verify_in_memory_job_inputs( self, job ): + """ Perform the same checks that happen via SQL for in-memory managed + jobs. + """ + if job.state == model.Job.states.DELETED: + return JOB_DELETED + elif job.state == model.Job.states.ERROR: + return JOB_ADMIN_DELETED + for dataset_assoc in job.input_datasets + job.input_library_datasets: + idata = dataset_assoc.dataset + if not idata: + continue + # don't run jobs for which the input dataset was deleted + if idata.deleted: + self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s (file: %s) was deleted before the job started" % ( idata.hid, idata.file_name ) ) + return JOB_INPUT_DELETED + # an error in the input data causes us to bail immediately + elif idata.state == idata.states.ERROR: + self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s is in error state" % ( idata.hid ) ) + return JOB_INPUT_ERROR + elif idata.state == idata.states.FAILED_METADATA: + self.job_wrappers.pop(job.id, self.job_wrapper( job )).fail( "input data %s failed to properly set metadata" % ( idata.hid ) ) + return JOB_INPUT_ERROR + elif idata.state != idata.states.OK and not ( idata.state == idata.states.SETTING_METADATA and job.tool_id is not None and job.tool_id == self.app.datatypes_registry.set_external_metadata_tool.id ): + # need to requeue + return JOB_WAIT + + # All inputs ready to go. + return None def __clear_job_count( self ): self.user_job_count = None diff -r 90a490a788c1718cd2abe9f5b65bdf8c6a081d82 -r 7b4946192456c1619829b6836a95716d8656817d lib/galaxy/jobs/mapper.py --- a/lib/galaxy/jobs/mapper.py +++ b/lib/galaxy/jobs/mapper.py @@ -16,6 +16,13 @@ self.failure_message = failure_message +class JobNotReadyException( Exception ): + + def __init__( self, job_state=None, message=None ): + self.job_state = job_state + self.message = message + + class JobRunnerMapper( object ): """ This class is responsible to managing the mapping of jobs diff -r 90a490a788c1718cd2abe9f5b65bdf8c6a081d82 -r 7b4946192456c1619829b6836a95716d8656817d lib/galaxy/jobs/runners/pulsar.py --- /dev/null +++ b/lib/galaxy/jobs/runners/pulsar.py @@ -0,0 +1,707 @@ +from __future__ import absolute_import # Need to import pulsar_client absolutely. + +import logging + +from galaxy import model +from galaxy.jobs.runners import AsynchronousJobState, AsynchronousJobRunner +from galaxy.jobs import ComputeEnvironment +from galaxy.jobs import JobDestination +from galaxy.jobs.command_factory import build_command +from galaxy.tools.deps import dependencies +from galaxy.util import string_as_bool_or_none +from galaxy.util.bunch import Bunch +from galaxy.util import specs + +import errno +from time import sleep +import os + +from pulsar.client import build_client_manager +from pulsar.client import url_to_destination_params +from pulsar.client import finish_job as pulsar_finish_job +from pulsar.client import submit_job as pulsar_submit_job +from pulsar.client import ClientJobDescription +from pulsar.client import PulsarOutputs +from pulsar.client import ClientOutputs +from pulsar.client import PathMapper + +log = logging.getLogger( __name__ ) + +__all__ = [ 'PulsarLegacyJobRunner', 'PulsarRESTJobRunner', 'PulsarMQJobRunner' ] + +NO_REMOTE_GALAXY_FOR_METADATA_MESSAGE = "Pulsar misconfiguration - Pulsar client configured to set metadata remotely, but remote Pulsar isn't properly configured with a galaxy_home directory." +NO_REMOTE_DATATYPES_CONFIG = "Pulsar client is configured to use remote datatypes configuration when setting metadata externally, but Pulsar is not configured with this information. Defaulting to datatypes_conf.xml." +GENERIC_REMOTE_ERROR = "Failed to communicate with remote job server." + +# Is there a good way to infer some default for this? Can only use +# url_for from web threads. https://gist.github.com/jmchilton/9098762 +DEFAULT_GALAXY_URL = "http://localhost:8080" + +PULSAR_PARAM_SPECS = dict( + transport=dict( + map=specs.to_str_or_none, + valid=specs.is_in("urllib", "curl", None), + default=None + ), + cache=dict( + map=specs.to_bool_or_none, + default=None, + ), + amqp_url=dict( + map=specs.to_str_or_none, + default=None, + ), + galaxy_url=dict( + map=specs.to_str_or_none, + default=DEFAULT_GALAXY_URL, + ), + manager=dict( + map=specs.to_str_or_none, + default=None, + ), + amqp_consumer_timeout=dict( + map=lambda val: None if val == "None" else float(val), + default=None, + ), + amqp_connect_ssl_ca_certs=dict( + map=specs.to_str_or_none, + default=None, + ), + amqp_connect_ssl_keyfile=dict( + map=specs.to_str_or_none, + default=None, + ), + amqp_connect_ssl_certfile=dict( + map=specs.to_str_or_none, + default=None, + ), + amqp_connect_ssl_cert_reqs=dict( + map=specs.to_str_or_none, + default=None, + ), + # http://kombu.readthedocs.org/en/latest/reference/kombu.html#kombu.Producer.… + amqp_publish_retry=dict( + map=specs.to_bool, + default=False, + ), + amqp_publish_priority=dict( + map=int, + valid=lambda x: 0 <= x and x <= 9, + default=0, + ), + # http://kombu.readthedocs.org/en/latest/reference/kombu.html#kombu.Exchange.… + amqp_publish_delivery_mode=dict( + map=str, + valid=specs.is_in("transient", "persistent"), + default="persistent", + ), + amqp_publish_retry_max_retries=dict( + map=int, + default=None, + ), + amqp_publish_retry_interval_start=dict( + map=int, + default=None, + ), + amqp_publish_retry_interval_step=dict( + map=int, + default=None, + ), + amqp_publish_retry_interval_max=dict( + map=int, + default=None, + ), +) + + +PARAMETER_SPECIFICATION_REQUIRED = object() +PARAMETER_SPECIFICATION_IGNORED = object() + + +class PulsarJobRunner( AsynchronousJobRunner ): + """ + Pulsar Job Runner + """ + runner_name = "PulsarJobRunner" + + def __init__( self, app, nworkers, **kwds ): + """Start the job runner """ + super( PulsarJobRunner, self ).__init__( app, nworkers, runner_param_specs=PULSAR_PARAM_SPECS, **kwds ) + self._init_worker_threads() + galaxy_url = self.runner_params.galaxy_url + if galaxy_url: + galaxy_url = galaxy_url.rstrip("/") + self.galaxy_url = galaxy_url + self.__init_client_manager() + self._monitor() + + def _monitor( self ): + # Extension point allow MQ variant to setup callback instead + self._init_monitor_thread() + + def __init_client_manager( self ): + client_manager_kwargs = {} + for kwd in 'manager', 'cache', 'transport': + client_manager_kwargs[ kwd ] = self.runner_params[ kwd ] + for kwd in self.runner_params.keys(): + if kwd.startswith( 'amqp_' ): + client_manager_kwargs[ kwd ] = self.runner_params[ kwd ] + self.client_manager = build_client_manager(**client_manager_kwargs) + + def url_to_destination( self, url ): + """Convert a legacy URL to a job destination""" + return JobDestination( runner="pulsar", params=url_to_destination_params( url ) ) + + def check_watched_item(self, job_state): + try: + client = self.get_client_from_state(job_state) + status = client.get_status() + except Exception: + # An orphaned job was put into the queue at app startup, so remote server went down + # either way we are done I guess. + self.mark_as_finished(job_state) + return None + job_state = self._update_job_state_for_status(job_state, status) + return job_state + + def _update_job_state_for_status(self, job_state, pulsar_status): + if pulsar_status == "complete": + self.mark_as_finished(job_state) + return None + if pulsar_status == "failed": + self.fail_job(job_state) + return None + if pulsar_status == "running" and not job_state.running: + job_state.running = True + job_state.job_wrapper.change_state( model.Job.states.RUNNING ) + return job_state + + def queue_job(self, job_wrapper): + job_destination = job_wrapper.job_destination + self._populate_parameter_defaults( job_destination ) + + command_line, client, remote_job_config, compute_environment = self.__prepare_job( job_wrapper, job_destination ) + + if not command_line: + return + + try: + dependencies_description = PulsarJobRunner.__dependencies_description( client, job_wrapper ) + rewrite_paths = not PulsarJobRunner.__rewrite_parameters( client ) + unstructured_path_rewrites = {} + if compute_environment: + unstructured_path_rewrites = compute_environment.unstructured_path_rewrites + + client_job_description = ClientJobDescription( + command_line=command_line, + input_files=self.get_input_files(job_wrapper), + client_outputs=self.__client_outputs(client, job_wrapper), + working_directory=job_wrapper.working_directory, + tool=job_wrapper.tool, + config_files=job_wrapper.extra_filenames, + dependencies_description=dependencies_description, + env=client.env, + rewrite_paths=rewrite_paths, + arbitrary_files=unstructured_path_rewrites, + ) + job_id = pulsar_submit_job(client, client_job_description, remote_job_config) + log.info("Pulsar job submitted with job_id %s" % job_id) + job_wrapper.set_job_destination( job_destination, job_id ) + job_wrapper.change_state( model.Job.states.QUEUED ) + except Exception: + job_wrapper.fail( "failure running job", exception=True ) + log.exception("failure running job %d" % job_wrapper.job_id) + return + + pulsar_job_state = AsynchronousJobState() + pulsar_job_state.job_wrapper = job_wrapper + pulsar_job_state.job_id = job_id + pulsar_job_state.old_state = True + pulsar_job_state.running = False + pulsar_job_state.job_destination = job_destination + self.monitor_job(pulsar_job_state) + + def __prepare_job(self, job_wrapper, job_destination): + """ Build command-line and Pulsar client for this job. """ + command_line = None + client = None + remote_job_config = None + compute_environment = None + try: + client = self.get_client_from_wrapper(job_wrapper) + tool = job_wrapper.tool + remote_job_config = client.setup(tool.id, tool.version) + rewrite_parameters = PulsarJobRunner.__rewrite_parameters( client ) + prepare_kwds = {} + if rewrite_parameters: + compute_environment = PulsarComputeEnvironment( client, job_wrapper, remote_job_config ) + prepare_kwds[ 'compute_environment' ] = compute_environment + job_wrapper.prepare( **prepare_kwds ) + self.__prepare_input_files_locally(job_wrapper) + remote_metadata = PulsarJobRunner.__remote_metadata( client ) + dependency_resolution = PulsarJobRunner.__dependency_resolution( client ) + metadata_kwds = self.__build_metadata_configuration(client, job_wrapper, remote_metadata, remote_job_config) + remote_command_params = dict( + working_directory=remote_job_config['working_directory'], + metadata_kwds=metadata_kwds, + dependency_resolution=dependency_resolution, + ) + remote_working_directory = remote_job_config['working_directory'] + # TODO: Following defs work for Pulsar, always worked for Pulsar but should be + # calculated at some other level. + remote_job_directory = os.path.abspath(os.path.join(remote_working_directory, os.path.pardir)) + remote_tool_directory = os.path.abspath(os.path.join(remote_job_directory, "tool_files")) + container = self._find_container( + job_wrapper, + compute_working_directory=remote_working_directory, + compute_tool_directory=remote_tool_directory, + compute_job_directory=remote_job_directory, + ) + command_line = build_command( + self, + job_wrapper=job_wrapper, + container=container, + include_metadata=remote_metadata, + include_work_dir_outputs=False, + remote_command_params=remote_command_params, + ) + except Exception: + job_wrapper.fail( "failure preparing job", exception=True ) + log.exception("failure running job %d" % job_wrapper.job_id) + + # If we were able to get a command line, run the job + if not command_line: + job_wrapper.finish( '', '' ) + + return command_line, client, remote_job_config, compute_environment + + def __prepare_input_files_locally(self, job_wrapper): + """Run task splitting commands locally.""" + prepare_input_files_cmds = getattr(job_wrapper, 'prepare_input_files_cmds', None) + if prepare_input_files_cmds is not None: + for cmd in prepare_input_files_cmds: # run the commands to stage the input files + if 0 != os.system(cmd): + raise Exception('Error running file staging command: %s' % cmd) + job_wrapper.prepare_input_files_cmds = None # prevent them from being used in-line + + def _populate_parameter_defaults( self, job_destination ): + updated = False + params = job_destination.params + for key, value in self.destination_defaults.iteritems(): + if key in params: + if value is PARAMETER_SPECIFICATION_IGNORED: + log.warn( "Pulsar runner in selected configuration ignores parameter %s" % key ) + continue + #if self.runner_params.get( key, None ): + # # Let plugin define defaults for some parameters - + # # for instance that way jobs_directory can be + # # configured next to AMQP url (where it belongs). + # params[ key ] = self.runner_params[ key ] + # continue + + if not value: + continue + + if value is PARAMETER_SPECIFICATION_REQUIRED: + raise Exception( "Pulsar destination does not define required parameter %s" % key ) + elif value is not PARAMETER_SPECIFICATION_IGNORED: + params[ key ] = value + updated = True + return updated + + def get_output_files(self, job_wrapper): + output_paths = job_wrapper.get_output_fnames() + return [ str( o ) for o in output_paths ] # Force job_path from DatasetPath objects. + + def get_input_files(self, job_wrapper): + input_paths = job_wrapper.get_input_paths() + return [ str( i ) for i in input_paths ] # Force job_path from DatasetPath objects. + + def get_client_from_wrapper(self, job_wrapper): + job_id = job_wrapper.job_id + if hasattr(job_wrapper, 'task_id'): + job_id = "%s_%s" % (job_id, job_wrapper.task_id) + params = job_wrapper.job_destination.params.copy() + for key, value in params.iteritems(): + if value: + params[key] = model.User.expand_user_properties( job_wrapper.get_job().user, value ) + + env = getattr( job_wrapper.job_destination, "env", [] ) + return self.get_client( params, job_id, env ) + + def get_client_from_state(self, job_state): + job_destination_params = job_state.job_destination.params + job_id = job_state.job_id + return self.get_client( job_destination_params, job_id ) + + def get_client( self, job_destination_params, job_id, env=[] ): + # Cannot use url_for outside of web thread. + #files_endpoint = url_for( controller="job_files", job_id=encoded_job_id ) + + encoded_job_id = self.app.security.encode_id(job_id) + job_key = self.app.security.encode_id( job_id, kind="jobs_files" ) + files_endpoint = "%s/api/jobs/%s/files?job_key=%s" % ( + self.galaxy_url, + encoded_job_id, + job_key + ) + get_client_kwds = dict( + job_id=str( job_id ), + files_endpoint=files_endpoint, + env=env + ) + return self.client_manager.get_client( job_destination_params, **get_client_kwds ) + + def finish_job( self, job_state ): + stderr = stdout = '' + job_wrapper = job_state.job_wrapper + try: + client = self.get_client_from_state(job_state) + run_results = client.full_status() + remote_working_directory = run_results.get("working_directory", None) + stdout = run_results.get('stdout', '') + stderr = run_results.get('stderr', '') + exit_code = run_results.get('returncode', None) + pulsar_outputs = PulsarOutputs.from_status_response(run_results) + # Use Pulsar client code to transfer/copy files back + # and cleanup job if needed. + completed_normally = \ + job_wrapper.get_state() not in [ model.Job.states.ERROR, model.Job.states.DELETED ] + cleanup_job = self.app.config.cleanup_job + client_outputs = self.__client_outputs(client, job_wrapper) + finish_args = dict( client=client, + job_completed_normally=completed_normally, + cleanup_job=cleanup_job, + client_outputs=client_outputs, + pulsar_outputs=pulsar_outputs ) + failed = pulsar_finish_job( **finish_args ) + if failed: + job_wrapper.fail("Failed to find or download one or more job outputs from remote server.", exception=True) + except Exception: + message = GENERIC_REMOTE_ERROR + job_wrapper.fail( message, exception=True ) + log.exception("failure finishing job %d" % job_wrapper.job_id) + return + if not PulsarJobRunner.__remote_metadata( client ): + self._handle_metadata_externally( job_wrapper, resolve_requirements=True ) + # Finish the job + try: + job_wrapper.finish( + stdout, + stderr, + exit_code, + remote_working_directory=remote_working_directory + ) + except Exception: + log.exception("Job wrapper finish method failed") + job_wrapper.fail("Unable to finish job", exception=True) + + def fail_job( self, job_state ): + """ + Seperated out so we can use the worker threads for it. + """ + self.stop_job( self.sa_session.query( self.app.model.Job ).get( job_state.job_wrapper.job_id ) ) + job_state.job_wrapper.fail( getattr( job_state, "fail_message", GENERIC_REMOTE_ERROR ) ) + + def check_pid( self, pid ): + try: + os.kill( pid, 0 ) + return True + except OSError, e: + if e.errno == errno.ESRCH: + log.debug( "check_pid(): PID %d is dead" % pid ) + else: + log.warning( "check_pid(): Got errno %s when attempting to check PID %d: %s" % ( errno.errorcode[e.errno], pid, e.strerror ) ) + return False + + def stop_job( self, job ): + #if our local job has JobExternalOutputMetadata associated, then our primary job has to have already finished + job_ext_output_metadata = job.get_external_output_metadata() + if job_ext_output_metadata: + pid = job_ext_output_metadata[0].job_runner_external_pid # every JobExternalOutputMetadata has a pid set, we just need to take from one of them + if pid in [ None, '' ]: + log.warning( "stop_job(): %s: no PID in database for job, unable to stop" % job.id ) + return + pid = int( pid ) + if not self.check_pid( pid ): + log.warning( "stop_job(): %s: PID %d was already dead or can't be signaled" % ( job.id, pid ) ) + return + for sig in [ 15, 9 ]: + try: + os.killpg( pid, sig ) + except OSError, e: + log.warning( "stop_job(): %s: Got errno %s when attempting to signal %d to PID %d: %s" % ( job.id, errno.errorcode[e.errno], sig, pid, e.strerror ) ) + return # give up + sleep( 2 ) + if not self.check_pid( pid ): + log.debug( "stop_job(): %s: PID %d successfully killed with signal %d" % ( job.id, pid, sig ) ) + return + else: + log.warning( "stop_job(): %s: PID %d refuses to die after signaling TERM/KILL" % ( job.id, pid ) ) + else: + # Remote kill + pulsar_url = job.job_runner_name + job_id = job.job_runner_external_id + log.debug("Attempt remote Pulsar kill of job with url %s and id %s" % (pulsar_url, job_id)) + client = self.get_client(job.destination_params, job_id) + client.kill() + + def recover( self, job, job_wrapper ): + """Recovers jobs stuck in the queued/running state when Galaxy started""" + job_state = self._job_state( job, job_wrapper ) + job_wrapper.command_line = job.get_command_line() + state = job.get_state() + if state in [model.Job.states.RUNNING, model.Job.states.QUEUED]: + log.debug( "(Pulsar/%s) is still in running state, adding to the Pulsar queue" % ( job.get_id()) ) + job_state.old_state = True + job_state.running = state == model.Job.states.RUNNING + self.monitor_queue.put( job_state ) + + def shutdown( self ): + super( PulsarJobRunner, self ).shutdown() + self.client_manager.shutdown() + + def _job_state( self, job, job_wrapper ): + job_state = AsynchronousJobState() + # TODO: Determine why this is set when using normal message queue updates + # but not CLI submitted MQ updates... + raw_job_id = job.get_job_runner_external_id() or job_wrapper.job_id + job_state.job_id = str( raw_job_id ) + job_state.runner_url = job_wrapper.get_job_runner_url() + job_state.job_destination = job_wrapper.job_destination + job_state.job_wrapper = job_wrapper + return job_state + + def __client_outputs( self, client, job_wrapper ): + work_dir_outputs = self.get_work_dir_outputs( job_wrapper ) + output_files = self.get_output_files( job_wrapper ) + client_outputs = ClientOutputs( + working_directory=job_wrapper.working_directory, + work_dir_outputs=work_dir_outputs, + output_files=output_files, + version_file=job_wrapper.get_version_string_path(), + ) + return client_outputs + + @staticmethod + def __dependencies_description( pulsar_client, job_wrapper ): + dependency_resolution = PulsarJobRunner.__dependency_resolution( pulsar_client ) + remote_dependency_resolution = dependency_resolution == "remote" + if not remote_dependency_resolution: + return None + requirements = job_wrapper.tool.requirements or [] + installed_tool_dependencies = job_wrapper.tool.installed_tool_dependencies or [] + return dependencies.DependenciesDescription( + requirements=requirements, + installed_tool_dependencies=installed_tool_dependencies, + ) + + @staticmethod + def __dependency_resolution( pulsar_client ): + dependency_resolution = pulsar_client.destination_params.get( "dependency_resolution", "local" ) + if dependency_resolution not in ["none", "local", "remote"]: + raise Exception("Unknown dependency_resolution value encountered %s" % dependency_resolution) + return dependency_resolution + + @staticmethod + def __remote_metadata( pulsar_client ): + remote_metadata = string_as_bool_or_none( pulsar_client.destination_params.get( "remote_metadata", False ) ) + return remote_metadata + + @staticmethod + def __use_remote_datatypes_conf( pulsar_client ): + """ When setting remote metadata, use integrated datatypes from this + Galaxy instance or use the datatypes config configured via the remote + Pulsar. + + Both options are broken in different ways for same reason - datatypes + may not match. One can push the local datatypes config to the remote + server - but there is no guarentee these datatypes will be defined + there. Alternatively, one can use the remote datatype config - but + there is no guarentee that it will contain all the datatypes available + to this Galaxy. + """ + use_remote_datatypes = string_as_bool_or_none( pulsar_client.destination_params.get( "use_remote_datatypes", False ) ) + return use_remote_datatypes + + @staticmethod + def __rewrite_parameters( pulsar_client ): + return string_as_bool_or_none( pulsar_client.destination_params.get( "rewrite_parameters", False ) ) or False + + def __build_metadata_configuration(self, client, job_wrapper, remote_metadata, remote_job_config): + metadata_kwds = {} + if remote_metadata: + remote_system_properties = remote_job_config.get("system_properties", {}) + remote_galaxy_home = remote_system_properties.get("galaxy_home", None) + if not remote_galaxy_home: + raise Exception(NO_REMOTE_GALAXY_FOR_METADATA_MESSAGE) + metadata_kwds['exec_dir'] = remote_galaxy_home + outputs_directory = remote_job_config['outputs_directory'] + configs_directory = remote_job_config['configs_directory'] + working_directory = remote_job_config['working_directory'] + # For metadata calculation, we need to build a list of of output + # file objects with real path indicating location on Galaxy server + # and false path indicating location on compute server. Since the + # Pulsar disables from_work_dir copying as part of the job command + # line we need to take the list of output locations on the Pulsar + # server (produced by self.get_output_files(job_wrapper)) and for + # each work_dir output substitute the effective path on the Pulsar + # server relative to the remote working directory as the + # false_path to send the metadata command generation module. + work_dir_outputs = self.get_work_dir_outputs(job_wrapper, job_working_directory=working_directory) + outputs = [Bunch(false_path=os.path.join(outputs_directory, os.path.basename(path)), real_path=path) for path in self.get_output_files(job_wrapper)] + for output in outputs: + for pulsar_workdir_path, real_path in work_dir_outputs: + if real_path == output.real_path: + output.false_path = pulsar_workdir_path + metadata_kwds['output_fnames'] = outputs + metadata_kwds['compute_tmp_dir'] = working_directory + metadata_kwds['config_root'] = remote_galaxy_home + default_config_file = os.path.join(remote_galaxy_home, 'universe_wsgi.ini') + metadata_kwds['config_file'] = remote_system_properties.get('galaxy_config_file', default_config_file) + metadata_kwds['dataset_files_path'] = remote_system_properties.get('galaxy_dataset_files_path', None) + if PulsarJobRunner.__use_remote_datatypes_conf( client ): + remote_datatypes_config = remote_system_properties.get('galaxy_datatypes_config_file', None) + if not remote_datatypes_config: + log.warn(NO_REMOTE_DATATYPES_CONFIG) + remote_datatypes_config = os.path.join(remote_galaxy_home, 'datatypes_conf.xml') + metadata_kwds['datatypes_config'] = remote_datatypes_config + else: + integrates_datatypes_config = self.app.datatypes_registry.integrated_datatypes_configs + # Ensure this file gets pushed out to the remote config dir. + job_wrapper.extra_filenames.append(integrates_datatypes_config) + + metadata_kwds['datatypes_config'] = os.path.join(configs_directory, os.path.basename(integrates_datatypes_config)) + return metadata_kwds + + +class PulsarLegacyJobRunner( PulsarJobRunner ): + destination_defaults = dict( + rewrite_parameters="false", + dependency_resolution="local", + ) + + +class PulsarMQJobRunner( PulsarJobRunner ): + destination_defaults = dict( + default_file_action="remote_transfer", + rewrite_parameters="true", + dependency_resolution="remote", + jobs_directory=PARAMETER_SPECIFICATION_REQUIRED, + url=PARAMETER_SPECIFICATION_IGNORED, + private_token=PARAMETER_SPECIFICATION_IGNORED + ) + + def _monitor( self ): + # This is a message queue driven runner, don't monitor + # just setup required callback. + self.client_manager.ensure_has_status_update_callback(self.__async_update) + + def __async_update( self, full_status ): + job_id = None + try: + job_id = full_status[ "job_id" ] + job, job_wrapper = self.app.job_manager.job_handler.job_queue.job_pair_for_id( job_id ) + job_state = self._job_state( job, job_wrapper ) + self._update_job_state_for_status(job_state, full_status[ "status" ] ) + except Exception: + log.exception( "Failed to update Pulsar job status for job_id %s" % job_id ) + raise + # Nothing else to do? - Attempt to fail the job? + + +class PulsarRESTJobRunner( PulsarJobRunner ): + destination_defaults = dict( + default_file_action="transfer", + rewrite_parameters="true", + dependency_resolution="remote", + url=PARAMETER_SPECIFICATION_REQUIRED, + ) + + +class PulsarComputeEnvironment( ComputeEnvironment ): + + def __init__( self, pulsar_client, job_wrapper, remote_job_config ): + self.pulsar_client = pulsar_client + self.job_wrapper = job_wrapper + self.local_path_config = job_wrapper.default_compute_environment() + self.unstructured_path_rewrites = {} + # job_wrapper.prepare is going to expunge the job backing the following + # computations, so precalculate these paths. + self._wrapper_input_paths = self.local_path_config.input_paths() + self._wrapper_output_paths = self.local_path_config.output_paths() + self.path_mapper = PathMapper(pulsar_client, remote_job_config, self.local_path_config.working_directory()) + self._config_directory = remote_job_config[ "configs_directory" ] + self._working_directory = remote_job_config[ "working_directory" ] + self._sep = remote_job_config[ "system_properties" ][ "separator" ] + self._tool_dir = remote_job_config[ "tools_directory" ] + version_path = self.local_path_config.version_path() + new_version_path = self.path_mapper.remote_version_path_rewrite(version_path) + if new_version_path: + version_path = new_version_path + self._version_path = version_path + + def output_paths( self ): + local_output_paths = self._wrapper_output_paths + + results = [] + for local_output_path in local_output_paths: + wrapper_path = str( local_output_path ) + remote_path = self.path_mapper.remote_output_path_rewrite( wrapper_path ) + results.append( self._dataset_path( local_output_path, remote_path ) ) + return results + + def input_paths( self ): + local_input_paths = self._wrapper_input_paths + + results = [] + for local_input_path in local_input_paths: + wrapper_path = str( local_input_path ) + # This will over-copy in some cases. For instance in the case of task + # splitting, this input will be copied even though only the work dir + # input will actually be used. + remote_path = self.path_mapper.remote_input_path_rewrite( wrapper_path ) + results.append( self._dataset_path( local_input_path, remote_path ) ) + return results + + def _dataset_path( self, local_dataset_path, remote_path ): + remote_extra_files_path = None + if remote_path: + remote_extra_files_path = "%s_files" % remote_path[ 0:-len( ".dat" ) ] + return local_dataset_path.with_path_for_job( remote_path, remote_extra_files_path ) + + def working_directory( self ): + return self._working_directory + + def config_directory( self ): + return self._config_directory + + def new_file_path( self ): + return self.working_directory() # Problems with doing this? + + def sep( self ): + return self._sep + + def version_path( self ): + return self._version_path + + def rewriter( self, parameter_value ): + unstructured_path_rewrites = self.unstructured_path_rewrites + if parameter_value in unstructured_path_rewrites: + # Path previously mapped, use previous mapping. + return unstructured_path_rewrites[ parameter_value ] + if parameter_value in unstructured_path_rewrites.itervalues(): + # Path is a rewritten remote path (this might never occur, + # consider dropping check...) + return parameter_value + + rewrite, new_unstructured_path_rewrites = self.path_mapper.check_for_arbitrary_rewrite( parameter_value ) + if rewrite: + unstructured_path_rewrites.update(new_unstructured_path_rewrites) + return rewrite + else: + # Did need to rewrite, use original path or value. + return parameter_value + + def unstructured_path_rewriter( self ): + return self.rewriter diff -r 90a490a788c1718cd2abe9f5b65bdf8c6a081d82 -r 7b4946192456c1619829b6836a95716d8656817d lib/galaxy/jobs/runners/util/__init__.py --- a/lib/galaxy/jobs/runners/util/__init__.py +++ b/lib/galaxy/jobs/runners/util/__init__.py @@ -1,7 +1,7 @@ """ This module and its submodules contains utilities for running external processes and interfacing with job managers. This module should contain -functionality shared between Galaxy and the LWR. +functionality shared between Galaxy and the Pulsar. """ from galaxy.util.bunch import Bunch diff -r 90a490a788c1718cd2abe9f5b65bdf8c6a081d82 -r 7b4946192456c1619829b6836a95716d8656817d lib/galaxy/jobs/runners/util/cli/factory.py --- a/lib/galaxy/jobs/runners/util/cli/factory.py +++ b/lib/galaxy/jobs/runners/util/cli/factory.py @@ -5,7 +5,7 @@ ) code_dir = 'lib' except ImportError: - from lwr.managers.util.cli import ( + from pulsar.managers.util.cli import ( CliInterface, split_params ) diff -r 90a490a788c1718cd2abe9f5b65bdf8c6a081d82 -r 7b4946192456c1619829b6836a95716d8656817d lib/galaxy/jobs/runners/util/cli/job/slurm.py --- a/lib/galaxy/jobs/runners/util/cli/job/slurm.py +++ b/lib/galaxy/jobs/runners/util/cli/job/slurm.py @@ -5,7 +5,7 @@ from galaxy.model import Job job_states = Job.states except ImportError: - # Not in Galaxy, map Galaxy job states to LWR ones. + # Not in Galaxy, map Galaxy job states to Pulsar ones. from galaxy.util import enum job_states = enum(RUNNING='running', OK='complete', QUEUED='queued') diff -r 90a490a788c1718cd2abe9f5b65bdf8c6a081d82 -r 7b4946192456c1619829b6836a95716d8656817d lib/galaxy/jobs/runners/util/cli/job/torque.py --- a/lib/galaxy/jobs/runners/util/cli/job/torque.py +++ b/lib/galaxy/jobs/runners/util/cli/job/torque.py @@ -7,7 +7,7 @@ from galaxy.model import Job job_states = Job.states except ImportError: - # Not in Galaxy, map Galaxy job states to LWR ones. + # Not in Galaxy, map Galaxy job states to Pulsar ones. from galaxy.util import enum job_states = enum(RUNNING='running', OK='complete', QUEUED='queued') diff -r 90a490a788c1718cd2abe9f5b65bdf8c6a081d82 -r 7b4946192456c1619829b6836a95716d8656817d lib/galaxy/objectstore/__init__.py --- a/lib/galaxy/objectstore/__init__.py +++ b/lib/galaxy/objectstore/__init__.py @@ -623,9 +623,9 @@ elif store == 'irods': from .rods import IRODSObjectStore return IRODSObjectStore(config=config, config_xml=config_xml) - elif store == 'lwr': - from .lwr import LwrObjectStore - return LwrObjectStore(config=config, config_xml=config_xml) + elif store == 'pulsar': + from .pulsar import PulsarObjectStore + return PulsarObjectStore(config=config, config_xml=config_xml) else: log.error("Unrecognized object store definition: {0}".format(store)) diff -r 90a490a788c1718cd2abe9f5b65bdf8c6a081d82 -r 7b4946192456c1619829b6836a95716d8656817d lib/galaxy/objectstore/lwr.py --- a/lib/galaxy/objectstore/lwr.py +++ /dev/null @@ -1,76 +0,0 @@ -from __future__ import absolute_import # Need to import lwr_client absolutely. -from ..objectstore import ObjectStore -try: - from galaxy.jobs.runners.lwr_client.manager import ObjectStoreClientManager -except ImportError: - from lwr.lwr_client.manager import ObjectStoreClientManager - - -class LwrObjectStore(ObjectStore): - """ - Object store implementation that delegates to a remote LWR server. - - This may be more aspirational than practical for now, it would be good to - Galaxy to a point that a handler thread could be setup that doesn't attempt - to access the disk files returned by a (this) object store - just passing - them along to the LWR unmodified. That modification - along with this - implementation and LWR job destinations would then allow Galaxy to fully - manage jobs on remote servers with completely different mount points. - - This implementation should be considered beta and may be dropped from - Galaxy at some future point or significantly modified. - """ - - def __init__(self, config, config_xml): - self.lwr_client = self.__build_lwr_client(config_xml) - - def exists(self, obj, **kwds): - return self.lwr_client.exists(**self.__build_kwds(obj, **kwds)) - - def file_ready(self, obj, **kwds): - return self.lwr_client.file_ready(**self.__build_kwds(obj, **kwds)) - - def create(self, obj, **kwds): - return self.lwr_client.create(**self.__build_kwds(obj, **kwds)) - - def empty(self, obj, **kwds): - return self.lwr_client.empty(**self.__build_kwds(obj, **kwds)) - - def size(self, obj, **kwds): - return self.lwr_client.size(**self.__build_kwds(obj, **kwds)) - - def delete(self, obj, **kwds): - return self.lwr_client.delete(**self.__build_kwds(obj, **kwds)) - - # TODO: Optimize get_data. - def get_data(self, obj, **kwds): - return self.lwr_client.get_data(**self.__build_kwds(obj, **kwds)) - - def get_filename(self, obj, **kwds): - return self.lwr_client.get_filename(**self.__build_kwds(obj, **kwds)) - - def update_from_file(self, obj, **kwds): - return self.lwr_client.update_from_file(**self.__build_kwds(obj, **kwds)) - - def get_store_usage_percent(self): - return self.lwr_client.get_store_usage_percent() - - def get_object_url(self, obj, extra_dir=None, extra_dir_at_root=False, alt_name=None): - return None - - def __build_kwds(self, obj, **kwds): - kwds['object_id'] = obj.id - return kwds - pass - - def __build_lwr_client(self, config_xml): - url = config_xml.get("url") - private_token = config_xml.get("private_token", None) - transport = config_xml.get("transport", None) - manager_options = dict(transport=transport) - client_options = dict(url=url, private_token=private_token) - lwr_client = ObjectStoreClientManager(**manager_options).get_client(client_options) - return lwr_client - - def shutdown(self): - pass diff -r 90a490a788c1718cd2abe9f5b65bdf8c6a081d82 -r 7b4946192456c1619829b6836a95716d8656817d lib/galaxy/objectstore/pulsar.py --- /dev/null +++ b/lib/galaxy/objectstore/pulsar.py @@ -0,0 +1,73 @@ +from __future__ import absolute_import # Need to import pulsar_client absolutely. +from ..objectstore import ObjectStore +from pulsar.client.manager import ObjectStoreClientManager + + +class PulsarObjectStore(ObjectStore): + """ + Object store implementation that delegates to a remote Pulsar server. + + This may be more aspirational than practical for now, it would be good to + Galaxy to a point that a handler thread could be setup that doesn't attempt + to access the disk files returned by a (this) object store - just passing + them along to the Pulsar unmodified. That modification - along with this + implementation and Pulsar job destinations would then allow Galaxy to fully + manage jobs on remote servers with completely different mount points. + + This implementation should be considered beta and may be dropped from + Galaxy at some future point or significantly modified. + """ + + def __init__(self, config, config_xml): + self.pulsar_client = self.__build_pulsar_client(config_xml) + + def exists(self, obj, **kwds): + return self.pulsar_client.exists(**self.__build_kwds(obj, **kwds)) + + def file_ready(self, obj, **kwds): + return self.pulsar_client.file_ready(**self.__build_kwds(obj, **kwds)) + + def create(self, obj, **kwds): + return self.pulsar_client.create(**self.__build_kwds(obj, **kwds)) + + def empty(self, obj, **kwds): + return self.pulsar_client.empty(**self.__build_kwds(obj, **kwds)) + + def size(self, obj, **kwds): + return self.pulsar_client.size(**self.__build_kwds(obj, **kwds)) + + def delete(self, obj, **kwds): + return self.pulsar_client.delete(**self.__build_kwds(obj, **kwds)) + + # TODO: Optimize get_data. + def get_data(self, obj, **kwds): + return self.pulsar_client.get_data(**self.__build_kwds(obj, **kwds)) + + def get_filename(self, obj, **kwds): + return self.pulsar_client.get_filename(**self.__build_kwds(obj, **kwds)) + + def update_from_file(self, obj, **kwds): + return self.pulsar_client.update_from_file(**self.__build_kwds(obj, **kwds)) + + def get_store_usage_percent(self): + return self.pulsar_client.get_store_usage_percent() + + def get_object_url(self, obj, extra_dir=None, extra_dir_at_root=False, alt_name=None): + return None + + def __build_kwds(self, obj, **kwds): + kwds['object_id'] = obj.id + return kwds + pass + + def __build_pulsar_client(self, config_xml): + url = config_xml.get("url") + private_token = config_xml.get("private_token", None) + transport = config_xml.get("transport", None) + manager_options = dict(transport=transport) + client_options = dict(url=url, private_token=private_token) + pulsar_client = ObjectStoreClientManager(**manager_options).get_client(client_options) + return pulsar_client + + def shutdown(self): + pass diff -r 90a490a788c1718cd2abe9f5b65bdf8c6a081d82 -r 7b4946192456c1619829b6836a95716d8656817d lib/galaxy/tools/__init__.py --- a/lib/galaxy/tools/__init__.py +++ b/lib/galaxy/tools/__init__.py @@ -1264,7 +1264,7 @@ executable = self.version_string_cmd.split()[0] abs_executable = os.path.abspath(os.path.join(self.tool_dir, executable)) command_line = self.version_string_cmd.replace(executable, abs_executable, 1) - self.version_string_cmd = self.interpreter + " " + command_line + self.version_string_cmd = version_cmd_interpreter + " " + command_line # Parallelism for tasks, read from tool config. parallelism = root.find("parallelism") if parallelism is not None and parallelism.get("method"): diff -r 90a490a788c1718cd2abe9f5b65bdf8c6a081d82 -r 7b4946192456c1619829b6836a95716d8656817d lib/galaxy/tools/deps/dependencies.py --- a/lib/galaxy/tools/deps/dependencies.py +++ b/lib/galaxy/tools/deps/dependencies.py @@ -8,7 +8,7 @@ related context required to resolve dependencies via the ToolShedPackageDependencyResolver. - This is meant to enable remote resolution of dependencies, by the LWR or + This is meant to enable remote resolution of dependencies, by the Pulsar or other potential remote execution mechanisms. """ @@ -39,7 +39,7 @@ @staticmethod def _toolshed_install_dependency_from_dict(as_dict): - # Rather than requiring full models in LWR, just use simple objects + # Rather than requiring full models in Pulsar, just use simple objects # containing only properties and associations used to resolve # dependencies for tool execution. repository_object = bunch.Bunch( diff -r 90a490a788c1718cd2abe9f5b65bdf8c6a081d82 -r 7b4946192456c1619829b6836a95716d8656817d lib/galaxy/util/__init__.py --- a/lib/galaxy/util/__init__.py +++ b/lib/galaxy/util/__init__.py @@ -417,17 +417,16 @@ return slug_base -def in_directory( file, directory ): +def in_directory( file, directory, local_path_module=os.path ): """ Return true, if the common prefix of both is equal to directory e.g. /a/b/c/d.rst and directory is /a/b, the common prefix is /a/b """ # Make both absolute. - directory = os.path.abspath( directory ) - file = os.path.abspath( file ) - - return os.path.commonprefix( [ file, directory ] ) == directory + directory = local_path_module.abspath(directory) + file = local_path_module.abspath(file) + return local_path_module.commonprefix([file, directory]) == directory def merge_sorted_iterables( operator, *iterables ): diff -r 90a490a788c1718cd2abe9f5b65bdf8c6a081d82 -r 7b4946192456c1619829b6836a95716d8656817d lib/galaxy/webapps/galaxy/controllers/admin_toolshed.py --- a/lib/galaxy/webapps/galaxy/controllers/admin_toolshed.py +++ b/lib/galaxy/webapps/galaxy/controllers/admin_toolshed.py @@ -1359,9 +1359,8 @@ else: # Entering this else block occurs only if the tool_shed_repository does not include any valid tools. if install_repository_dependencies: - repository_dependencies = \ - repository_dependency_util.get_repository_dependencies_for_installed_tool_shed_repository( trans.app, - tool_shed_repository ) + repository_dependencies = rdm.get_repository_dependencies_for_installed_tool_shed_repository( trans.app, + tool_shed_repository ) else: repository_dependencies = None if metadata: @@ -1579,9 +1578,9 @@ raw_text = common_util.tool_shed_get( trans.app, tool_shed_url, url ) readme_files_dict = json.from_json_string( raw_text ) tool_dependencies = metadata.get( 'tool_dependencies', None ) - repository_dependencies = \ - repository_dependency_util.get_repository_dependencies_for_installed_tool_shed_repository( trans.app, - tool_shed_repository ) + rdm = RepositoryDependencyManager( trans.app ) + repository_dependencies = rdm.get_repository_dependencies_for_installed_tool_shed_repository( trans.app, + tool_shed_repository ) repo_info_dict = \ repository_maintenance_util.create_repo_info_dict( app=trans.app, repository_clone_url=repository_clone_url, diff -r 90a490a788c1718cd2abe9f5b65bdf8c6a081d82 -r 7b4946192456c1619829b6836a95716d8656817d lib/galaxy/webapps/tool_shed/api/repositories.py --- a/lib/galaxy/webapps/tool_shed/api/repositories.py +++ b/lib/galaxy/webapps/tool_shed/api/repositories.py @@ -10,14 +10,14 @@ from galaxy.web.base.controller import BaseAPIController from galaxy.web.base.controller import HTTPBadRequest from galaxy.web.framework.helpers import time_ago +from tool_shed.capsule import capsule_manager import tool_shed.repository_types.util as rt_util -import tool_shed.util.shed_util_common as suc from tool_shed.util import basic_util from tool_shed.util import encoding_util from tool_shed.util import hg_util -from tool_shed.util import import_util from tool_shed.util import metadata_util from tool_shed.util import repository_maintenance_util +from tool_shed.util import shed_util_common as suc from tool_shed.util import tool_util log = logging.getLogger( __name__ ) @@ -248,10 +248,14 @@ except tarfile.ReadError, e: log.debug( 'Error opening capsule file %s: %s' % ( str( capsule_file_name ), str( e ) ) ) return {} + irm = capsule_manager.ImportRepositoryManager( trans.app, + trans.request.host, + trans.user, + trans.user_is_admin() ) capsule_dict[ 'tar_archive' ] = tar_archive capsule_dict[ 'capsule_file_name' ] = capsule_file_name - capsule_dict = import_util.extract_capsule_files( **capsule_dict ) - capsule_dict = import_util.validate_capsule( **capsule_dict ) + capsule_dict = irm.extract_capsule_files( **capsule_dict ) + capsule_dict = irm.validate_capsule( **capsule_dict ) status = capsule_dict.get( 'status', 'error' ) if status == 'error': log.debug( 'The capsule contents are invalid and cannot be imported:<br/>%s' % \ @@ -263,15 +267,12 @@ return {} file_path = encoding_util.tool_shed_decode( encoded_file_path ) export_info_file_path = os.path.join( file_path, 'export_info.xml' ) - export_info_dict = import_util.get_export_info_dict( export_info_file_path ) + export_info_dict = irm.get_export_info_dict( export_info_file_path ) manifest_file_path = os.path.join( file_path, 'manifest.xml' ) # The manifest.xml file has already been validated, so no error_message should be returned here. - repository_info_dicts, error_message = import_util.get_repository_info_from_manifest( manifest_file_path ) + repository_info_dicts, error_message = irm.get_repository_info_from_manifest( manifest_file_path ) # Determine the status for each exported repository archive contained within the capsule. - repository_status_info_dicts = import_util.get_repository_status_from_tool_shed( trans.app, - trans.user, - trans.user_is_admin(), - repository_info_dicts ) + repository_status_info_dicts = irm.get_repository_status_from_tool_shed( repository_info_dicts ) # Generate a list of repository name / import results message tuples for display after the capsule is imported. import_results_tups = [] # Only create repositories that do not yet exist and that the current user is authorized to create. The @@ -280,12 +281,9 @@ # Add the capsule_file_name and encoded_file_path to the repository_status_info_dict. repository_status_info_dict[ 'capsule_file_name' ] = capsule_file_name repository_status_info_dict[ 'encoded_file_path' ] = encoded_file_path - import_results_tups = import_util.create_repository_and_import_archive( trans.app, - trans.request.host, - trans.user, - repository_status_info_dict, - import_results_tups ) - import_util.check_status_and_reset_downloadable( trans.app, import_results_tups ) + import_results_tups = irm.create_repository_and_import_archive( repository_status_info_dict, + import_results_tups ) + irm.check_status_and_reset_downloadable( import_results_tups ) basic_util.remove_dir( file_path ) # NOTE: the order of installation is defined in import_results_tups, but order will be lost # when transferred to return_dict. diff -r 90a490a788c1718cd2abe9f5b65bdf8c6a081d82 -r 7b4946192456c1619829b6836a95716d8656817d lib/galaxy/webapps/tool_shed/api/repository_revisions.py --- a/lib/galaxy/webapps/tool_shed/api/repository_revisions.py +++ b/lib/galaxy/webapps/tool_shed/api/repository_revisions.py @@ -5,7 +5,7 @@ from galaxy import util from galaxy.model.orm import and_, not_, select from galaxy.web.base.controller import BaseAPIController, HTTPBadRequest -from tool_shed.util import export_util +from tool_shed.capsule import capsule_manager from tool_shed.util import hg_util import tool_shed.util.shed_util_common as suc @@ -53,14 +53,14 @@ log.debug( error_message ) return None, error_message repository_id = trans.security.encode_id( repository.id ) - return export_util.export_repository( trans, - tool_shed_url, - repository_id, - str( repository.name ), - changeset_revision, - file_type, - export_repository_dependencies, - api=True ) + erm = capsule_manager.ExportRepositoryManager( app=trans.app, + user=trans.user, + tool_shed_url=tool_shed_url, + repository=repository, + changeset_revision=changeset_revision, + export_repository_dependencies=export_repository_dependencies, + using_api=True ) + return erm.export_repository() def __get_value_mapper( self, trans ): value_mapper = { 'id' : trans.security.encode_id, diff -r 90a490a788c1718cd2abe9f5b65bdf8c6a081d82 -r 7b4946192456c1619829b6836a95716d8656817d lib/galaxy/webapps/tool_shed/controllers/repository.py --- a/lib/galaxy/webapps/tool_shed/controllers/repository.py +++ b/lib/galaxy/webapps/tool_shed/controllers/repository.py @@ -14,20 +14,19 @@ from galaxy.web.framework.helpers import grids from galaxy.util import json from galaxy.model.orm import and_ -import tool_shed.util.shed_util_common as suc +from tool_shed.capsule import capsule_manager from tool_shed.util import basic_util from tool_shed.util import common_util from tool_shed.util import container_util from tool_shed.util import encoding_util -from tool_shed.util import export_util from tool_shed.util import hg_util -from tool_shed.util import import_util from tool_shed.util import metadata_util from tool_shed.util import readme_util from tool_shed.util import repository_dependency_util from tool_shed.util import repository_maintenance_util from tool_shed.util import review_util from tool_shed.util import search_util +from tool_shed.util import shed_util_common as suc from tool_shed.util import tool_dependency_util from tool_shed.util import tool_util from tool_shed.util import workflow_util @@ -1149,7 +1148,7 @@ # server account's .hgrc file to include the following setting: # [web] # allow_archive = bz2, gz, zip - file_type_str = export_util.get_file_type_str( changeset_revision, file_type ) + file_type_str = basic_util.get_file_type_str( changeset_revision, file_type ) repository.times_downloaded += 1 trans.sa_session.add( repository ) trans.sa_session.flush() @@ -1172,14 +1171,14 @@ file_type = 'gz' export_repository_dependencies = CheckboxField.is_checked( export_repository_dependencies ) tool_shed_url = web.url_for( '/', qualified=True ) - repositories_archive, error_message = export_util.export_repository( trans, - tool_shed_url, - repository_id, - str( repository.name ), - changeset_revision, - file_type, - export_repository_dependencies, - api=False ) + erm = capsule_manager.ExportRepositoryManager( app=trans.app, + user=trans.user, + tool_shed_url=tool_shed_url, + repository=repository, + changeset_revision=changeset_revision, + export_repository_dependencies=export_repository_dependencies, + using_api=False ) + repositories_archive, error_message = erm.export_repository() repositories_archive_filename = os.path.basename( repositories_archive.name ) if error_message: message = error_message @@ -1968,15 +1967,16 @@ encoded_file_path = kwd.get( 'encoded_file_path', None ) file_path = encoding_util.tool_shed_decode( encoded_file_path ) export_info_file_path = os.path.join( file_path, 'export_info.xml' ) - export_info_dict = import_util.get_export_info_dict( export_info_file_path ) + irm = capsule_manager.ImportRepositoryManager( trans.app, + trans.request.host, + trans.user, + trans.user_is_admin() ) + export_info_dict = irm.get_export_info_dict( export_info_file_path ) manifest_file_path = os.path.join( file_path, 'manifest.xml' ) # The manifest.xml file has already been validated, so no error_message should be returned here. - repository_info_dicts, error_message = import_util.get_repository_info_from_manifest( manifest_file_path ) + repository_info_dicts, error_message = irm.get_repository_info_from_manifest( manifest_file_path ) # Determine the status for each exported repository archive contained within the capsule. - repository_status_info_dicts = import_util.get_repository_status_from_tool_shed( trans.app, - trans.user, - trans.user_is_admin(), - repository_info_dicts ) + repository_status_info_dicts = irm.get_repository_status_from_tool_shed( repository_info_dicts ) if 'import_capsule_button' in kwd: # Generate a list of repository name / import results message tuples for display after the capsule is imported. import_results_tups = [] @@ -1986,12 +1986,9 @@ # Add the capsule_file_name and encoded_file_path to the repository_status_info_dict. repository_status_info_dict[ 'capsule_file_name' ] = capsule_file_name repository_status_info_dict[ 'encoded_file_path' ] = encoded_file_path - import_results_tups = import_util.create_repository_and_import_archive( trans.app, - trans.request.host, - trans.user, - repository_status_info_dict, - import_results_tups ) - import_util.check_status_and_reset_downloadable( trans.app, import_results_tups ) + import_results_tups = irm.create_repository_and_import_archive( repository_status_info_dict, + import_results_tups ) + irm.check_status_and_reset_downloadable( import_results_tups ) basic_util.remove_dir( file_path ) return trans.fill_template( '/webapps/tool_shed/repository/import_capsule_results.mako', export_info_dict=export_info_dict, @@ -3080,13 +3077,17 @@ status = kwd.get( 'status', 'done' ) url = kwd.get( 'url', '' ) if 'upload_capsule_button' in kwd: - capsule_dict = import_util.upload_capsule( **kwd ) + irm = capsule_manager.ImportRepositoryManager( trans.app, + trans.request.host, + trans.user, + trans.user_is_admin() ) + capsule_dict = irm.upload_capsule( **kwd ) status = capsule_dict.get( 'status', 'error' ) if status == 'error': message = capsule_dict.get( 'error_message', '' ) else: - capsule_dict = import_util.extract_capsule_files( **capsule_dict ) - capsule_dict = import_util.validate_capsule( **capsule_dict ) + capsule_dict = irm.extract_capsule_files( **capsule_dict ) + capsule_dict = irm.validate_capsule( **capsule_dict ) status = capsule_dict.get( 'status', 'error' ) if status == 'ok': return trans.response.send_redirect( web.url_for( controller='repository', diff -r 90a490a788c1718cd2abe9f5b65bdf8c6a081d82 -r 7b4946192456c1619829b6836a95716d8656817d lib/galaxy/webapps/tool_shed/controllers/upload.py --- a/lib/galaxy/webapps/tool_shed/controllers/upload.py +++ b/lib/galaxy/webapps/tool_shed/controllers/upload.py @@ -8,13 +8,14 @@ from galaxy import util from galaxy import web from galaxy.datatypes import checkers +from tool_shed.dependencies import dependency_manager import tool_shed.repository_types.util as rt_util -import tool_shed.util.shed_util_common as suc from tool_shed.util import basic_util from tool_shed.util import commit_util from tool_shed.util import hg_util from tool_shed.util import metadata_util from tool_shed.util import repository_dependency_util +from tool_shed.util import shed_util_common as suc from tool_shed.util import tool_dependency_util from tool_shed.util import tool_util from tool_shed.util import xml_util @@ -95,6 +96,8 @@ uploaded_file_filename = os.path.split( file_data.filename )[ -1 ] isempty = os.path.getsize( os.path.abspath( uploaded_file_name ) ) == 0 if uploaded_file or uploaded_directory: + rdah = dependency_manager.RepositoryDependencyAttributeHandler( trans.app, unpopulate=False ) + tdah = dependency_manager.ToolDependencyAttributeHandler( trans.app, unpopulate=False ) ok = True isgzip = False isbz2 = False @@ -124,6 +127,8 @@ if istar: ok, message, files_to_remove, content_alert_str, undesirable_dirs_removed, undesirable_files_removed = \ self.upload_tar( trans, + rdah, + tdah, repository, tar, uploaded_file, @@ -134,6 +139,8 @@ elif uploaded_directory: ok, message, files_to_remove, content_alert_str, undesirable_dirs_removed, undesirable_files_removed = \ self.upload_directory( trans, + rdah, + tdah, repository, uploaded_directory, upload_point, @@ -164,12 +171,9 @@ full_path = os.path.abspath( os.path.join( repo_dir, uploaded_file_filename ) ) # Move some version of the uploaded file to the load_point within the repository hierarchy. if uploaded_file_filename in [ rt_util.REPOSITORY_DEPENDENCY_DEFINITION_FILENAME ]: - # Inspect the contents of the file to see if changeset_revision values are missing and if so, - # set them appropriately. - altered, root_elem, error_message = \ - commit_util.handle_repository_dependencies_definition( trans.app, - uploaded_file_name, - unpopulate=False ) + # Inspect the contents of the file to see if toolshed or changeset_revision attributes + # are missing and if so, set them appropriately. + altered, root_elem, error_message = rdah.handle_tag_attributes( uploaded_file_name ) if error_message: ok = False message = error_message @@ -179,12 +183,10 @@ shutil.move( tmp_filename, full_path ) else: shutil.move( uploaded_file_name, full_path ) - elif uploaded_file_filename in [ rt_util.REPOSITORY_DEPENDENCY_DEFINITION_FILENAME, - rt_util.TOOL_DEPENDENCY_DEFINITION_FILENAME ]: + elif uploaded_file_filename in [ rt_util.TOOL_DEPENDENCY_DEFINITION_FILENAME ]: # Inspect the contents of the file to see if changeset_revision values are # missing and if so, set them appropriately. - altered, root_elem, error_message = \ - commit_util.handle_tool_dependencies_definition( trans.app, uploaded_file_name ) + altered, root_elem, error_message = tdah.handle_tag_attributes( uploaded_file_name ) if error_message: ok = False message = error_message @@ -333,7 +335,7 @@ message=message, status=status ) - def upload_directory( self, trans, repository, uploaded_directory, upload_point, remove_repo_files_not_in_tar, + def upload_directory( self, trans, rdah, tdah, repository, uploaded_directory, upload_point, remove_repo_files_not_in_tar, commit_message, new_repo_alert ): repo_dir = repository.repo_path( trans.app ) repo = hg_util.get_repo_for_repository( trans.app, repository=None, repo_path=repo_dir, create=False ) @@ -364,20 +366,18 @@ if ok: uploaded_file_name = os.path.abspath( os.path.join( root, uploaded_file ) ) if os.path.split( uploaded_file_name )[ -1 ] == rt_util.REPOSITORY_DEPENDENCY_DEFINITION_FILENAME: - # Inspect the contents of the file to see if changeset_revision values are missing and - # if so, set them appropriately. - altered, root_elem, error_message = \ - commit_util.handle_repository_dependencies_definition( trans.app, - uploaded_file_name, - unpopulate=False ) + # Inspect the contents of the file to see if toolshed or changeset_revision + # attributes are missing and if so, set them appropriately. + altered, root_elem, error_message = rdah.handle_tag_attributes( uploaded_file_name ) if error_message: return False, error_message, [], '', [], [] elif altered: tmp_filename = xml_util.create_and_write_tmp_file( root_elem ) shutil.move( tmp_filename, uploaded_file_name ) elif os.path.split( uploaded_file_name )[ -1 ] == rt_util.TOOL_DEPENDENCY_DEFINITION_FILENAME: - # Inspect the contents of the file to see if changeset_revision values are missing and if so, set them appropriately. - altered, root_elem, error_message = commit_util.handle_tool_dependencies_definition( trans.app, uploaded_file_name ) + # Inspect the contents of the file to see if toolshed or changeset_revision + # attributes are missing and if so, set them appropriately. + altered, root_elem, error_message = tdah.handle_tag_attributes( uploaded_file_name ) if error_message: return False, error_message, [], '', [], [] if altered: @@ -406,7 +406,8 @@ undesirable_dirs_removed, undesirable_files_removed ) - def upload_tar( self, trans, repository, tar, uploaded_file, upload_point, remove_repo_files_not_in_tar, commit_message, new_repo_alert ): + def upload_tar( self, trans, rdah, tdah, repository, tar, uploaded_file, upload_point, remove_repo_files_not_in_tar, + commit_message, new_repo_alert ): # Upload a tar archive of files. repo_dir = repository.repo_path( trans.app ) repo = hg_util.get_repo_for_repository( trans.app, repository=None, repo_path=repo_dir, create=False ) @@ -442,18 +443,18 @@ for filename in filenames_in_archive: uploaded_file_name = os.path.join( full_path, filename ) if os.path.split( uploaded_file_name )[ -1 ] == rt_util.REPOSITORY_DEPENDENCY_DEFINITION_FILENAME: - # Inspect the contents of the file to see if changeset_revision values are missing and if so, set them appropriately. - altered, root_elem, error_message = commit_util.handle_repository_dependencies_definition( trans.app, - uploaded_file_name, - unpopulate=False ) + # Inspect the contents of the file to see if toolshed or changeset_revision attributes + # are missing and if so, set them appropriately. + altered, root_elem, error_message = rdah.handle_tag_attributes( uploaded_file_name ) if error_message: return False, error_message, [], '', [], [] elif altered: tmp_filename = xml_util.create_and_write_tmp_file( root_elem ) shutil.move( tmp_filename, uploaded_file_name ) elif os.path.split( uploaded_file_name )[ -1 ] == rt_util.TOOL_DEPENDENCY_DEFINITION_FILENAME: - # Inspect the contents of the file to see if changeset_revision values are missing and if so, set them appropriately. - altered, root_elem, error_message = commit_util.handle_tool_dependencies_definition( trans.app, uploaded_file_name ) + # Inspect the contents of the file to see if toolshed or changeset_revision + # attributes are missing and if so, set them appropriately. + altered, root_elem, error_message = tdah.handle_tag_attributes( uploaded_file_name ) if error_message: return False, error_message, [], '', [], [] if altered: diff -r 90a490a788c1718cd2abe9f5b65bdf8c6a081d82 -r 7b4946192456c1619829b6836a95716d8656817d lib/galaxy/webapps/tool_shed/framework/middleware/hg.py --- a/lib/galaxy/webapps/tool_shed/framework/middleware/hg.py +++ b/lib/galaxy/webapps/tool_shed/framework/middleware/hg.py @@ -11,7 +11,6 @@ from galaxy.util import asbool from galaxy.util.hash_util import new_secure_hash -from tool_shed.util import commit_util from tool_shed.util import hg_util import tool_shed.repository_types.util as rt_util @@ -54,13 +53,15 @@ connection = engine.connect() path_info = environ[ 'PATH_INFO' ].lstrip( '/' ) user_id, repository_name = self.__get_user_id_repository_name_from_path_info( connection, path_info ) - sql_cmd = "SELECT times_downloaded FROM repository WHERE user_id = %d AND name = '%s'" % ( user_id, repository_name.lower() ) + sql_cmd = "SELECT times_downloaded FROM repository WHERE user_id = %d AND name = '%s'" % \ + ( user_id, repository_name.lower() ) result_set = connection.execute( sql_cmd ) for row in result_set: # Should only be 1 row... times_downloaded = row[ 'times_downloaded' ] times_downloaded += 1 - sql_cmd = "UPDATE repository SET times_downloaded = %d WHERE user_id = %d AND name = '%s'" % ( times_downloaded, user_id, repository_name.lower() ) + sql_cmd = "UPDATE repository SET times_downloaded = %d WHERE user_id = %d AND name = '%s'" % \ + ( times_downloaded, user_id, repository_name.lower() ) connection.execute( sql_cmd ) connection.close() elif cmd in [ 'unbundle', 'pushkey' ]: @@ -132,7 +133,7 @@ if filename and isinstance( filename, str ): if filename == rt_util.REPOSITORY_DEPENDENCY_DEFINITION_FILENAME: # Make sure the any complex repository dependency definitions contain valid <repository> tags. - is_valid, error_msg = commit_util.repository_tags_are_valid( filename, change_list ) + is_valid, error_msg = self.repository_tags_are_valid( filename, change_list ) if not is_valid: log.debug( error_msg ) return self.__display_exception_remotely( start_response, error_msg ) @@ -151,7 +152,7 @@ if filename and isinstance( filename, str ): if filename == rt_util.TOOL_DEPENDENCY_DEFINITION_FILENAME: # Make sure the any complex repository dependency definitions contain valid <repository> tags. - is_valid, error_msg = commit_util.repository_tags_are_valid( filename, change_list ) + is_valid, error_msg = self.repository_tags_are_valid( filename, change_list ) if not is_valid: log.debug( error_msg ) return self.__display_exception_remotely( start_response, error_msg ) @@ -174,7 +175,7 @@ rt_util.TOOL_DEPENDENCY_DEFINITION_FILENAME ]: # We check both files since tool dependency definitions files can contain complex # repository dependency definitions. - is_valid, error_msg = commit_util.repository_tags_are_valid( filename, change_list ) + is_valid, error_msg = self.repository_tags_are_valid( filename, change_list ) if not is_valid: log.debug( error_msg ) return self.__display_exception_remotely( start_response, error_msg ) @@ -186,6 +187,54 @@ return result.wsgi_application( environ, start_response ) return self.app( environ, start_response ) + def __authenticate( self, username, password ): + db_password = None + # Instantiate a database connection + engine = sqlalchemy.create_engine( self.db_url ) + connection = engine.connect() + result_set = connection.execute( "select email, password from galaxy_user where username = '%s'" % username.lower() ) + for row in result_set: + # Should only be 1 row... + db_email = row[ 'email' ] + db_password = row[ 'password' ] + connection.close() + if db_password: + # Check if password matches db_password when hashed. + return new_secure_hash( text_type=password ) == db_password + return False + + def __authenticate_remote_user( self, environ, username, password ): + """ + Look after a remote user and "authenticate" - upstream server should already have achieved + this for us, but we check that the user exists at least. Hg allow_push = must include username + - some versions of mercurial blow up with 500 errors. + """ + db_username = None + ru_email = environ[ 'HTTP_REMOTE_USER' ].lower() + ## Instantiate a database connection... + engine = sqlalchemy.create_engine( self.db_url ) + connection = engine.connect() + result_set = connection.execute( "select email, username, password from galaxy_user where email = '%s'" % ru_email ) + for row in result_set: + # Should only be 1 row... + db_email = row[ 'email' ] + db_password = row[ 'password' ] + db_username = row[ 'username' ] + connection.close() + if db_username: + # We could check the password here except that the function galaxy.web.framework.get_or_create_remote_user() + # does some random generation of a password - so that no-one knows the password and only the hash is stored... + return db_username == username + return False + + def __basic_authentication( self, environ, username, password ): + """The environ parameter is needed in basic authentication. We also check it if use_remote_user is true.""" + if asbool( self.config.get( 'use_remote_user', False ) ): + assert "HTTP_REMOTE_USER" in environ, "use_remote_user is set but no HTTP_REMOTE_USER variable" + return self.__authenticate_remote_user( environ, username, password ) + else: + return self.__authenticate( username, password ) + def __display_exception_remotely( self, start_response, msg ): # Display the exception to the remote user's command line. status = "500 %s" % msg @@ -213,49 +262,37 @@ user_id = row[ 'id' ] return user_id, repository_name - def __basic_authentication( self, environ, username, password ): - """The environ parameter is needed in basic authentication. We also check it if use_remote_user is true.""" - if asbool( self.config.get( 'use_remote_user', False ) ): - assert "HTTP_REMOTE_USER" in environ, "use_remote_user is set but no HTTP_REMOTE_USER variable" - return self.__authenticate_remote_user( environ, username, password ) - else: - return self.__authenticate( username, password ) - - def __authenticate( self, username, password ): - db_password = None - # Instantiate a database connection - engine = sqlalchemy.create_engine( self.db_url ) - connection = engine.connect() - result_set = connection.execute( "select email, password from galaxy_user where username = '%s'" % username.lower() ) - for row in result_set: - # Should only be 1 row... - db_email = row[ 'email' ] - db_password = row[ 'password' ] - connection.close() - if db_password: - # Check if password matches db_password when hashed. - return new_secure_hash( text_type=password ) == db_password - return False - - def __authenticate_remote_user( self, environ, username, password ): + def repository_tag_is_valid( self, filename, line ): """ - Look after a remote user and "authenticate" - upstream server should already have achieved this for us, but we check that the - user exists at least. Hg allow_push = must include username - some versions of mercurial blow up with 500 errors. + Checks changes made to <repository> tags in a dependency definition file being pushed to the + Tool Shed from the command line to ensure that all required attributes exist. """ - db_username = None - ru_email = environ[ 'HTTP_REMOTE_USER' ].lower() - ## Instantiate a database connection... - engine = sqlalchemy.create_engine( self.db_url ) - connection = engine.connect() - result_set = connection.execute( "select email, username, password from galaxy_user where email = '%s'" % ru_email ) - for row in result_set: - # Should only be 1 row... - db_email = row[ 'email' ] - db_password = row[ 'password' ] - db_username = row[ 'username' ] - connection.close() - if db_username: - # We could check the password here except that the function galaxy.web.framework.get_or_create_remote_user() does some random generation of - # a password - so that no-one knows the password and only the hash is stored... - return db_username == username - return False + required_attributes = [ 'toolshed', 'name', 'owner', 'changeset_revision' ] + defined_attributes = line.split() + for required_attribute in required_attributes: + defined = False + for defined_attribute in defined_attributes: + if defined_attribute.startswith( required_attribute ): + defined = True + break + if not defined: + error_msg = 'The %s file contains a <repository> tag that is missing the required attribute %s. ' % \ + ( filename, required_attribute ) + error_msg += 'Automatically populating dependency definition attributes occurs only when using ' + error_msg += 'the Tool Shed upload utility. ' + return False, error_msg + return True, '' + + def repository_tags_are_valid( self, filename, change_list ): + """ + Make sure the any complex repository dependency definitions contain valid <repository> tags when pushing + changes to the tool shed on the command line. + """ + tag = '<repository' + for change_dict in change_list: + lines = get_change_lines_in_file_for_tag( tag, change_dict ) + for line in lines: + is_valid, error_msg = repository_tag_is_valid( filename, line ) + if not is_valid: + return False, error_msg + return True, '' This diff is so big that we needed to truncate the remainder. https://bitbucket.org/galaxy/galaxy-central/commits/4a8a3951b020/ Changeset: 4a8a3951b020 User: dannon Date: 2014-06-25 00:34:32 Summary: Make display chunk size configurable in universe. Defaults to 64k. Affected #: 3 files diff -r 7b4946192456c1619829b6836a95716d8656817d -r 4a8a3951b020ae7de3fc74668ab3399e1a4ba3ef lib/galaxy/config.py --- a/lib/galaxy/config.py +++ b/lib/galaxy/config.py @@ -376,6 +376,8 @@ self.fluent_port = int( kwargs.get( 'fluent_port', 24224 ) ) # visualization plugin framework self.visualization_plugins_directory = kwargs.get( 'visualization_plugins_directory', None ) + # Default chunk size for chunkable datatypes -- 64k + self.display_chunk_size = int( kwargs.get( 'display_chunk_size', 65536) ) @property def sentry_dsn_public( self ): diff -r 7b4946192456c1619829b6836a95716d8656817d -r 4a8a3951b020ae7de3fc74668ab3399e1a4ba3ef lib/galaxy/datatypes/tabular.py --- a/lib/galaxy/datatypes/tabular.py +++ b/lib/galaxy/datatypes/tabular.py @@ -25,7 +25,6 @@ # All tabular data is chunkable. CHUNKABLE = True - CHUNK_SIZE = 10000 """Add metadata elements""" MetadataElement( name="comment_lines", default=0, desc="Number of comment lines", readonly=False, optional=True, no_value=0 ) @@ -262,13 +261,13 @@ def get_chunk(self, trans, dataset, chunk): ck_index = int(chunk) f = open(dataset.file_name) - f.seek(ck_index * self.CHUNK_SIZE) + f.seek(ck_index * trans.app.config.display_chunk_size) # If we aren't at the start of the file, seek to next newline. Do this better eventually. if f.tell() != 0: cursor = f.read(1) while cursor and cursor != '\n': cursor = f.read(1) - ck_data = f.read(self.CHUNK_SIZE) + ck_data = f.read(trans.app.config.display_chunk_size) cursor = f.read(1) while cursor and ck_data[-1] != '\n': ck_data += cursor diff -r 7b4946192456c1619829b6836a95716d8656817d -r 4a8a3951b020ae7de3fc74668ab3399e1a4ba3ef universe_wsgi.ini.sample --- a/universe_wsgi.ini.sample +++ b/universe_wsgi.ini.sample @@ -462,6 +462,10 @@ #enable_cloud_launch = False #cloudlaunch_default_ami = ami-a7dbf6ce +# Incremental Display Options + +#display_chunk_size = 65536 + # -- Advanced proxy features # For help on configuring the Advanced proxy features, see: https://bitbucket.org/galaxy/galaxy-central/commits/dd315abced98/ Changeset: dd315abced98 User: dannon Date: 2014-06-25 00:34:45 Summary: Merge. Affected #: 2 files diff -r 4a8a3951b020ae7de3fc74668ab3399e1a4ba3ef -r dd315abced98d580b24d8718c65243c4185c0c4c static/scripts/mvc/history/history-panel.js --- a/static/scripts/mvc/history/history-panel.js +++ b/static/scripts/mvc/history/history-panel.js @@ -249,7 +249,7 @@ actions.push( { // TODO: Only show quick pair if two things selected. html: _l( 'Build List of Dataset Pairs (Experimental)' ), - func: panel._showPairedCollectionModal.bind( panel ) + func: _.bind( panel._showPairedCollectionModal, panel ) }); return new PopupMenu( $where.find( '.history-dataset-action-popup-btn' ), actions ); }, diff -r 4a8a3951b020ae7de3fc74668ab3399e1a4ba3ef -r dd315abced98d580b24d8718c65243c4185c0c4c test/casperjs/modules/api.js --- a/test/casperjs/modules/api.js +++ b/test/casperjs/modules/api.js @@ -555,7 +555,7 @@ }; /** amount of time allowed to upload a file (before erroring) */ -ToolsAPI.prototype.DEFAULT_UPLOAD_TIMEOUT = 12000; +ToolsAPI.prototype.DEFAULT_UPLOAD_TIMEOUT = 30 * 1000; /** add two casperjs steps - upload a file, wait for the job to complete, and run 'then' when they are */ ToolsAPI.prototype.thenUpload = function thenUpload( historyId, options, then ){ @@ -569,6 +569,9 @@ this.debug( 'uploadedId: ' + uploadedId ); }); + spaceghost.debug( '---------------------------------------------------------- timeout: ' + ( options.timeout || spaceghost.api.tools.DEFAULT_UPLOAD_TIMEOUT ) ); + spaceghost.debug( 'timeout: ' + options.timeout ); + spaceghost.debug( 'timeout: ' + spaceghost.api.tools.DEFAULT_UPLOAD_TIMEOUT ); spaceghost.then( function(){ this.waitFor( function testHdaState(){ https://bitbucket.org/galaxy/galaxy-central/commits/06b64928f4a4/ Changeset: 06b64928f4a4 User: dannon Date: 2014-06-25 00:43:13 Summary: Clean up config.py (pep8, imports, misc). Affected #: 1 file diff -r dd315abced98d580b24d8718c65243c4185c0c4c -r 06b64928f4a4da6384d0717fc9bb650aa5e307eb lib/galaxy/config.py --- a/lib/galaxy/config.py +++ b/lib/galaxy/config.py @@ -15,13 +15,12 @@ from galaxy.web.formatting import expand_pretty_datetime_format from galaxy.util import string_as_bool from galaxy.util import listify -from galaxy.util import parse_xml from galaxy.util.dbkeys import GenomeBuilds from galaxy import eggs -import pkg_resources log = logging.getLogger( __name__ ) + def resolve_path( path, root ): """If 'path' is relative make absolute by prepending 'root'""" if not( os.path.isabs( path ) ): @@ -40,9 +39,9 @@ self.config_dict = kwargs self.root = kwargs.get( 'root_dir', '.' ) # Collect the umask and primary gid from the environment - self.umask = os.umask( 077 ) # get the current umask - os.umask( self.umask ) # can't get w/o set, so set it back - self.gid = os.getgid() # if running under newgrp(1) we'll need to fix the group of data created on the cluster + self.umask = os.umask( 077 ) # get the current umask + os.umask( self.umask ) # can't get w/o set, so set it back + self.gid = os.getgid() # if running under newgrp(1) we'll need to fix the group of data created on the cluster # Database related configuration self.database = resolve_path( kwargs.get( "database_file", "database/universe.sqlite" ), self.root ) @@ -75,7 +74,7 @@ self.enable_unique_workflow_defaults = string_as_bool( kwargs.get( 'enable_unique_workflow_defaults', False ) ) self.tool_path = resolve_path( kwargs.get( "tool_path", "tools" ), self.root ) self.tool_data_path = resolve_path( kwargs.get( "tool_data_path", "tool-data" ), os.getcwd() ) - self.len_file_path = resolve_path( kwargs.get( "len_file_path", os.path.join( self.tool_data_path, 'shared','ucsc','chrom') ), self.root ) + self.len_file_path = resolve_path( kwargs.get( "len_file_path", os.path.join( self.tool_data_path, 'shared', 'ucsc', 'chrom') ), self.root ) self.test_conf = resolve_path( kwargs.get( "test_conf", "" ), self.root ) # The value of migrated_tools_config is the file reserved for containing only those tools that have been eliminated from the distribution # and moved to the tool shed. @@ -169,7 +168,7 @@ self.admin_users = kwargs.get( "admin_users", "" ) self.admin_users_list = [u.strip() for u in self.admin_users.split(',') if u] self.reset_password_length = int( kwargs.get('reset_password_length', '15') ) - self.mailing_join_addr = kwargs.get('mailing_join_addr',"galaxy-announce-join(a)bx.psu.edu") + self.mailing_join_addr = kwargs.get('mailing_join_addr', 'galaxy-announce-join(a)bx.psu.edu') self.error_email_to = kwargs.get( 'error_email_to', None ) self.activation_email = kwargs.get( 'activation_email', None ) self.user_activation_on = string_as_bool( kwargs.get( 'user_activation_on', False ) ) @@ -271,9 +270,9 @@ self.object_store_cache_path = resolve_path( kwargs.get( "object_store_cache_path", "database/object_store_cache" ), self.root ) # Handle AWS-specific config options for backward compatibility if kwargs.get( 'aws_access_key', None) is not None: - self.os_access_key= kwargs.get( 'aws_access_key', None ) - self.os_secret_key= kwargs.get( 'aws_secret_key', None ) - self.os_bucket_name= kwargs.get( 's3_bucket', None ) + self.os_access_key = kwargs.get( 'aws_access_key', None ) + self.os_secret_key = kwargs.get( 'aws_secret_key', None ) + self.os_bucket_name = kwargs.get( 's3_bucket', None ) self.os_use_reduced_redundancy = kwargs.get( 'use_reduced_redundancy', False ) else: self.os_access_key = kwargs.get( 'os_access_key', None ) @@ -454,19 +453,13 @@ except Exception, e: raise ConfigurationError( "Unable to create missing directory: %s\n%s" % ( path, e ) ) # Create the directories that it makes sense to create - for path in self.file_path, \ - self.new_file_path, \ - self.job_working_directory, \ - self.cluster_files_directory, \ - self.template_cache, \ - self.ftp_upload_dir, \ - self.library_import_dir, \ - self.user_library_import_dir, \ - self.nginx_upload_store, \ - './static/genetrack/plots', \ - self.whoosh_index_dir, \ - self.object_store_cache_path, \ - os.path.join( self.tool_data_path, 'shared', 'jars' ): + for path in (self.file_path, self.new_file_path, + self.job_working_directory, self.cluster_files_directory, + self.template_cache, self.ftp_upload_dir, + self.library_import_dir, self.user_library_import_dir, + self.nginx_upload_store, './static/genetrack/plots', + self.whoosh_index_dir, self.object_store_cache_path, + os.path.join( self.tool_data_path, 'shared', 'jars' )): self._ensure_directory( path ) # Check that required files exist tool_configs = self.tool_configs @@ -482,7 +475,7 @@ if key in self.deprecated_options: log.warning( "Config option '%s' is deprecated and will be removed in a future release. Please consult the latest version of the sample configuration file." % key ) - def is_admin_user( self,user ): + def is_admin_user( self, user ): """ Determine if the provided user is listed in `admin_users`. @@ -497,12 +490,13 @@ """ return resolve_path( path, self.root ) + def get_database_engine_options( kwargs, model_prefix='' ): """ Allow options for the SQLAlchemy database engine to be passed by using the prefix "database_engine_option". """ - conversions = { + conversions = { 'convert_unicode': string_as_bool, 'pool_timeout': int, 'echo': string_as_bool, @@ -524,6 +518,7 @@ rval[ key ] = value return rval + def configure_logging( config ): """ Allow some basic logging configuration to be read from ini file. @@ -558,7 +553,7 @@ root.addHandler( handler ) # If sentry is configured, also log to it if config.sentry_dsn: - pkg_resources.require( "raven" ) + eggs.require( "raven" ) from raven.handlers.logging import SentryHandler sentry_handler = SentryHandler( config.sentry_dsn ) sentry_handler.setLevel( logging.WARN ) https://bitbucket.org/galaxy/galaxy-central/commits/13a9176d15e0/ Changeset: 13a9176d15e0 User: dannon Date: 2014-07-08 19:49:40 Summary: Merge from stable. Affected #: 1 file diff -r 5874f6bc02f92b0d09184e04e8fdc89bd2fda638 -r 13a9176d15e0fe13af359b5494ec2d8830d22e04 lib/galaxy/jobs/metrics/formatting.py --- a/lib/galaxy/jobs/metrics/formatting.py +++ b/lib/galaxy/jobs/metrics/formatting.py @@ -15,4 +15,4 @@ elif value < 3600: return "%s minutes" % ( value / 60 ) else: - return "%s days and %s minutes" % ( value / 3600, ( value % 3600 ) / 60 ) + return "%s hours and %s minutes" % ( value / 3600, ( value % 3600 ) / 60 ) https://bitbucket.org/galaxy/galaxy-central/commits/158302d14057/ Changeset: 158302d14057 User: dannon Date: 2014-07-08 19:50:05 Summary: Merge. Affected #: 4 files diff -r 13a9176d15e0fe13af359b5494ec2d8830d22e04 -r 158302d1405755306533ad44b2284acad31e421a lib/galaxy/config.py --- a/lib/galaxy/config.py +++ b/lib/galaxy/config.py @@ -15,13 +15,12 @@ from galaxy.web.formatting import expand_pretty_datetime_format from galaxy.util import string_as_bool from galaxy.util import listify -from galaxy.util import parse_xml from galaxy.util.dbkeys import GenomeBuilds from galaxy import eggs -import pkg_resources log = logging.getLogger( __name__ ) + def resolve_path( path, root ): """If 'path' is relative make absolute by prepending 'root'""" if not( os.path.isabs( path ) ): @@ -40,9 +39,9 @@ self.config_dict = kwargs self.root = kwargs.get( 'root_dir', '.' ) # Collect the umask and primary gid from the environment - self.umask = os.umask( 077 ) # get the current umask - os.umask( self.umask ) # can't get w/o set, so set it back - self.gid = os.getgid() # if running under newgrp(1) we'll need to fix the group of data created on the cluster + self.umask = os.umask( 077 ) # get the current umask + os.umask( self.umask ) # can't get w/o set, so set it back + self.gid = os.getgid() # if running under newgrp(1) we'll need to fix the group of data created on the cluster # Database related configuration self.database = resolve_path( kwargs.get( "database_file", "database/universe.sqlite" ), self.root ) @@ -75,7 +74,7 @@ self.enable_unique_workflow_defaults = string_as_bool( kwargs.get( 'enable_unique_workflow_defaults', False ) ) self.tool_path = resolve_path( kwargs.get( "tool_path", "tools" ), self.root ) self.tool_data_path = resolve_path( kwargs.get( "tool_data_path", "tool-data" ), os.getcwd() ) - self.len_file_path = resolve_path( kwargs.get( "len_file_path", os.path.join( self.tool_data_path, 'shared','ucsc','chrom') ), self.root ) + self.len_file_path = resolve_path( kwargs.get( "len_file_path", os.path.join( self.tool_data_path, 'shared', 'ucsc', 'chrom') ), self.root ) self.test_conf = resolve_path( kwargs.get( "test_conf", "" ), self.root ) # The value of migrated_tools_config is the file reserved for containing only those tools that have been eliminated from the distribution # and moved to the tool shed. @@ -169,7 +168,7 @@ self.admin_users = kwargs.get( "admin_users", "" ) self.admin_users_list = [u.strip() for u in self.admin_users.split(',') if u] self.reset_password_length = int( kwargs.get('reset_password_length', '15') ) - self.mailing_join_addr = kwargs.get('mailing_join_addr',"galaxy-announce-join(a)bx.psu.edu") + self.mailing_join_addr = kwargs.get('mailing_join_addr', 'galaxy-announce-join(a)bx.psu.edu') self.error_email_to = kwargs.get( 'error_email_to', None ) self.activation_email = kwargs.get( 'activation_email', None ) self.user_activation_on = string_as_bool( kwargs.get( 'user_activation_on', False ) ) @@ -271,9 +270,9 @@ self.object_store_cache_path = resolve_path( kwargs.get( "object_store_cache_path", "database/object_store_cache" ), self.root ) # Handle AWS-specific config options for backward compatibility if kwargs.get( 'aws_access_key', None) is not None: - self.os_access_key= kwargs.get( 'aws_access_key', None ) - self.os_secret_key= kwargs.get( 'aws_secret_key', None ) - self.os_bucket_name= kwargs.get( 's3_bucket', None ) + self.os_access_key = kwargs.get( 'aws_access_key', None ) + self.os_secret_key = kwargs.get( 'aws_secret_key', None ) + self.os_bucket_name = kwargs.get( 's3_bucket', None ) self.os_use_reduced_redundancy = kwargs.get( 'use_reduced_redundancy', False ) else: self.os_access_key = kwargs.get( 'os_access_key', None ) @@ -376,6 +375,8 @@ self.fluent_port = int( kwargs.get( 'fluent_port', 24224 ) ) # visualization plugin framework self.visualization_plugins_directory = kwargs.get( 'visualization_plugins_directory', None ) + # Default chunk size for chunkable datatypes -- 64k + self.display_chunk_size = int( kwargs.get( 'display_chunk_size', 65536) ) @property def sentry_dsn_public( self ): @@ -452,19 +453,13 @@ except Exception, e: raise ConfigurationError( "Unable to create missing directory: %s\n%s" % ( path, e ) ) # Create the directories that it makes sense to create - for path in self.file_path, \ - self.new_file_path, \ - self.job_working_directory, \ - self.cluster_files_directory, \ - self.template_cache, \ - self.ftp_upload_dir, \ - self.library_import_dir, \ - self.user_library_import_dir, \ - self.nginx_upload_store, \ - './static/genetrack/plots', \ - self.whoosh_index_dir, \ - self.object_store_cache_path, \ - os.path.join( self.tool_data_path, 'shared', 'jars' ): + for path in (self.file_path, self.new_file_path, + self.job_working_directory, self.cluster_files_directory, + self.template_cache, self.ftp_upload_dir, + self.library_import_dir, self.user_library_import_dir, + self.nginx_upload_store, './static/genetrack/plots', + self.whoosh_index_dir, self.object_store_cache_path, + os.path.join( self.tool_data_path, 'shared', 'jars' )): self._ensure_directory( path ) # Check that required files exist tool_configs = self.tool_configs @@ -480,7 +475,7 @@ if key in self.deprecated_options: log.warning( "Config option '%s' is deprecated and will be removed in a future release. Please consult the latest version of the sample configuration file." % key ) - def is_admin_user( self,user ): + def is_admin_user( self, user ): """ Determine if the provided user is listed in `admin_users`. @@ -495,12 +490,13 @@ """ return resolve_path( path, self.root ) + def get_database_engine_options( kwargs, model_prefix='' ): """ Allow options for the SQLAlchemy database engine to be passed by using the prefix "database_engine_option". """ - conversions = { + conversions = { 'convert_unicode': string_as_bool, 'pool_timeout': int, 'echo': string_as_bool, @@ -522,6 +518,7 @@ rval[ key ] = value return rval + def configure_logging( config ): """ Allow some basic logging configuration to be read from ini file. @@ -556,7 +553,7 @@ root.addHandler( handler ) # If sentry is configured, also log to it if config.sentry_dsn: - pkg_resources.require( "raven" ) + eggs.require( "raven" ) from raven.handlers.logging import SentryHandler sentry_handler = SentryHandler( config.sentry_dsn ) sentry_handler.setLevel( logging.WARN ) diff -r 13a9176d15e0fe13af359b5494ec2d8830d22e04 -r 158302d1405755306533ad44b2284acad31e421a lib/galaxy/datatypes/tabular.py --- a/lib/galaxy/datatypes/tabular.py +++ b/lib/galaxy/datatypes/tabular.py @@ -25,7 +25,6 @@ # All tabular data is chunkable. CHUNKABLE = True - CHUNK_SIZE = 10000 """Add metadata elements""" MetadataElement( name="comment_lines", default=0, desc="Number of comment lines", readonly=False, optional=True, no_value=0 ) @@ -262,13 +261,13 @@ def get_chunk(self, trans, dataset, chunk): ck_index = int(chunk) f = open(dataset.file_name) - f.seek(ck_index * self.CHUNK_SIZE) + f.seek(ck_index * trans.app.config.display_chunk_size) # If we aren't at the start of the file, seek to next newline. Do this better eventually. if f.tell() != 0: cursor = f.read(1) while cursor and cursor != '\n': cursor = f.read(1) - ck_data = f.read(self.CHUNK_SIZE) + ck_data = f.read(trans.app.config.display_chunk_size) cursor = f.read(1) while cursor and ck_data[-1] != '\n': ck_data += cursor diff -r 13a9176d15e0fe13af359b5494ec2d8830d22e04 -r 158302d1405755306533ad44b2284acad31e421a lib/galaxy/webapps/galaxy/controllers/user.py --- a/lib/galaxy/webapps/galaxy/controllers/user.py +++ b/lib/galaxy/webapps/galaxy/controllers/user.py @@ -1092,7 +1092,7 @@ def reset_password( self, trans, email=None, **kwd ): if trans.app.config.smtp_server is None: return trans.show_error_message( "Mail is not configured for this Galaxy instance. Please contact your local Galaxy administrator." ) - message = util.restore_text( kwd.get( 'message', '' ) ) + message = util.sanitize_text(util.restore_text( kwd.get( 'message', '' ) )) status = 'done' if kwd.get( 'reset_password_button', False ): reset_user = trans.sa_session.query( trans.app.model.User ).filter( trans.app.model.User.table.c.email == email ).first() diff -r 13a9176d15e0fe13af359b5494ec2d8830d22e04 -r 158302d1405755306533ad44b2284acad31e421a universe_wsgi.ini.sample --- a/universe_wsgi.ini.sample +++ b/universe_wsgi.ini.sample @@ -462,6 +462,10 @@ #enable_cloud_launch = False #cloudlaunch_default_ami = ami-a7dbf6ce +# Incremental Display Options + +#display_chunk_size = 65536 + # -- Advanced proxy features # For help on configuring the Advanced proxy features, see: Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: dannon: Merged in nsoranzo/galaxy-central/stable (pull request #436)
by commits-noreply＠bitbucket.org 08 Jul '14

08 Jul '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/b1441a96b840/ Changeset: b1441a96b840 Branch: stable User: dannon Date: 2014-07-08 19:47:15 Summary: Merged in nsoranzo/galaxy-central/stable (pull request #436) Fix seconds_to_str(). Affected #: 1 file diff -r d9ab8058b2748de6b4b1a17646c2d7456d709877 -r b1441a96b8401fd14ab3364c5d6dfa0733df263e lib/galaxy/jobs/metrics/formatting.py --- a/lib/galaxy/jobs/metrics/formatting.py +++ b/lib/galaxy/jobs/metrics/formatting.py @@ -15,4 +15,4 @@ elif value < 3600: return "%s minutes" % ( value / 60 ) else: - return "%s days and %s minutes" % ( value / 3600, ( value % 3600 ) / 60 ) + return "%s hours and %s minutes" % ( value / 3600, ( value % 3600 ) / 60 ) Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: 2 new changesets
by commits-noreply＠bitbucket.org 08 Jul '14

08 Jul '14

2 new commits in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/04a53a8605f1/ Changeset: 04a53a8605f1 Branch: stable User: nsoranzo Date: 2014-07-08 17:58:10 Summary: Fix seconds_to_str(). Affected #: 1 file diff -r c4519bae84d32d638d8fe168c6fbcab40f4e8449 -r 04a53a8605f18d28e4e084bc8352885611a87876 lib/galaxy/jobs/metrics/formatting.py --- a/lib/galaxy/jobs/metrics/formatting.py +++ b/lib/galaxy/jobs/metrics/formatting.py @@ -15,4 +15,4 @@ elif value < 3600: return "%s minutes" % ( value / 60 ) else: - return "%s days and %s minutes" % ( value / 3600, ( value % 3600 ) / 60 ) + return "%s hours and %s minutes" % ( value / 3600, ( value % 3600 ) / 60 ) https://bitbucket.org/galaxy/galaxy-central/commits/b1441a96b840/ Changeset: b1441a96b840 Branch: stable User: dannon Date: 2014-07-08 19:47:15 Summary: Merged in nsoranzo/galaxy-central/stable (pull request #436) Fix seconds_to_str(). Affected #: 1 file diff -r d9ab8058b2748de6b4b1a17646c2d7456d709877 -r b1441a96b8401fd14ab3364c5d6dfa0733df263e lib/galaxy/jobs/metrics/formatting.py --- a/lib/galaxy/jobs/metrics/formatting.py +++ b/lib/galaxy/jobs/metrics/formatting.py @@ -15,4 +15,4 @@ elif value < 3600: return "%s minutes" % ( value / 60 ) else: - return "%s days and %s minutes" % ( value / 3600, ( value % 3600 ) / 60 ) + return "%s hours and %s minutes" % ( value / 3600, ( value % 3600 ) / 60 ) Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: 6 new changesets
by commits-noreply＠bitbucket.org 08 Jul '14

08 Jul '14

6 new commits in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/21b53ebad211/ Changeset: 21b53ebad211 Branch: stable User: jmchilton Date: 2014-06-29 20:24:32 Summary: Spelling fix in method name. Affected #: 2 files diff -r 65e5e5b72893445eef5c4d2f43350df72e89b436 -r 21b53ebad2119dd9968c72f67b53d069ebc17902 lib/galaxy/tools/__init__.py --- a/lib/galaxy/tools/__init__.py +++ b/lib/galaxy/tools/__init__.py @@ -2874,7 +2874,7 @@ Find any additional datasets generated by a tool and attach (for cases where number of outputs is not known in advance). """ - return output_collect.collect_primary_datatasets( self, output, job_working_directory ) + return output_collect.collect_primary_datasets( self, output, job_working_directory ) def to_dict( self, trans, link_details=False, io_details=False ): """ Returns dict of tool. """ diff -r 65e5e5b72893445eef5c4d2f43350df72e89b436 -r 21b53ebad2119dd9968c72f67b53d069ebc17902 lib/galaxy/tools/parameters/output_collect.py --- a/lib/galaxy/tools/parameters/output_collect.py +++ b/lib/galaxy/tools/parameters/output_collect.py @@ -14,7 +14,7 @@ DEFAULT_EXTRA_FILENAME_PATTERN = r"primary_DATASET_ID_(?P<designation>[^_]+)_(?P<visible>[^_]+)_(?P<ext>[^_]+)(_(?P<dbkey>[^_]+))?" -def collect_primary_datatasets( tool, output, job_working_directory ): +def collect_primary_datasets( tool, output, job_working_directory ): app = tool.app sa_session = tool.sa_session new_primary_datasets = {} https://bitbucket.org/galaxy/galaxy-central/commits/c14bdcb9ae96/ Changeset: c14bdcb9ae96 Branch: stable User: jmchilton Date: 2014-06-29 20:24:32 Summary: Allow discovered datasets to use input data format in 'ext' definition. Affected #: 4 files diff -r 21b53ebad2119dd9968c72f67b53d069ebc17902 -r c14bdcb9ae9643525c0797090f293030ea899948 lib/galaxy/jobs/__init__.py --- a/lib/galaxy/jobs/__init__.py +++ b/lib/galaxy/jobs/__init__.py @@ -1065,6 +1065,13 @@ out_data = dict( [ ( da.name, da.dataset ) for da in job.output_datasets ] ) inp_data.update( [ ( da.name, da.dataset ) for da in job.input_library_datasets ] ) out_data.update( [ ( da.name, da.dataset ) for da in job.output_library_datasets ] ) + input_ext = 'data' + for _, data in inp_data.items(): + # For loop odd, but sort simulating behavior in galaxy.tools.actions + if not data: + continue + input_ext = data.ext + param_dict = dict( [ ( p.name, p.value ) for p in job.parameters ] ) # why not re-use self.param_dict here? ##dunno...probably should, this causes tools.parameters.basic.UnvalidatedValue to be used in following methods instead of validated and transformed values during i.e. running workflows param_dict = self.tool.params_from_strings( param_dict, self.app ) # Check for and move associated_files @@ -1075,7 +1082,7 @@ # Create generated output children and primary datasets and add to param_dict collected_datasets = { 'children': self.tool.collect_child_datasets(out_data, self.working_directory), - 'primary': self.tool.collect_primary_datasets(out_data, self.working_directory) + 'primary': self.tool.collect_primary_datasets(out_data, self.working_directory, input_ext) } param_dict.update({'__collected_datasets__': collected_datasets}) # Certain tools require tasks to be completed after job execution diff -r 21b53ebad2119dd9968c72f67b53d069ebc17902 -r c14bdcb9ae9643525c0797090f293030ea899948 lib/galaxy/tools/__init__.py --- a/lib/galaxy/tools/__init__.py +++ b/lib/galaxy/tools/__init__.py @@ -2869,12 +2869,12 @@ self.sa_session.flush() return children - def collect_primary_datasets( self, output, job_working_directory ): + def collect_primary_datasets( self, output, job_working_directory, input_ext ): """ Find any additional datasets generated by a tool and attach (for cases where number of outputs is not known in advance). """ - return output_collect.collect_primary_datasets( self, output, job_working_directory ) + return output_collect.collect_primary_datasets( self, output, job_working_directory, input_ext ) def to_dict( self, trans, link_details=False, io_details=False ): """ Returns dict of tool. """ diff -r 21b53ebad2119dd9968c72f67b53d069ebc17902 -r c14bdcb9ae9643525c0797090f293030ea899948 lib/galaxy/tools/parameters/output_collect.py --- a/lib/galaxy/tools/parameters/output_collect.py +++ b/lib/galaxy/tools/parameters/output_collect.py @@ -14,7 +14,7 @@ DEFAULT_EXTRA_FILENAME_PATTERN = r"primary_DATASET_ID_(?P<designation>[^_]+)_(?P<visible>[^_]+)_(?P<ext>[^_]+)(_(?P<dbkey>[^_]+))?" -def collect_primary_datasets( tool, output, job_working_directory ): +def collect_primary_datasets( tool, output, job_working_directory, input_ext ): app = tool.app sa_session = tool.sa_session new_primary_datasets = {} @@ -66,6 +66,8 @@ designation = fields_match.designation visible = fields_match.visible ext = fields_match.ext + if ext == "input": + ext = input_ext dbkey = fields_match.dbkey # Create new primary dataset primary_data = app.model.HistoryDatasetAssociation( extension=ext, diff -r 21b53ebad2119dd9968c72f67b53d069ebc17902 -r c14bdcb9ae9643525c0797090f293030ea899948 test/functional/tools/multi_output_configured.xml --- a/test/functional/tools/multi_output_configured.xml +++ b/test/functional/tools/multi_output_configured.xml @@ -8,19 +8,24 @@ echo "1" > subdir2/CUSTOM_1.txt; echo "2" > subdir2/CUSTOM_2.tabular; echo "3" > subdir2/CUSTOM_3.txt; + mkdir subdir3; + echo "Foo" > subdir3/Foo; </command><inputs> - <param name="input" type="integer" value="7" /> + <param name="num_param" type="integer" value="7" /> + <param name="input" type="data" /></inputs><outputs><data format="txt" name="report"><discover_datasets pattern="__designation_and_ext__" directory="subdir1" /><discover_datasets pattern="CUSTOM_(?P<designation>.+)\.(?P<ext>.+)" directory="subdir2" /> + <discover_datasets pattern="__designation__" directory="subdir3" ext="input" /></data></outputs><tests><test> - <param name="input" value="7" /> + <param name="num_param" value="7" /> + <param name="input" ftype="txt" value="simple_line.txt"/><output name="report"><assert_contents><has_line line="Hello" /> @@ -37,6 +42,9 @@ <discovered_dataset designation="2" ftype="tabular"><assert_contents><has_line line="2" /></assert_contents></discovered_dataset> + <discovered_dataset designation="Foo" ftype="txt"> + <assert_contents><has_line line="Foo" /></assert_contents> + </discovered_dataset></output></test></tests> https://bitbucket.org/galaxy/galaxy-central/commits/b6fae2cd673c/ Changeset: b6fae2cd673c User: davebgx Date: 2014-07-08 19:17:22 Summary: Remove deprecated "data admin" code, since the functionality was implemented better with Dan's data manager framework. Affected #: 14 files diff -r 4eeb41f2fa7569e1d8652d9a0be3a1f1d5c6c2c0 -r b6fae2cd673c6a286138b9e4628f7c4ef0abb099 lib/galaxy/app.py --- a/lib/galaxy/app.py +++ b/lib/galaxy/app.py @@ -12,7 +12,6 @@ from galaxy.visualization.data_providers.registry import DataProviderRegistry from galaxy.visualization.registry import VisualizationsRegistry from galaxy.tools.imp_exp import load_history_imp_exp_tools -from galaxy.tools.genome_index import load_genome_index_tools from galaxy.sample_tracking import external_service_types from galaxy.openid.providers import OpenIDProviders from galaxy.tools.data_manager.manager import DataManagers @@ -93,8 +92,6 @@ self.datatypes_registry.load_external_metadata_tool( self.toolbox ) # Load history import/export tools. load_history_imp_exp_tools( self.toolbox ) - # Load genome indexer tool. - load_genome_index_tools( self.toolbox ) # visualizations registry: associates resources with visualizations, controls how to render self.visualizations_registry = None if self.config.visualization_plugins_directory: diff -r 4eeb41f2fa7569e1d8652d9a0be3a1f1d5c6c2c0 -r b6fae2cd673c6a286138b9e4628f7c4ef0abb099 lib/galaxy/jobs/__init__.py --- a/lib/galaxy/jobs/__init__.py +++ b/lib/galaxy/jobs/__init__.py @@ -1192,9 +1192,6 @@ param_dict = self.tool.params_from_strings( param_dict, self.app ) # Check for and move associated_files self.tool.collect_associated_files(out_data, self.working_directory) - gitd = self.sa_session.query( model.GenomeIndexToolData ).filter_by( job=job ).first() - if gitd: - self.tool.collect_associated_files({'': gitd}, self.working_directory) # Create generated output children and primary datasets and add to param_dict collected_datasets = { 'children': self.tool.collect_child_datasets(out_data, self.working_directory), @@ -1248,7 +1245,6 @@ self.external_output_metadata.cleanup_external_metadata( self.sa_session ) galaxy.tools.imp_exp.JobExportHistoryArchiveWrapper( self.job_id ).cleanup_after_job( self.sa_session ) galaxy.tools.imp_exp.JobImportHistoryArchiveWrapper( self.app, self.job_id ).cleanup_after_job() - galaxy.tools.genome_index.GenomeIndexToolWrapper( self.job_id ).postprocessing( self.sa_session, self.app ) if delete_files: self.app.object_store.delete(self.get_job(), base_dir='job_work', entire_dir=True, dir_only=True, extra_dir=str(self.job_id)) except: @@ -1351,10 +1347,8 @@ dataset_path_rewriter = self.dataset_path_rewriter job = self.get_job() - # Job output datasets are combination of history, library, jeha and gitd datasets. + # Job output datasets are combination of history, library, and jeha datasets. special = self.sa_session.query( model.JobExportHistoryArchive ).filter_by( job=job ).first() - if not special: - special = self.sa_session.query( model.GenomeIndexToolData ).filter_by( job=job ).first() false_path = None results = [] diff -r 4eeb41f2fa7569e1d8652d9a0be3a1f1d5c6c2c0 -r b6fae2cd673c6a286138b9e4628f7c4ef0abb099 lib/galaxy/jobs/deferred/genome_index.py --- a/lib/galaxy/jobs/deferred/genome_index.py +++ /dev/null @@ -1,43 +0,0 @@ -""" -Module for managing genome transfer jobs. -""" -from __future__ import with_statement - -import logging, shutil, gzip, bz2, zipfile, tempfile, tarfile, sys, os - -from galaxy import eggs -from sqlalchemy import and_ -from data_transfer import * - -log = logging.getLogger( __name__ ) - -__all__ = [ 'GenomeIndexPlugin' ] - -class GenomeIndexPlugin( DataTransfer ): - - def __init__( self, app ): - super( GenomeIndexPlugin, self ).__init__( app ) - self.app = app - self.tool = app.toolbox.tools_by_id['__GENOME_INDEX__'] - self.sa_session = app.model.context.current - - def create_job( self, trans, path, indexes, dbkey, intname ): - params = dict( user=trans.user.id, path=path, indexes=indexes, dbkey=dbkey, intname=intname ) - deferred = trans.app.model.DeferredJob( state = self.app.model.DeferredJob.states.NEW, plugin = 'GenomeIndexPlugin', params = params ) - self.sa_session.add( deferred ) - self.sa_session.flush() - log.debug( 'Job created, id %d' % deferred.id ) - return deferred.id - - def check_job( self, job ): - log.debug( 'Job check' ) - return 'ready' - - def run_job( self, job ): - incoming = dict( path=os.path.abspath( job.params[ 'path' ] ), indexer=job.params[ 'indexes' ][0], user=job.params[ 'user' ] ) - indexjob = self.tool.execute( self, set_output_hid=False, history=None, incoming=incoming, transfer=None, deferred=job ) - job.params[ 'indexjob' ] = indexjob[0].id - job.state = self.app.model.DeferredJob.states.RUNNING - self.sa_session.add( job ) - self.sa_session.flush() - return self.app.model.DeferredJob.states.RUNNING diff -r 4eeb41f2fa7569e1d8652d9a0be3a1f1d5c6c2c0 -r b6fae2cd673c6a286138b9e4628f7c4ef0abb099 lib/galaxy/jobs/deferred/genome_transfer.py --- a/lib/galaxy/jobs/deferred/genome_transfer.py +++ /dev/null @@ -1,250 +0,0 @@ -""" -Module for managing genome transfer jobs. -""" -from __future__ import with_statement - -import logging, shutil, gzip, bz2, zipfile, tempfile, tarfile, sys - -from galaxy import eggs -from sqlalchemy import and_ - -from galaxy.util.odict import odict -from galaxy.workflow.modules import module_factory -from galaxy.jobs.actions.post import ActionBox - -from galaxy.tools.parameters import visit_input_values -from galaxy.tools.parameters.basic import DataToolParameter -from galaxy.tools.data import ToolDataTableManager - -from galaxy.datatypes.checkers import * -from galaxy.datatypes.sequence import Fasta -from data_transfer import * - -log = logging.getLogger( __name__ ) - -__all__ = [ 'GenomeTransferPlugin' ] - -class GenomeTransferPlugin( DataTransfer ): - - locations = {} - - def __init__( self, app ): - super( GenomeTransferPlugin, self ).__init__( app ) - self.app = app - self.tool = app.toolbox.tools_by_id['__GENOME_INDEX__'] - self.sa_session = app.model.context.current - tdtman = ToolDataTableManager( app.config.tool_data_path ) - xmltree = tdtman.load_from_config_file( app.config.tool_data_table_config_path, app.config.tool_data_path ) - for node in xmltree: - table = node.get('name') - location = node.findall('file')[0].get('path') - self.locations[table] = location - - def create_job( self, trans, url, dbkey, intname, indexes ): - job = trans.app.transfer_manager.new( protocol='http', url=url ) - params = dict( user=trans.user.id, transfer_job_id=job.id, protocol='http', type='init_transfer', url=url, dbkey=dbkey, indexes=indexes, intname=intname, liftover=None ) - deferred = trans.app.model.DeferredJob( state = self.app.model.DeferredJob.states.NEW, plugin = 'GenomeTransferPlugin', params = params ) - self.sa_session.add( deferred ) - self.sa_session.flush() - return deferred.id - - def check_job( self, job ): - if job.params['type'] == 'init_transfer': - if not hasattr(job, 'transfer_job'): - job.transfer_job = self.sa_session.query( self.app.model.TransferJob ).get( int( job.params[ 'transfer_job_id' ] ) ) - else: - self.sa_session.refresh( job.transfer_job ) - if job.transfer_job.state == 'done': - transfer = job.transfer_job - transfer.state = 'downloaded' - job.params['type'] = 'extract_transfer' - self.sa_session.add( job ) - self.sa_session.add( transfer ) - self.sa_session.flush() - return self.job_states.READY - elif job.transfer_job.state == 'running': - return self.job_states.WAIT - elif job.transfer_job.state == 'new': - assert job.params[ 'protocol' ] in [ 'http', 'ftp', 'https' ], 'Unknown protocol %s' % job.params[ 'protocol' ] - self.app.transfer_manager.run( job.transfer_job ) - self.sa_session.add( job.transfer_job ) - self.sa_session.flush() - return self.job_states.WAIT - else: - log.error( "An error occurred while downloading from %s" % job.params[ 'url' ] ) - return self.job_states.INVALID - elif job.params[ 'type' ] == 'extract_transfer': - return self.job_states.READY - - def get_job_status( self, jobid ): - job = self.sa_session.query( self.app.model.DeferredJob ).get( int( jobid ) ) - if 'transfer_job_id' in job.params: - if not hasattr( job, 'transfer_job' ): - job.transfer_job = self.sa_session.query( self.app.model.TransferJob ).get( int( job.params[ 'transfer_job_id' ] ) ) - else: - self.sa_session.refresh( job.transfer_job ) - return job - - def run_job( self, job ): - params = job.params - dbkey = params[ 'dbkey' ] - if not hasattr( job, 'transfer_job' ): - job.transfer_job = self.sa_session.query( self.app.model.TransferJob ).get( int( job.params[ 'transfer_job_id' ] ) ) - else: - self.sa_session.refresh( job.transfer_job ) - transfer = job.transfer_job - if params[ 'type' ] == 'extract_transfer': - CHUNK_SIZE = 2**20 - destpath = os.path.join( self.app.config.get( 'genome_data_path', 'tool-data/genome' ), job.params[ 'dbkey' ], 'seq' ) - destfile = '%s.fa' % job.params[ 'dbkey' ] - destfilepath = os.path.join( destpath, destfile ) - tmpprefix = '%s_%s_download_unzip_' % ( job.params['dbkey'], job.params[ 'transfer_job_id' ] ) - tmppath = os.path.dirname( os.path.abspath( transfer.path ) ) - if not os.path.exists( destpath ): - os.makedirs( destpath ) - protocol = job.params[ 'protocol' ] - data_type = self._check_compress( transfer.path ) - if data_type is None: - sniffer = Fasta() - if sniffer.sniff( transfer.path ): - data_type = 'fasta' - fd, uncompressed = tempfile.mkstemp( prefix=tmpprefix, dir=tmppath, text=False ) - if data_type in [ 'tar.gzip', 'tar.bzip' ]: - fp = open( transfer.path, 'r' ) - tar = tarfile.open( mode = 'r:*', bufsize = CHUNK_SIZE, fileobj = fp ) - files = tar.getmembers() - for filename in files: - z = tar.extractfile(filename) - while 1: - try: - chunk = z.read( CHUNK_SIZE ) - except IOError: - os.close( fd ) - log.error( 'Problem decompressing compressed data' ) - exit() - if not chunk: - break - os.write( fd, chunk ) - os.write( fd, '\n' ) - os.close( fd ) - tar.close() - fp.close() - elif data_type == 'gzip': - compressed = gzip.open( transfer.path, mode = 'rb' ) - while 1: - try: - chunk = compressed.read( CHUNK_SIZE ) - except IOError: - compressed.close() - log.error( 'Problem decompressing compressed data' ) - exit() - if not chunk: - break - os.write( fd, chunk ) - os.close( fd ) - compressed.close() - elif data_type == 'bzip': - compressed = bz2.BZ2File( transfer.path, mode = 'r' ) - while 1: - try: - chunk = compressed.read( CHUNK_SIZE ) - except IOError: - compressed.close() - log.error( 'Problem decompressing compressed data' ) - exit() - if not chunk: - break - os.write( fd, chunk ) - os.close( fd ) - compressed.close() - elif data_type == 'zip': - uncompressed_name = None - unzipped = False - z = zipfile.ZipFile( transfer.path ) - z.debug = 3 - for name in z.namelist(): - if name.endswith('/'): - continue - zipped_file = z.open( name ) - while 1: - try: - chunk = zipped_file.read( CHUNK_SIZE ) - except IOError: - os.close( fd ) - log.error( 'Problem decompressing zipped data' ) - return self.app.model.DeferredJob.states.INVALID - if not chunk: - break - os.write( fd, chunk ) - zipped_file.close() - os.close( fd ) - z.close() - elif data_type == 'fasta': - uncompressed = transfer.path - else: - job.state = self.app.model.DeferredJob.states.INVALID - log.error( "Unrecognized compression format for file %s." % transfer.path ) - self.sa_session.add( job ) - self.sa_session.flush() - return - shutil.move( uncompressed, destfilepath ) - if os.path.exists( transfer.path ): - os.remove( transfer.path ) - os.chmod( destfilepath, 0644 ) - fastaline = '\t'.join( [ dbkey, dbkey, params[ 'intname' ], os.path.abspath( destfilepath ) ] ) - self._add_line( 'all_fasta', fastaline ) - if params[ 'indexes' ] is not None: - job.state = self.app.model.DeferredJob.states.WAITING - job.params[ 'indexjobs' ] = [] - else: - job.state = self.app.model.DeferredJob.states.OK - job.params[ 'type' ] = 'finish_transfer' - transfer.path = os.path.abspath(destfilepath) - transfer.state = 'done' - self.sa_session.add( job ) - self.sa_session.add( transfer ) - if transfer.state == 'done': - if params[ 'indexes' ] is not None: - for indexer in params[ 'indexes' ]: - incoming = dict(indexer=indexer, dbkey=params[ 'dbkey' ], intname=params[ 'intname' ], path=transfer.path, user=params['user'] ) - deferred = self.tool.execute( self, set_output_hid=False, history=None, incoming=incoming, transfer=transfer, deferred=job ) - job.params[ 'indexjobs' ].append( deferred[0].id ) - else: - job.state = self.app.model.DeferredJob.states.OK - self.sa_session.add( job ) - self.sa_session.flush() - return self.app.model.DeferredJob.states.OK - - def _check_compress( self, filepath ): - retval = '' - if tarfile.is_tarfile( filepath ): - retval = 'tar.' - if check_zip( filepath ): - return 'zip' - is_bzipped, is_valid = check_bz2( filepath ) - if is_bzipped and is_valid: - return retval + 'bzip' - is_gzipped, is_valid = check_gzip( filepath ) - if is_gzipped and is_valid: - return retval + 'gzip' - return None - - def _add_line( self, locfile, newline ): - filepath = self.locations[ locfile ] - origlines = [] - output = [] - comments = [] - with open( filepath, 'r' ) as destfile: - for line in destfile: - if line.startswith( '#' ): - comments.append( line.strip() ) - else: - origlines.append( line.strip() ) - if newline not in origlines: - origlines.append( newline ) - output.extend( comments ) - origlines.sort() - output.extend( origlines ) - with open( filepath, 'w+' ) as destfile: - destfile.write( '\n'.join( output ) ) - diff -r 4eeb41f2fa7569e1d8652d9a0be3a1f1d5c6c2c0 -r b6fae2cd673c6a286138b9e4628f7c4ef0abb099 lib/galaxy/jobs/deferred/liftover_transfer.py --- a/lib/galaxy/jobs/deferred/liftover_transfer.py +++ /dev/null @@ -1,158 +0,0 @@ -""" -Module for managing genome transfer jobs. -""" -from __future__ import with_statement - -import logging, shutil, gzip, tempfile, sys - -from galaxy import eggs -from sqlalchemy import and_ - -from galaxy.util.odict import odict -from galaxy.workflow.modules import module_factory -from galaxy.jobs.actions.post import ActionBox - -from galaxy.tools.parameters import visit_input_values -from galaxy.tools.parameters.basic import DataToolParameter - -from galaxy.datatypes.checkers import * - -from data_transfer import * - -log = logging.getLogger( __name__ ) - -__all__ = [ 'LiftOverTransferPlugin' ] - -class LiftOverTransferPlugin( DataTransfer ): - - locations = {} - - def __init__( self, app ): - super( LiftOverTransferPlugin, self ).__init__( app ) - self.app = app - self.sa_session = app.model.context.current - - def create_job( self, trans, url, dbkey, from_genome, to_genome, destfile, parentjob ): - job = trans.app.transfer_manager.new( protocol='http', url=url ) - params = dict( user=trans.user.id, transfer_job_id=job.id, protocol='http', - type='init_transfer', dbkey=dbkey, from_genome=from_genome, - to_genome=to_genome, destfile=destfile, parentjob=parentjob ) - deferred = trans.app.model.DeferredJob( state = self.app.model.DeferredJob.states.NEW, plugin = 'LiftOverTransferPlugin', params = params ) - self.sa_session.add( deferred ) - self.sa_session.flush() - return deferred.id - - def check_job( self, job ): - if job.params['type'] == 'init_transfer': - if not hasattr(job, 'transfer_job'): - job.transfer_job = self.sa_session.query( self.app.model.TransferJob ).get( int( job.params[ 'transfer_job_id' ] ) ) - else: - self.sa_session.refresh( job.transfer_job ) - if job.transfer_job.state == 'done': - transfer = job.transfer_job - transfer.state = 'downloaded' - job.params['type'] = 'extract_transfer' - self.sa_session.add( job ) - self.sa_session.add( transfer ) - self.sa_session.flush() - return self.job_states.READY - elif job.transfer_job.state == 'running': - return self.job_states.WAIT - elif job.transfer_job.state == 'new': - assert job.params[ 'protocol' ] in [ 'http', 'ftp', 'https' ], 'Unknown protocol %s' % job.params[ 'protocol' ] - ready = True - parent = self.sa_session.query( self.app.model.DeferredJob ).get( int( job.params[ 'parentjob' ] ) ) - if not hasattr( parent, 'transfer_job' ): - parent.transfer_job = self.sa_session.query( self.app.model.TransferJob ).get( int( parent.params[ 'transfer_job_id' ] ) ) - if parent.transfer_job.state not in [ 'ok', 'error', 'done' ]: - ready = False - for lo_job in parent.params[ 'liftover' ]: - liftoverjob = self.sa_session.query( self.app.model.TransferJob ).get( int( lo_job ) ) - if liftoverjob: - if liftoverjob.state not in [ 'ok', 'error', 'new', 'done' ]: - ready = False - if ready: - self.app.transfer_manager.run( job.transfer_job ) - self.sa_session.add( job.transfer_job ) - self.sa_session.flush() - return self.job_states.WAIT - else: - log.error( "An error occurred while downloading from %s" % job.transfer_job.params[ 'url' ] ) - return self.job_states.INVALID - elif job.params[ 'type' ] == 'extract_transfer': - return self.job_states.READY - - def get_job_status( self, jobid ): - job = self.sa_session.query( self.app.model.DeferredJob ).get( int( jobid ) ) - return job - - def run_job( self, job ): - params = job.params - dbkey = params[ 'dbkey' ] - source = params[ 'from_genome' ] - target = params[ 'to_genome' ] - if not hasattr( job, 'transfer_job' ): - job.transfer_job = self.sa_session.query( self.app.model.TransferJob ).get( int( job.params[ 'transfer_job_id' ] ) ) - else: - self.sa_session.refresh( job.transfer_job ) - transfer = job.transfer_job - if params[ 'type' ] == 'extract_transfer': - CHUNK_SIZE = 2**20 - destpath = os.path.join( self.app.config.get( 'genome_data_path', 'tool-data/genome' ), source, 'liftOver' ) - if not os.path.exists( destpath ): - os.makedirs( destpath ) - destfile = job.params[ 'destfile' ] - destfilepath = os.path.join( destpath, destfile ) - tmpprefix = '%s_%s_download_unzip_' % ( job.params['dbkey'], job.params[ 'transfer_job_id' ] ) - tmppath = os.path.dirname( os.path.abspath( transfer.path ) ) - if not os.path.exists( destpath ): - os.makedirs( destpath ) - fd, uncompressed = tempfile.mkstemp( prefix=tmpprefix, dir=tmppath, text=False ) - chain = gzip.open( transfer.path, 'rb' ) - while 1: - try: - chunk = chain.read( CHUNK_SIZE ) - except IOError: - os.close( fd ) - log.error( 'Problem decompressing compressed data' ) - exit() - if not chunk: - break - os.write( fd, chunk ) - os.close( fd ) - chain.close() - # Replace the gzipped file with the decompressed file if it's safe to do so - shutil.move( uncompressed, destfilepath ) - os.remove( transfer.path ) - os.chmod( destfilepath, 0644 ) - locline = '\t'.join( [ source, target, os.path.abspath( destfilepath ) ] ) - self._add_line( locline ) - job.state = self.app.model.DeferredJob.states.OK - job.params[ 'type' ] = 'finish_transfer' - transfer.path = os.path.abspath(destfilepath) - transfer.state = 'done' - parentjob = self.sa_session.query( self.app.model.DeferredJob ).get( int( job.params[ 'parentjob' ] ) ) - finished = True - for i in parentjob.params[ 'liftover' ]: - sibling = self.sa_session.query( self.app.model.DeferredJob ).get( int( i ) ) - if sibling.state not in [ 'done', 'ok', 'error' ]: - finished = False - if finished: - parentjob.state = self.app.model.DeferredJob.states.OK - self.sa_session.add( parentjob ) - self.sa_session.add( job ) - self.sa_session.add( transfer ) - self.sa_session.flush() - return self.app.model.DeferredJob.states.OK - - def _add_line( self, newline ): - filepath = 'tool-data/liftOver.loc' - origlines = [] - with open( filepath, 'r' ) as destfile: - for line in destfile: - origlines.append( line.strip() ) - if newline not in origlines: - origlines.append( newline ) - with open( filepath, 'w+' ) as destfile: - destfile.write( '\n'.join( origlines ) ) - diff -r 4eeb41f2fa7569e1d8652d9a0be3a1f1d5c6c2c0 -r b6fae2cd673c6a286138b9e4628f7c4ef0abb099 lib/galaxy/tools/actions/index_genome.py --- a/lib/galaxy/tools/actions/index_genome.py +++ /dev/null @@ -1,67 +0,0 @@ -import tempfile -from __init__ import ToolAction -from galaxy.util.odict import odict -from galaxy.tools.genome_index import * - -import logging -log = logging.getLogger( __name__ ) - -class GenomeIndexToolAction( ToolAction ): - """Tool action used for exporting a history to an archive. """ - - def execute( self, tool, trans, *args, **kwargs ): - # - # Get genome to index. - # - incoming = kwargs['incoming'] - # - # Create the job and output dataset objects - # - job = trans.app.model.Job() - job.tool_id = tool.id - job.user_id = incoming['user'] - start_job_state = job.state # should be job.states.NEW - job.state = job.states.WAITING # we need to set job state to something other than NEW, - # or else when tracking jobs in db it will be picked up - # before we have added input / output parameters - trans.sa_session.add( job ) - - # Create dataset that will serve as archive. - temp_dataset = trans.app.model.Dataset( state=trans.app.model.Dataset.states.NEW ) - trans.sa_session.add( temp_dataset ) - - trans.sa_session.flush() # ensure job.id and archive_dataset.id are available - trans.app.object_store.create( temp_dataset ) # set the object store id, create dataset (because galaxy likes having datasets) - - # - # Setup job and job wrapper. - # - - # Add association for keeping track of index jobs, transfer jobs, and so on. - user = trans.sa_session.query( trans.app.model.User ).get( int( incoming['user'] ) ) - assoc = trans.app.model.GenomeIndexToolData( job=job, dataset=temp_dataset, fasta_path=incoming['path'], \ - indexer=incoming['indexer'], user=user, \ - deferred_job=kwargs['deferred'], transfer_job=kwargs['transfer'] ) - trans.sa_session.add( assoc ) - - job_wrapper = GenomeIndexToolWrapper( job ) - cmd_line = job_wrapper.setup_job( assoc ) - - # - # Add parameters to job_parameter table. - # - incoming[ '__GENOME_INDEX_COMMAND__' ] = cmd_line - for name, value in tool.params_to_strings( incoming, trans.app ).iteritems(): - job.add_parameter( name, value ) - - job.state = start_job_state # job inputs have been configured, restore initial job state - job.set_handler(tool.get_job_handler(None)) - trans.sa_session.flush() - - - # Queue the job for execution - trans.app.job_queue.put( job.id, tool.id ) - log.info( "Added genome index job to the job queue, id: %s" % str( job.id ) ) - - return job, odict() - diff -r 4eeb41f2fa7569e1d8652d9a0be3a1f1d5c6c2c0 -r b6fae2cd673c6a286138b9e4628f7c4ef0abb099 lib/galaxy/tools/genome_index/__init__.py --- a/lib/galaxy/tools/genome_index/__init__.py +++ /dev/null @@ -1,243 +0,0 @@ -from __future__ import with_statement - -import json -import logging -import os -import shutil -import tarfile -import tempfile - -from galaxy import model, util -from galaxy.web.framework.helpers import to_unicode -from galaxy.model.item_attrs import UsesAnnotations -from galaxy.util.json import * -from galaxy.web.base.controller import UsesHistoryMixin -from galaxy.tools.data import ToolDataTableManager - - -log = logging.getLogger(__name__) - -def load_genome_index_tools( toolbox ): - """ Adds tools for indexing genomes via the main job runner. """ - # Create XML for loading the tool. - tool_xml_text = """ - <tool id="__GENOME_INDEX__" name="Index Genome" version="0.1" tool_type="genome_index"> - <type class="GenomeIndexTool" module="galaxy.tools"/> - <action module="galaxy.tools.actions.index_genome" class="GenomeIndexToolAction"/> - <command>$__GENOME_INDEX_COMMAND__ $output_file $output_file.files_path "$__app__.config.rsync_url" "$__app__.config.tool_data_path"</command> - <inputs> - <param name="__GENOME_INDEX_COMMAND__" type="hidden"/> - </inputs> - <outputs> - <data format="txt" name="output_file"/> - </outputs> - <stdio> - <exit_code range="1:" err_level="fatal" /> - </stdio> - </tool> - """ - - # Load index tool. - tmp_name = tempfile.NamedTemporaryFile() - tmp_name.write( tool_xml_text ) - tmp_name.flush() - genome_index_tool = toolbox.load_tool( tmp_name.name ) - toolbox.tools_by_id[ genome_index_tool.id ] = genome_index_tool - log.debug( "Loaded genome index tool: %s", genome_index_tool.id ) - -class GenomeIndexToolWrapper( object ): - """ Provides support for performing jobs that index a genome. """ - def __init__( self, job_id ): - self.locations = dict() - self.job_id = job_id - - def setup_job( self, genobj ): - """ Perform setup for job to index a genome and return an archive. Method generates - attribute files, sets the corresponding attributes in the associated database - object, and returns a command line for running the job. The command line - includes the command, inputs, and options; it does not include the output - file because it must be set at runtime. """ - - # - # Create and return command line for running tool. - # - scriptpath = os.path.join( os.path.abspath( os.getcwd() ), "lib/galaxy/tools/genome_index/index_genome.py" ) - return "python %s %s %s" % ( scriptpath, genobj.indexer, genobj.fasta_path ) - - def postprocessing( self, sa_session, app ): - """ Finish the job, move the finished indexes to their final resting place, - and update the .loc files where applicable. """ - gitd = sa_session.query( model.GenomeIndexToolData ).filter_by( job_id=self.job_id ).first() - indexdirs = dict( bfast='bfast_index', bowtie='bowtie_index', bowtie2='bowtie2_index', - bwa='bwa_index', perm='perm_%s_index', picard='srma_index', sam='sam_index' ) - - - if gitd: - fp = open( gitd.dataset.get_file_name(), 'r' ) - deferred = sa_session.query( model.DeferredJob ).filter_by( id=gitd.deferred_job_id ).first() - try: - logloc = json.load( fp ) - except ValueError: - deferred.state = app.model.DeferredJob.states.ERROR - sa_session.add( deferred ) - sa_session.flush() - log.debug( 'Indexing job failed, setting deferred job state to error.' ) - return False - finally: - fp.close() - destination = None - tdtman = ToolDataTableManager( app.config.tool_data_path ) - xmltree = tdtman.load_from_config_file( app.config.tool_data_table_config_path, app.config.tool_data_path ) - for node in xmltree: - table = node.get('name') - location = node.findall('file')[0].get('path') - self.locations[table] = os.path.abspath( location ) - locbase = os.path.abspath( os.path.split( self.locations['all_fasta'] )[0] ) - params = deferred.params - dbkey = params[ 'dbkey' ] - basepath = os.path.join( os.path.abspath( app.config.genome_data_path ), dbkey ) - intname = params[ 'intname' ] - indexer = gitd.indexer - workingdir = os.path.abspath( gitd.dataset.extra_files_path ) - location = [] - indexdata = gitd.dataset.extra_files_path - if indexer == '2bit': - indexdata = os.path.join( workingdir, '%s.2bit' % dbkey ) - destination = os.path.join( basepath, 'seq', '%s.2bit' % dbkey ) - location.append( dict( line='\t'.join( [ 'seq', dbkey, destination ] ), file= os.path.join( locbase, 'alignseq.loc' ) ) ) - elif indexer == 'bowtie': - self._ex_tar( workingdir, 'cs.tar' ) - destination = os.path.join( basepath, 'bowtie_index' ) - for var in [ 'nt', 'cs' ]: - for line in logloc[ var ]: - idx = line - if var == 'nt': - locfile = self.locations[ 'bowtie_indexes' ] - locdir = os.path.join( destination, idx ) - else: - locfile = self.locations[ 'bowtie_indexes_color' ] - locdir = os.path.join( destination, var, idx ) - location.append( dict( line='\t'.join( [ dbkey, dbkey, intname, locdir ] ), file=locfile ) ) - elif indexer == 'bowtie2': - destination = os.path.join( basepath, 'bowtie2_index' ) - for line in logloc[ 'nt' ]: - idx = line - locfile = self.locations[ 'bowtie2_indexes' ] - locdir = os.path.join( destination, idx ) - location.append( dict( line='\t'.join( [ dbkey, dbkey, intname, locdir ] ), file=locfile ) ) - elif indexer == 'bwa': - self._ex_tar( workingdir, 'cs.tar' ) - destination = os.path.join( basepath, 'bwa_index' ) - for var in [ 'nt', 'cs' ]: - for line in logloc[ var ]: - idx = line - if var == 'nt': - locfile = self.locations[ 'bwa_indexes' ] - locdir = os.path.join( destination, idx ) - else: - locfile = self.locations[ 'bwa_indexes_color' ] - locdir = os.path.join( destination, var, idx ) - location.append( dict( line='\t'.join( [ dbkey, dbkey, intname, locdir ] ), file=locfile ) ) - elif indexer == 'perm': - self._ex_tar( workingdir, 'cs.tar' ) - destination = os.path.join( basepath, 'perm_index' ) - for var in [ 'nt', 'cs' ]: - for line in logloc[ var ]: - idx = line.pop() - if var == 'nt': - locfile = self.locations[ 'perm_base_indexes' ] - locdir = os.path.join( destination, idx ) - else: - locfile = self.locations[ 'perm_color_indexes' ] - locdir = os.path.join( destination, var, idx ) - line.append( locdir ) - location.append( dict( line='\t'.join( line ), file=locfile ) ) - elif indexer == 'picard': - destination = os.path.join( basepath, 'srma_index' ) - for var in [ 'nt' ]: - for line in logloc[ var ]: - idx = line - locfile = self.locations[ 'picard_indexes' ] - locdir = os.path.join( destination, idx ) - location.append( dict( line='\t'.join( [ dbkey, dbkey, intname, locdir ] ), file=locfile ) ) - elif indexer == 'sam': - destination = os.path.join( basepath, 'sam_index' ) - for var in [ 'nt' ]: - for line in logloc[ var ]: - locfile = self.locations[ 'sam_fa_indexes' ] - locdir = os.path.join( destination, line ) - location.append( dict( line='\t'.join( [ 'index', dbkey, locdir ] ), file=locfile ) ) - - if destination is not None and os.path.exists( os.path.split( destination )[0] ) and not os.path.exists( destination ): - log.debug( 'Moving %s to %s' % ( indexdata, destination ) ) - shutil.move( indexdata, destination ) - if indexer not in [ '2bit' ]: - genome = '%s.fa' % dbkey - target = os.path.join( destination, genome ) - fasta = os.path.abspath( os.path.join( basepath, 'seq', genome ) ) - self._check_link( fasta, target ) - if os.path.exists( os.path.join( destination, 'cs' ) ): - target = os.path.join( destination, 'cs', genome ) - fasta = os.path.abspath( os.path.join( basepath, 'seq', genome ) ) - self._check_link( fasta, target ) - for line in location: - self._add_line( line[ 'file' ], line[ 'line' ] ) - deferred.state = app.model.DeferredJob.states.OK - sa_session.add( deferred ) - sa_session.flush() - - - def _check_link( self, targetfile, symlink ): - target = os.path.relpath( targetfile, os.path.dirname( symlink ) ) - filename = os.path.basename( targetfile ) - if not os.path.exists( targetfile ): # this should never happen. - raise Exception, "%s not found. Unable to proceed without a FASTA file. Aborting." % targetfile - if os.path.exists( symlink ) and os.path.islink( symlink ): - if os.path.realpath( symlink ) == os.path.abspath( targetfile ): # symlink exists, points to the correct FASTA file. - return - else: # no it doesn't. Make a new one, and this time do it right. - os.remove( symlink ) - os.symlink( target, symlink ) - return - elif not os.path.exists( symlink ): # no symlink to the FASTA file. Create one. - os.symlink( target, symlink ) - return - elif os.path.exists( symlink ) and not os.path.islink( symlink ): - if self._hash_file( targetfile ) == self._hash_file( symlink ): # files are identical. No need to panic. - return - else: - if os.path.getsize( symlink ) == 0: # somehow an empty file got copied instead of the symlink. Delete with extreme prejudice. - os.remove( symlink ) - os.symlink( target, symlink ) - return - else: - raise Exception, "Regular file %s exists, is not empty, contents do not match %s." % ( symlink, targetfile ) - - def _hash_file( self, filename ): - import hashlib - md5 = hashlib.md5() - with open( filename, 'rb' ) as f: - for chunk in iter( lambda: f.read( 8192 ), '' ): - md5.update( chunk ) - return md5.digest() - - - def _ex_tar( self, directory, filename ): - fh = tarfile.open( os.path.join( directory, filename ) ) - fh.extractall( path=directory ) - fh.close() - os.remove( os.path.join( directory, filename ) ) - - def _add_line( self, locfile, newline ): - filepath = locfile - origlines = [] - output = [] - comments = [] - with open( filepath, 'r' ) as destfile: - for line in destfile: - origlines.append( line.strip() ) - if newline not in origlines: - origlines.append( newline ) - with open( filepath, 'w+' ) as destfile: - origlines.append( '' ) - destfile.write( '\n'.join( origlines ) ) diff -r 4eeb41f2fa7569e1d8652d9a0be3a1f1d5c6c2c0 -r b6fae2cd673c6a286138b9e4628f7c4ef0abb099 lib/galaxy/tools/genome_index/index_genome.py --- a/lib/galaxy/tools/genome_index/index_genome.py +++ /dev/null @@ -1,327 +0,0 @@ -#!/usr/bin/env python -""" -Export a history to an archive file using attribute files. - -usage: %prog history_attrs dataset_attrs job_attrs out_file - -G, --gzip: gzip archive file -""" -from __future__ import with_statement - -import json -import optparse -import os -import shlex -import shutil -import subprocess -import sys -import tarfile -import tempfile -import time - - -class ManagedIndexer(): - def __init__( self, output_file, infile, workingdir, rsync_url, tooldata ): - self.tooldatapath = os.path.abspath( tooldata ) - self.workingdir = os.path.abspath( workingdir ) - self.outfile = open( os.path.abspath( output_file ), 'w' ) - self.basedir = os.path.split( self.workingdir )[0] - self.fasta = os.path.abspath( infile ) - self.locations = dict( nt=[], cs=[] ) - self.log = [] - self.rsync_opts = '-aclSzq' - self.rsync_url = rsync_url - self.indexers = { - 'bwa': '_bwa', - 'bowtie': '_bowtie', - 'bowtie2': '_bowtie2', - '2bit': '_twobit', - 'perm': '_perm', - 'bfast': '_bfast', - 'picard': '_picard', - 'sam': '_sam' - } - if not os.path.exists( self.workingdir ): - os.makedirs( self.workingdir ) - self.logfile = open( os.path.join( self.workingdir, 'ManagedIndexer.log' ), 'w+' ) - - def run_indexer( self, indexer ): - self.fapath = self.fasta - self.fafile = os.path.basename( self.fapath ) - self.genome = os.path.splitext( self.fafile )[0] - with WithChDir( self.basedir ): - if indexer not in self.indexers: - sys.stderr.write( 'The requested indexing function does not exist' ) - exit(127) - else: - with WithChDir( self.workingdir ): - self._log( 'Running indexer %s.' % indexer ) - result = getattr( self, self.indexers[ indexer ] )() - if result in [ None, False ]: - sys.stderr.write( 'Error running indexer %s, %s' % ( indexer, result ) ) - self._flush_files() - exit(1) - else: - self._log( self.locations ) - self._log( 'Indexer %s completed successfully.' % indexer ) - self._flush_files() - exit(0) - - def _check_link( self ): - self._log( 'Checking symlink to %s' % self.fafile ) - if not os.path.exists( self.fafile ): - self._log( 'Symlink not found, creating' ) - os.symlink( os.path.relpath( self.fapath ), self.fafile ) - - def _do_rsync( self, idxpath ): - self._log( 'Trying rsync at %s/%s%s' % ( self.rsync_url, self.genome, idxpath ) ) - result = subprocess.call( shlex.split( 'rsync %s %s/%s%s .' % ( self.rsync_opts, self.rsync_url, self.genome, idxpath ) ), stderr=self.logfile ) - if result != 0: - self._log( 'Rsync failed or index not found. Generating.' ) - else: - self._log( 'Rsync succeeded.' ) - return result - - def _flush_files( self ): - json.dump( self.locations, self.outfile ) - self.outfile.close() - self.logfile.close() - - def _log( self, stuff ): - timestamp = time.strftime('%Y-%m-%d %H:%M:%S %z') - self.logfile.write( "[%s] %s\n" % (timestamp, stuff) ) - - def _bwa( self ): - result = self._do_rsync( '/bwa_index/' ) - if result == 0: - self.locations[ 'nt' ].append( self.fafile ) - return self._bwa_cs() - else: - self._check_link() - command = shlex.split( 'bwa index -a bwtsw %s' % self.fafile ) - result = subprocess.call( command, stderr=self.logfile, stdout=self.logfile ) - if result != 0: - newcommand = shlex.split( 'bwa index -c %s' % self.fafile ) - result = call( newcommand, stderr=self.logfile, stdout=self.logfile ) - if result == 0: - self.locations[ 'nt' ].append( self.fafile ) - os.remove( self.fafile ) - return self._bwa_cs() - else: - self._log( 'BWA (base) exited with code %s' % result ) - return False - - def _bwa_cs( self ): - if not os.path.exists( os.path.join( self.workingdir, 'cs' ) ): - os.makedirs( 'cs' ) - with WithChDir( 'cs' ): - self._check_link() - command = shlex.split( 'bwa index -a bwtsw -c %s' % self.fafile ) - result = subprocess.call( command, stderr=self.logfile, stdout=self.logfile ) - if result != 0: - newcommand = shlex.split( 'bwa index -c %s' % self.fafile ) - result = call( newcommand, stderr=self.logfile, stdout=self.logfile ) - if result == 0: - self.locations[ 'cs' ].append( self.fafile ) - os.remove( self.fafile ) - else: - self._log( 'BWA (color) exited with code %s' % result ) - return False - else: - self.locations[ 'cs' ].append( self.fafile ) - os.remove( self.fafile ) - else: - self.locations[ 'cs' ].append( self.fafile ) - temptar = tarfile.open( 'cs.tar', 'w' ) - temptar.add( 'cs' ) - temptar.close() - shutil.rmtree( 'cs' ) - return True - - - def _bowtie( self ): - result = self._do_rsync( '/bowtie_index/' ) - if result == 0: - self.locations[ 'nt' ].append( self.genome ) - return self._bowtie_cs() - else: - self._check_link() - command = shlex.split( 'bowtie-build -f %s %s' % ( self.fafile, self.genome ) ) - result = subprocess.call( command, stderr=self.logfile, stdout=self.logfile ) - if result == 0: - self.locations[ 'nt' ].append( self.genome ) - os.remove( self.fafile ) - return self._bowtie_cs() - else: - self._log( 'Bowtie (base) exited with code %s' % result ) - return False - - def _bowtie_cs( self ): - indexdir = os.path.join( os.getcwd(), 'cs' ) - if not ( os.path.exists( indexdir ) ): - os.makedirs( indexdir ) - with WithChDir( indexdir ): - self._check_link() - command = shlex.split( 'bowtie-build -C -f %s %s' % ( self.fafile, self.genome ) ) - result = subprocess.call( command, stderr=self.logfile, stdout=self.logfile ) - if result == 0: - self.locations[ 'cs' ].append( self.genome ) - else: - self._log( 'Bowtie (color) exited with code %s' % result ) - return False - os.remove( os.path.join( indexdir, self.fafile ) ) - else: - self.locations[ 'cs' ].append( self.genome ) - temptar = tarfile.open( 'cs.tar', 'w' ) - temptar.add( 'cs' ) - temptar.close() - shutil.rmtree( 'cs' ) - return True - - - def _bowtie2( self ): - result = self._do_rsync( '/bowtie2_index/' ) - if result == 0: - self.locations[ 'nt' ].append( self.fafile ) - return True - ref_base = os.path.splitext(self.fafile)[0] - self._check_link() - command = shlex.split( 'bowtie2-build %s %s' % ( self.fafile, ref_base ) ) - result = subprocess.call( command, stderr=self.logfile, stdout=self.logfile ) - if result == 0: - self.locations[ 'nt' ].append( ref_base ) - os.remove( self.fafile ) - return True - else: - self._log( 'Bowtie2 exited with code %s' % result ) - return False - - def _twobit( self ): - """Index reference files using 2bit for random access. - """ - result = self._do_rsync( '/seq/%s.2bit' % self.genome ) - if result == 0: - self.locations['nt'].append( "%s.2bit" % self.genome ) - return True - else: - out_file = "%s.2bit" % self.genome - self._check_link() - command = shlex.split( 'faToTwoBit %s %s' % ( self.fafile, out_file ) ) - result = subprocess.call( command, stderr=self.logfile, stdout=self.logfile ) - if result == 0: - self.locations['nt'].append( out_file ) - os.remove( self.fafile ) - return True - else: - self._log( 'faToTwoBit exited with code %s' % result ) - return False - - def _perm( self ): - result = self._do_rsync( '/perm_index/' ) - self._check_link() - genome = self.genome - read_length = 50 - for seed in [ 'F3', 'F4' ]: - key = '%s_%s_%s' % (self.genome, seed, read_length) - desc = '%s: seed=%s, read length=%s' % (self.genome, seed, read_length) - index = "%s_base_%s_%s.index" % (self.genome, seed, read_length) - if not os.path.exists( index ): - command = shlex.split("PerM %s %s --readFormat fastq --seed %s -m -s %s" % (self.fafile, read_length, seed, index)) - result = subprocess.call( command ) - if result != 0: - self._log( 'PerM (base) exited with code %s' % result ) - return False - self.locations[ 'nt' ].append( [ key, desc, index ] ) - os.remove( self.fafile ) - return self._perm_cs() - - def _perm_cs( self ): - genome = self.genome - read_length = 50 - if not os.path.exists( 'cs' ): - os.makedirs( 'cs' ) - with WithChDir( 'cs' ): - self._check_link() - for seed in [ 'F3', 'F4' ]: - key = '%s_%s_%s' % (genome, seed, read_length) - desc = '%s: seed=%s, read length=%s' % (genome, seed, read_length) - index = "%s_color_%s_%s.index" % (genome, seed, read_length) - if not os.path.exists( index ): - command = shlex.split("PerM %s %s --readFormat csfastq --seed %s -m -s %s" % (self.fafile, read_length, seed, index)) - result = subprocess.call( command, stderr=self.logfile, stdout=self.logfile ) - if result != 0: - self._log( 'PerM (color) exited with code %s' % result ) - return False - self.locations[ 'cs' ].append( [ key, desc, index ] ) - os.remove( self.fafile ) - temptar = tarfile.open( 'cs.tar', 'w' ) - temptar.add( 'cs' ) - temptar.close() - shutil.rmtree( 'cs' ) - return True - - def _picard( self ): - result = self._do_rsync( '/srma_index/' ) - if result == 0 and os.path.exists( '%s.dict' % self.genome): - self.locations[ 'nt' ].append( self.fafile ) - return True - local_ref = self.fafile - srma = os.path.abspath( os.path.join( self.tooldatapath, 'shared/jars/picard/CreateSequenceDictionary.jar' ) ) - genome = os.path.splitext( self.fafile )[0] - self._check_link() - if not os.path.exists( '%s.fai' % self.fafile ) and not os.path.exists( '%s.fai' % self.genome ): - command = shlex.split( 'samtools faidx %s' % self.fafile ) - subprocess.call( command, stderr=self.logfile ) - command = shlex.split( "java -jar %s R=%s O=%s.dict URI=%s" \ - % ( srma, local_ref, genome, local_ref ) ) - if not os.path.exists( '%s.dict' % self.genome ): - result = subprocess.call( command, stderr=self.logfile, stdout=self.logfile ) - self._log( ' '.join( command ) ) - if result != 0: - self._log( 'Picard exited with code %s' % result ) - return False - self.locations[ 'nt' ].append( self.fafile ) - os.remove( self.fafile ) - return True - - def _sam( self ): - local_ref = self.fafile - local_file = os.path.splitext( self.fafile )[ 0 ] - print 'Trying rsync' - result = self._do_rsync( '/sam_index/' ) - if result == 0 and ( os.path.exists( '%s.fai' % self.fafile ) or os.path.exists( '%s.fai' % self.genome ) ): - self.locations[ 'nt' ].append( '%s.fai' % local_ref ) - return True - self._check_link() - print 'Trying indexer' - command = shlex.split("samtools faidx %s" % local_ref) - result = subprocess.call( command, stderr=self.logfile, stdout=self.logfile ) - if result != 0: - self._log( 'SAM exited with code %s' % result ) - return False - else: - self.locations[ 'nt' ].append( '%s.fai' % local_ref ) - os.remove( local_ref ) - return True - -class WithChDir(): - def __init__( self, target ): - self.working = target - self.previous = os.getcwd() - def __enter__( self ): - os.chdir( self.working ) - def __exit__( self, *args ): - os.chdir( self.previous ) - - -if __name__ == "__main__": - # Parse command line. - parser = optparse.OptionParser() - (options, args) = parser.parse_args() - indexer, infile, outfile, working_dir, rsync_url, tooldata = args - - # Create archive. - idxobj = ManagedIndexer( outfile, infile, working_dir, rsync_url, tooldata ) - returncode = idxobj.run_indexer( indexer ) - if not returncode: - exit(1) - exit(0) diff -r 4eeb41f2fa7569e1d8652d9a0be3a1f1d5c6c2c0 -r b6fae2cd673c6a286138b9e4628f7c4ef0abb099 lib/galaxy/webapps/galaxy/controllers/data_admin.py --- a/lib/galaxy/webapps/galaxy/controllers/data_admin.py +++ /dev/null @@ -1,299 +0,0 @@ -import ftplib -import json -import sys -from galaxy import model, util -from galaxy.jobs import transfer_manager -from galaxy.model.orm import * -from galaxy.web.base.controller import * -from galaxy.web.framework.helpers import grids, iff, time_ago -from library_common import get_comptypes, lucene_search, whoosh_search - - -# Older py compatibility -try: - set() -except: - from sets import Set as set - -import logging -log = logging.getLogger( __name__ ) - -class DataAdmin( BaseUIController ): - jobstyles = dict( - done='panel-done-message', - waiting='state-color-waiting', - running='state-color-running', - downloaded='state-color-running', - new='state-color-new', - ok='panel-done-message', - error='panel-error-message', - queued='state-color-waiting' - ) - - @web.expose - @web.require_admin - def manage_data( self, trans, **kwd ): - if trans.app.config.get_bool( 'enable_beta_job_managers', False ) == False: - return trans.fill_template( '/admin/data_admin/generic_error.mako', message='This feature requires that enable_beta_job_managers be set to True in your Galaxy configuration.' ) - if 'all_fasta' not in trans.app.tool_data_tables.data_tables: - return trans.fill_template( '/admin/data_admin/generic_error.mako', message='The local data manager requires that an all_fasta entry exists in your tool_data_table_conf.xml.' ) - indextable = {} - dbkeys = [] - labels = { 'bowtie_indexes': 'Bowtie', 'bowtie2_indexes': 'Bowtie 2', 'bwa_indexes': 'BWA', 'srma_indexes': 'Picard', 'sam_fa_indexes': 'SAM', 'perm_base_indexes': 'PerM' } - tablenames = { 'Bowtie': 'bowtie_indexes', 'Bowtie 2': 'bowtie2_indexes', 'BWA': 'bwa_indexes', 'Picard': 'srma_indexes', 'SAM': 'sam_fa_indexes', 'PerM': 'perm_base_indexes' } - indexfuncs = dict( bowtie_indexes='bowtie', bowtie2_indexes='bowtie2', bwa_indexes='bwa', srma_indexes='picard', sam_fa_indexes='sam', perm_base_indexes='perm' ) - for genome in trans.app.tool_data_tables.data_tables[ 'all_fasta' ].data: - dbkey = genome[0] - dbkeys.append( dbkey ) - indextable[ dbkey ] = dict( indexes=dict(), name=genome[2], path=genome[3] ) - for genome in indextable: - for label in labels: - indextable[ genome ][ 'indexes' ][ label ] = 'Generate' - if label not in trans.app.tool_data_tables.data_tables: - indextable[ genome ][ 'indexes' ][ label ] = 'Disabled' - else: - for row in trans.app.tool_data_tables.data_tables[ label ].data: - if genome in row or row[0].startswith( genome ): - indextable[ genome ][ 'indexes' ][ label ] = 'Generated' - jobgrid = [] - sa_session = trans.app.model.context.current - jobs = sa_session.query( model.GenomeIndexToolData ).order_by( model.GenomeIndexToolData.created_time.desc() ).filter_by( user_id=trans.get_user().id ).group_by( model.GenomeIndexToolData.deferred ).limit( 20 ).all() - prevjobid = 0 - for job in jobs: - if prevjobid == job.deferred.id: - continue - prevjobid = job.deferred.id - state = job.deferred.state - params = job.deferred.params - if job.transfer is not None: - jobtype = 'download' - else: - jobtype = 'index' - indexers = ', '.join( params['indexes'] ) - jobgrid.append( dict( jobtype=jobtype, indexers=indexers, rowclass=state, deferred=job.deferred.id, state=state, intname=job.deferred.params[ 'intname' ], dbkey=job.deferred.params[ 'dbkey' ] ) ) - styles = dict( Generate=self.jobstyles['new'], Generated=self.jobstyles['ok'], Disabled=self.jobstyles['error'] ) - return trans.fill_template( '/admin/data_admin/local_data.mako', jobgrid=jobgrid, indextable=indextable, labels=labels, dbkeys=dbkeys, styles=styles, indexfuncs=indexfuncs ) - - @web.expose - @web.require_admin - def add_genome( self, trans, **kwd ): - if trans.app.config.get_bool( 'enable_beta_job_managers', False ) == False: - return trans.fill_template( '/admin/data_admin/generic_error.mako', message='This feature requires that enable_beta_job_managers be set to True in your Galaxy configuration.' ) - dbkeys = trans.ucsc_builds - ensemblkeys = trans.ensembl_builds - ncbikeys = trans.ncbi_builds - return trans.fill_template( '/admin/data_admin/data_form.mako', dbkeys=dbkeys, ensembls=ensemblkeys, ncbi=ncbikeys ) - - @web.expose - @web.require_admin - def genome_search( self, trans, **kwd ): - results = list() - ncbikeys = trans.ncbi_builds - params = util.Params( kwd ) - search = params.get( 'q', None ) - limit = params.get( 'limit', None ) - if search is not None: - query = search.lower() - for row in ncbikeys: - if query in row[ 'name' ].lower() or query in row[ 'dbkey' ].lower(): - result = '|'.join( [ ': '.join( [ row[ 'dbkey' ], row[ 'name' ] ] ), row[ 'dbkey' ] ] ) - results.append( result ) - if len( results ) >= limit: - break - return trans.fill_template( '/admin/data_admin/ajax_status.mako', json='\n'.join( results ) ) - - @web.expose - @web.require_admin - def index_build( self, trans, **kwd ): - """Index a previously downloaded genome.""" - params = util.Params( kwd ) - path = os.path.abspath( params.get( 'path', None ) ) - indexes = [ params.get( 'indexes', None ) ] - dbkey = params.get( 'dbkey', None ) - intname = params.get( 'longname', None ) - indexjob = trans.app.job_manager.deferred_job_queue.plugins['GenomeIndexPlugin'].create_job( trans, path, indexes, dbkey, intname ) - return indexjob - - @web.expose - @web.require_admin - def download_build( self, trans, **kwd ): - """Download a genome from a remote source and add it to the library.""" - params = util.Params( kwd ) - paramdict = build_param_dict( params, trans ) - if paramdict[ 'status' ] == 'error': - return trans.fill_template( '/admin/data_admin/generic_error.mako', message=paramdict[ 'message' ] ) - url = paramdict[ 'url' ] - liftover = paramdict[ 'liftover' ] - dbkey = paramdict[ 'dbkey' ] - indexers = paramdict[ 'indexers' ] - longname = paramdict[ 'longname' ] - dbkeys = dict() - protocol = 'http' - if url is None: - return trans.fill_template( '/admin/data_admin/generic_error.mako', message='Unable to generate a valid URL with the specified parameters.' ) - jobid = trans.app.job_manager.deferred_job_queue.plugins['GenomeTransferPlugin'].create_job( trans, url, dbkey, longname, indexers ) - chainjob = [] - if liftover is not None: - for chain in liftover: - liftover_url = u'ftp://hgdownload.cse.ucsc.edu%s' % chain[0] - from_genome = chain[1] - to_genome = chain[2] - destfile = liftover_url.split('/')[-1].replace('.gz', '') - lochain = trans.app.job_manager.deferred_job_queue.plugins['LiftOverTransferPlugin'].create_job( trans, liftover_url, dbkey, from_genome, to_genome, destfile, jobid ) - chainjob.append( lochain ) - job = trans.app.job_manager.deferred_job_queue.plugins['GenomeTransferPlugin'].get_job_status( jobid ) - job.params['liftover'] = chainjob - trans.app.model.context.current.add( job ) - trans.app.model.context.current.flush() - return trans.response.send_redirect( web.url_for( controller='data_admin', - action='monitor_status', - job=jobid ) ) - - @web.expose - @web.require_admin - def monitor_status( self, trans, **kwd ): - params = util.Params( kwd ) - jobid = params.get( 'job', '' ) - deferred = trans.app.model.context.current.query( model.DeferredJob ).filter_by( id=jobid ).first() - if deferred is None: - return trans.fill_template( '/admin/data_admin/generic_error.mako', message='Invalid genome downloader job specified.' ) - gname = deferred.params[ 'intname' ] - indexers = ', '.join( deferred.params[ 'indexes' ] ) - jobs = self._get_jobs( deferred, trans ) - jsonjobs = json.dumps( jobs ) - return trans.fill_template( '/admin/data_admin/download_status.mako', name=gname, indexers=indexers, mainjob=jobid, jobs=jobs, jsonjobs=jsonjobs ) - - @web.expose - @web.require_admin - def get_jobs( self, trans, **kwd ): - sa_session = trans.app.model.context.current - jobs = [] - params = util.Params( kwd ) - jobid = params.get( 'jobid', '' ) - job = sa_session.query( model.DeferredJob ).filter_by( id=jobid ).first() - jobs = self._get_jobs( job, trans ) - return trans.fill_template( '/admin/data_admin/ajax_status.mako', json=json.dumps( jobs ) ) - - def _get_job( self, jobid, jobtype, trans ): - sa = trans.app.model.context.current - if jobtype == 'liftover': - liftoverjob = sa.query( model.DeferredJob ).filter_by( id=jobid ).first() - job = sa.query( model.TransferJob ).filter_by( id=liftoverjob.params[ 'transfer_job_id' ] ).first() - joblabel = 'Download liftOver (%s to %s)' % ( liftoverjob.params[ 'from_genome' ], liftoverjob.params[ 'to_genome' ] ) - elif jobtype == 'transfer': - job = sa.query( model.TransferJob ).filter_by( id=jobid ).first() - joblabel = 'Download Genome' - elif jobtype == 'deferred': - job = sa.query( model.DeferredJob ).filter_by( id=jobid ).first() - joblabel = 'Main Controller' - elif jobtype == 'index': - job = sa.query( model.Job ).filter_by( id=jobid.job_id ).first() - joblabel = 'Index Genome (%s)' % jobid.indexer - return dict( status=job.state, jobid=job.id, style=self.jobstyles[job.state], type=jobtype, label=joblabel ) - - def _get_jobs( self, deferredjob, trans ): - jobs = [] - idxjobs = [] - sa_session = trans.app.model.context.current - job = sa_session.query( model.GenomeIndexToolData ).filter_by( deferred=deferredjob ).first() - jobs.append( self._get_job( deferredjob.id, 'deferred', trans ) ) - if 'transfer_job_id' in deferredjob.params: #hasattr( job, 'transfer' ) and job.transfer is not None: # This is a transfer job, check for indexers - jobs.append( self._get_job( deferredjob.params[ 'transfer_job_id' ], 'transfer', trans ) ) - if hasattr( job, 'deferred' ): - idxjobs = sa_session.query( model.GenomeIndexToolData ).filter_by( deferred=job.deferred, transfer=job.transfer ).all() - if deferredjob.params.has_key( 'liftover' ) and deferredjob.params[ 'liftover' ] is not None: - for jobid in deferredjob.params[ 'liftover' ]: - jobs.append( self._get_job( jobid, 'liftover', trans ) ) - for idxjob in idxjobs: - jobs.append( self._get_job( idxjob, 'index', trans ) ) - return jobs - -def build_param_dict( params, trans ): - - source = params.get('source', '') - longname = params.get('longname', None) - if not isinstance( params.get( 'indexers', None ), list ): - indexers = [ params.get( 'indexers', None ) ] - else: - indexers = params.get( 'indexers', None ) - if indexers is not None: - if indexers == [None]: - indexers = None - url = None - liftover = None - newlift = [] - dbkey = params.get( 'dbkey', None ) - dbkeys = dict() - protocol = 'http' - - if source == 'NCBI': - build = params.get('ncbi_name', '') - dbkey = build.split( ': ' )[0] - longname = build.split( ': ' )[-1] - url = 'http://togows.dbcls.jp/entry/ncbi-nucleotide/%s.fasta' % dbkey - elif source == 'URL': - dbkey = params.get( 'url_dbkey', '' ) - url = params.get( 'url', None ) - longname = params.get( 'longname', None ) - elif source == 'UCSC': - longname = None - for build in trans.ucsc_builds: - if dbkey == build[0]: - dbkey = build[0] - longname = build[1] - break - if dbkey == '?': - return dict( status='error', message='An invalid build was specified.' ) - ftp = ftplib.FTP('hgdownload.cse.ucsc.edu') - ftp.login('anonymous', trans.get_user().email) - checker = [] - liftover = [] - newlift = [] - ftp.retrlines('NLST /goldenPath/%s/liftOver/*.chain.gz' % dbkey, liftover.append) - try: - for chain in liftover: - lifts = [] - fname = chain.split( '/' )[-1] - organisms = fname.replace( '.over.chain.gz', '' ).split( 'To' ) - lifts.append( [ organisms[0], organisms[1][0].lower() + organisms[1][1:] ] ) - lifts.append( [ organisms[1][0].lower() + organisms[1][1:], organisms[0] ] ) - for organism in lifts: - remotepath = '/goldenPath/%s/liftOver/%sTo%s.over.chain.gz' % ( organism[0], organism[0], organism[1][0].upper() + organism[1][1:] ) - localfile = '%sTo%s.over.chain' % ( organism[0], organism[1][0].upper() + organism[1][1:] ) - localpath = os.path.join( trans.app.config.get( 'genome_data_path', 'tool-data/genome' ), organism[0], 'liftOver', localfile ) - if not os.path.exists( localpath ) or os.path.getsize( localpath ) == 0: - newlift.append( [ remotepath, organism[0], organism[1] ] ) - except: - newlift = None - pass - ftp.retrlines('NLST /goldenPath/%s/bigZips/' % dbkey, checker.append) - ftp.quit() - for filename in [ dbkey, 'chromFa' ]: - for extension in [ '.tar.gz', '.tar.bz2', '.zip', '.fa.gz', '.fa.bz2' ]: - testfile = '/goldenPath/%s/bigZips/%s%s' % ( dbkey, filename, extension ) - if testfile in checker: - url = 'ftp://hgdownload.cse.ucsc.edu%s' % testfile - break; - else: - continue - if url is None: - message = 'The genome %s was not found on the UCSC server.' % dbkey - status = 'error' - return dict( status=status, message=message ) - - elif source == 'Ensembl': - dbkey = params.get( 'ensembl_dbkey', None ) - if dbkey == '?': - return dict( status='error', message='An invalid build was specified.' ) - for build in trans.ensembl_builds: - if build[ 'dbkey' ] == dbkey: - dbkey = build[ 'dbkey' ] - release = build[ 'release' ] - pathname = '_'.join( build[ 'name' ].split(' ')[0:2] ) - longname = build[ 'name' ].replace('_', ' ') - break - url = 'ftp://ftp.ensembl.org/pub/release-%s/fasta/%s/dna/%s.%s.%s.dna.toplevel.fa.…' % ( release, pathname.lower(), pathname, dbkey, release ) - - params = dict( status='ok', dbkey=dbkey, datatype='fasta', url=url, user=trans.user.id, liftover=newlift, longname=longname, indexers=indexers ) - - return params diff -r 4eeb41f2fa7569e1d8652d9a0be3a1f1d5c6c2c0 -r b6fae2cd673c6a286138b9e4628f7c4ef0abb099 templates/admin/data_admin/ajax_status.mako --- a/templates/admin/data_admin/ajax_status.mako +++ /dev/null @@ -1,1 +0,0 @@ -${json} \ No newline at end of file diff -r 4eeb41f2fa7569e1d8652d9a0be3a1f1d5c6c2c0 -r b6fae2cd673c6a286138b9e4628f7c4ef0abb099 templates/admin/data_admin/data_form.mako --- a/templates/admin/data_admin/data_form.mako +++ /dev/null @@ -1,178 +0,0 @@ -<%inherit file="/base.mako"/> -<%namespace file="/message.mako" import="render_msg" /> -<%namespace file="/library/common/common.mako" import="common_javascripts" /> - -<%! - def inherit(context): - if context.get('use_panels'): - return '/webapps/galaxy/base_panels.mako' - else: - return '/base.mako' -%> -<%inherit file="${inherit(context)}"/> - -<%def name="init()"> -<% - self.has_left_panel=False - self.has_right_panel=False - self.message_box_visible=False - self.active_view="user" - self.overlay_visible=False - self.has_accessible_datasets = False -%> -</%def> -<%def name="stylesheets()"> - ${parent.stylesheets()} - ${h.css( "autocomplete_tagging" )} -</%def> -<%def name="javascripts()"> - ${parent.javascripts()} - ${h.js("libs/jquery/jquery.autocomplete", "galaxy.autocom_tagging" )} -</%def> -## -## Override methods from base.mako and base_panels.mako -## -<%def name="center_panel()"> - <div style="overflow: auto; height: 100%;"> - <div class="page-container" style="padding: 10px;"> - ${render_content()} - </div> - </div> -</%def> -<style type="text/css"> - .params-block { display: none; } -</style> -<div class="toolForm"> - %if message: - <div class="${status}">${message}</div> - %endif - <div class="toolFormTitle">Get build from a remote server</div> - <div class="toolFormBody"> - <form name="download_build" action="${h.url_for( controller='data_admin', action='download_build' )}" enctype="multipart/form-data" method="post"> - <div class="form-row"> - <label for="source">Data Source</label> - <select id="datasource" name="source" label="Data Source"> - <option value="UCSC">UCSC</option> - <option value="URL">Direct Link</option> - <option value="NCBI">NCBI</option> - <option value="Ensembl">EnsemblGenome</option> - </select> - <div style="clear: both;"> </div> - </div> - <div class="form-row"> - <label for="indexers">Indexers</label> - <select name="indexers" multiple style="width: 200px; height: 125px;"> - <option value="2bit" selected>TwoBit</option> - <option value="bowtie">Bowtie</option> - <option value="bowtie2">Bowtie 2</option> - <option value="bwa">BWA</option> - <option value="perm">PerM</option> - <option value="picard">Picard</option> - <option value="sam">sam</option> - </select> - <div class="toolParamHelp" style="clear: both;"> - Select the indexers you want to run on the FASTA file after downloading. - </div> - </div> - <h2>Parameters</h2> - <div id="params_URL" class="params-block"> - <div class="form-row"> - <label for="longname">Long Name</label> - <input name="longname" type="text" label="Long Name" /> - <div style="clear: both;"> </div> - <div class="toolParamHelp" style="clear: both;"> - A descriptive name for this build. - </div> - </div> - <div class="form-row"> - <label for="url_dbkey">DB Key</label> - <input name="url_dbkey" type="text" label="DB Key" /> - <div style="clear: both;"> </div> - <div class="toolParamHelp" style="clear: both;"> - The internal DB key for this build. WARNING: Using a value that already exists in one or more .loc files may have unpredictable results. - </div> - </div> - <div id="dlparams"> - <div class="form-row"> - <label for="url">URL</label> - <input name="url" type="text" label="URL" /> - <div style="clear: both;"> </div> - <div class="toolParamHelp" style="clear: both;"> - The URL to download this build from. - </div> - </div> - </div> - </div> - <div id="params_NCBI" class="params-block"> - <div class="form-row"> - <label>Genome:</label> - <div class="form-row-input"> - <input type="text" class="text-and-autocomplete-select ac_input" size="40" name="ncbi_name" id="ncbi_name" value="" /> - </div> - <div class="toolParamHelp" style="clear: both;"> - If you can't find the build you want in this list, open a terminal and execute - <pre>sh cron/updatencbi.sh</pre> - in your galaxy root directory. - </div> - </div> - </div> - <div id="params_Ensembl" class="params-block"> - <div class="form-row"> - <label>Genome:</label> - <div class="form-row-input"> - <select name="ensembl_dbkey" last_selected_value="?"> - %for dbkey in ensembls: - <option value="${dbkey['dbkey']}">${dbkey['dbkey']} - ${dbkey['name']}</option> - %endfor - </select> - </div> - <div class="toolParamHelp" style="clear: both;"> - If you can't find the build you want in this list, open a terminal and execute - <pre>sh cron/updateensembl.sh</pre> - in your galaxy root directory. - </div> - </div> - </div> - <div id="params_UCSC" class="params-block"> - <div class="form-row"> - <label>Genome:</label> - <div class="form-row-input"> - <select name="dbkey" last_selected_value="?"> - %for dbkey in dbkeys: - %if dbkey[0] == last_used_build: - <option value="${dbkey[0]}" selected>${dbkey[1]}</option> - %else: - <option value="${dbkey[0]}">${dbkey[1]}</option> - %endif - %endfor - </select> - </div> - <div class="toolParamHelp" style="clear: both;"> - If you can't find the build you want in this list, open a terminal and execute - <pre>sh cron/updateucsc.sh</pre> - in your galaxy root directory. - </div> - </div> - </div> - <div class="form-row"> - <input type="submit" class="primary-button" name="runtool_btn" value="Download and index"/> - </div> - <script type="text/javascript"> - $(document).ready(function() { - checkDataSource(); - }); - $('#datasource').change(function() { - checkDataSource(); - }); - function checkDataSource() { - var ds = $('#datasource').val(); - $('.params-block').each(function() { - $(this).hide(); - }); - $('#params_' + ds).show(); - }; - - var ac = $('#ncbi_name').autocomplete( $('#ncbi_name'), { minChars: 3, max: 100, url: '${h.url_for( controller='data_admin', action='genome_search' )}' } ); - </script> - </form> -</div> diff -r 4eeb41f2fa7569e1d8652d9a0be3a1f1d5c6c2c0 -r b6fae2cd673c6a286138b9e4628f7c4ef0abb099 templates/admin/data_admin/download_status.mako --- a/templates/admin/data_admin/download_status.mako +++ /dev/null @@ -1,110 +0,0 @@ -<%namespace file="/library/common/library_item_info.mako" import="render_library_item_info" /> -<%namespace file="/library/common/common.mako" import="render_actions_on_multiple_items" /> -<%namespace file="/library/common/common.mako" import="render_compression_types_help" /> -<%namespace file="/library/common/common.mako" import="common_javascripts" /> - -<%! - def inherit(context): - if context.get('use_panels'): - return '/webapps/galaxy/base_panels.mako' - else: - return '/base.mako' -%> -<%inherit file="${inherit(context)}"/> - -<%def name="init()"> -<% - self.has_left_panel=False - self.has_right_panel=False - self.message_box_visible=False - self.active_view="user" - self.overlay_visible=False - self.has_accessible_datasets = False -%> -</%def> - -## -## Override methods from base.mako and base_panels.mako -## -<%def name="center_panel()"> - <div style="overflow: auto; height: 100%;"> - <div class="page-container" style="padding: 10px;"> - ${render_content()} - </div> - </div> -</%def> -<p>${name} been added to the job queue - %if indexers: - to be indexed with ${indexers} - %endif - </p> -<table id="jobStatus"> -</table> -<p><a href="${h.url_for( controller='data_admin', action='manage_data' )}">Overview</a>.</p> -<p><a href="${h.url_for( controller='data_admin', action='add_genome' )}">Download form</a>.</p> -<script type="text/javascript"> - jobs = ${jsonjobs} - finalstates = new Array('done', 'error', 'ok'); - - function makeHTML(jobrow) { - jc = 'jobrow ' + jobrow['style']; - djid = jobrow['jobid']; - jt = jobrow['type']; - idval = jt + '-job-' + djid; - return '<tr id="' + idval + '" class="' + jc + '" data-status="' + jobrow['status'] + '" data-jobid="' + djid + '" data-jobtype="' + jt + '">' + - '<td style="padding: 0px 5px 0px 30px;">' + jobrow['label'] + '</td>' + - '<td style="padding: 0px 5px;">' + jobrow['status'] + '</td></tr>'; - } - - function checkJobs() { - var alldone = true; - var mainjob; - $('.jobrow').each(function() { - status = $(this).attr('data-status'); - if ($(this).attr('data-jobtype') == 'deferred') { - mainjob = $(this).attr('data-jobid'); - } - if ($.inArray(status, finalstates) == -1) { - alldone = false; - } - }); - if (!alldone) { - checkForNewJobs(mainjob); - $('#jobStatus').delay(3000).queue(function(n) { - checkJobs(); - n(); - }); - } - } - - function checkForNewJobs(mainjob) { - $.get('${h.url_for( controller='data_admin', action='get_jobs' )}', { jobid: mainjob }, function(data) { - jsondata = JSON.parse(data); - for (i in jsondata) { - currentjob = jsondata[i] - if (jobs[i] == undefined) { - $('#jobStatus').append(makeHTML(jsondata[i])); - jobs.push(jsondata[i]); - } - $('#' + currentjob['type'] + '-job-' + currentjob['jobid']).replaceWith(makeHTML(currentjob)); - } - }); - } - - $(document).ready(function() { - for (job in jobs) { - jobrow = jobs[job]; - $('#jobStatus').append(makeHTML(jobrow)); - if (jobrow['type'] == 'deferred') { - $('#jobStatus').delay(5000).queue(function(n) { - checkForNewJobs(jobrow['jobid']); - n(); - }).fadeIn(); - } - } - $('#jobStatus').delay(3000).queue(function(n) { - checkJobs(); - n(); - }).fadeIn(); - }); -</script> \ No newline at end of file diff -r 4eeb41f2fa7569e1d8652d9a0be3a1f1d5c6c2c0 -r b6fae2cd673c6a286138b9e4628f7c4ef0abb099 templates/admin/data_admin/generic_error.mako --- a/templates/admin/data_admin/generic_error.mako +++ /dev/null @@ -1,35 +0,0 @@ -<%inherit file="/base.mako"/> -<%namespace file="/message.mako" import="render_msg" /> -<%namespace file="/library/common/common.mako" import="common_javascripts" /> - -<%! - def inherit(context): - if context.get('use_panels'): - return '/webapps/galaxy/base_panels.mako' - else: - return '/base.mako' -%> -<%inherit file="${inherit(context)}"/> - -<%def name="init()"> -<% - self.has_left_panel=False - self.has_right_panel=False - self.message_box_visible=False - self.active_view="user" - self.overlay_visible=False - self.has_accessible_datasets = False -%> -</%def> -<%def name="stylesheets()"> - ${parent.stylesheets()} - ${h.css( "autocomplete_tagging" )} -</%def> -<%def name="javascripts()"> - ${parent.javascripts()} - ${h.js("libs/jquery/jquery.autocomplete", "galaxy.autocom_tagging" )} -</%def> -## -## Override methods from base.mako and base_panels.mako -## -<p class="panel-error-message">${message}</p> \ No newline at end of file diff -r 4eeb41f2fa7569e1d8652d9a0be3a1f1d5c6c2c0 -r b6fae2cd673c6a286138b9e4628f7c4ef0abb099 templates/admin/data_admin/local_data.mako --- a/templates/admin/data_admin/local_data.mako +++ /dev/null @@ -1,169 +0,0 @@ -<%inherit file="/base.mako"/> -<%namespace file="/message.mako" import="render_msg" /> -<%namespace file="/library/common/common.mako" import="common_javascripts" /> - -<%! - def inherit(context): - if context.get('use_panels'): - return '/webapps/galaxy/base_panels.mako' - else: - return '/base.mako' -%> -<%inherit file="${inherit(context)}"/> - -<%def name="init()"> -<% - self.has_left_panel=False - self.has_right_panel=False - self.message_box_visible=False - self.active_view="user" - self.overlay_visible=False - self.has_accessible_datasets = False -%> -</%def> -<%def name="stylesheets()"> - ${parent.stylesheets()} - ${h.css( "autocomplete_tagging" )} -</%def> -<%def name="javascripts()"> - ${parent.javascripts()} - ${h.js("libs/jquery/jquery.autocomplete", "galaxy.autocom_tagging" )} -</%def> -## -## Override methods from base.mako and base_panels.mako -## -<%def name="center_panel()"> - <div style="overflow: auto; height: 100%;"> - <div class="page-container" style="padding: 10px;"> - ${render_content()} - </div> - </div> -</%def> -<style type="text/css"> - .params-block { display: none; } - td, th { padding-left: 10px; padding-right: 10px; } - td.state-color-new { text-decoration: underline; } - td.panel-done-message { background-image: none; padding: 0px 10px 0px 10px; } - td.panel-error-message { background-image: none; padding: 0px 10px 0px 10px; } -</style> -<div class="toolForm"> - %if message: - <div class="${status}">${message}</div> - %endif - <div class="toolFormTitle">Currently tracked builds <a class="action-button" href="${h.url_for( controller='data_admin', action='add_genome' )}">Add new</a></div> - <div class="toolFormBody"> - <h2>Locally cached data:</h2> - <h3>NOTE: Indexes generated here will not be reflected in the table until Galaxy is restarted.</h3> - <table id="locfiles"> - <tr> - <th>DB Key</th> - <th>Name</th> - %for label in labels: - <th>${labels[label]}</th> - %endfor - </tr> - %for dbkey in sorted(dbkeys): - <tr> - <td>${dbkey}</td> - <td>${indextable[dbkey]['name']}</td> - %for label in labels: - <td id="${dbkey}-${indexfuncs[label]}" class="indexcell ${styles[indextable[dbkey]['indexes'][label]]}" data-fapath="${indextable[dbkey]['path']}" data-longname="${indextable[dbkey]['name']}" data-index="${indexfuncs[label]}" data-dbkey="${dbkey}">${indextable[dbkey]['indexes'][label]}</td> - %endfor - - </tr> - %endfor - </table> - <h2>Recent jobs:</h2> - <p>Click the job ID to see job details and the status of any individual sub-jobs. Note that this list only shows jobs initiated by your account.</p> - <div id="recentJobs"> - %for job in jobgrid: - <div id="job-${job['deferred']}" data-dbkey="${job['dbkey']}" data-name="${job['intname']}" data-indexes="${job['indexers']}" data-jobid="${job['deferred']}" data-state="${job['state']}" class="historyItem-${job['state']} historyItemWrapper historyItem"> - <p>Job ID <a href="${h.url_for( controller='data_admin', action='monitor_status', job=job['deferred'] )}">${job['deferred']}</a>: - %if job['jobtype'] == 'download': - Download <em>${job['intname']}</em> - %if job['indexers']: - and index with ${job['indexers']} - %endif - %else: - Index <em>${job['intname']}</em> with ${job['indexers']} - %endif - </p> - </div> - %endfor - </div> -</div> -<script type="text/javascript"> - finalstates = new Array('done', 'error', 'ok'); - $('.indexcell').click(function() { - status = $(this).html(); - elem = $(this); - if (status != 'Generate') { - return; - } - longname = $(this).attr('data-longname'); - dbkey = $(this).attr('data-dbkey'); - indexes = $(this).attr('data-index'); - path = $(this).attr('data-fapath'); - $.post('${h.url_for( controller='data_admin', action='index_build' )}', { longname: longname, dbkey: dbkey, indexes: indexes, path: path }, function(data) { - if (data == 'ERROR') { - alert('There was an error.'); - } - else { - elem.html('Generating'); - elem.attr('class', 'indexcell state-color-running'); - } - newhtml = '<div data-dbkey="' + dbkey + '" data-name="' + longname + '" data-indexes="' + indexes + '" id="job-' + data + '" class="historyItem-new historyItemWrapper historyItem">' + - '<p>Job ID <a href="${h.url_for( controller='data_admin', action='monitor_status')}?job=' + data + '">' + data + '</a>: ' + - 'Index <em>' + longname + '</em> with ' + indexes + '</p></div>'; - $('#recentJobs').prepend(newhtml); - $('#job-' + data).delay(3000).queue(function(n) { - checkJob(data); - n(); - }); - }); - }); - - function checkJob(jobid) { - $.get('${h.url_for( controller='data_admin', action='get_jobs' )}', { jobid: jobid }, function(data) { - jsondata = JSON.parse(data)[0]; - jsondata["name"] = $('#job-' + jobid).attr('data-name'); - jsondata["dbkey"] = $('#job-' + jobid).attr('data-dbkey'); - jsondata["indexes"] = $('#job-' + jobid).attr('data-indexes'); - tdid = jq(jsondata["dbkey"] + '-' + jsondata["indexes"]); - newhtml = makeNewHTML(jsondata); - $('#job-' + jobid).replaceWith(newhtml); - if ($.inArray(jsondata["status"], finalstates) == -1) { - $('#job-' + jobid).delay(3000).queue(function(n) { - checkJob(jobid); - n(); - }); - } - if (jsondata["status"] == 'done' || jsondata["status"] == 'ok') { - elem = $(tdid); - elem.html('Generated'); - elem.attr('class', 'indexcell panel-done-message'); - } - }); - } - - function makeNewHTML(jsondata) { - newhtml = '<div data-dbkey="' + jsondata["dbkey"] + '" data-name="' + jsondata["name"] + '" data-indexes="' + jsondata["indexes"] + '" id="job-' + jsondata["jobid"] + '" class="historyItem-' + jsondata["status"] + ' historyItemWrapper historyItem">' + - '<p>Job ID <a href="${h.url_for( controller='data_admin', action='monitor_status')}?job=' + jsondata["jobid"] + '">' + jsondata["jobid"] + '</a>: ' + - 'Index <em>' + jsondata["name"] + '</em> with ' + jsondata["indexes"] + '</p></div>'; - return newhtml; - } - - $(document).ready(function() { - $('.historyItem').each(function() { - state = $(this).attr('data-state'); - jobid = $(this).attr('data-jobid'); - if ($.inArray(state, finalstates) == -1) { - checkJob(jobid); - } - }); - }); - - function jq(id) { - return '#' + id.replace(/(:|\.)/g,'\\$1'); - } -</script> \ No newline at end of file https://bitbucket.org/galaxy/galaxy-central/commits/d9ab8058b274/ Changeset: d9ab8058b274 Branch: stable User: davebgx Date: 2014-07-08 19:21:48 Summary: Merge stables. Affected #: 4 files diff -r c4519bae84d32d638d8fe168c6fbcab40f4e8449 -r d9ab8058b2748de6b4b1a17646c2d7456d709877 lib/galaxy/jobs/__init__.py --- a/lib/galaxy/jobs/__init__.py +++ b/lib/galaxy/jobs/__init__.py @@ -1065,6 +1065,13 @@ out_data = dict( [ ( da.name, da.dataset ) for da in job.output_datasets ] ) inp_data.update( [ ( da.name, da.dataset ) for da in job.input_library_datasets ] ) out_data.update( [ ( da.name, da.dataset ) for da in job.output_library_datasets ] ) + input_ext = 'data' + for _, data in inp_data.items(): + # For loop odd, but sort simulating behavior in galaxy.tools.actions + if not data: + continue + input_ext = data.ext + param_dict = dict( [ ( p.name, p.value ) for p in job.parameters ] ) # why not re-use self.param_dict here? ##dunno...probably should, this causes tools.parameters.basic.UnvalidatedValue to be used in following methods instead of validated and transformed values during i.e. running workflows param_dict = self.tool.params_from_strings( param_dict, self.app ) # Check for and move associated_files @@ -1075,7 +1082,7 @@ # Create generated output children and primary datasets and add to param_dict collected_datasets = { 'children': self.tool.collect_child_datasets(out_data, self.working_directory), - 'primary': self.tool.collect_primary_datasets(out_data, self.working_directory) + 'primary': self.tool.collect_primary_datasets(out_data, self.working_directory, input_ext) } param_dict.update({'__collected_datasets__': collected_datasets}) # Certain tools require tasks to be completed after job execution diff -r c4519bae84d32d638d8fe168c6fbcab40f4e8449 -r d9ab8058b2748de6b4b1a17646c2d7456d709877 lib/galaxy/tools/__init__.py --- a/lib/galaxy/tools/__init__.py +++ b/lib/galaxy/tools/__init__.py @@ -2869,12 +2869,12 @@ self.sa_session.flush() return children - def collect_primary_datasets( self, output, job_working_directory ): + def collect_primary_datasets( self, output, job_working_directory, input_ext ): """ Find any additional datasets generated by a tool and attach (for cases where number of outputs is not known in advance). """ - return output_collect.collect_primary_datatasets( self, output, job_working_directory ) + return output_collect.collect_primary_datasets( self, output, job_working_directory, input_ext ) def to_dict( self, trans, link_details=False, io_details=False ): """ Returns dict of tool. """ diff -r c4519bae84d32d638d8fe168c6fbcab40f4e8449 -r d9ab8058b2748de6b4b1a17646c2d7456d709877 lib/galaxy/tools/parameters/output_collect.py --- a/lib/galaxy/tools/parameters/output_collect.py +++ b/lib/galaxy/tools/parameters/output_collect.py @@ -14,7 +14,7 @@ DEFAULT_EXTRA_FILENAME_PATTERN = r"primary_DATASET_ID_(?P<designation>[^_]+)_(?P<visible>[^_]+)_(?P<ext>[^_]+)(_(?P<dbkey>[^_]+))?" -def collect_primary_datatasets( tool, output, job_working_directory ): +def collect_primary_datasets( tool, output, job_working_directory, input_ext ): app = tool.app sa_session = tool.sa_session new_primary_datasets = {} @@ -66,6 +66,8 @@ designation = fields_match.designation visible = fields_match.visible ext = fields_match.ext + if ext == "input": + ext = input_ext dbkey = fields_match.dbkey # Create new primary dataset primary_data = app.model.HistoryDatasetAssociation( extension=ext, diff -r c4519bae84d32d638d8fe168c6fbcab40f4e8449 -r d9ab8058b2748de6b4b1a17646c2d7456d709877 test/functional/tools/multi_output_configured.xml --- a/test/functional/tools/multi_output_configured.xml +++ b/test/functional/tools/multi_output_configured.xml @@ -8,19 +8,24 @@ echo "1" > subdir2/CUSTOM_1.txt; echo "2" > subdir2/CUSTOM_2.tabular; echo "3" > subdir2/CUSTOM_3.txt; + mkdir subdir3; + echo "Foo" > subdir3/Foo; </command><inputs> - <param name="input" type="integer" value="7" /> + <param name="num_param" type="integer" value="7" /> + <param name="input" type="data" /></inputs><outputs><data format="txt" name="report"><discover_datasets pattern="__designation_and_ext__" directory="subdir1" /><discover_datasets pattern="CUSTOM_(?P<designation>.+)\.(?P<ext>.+)" directory="subdir2" /> + <discover_datasets pattern="__designation__" directory="subdir3" ext="input" /></data></outputs><tests><test> - <param name="input" value="7" /> + <param name="num_param" value="7" /> + <param name="input" ftype="txt" value="simple_line.txt"/><output name="report"><assert_contents><has_line line="Hello" /> @@ -37,6 +42,9 @@ <discovered_dataset designation="2" ftype="tabular"><assert_contents><has_line line="2" /></assert_contents></discovered_dataset> + <discovered_dataset designation="Foo" ftype="txt"> + <assert_contents><has_line line="Foo" /></assert_contents> + </discovered_dataset></output></test></tests> https://bitbucket.org/galaxy/galaxy-central/commits/5874f6bc02f9/ Changeset: 5874f6bc02f9 User: davebgx Date: 2014-07-08 19:22:06 Summary: Merge stable. Affected #: 2 files https://bitbucket.org/galaxy/galaxy-central/commits/6b1415cbc145/ Changeset: 6b1415cbc145 Branch: daniel_blanchard/update-drmaa-python-version-in-eggsini-t-1402925026891 User: davebgx Date: 2014-07-08 19:22:35 Summary: Close branch daniel_blanchard/update-drmaa-python-version-in-eggsini-t-1402925026891 Affected #: 0 files Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: dan: Expose OutputParameterJSONTool (tool_type='output_parameter_json') for direct usage. Previously used as inherited by data source and data manager tools.
by commits-noreply＠bitbucket.org 07 Jul '14

07 Jul '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/4eeb41f2fa75/ Changeset: 4eeb41f2fa75 User: dan Date: 2014-07-07 17:11:30 Summary: Expose OutputParameterJSONTool (tool_type='output_parameter_json') for direct usage. Previously used as inherited by data source and data manager tools. Affected #: 1 file diff -r 86308479b291e2295ecba6bcb3c33864abc18860 -r 4eeb41f2fa7569e1d8652d9a0be3a1f1d5c6c2c0 lib/galaxy/tools/__init__.py --- a/lib/galaxy/tools/__init__.py +++ b/lib/galaxy/tools/__init__.py @@ -3213,7 +3213,9 @@ # Populate tool_type to ToolClass mappings tool_types = {} -for tool_class in [ Tool, DataDestinationTool, SetMetadataTool, DataSourceTool, AsyncDataSourceTool, DataManagerTool ]: +for tool_class in [ Tool, SetMetadataTool, OutputParameterJSONTool, + DataManagerTool, DataSourceTool, AsyncDataSourceTool, + DataDestinationTool ]: tool_types[ tool_class.tool_type ] = tool_class Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0

commit/galaxy-central: 2 new changesets
by commits-noreply＠bitbucket.org 07 Jul '14

07 Jul '14

2 new commits in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/0b0f39f80bb9/ Changeset: 0b0f39f80bb9 User: jmchilton Date: 2014-07-07 04:08:38 Summary: Docker - Extract helper method defining prefix for Docker CLIs. Affected #: 2 files diff -r 6d50cd22380253129121e3e7359a6286ee9b1663 -r 0b0f39f80bb916669ddaac3f17054b26525f906f lib/galaxy/tools/deps/containers.py --- a/lib/galaxy/tools/deps/containers.py +++ b/lib/galaxy/tools/deps/containers.py @@ -191,6 +191,14 @@ # TODO: Remove redundant volumes... volumes = docker_util.DockerVolume.volumes_from_str(volumes_raw) volumes_from = self.destination_info.get("docker_volumes_from", docker_util.DEFAULT_VOLUMES_FROM) + + docker_host_props = dict( + docker_cmd=prop("cmd", docker_util.DEFAULT_DOCKER_COMMAND), + sudo=asbool(prop("sudo", docker_util.DEFAULT_SUDO)), + sudo_cmd=prop("sudo_cmd", docker_util.DEFAULT_SUDO_COMMAND), + host=prop("host", docker_util.DEFAULT_HOST), + ) + return docker_util.build_docker_run_command( command, self.container_id, @@ -198,11 +206,8 @@ volumes_from=volumes_from, env_directives=env_directives, working_directory=working_directory, - docker_cmd=prop("cmd", docker_util.DEFAULT_DOCKER_COMMAND), - sudo=asbool(prop("sudo", docker_util.DEFAULT_SUDO)), - sudo_cmd=prop("sudo_cmd", docker_util.DEFAULT_SUDO_COMMAND), - host=prop("host", docker_util.DEFAULT_HOST), - net=prop("net", "none") # By default, docker instance has networking disabled + net=prop("net", "none"), # By default, docker instance has networking disabled + **docker_host_props ) def __expand_str(self, value): diff -r 6d50cd22380253129121e3e7359a6286ee9b1663 -r 0b0f39f80bb916669ddaac3f17054b26525f906f lib/galaxy/tools/deps/docker_util.py --- a/lib/galaxy/tools/deps/docker_util.py +++ b/lib/galaxy/tools/deps/docker_util.py @@ -54,24 +54,19 @@ container_command, image, tag=None, - docker_cmd=DEFAULT_DOCKER_COMMAND, volumes=[], volumes_from=DEFAULT_VOLUMES_FROM, memory=DEFAULT_MEMORY, env_directives=[], working_directory=DEFAULT_WORKING_DIRECTORY, + name=None, + net=DEFAULT_NET, + docker_cmd=DEFAULT_DOCKER_COMMAND, sudo=DEFAULT_SUDO, sudo_cmd=DEFAULT_SUDO_COMMAND, - name=None, host=DEFAULT_HOST, - net=DEFAULT_NET, ): - command_parts = [] - if sudo: - command_parts.append(sudo_cmd) - command_parts.append(docker_cmd) - if host: - command_parts.append(["-H", host]) + command_parts = __docker_prefix(docker_cmd, sudo, sudo_cmd, host) command_parts.append("run") for env_directive in env_directives: command_parts.extend(["-e", env_directive]) @@ -93,3 +88,15 @@ command_parts.append(full_image) command_parts.append(container_command) return " ".join(command_parts) + + +def __docker_prefix(docker_cmd, sudo, sudo_cmd, host): + """ Prefix to issue a docker command. + """ + command_parts = [] + if sudo: + command_parts.append(sudo_cmd) + command_parts.append(docker_cmd) + if host: + command_parts.append(["-H", host]) + return command_parts https://bitbucket.org/galaxy/galaxy-central/commits/86308479b291/ Changeset: 86308479b291 User: jmchilton Date: 2014-07-07 04:08:38 Summary: Docker - Explicitly search for and pull images (if needed) prior to use. This should prevent successful docker caching from interfering with job outputs on the first execution of tools. This needs to be better - the output should be capture in such a way that admins can view but it doesn't interfer with the job and there should be an option to cache on the handler (instead of in the job) - for shared clusters without web access. Affected #: 2 files diff -r 0b0f39f80bb916669ddaac3f17054b26525f906f -r 86308479b291e2295ecba6bcb3c33864abc18860 lib/galaxy/tools/deps/containers.py --- a/lib/galaxy/tools/deps/containers.py +++ b/lib/galaxy/tools/deps/containers.py @@ -199,7 +199,8 @@ host=prop("host", docker_util.DEFAULT_HOST), ) - return docker_util.build_docker_run_command( + cache_command = docker_util.build_docker_cache_command(self.container_id, **docker_host_props) + run_command = docker_util.build_docker_run_command( command, self.container_id, volumes=volumes, @@ -209,6 +210,7 @@ net=prop("net", "none"), # By default, docker instance has networking disabled **docker_host_props ) + return "%s\n%s" % (cache_command, run_command) def __expand_str(self, value): if not value: diff -r 0b0f39f80bb916669ddaac3f17054b26525f906f -r 86308479b291e2295ecba6bcb3c33864abc18860 lib/galaxy/tools/deps/docker_util.py --- a/lib/galaxy/tools/deps/docker_util.py +++ b/lib/galaxy/tools/deps/docker_util.py @@ -50,6 +50,24 @@ return ":".join([self.from_path, self.to_path, self.how]) +def build_docker_cache_command( + image, + docker_cmd=DEFAULT_DOCKER_COMMAND, + sudo=DEFAULT_SUDO, + sudo_cmd=DEFAULT_SUDO_COMMAND, + host=DEFAULT_HOST, +): + inspect_command_parts = __docker_prefix(docker_cmd, sudo, sudo_cmd, host) + inspect_command_parts.extend(["inspect", image]) + inspect_image_command = " ".join(inspect_command_parts) + + pull_command_parts = __docker_prefix(docker_cmd, sudo, sudo_cmd, host) + pull_command_parts.extend(["pull", image]) + pull_image_command = " ".join(pull_command_parts) + cache_command = "%s > /dev/null 2>&1\n[ $? -ne 0 ] && %s > /dev/null 2>&1\n" % (inspect_image_command, pull_image_command) + return cache_command + + def build_docker_run_command( container_command, image, Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.

1 0