Refactoring run_flow_on_task and doc add for run_model (#516)

* Documentation fix

* Add doc for run_model_on_task

* Initial additions

* Added functions to cache flows

* Tweaking a function from flow which will be used to create a task dict as a pre step for publish

* Undo 22b1e62.

* PEP8 compliance.

* Add (unused) flag to (not) upload flow. Rename get_seeded_model method as the name did not reflect the functionality.

* Add RunExistsError.

* RunsExistsError now correctly allows multiple runs, reflected in name.

* Towards offline run_model_on_task

* Fix name.

* Py3 style.

* Fix typo.

* Allow run flow locally. Caching and upload not implemented.

* Clean up test with new Error type.

* Check if flow exists before uploading.

* Remove one-line method that was only called from other method.

* Change error type. Add typehint.

* Fix imports.

* Publish flow if flow_id is None.

* Do not allow for mutable parameter.

* Fill in parameter_settings based on the referenced flow.

* Allow parameters to be extracted for model which is not part of the object.

* Can not use reinstantiated model.

* to/from filesystem methods.

* When (de)serializing, if a local flow was used, also (de)serialize the flow.

* When loading a locally stored run, do not force fields for which the flow is required to have been uploaded.

* Updated publish_error for new publish.

* Use mock for existing_flow

* Add documentation on the offline functionality.

* Disable two unit tests for now.

* Fix typo.

* PEP8.

* Remove old check.

* Update to reflect the change that uploading the flow is no longer default behavior.

* Fixed an error where non-existant flows still got the treatment to check for duplicates.

* Make tests actually fully local. Update for new parameter order.

* Type hints. Explicitly check for int rather than implicit cast of int to bool.

* Add errors for inconsistencies between local flows and server information.

* Now only sets hyperparameters if sync happened.

* Always sync with server if we know the flow to exist on the server.

* Update vanilla test. Add test for local flow upload after file stored to disk.

* Raise an error if `flow.publish` is called on a flow with different local id than the one known on the server.

* Add tests to verify identical behavior if run is loaded from disk instead.

* Line too long.

* Docs, typehint. Remove unused method publish_flow_is_necessary.

* Changed summary as suggested by @mfeurer.

* Type hints.

* Fix naming inconsistency between from_filesystem and to_filesystem.

* Updated for the new parametername.

* Function signature formatting improvements.

* Consistent spacing around colons. Add parameter description of `from_server`

* Add missing parenthesis.

* Doc changes, typehint.

* Remove check for flow as I think it is outdated.

* PrivateDatasetError and RunsExistError now prefixed with 'OpenML'

* Updated unit test to verify flows existence before/after run_model_on_task and publish.

* Start for testing model on downloaded flow.

* Explicit test for none as other __len__ can get invoked on some models to test for truthiness.

* Unit test now downloads flow after ensuring it exists.

* Test with run_flow_on_task instead so a sentinel can be added to the flow to ensure it does not exist on the server.

* Fixed a bug where run.flow_id would be set to False instead of None if associated flow did not exist but was also not uploaded. This gave errors at publish-time.

* Fix typo.

Author: ArlindKadra

examples/flows_and_runs_tutorial.py

@@ -89,6 +89,33 @@
 myrun = run.publish()
 print("Uploaded to http://test.openml.org/r/" + str(myrun.run_id))
 
+###############################################################################
+# Running flows on tasks offline for later upload
+# ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+# For those scenarios where there is no access to internet, it is possible to run
+# a model on a task without uploading results or flows to the server immediately.
+
+# To perform the following line offline, it is required to have been called before
+# such that the task is cached on the local openml cache directory:
+task = openml.tasks.get_task(6)
+
+# The following lines can then be executed offline:
+run = openml.runs.run_model_on_task(
+    pipe,
+    task,
+    avoid_duplicate_runs=False,
+    upload_flow=False)
+
+# The run may be stored offline, and the flow will be stored along with it:
+run.to_filesystem(directory='myrun')
+
+# They made later be loaded and uploaded
+run = openml.runs.OpenMLRun.from_filesystem(directory='myrun')
+run.publish()
+
+# Publishing the run will automatically upload the related flow if
+# it does not yet exist on the server.
+
 ############################################################################
 # Challenge
 # ^^^^^^^^^

openml/datasets/functions.py

@@ -24,7 +24,7 @@
     OpenMLCacheException,
     OpenMLHashException,
     OpenMLServerException,
-    PrivateDatasetError,
+    OpenMLPrivateDatasetError,
 )
 from ..utils import (
     _create_cache_directory,
@@ -360,7 +360,7 @@ def get_dataset(dataset_id):
             # if there was an exception,
             # check if the user had access to the dataset
             if e.code == 112:
-                raise PrivateDatasetError(e.message) from None
+                raise OpenMLPrivateDatasetError(e.message) from None
             else:
                 raise e
         finally:

openml/exceptions.py

@@ -1,29 +1,29 @@
 class PyOpenMLError(Exception):
-    def __init__(self, message):
+    def __init__(self, message: str):
         self.message = message
-        super(PyOpenMLError, self).__init__(message)
+        super().__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):
-        super(OpenMLServerError, self).__init__(message)
+    def __init__(self, message: str):
+        super().__init__(message)
 
 
 class OpenMLServerException(OpenMLServerError):
     """exception for when the result of the server was
        not 200 (e.g., listing call w/o results). """
 
     # Code needs to be optional to allow the exceptino to be picklable:
-    # https://stackoverflow.com/questions/16244923/how-to-make-a-custom-exception-class-with-multiple-init-args-pickleable
-    def __init__(self, message, code=None, additional=None, url=None):
+    # https://stackoverflow.com/questions/16244923/how-to-make-a-custom-exception-class-with-multiple-init-args-pickleable  # noqa: E501
+    def __init__(self, message: str, code: str = None, additional: str = None, url: str = None):
         self.message = message
         self.code = code
         self.additional = additional
         self.url = url
-        super(OpenMLServerException, self).__init__(message)
+        super().__init__(message)
 
     def __str__(self):
         return '%s returned code %s: %s' % (
@@ -38,16 +38,25 @@ class OpenMLServerNoResult(OpenMLServerException):
 
 class OpenMLCacheException(PyOpenMLError):
     """Dataset / task etc not found in cache"""
-    def __init__(self, message):
-        super(OpenMLCacheException, self).__init__(message)
+    def __init__(self, message: str):
+        super().__init__(message)
 
 
 class OpenMLHashException(PyOpenMLError):
     """Locally computed hash is different than hash announced by the server."""
     pass
 
 
-class PrivateDatasetError(PyOpenMLError):
+class OpenMLPrivateDatasetError(PyOpenMLError):
     """ Exception thrown when the user has no rights to access the dataset. """
-    def __init__(self, message):
-        super(PrivateDatasetError, self).__init__(message)
+    def __init__(self, message: str):
+        super().__init__(message)
+
+
+class OpenMLRunsExistError(PyOpenMLError):
+    """ Indicates run(s) already exists on the server when they should not be duplicated. """
+    def __init__(self, run_ids: set, message: str):
+        if len(run_ids) < 1:
+            raise ValueError("Set of run ids must be non-empty.")
+        self.run_ids = run_ids
+        super().__init__(message)

openml/flows/flow.py

@@ -1,8 +1,10 @@
 from collections import OrderedDict
+import os
 
 import xmltodict
 
 import openml._api_calls
+import openml.exceptions
 from ..utils import extract_xml_tags
 
 
@@ -128,7 +130,7 @@ def __init__(self, name, description, model, components, parameters,
         self.dependencies = dependencies
         self.flow_id = flow_id
 
-    def _to_xml(self):
+    def _to_xml(self) -> str:
         """Generate xml representation of self for upload to server.
 
         Returns
@@ -144,7 +146,7 @@ def _to_xml(self):
         flow_xml = flow_xml.split('\n', 1)[-1]
         return flow_xml
 
-    def _to_dict(self):
+    def _to_dict(self) -> dict:
         """ Helper function used by _to_xml and itself.
 
         Creates a dictionary representation of self which can be serialized
@@ -312,8 +314,32 @@ def _from_dict(cls, xml_dict):
 
         return flow
 
-    def publish(self):
-        """Publish flow to OpenML server.
+    def to_filesystem(self, output_directory: str) -> None:
+        os.makedirs(output_directory, exist_ok=True)
+        if 'flow.xml' in os.listdir(output_directory):
+            raise ValueError('Output directory already contains a flow.xml file.')
+
+        run_xml = self._to_xml()
+        with open(os.path.join(output_directory, 'flow.xml'), 'w') as f:
+            f.write(run_xml)
+
+    @classmethod
+    def from_filesystem(cls, input_directory) -> 'OpenMLFlow':
+        with open(os.path.join(input_directory, 'flow.xml'), 'r') as f:
+            xml_string = f.read()
+        return OpenMLFlow._from_dict(xmltodict.parse(xml_string))
+
+    def publish(self, raise_error_if_exists: bool = False) -> 'OpenMLFlow':
+        """ Publish this flow to OpenML server.
+
+        Raises a PyOpenMLError if the flow exists on the server, but
+        `self.flow_id` does not match the server known flow id.
+
+        Parameters
+        ----------
+        raise_error_if_exists : bool, optional (default=False)
+            If True, raise PyOpenMLError if the flow exists on the server.
+            If False, update the local flow to match the server flow.
 
         Returns
         -------
@@ -326,16 +352,27 @@ def publish(self):
         # instantiate an OpenMLFlow.
         import openml.flows.functions
 
-        xml_description = self._to_xml()
+        flow_id = openml.flows.functions.flow_exists(self.name, self.external_version)
+        if not flow_id:
+            if self.flow_id:
+                raise openml.exceptions.PyOpenMLError("Flow does not exist on the server, "
+                                                      "but 'flow.flow_id' is not None.")
+            xml_description = self._to_xml()
+            file_elements = {'description': xml_description}
+            return_value = openml._api_calls._perform_api_call(
+                "flow/",
+                'post',
+                file_elements=file_elements,
+            )
+            server_response = xmltodict.parse(return_value)
+            flow_id = int(server_response['oml:upload_flow']['oml:id'])
+        elif raise_error_if_exists:
+            error_message = "This OpenMLFlow already exists with id: {}.".format(flow_id)
+            raise openml.exceptions.PyOpenMLError(error_message)
+        elif self.flow_id is not None and self.flow_id != flow_id:
+            raise openml.exceptions.PyOpenMLError("Local flow_id does not match server flow_id: "
+                                                  "'{}' vs '{}'".format(self.flow_id, flow_id))
 
-        file_elements = {'description': xml_description}
-        return_value = openml._api_calls._perform_api_call(
-            "flow/",
-            'post',
-            file_elements=file_elements,
-        )
-        server_response = xmltodict.parse(return_value)
-        flow_id = int(server_response['oml:upload_flow']['oml:id'])
         flow = openml.flows.functions.get_flow(flow_id)
         _copy_server_fields(flow, self)
         try: