Warning

This document is for an old release 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 itertools import chain

from galaxy.util.topsort import (
CycleError,
topsort,
topsort_levels,
)

[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.
"""
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

comment.order_index = i

return workflow.has_cycles

"""
Perform topological sort of the steps and comments,
"""
position_data_available = bool(steps)

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 comment.type == "freehand":
else:

# consider comments to find normalization position

# normalize comments by min_left and min_top
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))

# 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:

[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: