GeneLab_RNASeq_Workflow/workflow_code/main.nf at main · nasa/GeneLab_RNASeq_Workflow · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
// main.nf
nextflow.enable.dsl=2

// Command for: 'nextflow run main.nf --version'
if (params.version) {
    println """${workflow.manifest.name}
Workflow Version: ${workflow.manifest.version}"""
    exit 0
}

// Print the pipeline version on start
println """
${workflow.manifest.name}
Workflow Version: ${workflow.manifest.version}
""".stripIndent()

// Debug warning
if (params.limit_samples_to || params.truncate_to || params.force_single_end || params.genome_subsample) {
    println("WARNING: Debugging options enabled!")
    println("Sample limit: ${params.limit_samples_to ?: 'Not set'}")
    println("Read truncation: ${params.truncate_to ? "First ${params.truncate_to} records" : 'Not set'}")
    println("Reference genome subsampling: ${params.genome_subsample ? "Region '${params.genome_subsample}'" : 'Not set'}")
    println("Force single-end analysis: ${params.force_single_end ? 'Yes' : 'No'}")
} else {
    println("No debugging options enabled")
}

include { RNASEQ } from './workflows/rnaseq.nf'
include { RNASEQ_MICROBES } from './workflows/rnaseq_microbes.nf'
include { STAGE_ANALYSIS } from './workflows/stage_analysis.nf'

include { GENERATE_MD5SUMS } from './modules/generate_md5sums.nf'
include { UPDATE_ASSAY_TABLE } from './modules/update_assay_table.nf'
include { PUBLISH_STAGED_ANALYSIS } from './modules/publish_staged_analysis.nf'

ch_dp_tools_plugin = params.dp_tools_plugin ?
    Channel.value(file(params.dp_tools_plugin)) :
    Channel.value(file(params.mode == 'microbes' ?
        "$projectDir/bin/dp_tools__NF_RCP_Bowtie2" :
        "$projectDir/bin/dp_tools__NF_RCP"))

ch_accession = params.accession ? Channel.value(params.accession) : null
ch_runsheet = params.runsheet_path ? Channel.fromPath(params.runsheet_path) : null
ch_isa_archive = params.isa_archive_path ? Channel.fromPath(params.isa_archive_path) : null
ch_reference_table = Channel.value(params.reference_table)
ch_api_url = Channel.value(params.api_url)

ch_truncate_to = Channel.value(params.truncate_to)
ch_genome_subsample = Channel.value(params.genome_subsample)
ch_force_single_end = Channel.value(params.force_single_end)

ch_reference_store_path = Channel.value(params.reference_store_path)
ch_derived_store_path = Channel.value(params.derived_store_path)

// set reference params
ch_reference_source = params.reference_source ? Channel.value(params.reference_source) : null
ch_reference_version = params.reference_version ? Channel.value(params.reference_version) : null
ch_reference_fasta = params.reference_fasta ? Channel.fromPath(params.reference_fasta) : null
ch_reference_gtf = params.reference_gtf ? Channel.fromPath(params.reference_gtf) : null

// Set outdir based on the presence of an accession input.. currently not implemented as nextflow.config's outdir is only used for nextflow run info
//  and ch_outdir is used for all processes' publishDir in order to set either ./results or ./{accession}
ch_outdir = params.outdir ? channel.fromPath(params.outdir, checkIfExists: true) : null

// Main workflows
workflow {
    if (params.mode == 'microbes') {
        RNASEQ_MICROBES(
            ch_outdir,
            ch_dp_tools_plugin,
            ch_reference_table,
            ch_accession,
            ch_isa_archive,
            ch_runsheet,
            ch_api_url,
            ch_force_single_end,
            ch_truncate_to,
            ch_reference_source,
            ch_reference_version,
            ch_reference_fasta,
            ch_reference_gtf,
            ch_reference_store_path,
            ch_derived_store_path
        )
    } else {
        RNASEQ(
            ch_outdir,
            ch_dp_tools_plugin,
            ch_reference_table,
            ch_accession,
            ch_isa_archive,
            ch_runsheet,
            ch_api_url,
            ch_force_single_end,
            ch_truncate_to,
            ch_reference_source,
            ch_reference_version,
            ch_reference_fasta,
            ch_reference_gtf,
            ch_reference_store_path,
            ch_derived_store_path
        )
    }
}

// Workflow that only runs the initial staging steps:
//  Get the runsheet and raw reads (with debug parameters applied if applicable) and publish them to the output directory
workflow STAGE_ONLY {
    main:
        STAGE_ANALYSIS(
            ch_outdir,
            ch_dp_tools_plugin,
            ch_accession,
            ch_isa_archive,
            ch_runsheet,
            ch_api_url
        )
        PUBLISH_STAGED_ANALYSIS(
            STAGE_ANALYSIS.out.ch_outdir,
            STAGE_ANALYSIS.out.runsheet_path,
            STAGE_ANALYSIS.out.raw_reads.map { it -> it[1] }.collect()
        )
}

workflow POST_PROCESSING {
  main:
    ch_processed_directory = Channel.fromPath("${ params.outdir }/${ params.accession }", checkIfExists: true)
    ch_runsheet = Channel.fromPath("${ params.outdir }/${ params.accession }/Metadata/*_runsheet.csv", checkIfExists: true)
    UPDATE_ASSAY_TABLE(ch_processed_directory)
    GENERATE_MD5SUMS(ch_processed_directory)
}