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