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