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