main.nf

#!/usr/bin/env nextflow
// Copyright 2021 Edinburgh Genome Foundry, University of Edinburgh
//
// This file is part of Sequeduct.
//
// Sequeduct is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.
//
// Sequeduct is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License along with Sequeduct. If not, see <https:www.gnu.org/licenses/>.

nextflow.enable.dsl=2


include { demultiplex_workflow } from "$projectDir/nextflow/sequeduct_demultiplex.nf"
include { singleplex_workflow } from "$projectDir/nextflow/sequeduct_demultiplex.nf"
include { combine_samplesheets } from "$projectDir/nextflow/sequeduct_demultiplex.nf"
include { preview_workflow } from "$projectDir/nextflow/sequeduct_preview.nf"
include { analysis_workflow } from "$projectDir/nextflow/sequeduct_analysis.nf"
include { review_denovo } from "$projectDir/nextflow/sequeduct_review.nf"
include { assemble_denovo } from "$projectDir/nextflow/sequeduct_assembly.nf"

///////////////////////////////////////////////////////////////////////////////

workflow demultiplex {
    Channel
        .fromPath(params.sample_sheet)
        .splitCsv(header: true)
        .map { row -> 
            def barcode_dir = row['Barcode_dir']  // a barcode is present only once
            def barcode_path = file("${params.fastq_dir}/${barcode_dir}")
            def fastq_files = barcode_path.listFiles()  // multiple FASTQ in each barcode

            def sample = row['Sample']  // entry may contain multiple samples
            // def entry = "${barcode_dir}_${sample}"  // unique key for each sample sheet entry
            def genbank_paths = sample.tokenize(';')  // separator character
            for (int i = 0; i < genbank_paths.size(); i++) {
                genbank_path = genbank_paths[i]
                genbank_paths[i] = file("${params.reference_dir}/${genbank_path}.gb") 
            }
            return [barcode_dir, barcode_path, fastq_files, sample, genbank_paths]
            }

       .branch {
            multi_genbank: it[4].size() > 1
            single_genbank: it[4].size() <= 1
        }
        .set { plex_branches_ch }

    singleplex_workflow(plex_branches_ch.single_genbank)
    demultiplex_workflow(plex_branches_ch.multi_genbank)

    combine_samplesheets(singleplex_workflow.out, demultiplex_workflow.out)
}

///////////////////////////////////////////////////////////////////////////////

workflow preview {

    Channel
        .fromPath(params.sample_sheet)
        .splitCsv(header: true)
        .unique { row -> row['Barcode_dir'] }
        .map { row ->
            def barcode_out = row['Barcode_dir'] + '_plots'
            def barcode_dir = row['Barcode_dir']  // a barcode is present only once -- no pooling
            def barcode_path = file("${params.fastq_dir}/${barcode_dir}")
            def fastq_files = barcode_path.listFiles()  // multiple FASTQ in each barcode
            return [barcode_out, barcode_path, fastq_files]
        }
        .set { input_ch }

    preview_workflow(input_ch)
}

///////////////////////////////////////////////////////////////////////////////

workflow analysis {

    Channel
    .fromPath(params.sample_sheet)
    .splitCsv(header: true)
    .map { row -> 
        def barcode_dir = row['Barcode_dir']  // a barcode is present only once -- no pooling
        def sample = row['Sample']  // a sample may be present multiple times, in different barcodes
        def entry = "${barcode_dir}_${sample}"  // unique key for each sample sheet entry
        def genbank_path = file("${params.reference_dir}/${sample}.gb")
        def barcode_path = file("${params.fastq_dir}/${barcode_dir}")
        def fastq_files = barcode_path.listFiles()  // multiple FASTQ in each barcode

        return [entry, barcode_dir, barcode_path, fastq_files, sample, genbank_path]
        }
    .set { entries_ch }

    Channel
        .fromPath(params.sample_sheet)
        .splitCsv(header: true)
        .unique { row -> row['Sample'] }
        .map { row -> file(params.reference_dir + '/' + row['Sample'] + '.gb') }
        .set { genbank_ch }

    analysis_workflow(entries_ch, genbank_ch)
}

///////////////////////////////////////////////////////////////////////////////

workflow review {

    Channel
        .fromPath(params.results_csv)
        .splitCsv(header: true)
        .filter { item -> item[params.denovo_columname] == params.denovo_true}
        
        .map { row -> 
            def barcode = row['Barcode']  // a barcode is present only once -- no pooling
            def sample = row['Sample']  // a sample may be present multiple times, in different barcodes
            def entry = "${barcode}_${sample}"  // unique key for each sample sheet entry
            def result = row["Result"]
            def genbank_path = file("${params.reference_dir}/${sample}.gb")
            def fastq_path = file("${params.fastq_filtered_dir}/${barcode}.fastq")
            return [entry, barcode, sample, result, genbank_path, fastq_path]
            }
        .set { entries_de_novo_ch }


    review_denovo(entries_de_novo_ch)
}

///////////////////////////////////////////////////////////////////////////////

workflow assembly {
    Channel
        .fromPath(params.assembly_sheet)
        .splitCsv(header: true)
        
        .map { row -> 
            def barcode = row['Barcode_dir']  // a barcode is present only once -- no pooling
            def length = row['Length']  // estimated length in kbp for canu genomeSize parameter
            def barcode_path = file("${params.fastq_dir}/${barcode}")
            def fastq_files = barcode_path.listFiles()
            return [barcode, barcode_path, fastq_files, length]
            }
        .set { entries_assembly_ch }

   assemble_denovo(entries_assembly_ch)
}