mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-10-24 12:13:33 +08:00
Merged in jmchilton/galaxy-central-fork-1 (pull request #348)
Enhance workflows API to allow extracting workflow from history.
This commit is contained in:
John Chilton
@@ -41,6 +41,7 @@ from galaxy.tools.parameters import RuntimeValue, visit_input_values
|
||||
from galaxy.tools.parameters.basic import DataToolParameter
|
||||
from galaxy.util.json import to_json_string
|
||||
from galaxy.workflow.modules import ToolModule
|
||||
from galaxy.workflow.build_util import attach_ordered_steps
|
||||
|
||||
|
||||
log = logging.getLogger( __name__ )
|
||||
@@ -1676,7 +1677,6 @@ class UsesStoredWorkflowMixin( SharableItemSecurityMixin, UsesAnnotations ):
|
||||
"""
|
||||
Creates a workflow from a dict. Created workflow is stored in the database and returned.
|
||||
"""
|
||||
from galaxy.webapps.galaxy.controllers.workflow import attach_ordered_steps
|
||||
|
||||
# Put parameters in workflow mode
|
||||
trans.workflow_building_mode = True
|
||||
|
||||
@@ -11,8 +11,10 @@ from galaxy import util
|
||||
from galaxy import web
|
||||
from galaxy.web import _future_expose_api as expose_api
|
||||
from galaxy.web.base.controller import BaseAPIController, url_for, UsesStoredWorkflowMixin
|
||||
from galaxy.web.base.controller import UsesHistoryMixin
|
||||
from galaxy.workflow.modules import module_factory
|
||||
from galaxy.workflow.run import invoke
|
||||
from galaxy.workflow.extract import extract_workflow
|
||||
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
@@ -51,7 +53,7 @@ def _update_step_parameters(step, param_map):
|
||||
step.state.inputs.update(param_dict)
|
||||
|
||||
|
||||
class WorkflowsAPIController(BaseAPIController, UsesStoredWorkflowMixin):
|
||||
class WorkflowsAPIController(BaseAPIController, UsesStoredWorkflowMixin, UsesHistoryMixin):
|
||||
|
||||
@web.expose_api
|
||||
def index(self, trans, **kwd):
|
||||
@@ -141,9 +143,11 @@ class WorkflowsAPIController(BaseAPIController, UsesStoredWorkflowMixin):
|
||||
"""
|
||||
POST /api/workflows
|
||||
|
||||
We're not creating workflows from the api. Just execute for now.
|
||||
Run or create workflows from the api.
|
||||
|
||||
However, we will import them if installed_repository_file is specified.
|
||||
If installed_repository_file or from_history_id is specified a new
|
||||
workflow will be created for this user. Otherwise, workflow_id must be
|
||||
specified and this API method will cause a workflow to execute.
|
||||
|
||||
:param installed_repository_file The path of a workflow to import. Either workflow_id or installed_repository_file must be specified
|
||||
:type installed_repository_file str
|
||||
@@ -165,6 +169,15 @@ class WorkflowsAPIController(BaseAPIController, UsesStoredWorkflowMixin):
|
||||
|
||||
:param replacement_params: A dictionary used when renaming datasets
|
||||
:type replacement_params: dict
|
||||
|
||||
:param from_history_id: Id of history to extract a workflow from. Should not be used with worfklow_id or installed_repository_file.
|
||||
:type from_history_id: str
|
||||
|
||||
:param job_ids: If from_history_id is set - this should be a list of jobs to include when extracting workflow from history.
|
||||
:type job_ids: str
|
||||
|
||||
:param dataset_ids: If from_history_id is set - this should be a list of HDA ids corresponding to workflow inputs when extracting workflow from history.
|
||||
:type dataset_ids: str
|
||||
"""
|
||||
|
||||
# Pull parameters out of payload.
|
||||
@@ -183,6 +196,24 @@ class WorkflowsAPIController(BaseAPIController, UsesStoredWorkflowMixin):
|
||||
cntrller='api',
|
||||
**payload)
|
||||
return result
|
||||
if 'from_history_id' in payload:
|
||||
from_history_id = payload.get( 'from_history_id' )
|
||||
history = self.get_history( trans, from_history_id, check_ownership=False, check_accessible=True )
|
||||
job_ids = map( trans.security.decode_id, payload.get( "job_ids", [] ) )
|
||||
dataset_ids = map( trans.security.decode_id, payload.get( "dataset_ids", [] ) )
|
||||
workflow_name = payload[ "workflow_name" ]
|
||||
stored_workflow = extract_workflow(
|
||||
trans=trans,
|
||||
user=trans.get_user(),
|
||||
history=history,
|
||||
job_ids=job_ids,
|
||||
dataset_ids=dataset_ids,
|
||||
workflow_name=workflow_name,
|
||||
)
|
||||
item = stored_workflow.to_dict( value_mapper={ "id": trans.security.encode_id } )
|
||||
item[ 'url' ] = url_for( 'workflow', id=item[ "id" ] )
|
||||
return item
|
||||
|
||||
trans.response.status = 403
|
||||
return "Either workflow_id or installed_repository_file must be specified"
|
||||
if 'installed_repository_file' in payload:
|
||||
|
||||
@@ -4,7 +4,6 @@ pkg_resources.require( "SVGFig" )
|
||||
import base64
|
||||
import httplib
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import sgmllib
|
||||
import svgfig
|
||||
@@ -20,12 +19,9 @@ from galaxy import web
|
||||
from galaxy.datatypes.data import Data
|
||||
from galaxy.model.item_attrs import UsesItemRatings
|
||||
from galaxy.model.mapping import desc
|
||||
from galaxy.tools.parameters.basic import DataToolParameter
|
||||
from galaxy.tools.parameters import visit_input_values
|
||||
from galaxy.tools.parameters.basic import DataToolParameter, DrillDownSelectToolParameter, SelectToolParameter, UnvalidatedValue
|
||||
from galaxy.tools.parameters.grouping import Conditional, Repeat
|
||||
from galaxy.util.odict import odict
|
||||
from galaxy.util.sanitize_html import sanitize_html
|
||||
from galaxy.util.topsort import CycleError, topsort, topsort_levels
|
||||
from galaxy.web import error, url_for
|
||||
from galaxy.web.base.controller import BaseUIController, SharableMixin, UsesStoredWorkflowMixin
|
||||
from galaxy.web.framework import form
|
||||
@@ -33,6 +29,14 @@ from galaxy.web.framework.helpers import grids, time_ago
|
||||
from galaxy.web.framework.helpers import to_unicode
|
||||
from galaxy.workflow.modules import module_factory
|
||||
from galaxy.workflow.run import invoke
|
||||
from galaxy.workflow.extract import summarize
|
||||
from galaxy.workflow.extract import extract_workflow
|
||||
from galaxy.workflow.steps import (
|
||||
attach_ordered_steps,
|
||||
order_workflow_steps,
|
||||
edgelist_for_workflow_steps,
|
||||
order_workflow_steps_with_levels,
|
||||
)
|
||||
|
||||
|
||||
class StoredWorkflowListGrid( grids.Grid ):
|
||||
@@ -1188,7 +1192,7 @@ class WorkflowController( BaseUIController, SharableMixin, UsesStoredWorkflowMix
|
||||
if not user:
|
||||
return trans.show_error_message( "Must be logged in to create workflows" )
|
||||
if ( job_ids is None and dataset_ids is None ) or workflow_name is None:
|
||||
jobs, warnings = get_job_dict( trans )
|
||||
jobs, warnings = summarize( trans )
|
||||
# Render
|
||||
return trans.fill_template(
|
||||
"workflow/build_from_current_history.mako",
|
||||
@@ -1197,82 +1201,13 @@ class WorkflowController( BaseUIController, SharableMixin, UsesStoredWorkflowMix
|
||||
history=history
|
||||
)
|
||||
else:
|
||||
# Ensure job_ids and dataset_ids are lists (possibly empty)
|
||||
if job_ids is None:
|
||||
job_ids = []
|
||||
elif type( job_ids ) is not list:
|
||||
job_ids = [ job_ids ]
|
||||
if dataset_ids is None:
|
||||
dataset_ids = []
|
||||
elif type( dataset_ids ) is not list:
|
||||
dataset_ids = [ dataset_ids ]
|
||||
# Convert both sets of ids to integers
|
||||
job_ids = [ int( id ) for id in job_ids ]
|
||||
dataset_ids = [ int( id ) for id in dataset_ids ]
|
||||
# Find each job, for security we (implicately) check that they are
|
||||
# associated witha job in the current history.
|
||||
jobs, warnings = get_job_dict( trans )
|
||||
jobs_by_id = dict( ( job.id, job ) for job in jobs.keys() )
|
||||
steps = []
|
||||
steps_by_job_id = {}
|
||||
hid_to_output_pair = {}
|
||||
# Input dataset steps
|
||||
for hid in dataset_ids:
|
||||
step = model.WorkflowStep()
|
||||
step.type = 'data_input'
|
||||
step.tool_inputs = dict( name="Input Dataset" )
|
||||
hid_to_output_pair[ hid ] = ( step, 'output' )
|
||||
steps.append( step )
|
||||
# Tool steps
|
||||
for job_id in job_ids:
|
||||
assert job_id in jobs_by_id, "Attempt to create workflow with job not connected to current history"
|
||||
job = jobs_by_id[ job_id ]
|
||||
tool = trans.app.toolbox.get_tool( job.tool_id )
|
||||
param_values = job.get_param_values( trans.app, ignore_errors=True ) # If a tool was updated and e.g. had a text value changed to an integer, we don't want a traceback here
|
||||
associations = cleanup_param_values( tool.inputs, param_values )
|
||||
step = model.WorkflowStep()
|
||||
step.type = 'tool'
|
||||
step.tool_id = job.tool_id
|
||||
step.tool_inputs = tool.params_to_strings( param_values, trans.app )
|
||||
# NOTE: We shouldn't need to do two passes here since only
|
||||
# an earlier job can be used as an input to a later
|
||||
# job.
|
||||
for other_hid, input_name in associations:
|
||||
if other_hid in hid_to_output_pair:
|
||||
other_step, other_name = hid_to_output_pair[ other_hid ]
|
||||
conn = model.WorkflowStepConnection()
|
||||
conn.input_step = step
|
||||
conn.input_name = input_name
|
||||
# Should always be connected to an earlier step
|
||||
conn.output_step = other_step
|
||||
conn.output_name = other_name
|
||||
steps.append( step )
|
||||
steps_by_job_id[ job_id ] = step
|
||||
# Store created dataset hids
|
||||
for assoc in job.output_datasets:
|
||||
hid_to_output_pair[ assoc.dataset.hid ] = ( step, assoc.name )
|
||||
# Workflow to populate
|
||||
workflow = model.Workflow()
|
||||
workflow.name = workflow_name
|
||||
# Order the steps if possible
|
||||
attach_ordered_steps( workflow, steps )
|
||||
# And let's try to set up some reasonable locations on the canvas
|
||||
# (these are pretty arbitrary values)
|
||||
levorder = order_workflow_steps_with_levels( steps )
|
||||
base_pos = 10
|
||||
for i, steps_at_level in enumerate( levorder ):
|
||||
for j, index in enumerate( steps_at_level ):
|
||||
step = steps[ index ]
|
||||
step.position = dict( top=( base_pos + 120 * j ),
|
||||
left=( base_pos + 220 * i ) )
|
||||
# Store it
|
||||
stored = model.StoredWorkflow()
|
||||
stored.user = user
|
||||
stored.name = workflow_name
|
||||
workflow.stored_workflow = stored
|
||||
stored.latest_workflow = workflow
|
||||
trans.sa_session.add( stored )
|
||||
trans.sa_session.flush()
|
||||
extract_workflow(
|
||||
trans,
|
||||
user=user,
|
||||
job_ids=job_ids,
|
||||
dataset_ids=dataset_ids,
|
||||
workflow_name=workflow_name
|
||||
)
|
||||
# Index page with message
|
||||
return trans.show_message( "Workflow '%s' created from current history." % workflow_name )
|
||||
## return trans.show_ok_message( "<p>Workflow '%s' created.</p><p><a target='_top' href='%s'>Click to load in workflow editor</a></p>"
|
||||
@@ -1692,148 +1627,6 @@ class WorkflowController( BaseUIController, SharableMixin, UsesStoredWorkflowMix
|
||||
|
||||
|
||||
## ---- Utility methods -------------------------------------------------------
|
||||
def attach_ordered_steps( workflow, steps ):
|
||||
ordered_steps = order_workflow_steps( steps )
|
||||
if ordered_steps:
|
||||
workflow.has_cycles = False
|
||||
for i, step in enumerate( ordered_steps ):
|
||||
step.order_index = i
|
||||
workflow.steps.append( step )
|
||||
else:
|
||||
workflow.has_cycles = True
|
||||
workflow.steps = steps
|
||||
|
||||
|
||||
def edgelist_for_workflow_steps( steps ):
|
||||
"""
|
||||
Create a list of tuples representing edges between ``WorkflowSteps`` based
|
||||
on associated ``WorkflowStepConnection``s
|
||||
"""
|
||||
edges = []
|
||||
steps_to_index = dict( ( step, i ) for i, step in enumerate( steps ) )
|
||||
for step in steps:
|
||||
edges.append( ( steps_to_index[step], steps_to_index[step] ) )
|
||||
for conn in step.input_connections:
|
||||
edges.append( ( steps_to_index[conn.output_step], steps_to_index[conn.input_step] ) )
|
||||
return edges
|
||||
|
||||
|
||||
def order_workflow_steps( steps ):
|
||||
"""
|
||||
Perform topological sort of the steps, return ordered or None
|
||||
"""
|
||||
position_data_available = True
|
||||
for step in steps:
|
||||
if not step.position or not 'left' in step.position or not 'top' in step.position:
|
||||
position_data_available = False
|
||||
if position_data_available:
|
||||
steps.sort(cmp=lambda s1, s2: cmp( math.sqrt(s1.position['left'] ** 2 + s1.position['top'] ** 2), math.sqrt(s2.position['left'] ** 2 + s2.position['top'] ** 2)))
|
||||
try:
|
||||
edges = edgelist_for_workflow_steps( steps )
|
||||
node_order = topsort( edges )
|
||||
return [ steps[i] for i in node_order ]
|
||||
except CycleError:
|
||||
return None
|
||||
|
||||
|
||||
def order_workflow_steps_with_levels( steps ):
|
||||
try:
|
||||
return topsort_levels( edgelist_for_workflow_steps( steps ) )
|
||||
except CycleError:
|
||||
return None
|
||||
|
||||
|
||||
class FakeJob( object ):
|
||||
"""
|
||||
Fake job object for datasets that have no creating_job_associations,
|
||||
they will be treated as "input" datasets.
|
||||
"""
|
||||
def __init__( self, dataset ):
|
||||
self.is_fake = True
|
||||
self.id = "fake_%s" % dataset.id
|
||||
|
||||
|
||||
def get_job_dict( trans ):
|
||||
"""
|
||||
Return a dictionary of Job -> [ Dataset ] mappings, for all finished
|
||||
active Datasets in the current history and the jobs that created them.
|
||||
"""
|
||||
history = trans.get_history()
|
||||
# Get the jobs that created the datasets
|
||||
warnings = set()
|
||||
jobs = odict()
|
||||
for dataset in history.active_datasets:
|
||||
# FIXME: Create "Dataset.is_finished"
|
||||
if dataset.state in ( 'new', 'running', 'queued' ):
|
||||
warnings.add( "Some datasets still queued or running were ignored" )
|
||||
continue
|
||||
|
||||
#if this hda was copied from another, we need to find the job that created the origial hda
|
||||
job_hda = dataset
|
||||
while job_hda.copied_from_history_dataset_association:
|
||||
job_hda = job_hda.copied_from_history_dataset_association
|
||||
|
||||
if not job_hda.creating_job_associations:
|
||||
jobs[ FakeJob( dataset ) ] = [ ( None, dataset ) ]
|
||||
|
||||
for assoc in job_hda.creating_job_associations:
|
||||
job = assoc.job
|
||||
if job in jobs:
|
||||
jobs[ job ].append( ( assoc.name, dataset ) )
|
||||
else:
|
||||
jobs[ job ] = [ ( assoc.name, dataset ) ]
|
||||
return jobs, warnings
|
||||
|
||||
|
||||
def cleanup_param_values( inputs, values ):
|
||||
"""
|
||||
Remove 'Data' values from `param_values`, along with metadata cruft,
|
||||
but track the associations.
|
||||
"""
|
||||
associations = []
|
||||
# dbkey is pushed in by the framework
|
||||
if 'dbkey' in values:
|
||||
del values['dbkey']
|
||||
root_values = values
|
||||
|
||||
# Recursively clean data inputs and dynamic selects
|
||||
def cleanup( prefix, inputs, values ):
|
||||
for key, input in inputs.items():
|
||||
if isinstance( input, ( SelectToolParameter, DrillDownSelectToolParameter ) ):
|
||||
if input.is_dynamic and not isinstance( values[key], UnvalidatedValue ):
|
||||
values[key] = UnvalidatedValue( values[key] )
|
||||
if isinstance( input, DataToolParameter ):
|
||||
tmp = values[key]
|
||||
values[key] = None
|
||||
# HACK: Nested associations are not yet working, but we
|
||||
# still need to clean them up so we can serialize
|
||||
# if not( prefix ):
|
||||
if tmp: # this is false for a non-set optional dataset
|
||||
if not isinstance(tmp, list):
|
||||
associations.append( ( tmp.hid, prefix + key ) )
|
||||
else:
|
||||
associations.extend( [ (t.hid, prefix + key) for t in tmp] )
|
||||
|
||||
# Cleanup the other deprecated crap associated with datasets
|
||||
# as well. Worse, for nested datasets all the metadata is
|
||||
# being pushed into the root. FIXME: MUST REMOVE SOON
|
||||
key = prefix + key + "_"
|
||||
for k in root_values.keys():
|
||||
if k.startswith( key ):
|
||||
del root_values[k]
|
||||
elif isinstance( input, Repeat ):
|
||||
group_values = values[key]
|
||||
for i, rep_values in enumerate( group_values ):
|
||||
rep_index = rep_values['__index__']
|
||||
cleanup( "%s%s_%d|" % (prefix, key, rep_index ), input.inputs, group_values[i] )
|
||||
elif isinstance( input, Conditional ):
|
||||
group_values = values[input.name]
|
||||
current_case = group_values['__current_case__']
|
||||
cleanup( "%s%s|" % ( prefix, key ), input.cases[current_case].inputs, group_values )
|
||||
cleanup( "", inputs, values )
|
||||
return associations
|
||||
|
||||
|
||||
def _build_workflow_on_str(instance_ds_names):
|
||||
# Returns suffix for new histories based on multi input iteration
|
||||
num_multi_inputs = len(instance_ds_names)
|
||||
|
||||
@@ -0,0 +1,209 @@
|
||||
""" This module contains functionality to aid in extracting workflows from
|
||||
histories.
|
||||
"""
|
||||
from galaxy.util.odict import odict
|
||||
from galaxy import model
|
||||
from galaxy.tools.parameters.basic import (
|
||||
DataToolParameter,
|
||||
DrillDownSelectToolParameter,
|
||||
SelectToolParameter,
|
||||
UnvalidatedValue
|
||||
)
|
||||
from galaxy.tools.parameters.grouping import (
|
||||
Conditional,
|
||||
Repeat
|
||||
)
|
||||
from .steps import (
|
||||
attach_ordered_steps,
|
||||
order_workflow_steps_with_levels
|
||||
)
|
||||
|
||||
WARNING_SOME_DATASETS_NOT_READY = "Some datasets still queued or running were ignored"
|
||||
|
||||
|
||||
def extract_workflow( trans, user, history=None, job_ids=None, dataset_ids=None, workflow_name=None ):
|
||||
steps = extract_steps( trans, history=history, job_ids=job_ids, dataset_ids=dataset_ids )
|
||||
# Workflow to populate
|
||||
workflow = model.Workflow()
|
||||
workflow.name = workflow_name
|
||||
# Order the steps if possible
|
||||
attach_ordered_steps( workflow, steps )
|
||||
# And let's try to set up some reasonable locations on the canvas
|
||||
# (these are pretty arbitrary values)
|
||||
levorder = order_workflow_steps_with_levels( steps )
|
||||
base_pos = 10
|
||||
for i, steps_at_level in enumerate( levorder ):
|
||||
for j, index in enumerate( steps_at_level ):
|
||||
step = steps[ index ]
|
||||
step.position = dict( top=( base_pos + 120 * j ),
|
||||
left=( base_pos + 220 * i ) )
|
||||
# Store it
|
||||
stored = model.StoredWorkflow()
|
||||
stored.user = user
|
||||
stored.name = workflow_name
|
||||
workflow.stored_workflow = stored
|
||||
stored.latest_workflow = workflow
|
||||
trans.sa_session.add( stored )
|
||||
trans.sa_session.flush()
|
||||
return stored
|
||||
|
||||
|
||||
def extract_steps( trans, history=None, job_ids=None, dataset_ids=None ):
|
||||
# Ensure job_ids and dataset_ids are lists (possibly empty)
|
||||
if job_ids is None:
|
||||
job_ids = []
|
||||
elif type( job_ids ) is not list:
|
||||
job_ids = [ job_ids ]
|
||||
if dataset_ids is None:
|
||||
dataset_ids = []
|
||||
elif type( dataset_ids ) is not list:
|
||||
dataset_ids = [ dataset_ids ]
|
||||
# Convert both sets of ids to integers
|
||||
job_ids = [ int( id ) for id in job_ids ]
|
||||
dataset_ids = [ int( id ) for id in dataset_ids ]
|
||||
# Find each job, for security we (implicately) check that they are
|
||||
# associated witha job in the current history.
|
||||
jobs, warnings = summarize( trans, history=history )
|
||||
jobs_by_id = dict( ( job.id, job ) for job in jobs.keys() )
|
||||
steps = []
|
||||
steps_by_job_id = {}
|
||||
hid_to_output_pair = {}
|
||||
# Input dataset steps
|
||||
for hid in dataset_ids:
|
||||
step = model.WorkflowStep()
|
||||
step.type = 'data_input'
|
||||
step.tool_inputs = dict( name="Input Dataset" )
|
||||
hid_to_output_pair[ hid ] = ( step, 'output' )
|
||||
steps.append( step )
|
||||
# Tool steps
|
||||
for job_id in job_ids:
|
||||
assert job_id in jobs_by_id, "Attempt to create workflow with job not connected to current history"
|
||||
job = jobs_by_id[ job_id ]
|
||||
tool_inputs, associations = step_inputs( trans, job )
|
||||
step = model.WorkflowStep()
|
||||
step.type = 'tool'
|
||||
step.tool_id = job.tool_id
|
||||
step.tool_inputs = tool_inputs
|
||||
# NOTE: We shouldn't need to do two passes here since only
|
||||
# an earlier job can be used as an input to a later
|
||||
# job.
|
||||
for other_hid, input_name in associations:
|
||||
if other_hid in hid_to_output_pair:
|
||||
other_step, other_name = hid_to_output_pair[ other_hid ]
|
||||
conn = model.WorkflowStepConnection()
|
||||
conn.input_step = step
|
||||
conn.input_name = input_name
|
||||
# Should always be connected to an earlier step
|
||||
conn.output_step = other_step
|
||||
conn.output_name = other_name
|
||||
steps.append( step )
|
||||
steps_by_job_id[ job_id ] = step
|
||||
# Store created dataset hids
|
||||
for assoc in job.output_datasets:
|
||||
hid_to_output_pair[ assoc.dataset.hid ] = ( step, assoc.name )
|
||||
return steps
|
||||
|
||||
|
||||
class FakeJob( object ):
|
||||
"""
|
||||
Fake job object for datasets that have no creating_job_associations,
|
||||
they will be treated as "input" datasets.
|
||||
"""
|
||||
def __init__( self, dataset ):
|
||||
self.is_fake = True
|
||||
self.id = "fake_%s" % dataset.id
|
||||
|
||||
|
||||
def summarize( trans, history=None ):
|
||||
""" Return mapping of job description to datasets for active items in
|
||||
supplied history - needed for building workflow from a history.
|
||||
|
||||
Formerly call get_job_dict in workflow web controller.
|
||||
"""
|
||||
if not history:
|
||||
history = trans.get_history()
|
||||
|
||||
# Get the jobs that created the datasets
|
||||
warnings = set()
|
||||
jobs = odict()
|
||||
for dataset in history.active_datasets:
|
||||
# FIXME: Create "Dataset.is_finished"
|
||||
if dataset.state in ( 'new', 'running', 'queued' ):
|
||||
warnings.add( WARNING_SOME_DATASETS_NOT_READY )
|
||||
continue
|
||||
|
||||
#if this hda was copied from another, we need to find the job that created the origial hda
|
||||
job_hda = dataset
|
||||
while job_hda.copied_from_history_dataset_association:
|
||||
job_hda = job_hda.copied_from_history_dataset_association
|
||||
|
||||
if not job_hda.creating_job_associations:
|
||||
jobs[ FakeJob( dataset ) ] = [ ( None, dataset ) ]
|
||||
|
||||
for assoc in job_hda.creating_job_associations:
|
||||
job = assoc.job
|
||||
if job in jobs:
|
||||
jobs[ job ].append( ( assoc.name, dataset ) )
|
||||
else:
|
||||
jobs[ job ] = [ ( assoc.name, dataset ) ]
|
||||
|
||||
return jobs, warnings
|
||||
|
||||
|
||||
def step_inputs( trans, job ):
|
||||
tool = trans.app.toolbox.get_tool( job.tool_id )
|
||||
param_values = job.get_param_values( trans.app, ignore_errors=True ) # If a tool was updated and e.g. had a text value changed to an integer, we don't want a traceback here
|
||||
associations = __cleanup_param_values( tool.inputs, param_values )
|
||||
tool_inputs = tool.params_to_strings( param_values, trans.app )
|
||||
return tool_inputs, associations
|
||||
|
||||
|
||||
def __cleanup_param_values( inputs, values ):
|
||||
"""
|
||||
Remove 'Data' values from `param_values`, along with metadata cruft,
|
||||
but track the associations.
|
||||
"""
|
||||
associations = []
|
||||
# dbkey is pushed in by the framework
|
||||
if 'dbkey' in values:
|
||||
del values['dbkey']
|
||||
root_values = values
|
||||
|
||||
# Recursively clean data inputs and dynamic selects
|
||||
def cleanup( prefix, inputs, values ):
|
||||
for key, input in inputs.items():
|
||||
if isinstance( input, ( SelectToolParameter, DrillDownSelectToolParameter ) ):
|
||||
if input.is_dynamic and not isinstance( values[key], UnvalidatedValue ):
|
||||
values[key] = UnvalidatedValue( values[key] )
|
||||
if isinstance( input, DataToolParameter ):
|
||||
tmp = values[key]
|
||||
values[key] = None
|
||||
# HACK: Nested associations are not yet working, but we
|
||||
# still need to clean them up so we can serialize
|
||||
# if not( prefix ):
|
||||
if tmp: # this is false for a non-set optional dataset
|
||||
if not isinstance(tmp, list):
|
||||
associations.append( ( tmp.hid, prefix + key ) )
|
||||
else:
|
||||
associations.extend( [ (t.hid, prefix + key) for t in tmp] )
|
||||
|
||||
# Cleanup the other deprecated crap associated with datasets
|
||||
# as well. Worse, for nested datasets all the metadata is
|
||||
# being pushed into the root. FIXME: MUST REMOVE SOON
|
||||
key = prefix + key + "_"
|
||||
for k in root_values.keys():
|
||||
if k.startswith( key ):
|
||||
del root_values[k]
|
||||
elif isinstance( input, Repeat ):
|
||||
group_values = values[key]
|
||||
for i, rep_values in enumerate( group_values ):
|
||||
rep_index = rep_values['__index__']
|
||||
cleanup( "%s%s_%d|" % (prefix, key, rep_index ), input.inputs, group_values[i] )
|
||||
elif isinstance( input, Conditional ):
|
||||
group_values = values[input.name]
|
||||
current_case = group_values['__current_case__']
|
||||
cleanup( "%s%s|" % ( prefix, key ), input.cases[current_case].inputs, group_values )
|
||||
cleanup( "", inputs, values )
|
||||
return associations
|
||||
|
||||
__all__ = [ summarize, extract_workflow ]
|
||||
@@ -0,0 +1,63 @@
|
||||
""" This module contains utility methods for reasoning about and ordering
|
||||
workflow steps.
|
||||
"""
|
||||
import math
|
||||
from galaxy.util.topsort import (
|
||||
CycleError,
|
||||
topsort,
|
||||
topsort_levels
|
||||
)
|
||||
|
||||
|
||||
def attach_ordered_steps( workflow, steps ):
|
||||
""" Attempt to topologically order steps and attach to workflow. If this
|
||||
fails - the workflow contains cycles so it mark it as such.
|
||||
"""
|
||||
ordered_steps = order_workflow_steps( steps )
|
||||
if ordered_steps:
|
||||
workflow.has_cycles = False
|
||||
for i, step in enumerate( ordered_steps ):
|
||||
step.order_index = i
|
||||
workflow.steps.append( step )
|
||||
else:
|
||||
workflow.has_cycles = True
|
||||
workflow.steps = steps
|
||||
|
||||
|
||||
def order_workflow_steps( steps ):
|
||||
"""
|
||||
Perform topological sort of the steps, return ordered or None
|
||||
"""
|
||||
position_data_available = True
|
||||
for step in steps:
|
||||
if not step.position or not 'left' in step.position or not 'top' in step.position:
|
||||
position_data_available = False
|
||||
if position_data_available:
|
||||
steps.sort(cmp=lambda s1, s2: cmp( math.sqrt(s1.position['left'] ** 2 + s1.position['top'] ** 2), math.sqrt(s2.position['left'] ** 2 + s2.position['top'] ** 2)))
|
||||
try:
|
||||
edges = edgelist_for_workflow_steps( steps )
|
||||
node_order = topsort( edges )
|
||||
return [ steps[i] for i in node_order ]
|
||||
except CycleError:
|
||||
return None
|
||||
|
||||
|
||||
def edgelist_for_workflow_steps( steps ):
|
||||
"""
|
||||
Create a list of tuples representing edges between ``WorkflowSteps`` based
|
||||
on associated ``WorkflowStepConnection``s
|
||||
"""
|
||||
edges = []
|
||||
steps_to_index = dict( ( step, i ) for i, step in enumerate( steps ) )
|
||||
for step in steps:
|
||||
edges.append( ( steps_to_index[step], steps_to_index[step] ) )
|
||||
for conn in step.input_connections:
|
||||
edges.append( ( steps_to_index[conn.output_step], steps_to_index[conn.input_step] ) )
|
||||
return edges
|
||||
|
||||
|
||||
def order_workflow_steps_with_levels( steps ):
|
||||
try:
|
||||
return topsort_levels( edgelist_for_workflow_steps( steps ) )
|
||||
except CycleError:
|
||||
return None
|
||||
@@ -67,6 +67,39 @@ class WorkflowsApiTestCase( api.ApiTestCase, TestsDatasets ):
|
||||
self._assert_status_code_is( run_workflow_response, 200 )
|
||||
self._wait_for_history( history_id, assert_ok=True )
|
||||
|
||||
def test_extract_from_history( self ):
|
||||
workflow = self.workflow_populator.load_workflow( name="test_for_extract" )
|
||||
workflow_request, history_id = self._setup_workflow_run( workflow )
|
||||
contents_response = self._get( "histories/%s/contents" % history_id )
|
||||
self._assert_status_code_is( contents_response, 200 )
|
||||
hda_ids = map( lambda c: c[ "id" ], contents_response.json() )
|
||||
|
||||
run_workflow_response = self._post( "workflows", data=workflow_request )
|
||||
self._assert_status_code_is( run_workflow_response, 200 )
|
||||
|
||||
self._wait_for_history( history_id, assert_ok=True )
|
||||
data = dict( history_id=history_id, tool_id="cat1" )
|
||||
jobs_response = self._get( "jobs", data=data )
|
||||
self._assert_status_code_is( jobs_response, 200 )
|
||||
cat1_job_id = jobs_response.json()[ 0 ][ "id" ]
|
||||
|
||||
contents_response = self._get( "history/%s/contents", data=data )
|
||||
create_from_data = dict(
|
||||
from_history_id=history_id,
|
||||
dataset_ids=dumps( hda_ids ),
|
||||
job_ids=dumps( [ cat1_job_id ] ),
|
||||
workflow_name="test import from history",
|
||||
)
|
||||
run_workflow_response = self._post( "workflows", data=create_from_data )
|
||||
self._assert_status_code_is( run_workflow_response, 200 )
|
||||
|
||||
new_workflow_id = run_workflow_response.json()[ "id" ]
|
||||
download_response = self._get( "workflows/%s/download" % new_workflow_id )
|
||||
self._assert_status_code_is( download_response, 200 )
|
||||
downloaded_workflow = download_response.json()
|
||||
self.assertEquals( downloaded_workflow[ "name" ], "test import from history" )
|
||||
assert len( downloaded_workflow[ "steps" ] ) == 3
|
||||
|
||||
def test_run_replace_params_by_tool( self ):
|
||||
workflow_request, history_id = self._setup_random_x2_workflow( "test_for_replace_tool_params" )
|
||||
workflow_request[ "parameters" ] = dumps( dict( random_lines1=dict( num_lines=5 ) ) )
|
||||
|
||||
@@ -0,0 +1,98 @@
|
||||
import unittest
|
||||
|
||||
from galaxy import model
|
||||
from galaxy.workflow import extract
|
||||
|
||||
UNDEFINED_JOB = object()
|
||||
|
||||
|
||||
class TestWorkflowExtractSummary( unittest.TestCase ):
|
||||
|
||||
def setUp( self ):
|
||||
self.history = MockHistory()
|
||||
self.trans = MockTrans( self.history )
|
||||
|
||||
def test_empty_history( self ):
|
||||
job_dict, warnings = extract.summarize( trans=self.trans )
|
||||
assert not warnings
|
||||
assert not job_dict
|
||||
|
||||
def test_summarize_returns_name_and_dataset_list( self ):
|
||||
# Create two jobs and three datasets, test they are groupped
|
||||
# by job correctly with correct output names.
|
||||
hda1 = MockHda()
|
||||
self.history.active_datasets.append( hda1 )
|
||||
hda2 = MockHda( job=hda1.job, output_name="out2" )
|
||||
self.history.active_datasets.append( hda2 )
|
||||
hda3 = MockHda( output_name="out3" )
|
||||
self.history.active_datasets.append( hda3 )
|
||||
|
||||
job_dict, warnings = extract.summarize( trans=self.trans )
|
||||
assert len( job_dict ) == 2
|
||||
assert not warnings
|
||||
self.assertEquals( job_dict[ hda1.job ], [ ( 'out1', hda1 ), ( 'out2', hda2 ) ] )
|
||||
self.assertEquals( job_dict[ hda3.job ], [ ( 'out3', hda3 ) ] )
|
||||
|
||||
def test_finds_original_job_if_copied( self ):
|
||||
hda = MockHda()
|
||||
derived_hda_1 = MockHda()
|
||||
derived_hda_1.copied_from_history_dataset_association = hda
|
||||
derived_hda_2 = MockHda()
|
||||
derived_hda_2.copied_from_history_dataset_association = derived_hda_1
|
||||
self.history.active_datasets.append( derived_hda_2 )
|
||||
job_dict, warnings = extract.summarize( trans=self.trans )
|
||||
assert not warnings
|
||||
assert len( job_dict ) == 1
|
||||
self.assertEquals( job_dict[ hda.job ], [ ('out1', derived_hda_2 ) ] )
|
||||
|
||||
def test_fake_job( self ):
|
||||
""" Fakes job if creating_job_associations is empty.
|
||||
"""
|
||||
hda = MockHda( job=UNDEFINED_JOB )
|
||||
self.history.active_datasets.append( hda )
|
||||
job_dict, warnings = extract.summarize( trans=self.trans )
|
||||
assert not warnings
|
||||
assert len( job_dict ) == 1
|
||||
fake_job = job_dict.keys()[ 0 ]
|
||||
assert fake_job.id.startswith( "fake_" )
|
||||
datasets = job_dict.values()[ 0 ]
|
||||
assert datasets == [ ( None, hda ) ]
|
||||
|
||||
def test_warns_and_skips_datasets_if_not_finished( self ):
|
||||
hda = MockHda( state='queued' )
|
||||
self.history.active_datasets.append( hda )
|
||||
job_dict, warnings = extract.summarize( trans=self.trans )
|
||||
assert warnings
|
||||
assert len( job_dict ) == 0
|
||||
|
||||
|
||||
class MockHistory( object ):
|
||||
|
||||
def __init__( self ):
|
||||
self.active_datasets = []
|
||||
|
||||
|
||||
class MockTrans( object ):
|
||||
|
||||
def __init__( self, history ):
|
||||
self.history = history
|
||||
|
||||
def get_history( self ):
|
||||
return self.history
|
||||
|
||||
|
||||
class MockHda( object ):
|
||||
|
||||
def __init__( self, state='ok', output_name='out1', job=None ):
|
||||
self.id = 123
|
||||
self.state = state
|
||||
self.copied_from_history_dataset_association = None
|
||||
if job is not UNDEFINED_JOB:
|
||||
if not job:
|
||||
job = model.Job()
|
||||
self.job = job
|
||||
assoc = model.JobToOutputDatasetAssociation( output_name, self )
|
||||
assoc.job = job
|
||||
self.creating_job_associations = [ assoc ]
|
||||
else:
|
||||
self.creating_job_associations = []
|
||||
Reference in New Issue
Block a user