Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_get_uncached_setup(self):
openml.config.cache_directory = self.static_cache_dir
with self.assertRaises(openml.exceptions.OpenMLCacheException):
openml.setups.functions._get_cached_setup(10)
def _get_cached_split(tid):
for cache_dir in [config.get_cache_dir(), config.get_private_dir()]:
task_cache_dir = os.path.join(cache_dir, "tasks")
split_file = os.path.join(task_cache_dir,
"tid_%d.arff" % int(tid))
try:
split = OpenMLSplit.from_arff_file(split_file)
return split
except (OSError, IOError):
continue
raise OpenMLCacheException("Split file for tid %d not "
"cached" % tid)
def _get_cached_setup(setup_id):
"""Load a run from the cache."""
cache_dir = config.get_cache_directory()
setup_cache_dir = os.path.join(cache_dir, "setups", str(setup_id))
try:
setup_file = os.path.join(setup_cache_dir, "description.xml")
with io.open(setup_file, encoding='utf8') as fh:
setup_xml = xmltodict.parse(fh.read())
setup = _create_setup_from_xml(setup_xml, output_format='object')
return setup
except (OSError, IOError):
raise openml.exceptions.OpenMLCacheException(
"Setup file for setup id %d not cached" % setup_id)
-------
OpenMLTask
"""
tid_cache_dir = openml.utils._create_cache_directory_for_id(
TASKS_CACHE_DIR_NAME,
tid
)
try:
with io.open(os.path.join(tid_cache_dir, "task.xml"), encoding='utf8')\
as fh:
return _create_task_from_xml(fh.read())
except (OSError, IOError):
openml.utils._remove_cache_dir_for_id(TASKS_CACHE_DIR_NAME,
tid_cache_dir)
raise OpenMLCacheException("Task file for tid %d not "
"cached" % tid)
def _get_cached_dataset_arff(dataset_id):
did_cache_dir = _create_cache_directory_for_id(
DATASETS_CACHE_DIR_NAME, dataset_id,
)
output_file = os.path.join(did_cache_dir, "dataset.arff")
try:
with io.open(output_file, encoding='utf8'):
pass
return output_file
except (OSError, IOError):
raise OpenMLCacheException("ARFF file for dataset id %d not "
"cached" % dataset_id)
def _get_cached_dataset_description(dataset_id):
did_cache_dir = _create_cache_directory_for_id(
DATASETS_CACHE_DIR_NAME, dataset_id,
)
description_file = os.path.join(did_cache_dir, "description.xml")
try:
with io.open(description_file, encoding='utf8') as fh:
dataset_xml = fh.read()
return xmltodict.parse(dataset_xml)["oml:data_set_description"]
except (IOError, OSError):
raise OpenMLCacheException(
"Dataset description for dataset id %d not "
"cached" % dataset_id)
Does the real work for get_flow. It returns a cached flow
instance if the flow exists locally, otherwise it downloads the
flow and returns an instance created from the xml representation.
Parameters
----------
flow_id : int
The OpenML flow id.
Returns
-------
OpenMLFlow
"""
try:
return _get_cached_flow(flow_id)
except OpenMLCacheException:
xml_file = os.path.join(
openml.utils._create_cache_directory_for_id(FLOWS_CACHE_DIR_NAME, flow_id),
"flow.xml",
)
flow_xml = openml._api_calls._perform_api_call("flow/%d" % flow_id, request_method='get')
with io.open(xml_file, "w", encoding='utf8') as fh:
fh.write(flow_xml)
return _create_flow_from_xml(flow_xml)
def _get_cached_dataset_qualities(dataset_id):
did_cache_dir = _create_cache_directory_for_id(
DATASETS_CACHE_DIR_NAME, dataset_id,
)
qualities_file = os.path.join(did_cache_dir, "qualities.xml")
try:
with io.open(qualities_file, encoding='utf8') as fh:
qualities_xml = fh.read()
qualities_dict = xmltodict.parse(qualities_xml)
return qualities_dict["oml:data_qualities"]['oml:quality']
except (IOError, OSError):
raise OpenMLCacheException("Dataset qualities for dataset id %d not "
"cached" % dataset_id)