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.config

@@ -14,6 +14,11 @@
     executor = 'local'
   }
 
+  withName:bam2fastq {
+    cpus = { 1 }
+    memory = { 6.GB + (task.attempt).GB }
+  }
+
   withName:SplitLanesR1 {
     cpus = { 1 }
     memory = { 1.GB }

conf/resources_juno.config

@@ -14,6 +14,10 @@
     memory = { 1.GB }
     executor = 'local'
   }
+  withName:bam2fastq {
+    cpus = { 1 }
+    memory = { 16.GB + (task.attempt).GB }
+  }
   withName:SplitLanesR1 {
     cpus = { 1 }
     memory = { 1.GB }

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)
@@ -61,7 +62,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
@@ -100,7 +101,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
@@ -130,13 +131,29 @@ workflow {
     inputMapping = validate_wf.out.inputMapping
     inputPairing = validate_wf.out.inputPairing
 
+    if (params.bam2fastq)
+    {
+      inputBam = inputMapping
+      outname  = 'bamMapping_realigned.tsv'
+      bam2fastq(inputBam)
+      fastqs = bam2fastq.out.fastqOutput
+			.map { idSample, targets, files_pe1, files_pe2
+			  -> 	def calculatedSize = (files_pe1.size() instanceof Collection) ? files_pe1.size() : 1
+				tuple(groupKey(idSample, calculatedSize), 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=false
+  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])
@@ -182,13 +175,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'

tests/tests.json

@@ -2,6 +2,10 @@
     "test_full": {
         "command": ["nextflow", "run", "dsl2.nf", "--mapping", "tests/mapping.tsv", "--pairing", "tests/pairing.tsv", "-profile", "test", "--workflows", "snv,sv,mutsig,lohhla,facets,msisensor,germSNV,germSV", "-resume", "--aggregate"],
         "checks": [{"type": "checkNumberOfLines", "filename": "test-data/test_result/cohort_level/default_cohort/sample_data.txt", "num_lines": 2}, {"type": "checkExitCode", "expected": 0}]
+    },    
+    "test_bam2fastq": {
+        "command": ["./nextflow", "run", "dsl2.nf", "--bamMapping", "bamMapping.tsv", "--bam2fastq", "--pairing", "tests/pairing.tsv", "-profile", "test", "--workflows", "snv,sv,mutsig,lohhla,facets,msisensor,germSNV,germSV", "-resume", "--aggregate"],
+        "checks": [{"type": "checkNumberOfLines", "filename": "test-data/test_result/cohort_level/default_cohort/sample_data.txt", "num_lines": 2}, {"type": "checkExitCode", "expected": 0}]
     },
     "test_aggregate_different_cohort": {
         "command": ["nextflow", "run", "dsl2.nf", "--mapping", "tests/mapping.tsv", "--pairing", "tests/pairing.tsv", "-profile", "test", "--workflows", "snv,sv,mutsig,lohhla,facets,msisensor, germSNV, germSV", "--aggregate", "tests/cohort.tsv", "-resume"],