Repository navigation
Expand file tree
/
Copy pathutils.py
More file actions
239 lines (195 loc) · 6.97 KB
/
Copy pathutils.py
File metadata and controls
239 lines (195 loc) · 6.97 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
import os
import xmltodict
import six
import shutil
import openml._api_calls
from . import config
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], six.string_types):
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 _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_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 _list_all(listing_call, *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.
*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
"""
# 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 = {}
# 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']
# check if the batch size is greater than the number of results that need to be returned.
if LIMIT is not None:
if BATCH_SIZE_ORIG > LIMIT:
BATCH_SIZE_ORIG = min(LIMIT, BATCH_SIZE_ORIG)
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,
**active_filters
)
except openml.exceptions.OpenMLServerNoResult:
# we want to return an empty dict in this case
break
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 _create_cache_directory(key):
cache = config.get_cache_directory()
cache_dir = os.path.join(cache, key)
try:
os.makedirs(cache_dir)
except:
pass
return cache_dir
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 = os.path.join(
_create_cache_directory(key), str(id_)
)
if os.path.exists(cache_dir) and os.path.isdir(cache_dir):
pass
elif os.path.exists(cache_dir) and not os.path.isdir(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 _create_lockfiles_dir():
dir = os.path.join(config.get_cache_directory(), 'locks')
try:
os.makedirs(dir)
except:
pass
return dir