Coverage for rest_api/utils.py: 21%
145 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-10 03:22 +0000
1from dataclasses import dataclass
2import pathlib
3import re
4import uuid
6from dateutil import parser
7from django.core.files.uploadedfile import InMemoryUploadedFile
8import pandas as pd
10from . import models
12IUPAC_CODES = {
13 "nt": {
14 "A": set("A"),
15 "C": set("C"),
16 "G": set("G"),
17 "T": set("T"),
18 "R": set("AGR"),
19 "Y": set("CTY"),
20 "S": set("GCS"),
21 "W": set("ATW"),
22 "K": set("GTK"),
23 "M": set("ACM"),
24 "B": set("CGTB"),
25 "D": set("AGTD"),
26 "H": set("ACTH"),
27 "V": set("ACGV"),
28 "N": set("ACGTRYSWKMBDHVN"),
29 "n": set("N"),
30 },
31 "aa": {
32 "A": set("A"),
33 "R": set("R"),
34 "N": set("N"),
35 "D": set("D"),
36 "C": set("C"),
37 "Q": set("Q"),
38 "E": set("E"),
39 "G": set("G"),
40 "H": set("H"),
41 "I": set("I"),
42 "L": set("L"),
43 "K": set("K"),
44 "M": set("M"),
45 "F": set("F"),
46 "P": set("P"),
47 "S": set("S"),
48 "T": set("T"),
49 "W": set("W"),
50 "Y": set("Y"),
51 "V": set("V"),
52 "U": set("U"),
53 "O": set("O"),
54 "B": set("DNB"),
55 "Z": set("EQZ"),
56 "J": set("ILJ"),
57 "Φ": set("VILFWYMΦ"),
58 "Ω": set("FWYHΩ"),
59 "Ψ": set("VILMΨ"),
60 "π": set("PGASπ"),
61 "ζ": set("STHNQEDKRζ"),
62 "+": set("KRH+"),
63 "-": set("DE-"),
64 "X": set("ARNDCQEGHILKMFPSTWYVUOBZJΦΩΨπζ+-X"),
65 "x": set("X"),
66 },
67}
69regexes = {
70 # 1. A435G (NT)
71 # 2. NC_026438.1:A435G (Replicon:NT)
72 # 3. SH:K53E (Gene:AA)
73 # 4. NC_026438.1:SH:K53E (Replicon:Gene:AA)
74 "snv": re.compile(
75 r"^(\^*)" # group 1: negation (optional)
76 r"(?:([A-Za-z0-9_.-]+):)?" # group 2: replicon accession (optional)
77 r"(?:([A-Za-z0-9_.-]+):)?" # group 3: gene symbol (optional)
78 r"([A-Z]+)" # group 4: Ref (NT oder AA)
79 r"([0-9]+)" # group 5: Position
80 r"(=?[A-Zxn]+)$" # group 6: Alt (NT oder AA)
81 ),
82 # 1. del:133177 or del:133177-133186 (NT)
83 # 2. NC_026438.1:del:133177-133186 (Replicon:NT)
84 # 3. SH:del:34 or SH:del:34-35 (Gene:AA)
85 # 4. NC_026438.1:SH:del:34-35 (Replicon:Gene:AA)
86 "del": re.compile(
87 r"^(\^*)" # group 1: negation(optional)
88 r"(?:([A-Za-z0-9_.]+):)?" # group 2: replicon accession (optional)
89 r"(?:([A-Za-z0-9_-]+):)?" # group 3: gene symbol (optional)
90 r"del:"
91 r"(=?[0-9]+)" # group 4: First deleted position
92 r"(?:-(=?[0-9]+))?$" # group 5: Last deleted position (optional)
93 ),
94}
97def write_to_file(_path: pathlib.Path, file_obj: InMemoryUploadedFile):
98 _path.parent.mkdir(exist_ok=True, parents=True)
99 with open(_path, "wb") as destination:
100 for chunk in file_obj.chunks():
101 destination.write(chunk)
104# distutils will no longer be part of the standard library,
105# here is the code for distutils.util.strtobool() (see the source code for 3.11.2).
106def strtobool(val):
107 """Convert a string representation of truth to true (1) or false (0).
108 True values are 'y', 'yes', 't', 'true', 'on', and '1'; false values
109 are 'n', 'no', 'f', 'false', 'off', and '0'. Raises ValueError if
110 'val' is anything else.
111 """
112 val = val.lower()
113 if val in ("y", "yes", "t", "true", "on", "1"):
114 return 1
115 elif val in ("n", "no", "f", "false", "off", "0"):
116 return 0
117 else:
118 raise ValueError("invalid truth value %r" % (val,))
121def resolve_ambiguous_NT_AA(type, char):
122 try:
123 selected_chars = list(IUPAC_CODES[type][char])
124 except KeyError:
125 raise KeyError(f"Invalid notation '{char}'.")
127 return selected_chars
130def define_profile(mutation: str, protein_symbols: set, replicons: set):
131 """
132 Parse and validate mutation profile.
133 Supported formats:
134 NT Mutations:
135 1. A435G (single-segment only)
136 2. NC_026438.1:A435G (multi-segment with replicon)
138 AA Mutations:
139 3. SH:K53E (gene symbol required)
140 4. NC_026438.1:SH:K53E (replicon + gene symbol)
142 Deletions:
143 5. del:133177 or del:133177-133186 (NT deletion)
144 6. NC_026438.1:del:133177-133186 (NT deletion with replicon)
145 7. SH:del:34 or SH:del:34-35 (AA deletion)
146 8. NC_026438.1:SH:del:34-35 (AA deletion with replicon+gene)
147 """
148 _query = {"label": ""}
149 match = None
150 for mutation_type, regex in regexes.items():
151 match = regex.match(mutation)
152 if match:
153 negate = match.group(1) # ^ for exclusion
154 first_id = match.group(2) # Could be replicon OR gene
155 second_id = match.group(3) # Could be gene OR None
156 # IMPORTANT: determine gene or replicon
157 replicon_accession = None
158 gene_name = None
159 # case 1: both groups, e.g. CY121680.1:HA:S220T
160 if first_id and second_id:
161 if first_id in replicons:
162 replicon_accession = first_id
163 if second_id in protein_symbols:
164 gene_name = second_id
165 else:
166 raise ValueError(
167 f"Invalid gene symbol: '{second_id}'. "
168 f"Valid genes: {', '.join(sorted(list(protein_symbols))[:10])}..."
169 )
170 else:
171 raise ValueError(
172 f"Invalid replicon accession: '{first_id}'. "
173 f"Valid replicons: {', '.join(sorted(list(replicons))[:10])}..."
174 )
176 # case 2: only one group, e.g HA:S220T or NC_026438.1:A435G
177 elif first_id and not second_id:
178 if first_id in protein_symbols:
179 gene_name = first_id
180 elif first_id in replicons:
181 replicon_accession = first_id
182 else:
183 raise ValueError(
184 f"Unknown identifier: '{first_id}'. "
185 f"Not a valid replicon or gene name.\n"
186 f"Valid genes: {', '.join(sorted(list(protein_symbols)))}\n"
187 f"Valid replicons: {', '.join(sorted(list(replicons)))}"
188 )
189 # case 3: no group, e.g A435G, → gene_name and replicon_accession == None
190 if mutation_type == "snv":
191 ref = match.group(4)
192 ref_pos = match.group(5)
193 alt = match.group(6)
195 # Validate alternates based on IUPAC codes
196 alt_is_aa = all(c in IUPAC_CODES["aa"] for c in alt)
197 alt_is_nt = all(c in IUPAC_CODES["nt"] for c in alt)
198 ref_is_aa = ref in IUPAC_CODES["aa"]
199 ref_is_nt = ref in IUPAC_CODES["nt"]
201 # Determine if AA or NT mutation
202 if gene_name:
203 # AA MUTATION (gene_symbol:ref_aa+pos+alt_aa)
204 if not alt_is_aa:
205 raise ValueError(
206 f"Invalid AA mutation: '{alt}' contains non-AA characters."
207 )
208 if not ref_is_aa:
209 raise ValueError(
210 f"Invalid AA reference: '{ref}' is not a valid amino acid."
211 )
213 _query["alt_aa"] = alt
214 _query["ref_aa"] = ref
215 _query["ref_pos"] = ref_pos
216 _query["protein_symbol"] = gene_name
217 _query["label"] = "SNP AA" if len(alt) == 1 else "Ins AA"
218 _query["replicon_accession"] = replicon_accession
220 else:
221 # NT MUTATION (ref_nuc+pos+alt_nuc)
222 if not alt_is_nt or not ref_is_nt:
223 error_message = (
224 f"Invalid NT mutation: '{alt}' contains non-NT characters."
225 if not alt_is_nt
226 else f"Invalid NT reference: '{ref}' is not a valid nucleotide."
227 )
229 if alt_is_aa and ref_is_aa:
230 error_message += (
231 "\nDid you mean an AA query? "
232 "AA mutations require a gene symbol (e.g., HA:S220T)."
233 )
234 raise ValueError(error_message)
236 _query["alt_nuc"] = alt
237 _query["ref_nuc"] = ref
238 _query["ref_pos"] = ref_pos
239 _query["label"] = "SNP Nt" if len(alt) == 1 else "Ins Nt"
240 _query["replicon_accession"] = replicon_accession
242 elif mutation_type == "del":
243 first_deleted = match.group(4)
244 last_deleted = match.group(5) if match.group(5) else ""
245 # if gene name is provided, it's an AA deletion
246 if gene_name:
247 # AA DELETION (gene:del:pos-pos)
248 _query["protein_symbol"] = gene_name
249 _query["label"] = "Del AA"
250 _query["first_deleted"] = first_deleted
251 _query["last_deleted"] = last_deleted
252 _query["replicon_accession"] = replicon_accession
253 # without gene name, it's an NT deletion
254 else:
255 # NT DELETION (del:pos-pos)
256 _query["label"] = "Del Nt"
257 _query["first_deleted"] = first_deleted
258 _query["last_deleted"] = last_deleted
259 _query["replicon_accession"] = replicon_accession
261 # Flag for exclusion
262 _query["exclude"] = True if negate else False
263 break
265 if not match:
266 raise ValueError(f"Invalid mutation notation '{mutation}'.")
268 return _query
271@dataclass
272class PropertyColumnMapping:
273 db_property_name: str
274 data_type: str
275 default: any
278def generate_job_ID(is_prop: bool):
279 job = str(uuid.uuid4())
281 if is_prop:
282 job_id = "cli_prop_" + job
283 else:
284 job_id = "cli_" + job
285 return job_id
288def parse_date(value):
289 # Generalized function to parse any date format
290 # exp. 2021-11-30T00:00:00, 2021-02-16 19:00:03 +0100
291 # 2/2/2021
292 if pd.isna(value) or not value or str(value).strip() == "":
293 return pd.NA
294 try:
295 parsed_date = parser.parse(value)
296 return parsed_date.strftime("%Y-%m-%d")
297 except ValueError:
298 raise ValueError(f"ValueError: Unable to parse date from '{value}'")
299 except TypeError:
300 raise TypeError(f"TypeError: Invalid type for date parsing - '{value}'")
301 # or return None?
302 # raise ValueError(f"Failed to parse date '{value}': {str(e)}") from e
305def get_distinct_gene_symbols(reference=None):
306 """
307 Helper method to get distinct gene symbols.
308 This method can be called from anywhere.
309 """
310 queryset = models.Gene.objects.distinct("symbol").values("symbol")
311 if reference:
312 queryset = queryset.filter(replicon__reference__accession=reference)
313 return [item["symbol"] for item in queryset if item["symbol"]]
316def get_distinct_replicon_accessions(reference=None):
317 """
318 Helper method to get distinct replicon accessions.
319 """
320 queryset = models.Replicon.objects.all()
321 if reference:
322 queryset = queryset.filter(reference__accession=reference)
323 qs = queryset.values_list("accession", flat=True).distinct()
324 return [acc for acc in qs if acc]
327def get_distinct_cds_accessions(reference=None, replicon=None):
328 """
329 Helper method to get distinct CDS accessions.
330 """
331 queryset = models.CDS.objects.all()
332 if replicon:
333 queryset = queryset.filter(gene__replicon__accession=replicon)
334 if reference:
335 queryset = queryset.filter(gene__replicon__reference__accession=reference)
336 qs = queryset.values_list("accession", flat=True).distinct()
337 return [acc for acc in qs if acc]
340def get_distinct_peptide_descriptions(reference=None):
341 """
342 Helper method to get distinct gene symbols.
343 This method can be called from anywhere.
344 """
345 queryset = models.Peptide.objects.distinct("description").values("description")
346 if reference:
347 queryset = queryset.filter(replicon__reference__accession=reference)
348 return [item["description"] for item in queryset]
351def parse_default_data(value):
352 # Convert "None", "null", or empty strings to Python None
353 if value in {"None", "null", ""}:
354 return None
355 return value