Merge pull request #6 from Zymo-Research/zliu-issue5

add downsampling step

main.nf

@@ -44,6 +44,9 @@ Channel.from( summary.collect{ [it.key, it.value] } )
  * PROCESS DEFINITION
  */
 include { check_design } from "./processes/check_design"
+include { downsample } from "./processes/downsample" addParams(
+    downsample_num: params.downsample_num
+)
 include { miqscore16s } from "./processes/miqscore16s" addParams(
     publish_dir: "${outdir}/miqscore16s",
     forward_primer_length: params.forward_primer_length,
@@ -62,13 +65,13 @@ workflow {
     check_design.out.checked_design
         .splitCsv( header: true )
         .first() // Only support 1 pair of FASTQ for now
-        .multiMap {
-            read_1     : file(it['read_1'])
-            read_2     : file(it['read_2'])
-            sample_name: it['sample']
+        .map {
+            [ it['sample'], file(it['read_1']), file(it['read_2']) ]
         }
         .set { input }
-    miqscore16s(input.sample_name, input.read_1, input.read_2)
+    downsample(input)
+    miqscore_input = params.downsample_num ? downsample.out.reads : input
+    miqscore16s(miqscore_input)
     miqscore16s.out.report.map { "${outdir}/miqscore16s/" + it.getName() }
         .collectFile(name: "${outdir}/download_data/file_locations.txt", newLine: true)
         .set { output_locations }

nextflow.config

@@ -19,7 +19,10 @@ params {
     max_memory = 60.GB
     max_cpus = 8
     max_time = 2.h
-
+
+    // Downsample
+    downsample_num = 100000
+
     // Other Defaults
     outdir = './results'
     name = false

nextflow_schema.json

@@ -86,6 +86,12 @@
             "type": "object",
             "description": "Less common options for the pipeline, typically set in a config file",
             "properties": {
+                "downsample_num": {
+                    "type": "integer",
+                    "description": "Number of read pairs to downsample to. If input has fewer reads than this, downsampling will not happen. Set to 0 to turn this off.",
+                    "minimum": 0,
+                    "default": 100000
+                },
                 "name": {
                     "type": "string",
                     "description": "Name for the pipeline run. If not specified, Nextflow will automatically generate a random mnemonic",

processes/downsample.nf

@@ -0,0 +1,29 @@
+// Downsample FASTQ files
+
+process downsample {
+    container 'quay.io/biocontainers/seqtk:1.4--he4a0461_1'
+
+    input:
+    tuple val(name), path(read_1), path(read_2)
+
+    when:
+    params.downsample_num
+
+    output:
+    tuple val(name), path("${name}_downsample_R1.fastq.gz"), path("${name}_downsample_R2.fastq.gz"), emit: reads
+
+    script:
+    """
+    readnum=\$((\$(zcat $read_1 | wc -l) / 4))
+    if ((\$readnum > $params.downsample_num))
+    then
+    seqtk sample -s1000 $read_1 $params.downsample_num > ${name}_downsample_R1.fastq
+    gzip ${name}_downsample_R1.fastq
+    seqtk sample -s1000 $read_2 $params.downsample_num > ${name}_downsample_R2.fastq
+    gzip ${name}_downsample_R2.fastq
+    else
+    [ ! -f ${name}_downsample_R1.fastq.gz ] && ln -s $read_1 ${name}_downsample_R1.fastq.gz
+    [ ! -f ${name}_downsample_R2.fastq.gz ] && ln -s $read_2 ${name}_downsample_R2.fastq.gz
+    fi
+    """
+}

processes/miqscore16s.nf

@@ -9,9 +9,7 @@ process miqscore16s {
     publishDir "${params.publish_dir}", mode: 'copy'
 
     input:
-    env SAMPLENAME
-    path read_1
-    path read_2
+    tuple val(name), path(read_1), path(read_2)
 
     output:
     path '*.html', emit: report
@@ -22,6 +20,7 @@ process miqscore16s {
     mv $read_1 /data/input/sequence/standard_submitted_R1.fastq
     mv $read_2 /data/input/sequence/standard_submitted_R2.fastq
     mkdir -p /data/output
+    export SAMPLENAME=${name}
     export FORWARDPRIMERLENGTH=${params.forward_primer_length}
     export REVERSEPRIMERLENGTH=${params.reverse_primer_length}
     export AMPLICONLENGTH=${params.amplicon_length}