Channels.fromPath get files at runtime

[axshen]

Hi everyone,

I am working using DSL 2 and running my nextflow workflows on a Slurm cluster.

I am developing a workflow where I have a process that generates a bunch of files. In the process that follows, I want to execute a program on each of those files as parallel processes (on separate nodes of the Slurm cluster). This appears to be the default behaviour when you use Channels. What I have tried to do is provide a Channel as an input to this process with Channels.fromPath to find the files that are created (this step is sofia in the snippet below). The issue I have is that Channels.fromPath appears to search the specified directory at the time of executing the pipeline and not when the process runs, so it does not find any files and the pipeline stops there.

Does anybody know if it is possible to create a Channel at runtime, or have ideas for alternative patterns that could work for this?

Here is the code snippet with the processes and workflow

// Create scripts for running SoFiA via SoFiAX
// This creates a number of parameter files for the execution of sofia 
// and writes them to the working directory
process s2p_setup {
    container = params.S2P_IMAGE
    containerOptions = '--bind /mnt/shared:/mnt/shared'
 
    input:
        val image_cube_file
        val sofia_parameter_file_template
 
    output:
        stdout emit: stdout
 
    script:
        """
        python3 -u /app/s2p_setup.py \
            $image_cube_file \
            $sofia_parameter_file_template \
            ${params.SOURCE_FINDING_RUN_NAME} \
            ${params.SOURCE_FINDING_NODE_SIZE} \
            ${params.WORKDIR}
        """
}
 
// Run source finding application (sofia) with the specified parameter file
process sofia {
    container = params.SOFIA_IMAGE
    containerOptions = '--bind /mnt/shared:/mnt/shared'
    
    input:
        val s2p_setup
        val param_file
 
    output:
        val param_file, emit: param_file
 
    script:
        """
        #!/bin/bash
       
        sofia $param_file
        """
}
 
workflow source_finding {
    take: cube
 
    main:
        // sofia
        generate_sofia_parameter_file_template(cube)
        s2p_setup(cube, generate_sofia_parameter_file_template.out.file)
        sofia(s2p_setup.out.stdout, Channel.fromPath("${params.WORKDIR}/sofia_*.par"))
}

Thanks in advance for any help!

[pditommaso]

What is creating the files at "${params.WORKDIR}/sofia_*.par" ?

[axshen]

The process s2p_setup

[pditommaso]

then s2p_setup should declare an output for that and used as the second parameter in the sofia process invocation

[axshen]

Thanks, I tried this and was unsure about the syntax for doing this but was able to arrive at the following:

process s2p_setup {
    container = params.S2P_IMAGE
    containerOptions = '--bind /mnt/shared:/mnt/shared'

    input:
        val image_cube_file
        val sofia_parameter_file_template

    output:
        val "${params.WORKDIR}/${params.SOFIAX_CONFIG_FILE}", emit: sofiax_config
        val file("${params.WORKDIR}/sofia_*.par"), emit: parameter_files

    script:
        """
        python3 -u /app/s2p_setup.py \
            $image_cube_file \
            $sofia_parameter_file_template \
            ${params.SOURCE_FINDING_RUN_NAME} \
            ${params.SOURCE_FINDING_NODE_SIZE} \
            ${params.WORKDIR}
        """
}

...


workflow source_finding {
    take: cube

    main:
        generate_sofia_parameter_file_template(cube)
        s2p_setup(cube, generate_sofia_parameter_file_template.out.file)
        sofia(s2p_setup.out.parameter_files.flatten())
        ...
}

which does the trick.