templates.py

from gwf import AnonymousTarget


def template_one(bam_file, bar=False):
    inputs = {'bam_file': bam_file}
    outputs = {'bar_file': 'bar.{bam_file}'.format(bam_file=bam_file)}
    if bar:
        options = {'cores': 16}
    else:
        options = {}
    spec = """
    cat {bam_file} > bar.{bam_file}
    echo bar >> {bam_file}
    """.format(bam_file=bam_file)
    return AnonymousTarget(
        inputs=inputs,
        outputs=outputs,
        options=options,
        spec=spec,
    )


def template_two(bar_file):
    inputs = {'bar_file': bar_file}
    outputs = {
        'fixed_file': 'fixed.{bar_file}'.format(bar_file=bar_file),
        'stat_file': 'stat.{bar_file}'.format(bar_file=bar_file),
    }
    options = {}
    spec = """
    cat {bar_file} > bar.{bar_file}
    echo bar >> {bar_file}
    """.format(bar_file=bar_file)
    return AnonymousTarget(
        inputs=inputs,
        outputs=outputs,
        options=options,
        spec=spec,
    )


def template_three(fixed_files):
    inputs = {'fixed_files': fixed_files}
    outputs = {'results_file': 'all.txt'}
    options = {}
    spec = """
    touch all.txt
    """
    return AnonymousTarget(
        inputs=inputs,
        outputs=outputs,
        options=options,
        spec=spec,
    )

workflow.py

from gwf import Workflow
from gwf.helpers import collect

from templates import (
    template_one,
    template_two,
    template_three,
)


BAM_FILES = [{'bam_file': 'sample1.bam'}, {'bam_file': 'sample2.bam'}]


gwf = Workflow()

A = gwf.map(template_one, BAM_FILES, extra={'bar': True})
B = gwf.map(template_two, A.outputs)
C = gwf.target_from_template('collect', template_three(**collect(B.outputs, ['fixed_file'])))

print('A.inputs {}'.format(A.inputs))
print('A.outputs {}'.format(A.outputs))
print('B.inputs {}'.format(B.inputs))
print('B.outputs {}'.format(B.outputs))
print('C.inputs {}'.format(C.inputs))
print('C.outputs {}'.format(C.outputs))

print()
print("Targets created:")
for target in gwf.targets.values():
    print(target)

# A.inputs [{'bam_file': 'sample1.bam'}, {'bam_file': 'sample2.bam'}]
# A.outputs [{'bar_file': 'bar.sample1.bam'}, {'bar_file': 'bar.sample2.bam'}]
# B.inputs [{'bar_file': 'bar.sample1.bam'}, {'bar_file': 'bar.sample2.bam'}]
# B.outputs [{'stat_file': 'stat.bar.sample1.bam', 'fixed_file': 'fixed.bar.sample1.bam'}, {'stat_file': 'stat.bar.sample2.bam', 'fixed_file': 'fixed.bar.sample2.bam'}]
# C.inputs {'fixed_files': ['fixed.bar.sample1.bam', 'fixed.bar.sample2.bam']}
# C.outputs {'results_file': 'all.txt'}

# Targets created:
# collect
# template_one.0
# template_two.1
# template_one.1
# template_two.0