This document is for an in-development version of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy.workflow.steps

""" This module contains utility methods for reasoning about and ordering
workflow steps.
import math

from galaxy.util.topsort import (

[docs]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) workflow.has_cycles = True if ordered_steps: workflow.has_cycles = False workflow.steps = ordered_steps for i, step in enumerate(workflow.steps): step.order_index = i return workflow.has_cycles
[docs]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 'left' not in step.position or 'top' not in step.position: position_data_available = False if position_data_available: steps.sort(key=lambda _: math.sqrt(_.position['left'] ** 2 + _.position['top'] ** 2)) try: edges = sorted(edgelist_for_workflow_steps(steps)) node_order = topsort(edges) return [steps[i] for i in node_order] except CycleError: return None
[docs]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: output_index = steps_to_index[conn.output_step] input_index = steps_to_index[conn.input_step] # self connection - a cycle not detectable by topsort function. if output_index == input_index: raise CycleError([], 0, 0) edges.append((output_index, input_index)) return edges
[docs]def order_workflow_steps_with_levels(steps): try: return topsort_levels(edgelist_for_workflow_steps(steps)) except CycleError: return None