-
Notifications
You must be signed in to change notification settings - Fork 21
/
zooma.py
122 lines (99 loc) · 3.92 KB
/
zooma.py
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
import requests
from requests import HTTPError
from sdrf_pipelines.zooma.ols import OlsClient
class OlsTerm(object):
def __init__(self, iri: str = None, term: str = None, ontology: str = None) -> None:
self._iri = iri
self._term = term
self._ontology = ontology
def __str__(self) -> str:
return "{} -- {} -- {}".format(self._term, self._ontology, self._iri)
class SlimOlsClient(object):
def __init__(self) -> None:
super().__init__()
self._ols_client = OlsClient()
@staticmethod
def get_term_from_url(url, page_size: int = 100, ontology: str = None):
"""
Return a list of terms by ontology
:param url:
:param page_size:
:param ontology:
:return:
"""
url += "&" + "size=" + str(page_size)
r = requests.get(url)
if r.status_code == 414:
raise HTTPError('URL do not exist in OLS')
json_response = r.json()
old_terms = json_response['_embedded']['terms']
old_terms = list(filter(lambda k: ontology in k['ontology_name'], old_terms))
return [OlsTerm(x['iri'], x['label'], x['ontology_name']) for x in old_terms]
class Zooma(object):
"""
A Python binding of the Zooma REST API
(http://data.bioontology.org/documentation)
"""
BASE_URL = 'https://www.ebi.ac.uk/spot/zooma/v2/api/services'
@staticmethod
def process_zumma_results(results):
"""
Get a list of results from a query to Zooma and return a list
of dictionaries containing the queryValue, confidence and ols_url
:param results: List of query terms
:return:
"""
ontology_terms = []
for result in results:
ols_term = {'queryValue': result['annotatedProperty']['propertyValue'], 'confidence': result['confidence'],
'ols_url': result['_links']['olslinks'][0]['href']}
ontology_terms.append(ols_term)
return ontology_terms
def recommender(self, text_or_keywords, **kwargs):
"""
# https://www.ebi.ac.uk/spot/zooma/docs/api.html
Recommender provides a set of ontology terms that match the provided text.
:param text_or_keywords: keyword to search
:param kwargs: filters for ontologies
:return:
"""
endpoint = '/annotate'
full_url = Zooma.BASE_URL + endpoint
payload = kwargs
payload['propertyValue'] = text_or_keywords
return self._zooma_api_request(full_url, 'get', payload)
def _zooma_api_request(self, url, method, payload={}):
global r, error_message
processed_payload = self._process_payload(payload)
if method == 'get':
r = requests.get(url, params=processed_payload)
elif method == 'post':
r = requests.post(url, data=processed_payload)
if r.status_code == 414:
raise HTTPError('Text is too long.')
json_response = r.json()
try:
# This will raise an HTTPError if the HTTP request returned an
# unsuccessful status code.
r.raise_for_status()
except HTTPError:
if 'errors' in json_response.keys():
error_messages = json_response['errors']
error_message = '\n'.join(error_messages)
elif 'error' in json_response.keys():
error_message = json_response['error']
raise HTTPError(error_message)
return json_response
@staticmethod
def process_value(value):
if type(value) is bool:
return str(value).lower()
return value
def _process_payload(self, payload):
"""
Turn boolean True to str 'true' and False to str 'false'. Otherwise,
server will ignore argument with boolean value.
:param payload:
:return:
"""
return {key: self.process_value(value) for key, value in payload.items()}