Merge pull request #1319 from nf-core/reinstate_oncomplete

Reinstate oncomplete error messages

CHANGELOG.md

@@ -72,6 +72,7 @@ Thank you to everyone else that has contributed by reporting bugs, enhancements
 - [PR #1302](https://github.com/nf-core/rnaseq/pull/1302) - Add missing files from Tximport processing
 - [PR #1304](https://github.com/nf-core/rnaseq/pull/1304) - Remove redundant gene TPM outputs
 - [PR #1317](https://github.com/nf-core/rnaseq/pull/1317) - Strip problematic ifEmpty()
+- [PR #1319](https://github.com/nf-core/rnaseq/pull/1319) - Reinstate oncomplete error messages
 
 ### Parameters
 

main.nf

@@ -124,6 +124,9 @@ workflow NFCORE_RNASEQ {
     ch_versions = ch_versions.mix(RNASEQ.out.versions)
 
     emit:
+    trim_status    = RNASEQ.out.trim_status    // channel: [id, boolean]
+    map_status     = RNASEQ.out.map_status     // channel: [id, boolean]
+    strand_status  = RNASEQ.out.strand_status  // channel: [id, boolean]
     multiqc_report = RNASEQ.out.multiqc_report // channel: /path/to/multiqc_report.html
     versions       = ch_versions               // channel: [version1, version2, ...]
 }
@@ -167,7 +170,10 @@ workflow {
         params.outdir,
         params.monochrome_logs,
         params.hook_url,
-        NFCORE_RNASEQ.out.multiqc_report
+        NFCORE_RNASEQ.out.multiqc_report,
+        NFCORE_RNASEQ.out.trim_status,
+        NFCORE_RNASEQ.out.map_status,
+        NFCORE_RNASEQ.out.strand_status
     )
 }
 

subworkflows/local/utils_nfcore_rnaseq_pipeline/main.nf

@@ -14,12 +14,12 @@ include { UTILS_NFVALIDATION_PLUGIN } from '../../nf-core/utils_nfvalidation_plu
 include { paramsSummaryMap          } from 'plugin/nf-validation'
 include { UTILS_NEXTFLOW_PIPELINE   } from '../../nf-core/utils_nextflow_pipeline'
 include { completionEmail           } from '../../nf-core/utils_nfcore_pipeline'
-include { completionSummary         } from '../../nf-core/utils_nfcore_pipeline'
 include { dashedLine                } from '../../nf-core/utils_nfcore_pipeline'
 include { nfCoreLogo                } from '../../nf-core/utils_nfcore_pipeline'
 include { imNotification            } from '../../nf-core/utils_nfcore_pipeline'
 include { UTILS_NFCORE_PIPELINE     } from '../../nf-core/utils_nfcore_pipeline'
 include { workflowCitation          } from '../../nf-core/utils_nfcore_pipeline'
+include { logColours                } from '../../nf-core/utils_nfcore_pipeline'
 
 /*
 ========================================================================================
@@ -85,22 +85,44 @@ workflow PIPELINE_INITIALISATION {
 ========================================================================================
 */
 
+def pass_mapped_reads  = [:]
+def pass_trimmed_reads = [:]
+def pass_strand_check  = [:]
+
 workflow PIPELINE_COMPLETION {
 
     take:
-    schema          //  string: Path to the JSON schema file
-    email           //  string: email address
-    email_on_fail   //  string: email address sent on pipeline failure
-    plaintext_email // boolean: Send plain-text email instead of HTML
-    outdir          //    path: Path to output directory where results will be published
-    monochrome_logs // boolean: Disable ANSI colour codes in log output
-    hook_url        //  string: hook URL for notifications
-    multiqc_report  //  string: Path to MultiQC report
+    schema             //  string: Path to the JSON schema file
+    email              //  string: email address
+    email_on_fail      //  string: email address sent on pipeline failure
+    plaintext_email    // boolean: Send plain-text email instead of HTML
+    outdir             //    path: Path to output directory where results will be published
+    monochrome_logs    // boolean: Disable ANSI colour codes in log output
+    hook_url           //  string: hook URL for notifications
+    multiqc_report     //  string: Path to MultiQC report
+    trim_status        // map: pass/fail status per sample for trimming
+    map_status         // map: pass/fail status per sample for mapping
+    strand_status      // map: pass/fail status per sample for strandedness check
 
     main:
 
     summary_params = paramsSummaryMap(workflow, parameters_schema: schema)
 
+    trim_status
+        .map{
+            id, status -> pass_trimmed_reads[id] = status
+        }
+
+    map_status
+        .map{
+            id, status -> pass_mapped_reads[id] = status
+        }
+
+    strand_status
+        .map{
+            id, status -> pass_strand_check[id] = status
+        }
+
     //
     // Completion email and summary
     //
@@ -109,7 +131,7 @@ workflow PIPELINE_COMPLETION {
             completionEmail(summary_params, email, email_on_fail, plaintext_email, outdir, monochrome_logs, multiqc_report.toList())
         }
 
-        completionSummary(monochrome_logs)
+        rnaseqSummary(monochrome_logs=monochrome_logs, pass_mapped_reads=pass_mapped_reads, pass_trimmed_reads=pass_trimmed_reads, pass_strand_check=pass_strand_check)
 
         if (hook_url) {
             imNotification(summary_params, hook_url)
@@ -571,3 +593,38 @@ def getInferexperimentStrandedness(inferexperiment_file, cutoff=30) {
     }
     return [ strandedness, sense, antisense, undetermined ]
 }
+
+//
+// Print pipeline summary on completion
+//
+def rnaseqSummary(monochrome_logs=true, pass_mapped_reads=[:], pass_trimmed_reads=[:], pass_strand_check=[:]) {
+    def colors = logColours(monochrome_logs)
+
+    def fail_mapped_count  = pass_mapped_reads.count  { key, value -> value == false }
+    def fail_trimmed_count = pass_trimmed_reads.count { key, value -> value == false }
+    def fail_strand_count  = pass_strand_check.count  { key, value -> value == false }
+    if (workflow.success) {
+        def color = colors.green
+        def status = []
+        if (workflow.stats.ignoredCount != 0) {
+            color = colors.yellow
+            status += ['with errored process(es)']
+        }
+        if (fail_mapped_count > 0 || fail_trimmed_count > 0) {
+            color = colors.yellow
+            status += ['with skipped sampl(es)']
+        }
+        log.info "-${colors.purple}[$workflow.manifest.name]${color} Pipeline completed successfully ${status.join(', ')}${colors.reset}-"
+        if (fail_trimmed_count > 0) {
+            log.info "-${colors.purple}[$workflow.manifest.name]${colors.red} Please check MultiQC report: ${fail_trimmed_count}/${pass_trimmed_reads.size()} samples skipped since they failed ${params.min_trimmed_reads} trimmed read threshold.${colors.reset}-"
+        }
+        if (fail_mapped_count > 0) {
+            log.info "-${colors.purple}[$workflow.manifest.name]${colors.red} Please check MultiQC report: ${fail_mapped_count}/${pass_mapped_reads.size()} samples skipped since they failed STAR ${params.min_mapped_reads}% mapped threshold.${colors.reset}-"
+        }
+        if (fail_strand_count > 0) {
+            log.info "-${colors.purple}[$workflow.manifest.name]${colors.red} Please check MultiQC report: ${fail_strand_count}/${pass_strand_check.size()} samples failed strandedness check.${colors.reset}-"
+        }
+    } else {
+        log.info "-${colors.purple}[$workflow.manifest.name]${colors.red} Pipeline completed with errors${colors.reset}-"
+    }
+}

subworkflows/local/utils_nfcore_rnaseq_pipeline/tests/main.pipeline_completion.workflow.nf.test

@@ -18,6 +18,9 @@ nextflow_workflow {
                 input[5] = true // monochrome_logs (boolean)
                 input[6] = null // hook_url (string)
                 input[7] = "${outputDir}/multiqc_report.html" // multiqc_report (string)
+                input[8] = Channel.of(['test_sample', true])
+                input[9] = Channel.of(['test_sample', true])
+                input[10] = Channel.of(['test_sample', true])
                 """
             }
         }

workflows/rnaseq/main.nf

@@ -106,6 +106,9 @@ workflow RNASEQ {
     main:
 
     ch_multiqc_files = Channel.empty()
+    ch_trim_status = Channel.empty()
+    ch_map_status = Channel.empty()
+    ch_strand_status = Channel.empty()
 
     //
     // Create channel from input file provided through params.input
@@ -191,6 +194,13 @@ workflow RNASEQ {
         ch_versions = ch_versions.mix(FASTQ_FASTQC_UMITOOLS_FASTP.out.versions)
     }
 
+    // Save trim status for workflow summary
+
+    ch_trim_status = ch_trim_read_count
+        .map {
+            meta, num_reads ->
+                return [ meta.id, num_reads > params.min_trimmed_reads.toFloat() ]
+        }
     //
     // Get list of samples that failed trimming threshold for MultiQC report
     //
@@ -528,6 +538,13 @@ workflow RNASEQ {
             .map { meta, align_log -> [ meta ] + getStarPercentMapped(params, align_log) }
             .set { ch_percent_mapped }
 
+        // Save status for workflow summary
+        ch_map_status = ch_percent_mapped
+            .map {
+                meta, mapped, pass ->
+                    return [ meta.id, pass ]
+            }
+
         ch_genome_bam
             .join(ch_percent_mapped, by: [0])
             .map { meta, ofile, mapped, pass -> if (pass) [ meta, ofile ] }
@@ -718,12 +735,25 @@ workflow RNASEQ {
             ch_multiqc_files = ch_multiqc_files.mix(BAM_RSEQC.out.tin_txt.collect{it[1]})
             ch_versions = ch_versions.mix(BAM_RSEQC.out.versions)
 
-            BAM_RSEQC
+            ch_inferexperiment_strand = BAM_RSEQC
                 .out
                 .inferexperiment_txt
                 .map {
                     meta, strand_log ->
                         def inferred_strand = getInferexperimentStrandedness(strand_log, 30)
+                        return [ meta, inferred_strand ]
+                }
+
+            // Save status for workflow summary
+            ch_strand_status = ch_inferexperiment_strand
+                .map{
+                    meta, inferred_strand ->
+                        return [ meta.id, meta.strandedness == inferred_strand[0]]
+                }
+
+            ch_inferexperiment_strand
+                .map{
+                    meta, inferred_strand ->
                         if (meta.strandedness != inferred_strand[0]) {
                             return [ "$meta.id\t$meta.strandedness\t${inferred_strand.join('\t')}" ]
                         }
@@ -817,6 +847,9 @@ workflow RNASEQ {
     }
 
     emit:
+    trim_status    = ch_trim_status    // channel: [id, boolean]
+    map_status     = ch_map_status     // channel: [id, boolean]
+    strand_status  = ch_strand_status  // channel: [id, boolean]
     multiqc_report = ch_multiqc_report // channel: /path/to/multiqc_report.html
     versions       = ch_versions       // channel: [ path(versions.yml) ]
 }