kkonganti@0: #!/usr/bin/env python3
kkonganti@0: 
kkonganti@0: import os
kkonganti@0: import sys
kkonganti@0: import glob
kkonganti@0: import argparse
kkonganti@0: import re
kkonganti@0: 
kkonganti@0: 
kkonganti@0: def parse_args(args=None):
kkonganti@0:     Description = "Generate samplesheet from a directory of FastQ files."
kkonganti@0:     Epilog = "Example usage: python fastq_dir_to_samplesheet.py <FASTQ_DIR> <SAMPLESHEET_FILE>"
kkonganti@0: 
kkonganti@0:     parser = argparse.ArgumentParser(description=Description, epilog=Epilog)
kkonganti@0:     parser.add_argument("FASTQ_DIR", help="Folder containing raw FastQ files.")
kkonganti@0:     parser.add_argument("SAMPLESHEET_FILE", help="Output samplesheet file.")
kkonganti@0:     parser.add_argument(
kkonganti@0:         "-st",
kkonganti@0:         "--strandedness",
kkonganti@0:         type=str,
kkonganti@0:         dest="STRANDEDNESS",
kkonganti@0:         default="unstranded",
kkonganti@0:         help="Value for 'strandedness' in samplesheet. Must be one of 'unstranded', 'forward', 'reverse'.",
kkonganti@0:     )
kkonganti@0:     parser.add_argument(
kkonganti@0:         "-r1",
kkonganti@0:         "--read1_extension",
kkonganti@0:         type=str,
kkonganti@0:         dest="READ1_EXTENSION",
kkonganti@0:         default="_R1_001.fastq.gz",
kkonganti@0:         help="File extension for read 1.",
kkonganti@0:     )
kkonganti@0:     parser.add_argument(
kkonganti@0:         "-r2",
kkonganti@0:         "--read2_extension",
kkonganti@0:         type=str,
kkonganti@0:         dest="READ2_EXTENSION",
kkonganti@0:         default="_R2_001.fastq.gz",
kkonganti@0:         help="File extension for read 2.",
kkonganti@0:     )
kkonganti@0:     parser.add_argument(
kkonganti@0:         "-se",
kkonganti@0:         "--single_end",
kkonganti@0:         dest="SINGLE_END",
kkonganti@0:         action="store_true",
kkonganti@0:         help="Single-end information will be auto-detected but this option forces paired-end FastQ files to be treated as single-end so only read 1 information is included in the samplesheet.",
kkonganti@0:     )
kkonganti@0:     parser.add_argument(
kkonganti@0:         "-sn",
kkonganti@0:         "--sanitise_name",
kkonganti@0:         dest="SANITISE_NAME",
kkonganti@0:         action="store_true",
kkonganti@0:         help="Whether to further sanitise FastQ file name to get sample id. Used in conjunction with --sanitise_name_delimiter and --sanitise_name_index.",
kkonganti@0:     )
kkonganti@0:     parser.add_argument(
kkonganti@0:         "-sd",
kkonganti@0:         "--sanitise_name_delimiter",
kkonganti@0:         type=str,
kkonganti@0:         dest="SANITISE_NAME_DELIMITER",
kkonganti@0:         default="_",
kkonganti@0:         help="Delimiter to use to sanitise sample name.",
kkonganti@0:     )
kkonganti@0:     parser.add_argument(
kkonganti@0:         "-si",
kkonganti@0:         "--sanitise_name_index",
kkonganti@0:         type=int,
kkonganti@0:         dest="SANITISE_NAME_INDEX",
kkonganti@0:         default=1,
kkonganti@0:         help="After splitting FastQ file name by --sanitise_name_delimiter all elements before this index (1-based) will be joined to create final sample name.",
kkonganti@0:     )
kkonganti@0:     return parser.parse_args(args)
kkonganti@0: 
kkonganti@0: 
kkonganti@0: def fastq_dir_to_samplesheet(
kkonganti@0:     fastq_dir,
kkonganti@0:     samplesheet_file,
kkonganti@0:     strandedness="unstranded",
kkonganti@0:     read1_extension="_R1_001.fastq.gz",
kkonganti@0:     read2_extension="_R2_001.fastq.gz",
kkonganti@0:     single_end=False,
kkonganti@0:     sanitise_name=False,
kkonganti@0:     sanitise_name_delimiter="_",
kkonganti@0:     sanitise_name_index=1,
kkonganti@0: ):
kkonganti@0:     def sanitize_sample(path, extension):
kkonganti@0:         """Retrieve sample id from filename"""
kkonganti@0:         sample = os.path.basename(path).replace(extension, "")
kkonganti@0:         if sanitise_name:
kkonganti@0:             if sanitise_name_index > 0:
kkonganti@0:                 sample = sanitise_name_delimiter.join(
kkonganti@0:                     os.path.basename(path).split(sanitise_name_delimiter)[
kkonganti@0:                         :sanitise_name_index
kkonganti@0:                     ]
kkonganti@0:                 )
kkonganti@0:             # elif sanitise_name_index == -1:
kkonganti@0:             #     sample = os.path.basename(path)[ :os.path.basename(path).index('.') ]
kkonganti@0:         return sample
kkonganti@0: 
kkonganti@0:     def get_fastqs(extension):
kkonganti@0:         """
kkonganti@0:         Needs to be sorted to ensure R1 and R2 are in the same order
kkonganti@0:         when merging technical replicates. Glob is not guaranteed to produce
kkonganti@0:         sorted results.
kkonganti@0:         See also https://stackoverflow.com/questions/6773584/how-is-pythons-glob-glob-ordered
kkonganti@0:         """
kkonganti@0:         abs_fq_files = glob.glob(os.path.join(fastq_dir, f"**", f"*{extension}"), recursive=True)
kkonganti@0:         return sorted(
kkonganti@0:             [
kkonganti@0:                 fq for _, fq in enumerate(abs_fq_files) if re.match('^((?!undetermined|unclassified|fail|class|qc|downloads).)*$', fq, flags=re.IGNORECASE)
kkonganti@0:             ]
kkonganti@0:         )
kkonganti@0: 
kkonganti@0:     read_dict = {}
kkonganti@0: 
kkonganti@0:     ## Get read 1 files
kkonganti@0:     for read1_file in get_fastqs(read1_extension):
kkonganti@0:         sample = sanitize_sample(read1_file, read1_extension)
kkonganti@0:         if sample not in read_dict:
kkonganti@0:             read_dict[sample] = {"R1": [], "R2": []}
kkonganti@0:         read_dict[sample]["R1"].append(read1_file)
kkonganti@0: 
kkonganti@0:     ## Get read 2 files
kkonganti@0:     if not single_end:
kkonganti@0:         for read2_file in get_fastqs(read2_extension):
kkonganti@0:             sample = sanitize_sample(read2_file, read2_extension)
kkonganti@0:             read_dict[sample]["R2"].append(read2_file)
kkonganti@0: 
kkonganti@0:     ## Write to file
kkonganti@0:     if len(read_dict) > 0:
kkonganti@0:         out_dir = os.path.dirname(samplesheet_file)
kkonganti@0:         if out_dir and not os.path.exists(out_dir):
kkonganti@0:             os.makedirs(out_dir)
kkonganti@0: 
kkonganti@0:         with open(samplesheet_file, "w") as fout:
kkonganti@0:             header = ["sample", "fq1", "fq2", "strandedness"]
kkonganti@0:             fout.write(",".join(header) + "\n")
kkonganti@0:             for sample, reads in sorted(read_dict.items()):
kkonganti@0:                 for idx, read_1 in enumerate(reads["R1"]):
kkonganti@0:                     read_2 = ""
kkonganti@0:                     if idx < len(reads["R2"]):
kkonganti@0:                         read_2 = reads["R2"][idx]
kkonganti@0:                     sample_info = ",".join([sample, read_1, read_2, strandedness])
kkonganti@0:                     fout.write(f"{sample_info}\n")
kkonganti@0:     else:
kkonganti@0:         error_str = (
kkonganti@0:             "\nWARNING: No FastQ files found so samplesheet has not been created!\n\n"
kkonganti@0:         )
kkonganti@0:         error_str += "Please check the values provided for the:\n"
kkonganti@0:         error_str += "  - Path to the directory containing the FastQ files\n"
kkonganti@0:         error_str += "  - '--read1_extension' parameter\n"
kkonganti@0:         error_str += "  - '--read2_extension' parameter\n"
kkonganti@0:         print(error_str)
kkonganti@0:         sys.exit(1)
kkonganti@0: 
kkonganti@0: 
kkonganti@0: def main(args=None):
kkonganti@0:     args = parse_args(args)
kkonganti@0: 
kkonganti@0:     strandedness = "unstranded"
kkonganti@0:     if args.STRANDEDNESS in ["unstranded", "forward", "reverse"]:
kkonganti@0:         strandedness = args.STRANDEDNESS
kkonganti@0: 
kkonganti@0:     fastq_dir_to_samplesheet(
kkonganti@0:         fastq_dir=args.FASTQ_DIR,
kkonganti@0:         samplesheet_file=args.SAMPLESHEET_FILE,
kkonganti@0:         strandedness=strandedness,
kkonganti@0:         read1_extension=args.READ1_EXTENSION,
kkonganti@0:         read2_extension=args.READ2_EXTENSION,
kkonganti@0:         single_end=args.SINGLE_END,
kkonganti@0:         sanitise_name=args.SANITISE_NAME,
kkonganti@0:         sanitise_name_delimiter=args.SANITISE_NAME_DELIMITER,
kkonganti@0:         sanitise_name_index=args.SANITISE_NAME_INDEX,
kkonganti@0:     )
kkonganti@0: 
kkonganti@0: 
kkonganti@0: if __name__ == "__main__":
kkonganti@0:     sys.exit(main())