Mercurial > repos > kkonganti > cfsan_bettercallsal

annotate 0.6.1/bin/fastq_dir_to_samplesheet.py @ 14:b0a37e88ecb5

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload"
author kkonganti
date Thu, 07 Sep 2023 10:13:31 -0400
parents 749faef1caa9
children
rev   line source
kkonganti@11 1 #!/usr/bin/env python3
kkonganti@11 2
kkonganti@11 3 import os
kkonganti@11 4 import sys
kkonganti@11 5 import glob
kkonganti@11 6 import argparse
kkonganti@11 7 import re
kkonganti@11 8
kkonganti@11 9
kkonganti@11 10 def parse_args(args=None):
kkonganti@11 11 Description = "Generate samplesheet from a directory of FastQ files."
kkonganti@11 12 Epilog = "Example usage: python fastq_dir_to_samplesheet.py <FASTQ_DIR> <SAMPLESHEET_FILE>"
kkonganti@11 13
kkonganti@11 14 parser = argparse.ArgumentParser(description=Description, epilog=Epilog)
kkonganti@11 15 parser.add_argument("FASTQ_DIR", help="Folder containing raw FastQ files.")
kkonganti@11 16 parser.add_argument("SAMPLESHEET_FILE", help="Output samplesheet file.")
kkonganti@11 17 parser.add_argument(
kkonganti@11 18 "-st",
kkonganti@11 19 "--strandedness",
kkonganti@11 20 type=str,
kkonganti@11 21 dest="STRANDEDNESS",
kkonganti@11 22 default="unstranded",
kkonganti@11 23 help="Value for 'strandedness' in samplesheet. Must be one of 'unstranded', 'forward', 'reverse'.",
kkonganti@11 24 )
kkonganti@11 25 parser.add_argument(
kkonganti@11 26 "-r1",
kkonganti@11 27 "--read1_extension",
kkonganti@11 28 type=str,
kkonganti@11 29 dest="READ1_EXTENSION",
kkonganti@11 30 default="_R1_001.fastq.gz",
kkonganti@11 31 help="File extension for read 1.",
kkonganti@11 32 )
kkonganti@11 33 parser.add_argument(
kkonganti@11 34 "-r2",
kkonganti@11 35 "--read2_extension",
kkonganti@11 36 type=str,
kkonganti@11 37 dest="READ2_EXTENSION",
kkonganti@11 38 default="_R2_001.fastq.gz",
kkonganti@11 39 help="File extension for read 2.",
kkonganti@11 40 )
kkonganti@11 41 parser.add_argument(
kkonganti@11 42 "-se",
kkonganti@11 43 "--single_end",
kkonganti@11 44 dest="SINGLE_END",
kkonganti@11 45 action="store_true",
kkonganti@11 46 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 47 )
kkonganti@11 48 parser.add_argument(
kkonganti@11 49 "-sn",
kkonganti@11 50 "--sanitise_name",
kkonganti@11 51 dest="SANITISE_NAME",
kkonganti@11 52 action="store_true",
kkonganti@11 53 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 54 )
kkonganti@11 55 parser.add_argument(
kkonganti@11 56 "-sd",
kkonganti@11 57 "--sanitise_name_delimiter",
kkonganti@11 58 type=str,
kkonganti@11 59 dest="SANITISE_NAME_DELIMITER",
kkonganti@11 60 default="_",
kkonganti@11 61 help="Delimiter to use to sanitise sample name.",
kkonganti@11 62 )
kkonganti@11 63 parser.add_argument(
kkonganti@11 64 "-si",
kkonganti@11 65 "--sanitise_name_index",
kkonganti@11 66 type=int,
kkonganti@11 67 dest="SANITISE_NAME_INDEX",
kkonganti@11 68 default=1,
kkonganti@11 69 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 70 )
kkonganti@11 71 return parser.parse_args(args)
kkonganti@11 72
kkonganti@11 73
kkonganti@11 74 def fastq_dir_to_samplesheet(
kkonganti@11 75 fastq_dir,
kkonganti@11 76 samplesheet_file,
kkonganti@11 77 strandedness="unstranded",
kkonganti@11 78 read1_extension="_R1_001.fastq.gz",
kkonganti@11 79 read2_extension="_R2_001.fastq.gz",
kkonganti@11 80 single_end=False,
kkonganti@11 81 sanitise_name=False,
kkonganti@11 82 sanitise_name_delimiter="_",
kkonganti@11 83 sanitise_name_index=1,
kkonganti@11 84 ):
kkonganti@11 85 def sanitize_sample(path, extension):
kkonganti@11 86 """Retrieve sample id from filename"""
kkonganti@11 87 sample = os.path.basename(path).replace(extension, "")
kkonganti@11 88 if sanitise_name:
kkonganti@11 89 if sanitise_name_index > 0:
kkonganti@11 90 sample = sanitise_name_delimiter.join(
kkonganti@11 91 os.path.basename(path).split(sanitise_name_delimiter)[
kkonganti@11 92 :sanitise_name_index
kkonganti@11 93 ]
kkonganti@11 94 )
kkonganti@11 95 # elif sanitise_name_index == -1:
kkonganti@11 96 # sample = os.path.basename(path)[ :os.path.basename(path).index('.') ]
kkonganti@11 97 return sample
kkonganti@11 98
kkonganti@11 99 def get_fastqs(extension):
kkonganti@11 100 """
kkonganti@11 101 Needs to be sorted to ensure R1 and R2 are in the same order
kkonganti@11 102 when merging technical replicates. Glob is not guaranteed to produce
kkonganti@11 103 sorted results.
kkonganti@11 104 See also https://stackoverflow.com/questions/6773584/how-is-pythons-glob-glob-ordered
kkonganti@11 105 """
kkonganti@11 106 abs_fq_files = glob.glob(os.path.join(fastq_dir, f"**", f"*{extension}"), recursive=True)
kkonganti@11 107 return sorted(
kkonganti@11 108 [
kkonganti@11 109 fq for _, fq in enumerate(abs_fq_files) if re.match('^((?!undetermined|unclassified|downloads).)*$', fq, flags=re.IGNORECASE)
kkonganti@11 110 ]
kkonganti@11 111 )
kkonganti@11 112
kkonganti@11 113 read_dict = {}
kkonganti@11 114
kkonganti@11 115 ## Get read 1 files
kkonganti@11 116 for read1_file in get_fastqs(read1_extension):
kkonganti@11 117 sample = sanitize_sample(read1_file, read1_extension)
kkonganti@11 118 if sample not in read_dict:
kkonganti@11 119 read_dict[sample] = {"R1": [], "R2": []}
kkonganti@11 120 read_dict[sample]["R1"].append(read1_file)
kkonganti@11 121
kkonganti@11 122 ## Get read 2 files
kkonganti@11 123 if not single_end:
kkonganti@11 124 for read2_file in get_fastqs(read2_extension):
kkonganti@11 125 sample = sanitize_sample(read2_file, read2_extension)
kkonganti@11 126 read_dict[sample]["R2"].append(read2_file)
kkonganti@11 127
kkonganti@11 128 ## Write to file
kkonganti@11 129 if len(read_dict) > 0:
kkonganti@11 130 out_dir = os.path.dirname(samplesheet_file)
kkonganti@11 131 if out_dir and not os.path.exists(out_dir):
kkonganti@11 132 os.makedirs(out_dir)
kkonganti@11 133
kkonganti@11 134 with open(samplesheet_file, "w") as fout:
kkonganti@11 135 header = ["sample", "fq1", "fq2", "strandedness"]
kkonganti@11 136 fout.write(",".join(header) + "\n")
kkonganti@11 137 for sample, reads in sorted(read_dict.items()):
kkonganti@11 138 for idx, read_1 in enumerate(reads["R1"]):
kkonganti@11 139 read_2 = ""
kkonganti@11 140 if idx < len(reads["R2"]):
kkonganti@11 141 read_2 = reads["R2"][idx]
kkonganti@11 142 sample_info = ",".join([sample, read_1, read_2, strandedness])
kkonganti@11 143 fout.write(f"{sample_info}\n")
kkonganti@11 144 else:
kkonganti@11 145 error_str = (
kkonganti@11 146 "\nWARNING: No FastQ files found so samplesheet has not been created!\n\n"
kkonganti@11 147 )
kkonganti@11 148 error_str += "Please check the values provided for the:\n"
kkonganti@11 149 error_str += " - Path to the directory containing the FastQ files\n"
kkonganti@11 150 error_str += " - '--read1_extension' parameter\n"
kkonganti@11 151 error_str += " - '--read2_extension' parameter\n"
kkonganti@11 152 print(error_str)
kkonganti@11 153 sys.exit(1)
kkonganti@11 154
kkonganti@11 155
kkonganti@11 156 def main(args=None):
kkonganti@11 157 args = parse_args(args)
kkonganti@11 158
kkonganti@11 159 strandedness = "unstranded"
kkonganti@11 160 if args.STRANDEDNESS in ["unstranded", "forward", "reverse"]:
kkonganti@11 161 strandedness = args.STRANDEDNESS
kkonganti@11 162
kkonganti@11 163 fastq_dir_to_samplesheet(
kkonganti@11 164 fastq_dir=args.FASTQ_DIR,
kkonganti@11 165 samplesheet_file=args.SAMPLESHEET_FILE,
kkonganti@11 166 strandedness=strandedness,
kkonganti@11 167 read1_extension=args.READ1_EXTENSION,
kkonganti@11 168 read2_extension=args.READ2_EXTENSION,
kkonganti@11 169 single_end=args.SINGLE_END,
kkonganti@11 170 sanitise_name=args.SANITISE_NAME,
kkonganti@11 171 sanitise_name_delimiter=args.SANITISE_NAME_DELIMITER,
kkonganti@11 172 sanitise_name_index=args.SANITISE_NAME_INDEX,
kkonganti@11 173 )
kkonganti@11 174
kkonganti@11 175
kkonganti@11 176 if __name__ == "__main__":
kkonganti@11 177 sys.exit(main())