Demo of failing run loop in ruffus

fail_loop_ruffus.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""

@contact: Yun Yan (yy1533@nyu.edu)
"""

from ruffus import *
from helper import *

#==============================================================================
# Parallel processing in ruffus
#==============================================================================
parameters = [['a.fa', 'a.sam', 1],
              ['b.fa', 'b.sam', 1],
              ['c.fa', 'c.sam', 1]]
@files(parameters)
def fa2sam(infile, outfile, idx):
    with open(outfile, "w") as fo:
        fo.write(str(idx))
pipeline_run(forcedtorun_tasks=[fa2sam])

#==============================================================================
# Attempt to force to use for loop in ruffus
# Question cannot be solved: where to set the task name to run?
#==============================================================================

# @for_loop: some task definition of ruffus here
for i in range(0, 3):
    parameters = [['a.fa', 'a.sam', i],
                  ['b.fa', 'b.sam', i],
                  ['c.fa', 'c.sam', i]]
    @files(parameters)
    def fa2sam(infile, outfile, idx):
        with open(outfile, "w") as fo:
            fo.write(str(idx))

pipeline_run(forcedtorun_tasks=[for_loop])

#==============================================================================
# More simple sample to state my point
#==============================================================================
def iplusj(i, j):
    return(i+j)

def param_generator():
    for i in range(0, 3):
        t = "Task" + str(i)
        j = i * 2
        yield([t, i, j])

## work
@parallel(param_generator)
def parallel_task(name, parami, paramj):
    with open(join_path("/Users/yunyan/Projects/Skywalker/ruffus101", name), "w") as fo:
        print(str(iplusj(parami, paramj)))
        fo.write(str(iplusj(parami, paramj)))

pipeline_run([parallel_task])

## not work
for id in range(0, 3):
    @parallel(param_generator)
    def parallel_task(name, parami, paramj):
        name = paste0("run", id, name)
        with open(join_path("/Users/yunyan/Projects/Skywalker/ruffus101", name), "w") as fo:
            print(str(iplusj(parami, paramj)))
            fo.write(str(iplusj(parami, paramj)))

    pipeline_run([parallel_task])