split function run_task into two functions

also parse parameters when running a flow on a task

fix publishing error

Author: mfeurer

openml/__init__.py

@@ -21,6 +21,7 @@
 from . import tasks
 from . import runs
 from . import flows
+from . import setups
 from .runs import OpenMLRun
 from .tasks import OpenMLTask, OpenMLSplit
 from .flows import OpenMLFlow
@@ -66,4 +67,4 @@ def populate_cache(task_ids=None, dataset_ids=None, flow_ids=None,
 
 __all__ = ['OpenMLDataset', 'OpenMLDataFeature', 'OpenMLRun',
            'OpenMLSplit', 'datasets', 'OpenMLTask', 'OpenMLFlow',
-           'config', 'runs', 'flows', 'tasks']
+           'config', 'runs', 'flows', 'tasks', 'setups']

openml/flows/__init__.py

@@ -1,4 +1,4 @@
-from .flow import OpenMLFlow
+from .flow import OpenMLFlow, _copy_server_fields
 
 from .sklearn_converter import sklearn_to_flow, flow_to_sklearn, _check_n_jobs
 from .functions import get_flow, list_flows, flow_exists, assert_flows_equal

openml/flows/flow.py

@@ -322,7 +322,16 @@ def _from_dict(cls, xml_dict):
         arguments['tags'] = tags
 
         arguments['model'] = None
-        return cls(**arguments)
+        flow = cls(**arguments)
+
+        if 'sklearn' in arguments['external_version']:
+            from .sklearn_converter import flow_to_sklearn
+            model = flow_to_sklearn(flow)
+        else:
+            model = None
+        flow.model = model
+
+        return flow
 
     def publish(self):
         """Publish flow to OpenML server.
@@ -332,32 +341,38 @@ def publish(self):
         self : OpenMLFlow
 
         """
+        import openml.flows.functions
 
         xml_description = self._to_xml()
 
         file_elements = {'description': xml_description}
         return_value = _perform_api_call("flow/", file_elements=file_elements)
-        self.flow_id = int(xmltodict.parse(return_value)['oml:upload_flow']['oml:id'])
+        flow_id = int(xmltodict.parse(return_value)['oml:upload_flow']['oml:id'])
+        flow = openml.flows.functions.get_flow(flow_id)
         try:
-            _check_flow(self)
+            openml.flows.functions.assert_flows_equal(self, flow)
         except ValueError as e:
             message = e.args[0]
             raise ValueError("Flow was not stored correctly on the server. "
                              "New flow ID is %d. Please check manually and "
                              "remove the flow if necessary! Error is:\n'%s'" %
-                             (self.flow_id, message))
+                             (flow_id, message))
+        _copy_server_fields(flow, self)
         return self
 
 
+def _copy_server_fields(source_flow, target_flow):
+    fields_added_by_the_server = ['flow_id', 'uploader', 'version',
+                                  'upload_date']
+    for field in fields_added_by_the_server:
+        setattr(target_flow, field, getattr(source_flow, field))
+
+    for name, component in source_flow.components.items():
+        assert name in target_flow.components
+        _copy_server_fields(component, target_flow.components[name])
+
+
 def _add_if_nonempty(dic, key, value):
     if value is not None:
         dic[key] = value
 
-
-def _check_flow(flow):
-    # Import is not possible at the top of the file as this would cause an
-    # ImportError due to an import cycle.
-    import openml.flows.functions
-
-    flow_copy = openml.flows.functions.get_flow(flow.flow_id)
-    openml.flows.functions.assert_flows_equal(flow, flow_copy)

openml/flows/functions.py

@@ -2,7 +2,7 @@
 import six
 
 from openml._api_calls import _perform_api_call
-from . import OpenMLFlow, flow_to_sklearn
+from . import OpenMLFlow
 
 
 def get_flow(flow_id):
@@ -24,9 +24,6 @@ def get_flow(flow_id):
     flow_dict = xmltodict.parse(flow_xml)
     flow = OpenMLFlow._from_dict(flow_dict)
 
-    if 'sklearn' in flow.external_version:
-        flow.model = flow_to_sklearn(flow)
-
     return flow
 
 
@@ -144,8 +141,7 @@ def assert_flows_equal(flow1, flow2):
         raise TypeError('Argument 2 must be of type OpenMLFlow, but is %s' %
                         type(flow2))
 
-    generated_by_the_server = ['flow_id', 'uploader', 'version',
-                               'upload_date', ]
+    generated_by_the_server = ['flow_id', 'uploader', 'version', 'upload_date']
     ignored_by_python_API = ['binary_url', 'binary_format', 'binary_md5',
                              'model']
 

openml/runs/__init__.py

@@ -1,6 +1,8 @@
 from .run import OpenMLRun
 from .trace import OpenMLRunTrace, OpenMLTraceIteration
-from .functions import (run_task, get_run, list_runs, get_runs, get_run_trace,
-                        initialize_model_from_run, initialize_model_from_trace)
+from .functions import (run_model_on_task, run_flow_on_task, get_run, list_runs,
+                        get_runs, get_run_trace, initialize_model_from_run,
+                        initialize_model_from_trace)
 
-__all__ = ['OpenMLRun', 'run_task', 'get_run', 'list_runs', 'get_runs']
+__all__ = ['OpenMLRun', 'run_model_on_task', 'run_flow_on_task', 'get_run',
+           'list_runs', 'get_runs']

openml/runs/functions.py

@@ -13,10 +13,11 @@
 
 from ..exceptions import PyOpenMLError
 from .. import config
-from ..flows import sklearn_to_flow, get_flow, flow_exists, _check_n_jobs
+from ..flows import sklearn_to_flow, get_flow, flow_exists, _check_n_jobs, \
+    _copy_server_fields
 from ..setups import setup_exists, initialize_model
 from ..exceptions import OpenMLCacheException, OpenMLServerException
-from .._api_calls import _perform_api_call, _file_id_to_url
+from .._api_calls import _perform_api_call
 from .run import OpenMLRun, _get_version_information
 from .trace import OpenMLRunTrace, OpenMLTraceIteration
 
@@ -25,7 +26,31 @@
 # circular imports
 
 
-def run_task(task, model, avoid_duplicate_runs=True, flow_tags=None, seed=None):
+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)
+
+
+def run_flow_on_task(task, flow, avoid_duplicate_runs=True, flow_tags=None,
+                     seed=None):
     """Performs a CV run on the dataset of the given task, using the split.
 
     Parameters
@@ -51,23 +76,18 @@ def run_task(task, model, avoid_duplicate_runs=True, flow_tags=None, seed=None):
     """
     if flow_tags is not None and not isinstance(flow_tags, list):
         raise ValueError("flow_tags should be list")
-    # TODO move this into its onwn module. While it somehow belongs here, it
-    # adds quite a lot of functionality which is better suited in other places!
-    # TODO why doesn't this accept a flow as input? - this would make this more flexible!
-    model = _get_seeded_model(model, seed)
-    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)
+    flow.model = _get_seeded_model(flow.model, seed=seed)
 
     # 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 and flow_id:
-        flow = get_flow(flow_id)
-        setup_id = setup_exists(flow, model)
+    if avoid_duplicate_runs:
+        flow_from_server = get_flow(flow.flow_id)
+        setup_id = setup_exists(flow_from_server)
         ids = _run_exists(task.task_id, setup_id)
         if ids:
             raise PyOpenMLError("Run already exists in server. Run id(s): %s" %str(ids))
+        _copy_server_fields(flow_from_server, flow)
 
     dataset = task.get_dataset()
 
@@ -79,18 +99,12 @@ def run_task(task, model, avoid_duplicate_runs=True, flow_tags=None, seed=None):
     run_environment = _get_version_information()
     tags = ['openml-python', run_environment[1]]
     # execute the run
-    run = OpenMLRun(task_id=task.task_id, flow_id=None, dataset_id=dataset.dataset_id, model=model, tags=tags)
-    res = _run_task_get_arffcontent(model, task, class_labels)
+    run = OpenMLRun(task_id=task.task_id, flow_id=None, 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
 
-    if flow_id == False:
-        # 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 = get_flow(flow_id)
-
     run.flow_id = flow.flow_id
     config.logger.info('Executed Task %d with Flow id: %d' % (task.task_id, run.flow_id))
 

openml/runs/run.py

@@ -1,4 +1,5 @@
 from collections import OrderedDict
+import json
 import sys
 import time
 
@@ -143,11 +144,6 @@ def _create_description_xml(self):
             XML description of run.
         """
 
-        # TODO: don't we have flow object in data structure? Use this one
-        downloaded_flow = openml.flows.get_flow(self.flow_id)
-
-        openml_param_settings = OpenMLRun._parse_parameters(self.model, downloaded_flow)
-
         # as a tag, it must be of the form ([a-zA-Z0-9_\-\.])+
         # so we format time from 'mm/dd/yy hh:mm:ss' to 'mm-dd-yy_hh.mm.ss'
         # well_formatted_time = time.strftime("%c").replace(
@@ -156,56 +152,82 @@ def _create_description_xml(self):
         #     [self.model.__module__ + "." + self.model.__class__.__name__]
         description = _to_dict(taskid=self.task_id, flow_id=self.flow_id,
                                setup_string=_create_setup_string(self.model),
-                               parameter_settings=openml_param_settings,
+                               parameter_settings=self.parameter_settings,
                                error_message=self.error_message,
                                detailed_evaluations=self.detailed_evaluations,
                                tags=self.tags)
         description_xml = xmltodict.unparse(description, pretty=True)
         return description_xml
 
     @staticmethod
-    def _parse_parameters(model, server_flow):
-        """Extracts all parameter settings from a model in OpenML format.
+    def _parse_parameters(flow):
+        """Extracts all parameter settings from the model inside a flow in
+        OpenML format.
 
         Parameters
         ----------
-        model
-            the scikit-learn model (fitted)
         flow
             openml flow object (containing flow ids, i.e., it has to be downloaded from the server)
 
         """
-        if server_flow.flow_id is None:
-            raise ValueError("The flow parameter needs to be downloaded from server")
+
+        # Depth-first search to check if all components were uploaded to the
+        # server before parsing the parameters
+        stack = list()
+        stack.append(flow)
+        while len(stack) > 0:
+            current = stack.pop()
+            if current.flow_id is None:
+                raise ValueError("Flow %s has no flow_id!" % current.name)
+            else:
+                for component in current.components.values():
+                    stack.append(component)
 
         def get_flow_dict(_flow):
             flow_map = {_flow.name: _flow.flow_id}
             for subflow in _flow.components:
                 flow_map.update(get_flow_dict(_flow.components[subflow]))
             return flow_map
 
-        def extract_parameters(_flow, _param_dict, _main_call=False, main_id=None):
+        def extract_parameters(_flow, _flow_dict, _main_call=False, main_id=None):
             # _flow is openml flow object, _param dict maps from flow name to flow id
             # for the main call, the param dict can be overridden (useful for unit tests / sentinels)
-            # this way, for flows without subflows we do not have to rely on _param_dict
+            # this way, for flows without subflows we do not have to rely on _flow_dict
             _params = []
             for _param_name in _flow.parameters:
                 _current = OrderedDict()
                 _current['oml:name'] = _param_name
-                _current['oml:value'] = _flow.parameters[_param_name]
+
+                _tmp = openml.flows.sklearn_to_flow(_flow.model.get_params()[_param_name])
+
+                # Try to filter out components which are handled further down!
+                if isinstance(_tmp, openml.flows.OpenMLFlow):
+                    continue
+                try:
+                    _tmp = json.dumps(_tmp)
+                except TypeError as e:
+                    # Python3.5 exception message:
+                    # <openml.flows.flow.OpenMLFlow object at 0x7fed87978160> is not JSON serializable
+                    # Python3.6 exception message:
+                    # Object of type 'OpenMLFlow' is not JSON serializable
+                    if 'OpenMLFlow' in e.args[0] and \
+                            'is not JSON serializable' in e.args[0]:
+                        continue
+
+                _current['oml:value'] = _tmp
                 if _main_call:
                     _current['oml:component'] = main_id
                 else:
-                    _current['oml:component'] = _param_dict[_flow.name]
+                    _current['oml:component'] = _flow_dict[_flow.name]
                 _params.append(_current)
+
             for _identifier in _flow.components:
-                _params.extend(extract_parameters(_flow.components[_identifier], _param_dict))
+                _params.extend(extract_parameters(_flow.components[_identifier], _flow_dict))
             return _params
 
-        flow_dict = get_flow_dict(server_flow)
-        local_flow = openml.flows.sklearn_to_flow(model)
+        flow_dict = get_flow_dict(flow)
+        parameters = extract_parameters(flow, flow_dict, True, flow.flow_id)
 
-        parameters = extract_parameters(local_flow, flow_dict, True, server_flow.flow_id)
         return parameters
 
 ################################################################################