From e2deebc24414bee930e3902e1feeaaadcdcab769 Mon Sep 17 00:00:00 2001 From: Anne Marie Noronha Date: Thu, 20 Jun 2024 16:14:10 -0400 Subject: [PATCH 1/6] add subworkflows for reading inputs with Channel.interval --- modules/function/read_inputs_interval.nf | 168 ++++++++++++++++++++ modules/subworkflow/AggregateFromProcess.nf | 7 +- modules/subworkflow/AggregateFromResult.nf | 7 +- modules/subworkflow/validate_wf.nf | 11 +- 4 files changed, 183 insertions(+), 10 deletions(-) create mode 100644 modules/function/read_inputs_interval.nf diff --git a/modules/function/read_inputs_interval.nf b/modules/function/read_inputs_interval.nf new file mode 100644 index 00000000..4071d268 --- /dev/null +++ b/modules/function/read_inputs_interval.nf @@ -0,0 +1,168 @@ +workflow watchMapping { + take: + read_inputs_channel + main: + def index = 0 + def interval_count = 0 + read_inputs_channel + .map{ + interval_count = it + 1 + index = 0 + file(params.bamMapping) + } + .splitCsv(sep: '\t', header: true) + .filter{ row -> + index += 1 + if (params.chunkSizeLimit > 0 ){ + index <= params.chunkSizeLimit*interval_count + }else{ + 1 + } + }.unique() + .map{ row -> + def idSample = row.SAMPLE + def target = row.TARGET + def fastqFile1 = file(row.FASTQ_PE1, checkIfExists: false) + def fastqFile2 = file(row.FASTQ_PE2, checkIfExists: false) + def numOfPairs = row.NUM_OF_PAIRS.toInteger() + if(!TempoUtils.checkTarget(target, params.assayType, validTargetsList)){} + if(!TempoUtils.checkNumberOfItem(row, 5, tsvFile)){} + + [idSample, numOfPairs, target, fastqFile1, fastqFile2] + } + .map{ idSample, numOfPairs, target, files_pe1, files_pe2 + -> tuple( groupKey(idSample, numOfPairs), target, files_pe1, files_pe2) + } + .transpose() + .unique() + .set{mapping_ch} + emit: + mapping_ch +} + +workflow watchBamMapping { + take: + read_inputs_channel + + main: + def index = 0 + def interval_count = 0 + read_inputs_channel + .map{ + interval_count = it + 1 + index = 0 + file(params.bamMapping) + } + .splitCsv(sep: '\t', header: true) + .filter{ row -> + index = index + 1 + if (params.chunkSizeLimit > 0 ){ + index <= params.chunkSizeLimit*interval_count + }else{ 1 } + + }.unique() + .map{ row -> + def idSample = row.SAMPLE + def target = row.TARGET + def bam = file(row.BAM, checkIfExists: false) + def bai = file(row.BAI, checkIfExists: false) + if(!TempoUtils.checkTarget(target, params.assayType, params.targetsMap.keySet()){} + if(!TempoUtils.checkNumberOfItem(row, 4, params.bamMapping)){} + + [idSample, target, bam, bai] + } + .map{ idSample, target, files_pe1, files_pe2 + -> tuple( groupKey(idSample, 1), target, files_pe1, files_pe2) + } + .transpose() + .unique() + .set{bamMapping_ch} + emit: + bamMapping_ch + +} + +workflow watchPairing { + take: + read_inputs_channel + main: + read_inputs_channel + .map{ params.pairing } + .splitCsv(sep: '\t', header: true) + .unique() + .map { row -> + def TUMOR_ID = row.TUMOR_ID + def NORMAL_ID = row.NORMAL_ID + if(!TempoUtils.checkNumberOfItem(row, 2, tsvFile)){} + + [TUMOR_ID, NORMAL_ID] + }.unique() + .set{pairing_ch} + emit: + pairing_ch + +} + +workflow watchAggregateWithResult { + take: + read_inputs_channel + main: + def index = 0 + def interval_count = 0 + + read_inputs_channel + .map{ + interval_count = it + 1 + index = 0 + file(params.aggregate) + }.splitCsv(sep: '\t', header: true) + .filter{ index, row -> + index += 1 + if (params.chunkSizeLimit > 0 ){ + index <= params.chunkSizeLimit*interval_count + } else { 1 } + .map{ row -> + def idNormal = row.NORMAL_ID + def idTumor = row.TUMOR_ID + def cohort = row.COHORT + def cohortSize = row.COHORT_SIZE.toInteger() + def path = row.PATH + if(!TempoUtils.checkNumberOfItem(row, 5, file(runAggregate))){} + [cohort, cohortSize, idTumor, idNormal, path] + }.map { cohort, cohortSize, idTumor, idNormal, path + -> tuple( groupKey(cohort, cohortSize), idTumor, idNormal, path) + }.transpose() + .unique() + .set{aggregate_ch} + + emit: + aggregate_ch + +} + +workflow watchAggregate { + take: + read_inputs_channel + main: + read_inputs_channel + .map{ params.aggregate } + .splitCsv(sep: '\t', header: true) + .unique() + .map{ row -> + def idNormal = row.NORMAL_ID + def idTumor = row.TUMOR_ID + def cohort = row.COHORT + def cohortSize = row.COHORT_SIZE.toInteger() + if(!TempoUtils.checkNumberOfItem(row, 4, tsvFile)){} + + [cohort, cohortSize, idTumor, idNormal] + } + .map { cohort, cohortSize, idTumor, idNormal + -> tuple( groupKey(cohort, cohortSize), idTumor, idNormal) + } + .transpose() + .unique() + .set{aggregate_ch} + emit: + aggregate_ch +} diff --git a/modules/subworkflow/AggregateFromProcess.nf b/modules/subworkflow/AggregateFromProcess.nf index d4d9326a..9f164d96 100644 --- a/modules/subworkflow/AggregateFromProcess.nf +++ b/modules/subworkflow/AggregateFromProcess.nf @@ -12,7 +12,8 @@ include { SomaticAggregateSvSignatures } from '../process/Aggregate/Somati include { SomaticAggregateHRDetect } from '../process/Aggregate/SomaticAggregateHRDetect' include { SomaticAggregateSVclone } from '../process/Aggregate/SomaticAggregateSVclone' include { CohortRunMultiQC } from '../process/Aggregate/CohortRunMultiQC' -include { watchMapping; watchBamMapping; watchPairing; watchAggregateWithResult; watchAggregate } from '../function/watch_inputs.nf' +//include { watchMapping; watchBamMapping; watchPairing; watchAggregateWithResult; watchAggregate } from '../function/watch_inputs.nf' +include { watchAggregate } from '../function/read_inputs_interval' workflow aggregateFromProcess { @@ -47,8 +48,8 @@ workflow aggregateFromProcess .set{inputAggregate} } else{ - watchAggregate(file(runAggregate, checkIfExists: false)) - .set{inputAggregate} + read_inputs_channel = Channel.interval('100s').view() + inputAggregate = watchAggregate(read_inputs_channel).aggregate_ch } } else { diff --git a/modules/subworkflow/AggregateFromResult.nf b/modules/subworkflow/AggregateFromResult.nf index 8a5a679b..78010bcf 100644 --- a/modules/subworkflow/AggregateFromResult.nf +++ b/modules/subworkflow/AggregateFromResult.nf @@ -12,7 +12,8 @@ include { SomaticAggregateSvSignatures } from '../process/Aggregate/Somati include { SomaticAggregateHRDetect } from '../process/Aggregate/SomaticAggregateHRDetect' include { SomaticAggregateSVclone } from '../process/Aggregate/SomaticAggregateSVclone' include { CohortRunMultiQC } from '../process/Aggregate/CohortRunMultiQC' -include { watchMapping; watchBamMapping; watchPairing; watchAggregateWithResult; watchAggregate } from '../function/watch_inputs.nf' +// include { watchMapping; watchBamMapping; watchPairing; watchAggregateWithResult; watchAggregate } from '../function/watch_inputs.nf' +include { watchAggregateWithResult } from '../function/read_inputs_interval' workflow aggregateFromResult { @@ -44,8 +45,8 @@ workflow aggregateFromResult .set{ inputAggregate } } else{ - watchAggregateWithResult(file(aggregateFile, checkIfExists: true)) - .set{ inputAggregate } + read_inputs_channel = Channel.interval('100s').view() + inputAggregate = watchAggregateWithResult(read_inputs_channel).aggregate_ch } inputAggregate.multiMap{ cohort, idTumor, idNormal, path -> diff --git a/modules/subworkflow/validate_wf.nf b/modules/subworkflow/validate_wf.nf index 110e5310..2fe2a5c9 100644 --- a/modules/subworkflow/validate_wf.nf +++ b/modules/subworkflow/validate_wf.nf @@ -1,12 +1,15 @@ include { CrossValidateSamples } from '../process/SampleValidation/CrossValidateSamples' -include { watchMapping; watchBamMapping; watchPairing; watchAggregateWithResult; watchAggregate } from '../function/watch_inputs.nf' +//include { watchMapping; watchPairing; watchAggregateWithResult; watchAggregate } from '../function/watch_inputs.nf' +include { watchMapping; watchBamMapping; watchPairing } from '../function/read_inputs_interval.nf' workflow validate_wf { main: referenceMap = params.referenceMap targetsMap = params.targetsMap - + if (params.watch == true) { + read_inputs_channel = Channel.interval('100s').view() + } TempoUtils.checkAssayType(params.assayType) target_id_list = targetsMap.keySet() if (params.watch == false) { @@ -15,7 +18,7 @@ workflow validate_wf } else if (params.watch == true) { mappingFile = params.mapping ? file(params.mapping, checkIfExists: false) : file(params.bamMapping, checkIfExists: false) - inputMapping = params.mapping ? watchMapping(mappingFile, params.assayType, target_id_list) : watchBamMapping(mappingFile, params.assayType, target_id_list) + inputMapping = params.mapping ? watchMapping(read_inputs_channel) : watchBamMapping(read_inputs_channel).bamMapping_ch } else{} if(params.pairing){ @@ -36,7 +39,7 @@ workflow validate_wf } else if (params.watch == true) { pairingFile = file(params.pairing, checkIfExists: false) - inputPairing = watchPairing(pairingFile) + inputPairing = watchPairing(read_inputs_channel).pairing_ch } else{} } From 8450cac93c82c6c6735286c02d7658138fccbe0d Mon Sep 17 00:00:00 2001 From: Anne Marie Noronha Date: Thu, 20 Jun 2024 16:38:22 -0400 Subject: [PATCH 2/6] fix syntax mistakes --- modules/function/read_inputs_interval.nf | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/modules/function/read_inputs_interval.nf b/modules/function/read_inputs_interval.nf index 4071d268..13764396 100644 --- a/modules/function/read_inputs_interval.nf +++ b/modules/function/read_inputs_interval.nf @@ -66,7 +66,7 @@ workflow watchBamMapping { def target = row.TARGET def bam = file(row.BAM, checkIfExists: false) def bai = file(row.BAI, checkIfExists: false) - if(!TempoUtils.checkTarget(target, params.assayType, params.targetsMap.keySet()){} + if(!TempoUtils.checkTarget(target, params.assayType, params.targetsMap.keySet())){} if(!TempoUtils.checkNumberOfItem(row, 4, params.bamMapping)){} [idSample, target, bam, bai] @@ -116,12 +116,12 @@ workflow watchAggregateWithResult { index = 0 file(params.aggregate) }.splitCsv(sep: '\t', header: true) - .filter{ index, row -> + .filter{ row -> index += 1 if (params.chunkSizeLimit > 0 ){ index <= params.chunkSizeLimit*interval_count } else { 1 } - .map{ row -> + }.map{ row -> def idNormal = row.NORMAL_ID def idTumor = row.TUMOR_ID def cohort = row.COHORT From 568c6e99ab0ede3f26b896230941ed4b60aafc90 Mon Sep 17 00:00:00 2001 From: Anne Marie Noronha Date: Thu, 31 Oct 2024 17:48:23 -0400 Subject: [PATCH 3/6] fix more bugs --- modules/function/read_inputs_interval.nf | 14 +++++++------- modules/subworkflow/AggregateFromProcess.nf | 4 ++-- modules/subworkflow/AggregateFromResult.nf | 2 +- modules/subworkflow/validate_wf.nf | 5 +++-- 4 files changed, 13 insertions(+), 12 deletions(-) diff --git a/modules/function/read_inputs_interval.nf b/modules/function/read_inputs_interval.nf index 13764396..75bf49ba 100644 --- a/modules/function/read_inputs_interval.nf +++ b/modules/function/read_inputs_interval.nf @@ -8,7 +8,7 @@ workflow watchMapping { .map{ interval_count = it + 1 index = 0 - file(params.bamMapping) + file(params.mapping) } .splitCsv(sep: '\t', header: true) .filter{ row -> @@ -26,7 +26,7 @@ workflow watchMapping { def fastqFile2 = file(row.FASTQ_PE2, checkIfExists: false) def numOfPairs = row.NUM_OF_PAIRS.toInteger() if(!TempoUtils.checkTarget(target, params.assayType, validTargetsList)){} - if(!TempoUtils.checkNumberOfItem(row, 5, tsvFile)){} + if(!TempoUtils.checkNumberOfItem(row, 5, params.mapping)){} [idSample, numOfPairs, target, fastqFile1, fastqFile2] } @@ -87,13 +87,13 @@ workflow watchPairing { read_inputs_channel main: read_inputs_channel - .map{ params.pairing } + .map{ file(params.pairing) } .splitCsv(sep: '\t', header: true) .unique() .map { row -> def TUMOR_ID = row.TUMOR_ID def NORMAL_ID = row.NORMAL_ID - if(!TempoUtils.checkNumberOfItem(row, 2, tsvFile)){} + if(!TempoUtils.checkNumberOfItem(row, 2, params.pairing)){} [TUMOR_ID, NORMAL_ID] }.unique() @@ -127,7 +127,7 @@ workflow watchAggregateWithResult { def cohort = row.COHORT def cohortSize = row.COHORT_SIZE.toInteger() def path = row.PATH - if(!TempoUtils.checkNumberOfItem(row, 5, file(runAggregate))){} + if(!TempoUtils.checkNumberOfItem(row, 5, file(params.aggregate))){} [cohort, cohortSize, idTumor, idNormal, path] }.map { cohort, cohortSize, idTumor, idNormal, path -> tuple( groupKey(cohort, cohortSize), idTumor, idNormal, path) @@ -145,7 +145,7 @@ workflow watchAggregate { read_inputs_channel main: read_inputs_channel - .map{ params.aggregate } + .map{ file(params.aggregate) } .splitCsv(sep: '\t', header: true) .unique() .map{ row -> @@ -153,7 +153,7 @@ workflow watchAggregate { def idTumor = row.TUMOR_ID def cohort = row.COHORT def cohortSize = row.COHORT_SIZE.toInteger() - if(!TempoUtils.checkNumberOfItem(row, 4, tsvFile)){} + if(!TempoUtils.checkNumberOfItem(row, 4, file(params.aggregate))){} [cohort, cohortSize, idTumor, idNormal] } diff --git a/modules/subworkflow/AggregateFromProcess.nf b/modules/subworkflow/AggregateFromProcess.nf index 9f164d96..0a64c4fc 100644 --- a/modules/subworkflow/AggregateFromProcess.nf +++ b/modules/subworkflow/AggregateFromProcess.nf @@ -48,8 +48,8 @@ workflow aggregateFromProcess .set{inputAggregate} } else{ - read_inputs_channel = Channel.interval('100s').view() - inputAggregate = watchAggregate(read_inputs_channel).aggregate_ch + read_inputs_channel = Channel.interval(params.touchInputsInterval * 60 + 's').view() + inputAggregate = watchAggregate(read_inputs_channel).aggregate_ch.view() } } else { diff --git a/modules/subworkflow/AggregateFromResult.nf b/modules/subworkflow/AggregateFromResult.nf index 78010bcf..680d2093 100644 --- a/modules/subworkflow/AggregateFromResult.nf +++ b/modules/subworkflow/AggregateFromResult.nf @@ -45,7 +45,7 @@ workflow aggregateFromResult .set{ inputAggregate } } else{ - read_inputs_channel = Channel.interval('100s').view() + read_inputs_channel = Channel.interval(params.touchInputsInterval * 60 + 's').view() inputAggregate = watchAggregateWithResult(read_inputs_channel).aggregate_ch } diff --git a/modules/subworkflow/validate_wf.nf b/modules/subworkflow/validate_wf.nf index 2fe2a5c9..9e578489 100644 --- a/modules/subworkflow/validate_wf.nf +++ b/modules/subworkflow/validate_wf.nf @@ -8,7 +8,7 @@ workflow validate_wf referenceMap = params.referenceMap targetsMap = params.targetsMap if (params.watch == true) { - read_inputs_channel = Channel.interval('100s').view() + read_inputs_channel = Channel.interval(params.touchInputsInterval * 60 + 's').view() } TempoUtils.checkAssayType(params.assayType) target_id_list = targetsMap.keySet() @@ -18,7 +18,7 @@ workflow validate_wf } else if (params.watch == true) { mappingFile = params.mapping ? file(params.mapping, checkIfExists: false) : file(params.bamMapping, checkIfExists: false) - inputMapping = params.mapping ? watchMapping(read_inputs_channel) : watchBamMapping(read_inputs_channel).bamMapping_ch + inputMapping = params.mapping ? watchMapping(read_inputs_channel).mapping_ch : watchBamMapping(read_inputs_channel).bamMapping_ch } else{} if(params.pairing){ @@ -40,6 +40,7 @@ workflow validate_wf else if (params.watch == true) { pairingFile = file(params.pairing, checkIfExists: false) inputPairing = watchPairing(read_inputs_channel).pairing_ch + inputPairing.view() } else{} } From fcb6e1891445706b169d5f90cc59f9fbf57880b1 Mon Sep 17 00:00:00 2001 From: Anne Marie Noronha Date: Wed, 6 Nov 2024 12:54:04 -0500 Subject: [PATCH 4/6] fix missing variable definition --- modules/function/read_inputs_interval.nf | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/modules/function/read_inputs_interval.nf b/modules/function/read_inputs_interval.nf index 75bf49ba..f0a62376 100644 --- a/modules/function/read_inputs_interval.nf +++ b/modules/function/read_inputs_interval.nf @@ -25,7 +25,7 @@ workflow watchMapping { def fastqFile1 = file(row.FASTQ_PE1, checkIfExists: false) def fastqFile2 = file(row.FASTQ_PE2, checkIfExists: false) def numOfPairs = row.NUM_OF_PAIRS.toInteger() - if(!TempoUtils.checkTarget(target, params.assayType, validTargetsList)){} + if(!TempoUtils.checkTarget(target, params.assayType, params.targetsMap.keySet())){} if(!TempoUtils.checkNumberOfItem(row, 5, params.mapping)){} [idSample, numOfPairs, target, fastqFile1, fastqFile2] From 9ca74f3946b920ee3c38ce4a4d2c8007ea2b9723 Mon Sep 17 00:00:00 2001 From: Anne Marie Noronha Date: Wed, 6 Nov 2024 17:48:10 -0500 Subject: [PATCH 5/6] clean up --- dsl2.nf | 11 -- modules/function/read_inputs_interval.nf | 184 ++++++++++---------- modules/subworkflow/AggregateFromProcess.nf | 1 - modules/subworkflow/AggregateFromResult.nf | 1 - modules/subworkflow/validate_wf.nf | 5 +- 5 files changed, 94 insertions(+), 108 deletions(-) diff --git a/dsl2.nf b/dsl2.nf index 72cea33a..5c282e09 100644 --- a/dsl2.nf +++ b/dsl2.nf @@ -20,7 +20,6 @@ params.startEpoch = new Date().getTime() //Utility Includes include { defineReferenceMap; loadTargetReferences } from './modules/function/define_maps' -include { touchInputs; watchMapping; watchBamMapping; watchPairing; watchAggregateWithResult; watchAggregate } from './modules/function/watch_inputs' pairingQc = params.pairing referenceMap = defineReferenceMap() @@ -111,16 +110,6 @@ workflow { exit 1 } - if (params.watch == true) { - epochMap = [:] - for (i in ["mapping","bamMapping","pairing","aggregate"]) { - if (file(params."${i}".toString()).exists()){ - epochMap[file(params."${i}").toRealPath()] = 0 - } - } - touchInputs(params.chunkSizeLimit, epochMap) - } - if (doWF_AggregateFromResult){ aggregateFromResult(runAggregate, multiqcWesConfig, multiqcWgsConfig, multiqcTempoLogo) } diff --git a/modules/function/read_inputs_interval.nf b/modules/function/read_inputs_interval.nf index f0a62376..04522f6d 100644 --- a/modules/function/read_inputs_interval.nf +++ b/modules/function/read_inputs_interval.nf @@ -5,37 +5,37 @@ workflow watchMapping { def index = 0 def interval_count = 0 read_inputs_channel - .map{ - interval_count = it + 1 - index = 0 - file(params.mapping) + .map{ + interval_count = it + 1 + index = 0 + file(params.mapping) + } + .splitCsv(sep: '\t', header: true) + .filter{ row -> + index += 1 + if (params.chunkSizeLimit > 0 ){ + index <= params.chunkSizeLimit*interval_count + }else{ + 1 } - .splitCsv(sep: '\t', header: true) - .filter{ row -> - index += 1 - if (params.chunkSizeLimit > 0 ){ - index <= params.chunkSizeLimit*interval_count - }else{ - 1 - } - }.unique() - .map{ row -> - def idSample = row.SAMPLE - def target = row.TARGET - def fastqFile1 = file(row.FASTQ_PE1, checkIfExists: false) - def fastqFile2 = file(row.FASTQ_PE2, checkIfExists: false) - def numOfPairs = row.NUM_OF_PAIRS.toInteger() - if(!TempoUtils.checkTarget(target, params.assayType, params.targetsMap.keySet())){} - if(!TempoUtils.checkNumberOfItem(row, 5, params.mapping)){} + }.unique() + .map{ row -> + def idSample = row.SAMPLE + def target = row.TARGET + def fastqFile1 = file(row.FASTQ_PE1, checkIfExists: false) + def fastqFile2 = file(row.FASTQ_PE2, checkIfExists: false) + def numOfPairs = row.NUM_OF_PAIRS.toInteger() + if(!TempoUtils.checkTarget(target, params.assayType, params.targetsMap.keySet())){} + if(!TempoUtils.checkNumberOfItem(row, 5, params.mapping)){} - [idSample, numOfPairs, target, fastqFile1, fastqFile2] - } - .map{ idSample, numOfPairs, target, files_pe1, files_pe2 - -> tuple( groupKey(idSample, numOfPairs), target, files_pe1, files_pe2) - } - .transpose() - .unique() - .set{mapping_ch} + [idSample, numOfPairs, target, fastqFile1, fastqFile2] + } + .map{ idSample, numOfPairs, target, files_pe1, files_pe2 + -> tuple( groupKey(idSample, numOfPairs), target, files_pe1, files_pe2) + } + .transpose() + .unique() + .set{mapping_ch} emit: mapping_ch } @@ -48,37 +48,37 @@ workflow watchBamMapping { def index = 0 def interval_count = 0 read_inputs_channel - .map{ - interval_count = it + 1 - index = 0 - file(params.bamMapping) - } - .splitCsv(sep: '\t', header: true) - .filter{ row -> - index = index + 1 - if (params.chunkSizeLimit > 0 ){ - index <= params.chunkSizeLimit*interval_count - }else{ 1 } - - }.unique() - .map{ row -> - def idSample = row.SAMPLE - def target = row.TARGET - def bam = file(row.BAM, checkIfExists: false) - def bai = file(row.BAI, checkIfExists: false) - if(!TempoUtils.checkTarget(target, params.assayType, params.targetsMap.keySet())){} - if(!TempoUtils.checkNumberOfItem(row, 4, params.bamMapping)){} + .map{ + interval_count = it + 1 + index = 0 + file(params.bamMapping) + } + .splitCsv(sep: '\t', header: true) + .filter{ row -> + index = index + 1 + if (params.chunkSizeLimit > 0 ){ + index <= params.chunkSizeLimit*interval_count + }else{ 1 } + + }.unique() + .map{ row -> + def idSample = row.SAMPLE + def target = row.TARGET + def bam = file(row.BAM, checkIfExists: false) + def bai = file(row.BAI, checkIfExists: false) + if(!TempoUtils.checkTarget(target, params.assayType, params.targetsMap.keySet())){} + if(!TempoUtils.checkNumberOfItem(row, 4, params.bamMapping)){} - [idSample, target, bam, bai] - } - .map{ idSample, target, files_pe1, files_pe2 - -> tuple( groupKey(idSample, 1), target, files_pe1, files_pe2) - } - .transpose() - .unique() - .set{bamMapping_ch} - emit: - bamMapping_ch + [idSample, target, bam, bai] + } + .map{ idSample, target, files_pe1, files_pe2 + -> tuple( groupKey(idSample, 1), target, files_pe1, files_pe2) + } + .transpose() + .unique() + .set{bamMapping_ch} + emit: + bamMapping_ch } @@ -113,27 +113,27 @@ workflow watchAggregateWithResult { read_inputs_channel .map{ interval_count = it + 1 - index = 0 + index = 0 file(params.aggregate) }.splitCsv(sep: '\t', header: true) .filter{ row -> index += 1 if (params.chunkSizeLimit > 0 ){ - index <= params.chunkSizeLimit*interval_count - } else { 1 } - }.map{ row -> - def idNormal = row.NORMAL_ID - def idTumor = row.TUMOR_ID - def cohort = row.COHORT - def cohortSize = row.COHORT_SIZE.toInteger() - def path = row.PATH - if(!TempoUtils.checkNumberOfItem(row, 5, file(params.aggregate))){} - [cohort, cohortSize, idTumor, idNormal, path] - }.map { cohort, cohortSize, idTumor, idNormal, path - -> tuple( groupKey(cohort, cohortSize), idTumor, idNormal, path) - }.transpose() - .unique() - .set{aggregate_ch} + index <= params.chunkSizeLimit*interval_count + } else { 1 } + }.map{ row -> + def idNormal = row.NORMAL_ID + def idTumor = row.TUMOR_ID + def cohort = row.COHORT + def cohortSize = row.COHORT_SIZE.toInteger() + def path = row.PATH + if(!TempoUtils.checkNumberOfItem(row, 5, file(params.aggregate))){} + [cohort, cohortSize, idTumor, idNormal, path] + }.map { cohort, cohortSize, idTumor, idNormal, path + -> tuple( groupKey(cohort, cohortSize), idTumor, idNormal, path) + }.transpose() + .unique() + .set{aggregate_ch} emit: aggregate_ch @@ -145,24 +145,24 @@ workflow watchAggregate { read_inputs_channel main: read_inputs_channel - .map{ file(params.aggregate) } - .splitCsv(sep: '\t', header: true) - .unique() - .map{ row -> - def idNormal = row.NORMAL_ID - def idTumor = row.TUMOR_ID - def cohort = row.COHORT - def cohortSize = row.COHORT_SIZE.toInteger() - if(!TempoUtils.checkNumberOfItem(row, 4, file(params.aggregate))){} + .map{ file(params.aggregate) } + .splitCsv(sep: '\t', header: true) + .unique() + .map{ row -> + def idNormal = row.NORMAL_ID + def idTumor = row.TUMOR_ID + def cohort = row.COHORT + def cohortSize = row.COHORT_SIZE.toInteger() + if(!TempoUtils.checkNumberOfItem(row, 4, file(params.aggregate))){} - [cohort, cohortSize, idTumor, idNormal] - } - .map { cohort, cohortSize, idTumor, idNormal - -> tuple( groupKey(cohort, cohortSize), idTumor, idNormal) - } - .transpose() - .unique() - .set{aggregate_ch} + [cohort, cohortSize, idTumor, idNormal] + } + .map { cohort, cohortSize, idTumor, idNormal + -> tuple( groupKey(cohort, cohortSize), idTumor, idNormal) + } + .transpose() + .unique() + .set{aggregate_ch} emit: aggregate_ch } diff --git a/modules/subworkflow/AggregateFromProcess.nf b/modules/subworkflow/AggregateFromProcess.nf index 0a64c4fc..d309c37d 100644 --- a/modules/subworkflow/AggregateFromProcess.nf +++ b/modules/subworkflow/AggregateFromProcess.nf @@ -12,7 +12,6 @@ include { SomaticAggregateSvSignatures } from '../process/Aggregate/Somati include { SomaticAggregateHRDetect } from '../process/Aggregate/SomaticAggregateHRDetect' include { SomaticAggregateSVclone } from '../process/Aggregate/SomaticAggregateSVclone' include { CohortRunMultiQC } from '../process/Aggregate/CohortRunMultiQC' -//include { watchMapping; watchBamMapping; watchPairing; watchAggregateWithResult; watchAggregate } from '../function/watch_inputs.nf' include { watchAggregate } from '../function/read_inputs_interval' workflow aggregateFromProcess diff --git a/modules/subworkflow/AggregateFromResult.nf b/modules/subworkflow/AggregateFromResult.nf index 680d2093..0cbe4960 100644 --- a/modules/subworkflow/AggregateFromResult.nf +++ b/modules/subworkflow/AggregateFromResult.nf @@ -12,7 +12,6 @@ include { SomaticAggregateSvSignatures } from '../process/Aggregate/Somati include { SomaticAggregateHRDetect } from '../process/Aggregate/SomaticAggregateHRDetect' include { SomaticAggregateSVclone } from '../process/Aggregate/SomaticAggregateSVclone' include { CohortRunMultiQC } from '../process/Aggregate/CohortRunMultiQC' -// include { watchMapping; watchBamMapping; watchPairing; watchAggregateWithResult; watchAggregate } from '../function/watch_inputs.nf' include { watchAggregateWithResult } from '../function/read_inputs_interval' workflow aggregateFromResult diff --git a/modules/subworkflow/validate_wf.nf b/modules/subworkflow/validate_wf.nf index 9e578489..da4793d9 100644 --- a/modules/subworkflow/validate_wf.nf +++ b/modules/subworkflow/validate_wf.nf @@ -1,5 +1,4 @@ include { CrossValidateSamples } from '../process/SampleValidation/CrossValidateSamples' -//include { watchMapping; watchPairing; watchAggregateWithResult; watchAggregate } from '../function/watch_inputs.nf' include { watchMapping; watchBamMapping; watchPairing } from '../function/read_inputs_interval.nf' workflow validate_wf @@ -8,7 +7,7 @@ workflow validate_wf referenceMap = params.referenceMap targetsMap = params.targetsMap if (params.watch == true) { - read_inputs_channel = Channel.interval(params.touchInputsInterval * 60 + 's').view() + read_inputs_channel = Channel.interval(params.touchInputsInterval * 60 + 's') } TempoUtils.checkAssayType(params.assayType) target_id_list = targetsMap.keySet() @@ -40,7 +39,7 @@ workflow validate_wf else if (params.watch == true) { pairingFile = file(params.pairing, checkIfExists: false) inputPairing = watchPairing(read_inputs_channel).pairing_ch - inputPairing.view() + inputPairing } else{} } From 2bee864ada9ffd8f1241d214db312ee9da51a936 Mon Sep 17 00:00:00 2001 From: Anne Marie Noronha Date: Thu, 7 Nov 2024 11:10:35 -0500 Subject: [PATCH 6/6] remove obsolete watch_inputs.nf --- modules/function/watch_inputs.nf | 155 ------------------------------- 1 file changed, 155 deletions(-) delete mode 100644 modules/function/watch_inputs.nf diff --git a/modules/function/watch_inputs.nf b/modules/function/watch_inputs.nf deleted file mode 100644 index 7c0f01e7..00000000 --- a/modules/function/watch_inputs.nf +++ /dev/null @@ -1,155 +0,0 @@ -def touchInputs(chunkSizeLimit, epochMap) { - new Timer().schedule({ - for ( i in epochMap.keySet() ){ - fileEpoch = file(i).lastModified() - if (( fileEpoch > epochMap[i]) || (chunkSizeLimit > 0 )) { - epochMap[i] = fileEpoch - "touch -ca ${i}".execute() - } - } -} as TimerTask, 15*1000, params.touchInputsInterval * 60 * 1000 ) // convert minutes to milliseconds -} - -def watchMapping(tsvFile, assayType, validTargetsList) { - def index = 0 - def limitInputLines = params.chunkSizeLimit - Channel.watchPath( tsvFile, 'create, modify' ) - .map{ row -> - def timeNow = new Date().getTime() - limitInputLines = params.chunkSizeLimit + ( ((timeNow - params.startEpoch)/60000) * (params.chunkSizeLimit / params.touchInputsInterval) ) - index = 0 - row - }.splitCsv(sep: '\t', header: true) - .map{ row -> - [index++] + row - }.filter{ row -> - if (params.chunkSizeLimit > 0 ){ - row[0] <= limitInputLines - } else { 1 } - }.map{ row -> - row[1] - }.unique() - .map{ row -> - def idSample = row.SAMPLE - def target = row.TARGET - def fastqFile1 = file(row.FASTQ_PE1, checkIfExists: false) - def fastqFile2 = file(row.FASTQ_PE2, checkIfExists: false) - def numOfPairs = row.NUM_OF_PAIRS.toInteger() - if(!TempoUtils.checkTarget(target, assayType, validTargetsList)){} - if(!TempoUtils.checkNumberOfItem(row, 5, tsvFile)){} - - [idSample, numOfPairs, target, fastqFile1, fastqFile2] - } - .map{ idSample, numOfPairs, target, files_pe1, files_pe2 - -> tuple( groupKey(idSample, numOfPairs), target, files_pe1, files_pe2) - } - .transpose() - .unique() -} - -def watchBamMapping(tsvFile, assayType, validTargetsList){ - def index = 0 - def limitInputLines = params.chunkSizeLimit - Channel.watchPath( tsvFile, 'create, modify' ) - .map{ row -> - def timeNow = new Date().getTime() - limitInputLines = params.chunkSizeLimit + ( ((timeNow - params.startEpoch)/60000) * (params.chunkSizeLimit / params.touchInputsInterval) ) - index = 0 - row - }.splitCsv(sep: '\t', header: true) - .map{ row -> - [index++] + row - }.filter{ row -> - if (params.chunkSizeLimit > 0 ){ - row[0] <= limitInputLines - } else { 1 } - }.map{ row -> - row[1] - }.unique() - .map{ row -> - def idSample = row.SAMPLE - def target = row.TARGET - def bam = file(row.BAM, checkIfExists: false) - def bai = file(row.BAI, checkIfExists: false) - if(!TempoUtils.checkTarget(target, assayType, validTargetsList)){} - if(!TempoUtils.checkNumberOfItem(row, 4, tsvFile)){} - - [idSample, target, bam, bai] - } - .map{ idSample, target, files_pe1, files_pe2 - -> tuple( groupKey(idSample, 1), target, files_pe1, files_pe2) - } - .transpose() - .unique() -} - -def watchPairing(tsvFile){ - Channel.watchPath( tsvFile, 'create, modify' ) - .splitCsv(sep: '\t', header: true) - .unique() - .map { row -> - def TUMOR_ID = row.TUMOR_ID - def NORMAL_ID = row.NORMAL_ID - if(!TempoUtils.checkNumberOfItem(row, 2, tsvFile)){} - - [TUMOR_ID, NORMAL_ID] - } - .unique() -} - -def watchAggregateWithResult(tsvFile) { - def index = 0 - def limitInputLines = params.chunkSizeLimit - Channel.watchPath(tsvFile, 'create, modify') - .map{ row -> - def timeNow = new Date().getTime() - limitInputLines = params.chunkSizeLimit + ( ((timeNow - params.startEpoch)/60000) * (params.chunkSizeLimit / params.touchInputsInterval) ) - index = 0 - row - }.splitCsv(sep: '\t', header: true) - .map{ row -> - [index++] + row - }.filter{ row -> - if (params.chunkSizeLimit > 0 ){ - row[0] <= limitInputLines - } else { 1 } - }.map{ row -> - row[1] - }.unique() - .map{ row -> - def idNormal = row.NORMAL_ID - def idTumor = row.TUMOR_ID - def cohort = row.COHORT - def cohortSize = row.COHORT_SIZE.toInteger() - def path = row.PATH - if(!TempoUtils.checkNumberOfItem(row, 5, file(runAggregate))){} - - [cohort, cohortSize, idTumor, idNormal, path] - } - .map { cohort, cohortSize, idTumor, idNormal, path - -> tuple( groupKey(cohort, cohortSize), idTumor, idNormal, path) - } - .transpose() - .unique() -} - -def watchAggregate(tsvFile) { - Channel.watchPath(tsvFile, 'create, modify') - .splitCsv(sep: '\t', header: true) - .unique() - .map{ row -> - def idNormal = row.NORMAL_ID - def idTumor = row.TUMOR_ID - def cohort = row.COHORT - def cohortSize = row.COHORT_SIZE.toInteger() - if(!TempoUtils.checkNumberOfItem(row, 4, tsvFile)){} - - [cohort, cohortSize, idTumor, idNormal] - } - .map { cohort, cohortSize, idTumor, idNormal - -> tuple( groupKey(cohort, cohortSize), idTumor, idNormal) - } - .transpose() - .unique() -} -