4
$\begingroup$

I have 36 samples in total in the bam_files folder, with the name like this 20230306_CH_EP_C01.md.bam

I expect the code to output all the 36 samples, one by one. But the run stopped after only running one sample. I have searched online, but did not find a solution. I am using nextflow version 2.

bam_chl = Channel
     .fromPath("bam_files/*.md.bam")
     .map{ item -> tuple(item.baseName.substring(0, 18), item) }
     

     
ref_chl = Channel
      .fromPath("sequences/hg38.fa")
ref_index_chl = Channel
      .fromPath("sequences/hg38.fa.fai")
amplicon_chl = Channel
      .fromPath("sequences/tp53_amplicon.hg38.interval_list")
target_chl = Channel
      .fromPath("sequences/TP53.hg38.interval_list")
     


process get_pcr_metric {

  publishDir path: 'pcr_metric', pattern: '*.csv'
  
  input:
  tuple val(pair_id), path(bam)
  path(ref)
  path(ref_index)
  path(amplicon)
  path(target)

  output:
  path("*.pcr_metrics.csv")

  """
  java -jar /picard.jar CollectTargetedPcrMetrics \
       I=$bam \
   O=${pair_id}_pcr_metrics.txt \
       R=$ref \
   AMPLICON_INTERVALS=$amplicon \
       TARGET_INTERVALS=$target
       
  sed -n -e 7p -e 8p ${pair_id}_pcr_metrics.txt |  sed 's/\t/,/g'   > ${pair_id}.pcr_metrics.csv
  
  

  """


}


workflow {

get_pcr_metric(bam_chl, ref_chl,ref_index_chl, amplicon_chl, target_chl)
}
$\endgroup$

2 Answers 2

3
$\begingroup$

The issue here is that the fromPath factory method produces a queue channel, but you don't want all of your input channels to be queue channels. Most of the time, what you want is one queue channel and one or more value channels when you require multiple input channels.

You could use the value factory method to create the value channels, but a simpler way is to just pass in some file objects. The latter works because a value channel is implicitly created by a process when it is invoked with a simple value1. For example:

params.ref_fasta = "./sequences/hg38.fa"
params.amplicon_intervals = "./sequences/tp53_amplicon_intervals.hg38.interval_list"
params.target_intervals = "./sequences/TP53.hg38.interval_list"

params.bam_files = "./bam_files/*.md.bam"
params.outdir = './results'
process get_pcr_metric {

    publishDir path: "${params.outdir}/pcr_metric", mode: 'copy'

    input:
    tuple val(pair_id), path(bam_file)
    path ref_fasta
    path amplicon_intervals
    path target_intervals

    output:
    path "${pair_id}_pcr_metrics.txt"

    """
    picard CollectTargetedPcrMetrics \\
         I="${bam_file}" \\
         O="${pair_id}_pcr_metrics.txt" \\
         R="${ref_fasta}" \\
         AMPLICON_INTERVALS="${amplicon_intervals}" \\
         TARGET_INTERVALS="${target_intervals}"
    """
}
workflow {

    ref_fasta = file( params.ref_fasta )
    amplicon_intervals = file( params.amplicon_intervals )
    target_intervals = file( params.target_intervals )

    Channel
        .fromPath( params.bam_files )
        .map { tuple( it.getBaseName(2), it ) }
        .set { bam_files }

    get_pcr_metric( bam_files, ref_fasta, amplicon_intervals, target_intervals )
}
$\endgroup$
2
$\begingroup$

The code can be made to work on all samples by changing the interface of the process and using the combine operator:

ref_chl = Channel
      .fromPath("sequences/hg38.fa")
ref_index_chl = Channel
      .fromPath("sequences/hg38.fa.fai")
amplicon_chl = Channel
      .fromPath("sequences/tp53_amplicon.hg38.interval_list")
target_chl = Channel
      .fromPath("sequences/TP53.hg38.interval_list")

bam_chl = Channel
     .fromPath("bam_files/*.md.bam")
     .map{ item -> tuple(item.baseName.substring(0, 18), item) }
     .combine(ref_chl)
     .combine(ref_index_chl)
     .combine(amplicon_chl)
     .combine(target_chl)

process get_pcr_metric {

  publishDir path: 'pcr_metric', pattern: '*.csv'
  
  input:
  tuple val(pair_id), path(bam), path(ref), path(ref_index), path(amplicon), path(target)

  output:
  path("*.pcr_metrics.csv")

  """
  java -jar /picard.jar CollectTargetedPcrMetrics \
       I=$bam \
   O=${pair_id}_pcr_metrics.txt \
       R=$ref \
   AMPLICON_INTERVALS=$amplicon \
       TARGET_INTERVALS=$target
       
  sed -n -e 7p -e 8p ${pair_id}_pcr_metrics.txt |  sed 's/\t/,/g'   > ${pair_id}.pcr_metrics.csv
  """
}

workflow {
  get_pcr_metric(bam_chl)
}

The combine operator as the name implies generates all combinations. The different file objects are then in one tuple.

Since only one Channel is passed to the process, it's now also possible to rewrite the workflow code as follows:

workflow {

  refs = [
    "sequences/hg38.fa",
    "sequences/hg38.fa.fai",
    "sequences/tp53_amplicon.hg38.interval_list",
    "sequences/TP53.hg38.interval_list"
  ]

  ref_chl = Channel.from(refs)
   | map{ pointer -> file(pointer) }
   | collect

  Channel.fromPath("bam_files/*.md.bam")
   | map{ item -> tuple(item.baseName.substring(0, 18), item) }
   | combine( ref_chl )
   | get_pcr_metric

}
$\endgroup$
1
  • $\begingroup$ You do not need the combine operator here: In general, multiple input channels should be used to process combinations of different inputs, using the each qualifier or value channels. $\endgroup$
    – Steve
    Jun 1, 2023 at 12:18

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.