Source code for watson_machine_learning_client.experiments

################################################################################
#
# Licensed Materials - Property of IBM
# (C) Copyright IBM Corp. 2017
# US Government Users Restricted Rights - Use, duplication disclosure restricted
# by GSA ADP Schedule Contract with IBM Corp.
#
################################################################################

from __future__ import print_function
import requests
import json
import re
import time
import copy
from watson_machine_learning_client.wml_client_error import MissingValue, WMLClientError, MissingMetaProp
from watson_machine_learning_client.href_definitions import is_uid
from watson_machine_learning_client.wml_resource import WMLResource
from multiprocessing import Pool
from watson_machine_learning_client.utils import print_text_header_h1, print_text_header_h2, EXPERIMENT_DETAILS_TYPE,  format_metrics, STR_TYPE, STR_TYPE_NAME, docstring_parameter, group_metrics, str_type_conv, meta_props_str_conv
from watson_machine_learning_client.hpo import HPOParameter, HPOMethodParam
from watson_machine_learning_client.metanames import ExperimentMetaNames
from watson_machine_learning_client.href_definitions import API_VERSION, SPACES,PIPELINES, LIBRARIES, EXPERIMENTS, RUNTIMES, DEPLOYMENTS

_DEFAULT_LIST_LENGTH = 50

[docs]class Experiments(WMLResource): """ Run new experiment. """ ConfigurationMetaNames = ExperimentMetaNames() """MetaNames for experiments creation.""" @staticmethod def _HPOParameter(name, values=None, max=None, min=None, step=None): return HPOParameter(name, values, max, min, step) @staticmethod def _HPOMethodParam(name=None, value=None): return HPOMethodParam(name, value) def __init__(self, client): WMLResource.__init__(self, __name__, client) self._experiments_uids_cache = {}
[docs] def store(self, meta_props): """ Create IBM Watson Machine Learning experiment. A deep learning experiment is a logical grouping of one or more deep learning experiment training runs. **Parameters** .. important:: #. **meta_props**: meta data of the experiment configuration. To see available meta names use:\n >>> client.experiments.ConfigurationMetaNames.get() **type**: dict\n **Output** .. important:: **returns**: stored experiment metadata\n Metatdata contains the Unique Id (UID) of experiment created\n **return type**: dict\n **Example** >>> metadata = { >>> client.experiments.ConfigurationMetaNames.NAME: 'my_experiment', >>> client.experiments.ConfigurationMetaNames.EVALUATION_METRICS: ['accuracy'], >>> client.experiments.ConfigurationMetaNames.TRAINING_REFERENCES: [ >>> { >>> 'pipeline': {'href': pipeline_href_1} >>> >>> }, >>> { >>> 'pipeline': {'href':pipeline_href_2} >>> }, >>> ] >>> } >>> experiment_details = client.experiments.store(meta_props=metadata) >>> experiment_href = client.experiments.get_href(experiment_details) """ ##For CP4D, check if either spce or project ID is set self._client._check_if_either_is_set() metaProps = self.ConfigurationMetaNames._generate_resource_metadata(meta_props) ##Check if default space is set if self._client.CAMS: if self._client.default_space_id is not None: metaProps['space'] = {'href': "/v4/spaces/"+self._client.default_space_id} elif self._client.default_project_id is not None: metaProps['project'] = {'href': "/v2/projects/"+self._client.default_project_id} else: raise WMLClientError("It is mandatory to set the space/project id. Use client.set.default_space(<SPACE_UID>)/client.set.default_project(<PROJECT_UID>) to proceed.") if not self._ICP: self.ConfigurationMetaNames._validate(meta_props) response_experiment_post = requests.post( self._href_definitions.get_experiments_href(), json=metaProps, headers=self._client._get_headers() ) else: self.ConfigurationMetaNames._validate(meta_props) response_experiment_post = requests.post( self._href_definitions.get_experiments_href(), json=metaProps, headers=self._client._get_headers(), verify=False ) return self._handle_response(201, u'saving experiment', response_experiment_post)
[docs] @docstring_parameter({'str_type': STR_TYPE_NAME}) def update(self, experiment_uid, changes): """ Updates existing experiment metadata. **Parameters** .. important:: #. **experiment_uid**: Unique Id of experiment for which definition should be updated\n **type**: str\n #. **changes**: elements which should be changed, where keys are ConfigurationMetaNames\n **type**: dict\n **Output** .. important:: **returns**: metadata of updated experiment\n **return type**: dict\n **Example** >>> metadata = { >>> client.experiments.ConfigurationMetaNames.NAME:"updated_exp" >>> } >>> exp_details = client.experiments.update(experiment_uid, changes=metadata) """ ##For CP4D, check if either spce or project ID is set self._client._check_if_either_is_set() experiment_uid = str_type_conv(experiment_uid) self._validate_type(experiment_uid, u'experiment_uid', STR_TYPE, True) self._validate_type(changes, u'changes', dict, True) meta_props_str_conv(changes) details = self._client.repository.get_details(experiment_uid) patch_payload = self.ConfigurationMetaNames._generate_patch_payload(details['entity'], changes, with_validation=True) url = self._href_definitions.get_experiment_href(experiment_uid) if not self._ICP: response = requests.patch(url, json=patch_payload, params = self._client._params(),headers=self._client._get_headers()) else: response = requests.patch(url, json=patch_payload, params = self._client._params(),headers=self._client._get_headers(), verify=False) updated_details = self._handle_response(200, u'experiment patch', response) return updated_details
[docs] @docstring_parameter({'str_type': STR_TYPE_NAME}) def get_details(self, experiment_uid=None, limit=None): """ Get metadata of experiment(s). If Unique ID of experiment (experiment_uid) is not specified all experiments metadata is returned. **Parameters** .. important:: #. **experiment_uid**: Unique ID of experiment. (optional)\n **type**: str\n #. **limit**: limit number of fetched records (optional)\n **type**: int\n **Output** .. important:: **returns**: experiment(s) metadata\n **return type**: dict\n dict (if UID is not None) or {"resources": [dict]} (if UID is None)\n .. note:: If Unique ID is not specified, all experiments metadata is fetched\n **Example** >>> experiment_details = client.experiments.get_details(experiment_uid) >>> experiment_details = client.experiments.get_details() """ ##For CP4D, check if either spce or project ID is set self._client._check_if_either_is_set() experiment_uid = str_type_conv(experiment_uid) url = self._href_definitions.get_experiments_href() return self._get_artifact_details(url, experiment_uid, limit, 'experiment')
[docs] @staticmethod @docstring_parameter({'str_type': STR_TYPE_NAME}) def get_uid(experiment_details): """ Get Unique ID(UID) of stored experiment. **Parameters** .. important:: #. **experiment_details**: Metadata of the stored experiment\n **type**: dict\n **Output** .. important:: **returns**: Unique ID of stored experiment\n **return type**: str\n **Example** >>> experiment_details = client.experiments.get_detailsf(experiment_uid) >>> experiment_uid = client.experiments.get_uid(experiment_details) """ Experiments._validate_type(experiment_details, u'experiment_details', object, True) Experiments._validate_type_of_details(experiment_details, EXPERIMENT_DETAILS_TYPE) return WMLResource._get_required_element_from_dict(experiment_details, u'experiment_details', [u'metadata', u'guid'])
[docs] @staticmethod @docstring_parameter({'str_type': STR_TYPE_NAME}) def get_href(experiment_details): """ Get href of stored experiment. **Parameters** .. important:: #. **experiment_details**: Metadata of the stored experiment\n **type**: dict\n **Output** .. important:: **returns**: href of stored experiment\n **return type**: str\n **Example** >>> experiment_details = client.experiments.get_detailsf(experiment_uid) >>> experiment_href = client.experiments.get_href(experiment_details) """ Experiments._validate_type(experiment_details, u'experiment_details', object, True) Experiments._validate_type_of_details(experiment_details, EXPERIMENT_DETAILS_TYPE) return WMLResource._get_required_element_from_dict(experiment_details, u'experiment_details', [u'metadata', u'href'])
[docs] @docstring_parameter({'str_type': STR_TYPE_NAME}) def delete(self, experiment_uid): """ Delete a stored experiment. **Parameters** .. important:: #. **experiment_uid**: Unique ID of the stored experiment\n **type**: str\n **Output** .. important:: **returns**: status ("SUCCESS" or "FAILED")\n **return type**: str\n **Example** >>> client.experiments.delete(experiment_uid) """ ##For CP4D, check if either spce or project ID is set self._client._check_if_either_is_set() experiment_uid = str_type_conv(experiment_uid) Experiments._validate_type(experiment_uid, u'experiment_uid', STR_TYPE, True) url = self._href_definitions.get_experiment_href(experiment_uid) if not self._ICP: response = requests.delete(url, params = self._client._params(),headers=self._client._get_headers()) else: response = requests.delete(url, params = self._client._params(),headers=self._client._get_headers(), verify=False) return self._handle_response(204, u'experiment deletion', response, False)
[docs] def list(self, limit=None): """ List stored experiments. If limit is set to None there will be only first 50 records shown. **Parameters** .. important:: #. **limit**: limit number of fetched records\n **type**: int\n **Output** .. important:: This method only prints the list of all experiments in a table format.\n **return type**: None\n **Example** >>> client.experiments.list() """ ##For CP4D, check if either spce or project ID is set self._client._check_if_either_is_set() experiment_resources = self.get_details(limit=limit)[u'resources'] experiment_values = [(m[u'metadata'][u'guid'], m[u'entity'][u'name'], m[u'metadata'][u'created_at']) for m in experiment_resources] self._list(experiment_values, [u'GUID', u'NAME', u'CREATED'], limit, _DEFAULT_LIST_LENGTH)
def _create_revision(self, experiment_id, meta_props): """ Creates a new experiment revision. :param meta_props: :param experiment_id: :return: stored experiment new revision details A way you might use this is: >>> metadata = { >>> client.experiments.ConfigurationMetaNames.NAME: 'my_experiment', >>> client.experiments.ConfigurationMetaNames.EVALUATION_METRICS: ['accuracy'], >>> client.experiments.ConfigurationMetaNames.TRAINING_REFERENCES: [ >>> { >>> 'pipeline': {'href': pipeline_href_1} >>> }, >>> { >>> 'pipeline': {'href':pipeline_href_2} >>> } >>> ], >>> } >>> experiment_artifact = client.experiments.create_revision(meta_props, experiment_id) >>> experiment_href = client.experiments.get_href(experiment_details) """ meta_props_str_conv(meta_props) meta_props = copy.deepcopy(meta_props) if not self._ICP: response_experiment_put = requests.put( self._href_definitions.get_experiment_href(experiment_id), json=self.ConfigurationMetaNames._generate_resource_metadata(meta_props), headers=self._client._get_headers() ) else: response_experiment_put = requests.put( self._href_definitions.get_experiment_href(experiment_id), json=self.ConfigurationMetaNames._generate_resource_metadata(meta_props), headers=self._client._get_headers(), verify=False ) return self._handle_response(200, u'saving experiment revision', response_experiment_put) @docstring_parameter({'str_type': STR_TYPE_NAME}) def _get_revision_details(self, experiment_uid, limit=None): """ Get metadata of stored experiments revisions. If experiment UID is specified metadata of all revisions of that experiment is returned. :param experiment_uid: Unique Id of stored experiment(optional)\n :type experiment_uid: {str_type}\n :param limit: limit number of fetched records (optional)\n :type limit: int\n :returns: stored experiment(s) metadata\n :rtype: dict\n A way you might use this is: \n >>> experiment_details = client.repository.get_revision_details(experiment_uid) >>> experiment_details = client.experiments.get_revision_details(experiment_uid) """ url = self._href_definitions.get_experiment_href(experiment_uid) + u'/' + 'revisions' if not self._ICP: experiment_uid = str_type_conv(experiment_uid) response_get = requests.get( url, headers=self._client._get_headers() ) else: response_get = requests.get( url, headers=self._client._get_headers(), verify=False ) return self._handle_response(200, u'Fetching experiment revisions details', response_get)
[docs] def clone(self, experiment_uid, space_id=None, action="copy", rev_id=None): """ Creates a new experiment identical with the given experiment either in the same space or in a new space. All dependent assets will be cloned too. **Parameters** .. important:: #. **model_id**: Unique Id of the experiment to be cloned:\n **type**: str\n #. **space_id**: Unique Id of the space to which the experiment needs to be cloned. (optional)\n **type**: str\n #. **action**: Action specifying "copy" or "move". (optional)\n **type**: str\n #. **rev_id**: Revision ID of the experiment. (optional)\n **type**: str\n **Output** .. important:: **returns**: Metadata of the experiment cloned.\n **return type**: dict\n **Example** >>> client.experiments.clone(experiment_uid=artifact_id,space_id=space_uid,action="copy") .. note:: * If revision id is not specified, all revisions of the artifact are cloned\n * Default value of the parameter action is copy\n * Unique Id of Space is mandatory for move action\n """ artifact = str_type_conv(experiment_uid) Experiments._validate_type(artifact, 'experiment_uid', STR_TYPE, True) space = str_type_conv(space_id) rev = str_type_conv(rev_id) action = str_type_conv(action) clone_meta = {} if space is not None: clone_meta["space"] = {"href": API_VERSION + SPACES + "/" + space} if action is not None: clone_meta["action"] = action if rev is not None: clone_meta["rev"] = rev url = self._href_definitions.get_experiment_href(experiment_uid) if not self._ICP: response_post = requests.post(url, json=clone_meta, headers=self._client._get_headers()) else: response_post = requests.post(url, json=clone_meta, headers=self._client._get_headers(), verify=False) details = self._handle_response(expected_status_code=200, operationName=u'cloning experiment', response=response_post) return details