opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-21 13:50:20 +08:00
Code Issues Packages Projects Releases Wiki Activity

Add implicit collection mapping support to PickValueModule

Browse Source 
Detect collection inputs via compute_collection_info and iterate
per-element slices so pick_value runs once per collection element.
For single-value modes output is a flat list; for all_non_null
output is list:list. Adds framework tests for both modes.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
John Chilton
2026-03-22 19:15:04 -04:00
co-authored by Claude Opus 4.6
parent fc69697dd8
commit 1263c8d946
5 changed files with 314 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/workflow/modules.py
+110 -21
View File
@@ -2067,20 +2067,9 @@ class PickValueModule(WorkflowModule):
return True
return False
def execute(
self, trans, progress: "WorkflowProgress", invocation_step, use_cached_job: bool = False
) -> Optional[bool]:
def _pick_from_replacements(self, trans, invocation_step, mode, replacements):
"""Apply pick logic to a list of replacement values. Returns the picked output."""
step = invocation_step.workflow_step
mode = step.tool_inputs.get("mode", "first_non_null") if step.tool_inputs else "first_non_null"
# Gather replacements from each named input terminal, in order
replacements = []
for input_dict in self.get_all_inputs():
replacement = progress.replacement_for_input(trans, step, input_dict)
if replacement is not NO_REPLACEMENT:
replacements.append(replacement)
# Separate non-null from null/skipped, preserving order
non_null = [r for r in replacements if not self._is_null_or_skipped(r)]
if mode == "first_non_null":
@@ -2091,13 +2080,12 @@ class PickValueModule(WorkflowModule):
workflow_step_id=step.id,
)
)
output = non_null[0]
return non_null[0]
elif mode == "first_or_skip":
if not non_null:
output = self._create_skipped_output(trans, invocation_step)
else:
output = non_null[0]
return self._create_skipped_output(trans, invocation_step)
return non_null[0]
elif mode == "the_only_non_null":
if len(non_null) != 1:
@@ -2107,20 +2095,77 @@ class PickValueModule(WorkflowModule):
workflow_step_id=step.id,
)
)
output = non_null[0]
return non_null[0]
elif mode == "all_non_null":
# CWL spec: all_non_null returns list of non-null values,
# which may be empty if all inputs are null.
output = self._create_collection_from_list(trans, invocation_step, non_null)
return self._create_collection_from_list(trans, invocation_step, non_null)
else:
raise ValueError(f"Unknown pick_value mode: {mode}")
def execute(
self, trans, progress: "WorkflowProgress", invocation_step, use_cached_job: bool = False
) -> Optional[bool]:
step = invocation_step.workflow_step
mode = step.tool_inputs.get("mode", "first_non_null") if step.tool_inputs else "first_non_null"
all_inputs = self.get_all_inputs()
collection_info = self.compute_collection_info(progress, step, all_inputs)
if collection_info:
output = self._execute_mapped(trans, invocation_step, mode, all_inputs, collection_info)
else:
# Gather replacements from each named input terminal, in order
replacements = []
for input_dict in all_inputs:
replacement = progress.replacement_for_input(trans, step, input_dict)
if replacement is not NO_REPLACEMENT:
replacements.append(replacement)
output = self._pick_from_replacements(trans, invocation_step, mode, replacements)
progress.set_step_outputs(invocation_step, {"output": output})
self._apply_post_job_actions(trans, step, output, progress.effective_replacement_dict())
return None
def _execute_mapped(self, trans, invocation_step, mode, all_inputs, collection_info):
"""Execute pick_value mapped over collection inputs."""
invocation = invocation_step.workflow_invocation
history = invocation.history
# Build a map of input_name -> input_dict for quick lookup
input_names = {d["name"] for d in all_inputs}
per_element_outputs = []
for iteration_elements, _when_value in collection_info.slice_collections():
# For each slice, extract per-element replacements
replacements = []
for input_dict in all_inputs:
name = input_dict["name"]
if iteration_elements and name in iteration_elements:
dce = iteration_elements[name]
replacement = dce.hda if hasattr(dce, "hda") and dce.hda else dce.child_collection
else:
# Input not part of the mapped collection — not connected
replacement = NO_REPLACEMENT
if replacement is not NO_REPLACEMENT:
replacements.append(replacement)
element_output = self._pick_from_replacements(trans, invocation_step, mode, replacements)
# Track the identifier from the first mapped input for naming
first_mapped = (
next(
(iteration_elements[n] for n in sorted(iteration_elements) if n in input_names),
None,
)
if iteration_elements
else None
)
identifier = first_mapped.element_identifier if first_mapped else str(len(per_element_outputs))
per_element_outputs.append((identifier, element_output))
# Build the output collection from per-element outputs
return self._create_mapped_output_collection(trans, history, mode, per_element_outputs)
def _create_skipped_output(self, trans, invocation_step):
"""Create a skipped HDA for first_or_skip when all inputs are null."""
invocation = invocation_step.workflow_invocation
@@ -2160,6 +2205,50 @@ class PickValueModule(WorkflowModule):
)
return hdca
def _create_mapped_output_collection(self, trans, history, mode, per_element_outputs):
"""Create an implicit output collection from per-element pick results.
For single-value modes (first_non_null, etc.), creates a flat list of HDAs.
For all_non_null mode, creates a list:list where each element is a sub-collection.
"""
collection_manager = trans.app.dataset_collection_manager
if mode == "all_non_null":
# Each element is an HDCA — build list:list
elements = []
for identifier, hdca in per_element_outputs:
elements.append(
dict(
name=identifier,
src="hdca",
id=hdca.id,
)
)
return collection_manager.create(
trans,
history,
name="Pick Value - mapped all non-null",
collection_type="list:list",
element_identifiers=elements,
)
else:
# Each element is an HDA — build flat list
elements = []
for identifier, hda in per_element_outputs:
elements.append(
dict(
name=identifier,
src="hda",
id=hda.id,
)
)
return collection_manager.create(
trans,
history,
name="Pick Value - mapped",
collection_type="list",
element_identifiers=elements,
)
def _apply_post_job_actions(self, trans, step, output, replacement_dict):
"""Apply post job actions directly to module output via ActionBox.
lib/galaxy_test/workflow/pick_value_all_non_null_mapped.gxwf-tests.yml
+55
View File
@@ -0,0 +1,55 @@
- doc: |
Map over pick_value(all_non_null) with both branches active.
Each input element produces a 2-element inner list.
Output is list:list with 2 outer elements.
job:
input_collection:
class: Collection
collection_type: list
elements:
- identifier: el1
class: File
contents: "line1"
filetype: bed
- identifier: el2
class: File
contents: "line2"
filetype: bed
when_a:
type: raw
value: true
when_b:
type: raw
value: true
outputs:
picked:
class: Collection
collection_type: list:list
element_count: 2
- doc: |
Map over pick_value(all_non_null) with one null branch.
Each element produces a 1-element inner list.
job:
input_collection:
class: Collection
collection_type: list
elements:
- identifier: el1
class: File
contents: "line1"
filetype: bed
- identifier: el2
class: File
contents: "line2"
filetype: bed
when_a:
type: raw
value: true
when_b:
type: raw
value: false
outputs:
picked:
class: Collection
collection_type: list:list
element_count: 2
lib/galaxy_test/workflow/pick_value_all_non_null_mapped.gxwf.yml
+38
View File
@@ -0,0 +1,38 @@
class: GalaxyWorkflow
inputs:
input_collection:
type: collection
collection_type: list
when_a:
type: boolean
when_b:
type: boolean
outputs:
picked:
outputSource: pick/output
steps:
branch_a:
tool_id: cat
in:
input1:
source: input_collection
when:
source: when_a
when: $(inputs.when)
branch_b:
tool_id: cat
in:
input1:
source: input_collection
when:
source: when_b
when: $(inputs.when)
pick:
type: pick_value
in:
input_0:
source: branch_a/out_file1
input_1:
source: branch_b/out_file1
state:
mode: all_non_null
lib/galaxy_test/workflow/pick_value_first_non_null_mapped.gxwf-tests.yml
+73
View File
@@ -0,0 +1,73 @@
- doc: |
Map over pick_value(first_non_null) with both branches active.
Each input element picks from branch_a, output is a list with same
element count as the input collection.
job:
input_collection:
class: Collection
collection_type: list
elements:
- identifier: el1
class: File
contents: "line1"
filetype: bed
- identifier: el2
class: File
contents: "line2"
filetype: bed
when_a:
type: raw
value: true
when_b:
type: raw
value: true
outputs:
picked:
class: Collection
collection_type: list
element_count: 2
elements:
el1:
asserts:
- that: has_text
text: "line1"
el2:
asserts:
- that: has_text
text: "line2"
- doc: |
Map over pick_value(first_non_null) with only branch_b active.
Each element should pick from branch_b (branch_a is skipped).
job:
input_collection:
class: Collection
collection_type: list
elements:
- identifier: el1
class: File
contents: "line1"
filetype: bed
- identifier: el2
class: File
contents: "line2"
filetype: bed
when_a:
type: raw
value: false
when_b:
type: raw
value: true
outputs:
picked:
class: Collection
collection_type: list
element_count: 2
elements:
el1:
asserts:
- that: has_text
text: "line1"
el2:
asserts:
- that: has_text
text: "line2"
lib/galaxy_test/workflow/pick_value_first_non_null_mapped.gxwf.yml
+38
View File
@@ -0,0 +1,38 @@
class: GalaxyWorkflow
inputs:
input_collection:
type: collection
collection_type: list
when_a:
type: boolean
when_b:
type: boolean
outputs:
picked:
outputSource: pick/output
steps:
branch_a:
tool_id: cat
in:
input1:
source: input_collection
when:
source: when_a
when: $(inputs.when)
branch_b:
tool_id: cat
in:
input1:
source: input_collection
when:
source: when_b
when: $(inputs.when)
pick:
type: pick_value
in:
input_0:
source: branch_a/out_file1
input_1:
source: branch_b/out_file1
state:
mode: first_non_null