Enhancement/interval channel

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)
   }

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.mapping)
+      }
+      .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)){}
+
+        [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{ 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, params.pairing)){}
+
+         [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{ 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}
+
+  emit:
+    aggregate_ch
+
+}
+
+workflow watchAggregate {
+  take:
+    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))){}
+
+        [cohort, cohortSize, idTumor, idNormal]
+      }
+      .map { cohort, cohortSize, idTumor, idNormal
+        -> tuple( groupKey(cohort, cohortSize), idTumor, idNormal)
+      }
+      .transpose()
+      .unique()
+      .set{aggregate_ch}
+  emit:
+  aggregate_ch
+}

modules/subworkflow/AggregateFromProcess.nf

@@ -12,7 +12,7 @@ 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
 {
@@ -47,8 +47,8 @@ workflow aggregateFromProcess
 		  .set{inputAggregate}
       }
       else{
-        watchAggregate(file(runAggregate, checkIfExists: false))
-	              .set{inputAggregate}
+	read_inputs_channel = Channel.interval(params.touchInputsInterval * 60 + 's').view()
+        inputAggregate = watchAggregate(read_inputs_channel).aggregate_ch.view()
       }
     }
     else {

modules/subworkflow/AggregateFromResult.nf

@@ -12,7 +12,7 @@ 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
 {
@@ -44,8 +44,8 @@ workflow aggregateFromResult
         .set{ inputAggregate }
     }
     else{
-      watchAggregateWithResult(file(aggregateFile, checkIfExists: true))
-        .set{ inputAggregate }
+      read_inputs_channel = Channel.interval(params.touchInputsInterval * 60 + 's').view()
+      inputAggregate = watchAggregateWithResult(read_inputs_channel).aggregate_ch
     }
 
     inputAggregate.multiMap{ cohort, idTumor, idNormal, path ->