kkonganti@101: // Define any required imports for this specific workflow
kkonganti@101: import java.nio.file.Paths
kkonganti@101: import nextflow.file.FileHelper
kkonganti@101: 
kkonganti@101: // Include any necessary methods
kkonganti@101: include { \
kkonganti@101:     summaryOfParams; stopNow; fastqEntryPointHelp; sendMail; \
kkonganti@101:     addPadding; wrapUpHelp   } from "${params.routines}"
kkonganti@101: include { seqkitrmdupHelp    } from "${params.toolshelp}${params.fs}seqkitrmdup"
kkonganti@101: include { kraken2Help        } from "${params.toolshelp}${params.fs}kraken2"
kkonganti@101: include { centrifugeHelp     } from "${params.toolshelp}${params.fs}centrifuge"
kkonganti@101: include { megahitHelp        } from "${params.toolshelp}${params.fs}megahit"
kkonganti@101: include { spadesHelp         } from "${params.toolshelp}${params.fs}spades"
kkonganti@101: include { serotypefinderHelp } from "${params.toolshelp}${params.fs}serotypefinder"
kkonganti@101: include { seqsero2Help       } from "${params.toolshelp}${params.fs}seqsero2"
kkonganti@101: include { mlstHelp           } from "${params.toolshelp}${params.fs}mlst"
kkonganti@101: include { abricateHelp       } from "${params.toolshelp}${params.fs}abricate"
kkonganti@101: 
kkonganti@101: // Exit if help requested before any subworkflows
kkonganti@101: if (params.help) {
kkonganti@101:     log.info help()
kkonganti@101:     exit 0
kkonganti@101: }
kkonganti@101: 
kkonganti@101: // Include any necessary modules and subworkflows
kkonganti@101: include { PROCESS_FASTQ           } from "${params.subworkflows}${params.fs}process_fastq"
kkonganti@101: include { FASTQC                  } from "${params.modules}${params.fs}fastqc${params.fs}main"
kkonganti@101: include { SEQKIT_RMDUP            } from "${params.modules}${params.fs}seqkit${params.fs}rmdup${params.fs}main"
kkonganti@101: include { CENTRIFUGE_CLASSIFY     } from "${params.modules}${params.fs}centrifuge${params.fs}classify${params.fs}main"
kkonganti@101: include { CENTRIFUGE_PROCESS      } from "${params.modules}${params.fs}centrifuge${params.fs}process${params.fs}main"
kkonganti@101: include { SEQKIT_GREP             } from "${params.modules}${params.fs}seqkit${params.fs}grep${params.fs}main"
kkonganti@101: include { MEGAHIT_ASSEMBLE        } from "${params.modules}${params.fs}megahit${params.fs}assemble${params.fs}main"
kkonganti@101: include { SPADES_ASSEMBLE         } from "${params.modules}${params.fs}spades${params.fs}assemble${params.fs}main"
kkonganti@101: include { KRAKEN2_CLASSIFY        } from "${params.modules}${params.fs}kraken2${params.fs}classify${params.fs}main"
kkonganti@101: include { KRAKEN2_EXTRACT_CONTIGS } from "${params.modules}${params.fs}kraken2${params.fs}extract_contigs${params.fs}main"
kkonganti@101: include { SEROTYPEFINDER          } from "${params.modules}${params.fs}serotypefinder${params.fs}main"
kkonganti@101: include { SEQSERO2                } from "${params.modules}${params.fs}seqsero2${params.fs}main"
kkonganti@101: include { MLST                    } from "${params.modules}${params.fs}mlst${params.fs}main"
kkonganti@101: include { ABRICATE_RUN            } from "${params.modules}${params.fs}abricate${params.fs}run${params.fs}main"
kkonganti@101: include { ABRICATE_SUMMARY        } from "${params.modules}${params.fs}abricate${params.fs}summary${params.fs}main"
kkonganti@101: include { TABLE_SUMMARY           } from "${params.modules}${params.fs}cat${params.fs}tables${params.fs}main"
kkonganti@101: include { MULTIQC                 } from "${params.modules}${params.fs}multiqc${params.fs}main"
kkonganti@101: include { DUMP_SOFTWARE_VERSIONS  } from "${params.modules}${params.fs}custom${params.fs}dump_software_versions${params.fs}main"
kkonganti@101: 
kkonganti@101: 
kkonganti@101: 
kkonganti@101: /*
kkonganti@101: ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
kkonganti@101:     INPUTS AND ANY CHECKS FOR THE CENTRIFLAKEN-HY WORKFLOW
kkonganti@101: ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
kkonganti@101: */
kkonganti@101: 
kkonganti@101: def kraken2_db_dir = file ( "${params.kraken2_db}" )
kkonganti@101: def centrifuge_x = file ( "${params.centrifuge_x}" )
kkonganti@101: def spades_custom_hmm = (params.spades_hmm ? file ( "${params.spades_hmm}" ) : false)
kkonganti@101: def reads_platform = 0
kkonganti@101: def abricate_dbs = [ 'ncbiamrplus', 'resfinder', 'megares', 'argannot' ]
kkonganti@101: 
kkonganti@101: reads_platform += (params.input ? 1 : 0)
kkonganti@101: 
kkonganti@101: if (!kraken2_db_dir.exists() || !centrifuge_x.getParent().exists()) {
kkonganti@101:     stopNow("Please check if the following absolute paths are valid:\n" +
kkonganti@101:         "${params.kraken2_db}\n${params.centrifuge_x}\n" +
kkonganti@101:         "Cannot proceed further!")
kkonganti@101: }
kkonganti@101: 
kkonganti@101: if (spades_custom_hmm && !spades_custom_hmm.exists()) {
kkonganti@101:     stopNow("Please check if the following SPAdes' custom HMM directory\n" +
kkonganti@101:         "path is valid:\n${params.spades_hmm}\nCannot proceed further!")
kkonganti@101: }
kkonganti@101: 
kkonganti@101: if (reads_platform < 1 || reads_platform == 0) {
kkonganti@101:     stopNow("Please mention at least one absolute path to input folder which contains\n" +
kkonganti@101:             "FASTQ files sequenced using the --input option.\n" +
kkonganti@101:         "Ex: --input (Illumina or Generic short reads in FASTQ format)")
kkonganti@101: }
kkonganti@101: 
kkonganti@101: if (params.centrifuge_extract_bug != params.kraken2_extract_bug) {
kkonganti@101:     stopNow("Please make sure that the bug to be extracted is same\n" +
kkonganti@101:     "for both --centrifuge_extract_bug and --kraken2_extract_bug options.")
kkonganti@101: }
kkonganti@101: 
kkonganti@101: /*
kkonganti@101: ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
kkonganti@101:     RUN THE CENTRIFLAKEN-HY WORKFLOW
kkonganti@101: ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
kkonganti@101: */
kkonganti@101: 
kkonganti@101: workflow CENTRIFLAKEN_HY {
kkonganti@101:     main:
kkonganti@101:         ch_asm_filtered_contigs = Channel.empty()
kkonganti@101:         ch_mqc_custom_tbl = Channel.empty()
kkonganti@101:         ch_dummy = Channel.fromPath("${params.dummyfile}")
kkonganti@101:         ch_dummy2 = Channel.fromPath("${params.dummyfile2}")
kkonganti@101: 
kkonganti@101:         log.info summaryOfParams()
kkonganti@101: 
kkonganti@101:         PROCESS_FASTQ()
kkonganti@101:             .processed_reads
kkonganti@101:             .map {
kkonganti@101:                 meta, fastq ->
kkonganti@101:                 meta.centrifuge_x = params.centrifuge_x
kkonganti@101:                 meta.kraken2_db = params.kraken2_db
kkonganti@101:                 [meta, fastq]
kkonganti@101:             }
kkonganti@101:             .set { ch_processed_reads }
kkonganti@101: 
kkonganti@101:         PROCESS_FASTQ
kkonganti@101:             .out
kkonganti@101:             .versions
kkonganti@101:             .set { software_versions }
kkonganti@101: 
kkonganti@101:         FASTQC ( ch_processed_reads )
kkonganti@101: 
kkonganti@101:         if (params.seqkit_rmdup_run) {
kkonganti@101:             SEQKIT_RMDUP ( ch_processed_reads )
kkonganti@101: 
kkonganti@101:             SEQKIT_RMDUP
kkonganti@101:                 .out
kkonganti@101:                 .fastx
kkonganti@101:                 .set { ch_processed_reads }
kkonganti@101: 
kkonganti@101:             software_versions
kkonganti@101:                 .mix ( SEQKIT_RMDUP.out.versions.ifEmpty(null) )
kkonganti@101:                 .set { software_versions }
kkonganti@101:         }
kkonganti@101: 
kkonganti@101:         CENTRIFUGE_CLASSIFY ( ch_processed_reads )
kkonganti@101: 
kkonganti@101:         CENTRIFUGE_PROCESS (
kkonganti@101:             CENTRIFUGE_CLASSIFY.out.report
kkonganti@101:                 .join( CENTRIFUGE_CLASSIFY.out.output )
kkonganti@101:         )
kkonganti@101: 
kkonganti@101:         ch_processed_reads.join ( CENTRIFUGE_PROCESS.out.extracted )
kkonganti@101:             .set { ch_centrifuge_extracted }
kkonganti@101: 
kkonganti@101:         SEQKIT_GREP ( ch_centrifuge_extracted )
kkonganti@101: 
kkonganti@101:         // As of 06/02/2022, with the upcoming newer versions of NextFlow, we will be able to do
kkonganti@101:         // allowNull: true for both input and output, but until then, we have to use dummy files.
kkonganti@101:         // and work arounds.
kkonganti@101:         // https://github.com/nextflow-io/nextflow/pull/2893
kkonganti@101:         if (params.spades_run) {
kkonganti@101:             SPADES_ASSEMBLE (
kkonganti@101:                 SEQKIT_GREP.out.fastx
kkonganti@101:                     .combine(ch_dummy)
kkonganti@101:                     .combine(ch_dummy2)
kkonganti@101:             )
kkonganti@101: 
kkonganti@101:             SPADES_ASSEMBLE
kkonganti@101:                 .out
kkonganti@101:                 .assembly
kkonganti@101:                 .set { ch_assembly }
kkonganti@101: 
kkonganti@101:             software_versions
kkonganti@101:                 .mix ( SPADES_ASSEMBLE.out.versions.ifEmpty(null) )
kkonganti@101:                 .set { software_versions }
kkonganti@101:         } else if (params.megahit_run) {
kkonganti@101:             MEGAHIT_ASSEMBLE (
kkonganti@101:                 SEQKIT_GREP.out.fastx
kkonganti@101:             )
kkonganti@101: 
kkonganti@101:             MEGAHIT_ASSEMBLE
kkonganti@101:                 .out
kkonganti@101:                 .assembly
kkonganti@101:                 .set { ch_assembly }
kkonganti@101: 
kkonganti@101:             software_versions
kkonganti@101:                 .mix ( MEGAHIT_ASSEMBLE.out.versions.ifEmpty(null) )
kkonganti@101:                 .set { software_versions }
kkonganti@101:         }
kkonganti@101: 
kkonganti@101:         ch_assembly
kkonganti@101:             .map {
kkonganti@101:                 meta, fastq ->
kkonganti@101:                 meta.is_assembly = true
kkonganti@101:                 [meta, fastq]
kkonganti@101:             }
kkonganti@101:             .set { ch_assembly }
kkonganti@101: 
kkonganti@101:         ch_assembly.ifEmpty { [ false, false ] }
kkonganti@101: 
kkonganti@101:         KRAKEN2_CLASSIFY ( ch_assembly )
kkonganti@101: 
kkonganti@101:         KRAKEN2_EXTRACT_CONTIGS (
kkonganti@101:             ch_assembly
kkonganti@101:                 .join( KRAKEN2_CLASSIFY.out.kraken_output ),
kkonganti@101:             params.kraken2_extract_bug
kkonganti@101:         )
kkonganti@101: 
kkonganti@101:         KRAKEN2_EXTRACT_CONTIGS
kkonganti@101:             .out
kkonganti@101:             .asm_filtered_contigs
kkonganti@101:             .map {
kkonganti@101:                 meta, fastq ->
kkonganti@101:                 meta.organism = params.kraken2_extract_bug.split(/\s+/)[0].capitalize()
kkonganti@101:                 meta.serotypefinder_db = params.serotypefinder_db
kkonganti@101:                 [meta, fastq]
kkonganti@101:             }
kkonganti@101:             .set { ch_asm_filtered_contigs }
kkonganti@101: 
kkonganti@101:         SEROTYPEFINDER ( ch_asm_filtered_contigs )
kkonganti@101: 
kkonganti@101:         SEQSERO2 ( ch_asm_filtered_contigs )
kkonganti@101: 
kkonganti@101:         MLST ( ch_asm_filtered_contigs )
kkonganti@101: 
kkonganti@101:         ABRICATE_RUN ( 
kkonganti@101:             ch_asm_filtered_contigs, 
kkonganti@101:             abricate_dbs
kkonganti@101:         )
kkonganti@101: 
kkonganti@101:         ABRICATE_RUN
kkonganti@101:             .out
kkonganti@101:             .abricated
kkonganti@101:             .map { meta, abres -> [ abricate_dbs, abres ] }
kkonganti@101:             .groupTuple(by: [0])
kkonganti@101:             .map { it -> tuple ( it[0], it[1].flatten() ) }
kkonganti@101:             .set { ch_abricated }
kkonganti@101: 
kkonganti@101:         ABRICATE_SUMMARY ( ch_abricated )
kkonganti@101: 
kkonganti@101:         CENTRIFUGE_CLASSIFY.out.kreport
kkonganti@101:             .map { meta, kreport -> [ kreport ] }
kkonganti@101:             .flatten()
kkonganti@101:             .concat ( 
kkonganti@101:                 KRAKEN2_CLASSIFY.out.kraken_report
kkonganti@101:                 .map { meta, kreport -> [ kreport ] }
kkonganti@101:                 .flatten(),
kkonganti@101:                 FASTQC.out.zip
kkonganti@101:                 .map { meta, zip -> [ zip ] }
kkonganti@101:                 .flatten()
kkonganti@101:             )
kkonganti@101:             .set { ch_mqc_classify }
kkonganti@101: 
kkonganti@101:         if (params.serotypefinder_run) {
kkonganti@101:             SEROTYPEFINDER
kkonganti@101:                 .out
kkonganti@101:                 .serotyped
kkonganti@101:                 .map { meta, tsv -> [ 'serotypefinder', tsv ] }
kkonganti@101:                 .groupTuple(by: [0])
kkonganti@101:                 .map { it -> tuple ( it[0], it[1].flatten() ) }
kkonganti@101:                 .set { ch_mqc_custom_tbl }
kkonganti@101:         } else if (params.seqsero2_run) {
kkonganti@101:             SEQSERO2
kkonganti@101:                 .out
kkonganti@101:                 .serotyped
kkonganti@101:                 .map { meta, tsv -> [ 'seqsero2', tsv ] }
kkonganti@101:                 .groupTuple(by: [0])
kkonganti@101:                 .map { it -> tuple ( it[0], it[1].flatten() ) }
kkonganti@101:                 .set { ch_mqc_custom_tbl }
kkonganti@101:         }
kkonganti@101: 
kkonganti@101:         ch_mqc_custom_tbl
kkonganti@101:             .concat (
kkonganti@101:                 ABRICATE_SUMMARY.out.ncbiamrplus.map{ it -> tuple ( it[0], it[1] )},
kkonganti@101:                 ABRICATE_SUMMARY.out.resfinder.map{ it -> tuple ( it[0], it[1] )},
kkonganti@101:                 ABRICATE_SUMMARY.out.megares.map{ it -> tuple ( it[0], it[1] )},
kkonganti@101:                 ABRICATE_SUMMARY.out.argannot.map{ it -> tuple ( it[0], it[1] )},
kkonganti@101:             )
kkonganti@101:             .groupTuple(by: [0])
kkonganti@101:             .map { it -> [ it[0], it[1].flatten() ]}
kkonganti@101:             .set { ch_mqc_custom_tbl }
kkonganti@101: 
kkonganti@101:         TABLE_SUMMARY ( ch_mqc_custom_tbl )
kkonganti@101: 
kkonganti@101:         DUMP_SOFTWARE_VERSIONS (
kkonganti@101:             software_versions
kkonganti@101:                 .mix (
kkonganti@101:                     FASTQC.out.versions,
kkonganti@101:                     CENTRIFUGE_CLASSIFY.out.versions,
kkonganti@101:                     CENTRIFUGE_PROCESS.out.versions,
kkonganti@101:                     SEQKIT_GREP.out.versions,
kkonganti@101:                     KRAKEN2_CLASSIFY.out.versions.ifEmpty(null),
kkonganti@101:                     KRAKEN2_EXTRACT_CONTIGS.out.versions.ifEmpty(null),
kkonganti@101:                     SEROTYPEFINDER.out.versions.ifEmpty(null),
kkonganti@101:                     SEQSERO2.out.versions.ifEmpty(null),
kkonganti@101:                     MLST.out.versions.ifEmpty(null),
kkonganti@101:                     ABRICATE_RUN.out.versions.ifEmpty(null),
kkonganti@101:                     ABRICATE_SUMMARY.out.versions.ifEmpty(null),
kkonganti@101:                     TABLE_SUMMARY.out.versions.ifEmpty(null)
kkonganti@101:                 )
kkonganti@101:                 .unique()
kkonganti@101:                 .collectFile(name: 'collected_versions.yml')
kkonganti@101:         )
kkonganti@101: 
kkonganti@101:         DUMP_SOFTWARE_VERSIONS
kkonganti@101:             .out
kkonganti@101:             .mqc_yml
kkonganti@101:             .concat (
kkonganti@101:                 ch_mqc_classify,
kkonganti@101:                 TABLE_SUMMARY.out.mqc_yml
kkonganti@101:             )
kkonganti@101:             .collect()
kkonganti@101:             .set { ch_multiqc }
kkonganti@101: 
kkonganti@101:         MULTIQC ( ch_multiqc )
kkonganti@101: }
kkonganti@101: 
kkonganti@101: /*
kkonganti@101: ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
kkonganti@101:     ON COMPLETE, SHOW GORY DETAILS OF ALL PARAMS WHICH WILL BE HELPFUL TO DEBUG
kkonganti@101: ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
kkonganti@101: */
kkonganti@101: 
kkonganti@101: workflow.onComplete {
kkonganti@101:     if (workflow.success) {
kkonganti@101:         // CREATE APPROPRIATE DIRECTORIES AND MOVE AS REQUESTED BY STAKEHOLDER(S)
kkonganti@101:         //
kkonganti@101:         // Nextflow's .moveTo will error out if directories contain files and it
kkonganti@101:         // would be complex to include logic to skip directories 
kkonganti@101:         //
kkonganti@101:         def final_intermediate_dir = "${params.output}${params.fs}${params.pipeline}-steps"
kkonganti@101:         def final_results_dir = "${params.output}${params.fs}${params.pipeline}-results"
kkonganti@101:         def kraken2_ext_contigs =  file( "${final_intermediate_dir}${params.fs}kraken2_extract_contigs", type: 'dir' )
kkonganti@101:         def final_intermediate = file( final_intermediate_dir, type: 'dir' )
kkonganti@101:         def final_results = file( final_results_dir, type: 'dir' )
kkonganti@101:         def pipeline_output = file( params.output, type: 'dir' )
kkonganti@101: 
kkonganti@101:         if ( !final_intermediate.exists() ) {
kkonganti@101:             final_intermediate.mkdirs()
kkonganti@101: 
kkonganti@101:             FileHelper.visitFiles(Paths.get("${params.output}"), '*') {
kkonganti@101:                 if ( !(it.name ==~ /^(${params.cfsanpipename}|multiqc|\.nextflow|${workflow.workDir.name}|${params.pipeline}).*/) ) {
kkonganti@101:                     FileHelper.movePath(
kkonganti@101:                         it, Paths.get( "${final_intermediate_dir}${params.fs}${it.name}" )
kkonganti@101:                     )
kkonganti@101:                 }
kkonganti@101:             }
kkonganti@101:         }
kkonganti@101: 
kkonganti@101:         if ( kraken2_ext_contigs.exists() && !final_results.exists() ) {
kkonganti@101:             final_results.mkdirs()
kkonganti@101: 
kkonganti@101:             FileHelper.movePath(
kkonganti@101:                 Paths.get( "${final_intermediate_dir}${params.fs}kraken2_extract_contigs" ),
kkonganti@101:                 Paths.get( "${final_results_dir}${params.fs}kraken2_extract_contigs" )
kkonganti@101:             )
kkonganti@101:         }
kkonganti@101: 
kkonganti@101:         sendMail()
kkonganti@101:     }
kkonganti@101: }
kkonganti@101: 
kkonganti@101: workflow.onError {
kkonganti@101:     sendMail()
kkonganti@101: }
kkonganti@101: 
kkonganti@101: /*
kkonganti@101: ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
kkonganti@101:     HELPER METHODS FOR CENTRIFLAKEN-HY WORKFLOW
kkonganti@101: ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
kkonganti@101: */
kkonganti@101: 
kkonganti@101: def help() {
kkonganti@101: 
kkonganti@101:     Map helptext = [:]
kkonganti@101: 
kkonganti@101:     helptext.putAll (
kkonganti@101:         fastqEntryPointHelp() +
kkonganti@101:         seqkitrmdupHelp(params).text +
kkonganti@101:         kraken2Help(params).text +
kkonganti@101:         centrifugeHelp(params).text +
kkonganti@101:         megahitHelp(params).text +
kkonganti@101:         spadesHelp(params).text +
kkonganti@101:         serotypefinderHelp(params).text +
kkonganti@101:         seqsero2Help(params).text +
kkonganti@101:         mlstHelp(params).text +
kkonganti@101:         abricateHelp(params).text +
kkonganti@101:         wrapUpHelp()
kkonganti@101:     )
kkonganti@101: 
kkonganti@101:     return addPadding(helptext)
kkonganti@101: }