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