Mercurial > repos > in_silico > cravat_vcf_convert
comparison cravat_convert/cravat_convert.py @ 1:152cc8feff8a draft
Uploaded
author | in_silico |
---|---|
date | Tue, 12 Jun 2018 11:05:27 -0400 |
parents | |
children |
comparison
equal
deleted
inserted
replaced
0:399f41a4bad6 | 1:152cc8feff8a |
---|---|
1 ''' | |
2 Convert a VCF format file to Cravat format file | |
3 ''' | |
4 | |
5 import os | |
6 import argparse | |
7 from vcf_converter import CravatConverter | |
8 ### | |
9 import ipdb | |
10 | |
11 | |
12 # File read/write configuration variables | |
13 vcf_sep = '\t' | |
14 cr_sep = '\t' | |
15 cr_newline = '\n' | |
16 | |
17 # VCF Headers mapped to their index position in a row of VCF values | |
18 vcf_mapping = { | |
19 'CHROM': 0, | |
20 'POS': 1, | |
21 'ID': 2, | |
22 'REF': 3, | |
23 'ALT': 4, | |
24 'QUAL': 5, | |
25 'FILTER': 6, | |
26 'INFO': 7, | |
27 'FORMAT': 8, | |
28 'NA00001': 9, | |
29 'NA00002': 10, | |
30 'NA00003': 11 | |
31 } | |
32 | |
33 | |
34 def get_args(): | |
35 parser = argparse.ArgumentParser() | |
36 parser.add_argument('--input', | |
37 '-i', | |
38 required = True, | |
39 help='Input path to a VCF file for conversion',) | |
40 parser.add_argument('--output', | |
41 '-o', | |
42 default = os.path.join(os.getcwd(), "cravat_converted.txt"), | |
43 help = 'Output path to write the cravat file to') | |
44 return parser.parse_args() | |
45 | |
46 | |
47 def convert(in_path, out_path=None): | |
48 if not out_path: | |
49 base, _ = os.path.split(in_path) | |
50 out_path = os.path.join(base, "cravat_converted.txt") | |
51 | |
52 with open(in_path, 'r') as in_file, \ | |
53 open(out_path, 'w') as out_file: | |
54 | |
55 # cr_count will be used to generate the 'TR' field of the cravat rows (first header) | |
56 cr_count = 0 | |
57 # VCF lines are always assumed to be '+' strand, as VCF doesn't specify that attribute | |
58 strand = '+' | |
59 # VCF converter. Adjusts position, reference, and alternate for Cravat formatting. | |
60 converter = CravatConverter() | |
61 | |
62 for line in in_file: | |
63 if line.startswith("#"): | |
64 continue | |
65 line = line.strip().split(vcf_sep) | |
66 # row is dict of VCF headers mapped to corresponding values of this line | |
67 row = { header: line[index] for header, index in vcf_mapping.items() } | |
68 for alt in row["ALT"].split(","): | |
69 new_pos, new_ref, new_alt = converter.extract_vcf_variant(strand, row["POS"], row["REF"], alt) | |
70 new_pos, new_ref, new_alt = str(new_pos), str(new_ref), str(new_alt) | |
71 cr_line = cr_sep.join([ | |
72 'TR' + str(cr_count), row['CHROM'], new_pos, strand, new_ref, new_alt, row['ID'] | |
73 ]) | |
74 out_file.write(cr_line + cr_newline) | |
75 cr_count += 1 | |
76 | |
77 | |
78 if __name__ == "__main__": | |
79 cli_args = get_args() | |
80 convert(cli_args.input, cli_args.output) |