publish flow after running it

Author: mfeurer

openml/exceptions.py

@@ -1,13 +1,14 @@
 class PyOpenMLError(Exception):
     def __init__(self, message):
+        self.message = message
         super(PyOpenMLError, self).__init__(message)
 
+
 class OpenMLServerError(PyOpenMLError):
     """class for when something is really wrong on the server
        (result did not parse to dict), contains unparsed error."""
 
     def __init__(self, message):
-        message = "OpenML Server error: " + message
         super(OpenMLServerError, self).__init__(message)
 
 #
@@ -18,7 +19,6 @@ class OpenMLServerException(OpenMLServerError):
     def __init__(self, code, message, additional=None):
         self.code = code
         self.additional = additional
-        message = "OpenML Server exception: " + message
         super(OpenMLServerException, self).__init__(message)
 
 

openml/flows/functions.py

@@ -145,7 +145,8 @@ def _check_flow_for_server_id(flow):
                 stack.append(component)
 
 
-def assert_flows_equal(flow1, flow2, ignore_parameters_on_older_children=None):
+def assert_flows_equal(flow1, flow2, ignore_parameters_on_older_children=None,
+                       ignore_parameters=False):
     """Check equality of two flows.
 
     Two flows are equal if their all keys which are not set by the server
@@ -160,6 +161,9 @@ def assert_flows_equal(flow1, flow2, ignore_parameters_on_older_children=None):
     ignore_parameters_on_older_children : str
         If set to ``OpenMLFlow.upload_date``, ignores parameters in a child
         flow if it's upload date predates the upload date of the parent flow.
+
+    ignore_parameters : bool
+        Whether to ignore parameter values when comparing flows.
     """
     if not isinstance(flow1, OpenMLFlow):
         raise TypeError('Argument 1 must be of type OpenMLFlow, but is %s' %
@@ -199,6 +203,8 @@ def assert_flows_equal(flow1, flow2, ignore_parameters_on_older_children=None):
                         ignore_parameters_on_older_children)
                     if upload_date_current_flow < upload_date_parent_flow:
                         continue
+                elif ignore_parameters:
+                    continue
 
             if attr1 != attr2:
                 raise ValueError("Flow %s: values for attribute '%s' differ: "

openml/runs/functions.py

@@ -11,6 +11,7 @@
 import six
 import xmltodict
 
+import openml
 from ..exceptions import PyOpenMLError
 from .. import config
 from ..flows import sklearn_to_flow, get_flow, flow_exists, _check_n_jobs, \
@@ -30,20 +31,6 @@ def run_model_on_task(task, model, avoid_duplicate_runs=True, flow_tags=None,
                       seed=None):
     flow = sklearn_to_flow(model)
 
-    # returns flow id if the flow exists on the server, False otherwise
-    flow_id = flow_exists(flow.name, flow.external_version)
-
-    if flow_id == False:
-        # TODO this is potential race condition! someone could upload the
-        # same flow in the meantime!
-        # means the flow did not exists. As we could run it, publish it now
-        flow = flow.publish()
-    else:
-        # flow already existed, download it from server
-        # TODO (neccessary? is this a post condition of this function)
-        flow_from_server = get_flow(flow_id)
-        _copy_server_fields(flow_from_server, flow)
-
     return run_flow_on_task(task=task, flow=flow,
                             avoid_duplicate_runs=avoid_duplicate_runs,
                             flow_tags=flow_tags, seed=seed)
@@ -82,6 +69,9 @@ def run_flow_on_task(task, flow, avoid_duplicate_runs=True, flow_tags=None,
     # skips the run if it already exists and the user opts for this in the config file.
     # also, if the flow is not present on the server, the check is not needed.
     if avoid_duplicate_runs:
+        if flow.flow_id is None:
+            raise ValueError('Cannot check if a run exists if the '
+                             'corresponding flow has not been published yet!')
         flow_from_server = get_flow(flow.flow_id)
         setup_id = setup_exists(flow_from_server)
         ids = _run_exists(task.task_id, setup_id)
@@ -98,18 +88,43 @@ def run_flow_on_task(task, flow, avoid_duplicate_runs=True, flow_tags=None,
 
     run_environment = _get_version_information()
     tags = ['openml-python', run_environment[1]]
+
     # execute the run
+    res = _run_task_get_arffcontent(flow.model, task, class_labels)
+
+    if flow.flow_id is None:
+        _publish_flow_if_necessary(flow)
+
     run = OpenMLRun(task_id=task.task_id, flow_id=flow.flow_id,
                     dataset_id=dataset.dataset_id, model=flow.model, tags=tags)
     run.parameter_settings = OpenMLRun._parse_parameters(flow)
-    res = _run_task_get_arffcontent(flow.model, task, class_labels)
+
     run.data_content, run.trace_content, run.trace_attributes, run.detailed_evaluations = res
 
     config.logger.info('Executed Task %d with Flow id: %d' % (task.task_id, run.flow_id))
 
     return run
 
 
+def _publish_flow_if_necessary(flow):
+    # try publishing the flow if one has to assume it doesn't exist yet. It
+    # might fail because it already exists, then the flow is currently not
+    # reused
+
+        try:
+            flow.publish()
+        except OpenMLServerException as e:
+            if e.message == "flow already exists":
+                flow_id = openml.flows.flow_exists(flow.name,
+                                                   flow.external_version)
+                server_flow = get_flow(flow_id)
+                openml.flows.flow._copy_server_fields(server_flow, flow)
+                openml.flows.assert_flows_equal(flow, server_flow,
+                                                ignore_parameters=True)
+            else:
+                raise e
+
+
 def get_run_trace(run_id):
     """Get the optimization trace object for a given run id.
 

tests/test_runs/test_run_functions.py

@@ -180,6 +180,17 @@ def _check_detailed_evaluations(self, detailed_evaluations, num_repeats, num_fol
                         self.assertGreater(evaluation, 0) # should take at least one millisecond (?)
                         self.assertLess(evaluation, max_time_allowed)
 
+    def test_avoid_duplicate_runs_with_unpublished_flow(self):
+        task_id = 115
+
+        clf = LogisticRegression()
+        flow = sklearn_to_flow(clf)
+        flow, _ = self._add_sentinel_to_flow_name(flow, None)
+        task = openml.tasks.get_task(task_id)
+
+        self.assertRaisesRegexp(ValueError, 'Cannot check if a run exists if the corresponding flow has not been published yet!',
+                                openml.runs.run_flow_on_task, task=task,
+                                flow=flow, avoid_duplicate_runs=True)
 
     def test_run_regression_on_classif_task(self):
         task_id = 115
@@ -202,6 +213,21 @@ def test_check_erronous_sklearn_flow_fails(self):
                                 openml.runs.run_model_on_task, task=task,
                                 model=clf)
 
+    def test__publish_flow_if_necessary(self):
+        task_id = 115
+        task = openml.tasks.get_task(task_id)
+
+        clf = LogisticRegression()
+        flow = sklearn_to_flow(clf)
+        flow, sentinel = self._add_sentinel_to_flow_name(flow, None)
+        openml.runs.functions._publish_flow_if_necessary(flow)
+        self.assertIsNotNone(flow.flow_id)
+
+        flow2 = sklearn_to_flow(clf)
+        flow2, _ = self._add_sentinel_to_flow_name(flow2, sentinel)
+        openml.runs.functions._publish_flow_if_necessary(flow2)
+        self.assertEqual(flow2.flow_id, flow.flow_id)
+
     def test_run_and_upload(self):
         # This unit test is ment to test the following functions, using a varity of flows:
         # - openml.runs.run_task()
@@ -277,7 +303,6 @@ def test_run_and_upload(self):
             self._check_detailed_evaluations(run.detailed_evaluations, 1, num_folds)
             pass
 
-
     def test_initialize_model_from_run(self):
         clf = sklearn.pipeline.Pipeline(steps=[('Imputer', Imputer(strategy='median')),
                                                ('VarianceThreshold', VarianceThreshold(threshold=0.05)),