How do I handle tuple of paths where the first path is a list of paths?

[Shians]

I am trying to write a pipeline that haplotypes some BAM files based on different varying VCF files. I cannot figure out how to put in the inputs in nice way. Initially I had something like

bam_files = Channel.fromPath("data/*.bam")
bam_indices = Channel.fromPath("data/*.bam.bai")

In process:

input:
  path(bam_files)
  path(bai_files)
  path(vcf_file)

the through was to run the process twice, but I found out that Nextflow did allow processes to be run twice. So the idea was to create a channel with two elements that contained

[[bam_files], [bai_files], vcf_file]

and change the input to

input:
  tuple path(bam_files), path(bai_files),  path(vcf_file)

but this produces a warning. So I am looking for advice on how such a situation should be handled properly. I am aware I could create a sub-module and import it under two different names, but this feels like a hack that is not very scalable.

[bentsherman]

What warning did you get?

[Shians]

Starting with

.
├── data
│   ├── file1.bam
│   ├── file1.bam.bai
│   ├── file2.bam
│   ├── file2.bam.bai
│   ├── file3.bam
│   ├── file3.bam.bai
│   ├── genome1.fa
│   └── genome2.fa
├── main.nf

So my initial idea was to do to this

process Test {
    input:
    path bam_files
    path bam_indices
    path genome

    output:
    stdout

    script:
    """
    echo bam_files: ${bam_files}
    echo bam_indices: ${bam_indices}
    echo genome: ${genome}
    """

}

workflow {
    bam_files = Channel.fromPath("data/*.bam") | collect
    bam_indices = Channel.fromPath("data/*.bam.bai") | collect

    Test(bam_files, bam_indices, Channel.fromPath("data/genome1.fa")) | view
    Test(bam_files, bam_indices, Channel.fromPath("data/genome2.fa")) | view
}

But that doesn't work because I cannot use Test twice. So instead I tried to make a Channel with tuples.

process Test {
    input:
    tuple path(bam_files), path(bam_indices), path(genome)
    
    output:
    stdout

    script:
    """
    echo bam_files: ${bam_files}
    echo bam_indices: ${bam_indices}
    echo genome: ${genome}
    """

}

workflow {
    bam_files = Channel.fromPath("data/*.bam") | collect | toList
    bam_indices = Channel.fromPath("data/*.bam.bai") | collect | toList

    input_channel = bam_files.combine(bam_indices).combine(Channel.fromPath("data/genome1.fa"))
        .combine(bam_files.combine(bam_indices).combine(Channel.fromPath("data/genome2.fa")))

    input_channel | view

    Test(input_channel) | view
}

which produced the warning

WARN: Input tuple does not match tuple declaration in process `Test` -- offending value: [[/Users/su.s/Programs/NextFlow/MultiPathIssue/data/file2.bam, /Users/su.s/Programs/NextFlow/MultiPathIssue/data/file3.bam, /Users/su.s/Programs/NextFlow/MultiPathIssue/data/file1.bam], [/Users/su.s/Programs/NextFlow/MultiPathIssue/data/file1.bam.bai, /Users/su.s/Programs/NextFlow/MultiPathIssue/data/file2.bam.bai, /Users/su.s/Programs/NextFlow/MultiPathIssue/data/file3.bam.bai], /Users/su.s/Programs/NextFlow/MultiPathIssue/data/genome1.fa, [/Users/su.s/Programs/NextFlow/MultiPathIssue/data/file2.bam, /Users/su.s/Programs/NextFlow/MultiPathIssue/data/file3.bam, /Users/su.s/Programs/NextFlow/MultiPathIssue/data/file1.bam], [/Users/su.s/Programs/NextFlow/MultiPathIssue/data/file1.bam.bai, /Users/su.s/Programs/NextFlow/MultiPathIssue/data/file2.bam.bai, /Users/su.s/Programs/NextFlow/MultiPathIssue/data/file3.bam.bai], /Users/su.s/Programs/NextFlow/MultiPathIssue/data/genome2.fa]

Upon further experimenting, I realized that what I created was not actually a channel of two elements, not entirely sure what it is, but I fixed it by using concat instead of combine

workflow {
    bam_files = Channel.fromPath("data/*.bam") | collect | toList
    bam_indices = Channel.fromPath("data/*.bam.bai") | collect | toList

    input_channel = bam_files.combine(bam_indices).combine(Channel.fromPath("data/genome1.fa"))
        .concat(bam_files.combine(bam_indices).combine(Channel.fromPath("data/genome2.fa")))

    input_channel | view

    Test(input_channel) | view
}

Works as expected. So a tuple containing a list of paths as an element seems to be accepted properly, though constructing such channels is quite awkward. Is there some way to write something like the following?

 input_channel = Channel.concat(
        Channel.combine(
            bam_files,
            bam_indices,
            Channel.fromPath("data/genome1.fa")
        ),
        Channel.combine(
            bam_files,
            bam_indices,
            Channel.fromPath("data/genome2.fa")
        )
    )

[bentsherman]

I would write it like this:

workflow {
    ch_bam = Channel.fromFilePairs("data/*.bam", flat: true)
    ch_bai = Channel.fromFilePairs("data/*.bam.bai", flat: true)
    ch_genome = Channel.fromPath("data/*.fa")

    ch_input = ch_bam
        .combine(ch_bai, by: 0)
        .combine(ch_genome)

    Test( ch_input )
}

That should give you a 4-tuple of (sample id, bam, bai, genome) for every combination of bam/bai pair and genome. Not sure if you wanted this or if you actually wanted to process all bam/bai pairs in a single job

[Shians]

I did actually want to have all my BAM files as a single input and only vary the genome. Thanks for letting me know about fromFilePairs, it seems useful as I had also previously struggled with manipulating channels into tuples to use with join/combine.

[bentsherman]

Yeah I figured it would be good to show the more complicated scenario. But in your case it should actually be simpler. Sometimes you can use the files() function instead of Channel.fromPath() which gives you more flexibility:

workflow {
    ch_bambai = Channel.of(
        tuple( files("data/*.bam"), files("data/*.bam.bai") )
    )
    ch_genome = Channel.fromPath("data/*.fa")

    ch_input = ch_bambai.combine(ch_genome)

    Test( ch_input )
}

That should give you a 3-tuple of (bams, bais, genome) for each genome file