From 2a1ff88cc7f5467d3669514302e78062ce8715e5 Mon Sep 17 00:00:00 2001
From: Chris Wright <chris.wright@nanoporetech.com>
Date: Mon, 12 Jul 2021 21:13:43 +0100
Subject: [PATCH] add fastqingress module

---
 bin/check_sample_sheet.py |  41 +++++++++
 lib/fastqingress.nf       | 170 ++++++++++++++++++++++++++++++++++++++
 main.nf                   |  32 +++----
 nextflow.config           |   2 +
 4 files changed, 229 insertions(+), 16 deletions(-)
 create mode 100755 bin/check_sample_sheet.py
 create mode 100644 lib/fastqingress.nf

diff --git a/bin/check_sample_sheet.py b/bin/check_sample_sheet.py
new file mode 100755
index 0000000..ae8fb76
--- /dev/null
+++ b/bin/check_sample_sheet.py
@@ -0,0 +1,41 @@
+#!/usr/bin/env python
+"""Script to check that sample sheet is well-formatted."""
+import argparse
+import sys
+
+import pandas as pd
+
+
+def main():
+    """Run entry point."""
+    parser = argparse.ArgumentParser()
+    parser.add_argument('sample_sheet')
+    parser.add_argument('output')
+    args = parser.parse_args()
+
+    try:
+        samples = pd.read_csv(args.sample_sheet, sep=None)
+        if 'alias' in samples.columns:
+            if 'sample_name' in samples.columns:
+                sys.stderr.write(
+                    "Warning: sample sheet contains both 'alias' and "
+                    'sample_name, using the former.')
+            samples['sample_name'] = samples['alias']
+        if 'barcode' not in samples.columns \
+                or 'sample_name' not in samples.columns:
+            raise IOError()
+    except Exception:
+        raise IOError(
+            "Could not parse sample sheet, it must contain two columns "
+            "named 'barcode' and 'sample_name' or 'alias'.")
+    # check duplicates
+    dup_bc = samples['barcode'].duplicated()
+    dup_sample = samples['sample_name'].duplicated()
+    if any(dup_bc) or any(dup_sample):
+        raise IOError(
+            "Sample sheet contains duplicate values.")
+    samples.to_csv(args.output, sep=",", index=False)
+
+
+if __name__ == '__main__':
+    main()
diff --git a/lib/fastqingress.nf b/lib/fastqingress.nf
new file mode 100644
index 0000000..30713a7
--- /dev/null
+++ b/lib/fastqingress.nf
@@ -0,0 +1,170 @@
+
+process checkSampleSheet {
+    label "artic"
+    cpus 1
+    input:
+        file "sample_sheet.txt"
+    output:
+        file "samples.txt"
+    """
+    check_sample_sheet.py sample_sheet.txt samples.txt
+    """
+}
+
+
+/**
+ * Load a sample sheet into a Nextflow channel to map barcodes
+ * to sample names.
+ *
+ * @param samples CSV file according to MinKNOW sample sheet specification
+ * @return A Nextflow Channel of tuples (barcode, sample name)
+ */ 
+def check_sample_sheet(samples)
+{
+    println("Checking sample sheet.")
+    sample_sheet = Channel.fromPath(samples, checkIfExists: true)
+    sample_sheet = checkSampleSheet(sample_sheet)
+        .splitCsv(header: true)
+        .map { row -> tuple(row.barcode, row.sample_name) }
+    return sample_sheet
+}
+
+
+/**
+ * Find fastq data using various globs. Wrapper around Nextflow `file`
+ * method.
+ *
+ * @param patten glob pattern for top level input folder.
+ * @param maxdepth maximum depth to traverse
+ * @return list of files.
+ */ 
+def find_fastq(pattern, maxdepth)
+{
+    files = []
+    extensions = ["fastq", "fastq.gz", "fq", "fq.gz"]
+    for (ext in extensions) {
+        files += file("${pattern}/*.${ext}", type: 'file', maxdepth: maxdepth)
+    }
+    return files
+}
+
+
+/**
+ * Rework EPI2ME flattened directory structure into standard form
+ * files are matched on barcode\d+ and moved into corresponding
+ * subdirectories ready for processing.
+ *
+ * @param input_folder Top-level input directory.
+ * @param output_folder Top-level output_directory.
+ * @return A File object representating the staging directory created
+ *     under output_folder
+ */ 
+def sanitize_fastq(input_folder, output_folder)
+{
+    println("Running sanitization.")
+    println(" - Moving files: ${input_folder} -> ${output_folder}")
+    staging = new File(output_folder)
+    staging.mkdirs()
+    files = find_fastq("${input_folder}/**/", 1)
+    for (fastq in files) {
+        fname = fastq.getFileName()
+        // find barcode
+        pattern = ~/barcode\d+/
+        matcher = fname =~ pattern
+        if (!matcher.find()) {
+            // not barcoded - leave alone
+            fastq.renameTo("${staging}/${fname}")
+        } else {
+            bc_dir = new File("${staging}/${matcher[0]}")
+            bc_dir.mkdirs()
+            fastq.renameTo("${staging}/${matcher[0]}/${fname}")
+        }
+    }
+    println(" - Finished sanitization.")
+    return staging
+}
+
+
+/**
+ * Resolves input folder containing barcode subdirectories
+ * or a flat set of fastq data to a Nextflow Channel. Removes barcode
+ * directories with no fastq files.
+ *
+ * @param input_folder Top level input folder to locate fastq data
+ * @param sample_sheet List of tuples mapping barcode to sample name
+ *     or a simple string for non-multiplexed data.
+ * @return Channel of tuples (path, sample_name)
+ */
+def resolve_barcode_structure(input_folder, sample_sheet)
+{
+    println("Checking input directory structure.")
+    barcode_dirs = file("$input_folder/barcode*", type: 'dir', maxdepth: 1)
+    not_barcoded = find_fastq("$input_folder/", 1)
+    samples = null
+    if (barcode_dirs) {
+        println(" - Found barcode directories")
+        // remove empty barcode_dirs
+        valid_barcode_dirs = []
+        invalid_barcode_dirs = []
+        for (d in barcode_dirs) {
+            if(!find_fastq(d, 1)) {
+                invalid_barcode_dirs << d
+            } else {
+                valid_barcode_dirs << d
+            }
+        }
+        if (invalid_barcode_dirs.size() > 0) {
+            println(" - Some barcode directories did not contain .fastq(.gz) files:")
+            for (d in invalid_barcode_dirs) {
+                println("   - ${d}")
+            }
+        }
+        // link sample names to barcode through sample sheet
+        if (!sample_sheet) {
+            sample_sheet = Channel
+                .fromPath(valid_barcode_dirs)
+                .filter(~/.*barcode[0-9]{1,3}$/)  // up to 192
+                .map { path -> tuple(path.baseName, path.baseName) }
+        }
+        samples = Channel
+            .fromPath(valid_barcode_dirs)
+            .filter(~/.*barcode[0-9]{1,3}$/)  // up to 192
+            .map { path -> tuple(path.baseName, path) }
+            .join(sample_sheet)
+            .map { barcode, path, sample -> tuple(path, sample) }
+    } else if (not_barcoded) {
+        println(" - Found fastq files, assuming single sample")
+        sample = (sample_sheet == null) ? "unknown" : sample_sheet
+        samples = Channel
+            .fromPath(input_folder, type: 'dir', maxDepth:1)
+            .map { path -> tuple(path, sample) }
+    }
+    return samples
+}
+
+
+/**
+ * Take an input directory and sample sheet to return a channel of
+ * named samples.
+ *
+ * @param input_folder Top level input folder to locate fastq data
+ * @param sample_sheet List of tuples mapping barcode to sample name
+ *     or a simple string for non-multiplexed data.
+ * @return Channel of tuples (path, sample_name)
+ */
+def fastq_ingress(input_folder, output_folder, samples, sanitize)
+{
+    // EPI2ME harness 
+    if (sanitize) {
+        staging = "${output_folder}/staging"
+        input_folder = sanitize_fastq(input_folder, staging)
+    }
+    // check sample sheet
+    sample_sheet = null
+    if (samples) {
+        sample_sheet = check_sample_sheet(samples)
+    }
+    // resolve whether we have demultiplexed data or single sample
+    data = resolve_barcode_structure(input_folder, sample_sheet)
+    return data
+}
diff --git a/main.nf b/main.nf
index 9d5ef80..2eed4e0 100644
--- a/main.nf
+++ b/main.nf
@@ -12,6 +12,7 @@
 
 nextflow.enable.dsl = 2
 
+include { fastq_ingress } from './lib/fastqingress' 
 
 def helpMessage(){
     log.info """
@@ -21,7 +22,9 @@ Usage:
     nextflow run epi2melabs/wf-template [options]
 
 Script Options:
-    --fastq        DIR     Path to directory containing FASTQ files (required)
+    --fastq        DIR     Path to FASTQ directory (required)
+    --samples      FILE    CSV file with columns named `barcode` and `sample_name`
+                           (or simply a sample name for non-multiplexed data).
     --out_dir      DIR     Path for output (default: $params.out_dir)
 """
 }
@@ -33,12 +36,12 @@ process summariseReads {
     label "pysam"
     cpus 1
     input:
-        file "input"
+        tuple path(directory), val(sample_name)
     output:
-        file "seqs.txt"
+        path "${sample_name}.stats"
     shell:
     """
-    fastcat -r seqs.txt input/*.fastq* > /dev/null
+    fastcat -s ${sample_name} -r ${sample_name}.stats -x ${directory} > /dev/null
     """
 }
 
@@ -46,9 +49,9 @@ process summariseReads {
 process makeReport {
     label "pysam"
     input:
-        file "seqs.txt"
+        path "seqs.txt"
     output:
-        file "wf-template-report.html"
+        path "wf-template-report.html"
     """
     report.py wf-template-report.html seqs.txt
     """
@@ -63,9 +66,9 @@ process output {
     label "pysam"
     publishDir "${params.out_dir}", mode: 'copy', pattern: "*"
     input:
-        file fname
+        path fname
     output:
-        file fname
+        path fname
     """
     echo "Writing output files"
     """
@@ -98,12 +101,9 @@ workflow {
         exit 1
     }
 
-    reads = file("$params.fastq/*.fastq*", type: 'file', maxdepth: 1)
-    if (reads) {
-        reads = Channel.fromPath(params.fastq, type: 'dir', checkIfExists: true)
-        results = pipeline(reads)
-        output(results)
-    } else {
-        println("No .fastq(.gz) files found under `${params.fastq}`.")
-    }
+    samples = fastq_ingress(
+        params.fastq, params.out_dir, params.samples, params.sanitize_fastq)
+
+    results = pipeline(samples)
+    output(results)
 }
diff --git a/nextflow.config b/nextflow.config
index 2289286..9f3f214 100644
--- a/nextflow.config
+++ b/nextflow.config
@@ -14,6 +14,8 @@ params {
     help = false
     fastq = null
     out_dir = "output"
+    samples = null
+    sanitize_fastq = false
     wfversion = "v0.0.6"
     aws_image_prefix = null
     aws_queue = null