-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathsnp_ranger.py
More file actions
156 lines (100 loc) · 4.1 KB
/
Copy pathsnp_ranger.py
File metadata and controls
156 lines (100 loc) · 4.1 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
@author: camilla eldridge
"""
import sys
import re
from typing import List, Dict
mauve_SNP_output = sys.argv[1]
first_gbk = sys.argv[2]
''' Get node ids and snp pos for sequence 1 only'''
contig_pos = []
with open(mauve_SNP_output, "r") as snps:
next(snps)
for line in snps:
if not line.strip():
continue
spline = line.split()
if len(spline) < 4:
continue
contig_id = "_".join(spline[1].split("_")[:2])
contig_pos.append(f"{contig_id},{spline[2]},{spline[3]}")
''' Get set of node ids with SNPS'''
snp_node_ids: set[str] = {node.split(",")[0] for node in contig_pos}
''' Get gbk entries for SNP nodes '''
gbk_snp_entries: List[str] = []
with open(first_gbk, "r") as gbk:
gbk1 = gbk.read().split("LOCUS")[1:]
for entry in gbk1:
at2 = entry.split("\n", 1)[0].split()[0]
if at2 in snp_node_ids:
gbk_snp_entries.append(entry)
''' Find annotated regions for each contig'''
def get_annotations(x: str) -> List[str]:
contig_cds: List[str] = []
NODEid = x.split()[0]
contig_cds.append(NODEid)
y = x.split("ORIGIN")[0][1:]
for Q in y.split("\n"):
to_app = ""
if "rRNA " in Q or " CDS " in Q or " tRNA " in Q:
to_app = re.sub(r'[^a-zA-Z0-9]', ' ', Q).replace("complement", "")
if "/product=" in Q:
to_app = to_app + Q.strip().replace(" ", "_").replace("/", "")
contig_cds.append(to_app)
return [item.strip() for item in contig_cds if item.strip()]
''' Get annotated entry in each contig '''
all_snp_conts: List[List[str]] = [get_annotations(C) for C in gbk_snp_entries]
''' Print out which nodes have a SNP but no annotation'''
all_snp_conts_replaced: List[str] = []
for E in all_snp_conts:
nod_id = str(E).split()[0]
if len(E) < 2: # If SNP but no annotation found.. print the node id.
print(f"{''.join(map(str, E))}: SNP here but no annotations found!")
print("-----")
else:
E = str(E).replace("CDS ", nod_id).replace("tRNA ", nod_id).replace("rRNA ", nod_id).replace("product=", "")
E = re.sub(r"[',\[\]\"()]", "", E)
all_snp_conts_replaced.append([r for r in E.split()[1:]])
''' Remove duplicate node ids from SNP list and add a split term'''
w: List[str] = []
for i in contig_pos:
i = i.split(",")
nod_id, nod_pos = i[0], i[1]
if nod_id not in w:
w.append("split_me")
w.append(nod_id)
w.append(nod_pos)
''' Get all SNP positions for each node '''
snp_node_positions: List[List[str]] = list(filter(None, [k.split() for k in " ".join(w).split("split_me")]))
''' Iterate through SNP positions and annotated regions for each node; a bit loopy - there is probably a better way to do this'''
snps_in_annot: Dict[str, List[Dict[str, str]]] = {}
for snp_val in snp_node_positions:
node = snp_val[0]
snps_in_annot[node] = []
for triplet in all_snp_conts_replaced:
triplet_node = triplet[0]
if triplet_node == node:
values = snp_val[1:]
for i in range(1, len(triplet), 4):
product = triplet[i + 2]
range_values = triplet[i:i + 2]
for value in values:
int_value = int(value)
if int(range_values[0]) <= int_value <= int(range_values[1]):
snps_in_annot[node].append({'product': product, 'value': value, 'range': range_values})
break
''' Remove entries with empty key or values (SNPs outside annotated range)'''
snps_in_annot = {k: v for k, v in snps_in_annot.items() if k and v}
''' Print out final results '''
for node, entries in snps_in_annot.items():
print(f"{node}")
for entry in entries:
product = entry['product']
value = entry['value']
value_range = entry['range']
print(f"Product: {product}")
print(f"Value: {value}")
print(f"Range: {', '.join(value_range)}")
print("-----")