added serialize run functionality

openml/runs/functions.py

@@ -1,4 +1,4 @@
-from collections import defaultdict
+import collections
 import io
 import json
 import os
@@ -33,18 +33,19 @@
 
 
 def run_model_on_task(task, model, avoid_duplicate_runs=True, flow_tags=None,
- seed=None):
+ seed=None, add_local_measures=True):
  """See ``run_flow_on_task for a documentation``."""
 
  flow = sklearn_to_flow(model)
 
  return run_flow_on_task(task=task, flow=flow,
  avoid_duplicate_runs=avoid_duplicate_runs,
- flow_tags=flow_tags, seed=seed)
+ flow_tags=flow_tags, seed=seed,
+ add_local_measures=add_local_measures)
 
 
 def run_flow_on_task(task, flow, avoid_duplicate_runs=True, flow_tags=None,
- seed=None):
+ seed=None, add_local_measures=True):
  """Run the model provided by the flow on the dataset defined by task.
 
  Takes the flow and repeat information into account. In case a flow is not
@@ -68,6 +69,9 @@ def run_flow_on_task(task, flow, avoid_duplicate_runs=True, flow_tags=None,
  A list of tags that the flow should have at creation.
  seed: int
  Models that are not seeded will get this seed.
+ add_local_measures : bool
+ Determines whether to calculate a set of evaluation measures locally,
+ to later verify server behaviour. Defaults to True
 
  Returns
  -------
@@ -100,7 +104,7 @@ def run_flow_on_task(task, flow, avoid_duplicate_runs=True, flow_tags=None,
  tags = ['openml-python', run_environment[1]]
 
  # execute the run
- res = _run_task_get_arffcontent(flow.model, task)
+ res = _run_task_get_arffcontent(flow.model, task, add_local_measures=add_local_measures)
 
  # in case the flow not exists, we will get a "False" back (which can be
  if not isinstance(flow.flow_id, int) or flow_id == False:
@@ -368,7 +372,7 @@ def _prediction_to_row(rep_no, fold_no, sample_no, row_id, correct_label,
  return arff_line
 
 
-def _run_task_get_arffcontent(model, task):
+def _run_task_get_arffcontent(model, task, add_local_measures):
 
  def _prediction_to_probabilities(y, model_classes):
  # y: list or numpy array of predictions
@@ -387,11 +391,11 @@ def _prediction_to_probabilities(y, model_classes):
  # this information is multiple times overwritten, but due to the ordering
  # of tne loops, eventually it contains the information based on the full
  # dataset size
- user_defined_measures_per_fold = defaultdict(lambda: defaultdict(dict))
+ user_defined_measures_per_fold = collections.defaultdict(lambda: collections.defaultdict(dict))
  # stores sample-based evaluation measures (sublevel of fold-based)
  # will also be filled on a non sample-based task, but the information
  # is the same as the fold-based measures, and disregarded in that case
- user_defined_measures_per_sample = defaultdict(lambda: defaultdict(lambda: defaultdict(dict)))
+ user_defined_measures_per_sample = collections.defaultdict(lambda: collections.defaultdict(lambda: collections.defaultdict(dict)))
 
  # sys.version_info returns a tuple, the following line compares the entry of tuples
  # https://docs.python.org/3.6/reference/expressions.html#value-comparisons
@@ -404,7 +408,9 @@ def _prediction_to_probabilities(y, model_classes):
  for fold_no in range(num_folds):
  for sample_no in range(num_samples):
  model_fold = sklearn.base.clone(model, safe=True)
- res =_run_model_on_fold(model_fold, task, rep_no, fold_no, sample_no, can_measure_runtime)
+ res = _run_model_on_fold(model_fold, task, rep_no, fold_no, sample_no,
+ can_measure_runtime=can_measure_runtime,
+ add_local_measures=add_local_measures)
  arff_datacontent_fold, arff_tracecontent_fold, user_defined_measures_fold, model_fold = res
 
  arff_datacontent.extend(arff_datacontent_fold)
@@ -430,7 +436,7 @@ def _prediction_to_probabilities(y, model_classes):
  user_defined_measures_per_sample
 
 
-def _run_model_on_fold(model, task, rep_no, fold_no, sample_no, can_measure_runtime):
+def _run_model_on_fold(model, task, rep_no, fold_no, sample_no, can_measure_runtime, add_local_measures):
  """Internal function that executes a model on a fold (and possibly
  subsample) of the dataset. It returns the data that is necessary
  to construct the OpenML Run object (potentially over more than
@@ -455,6 +461,9 @@ def _run_model_on_fold(model, task, rep_no, fold_no, sample_no, can_measure_runt
  can_measure_runtime : bool
  Wether we are allowed to measure runtime (requires: Single node
  computation and Python >= 3.3)
+ add_local_measures : bool
+ Determines whether to calculate a set of measures (i.e., predictive
+ accuracy) locally, to later verify server behaviour
 
  Returns
  -------
@@ -547,7 +556,8 @@ def _prediction_to_probabilities(y, model_classes):
  def _calculate_local_measure(sklearn_fn, openml_name):
  user_defined_measures[openml_name] = sklearn_fn(testY, PredY)
 
- _calculate_local_measure(sklearn.metrics.accuracy_score, 'predictive_accuracy')
+ if add_local_measures:
+ _calculate_local_measure(sklearn.metrics.accuracy_score, 'predictive_accuracy')
 
  arff_datacontent = []
  for i in range(0, len(test_indices)):
@@ -694,7 +704,7 @@ def obtain_field(xml_obj, fieldname, from_server, cast=None):
  else:
  raise AttributeError('Run XML does not contain required (server) field: ', fieldname)
 
- run = xmltodict.parse(xml, force_list=['oml:file', 'oml:evaluation'])["oml:run"]
+ run = xmltodict.parse(xml, force_list=['oml:file', 'oml:evaluation', 'oml:parameter_setting'])["oml:run"]
  run_id = obtain_field(run, 'oml:run_id', from_server, cast=int)
  uploader = obtain_field(run, 'oml:uploader', from_server, cast=int)
  uploader_name = obtain_field(run, 'oml:uploader_name', from_server)
@@ -712,13 +722,16 @@ def obtain_field(xml_obj, fieldname, from_server, cast=None):
  setup_id = obtain_field(run, 'oml:setup_id', from_server, cast=int)
  setup_string = obtain_field(run, 'oml:setup_string', from_server)
 
- parameters = dict()
- if 'oml:parameter_settings' in run:
- parameter_settings = run['oml:parameter_settings']
- for parameter_dict in parameter_settings:
- key = parameter_dict['oml:name']
- value = parameter_dict['oml:value']
- parameters[key] = value
+ parameters = []
+ if 'oml:parameter_setting' in run:
+ obtained_parameter_settings = run['oml:parameter_setting']
+ for parameter_dict in obtained_parameter_settings:
+ current_parameter = collections.OrderedDict()
+ current_parameter['oml:name'] = parameter_dict['oml:name']
+ current_parameter['oml:value'] = parameter_dict['oml:value']
+ if 'oml:component' in parameter_dict:
+ current_parameter['oml:component'] = parameter_dict['oml:component']
+ parameters.append(current_parameter)
 
  if 'oml:input_data' in run:
  dataset_id = int(run['oml:input_data']['oml:dataset']['oml:did'])
@@ -727,10 +740,11 @@ def obtain_field(xml_obj, fieldname, from_server, cast=None):
 
  files = dict()
  evaluations = dict()
- fold_evaluations = defaultdict(lambda: defaultdict(dict))
- sample_evaluations = defaultdict(lambda: defaultdict(lambda: defaultdict(dict)))
+ fold_evaluations = collections.defaultdict(lambda: collections.defaultdict(dict))
+ sample_evaluations = collections.defaultdict(lambda: collections.defaultdict(lambda: collections.defaultdict(dict)))
  if 'oml:output_data' not in run:
- raise ValueError('Run does not contain output_data (OpenML server error?)')
+ if from_server:
+ raise ValueError('Run does not contain output_data (OpenML server error?)')
  else:
  output_data = run['oml:output_data']
  if 'oml:file' in output_data:

openml/runs/run.py

@@ -1,10 +1,13 @@
 from collections import OrderedDict
+import errno
 import json
+import pickle
 import sys
 import time
 import numpy as np
 
 import arff
+import os
 import xmltodict
 
 import openml
@@ -65,6 +68,96 @@ def __str__(self):
  def _repr_pretty_(self, pp, cycle):
  pp.text(str(self))
 
+ @classmethod
+ def from_filesystem(cls, folder):
+ """
+ The inverse of the to_filesystem method. Instantiates an OpenMLRun
+ object based on files stored on the file system.
+
+ Parameters
+ ----------
+ folder : str
+ a path leading to the folder where the results
+ are stored
+
+ Returns
+ -------
+ run : OpenMLRun
+ the re-instantiated run object
+ """
+ if not os.path.isdir(folder):
+ raise ValueError('Could not find folder')
+
+ description_path = os.path.join(folder, 'description.xml')
+ predictions_path = os.path.join(folder, 'predictions.arff')
+ trace_path = os.path.join(folder, 'trace.arff')
+ model_path = os.path.join(folder, 'model.pkl')
+
+ if not os.path.isfile(description_path):
+ raise ValueError('Could not find description.xml')
+ if not os.path.isfile(predictions_path):
+ raise ValueError('Could not find predictions.arff')
+ if not os.path.isfile(model_path):
+ raise ValueError('Could not find model.pkl')
+
+ with open(description_path, 'r') as fp:
+ run = openml.runs.functions._create_run_from_xml(fp.read(), from_server=False)
+
+ with open(predictions_path, 'r') as fp:
+ predictions = arff.load(fp)
+ run.data_content = predictions['data']
+
+ with open(model_path, 'rb') as fp:
+ run.model = pickle.load(fp)
+
+ if os.path.isfile(trace_path):
+ with open(trace_path, 'r') as fp:
+ trace = arff.load(fp)
+ run.trace_attributes = trace['attributes']
+ run.trace_content = trace['data']
+
+ return run
+
+ def to_filesystem(self, output_directory):
+ """
+ The inverse of the from_filesystem method. Serializes a run
+ on the filesystem, to be uploaded later.
+
+ Parameters
+ ----------
+ folder : str
+ a path leading to the folder where the results
+ will be stored. Should be empty
+ """
+ if self.data_content is None or self.model is None:
+ raise ValueError('Run should have been executed (and contain model / predictions)')
+
+ try:
+ os.makedirs(output_directory)
+ except OSError as e:
+ if e.errno == errno.EEXIST:
+ pass
+ else:
+ raise e
+
+ if not os.listdir(output_directory) == []:
+ raise ValueError('Output directory should be empty')
+
+ run_xml = self._create_description_xml()
+ predictions_arff = arff.dumps(self._generate_arff_dict())
+
+ with open(os.path.join(output_directory, 'description.xml'), 'w') as f:
+ f.write(run_xml)
+ with open(os.path.join(output_directory, 'predictions.arff'), 'w') as f:
+ f.write(predictions_arff)
+ with open(os.path.join(output_directory, 'model.pkl'), 'wb') as f:
+ pickle.dump(self.model, f)
+
+ if self.trace_content is not None:
+ trace_arff = arff.dumps(self._generate_trace_arff_dict())
+ with open(os.path.join(output_directory, 'trace.arff'), 'w') as f:
+ f.write(trace_arff)
+
  def _generate_arff_dict(self):
  """Generates the arff dictionary for uploading predictions to the server.
 
@@ -109,11 +202,11 @@ def _generate_trace_arff_dict(self):
  Contains information about the optimization trace.
  """
  if self.trace_content is None or len(self.trace_content) == 0:
- raise ValueError('No trace content avaiable.')
+ raise ValueError('No trace content available.')
  if len(self.trace_attributes) != len(self.trace_content[0]):
  raise ValueError('Trace_attributes and trace_content not compatible')
 
- arff_dict = {}
+ arff_dict = dict()
  arff_dict['attributes'] = self.trace_attributes
  arff_dict['data'] = self.trace_content
  arff_dict['relation'] = 'openml_task_' + str(self.task_id) + '_predictions'
@@ -447,7 +540,8 @@ def _to_dict(taskid, flow_id, setup_string, error_message, parameter_settings,
  description['oml:run']['oml:parameter_setting'] = parameter_settings
  if tags is not None:
  description['oml:run']['oml:tag'] = tags # Tags describing the run
- if fold_evaluations is not None or sample_evaluations is not None:
+ if (fold_evaluations is not None and len(fold_evaluations) > 0) or \
+ (sample_evaluations is not None and len(sample_evaluations) > 0):
  description['oml:run']['oml:output_data'] = dict()
  description['oml:run']['oml:output_data']['oml:evaluation'] = list()
  if fold_evaluations is not None:

tests/test_runs/test_run.py

@@ -1,8 +1,12 @@
+import numpy as np
+import random
+import os
 from time import time
 
+from sklearn.tree import DecisionTreeClassifier
 from sklearn.ensemble import RandomForestClassifier, AdaBoostClassifier
 from sklearn.linear_model import LogisticRegression
-from sklearn.model_selection import RandomizedSearchCV, StratifiedKFold
+from sklearn.model_selection import GridSearchCV, RandomizedSearchCV, StratifiedKFold
 
 from openml.testing import TestBase
 from openml.flows.sklearn_converter import sklearn_to_flow
@@ -66,3 +70,58 @@ def test_tagging(self):
  run.remove_tag(tag)
  run_list = openml.runs.list_runs(tag=tag)
  self.assertEqual(len(run_list), 0)
+
+ def _test_run_obj_equals(self, run, run_prime):
+ for dictionary in ['evaluations', 'fold_evaluations', 'sample_evaluations']:
+ if getattr(run, dictionary) is not None:
+ self.assertDictEqual(getattr(run, dictionary), getattr(run_prime, dictionary))
+ else:
+ # should be none or empty
+ other = getattr(run_prime, dictionary)
+ if other is not None:
+ self.assertDictEqual(other, dict())
+ self.assertEqual(run._create_description_xml(), run_prime._create_description_xml())
+
+ numeric_part = np.array(np.array(run.data_content)[:, 0:-2], dtype=float)
+ numeric_part_prime = np.array(np.array(run_prime.data_content)[:, 0:-2], dtype=float)
+ string_part = np.array(run.data_content)[:, -2:]
+ string_part_prime = np.array(run_prime.data_content)[:, -2:]
+ # JvR: Python 2.7 requires an almost equal check, rather than an equals check
+ np.testing.assert_array_almost_equal(numeric_part, numeric_part_prime)
+ np.testing.assert_array_equal(string_part, string_part_prime)
+
+ if run.trace_content is not None:
+ numeric_part = np.array(np.array(run.trace_content)[:, 0:-2], dtype=float)
+ numeric_part_prime = np.array(np.array(run_prime.trace_content)[:, 0:-2], dtype=float)
+ string_part = np.array(run.trace_content)[:, -2:]
+ string_part_prime = np.array(run_prime.trace_content)[:, -2:]
+ # JvR: Python 2.7 requires an almost equal check, rather than an equals check
+ np.testing.assert_array_almost_equal(numeric_part, numeric_part_prime)
+ np.testing.assert_array_equal(string_part, string_part_prime)
+ else:
+ self.assertIsNone(run_prime.trace_content)
+
+ def test_to_from_filesystem_vanilla(self):
+ model = DecisionTreeClassifier(max_depth=1)
+ task = openml.tasks.get_task(119)
+ run = openml.runs.run_model_on_task(task, model, add_local_measures=False)
+
+ cache_path = os.path.join(self.workdir, 'runs', str(random.getrandbits(128)))
+ run.to_filesystem(cache_path)
+
+ run_prime = openml.runs.OpenMLRun.from_filesystem(cache_path)
+ self._test_run_obj_equals(run, run_prime)
+ run_prime.publish()
+
+ def test_to_from_filesystem_search(self):
+ model = GridSearchCV(estimator=DecisionTreeClassifier(), param_grid={"max_depth": [1, 2, 3, 4, 5]})
+
+ task = openml.tasks.get_task(119)
+ run = openml.runs.run_model_on_task(task, model, add_local_measures=False)
+
+ cache_path = os.path.join(self.workdir, 'runs', str(random.getrandbits(128)))
+ run.to_filesystem(cache_path)
+
+ run_prime = openml.runs.OpenMLRun.from_filesystem(cache_path)
+ self._test_run_obj_equals(run, run_prime)
+ run_prime.publish()