Enhancement/bam2fastq

conf/containers.config

@@ -9,6 +9,9 @@
  //------------- Read alignment
 
  process {
+  withName:bam2fastq {
+    container = "broadinstitute/gatk:4.1.9.0"
+  }
   withName:AlignReads {
     container = "cmopipeline/fastp-bwa-samtools:2.0.0"
   }

conf/resources_juno_genome.config

@@ -9,6 +9,10 @@
 //------------- Read alignment
 
  process {
+  withName:bam2fastq {
+    cpus = { 1 }
+    memory = { 16.GB + (task.attempt).GB }
+  }
   withName:CrossValidateSamples {
     cpus = { 1 }
     memory = { 1.GB }

dsl2.nf

@@ -27,6 +27,7 @@ referenceMap = defineReferenceMap()
 targetsMap   = loadTargetReferences()
 
 //Sub-workflow Includes
+include { bam2fastq }            from './modules/process/Alignment/bam2fastq'
 include { validate_wf }          from './modules/subworkflow/validate_wf'         addParams(referenceMap: referenceMap, targetsMap: targetsMap)
 include { alignment_wf }         from './modules/subworkflow/alignment_wf'        addParams(referenceMap: referenceMap, targetsMap: targetsMap)
 include { manta_wf }             from './modules/subworkflow/manta_wf'            addParams(referenceMap: referenceMap, targetsMap: targetsMap)
@@ -58,7 +59,7 @@ WFs = (!params.mapping && !params.bamMapping && aggregateParamIsFile) ? ['snv','
 
 workflow {
   //Set flags for when each pipeline is required to run.
-  doWF_align           = (params.mapping) ? true : false
+  doWF_align           = params.mapping || (params.bam2fastq && params.bamMapping) ? true : false
   doWF_manta           = ['snv', 'sv', 'mutsig'].any(it -> it in WFs) ? true : false
   doWF_scatter         = ['snv', 'sv', 'mutsig', 'germsnv'].any(it -> it in WFs) ? true : false
   doWF_germSNV         = 'germsnv' in WFs ? true : false
@@ -96,7 +97,7 @@ workflow {
       exit 1
   }
 
-  if (params.bamMapping && WFs == ['']){
+  if (params.bamMapping && !params.bam2fastq && WFs == ['']){
       println "ERROR: No sub-workflows to run.."
       println "\tPlease provide sub-workflows using --workflow parameters."
       exit 1
@@ -126,13 +127,27 @@ workflow {
     inputMapping = validate_wf.out.inputMapping
     inputPairing = validate_wf.out.inputPairing
 
+    if (params.bam2fastq)
+    {
+      inputBam = inputMapping
+      bam2fastq(inputBam)
+      fastqs = bam2fastq.out.fastqOutput
+			.map { idSample, targets, files_pe1, files_pe2
+			  -> tuple(groupKey(idSample, files_pe1.size()), targets, files_pe1, files_pe2)
+			}
+			.transpose()
+    }
+    else {
+      fastqs = inputMapping
+    }
+
     if (doWF_align)
     {
-      alignment_wf(inputMapping)
+      alignment_wf(fastqs)
     }
 
     //Handle input bams as coming originally from bams, or from an alignment this run.
-    if (params.bamMapping) {
+    if (params.bamMapping && !params.bam2fastq) {
       inputBam = inputMapping
       if (doWF_QC){
         inputMapping.map{idSample, target, bam, bai ->

modules/process/Alignment/bam2fastq.nf

@@ -0,0 +1,36 @@
+process bam2fastq {
+  tag "${idSample}"
+
+  input:
+    tuple val(idSample), val(target), file(bam), val(bai)
+
+  output:
+    tuple val(idSample), val(target), path("*_1.fastq.gz"), path("*_2.fastq.gz"), emit: fastqOutput
+
+  script:
+  inputSize = bam.size()
+  if (workflow.profile == "juno") {
+    if (inputSize > 80.GB) {
+      task.time = { params.maxWallTime }
+    }
+    else if (inputSize < 40.GB) {
+      task.time = task.exitStatus.toString() in params.wallTimeExitCode ? { params.medWallTime } : { params.minWallTime }
+    }
+    else {
+      task.time = task.exitStatus.toString() in params.wallTimeExitCode ? { params.maxWallTime } : { params.medWallTime }
+    }
+    // if it's the last time to try, use 500h as time limit no matter for what reason it failed before
+    task.time = task.attempt < 3 ? task.time : { params.maxWallTime }
+  }
+    memMultiplier = params.mem_per_core ? task.cpus : 1
+    // when increase memory requested from system every time it retries, keep java Xmx steady, in order to give more memory for java garbadge collection
+    originalMem = task.attempt ==1 ? task.memory : originalMem
+    maxMem = (memMultiplier * originalMem.toString().split(" ")[0].toInteger() - 3)
+    maxMem = maxMem < 4 ? 5 : maxMem
+    javaOptions    = "--java-options '-Xmx" + originalMem.toString().split(" ")[0].toInteger() * memMultiplier + "g'"
+
+  """
+  gatk SamToFastq ${javaOptions} VALIDATION_STRINGENCY=LENIENT I=${bam} RG_TAG=ID OUTPUT_PER_RG=true COMPRESS_OUTPUTS_PER_RG=true OUTPUT_DIR=./ INCLUDE_NON_PF_READS=true INCLUDE_NON_PRIMARY_ALIGNMENTS=true
+  ls *.fastq.gz | xargs -I {} -n1 mv {} `basename ${idSample}`@{}
+  """
+}

modules/subworkflow/alignment_wf.nf

@@ -12,13 +12,6 @@ workflow alignment_wf
     referenceMap = params.referenceMap
     targetsMap   = params.targetsMap
 
-    if (params.bamMapping)
-    {
-      println "Alignment workflow cannot accept bam files for input."
-      exit 1
-    }
-    if(params.mapping)
-    {
       // Parse input FASTQ mapping
       if (params.watch != true) {
         inputMapping.groupTuple(by: [0])
@@ -180,13 +173,6 @@ workflow alignment_wf
             out.println "${obj[0]}\t${obj[1]}\t${obj[2]}\t${obj[3]}"
         }
       }
-    }
-    else{
-      if(params.pairing){
-        println "ERROR: When --pairing [tsv], --mapping [tsv] must be provided."
-        exit 1
-      }
-    }
 
 
   emit:

nextflow.config

@@ -38,6 +38,7 @@ params {
   pairing = false
   bamMapping = false
   splitLanes = true
+  bam2fastq = false
   QC = false
   aggregate = false
   fileTracking = 'fileTracking.tsv'