move pathway related logic out of differential and correlation subworkflows

conf/modules.config

@@ -507,10 +507,7 @@ process {
             "--number_of_cutoffs ${params.propr_ncutoffs}"
         ].join(' ').trim() }
         publishDir = [
-            path: {
-                meta.args_cor ? "${params.outdir}/correlation_analysis/propr-${meta.args_cor.replace('--','').replace(' ', '_')}" :
-                                "${params.outdir}/correlation_analysis/propr"
-            },
+            path: { "${params.outdir}/correlation_analysis/propr-${meta.pathway_name ?: params.pathway}" },
             mode: params.publish_dir_mode,
             saveAs: { filename -> filename.equals('versions.yml') ? null : filename }
         ]
@@ -534,10 +531,7 @@ process {
             "--weighted_degree ${params.propd_weighted_degree}"
         ].join(' ').trim() }
         publishDir = [
-            path: {
-                meta.args_diff ? "${params.outdir}/differential_analysis/propd-${meta.args_diff.replace('--','').replace(' ', '_')}/${meta.contrast}" :
-                                "${params.outdir}/differential_analysis/propd/${meta.contrast}"
-            },
+            path: { "${params.outdir}/differential_analysis/propr-${meta.pathway_name ?: params.pathway}" },
             mode: params.publish_dir_mode,
             saveAs: { filename -> filename.equals('versions.yml') ? null : filename }
         ]
@@ -555,15 +549,7 @@ process {
             "--permutation ${params.grea_permutation}"
         ].join(' ').trim() }
         publishDir = [
-            path: {
-                meta.args_enr ? "${params.outdir}/enrichment_analysis/grea-${meta.args_enr.replace('--','').replace(' ', '_')}" :
-                                "${params.outdir}/enrichment_analysis/grea"
-            },
-            mode: params.publish_dir_mode,
-            saveAs: { filename -> filename.equals('versions.yml') ? null : filename }
-        ]
-        publishDir = [
-            path: { "${params.outdir}/enrichment_analysis/${task.process.tokenize(':')[-1].tokenize('_')[0].toLowerCase()}" },
+            path: { "${params.outdir}/enrichment_analysis/propr-${meta.pathway_name ?: params.pathway}" },
             mode: params.publish_dir_mode,
             saveAs: { filename -> filename.equals('versions.yml') ? null : filename }
         ]

subworkflows/local/correlation/main.nf

@@ -3,17 +3,9 @@
 //
 include {PROPR_PROPR as PROPR} from "../../../modules/local/propr/propr/main.nf"
 
-def correct_meta_data = { meta, data, pathway ->
-    def meta_clone = meta.clone() + pathway
-    meta_clone.remove('cor_method')
-    meta_clone.remove('args_cor')
-    return [meta_clone, data]
-}
-
 workflow CORRELATION {
     take:
-    ch_tools        // [ pathway_name, correlation_map ]
-    ch_counts
+    ch_counts       // [ meta, counts] with meta keys: method, args_cor
 
     main:
 
@@ -22,30 +14,18 @@ workflow CORRELATION {
     ch_adjacency = Channel.empty()
 
     // branch tools to select the correct correlation analysis method
-    ch_tools
+    ch_counts
         .branch {
-            propr:  it[1]["cor_method"] == "propr"
+            propr:  it[0]["method"] == "propr"
         }
-        .set { ch_tools_single }
+        .set { ch_counts }
 
     // ----------------------------------------------------
     // Perform correlation analysis with propr
     // ----------------------------------------------------
-
-    ch_counts
-        .combine(ch_tools_single.propr)
-        .multiMap {
-            metacounts, counts, pathway, metatools ->
-                input:   [ metacounts+metatools, counts ]
-                pathway: [ metacounts+metatools, pathway ]
-        }
-        .set { ch_counts_propr }
-
-    PROPR(ch_counts_propr.input.unique())
-    ch_matrix    = PROPR.out.matrix
-                        .join(ch_counts_propr.pathway).map(correct_meta_data).mix(ch_matrix)
-    ch_adjacency = PROPR.out.adjacency
-                        .join(ch_counts_propr.pathway).map(correct_meta_data).mix(ch_adjacency)
+    PROPR(ch_counts.propr.unique())
+    ch_matrix    = PROPR.out.matrix.mix(ch_matrix)
+    ch_adjacency = PROPR.out.adjacency.mix(ch_adjacency)
 
     // TODO: divide propr module into cor, propr, pcor, pcorbshrink, etc.
 

subworkflows/local/differential/main.nf

@@ -6,17 +6,9 @@ include { DESEQ2_DIFFERENTIAL } from '../../../modules/nf-core/deseq2/differenti
 include { LIMMA_DIFFERENTIAL } from '../../../modules/nf-core/limma/differential/main'
 include { FILTER_DIFFTABLE as FILTER_DIFFTABLE_LIMMA } from '../../../modules/local/filter_difftable'
 
-def correct_meta_data = { meta, data, pathway ->
-    def meta_clone = meta.clone() + pathway
-    meta_clone.remove('diff_method')
-    meta_clone.remove('args_diff')
-    return [meta_clone, data]
-}
-
 workflow DIFFERENTIAL {
     take:
-    ch_tools              // [ pathway_name, differential_map ]
-    ch_counts             // [ meta_exp, counts ]
+    ch_counts             // [ meta_exp, counts ] with meta keys: method, args_diff
     ch_samplesheet        // [ meta_exp, samplesheet ]
     ch_contrasts          // [ meta_contrast, contrast_variable, reference, target ]
 
@@ -30,13 +22,13 @@ workflow DIFFERENTIAL {
     ch_adjacency                 = Channel.empty()
 
     // branch tools to select the correct differential analysis method
-    ch_tools
+    ch_counts
         .branch {
-            propd:  it[1]["diff_method"] == "propd"
-            deseq2: it[1]["diff_method"] == "deseq2"
-            limma:  it[1]["diff_method"] == "limma"
+            propd:  it[0]["method"] == "propd"
+            deseq2: it[0]["method"] == "deseq2"
+            limma:  it[0]["method"] == "limma"
         }
-        .set { ch_tools_single }
+        .set { ch_counts }
 
     // ----------------------------------------------------
     // Perform differential analysis with propd
@@ -45,29 +37,23 @@ workflow DIFFERENTIAL {
     // TODO propd currently don't support blocking, so we should not run propd with same contrast_variable, reference and target,
     // but different blocking variable, since it will simply run the same analysis again.
 
-    ch_counts
-        .join(ch_samplesheet)
+    ch_counts.propd
+        .combine(ch_samplesheet)
+        .filter{ meta_counts, counts, meta_samplesheet, samplesheet -> meta_counts.subMap(meta_samplesheet.keySet()) == meta_samplesheet }
         .combine(ch_contrasts)
-        .combine(ch_tools_single.propd)
-        .multiMap {
-            meta_data, counts, samplesheet, meta_contrast, contrast_variable, reference, target, pathway, meta_tools ->
-                def meta = meta_data.clone() + ['contrast': meta_contrast.id] + meta_tools.clone()
-                input:   [ meta, counts, samplesheet, contrast_variable, reference, target ]
-                pathway: [ meta, pathway ]
+        .map {
+            meta_data, counts, meta_samplesheet, samplesheet, meta_contrast, contrast_variable, reference, target ->
+                def meta = meta_data.clone() + ['contrast': meta_contrast.id]
+                return [ meta, counts, samplesheet, contrast_variable, reference, target ]
         }
         .set { ch_propd }
 
-    PROPD(ch_propd.input.unique())
-    ch_results_pairwise          = PROPD.out.results
-                                        .join(ch_propd.pathway).map(correct_meta_data).mix(ch_results_pairwise)
-    ch_results_pairwise_filtered = PROPD.out.results_filtered
-                                        .join(ch_propd.pathway).map(correct_meta_data).mix(ch_results_pairwise_filtered)
-    ch_results_genewise          = PROPD.out.connectivity
-                                        .join(ch_propd.pathway).map(correct_meta_data).mix(ch_results_genewise)
-    ch_results_genewise_filtered = PROPD.out.hub_genes
-                                        .join(ch_propd.pathway).map(correct_meta_data).mix(ch_results_genewise_filtered)
-    ch_adjacency                 = PROPD.out.adjacency
-                                        .join(ch_propd.pathway).map(correct_meta_data).mix(ch_adjacency)
+    PROPD(ch_propd.unique())
+    ch_results_pairwise          = PROPD.out.results.mix(ch_results_pairwise)
+    ch_results_pairwise_filtered = PROPD.out.results_filtered.mix(ch_results_pairwise_filtered)
+    ch_results_genewise          = PROPD.out.connectivity.mix(ch_results_genewise)
+    ch_results_genewise_filtered = PROPD.out.hub_genes.mix(ch_results_genewise_filtered)
+    ch_adjacency                 = PROPD.out.adjacency.mix(ch_adjacency)
 
     // ----------------------------------------------------
     // Perform differential analysis with DESeq2
@@ -98,17 +84,16 @@ workflow DIFFERENTIAL {
 
     // combine the input channels with the tools information
     // in this way, limma will only be run if the user have specified it, as informed by ch_tools
-    ch_counts
-        .join(ch_samplesheet)
+    ch_counts.limma
+        .combine(ch_samplesheet)
+        .filter{ meta_counts, counts, meta_samplesheet, samplesheet -> meta_counts.subMap(meta_samplesheet.keySet()) == meta_samplesheet }
         .combine(ch_contrasts)
-        .combine(ch_tools_single.limma)
         .unique()
         .multiMap {
-            meta_data, counts, samplesheet, meta_contrast, contrast_variable, reference, target, pathway, meta_tools ->
-                def meta = meta_data.clone() + meta_contrast.clone() + meta_tools.clone()
+            meta_data, counts, meta_samplesheet, samplesheet, meta_contrast, contrast_variable, reference, target ->
+                def meta = meta_data.clone() + meta_contrast.clone()
                 input1:  [ meta, contrast_variable, reference, target ]
                 input2:  [ meta, samplesheet, counts ]
-                pathway: [ meta, pathway ]
         }
         .set { ch_limma }
 
@@ -127,10 +112,8 @@ workflow DIFFERENTIAL {
     )
 
     // collect results
-    ch_results_genewise = LIMMA_DIFFERENTIAL.out.results
-                            .join(ch_limma.pathway).map(correct_meta_data).mix(ch_results_genewise)
-    ch_results_genewise_filtered = FILTER_DIFFTABLE_LIMMA.out.filtered
-                            .join(ch_limma.pathway).map(correct_meta_data).mix(ch_results_genewise_filtered)
+    ch_results_genewise = LIMMA_DIFFERENTIAL.out.results.mix(ch_results_genewise)
+    ch_results_genewise_filtered = FILTER_DIFFTABLE_LIMMA.out.filtered.mix(ch_results_genewise_filtered)
 
     emit:
     results_pairwise          = ch_results_pairwise

subworkflows/local/enrichment/main.nf

@@ -7,17 +7,23 @@ include { GPROFILER2_GOST } from "../../../modules/nf-core/gprofiler2/gost/main.
 
 workflow ENRICHMENT {
     take:
-    ch_tools        // [ pathway_name, enrichment_map ]
-    ch_counts
-    ch_results_genewise
-    ch_results_genewise_filtered
-    ch_adjacency
+    ch_counts                       // [ meta, counts] with meta keys: method, args_cor
+    ch_results_genewise             // [ meta, results] with meta keys: method, args_cor
+    ch_results_genewise_filtered    // [ meta, results] with meta keys: method, args_cor
+    ch_adjacency                    // [ meta, adj_matrix] with meta keys: method, args_cor
     // TODO: add ch_gm when provided by user, etc.
 
     main:
 
     // initialize empty results channels
     ch_enriched = Channel.empty()
+    ch_gmt      = Channel.empty()
+
+    ch_adjacency
+        .branch {
+            grea: it[0]["method"] == "grea"
+        }
+        .set { ch_adjacency }
 
     // ----------------------------------------------------
     // Construct gene set database
@@ -32,21 +38,7 @@ workflow ENRICHMENT {
     // Perform enrichment analysis with GREA
     // ----------------------------------------------------
 
-    ch_adjacency
-        .map { meta, matrix -> [meta.subMap(["pathway_name"]), meta, matrix] }
-        .join(ch_tools, by: [0])
-        .map {
-            pathway_name, meta, matrix, meta_tools ->
-                def new_meta = meta.clone() + meta_tools.clone()
-                [ new_meta, matrix ]
-            }
-        .branch {
-            grea:  it[0]["enr_method"] == "grea"
-            gsea: it[0]["enr_method"] == "gsea"
-        }
-        .set { ch_adjacency }
-
-    GREA(ch_adjacency.grea, ch_gmt.collect()) //currently, ch_gmt.collect() returns an empty channel, so this does not run
+    GREA(ch_adjacency.grea.unique(), ch_gmt.collect())
     ch_enriched = ch_enriched.mix(GREA.out.results)
 
     // ----------------------------------------------------
@@ -72,19 +64,12 @@ workflow ENRICHMENT {
         ch_background = Channel.from(file(params.gprofiler2_background_file, checkIfExists: true))
     }
 
-    // rearrage channel for GPROFILER2_GOST process
+    // rearrange channel for GPROFILER2_GOST process
     ch_gmt = ch_gmt.map { meta, gmt -> gmt }
 
     ch_results_genewise_filtered
-        .map { meta, matrix -> [meta.subMap(["pathway_name"]), meta, matrix] }
-        .join(ch_tools, by: [0])
-        .map {
-            pathway_name, meta, matrix, meta_tools ->
-                def new_meta = meta.clone() + meta_tools.clone()
-                [ new_meta, matrix ]
-        }
-        .filter {
-            it[0].enr_method == "gprofiler2"
+        .branch {
+            grea: it[0]["method"] == "gprofiler2"
         }
         .set { ch_results_genewise_filtered }