diff --git a/conf/containers.config b/conf/containers.config index ea228832..6e3ab029 100755 --- a/conf/containers.config +++ b/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" } diff --git a/conf/resources.config b/conf/resources.config index 0ceeade0..7b7f0459 100755 --- a/conf/resources.config +++ b/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 } diff --git a/conf/resources_juno.config b/conf/resources_juno.config index 556b0070..e700a301 100755 --- a/conf/resources_juno.config +++ b/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 } diff --git a/conf/resources_juno_genome.config b/conf/resources_juno_genome.config index 5eda48c7..02d46b1c 100644 --- a/conf/resources_juno_genome.config +++ b/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 } diff --git a/dsl2.nf b/dsl2.nf index 72cea33a..6856cbc6 100644 --- a/dsl2.nf +++ b/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 -> diff --git a/modules/process/Alignment/bam2fastq.nf b/modules/process/Alignment/bam2fastq.nf new file mode 100644 index 00000000..c0294095 --- /dev/null +++ b/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}`@{} + """ +} diff --git a/modules/subworkflow/alignment_wf.nf b/modules/subworkflow/alignment_wf.nf index 5813d994..715dcad7 100644 --- a/modules/subworkflow/alignment_wf.nf +++ b/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: diff --git a/nextflow.config b/nextflow.config index 8856eda0..33135565 100755 --- a/nextflow.config +++ b/nextflow.config @@ -38,6 +38,7 @@ params { pairing = false bamMapping = false splitLanes = true + bam2fastq = false QC = false aggregate = false fileTracking = 'fileTracking.tsv' diff --git a/tests/tests.json b/tests/tests.json index 4b1e35cf..36738bf8 100644 --- a/tests/tests.json +++ b/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"],