Source code for galaxy.workflow.steps

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

import math
from itertools import chain

from galaxy.util.topsort import (

[docs]def attach_ordered_steps(workflow): """Attempt to topologically order steps and comments, and attach to workflow. If ordering Steps fails - the workflow contains cycles so it mark it as such. """ ordered_steps, ordered_comments = order_workflow_steps(workflow.steps, workflow.comments) 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 workflow.comments = ordered_comments for i, comment in enumerate(workflow.comments): comment.order_index = i return workflow.has_cycles
[docs]def order_workflow_steps(steps, comments): """ Perform topological sort of the steps and comments, return (ordered steps, ordered comments) or (None, ordered comments) """ position_data_available = bool(steps) ordered_comments = comments for step in steps: if step.subworkflow: attach_ordered_steps(step.subworkflow) if not step.position or "left" not in step.position or "top" not in step.position: position_data_available = False if position_data_available: # find minimum left and top values to normalize position min_left = min(step.position["left"] for step in steps) min_top = min(step.position["top"] for step in steps) if comments: freehand_comments = [] sortable_comments = [] for comment in comments: if comment.type == "freehand": freehand_comments.append(comment) else: sortable_comments.append(comment) if sortable_comments: # consider comments to find normalization position min_left_comments = min(comment.position[0] for comment in sortable_comments) min_top_comments = min(comment.position[1] for comment in sortable_comments) min_left = min(min_left_comments, min_left) min_top = min(min_top_comments, min_top) # normalize comments by min_left and min_top for comment in chain(sortable_comments, freehand_comments): comment.position = [comment.position[0] - min_left, comment.position[1] - min_top] # order by Euclidean distance to the origin sortable_comments.sort(key=lambda comment: math.sqrt(comment.position[0] ** 2 + comment.position[1] ** 2)) # replace comments list with sorted comments ordered_comments = freehand_comments ordered_comments.extend(sortable_comments) # normalize steps by min_left and min_top for step in steps: step.position = { "left": step.position["left"] - min_left, "top": step.position["top"] - min_top, # other position attributes can be discarded if present } # order by Euclidean distance to the origin (i.e. pre-normalized (min_left, min_top)) 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], ordered_comments) except CycleError: return (None, ordered_comments)
[docs]def has_cycles(workflow): try: topsort(sorted(edgelist_for_workflow_steps(workflow.steps))) return False except CycleError: return True
[docs]def edgelist_for_workflow_steps(steps): """ Create a list of tuples representing edges between ``WorkflowStep`` s based on associated ``WorkflowStepConnection`` s """ edges = [] steps_to_index = {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