-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathworms.py
218 lines (183 loc) · 8.05 KB
/
worms.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
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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
#!/usr/bin/env python
'''Client to access the web services of the WoRMS database (http://www.marinespecies.org/).
For a full description refer to: http://marinespecies.org/aphia.php?p=soap
Usage:
from worms import Aphia
aphia = Aphia()
results = aphia.get_aphia_records('Priapulus caudatus')
print(results)
Services:
[y] getAphiaID
[y] getAphiaRecords
[y] getAphiaNameByID
[y] getAphiaRecordByID
[ ] getAphiaRecordsByIDs
[y] getAphiaRecordByExtID
[y] getExtIDbyAphiaID
[y] getAphiaRecordsByNames
[y] getAphiaRecordsByVernacular
[y] getAphiaRecordsByDate
[y] getAphiaClassificationByID
[y] getSourcesByAphiaID
[y] getAphiaSynonymsByID
[y] getAphiaVernacularsByID
[y] getAphiaChildrenByID
[y] matchAphiaRecordsByNames
[y] getAphiaDistributionsByID
[ ] getAphiaTaxonRanksByID
[ ] getAphiaTaxonRanksByName
[ ] getAphiaRecordsByTaxonRankID
[ ] getAphiaAttributeKeysByID
[ ] getAphiaAttributeValuesByCategoryID
[ ] getAphiaIDsByAttributeKeyID
[ ] getAphiaAttributesByAphiaID
'''
# TODO: Migrate to REST
# TODO: Allow passing additional arguments
from suds import null, WebFault
from suds.client import Client
import logging
# Create logger.
logger = logging.getLogger('worms')
logger.setLevel(logging.DEBUG)
logger.propagate = False
formatter = logging.Formatter('[%(levelname)s] %(asctime)s @ %(module)s %(funcName)s (l%(lineno)d): %(message)s')
# Console handler for logger.
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.DEBUG)
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)
class Aphia:
'''WoRMS interactor.'''
def __init__(self):
self.url = 'http://www.marinespecies.org/aphia.php?p=soap&wsdl=1'
logger.info('Initiating contact with WoRMS...')
try:
self.client = Client(self.url)
logger.info('Connected to WoRMS web services.')
except:
print('Could not connect to client!')
def wire(self, service, query, attempt=0):
'''Manage re-connections between client and WoRMS.'''
try:
results = service(query)
except:
while attempt < 3:
logger.warning('Could not connect... try=%d' % attempt)
attempt += 1
self.wire(service, query, attempt)
logger.critical('Closing up the connection. I failed.')
results = None
return results
def get_aphia_id(self, query):
'''Get the AphiaID for a given name.'''
logger.info('Searching for the name "%s"', query)
results = self.wire(self.client.service.getAphiaID, query)
return results
def get_aphia_records(self, query):
'''Get one or more matching AphiaRecords for a given name.'''
logger.info('Searching for the name "%s"', query)
results = self.wire(self.client.service.getAphiaRecords, query)
self.show_results(results)
return results
def get_aphia_name_by_id(self, query):
'''Get the correct name for a given AphiaID.'''
logger.info('Searching for the ID "%s"', query)
results = self.wire(self.client.service.getAphiaNameByID, query)
return results
def get_aphia_record_by_id(self, query):
'''Get the complete AphiaRecord for a given AphiaID.'''
logger.info('Searching for the ID "%s"', query)
results = self.wire(self.client.service.getAphiaRecordByID, query)
return results
def get_aphia_record_by_external_id(self, query, dbtype=''):
'''Get the Aphia Record for a given external identifier.'''
#TODO define type parameter.
logger.info('Searching for the ID "%s"', query)
results = self.wire(self.client.service.getAphiaRecordByExtID, query)
return results
def get_external_id_by_aphia_id(self, query):
'''Get any external identifier(s) for a given AphiaID.'''
logger.info('Searching for the ID "%s"', query)
results = self.wire(self.client.service.getExtIDbyAphiaID, query)
return results
def get_aphia_records_by_names(self, query):
'''For each given scientific name, try to find one or more AphiaRecords.
This allows you to match multiple names in one call. Limited to 500
names at once for performance reasons.
'''
logger.info('Searching for the name(s) "%s"', query)
results = self.wire(self.client.service.getAphiaRecordsByNames, query)
return results
def get_aphia_records_by_vernacular(self, query):
'''Get one or more Aphia Records for a given vernacular.'''
logger.info('Searching for the name "%s"', query)
results = self.wire(self.client.service.getAphiaRecordsByVernacular, query)
return results
def get_aphia_records_by_date(self, query):
'''Lists all AphiaRecords (taxa) modified or added between a specific time interval.'''
#TODO Define startdate parameters.
logger.info('Searching between dates "%s"', query)
results = self.wire(self.client.service.getAphiaRecordsByDate, query)
return results
def get_aphia_classification_by_id(self, query):
'''Get the complete classification for one taxon.
This also includes any sub or super ranks.
'''
logger.info('Searching for the ID "%s"', query)
results = self.wire(self.client.service.getAphiaClassificationByID, query)
return results
def get_sources_by_aphia_id(self, query):
'''Get one or more sources/references including links, for one AphiaID.'''
logger.info('Searching for the ID "%s"', query)
results = self.wire(self.client.service.getSourcesByAphiaID, query)
return results
def get_aphia_synonyms_by_id(self, query):
'''Get all synonyms for a given AphiaID.'''
logger.info('Searching for the ID "%s"', query)
results = self.wire(self.client.service.getAphiaSynonymsByID, query)
return results
def get_aphia_vernaculars_by_id(self, query):
'''Get all vernaculars for a given AphiaID.'''
logger.info('Searching for the ID "%s"', query)
results = self.wire(self.client.service.getAphiaVernacularsByID, query)
return results
def get_aphia_children_by_id(self, query):
'''Get the direct children for a given AphiaID.'''
logger.info('Searching for the ID "%s"', query)
results = self.wire(self.client.service.getAphiaChildrenByID, query)
return results
def match_aphia_records_by_names(self, query):
'''For each given scientific name (may include authority), try to find one or more AphiaRecords, using the TAXAMATCH fuzzy matching algorithm by Tony Rees.
This allows you to (fuzzy) match multiple names in one call. Limited to
50 names at once for performance reasons.
'''
logger.info('Searching for the name(s) "%s"', query)
results = self.wire(self.client.service.matchAphiaRecordsByNames, query)
return results
def get_aphia_distributions_by_id(self, query):
'''Get all distributions for a given AphiaID.'''
logger.info('Searching for the ID "%s"', query)
results = self.wire(self.client.service.getAphiaDistributionsByID, query)
return results
def show_results(self, results):
'''Print relevant information about the results.'''
if results:
logger.info('Found {} record(s)'.format(len(results)))
for result in results:
self.print_record(result)
else:
logger.info('Found no records.')
def print_record(self, record, pre=''):
'''Print taxon information.'''
# logger.info('ID: {0}, Name: {1}, Authority: {2}, Rank: {3}, Status: {4}'.format(
logger.info('{0}{1} / {2} / {3} / {4} / {5}'.format(
pre,
record['AphiaID'],
record['scientificname'],
record['authority'],
record['rank'],
record['status'])
)
if __name__ == '__main__':
print('Command line not implemented yet.')