forked from openml/openml-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathutils.py
More file actions
416 lines (344 loc) · 13 KB
/
utils.py
File metadata and controls
416 lines (344 loc) · 13 KB
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
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
# License: BSD 3-Clause
import os
import xmltodict
import shutil
from typing import TYPE_CHECKING, List, Tuple, Union, Type
import warnings
import pandas as pd
from functools import wraps
import collections
import openml
import openml._api_calls
import openml.exceptions
from . import config
# Avoid import cycles: https://mypy.readthedocs.io/en/latest/common_issues.html#import-cycles
if TYPE_CHECKING:
from openml.base import OpenMLBase
oslo_installed = False
try:
# Currently, importing oslo raises a lot of warning that it will stop working
# under python3.8; remove this once they disappear
with warnings.catch_warnings():
warnings.simplefilter("ignore")
from oslo_concurrency import lockutils
oslo_installed = True
except ImportError:
pass
def extract_xml_tags(xml_tag_name, node, allow_none=True):
"""Helper to extract xml tags from xmltodict.
Parameters
----------
xml_tag_name : str
Name of the xml tag to extract from the node.
node : object
Node object returned by ``xmltodict`` from which ``xml_tag_name``
should be extracted.
allow_none : bool
If ``False``, the tag needs to exist in the node. Will raise a
``ValueError`` if it does not.
Returns
-------
object
"""
if xml_tag_name in node and node[xml_tag_name] is not None:
if isinstance(node[xml_tag_name], dict):
rval = [node[xml_tag_name]]
elif isinstance(node[xml_tag_name], str):
rval = [node[xml_tag_name]]
elif isinstance(node[xml_tag_name], list):
rval = node[xml_tag_name]
else:
raise ValueError("Received not string and non list as tag item")
return rval
else:
if allow_none:
return None
else:
raise ValueError("Could not find tag '%s' in node '%s'" % (xml_tag_name, str(node)))
def _get_rest_api_type_alias(oml_object: "OpenMLBase") -> str:
"""Return the alias of the openml entity as it is defined for the REST API."""
rest_api_mapping: List[Tuple[Union[Type, Tuple], str]] = [
(openml.datasets.OpenMLDataset, "data"),
(openml.flows.OpenMLFlow, "flow"),
(openml.tasks.OpenMLTask, "task"),
(openml.runs.OpenMLRun, "run"),
((openml.study.OpenMLStudy, openml.study.OpenMLBenchmarkSuite), "study"),
]
_, api_type_alias = [
(python_type, api_alias)
for (python_type, api_alias) in rest_api_mapping
if isinstance(oml_object, python_type)
][0]
return api_type_alias
def _tag_openml_base(oml_object: "OpenMLBase", tag: str, untag: bool = False):
api_type_alias = _get_rest_api_type_alias(oml_object)
_tag_entity(api_type_alias, oml_object.id, tag, untag)
def _tag_entity(entity_type, entity_id, tag, untag=False):
"""
Function that tags or untags a given entity on OpenML. As the OpenML
API tag functions all consist of the same format, this function covers
all entity types (currently: dataset, task, flow, setup, run). Could
be used in a partial to provide dataset_tag, dataset_untag, etc.
Parameters
----------
entity_type : str
Name of the entity to tag (e.g., run, flow, data)
entity_id : int
OpenML id of the entity
tag : str
The tag
untag : bool
Set to true if needed to untag, rather than tag
Returns
-------
tags : list
List of tags that the entity is (still) tagged with
"""
legal_entities = {"data", "task", "flow", "setup", "run"}
if entity_type not in legal_entities:
raise ValueError("Can't tag a %s" % entity_type)
uri = "%s/tag" % entity_type
main_tag = "oml:%s_tag" % entity_type
if untag:
uri = "%s/untag" % entity_type
main_tag = "oml:%s_untag" % entity_type
post_variables = {"%s_id" % entity_type: entity_id, "tag": tag}
result_xml = openml._api_calls._perform_api_call(uri, "post", post_variables)
result = xmltodict.parse(result_xml, force_list={"oml:tag"})[main_tag]
if "oml:tag" in result:
return result["oml:tag"]
else:
# no tags, return empty list
return []
def _delete_entity(entity_type, entity_id):
"""
Function that deletes a given entity on OpenML. As the OpenML
API tag functions all consist of the same format, this function covers
all entity types that can be deleted (currently: dataset, task, flow,
run, study and user).
Parameters
----------
entity_type : str
Name of the entity to tag (e.g., run, flow, data)
entity_id : int
OpenML id of the entity
Returns
-------
bool
True iff the deletion was successful. False otherwse
"""
legal_entities = {
"data",
"flow",
"task",
"run",
"study",
"user",
}
if entity_type not in legal_entities:
raise ValueError("Can't delete a %s" % entity_type)
url_suffix = "%s/%d" % (entity_type, entity_id)
try:
result_xml = openml._api_calls._perform_api_call(url_suffix, "delete")
result = xmltodict.parse(result_xml)
return f"oml:{entity_type}_delete" in result
except openml.exceptions.OpenMLServerException as e:
# https://github.com/openml/OpenML/blob/21f6188d08ac24fcd2df06ab94cf421c946971b0/openml_OS/views/pages/api_new/v1/xml/pre.php
# Most exceptions are descriptive enough to be raised as their standard
# OpenMLServerException, however there are two cases where we add information:
# - a generic "failed" message, we direct them to the right issue board
# - when the user successfully authenticates with the server,
# but user is not allowed to take the requested action,
# in which case we specify a OpenMLNotAuthorizedError.
by_other_user = [323, 353, 393, 453, 594]
has_dependent_entities = [324, 326, 327, 328, 354, 454, 464, 595]
unknown_reason = [325, 355, 394, 455, 593]
if e.code in by_other_user:
raise openml.exceptions.OpenMLNotAuthorizedError(
message=(
f"The {entity_type} can not be deleted because it was not uploaded by you."
),
) from e
if e.code in has_dependent_entities:
raise openml.exceptions.OpenMLNotAuthorizedError(
message=(
f"The {entity_type} can not be deleted because "
f"it still has associated entities: {e.message}"
)
) from e
if e.code in unknown_reason:
raise openml.exceptions.OpenMLServerError(
message=(
f"The {entity_type} can not be deleted for unknown reason,"
" please open an issue at: https://github.com/openml/openml/issues/new"
),
) from e
raise
def _list_all(listing_call, output_format="dict", *args, **filters):
"""Helper to handle paged listing requests.
Example usage:
``evaluations = list_all(list_evaluations, "predictive_accuracy", task=mytask)``
Parameters
----------
listing_call : callable
Call listing, e.g. list_evaluations.
output_format : str, optional (default='dict')
The parameter decides the format of the output.
- If 'dict' the output is a dict of dict
- If 'dataframe' the output is a pandas DataFrame
*args : Variable length argument list
Any required arguments for the listing call.
**filters : Arbitrary keyword arguments
Any filters that can be applied to the listing function.
additionally, the batch_size can be specified. This is
useful for testing purposes.
Returns
-------
dict or dataframe
"""
# eliminate filters that have a None value
active_filters = {key: value for key, value in filters.items() if value is not None}
page = 0
result = collections.OrderedDict()
if output_format == "dataframe":
result = pd.DataFrame()
# Default batch size per paging.
# This one can be set in filters (batch_size), but should not be
# changed afterwards. The derived batch_size can be changed.
BATCH_SIZE_ORIG = 10000
if "batch_size" in active_filters:
BATCH_SIZE_ORIG = active_filters["batch_size"]
del active_filters["batch_size"]
# max number of results to be shown
LIMIT = None
offset = 0
if "size" in active_filters:
LIMIT = active_filters["size"]
del active_filters["size"]
if LIMIT is not None and BATCH_SIZE_ORIG > LIMIT:
BATCH_SIZE_ORIG = LIMIT
if "offset" in active_filters:
offset = active_filters["offset"]
del active_filters["offset"]
batch_size = BATCH_SIZE_ORIG
while True:
try:
current_offset = offset + BATCH_SIZE_ORIG * page
new_batch = listing_call(
*args,
limit=batch_size,
offset=current_offset,
output_format=output_format,
**active_filters,
)
except openml.exceptions.OpenMLServerNoResult:
# we want to return an empty dict in this case
break
if output_format == "dataframe":
if len(result) == 0:
result = new_batch
else:
result = pd.concat([result, new_batch], ignore_index=True)
else:
# For output_format = 'dict' or 'object'
result.update(new_batch)
if len(new_batch) < batch_size:
break
page += 1
if LIMIT is not None:
# check if the number of required results has been achieved
# always do a 'bigger than' check,
# in case of bugs to prevent infinite loops
if len(result) >= LIMIT:
break
# check if there are enough results to fulfill a batch
if BATCH_SIZE_ORIG > LIMIT - len(result):
batch_size = LIMIT - len(result)
return result
def _get_cache_dir_for_key(key):
cache = config.get_cache_directory()
return os.path.join(cache, key)
def _create_cache_directory(key):
cache_dir = _get_cache_dir_for_key(key)
try:
os.makedirs(cache_dir, exist_ok=True)
except Exception as e:
raise openml.exceptions.OpenMLCacheException(
f"Cannot create cache directory {cache_dir}."
) from e
return cache_dir
def _get_cache_dir_for_id(key, id_, create=False):
if create:
cache_dir = _create_cache_directory(key)
else:
cache_dir = _get_cache_dir_for_key(key)
return os.path.join(cache_dir, str(id_))
def _create_cache_directory_for_id(key, id_):
"""Create the cache directory for a specific ID
In order to have a clearer cache structure and because every task
is cached in several files (description, split), there
is a directory for each task witch the task ID being the directory
name. This function creates this cache directory.
This function is NOT thread/multiprocessing safe.
Parameters
----------
key : str
id_ : int
Returns
-------
str
Path of the created dataset cache directory.
"""
cache_dir = _get_cache_dir_for_id(key, id_, create=True)
if os.path.isdir(cache_dir):
pass
elif os.path.exists(cache_dir):
raise ValueError("%s cache dir exists but is not a directory!" % key)
else:
os.makedirs(cache_dir)
return cache_dir
def _remove_cache_dir_for_id(key, cache_dir):
"""Remove the task cache directory
This function is NOT thread/multiprocessing safe.
Parameters
----------
key : str
cache_dir : str
"""
try:
shutil.rmtree(cache_dir)
except (OSError, IOError):
raise ValueError(
"Cannot remove faulty %s cache directory %s."
"Please do this manually!" % (key, cache_dir)
)
def thread_safe_if_oslo_installed(func):
if oslo_installed:
@wraps(func)
def safe_func(*args, **kwargs):
# Lock directories use the id that is passed as either positional or keyword argument.
id_parameters = [parameter_name for parameter_name in kwargs if "_id" in parameter_name]
if len(id_parameters) == 1:
id_ = kwargs[id_parameters[0]]
elif len(args) > 0:
id_ = args[0]
else:
raise RuntimeError(
"An id must be specified for {}, was passed: ({}, {}).".format(
func.__name__, args, kwargs
)
)
# The [7:] gets rid of the 'openml.' prefix
lock_name = "{}.{}:{}".format(func.__module__[7:], func.__name__, id_)
with lockutils.external_lock(name=lock_name, lock_path=_create_lockfiles_dir()):
return func(*args, **kwargs)
return safe_func
else:
return func
def _create_lockfiles_dir():
dir = os.path.join(config.get_cache_directory(), "locks")
try:
os.makedirs(dir)
except OSError:
pass
return dir