1
|
|
|
import abc |
|
|
|
|
2
|
|
|
import gzip |
3
|
|
|
from pathlib import Path |
4
|
|
|
from dataclasses import dataclass |
5
|
|
|
from typing import Union, Sequence, Optional |
6
|
|
|
|
7
|
|
|
import defusedxml.ElementTree as Xml |
|
|
|
|
8
|
|
|
import orjson |
|
|
|
|
9
|
|
|
from pocketutils.tools.common_tools import CommonTools |
|
|
|
|
10
|
|
|
|
11
|
|
|
from mandos.model.settings import QUERY_EXECUTORS, MANDOS_SETTINGS |
12
|
|
|
from pocketutils.core.dot_dict import NestedDotDict |
|
|
|
|
13
|
|
|
from pocketutils.core.query_utils import QueryExecutor |
|
|
|
|
14
|
|
|
|
15
|
|
|
from mandos.model import Api |
|
|
|
|
16
|
|
|
|
17
|
|
|
|
18
|
|
|
def _is_float(s): |
|
|
|
|
19
|
|
|
try: |
20
|
|
|
float(s) |
21
|
|
|
return True |
22
|
|
|
except ValueError: |
23
|
|
|
return False |
24
|
|
|
|
25
|
|
|
|
26
|
|
|
@dataclass(frozen=True, repr=True, order=True) |
|
|
|
|
27
|
|
|
class HmdbProperty: |
28
|
|
|
name: str |
29
|
|
|
source: str |
30
|
|
|
value: Union[None, str, int, float, bool] |
31
|
|
|
experimental: bool |
32
|
|
|
|
33
|
|
|
|
34
|
|
|
class HmdbApi(Api, metaclass=abc.ABCMeta): |
|
|
|
|
35
|
|
|
def fetch(self, hmdb_id: str) -> NestedDotDict: |
|
|
|
|
36
|
|
|
raise NotImplementedError() |
37
|
|
|
|
38
|
|
|
def fetch_properties(self, hmdb_id: str) -> Sequence[HmdbProperty]: |
|
|
|
|
39
|
|
|
raise NotImplementedError() |
40
|
|
|
|
41
|
|
|
def fetch_tissues(self, hmdb_id: str) -> Sequence[HmdbProperty]: |
|
|
|
|
42
|
|
|
raise NotImplementedError() |
43
|
|
|
|
44
|
|
|
|
45
|
|
|
class JsonBackedHmdbApi(HmdbApi, metaclass=abc.ABCMeta): |
|
|
|
|
46
|
|
|
def fetch_properties(self, hmdb_id: str) -> Sequence[HmdbProperty]: |
47
|
|
|
data = self.fetch(hmdb_id) |
48
|
|
|
pred = data.sub("metabolite.predicted_properties.property") |
49
|
|
|
exp = data.sub("metabolite.experimental_properties.property") |
50
|
|
|
items = [self._prop(x, True) for x in exp] |
51
|
|
|
items += [self._prop(x, False) for x in pred] |
52
|
|
|
return items |
53
|
|
|
|
54
|
|
|
def fetch_tissues(self, hmdb_id: str) -> Sequence[str]: |
55
|
|
|
data = self.fetch(hmdb_id) |
56
|
|
|
tissues = data.get_list_as("metabolite.biological_properties.tissue_locations.tissue", str) |
57
|
|
|
return [] if tissues is None else tissues |
58
|
|
|
|
59
|
|
|
def _prop(self, x: NestedDotDict, experimental: bool): |
|
|
|
|
60
|
|
|
value = x.req_as("value", str) |
61
|
|
|
if value.isdigit(): |
62
|
|
|
value = int(value) |
63
|
|
|
elif value.lower() == "true": |
64
|
|
|
value = True |
65
|
|
|
elif value.lower() == "false": |
66
|
|
|
value = False |
67
|
|
|
elif CommonTools.is_probable_null(value): |
68
|
|
|
value = None |
69
|
|
|
elif _is_float(value): |
70
|
|
|
value = float(value) |
71
|
|
|
return HmdbProperty( |
72
|
|
|
name=x["kind"], value=value, source=x["source"], experimental=experimental |
73
|
|
|
) |
74
|
|
|
|
75
|
|
|
|
76
|
|
|
class QueryingHmdbApi(JsonBackedHmdbApi): |
|
|
|
|
77
|
|
|
def __init__(self, executor: QueryExecutor = QUERY_EXECUTORS.hmdb): |
78
|
|
|
self._executor = executor |
79
|
|
|
|
80
|
|
|
def fetch(self, hmdb_id: str) -> NestedDotDict: |
81
|
|
|
# e.g. https://hmdb.ca/metabolites/HMDB0001925.xml |
82
|
|
|
url = f"https://hmdb.ca/metabolites/{hmdb_id}.xml" |
83
|
|
|
data = self._executor(url) |
84
|
|
|
data = self._to_json(data) |
85
|
|
|
return NestedDotDict(data) |
86
|
|
|
|
87
|
|
|
def _to_json(self, xml): |
88
|
|
|
response = {} |
89
|
|
|
for child in list(xml): |
90
|
|
|
if len(list(child)) > 0: |
91
|
|
|
response[child.tag] = self._to_json(child) |
92
|
|
|
else: |
93
|
|
|
response[child.tag] = child.text or "" |
94
|
|
|
return response |
95
|
|
|
|
96
|
|
|
|
97
|
|
|
class CachingHmdbApi(JsonBackedHmdbApi): |
|
|
|
|
98
|
|
|
def __init__( |
99
|
|
|
self, query: Optional[QueryingHmdbApi], cache_dir: Path = MANDOS_SETTINGS.hmdb_cache_path |
|
|
|
|
100
|
|
|
): |
101
|
|
|
self._query = query |
102
|
|
|
self._cache_dir = cache_dir |
103
|
|
|
|
104
|
|
|
def fetch(self, hmdb_id: str) -> NestedDotDict: |
105
|
|
|
path = self.path(hmdb_id) |
106
|
|
|
if not path.exists(): |
|
|
|
|
107
|
|
|
return self._read_json(path) |
108
|
|
|
else: |
109
|
|
|
data = self._query.fetch(hmdb_id) |
110
|
|
|
self._write_json(data, path) |
111
|
|
|
return data |
112
|
|
|
|
113
|
|
|
def path(self, hmdb_id: str): |
|
|
|
|
114
|
|
|
return (self._cache_dir / hmdb_id).with_suffix(".json.gz") |
115
|
|
|
|
116
|
|
|
def _write_json(self, data: NestedDotDict, path: Path) -> None: |
|
|
|
|
117
|
|
|
path.write_bytes(gzip.compress(data.to_json())) |
118
|
|
|
|
119
|
|
|
def _read_json(self, path: Path) -> NestedDotDict: |
|
|
|
|
120
|
|
|
deflated = gzip.decompress(path.read_bytes()) |
121
|
|
|
read = orjson.loads(deflated) |
122
|
|
|
return NestedDotDict(read) |
123
|
|
|
|
124
|
|
|
|
125
|
|
|
__all__ = ["HmdbApi", "QueryingHmdbApi", "HmdbProperty"] |
126
|
|
|
|