kkonganti@0: // Include any necessary methods and modules
kkonganti@0: include { stopNow; validateParamsForFASTQ } from "${params.routines}"
kkonganti@0: include { GEN_SAMPLESHEET                 } from "${params.modules}${params.fs}gen_samplesheet${params.fs}main"
kkonganti@0: include { SAMPLESHEET_CHECK               } from "${params.modules}${params.fs}samplesheet_check${params.fs}main"
kkonganti@0: include { CAT_FASTQ                       } from "${params.modules}${params.fs}cat${params.fs}fastq${params.fs}main"
kkonganti@0: include { SEQKIT_SEQ                      } from "${params.modules}${params.fs}seqkit${params.fs}seq${params.fs}main"
kkonganti@0: 
kkonganti@0: // Validate 4 required workflow parameters if
kkonganti@0: // FASTQ files are the input for the
kkonganti@0: // entry point.
kkonganti@0: validateParamsForFASTQ()
kkonganti@0: 
kkonganti@0: // Start the subworkflow
kkonganti@0: workflow PROCESS_FASTQ {
kkonganti@0:     main:
kkonganti@0:         versions = Channel.empty()
kkonganti@0:         input_ch = Channel.empty()
kkonganti@0:         reads = Channel.empty()
kkonganti@0: 
kkonganti@0:         def input = file( (params.input ?: params.metadata) )
kkonganti@0: 
kkonganti@0:         if (params.input) {
kkonganti@0:             def fastq_files = []
kkonganti@0: 
kkonganti@0:             if (params.fq_suffix == null) {
kkonganti@0:             stopNow("We need to know what suffix the FASTQ files ends with inside the\n" +
kkonganti@0:                 "directory. Please use the --fq_suffix option to indicate the file\n" +
kkonganti@0:                 "suffix by which the files are to be collected to run the pipeline on.")
kkonganti@0:             }
kkonganti@0: 
kkonganti@0:             if (params.fq_strandedness == null) {
kkonganti@0:                 stopNow("We need to know if the FASTQ files inside the directory\n" +
kkonganti@0:                     "are sequenced using stranded or non-stranded sequencing. This is generally\n" +
kkonganti@0:                     "required if the sequencing experiment is RNA-SEQ. For almost all of the other\n" +
kkonganti@0:                     "cases, you can probably use the --fq_strandedness unstranded option to indicate\n" +
kkonganti@0:                     "that the reads are unstranded.")
kkonganti@0:             }
kkonganti@0: 
kkonganti@0:             if (params.fq_filename_delim == null || params.fq_filename_delim_idx == null) {
kkonganti@0:                 stopNow("We need to know the delimiter of the filename of the FASTQ files.\n" +
kkonganti@0:                     "By default the filename delimiter is _ (underscore). This delimiter character\n" +
kkonganti@0:                     "is used to split and assign a group name. The group name can be controlled by\n" +
kkonganti@0:                     "using the --fq_filename_delim_idx option (1-based). For example, if the FASTQ\n" +
kkonganti@0:                     "filename is WT_REP1_001.fastq, then to create a group WT, use the following\n" +
kkonganti@0:                     "options: --fq_filename_delim _ --fq_filename_delim_idx 1")
kkonganti@0:             }
kkonganti@0: 
kkonganti@0:             if (!input.exists()) {
kkonganti@0:                 stopNow("The input directory,\n${params.input}\ndoes not exist!")
kkonganti@0:             }
kkonganti@0: 
kkonganti@0:             input.eachFileRecurse {
kkonganti@0:                 it.name.endsWith("${params.fq_suffix}") ? fastq_files << it : fastq_files << null
kkonganti@0:             }
kkonganti@0: 
kkonganti@0:             if (fastq_files.findAll{ it != null }.size() == 0) {
kkonganti@0:                 stopNow("The input directory,\n${params.input}\nis empty! or does not " +
kkonganti@0:                     "have FASTQ files ending with the suffix: ${params.fq_suffix}")
kkonganti@0:             }
kkonganti@0:             
kkonganti@0:             GEN_SAMPLESHEET( Channel.fromPath(params.input, type: 'dir') )
kkonganti@0:             GEN_SAMPLESHEET.out.csv.set{ input_ch }
kkonganti@0:             versions.mix( GEN_SAMPLESHEET.out.versions )
kkonganti@0:                 .set { versions }
kkonganti@0:         } else if (params.metadata) {
kkonganti@0:             if (!input.exists()) {
kkonganti@0:                 stopNow("The metadata CSV file,\n${params.metadata}\ndoes not exist!")
kkonganti@0:             }
kkonganti@0: 
kkonganti@0:             if (input.size() <= 0) {
kkonganti@0:                 stopNow("The metadata CSV file,\n${params.metadata}\nis empty!")
kkonganti@0:             }
kkonganti@0: 
kkonganti@0:             Channel.fromPath(params.metadata, type: 'file')
kkonganti@0:                 .set { input_ch }
kkonganti@0:         }
kkonganti@0: 
kkonganti@0:         SAMPLESHEET_CHECK( input_ch )
kkonganti@0:             .csv
kkonganti@0:             .splitCsv( header: true, sep: ',')
kkonganti@0:             .map { create_fastq_channel(it) }
kkonganti@0:             .groupTuple(by: [0])
kkonganti@0:             .branch {
kkonganti@0:                 meta, fastq ->
kkonganti@0:                     single   : fastq.size() == 1
kkonganti@0:                         return [ meta, fastq.flatten() ]
kkonganti@0:                     multiple : fastq.size() > 1
kkonganti@0:                         return [ meta, fastq.flatten() ]
kkonganti@0:             }
kkonganti@0:             .set { reads }
kkonganti@0: 
kkonganti@0:         CAT_FASTQ( reads.multiple )
kkonganti@0:             .catted_reads
kkonganti@0:             .mix( reads.single )
kkonganti@0:             .set { processed_reads }
kkonganti@0: 
kkonganti@0:         if (params.fq_filter_by_len.toInteger() > 0) {
kkonganti@0:             SEQKIT_SEQ( processed_reads )
kkonganti@0:                 .fastx
kkonganti@0:                 .set { processed_reads }
kkonganti@0: 
kkonganti@0:             versions.mix( SEQKIT_SEQ.out.versions.first().ifEmpty(null) )
kkonganti@0:                 .set { versions }
kkonganti@0:         }
kkonganti@0: 
kkonganti@0:         versions.mix(
kkonganti@0:             SAMPLESHEET_CHECK.out.versions,
kkonganti@0:             CAT_FASTQ.out.versions.first().ifEmpty(null)
kkonganti@0:         )
kkonganti@0:         .set { versions }
kkonganti@0: 
kkonganti@0:     emit:
kkonganti@0:         processed_reads
kkonganti@0:         versions
kkonganti@0: }
kkonganti@0: 
kkonganti@0: // Function to get list of [ meta, [ fq1, fq2 ] ]
kkonganti@0: def create_fastq_channel(LinkedHashMap row) {
kkonganti@0: 
kkonganti@0:     def meta = [:]
kkonganti@0:     meta.id           = row.sample
kkonganti@0:     meta.single_end   = row.single_end.toBoolean()
kkonganti@0:     meta.strandedness = row.strandedness
kkonganti@0:     meta.id = meta.id.split(params.fq_filename_delim)[0..params.fq_filename_delim_idx.toInteger() - 1]
kkonganti@0:         .join(params.fq_filename_delim)
kkonganti@0:     meta.id = (meta.id =~ /\./ ? meta.id.take(meta.id.indexOf('.')) : meta.id)
kkonganti@0: 
kkonganti@0:     def array = []
kkonganti@0: 
kkonganti@0:     if (!file(row.fq1).exists()) {
kkonganti@0:         stopNow("Please check input metadata CSV. The following Read 1 FASTQ file does not exist!" +
kkonganti@0:             "\n${row.fq1}")
kkonganti@0:     }
kkonganti@0:     if (meta.single_end) {
kkonganti@0:         array = [ meta, [ file(row.fq1) ] ]
kkonganti@0:     } else {
kkonganti@0:         if (!file(row.fq2).exists()) {
kkonganti@0:             stopNow("Please check input metadata CSV. The following Read 2 FASTQ file does not exist!" +
kkonganti@0:                 "\n${row.fq2}")
kkonganti@0:         }
kkonganti@0:         array = [ meta, [ file(row.fq1), file(row.fq2) ] ]
kkonganti@0:     }
kkonganti@0:     return array
kkonganti@0: }