text: # ---------------------------------------------------------
  # Copyright (c) Microsoft Corporation. All rights reserved.
  # ---------------------------------------------------------
  
  """Contains functionality for managing experiment metrics and artifacts in Azure Machine Learning."""
  
  import logging
  import os
  import re
  import json
  import sys
  import time
  
  from functools import wraps
  
  from azureml._base_sdk_common._docstring_wrapper import experimental
  from six import raise_from
  from inspect import isgeneratorfunction
  from types import GeneratorType
  from uuid import uuid4
  
  from azureml._base_sdk_common.tracking import global_tracking_info_registry
  from azureml._logging.chained_identity import ChainedIdentity
  from azureml._run_impl.run_base import _RunBase
  from azureml._history.utils.constants import OUTPUTS_DIR
  from azureml._restclient.constants import BASE_RUN_SOURCE, RUN_ORIGIN, SDK_TARGET, RunStatus
  from azureml._restclient.utils import camel_case_transformer, create_session_with_retry
  from azureml._restclient.models.run_type_v2 import RunTypeV2
  from azureml._model_management._util import model_name_validation
  
  from azureml.core.environment import Environment
  from azureml.core.datastore import Datastore
  from azureml.core.keyvault import Keyvault
  from azureml.data.abstract_dataset import AbstractDataset
  from azureml.data.datapath import DataPath
  from azureml.exceptions import RunEnvironmentException, UserErrorException, ExperimentExecutionException, \
      ActivityFailedException, SnapshotException
  from azureml._file_utils import download_file_stream
  
  RUNNING_STATES = RunStatus.get_running_statuses()
  POST_PROCESSING_STATES = RunStatus.get_post_processing_statuses()
  
  module_logger = logging.getLogger(__name__)
  
  
  def _check_for_data_container_id(func):
      @wraps(func)
      def wrapped(self, *args, **kwargs):
          if self._container is None:
              raise UserErrorException("{} doesn't have a data container associated with it yet - "
                                       "therefore, the {} cannot upload files, or log file backed metrics."
                                       .format(self, self.__class__.__name__))
          return func(self, *args, **kwargs)
  
      return wrapped
  
  
  class Run(_RunBase):
      """Defines the base class for all Azure Machine Learning experiment runs.
  
      A *run* represents a single trial of an experiment. Runs are used to monitor the asynchronous
      execution of a trial, log metrics and store output of the trial, and to analyze results and access
      artifacts generated by the trial.
  
      Run objects are created when you submit a script to train a model in many different scenarios in
      Azure Machine Learning, including HyperDrive runs, Pipeline runs, and AutoML runs.
      A Run object is also created when you :func:`azureml.core.experiment.Experiment.submit` or
      :func:`azureml.core.experiment.Experiment.start_logging` with the :class:`azureml.core.Experiment` class.
  
      To get started with experiments and runs, see
  
      * `Create and manage environments for training and
        deployment <https://docs.microsoft.com/azure/machine-learning/how-to-use-environments>`_
      * `How to manage runs <https://docs.microsoft.com/azure/machine-learning/how-to-manage-runs>`_
  
      .. remarks::
  
          A *run* represents a single trial of an experiment.  A Run object is used to monitor the
          asynchronous execution of a trial, log metrics and store output of the trial,
          and to analyze results and access artifacts generated by the trial.
  
          Run is used inside of your experimentation code to log metrics and artifacts to the Run History service.
  
          Run is used outside of your experiments to monitor progress and to query and analyze the metrics and
          results that were generated.
  
          The functionality of Run includes:
  
          *  Storing and retrieving metrics and data
          *  Uploading and downloading files
          *  Using tags as well as the child hierarchy for easy lookup of past runs
          *  Registering stored model files as a model that can be operationalized
          *  Storing, modifying, and retrieving properties of a run
          *  Loading the current run from a remote environment with the :meth:`azureml.core.run.Run.get_context` method
          *  Efficiently snapshotting a file or directory for reproducibility
  
          This class works with the :class:`azureml.core.Experiment` in these scenarios:
  
          * Creating a run by executing code using :func:`azureml.core.Experiment.submit`
          * Creating a run interactively in a notebook using :func:`azureml.core.Experiment.start_logging`
          * Logging metrics and uploading artifacts in your experiment, such as when using :func:`log`
          * Reading metrics and downloading artifacts when analyzing experimental results, such as when using
            :func:`get_metrics`
  
          To submit a run, create a configuration object that describes how the experiment is run. Here are examples
          of the different configuration objects you can use:
  
          * :class:`azureml.core.ScriptRunConfig`
          * :class:`azureml.train.automl.automlconfig.AutoMLConfig`
          * :class:`azureml.train.hyperdrive.HyperDriveConfig`
          * :class:`azureml.pipeline.core.Pipeline`
          * :class:`azureml.pipeline.core.PublishedPipeline`
          * :class:`azureml.pipeline.core.PipelineEndpoint`
  
          The following metrics can be added to a run while training an experiment.
  
          * Scalar
              * Log a numerical or string value to the run with the given name using :meth:`log`. Logging
                a metric to a run causes that metric to be stored in the run record in the experiment. You
                can log the same metric multiple times within a run, the result being considered a vector
                of that metric.
              * Example: ``run.log("accuracy", 0.95)``
          * List
              * Log a list of values to the run with the given name using :meth:`log_list`.
              * Example: ``run.log_list("accuracies", [0.6, 0.7, 0.87])``
          * Row
              * Using :meth:`log_row` creates a metric with multiple columns as described in ``kwargs``.
                Each named parameter generates a column with the value specified. ``log_row`` can be called
                once to log an arbitrary tuple, or multiple times in a loop to generate a complete table.
              * Example: ``run.log_row("Y over X", x=1, y=0.4)``
          * Table
              * Log a dictionary object to the run with the given name using :meth:`log_table`.
              * Example: ``run.log_table("Y over X", {"x":[1, 2, 3], "y":[0.6, 0.7, 0.89]})``
          * Image
              * Log an image to the run record. Use :meth:`log_image` to log an image file or a matplotlib plot to
                the run. These images will be visible and comparable in the run record.
              * Example: ``run.log_image("ROC", path)``
  
      :param experiment: The containing experiment.
      :type experiment: azureml.core.Experiment
      :param run_id: The ID for the run.
      :type run_id: str
      :param outputs: The outputs to be tracked.
      :type outputs: str
      :param _run_dto: Internal use only.
      :type _run_dto: azureml._restclient.models.run_dto.RunDto
      :param kwargs: A dictionary of additional configuration parameters.
      :type kwargs: dict
      """
  
      _RUNSOURCE_PROPERTY = BASE_RUN_SOURCE
      _run_source_initializers = {}
  
      # default timeout of session for getting content in the run,
      # the 1st element is conn timeout, the 2nd is the read timeout.
      _DEFAULT_GET_CONTENT_TIMEOUT = (5, 120)
      _WAIT_COMPLETION_POLLING_INTERVAL_MIN = os.environ.get("AZUREML_RUN_POLLING_INTERVAL_MIN", 2)
      _WAIT_COMPLETION_POLLING_INTERVAL_MAX = os.environ.get("AZUREML_RUN_POLLING_INTERVAL_MAX", 60)
  
      def __init__(self, experiment, run_id, outputs=None, **kwargs):
          """Initialize the Run object.
  
          :param experiment: The containing experiment.
          :type experiment: azureml.core.Experiment
          :param run_id: The ID for the run.
          :type run_id: str
          :param outputs: The outputs to be tracked.
          :type outputs: str
          :param kwargs: A dictionary of additional configuration parameters.
          :type kwargs: dict
  
          """
          super(Run, self).__init__(experiment, run_id, outputs=outputs, **kwargs)
          self._parent_run = None
  
      @property
      def _container(self):
          """Return the container for files uploaded by the run.
  
          :return: Container string for the run files.
          :rtype: str
          """
          return self._run_dto.get("data_container_id", None)
  
      @staticmethod
      def _dto_to_run(experiment, run_dto, outputs=None, **kwargs):
          """Create a Run object from the service representation.
  
          :param experiment: The containing experiment.
          :type experiment: azureml.core.Experiment
          :param run_dto:
          :type run_dto: azureml._restclient.models.run_dto.RunDto
          :param outputs: Outputs directory to be tracked.
          :type outputs: str
          :param kwargs:
          :type kwargs: dict
          :return: Returns the Run object
          :rtype: Run
          """
          return Run(experiment, run_dto.run_id, outputs=outputs, _run_dto=run_dto, **kwargs)
  
      @classmethod
      def _load_scope(cls):
          """Load the current context from the environment.
  
          :return: experiment, run_id, url
          :rtype: azureml.core.Experiment, str, str
          """
          from .authentication import AzureMLTokenAuthentication
          from .experiment import Experiment
          from .workspace import Workspace
  
          try:
              # Load authentication scope environment variables
              subscription_id = os.environ['AZUREML_ARM_SUBSCRIPTION']
              run_id = os.environ["AZUREML_RUN_ID"]
              resource_group = os.environ["AZUREML_ARM_RESOURCEGROUP"]
              workspace_name = os.environ["AZUREML_ARM_WORKSPACE_NAME"]
              experiment_name = os.environ["AZUREML_ARM_PROJECT_NAME"]
              experiment_id = os.environ.get("AZUREML_EXPERIMENT_ID")
              workspace_id = os.environ.get("AZUREML_WORKSPACE_ID")
  
              if experiment_id is None:
                  module_logger.warning("experiment_id cannot be found in env variable.")
              # Initialize an AMLToken auth, authorized for the current run
              token, token_expiry_time = AzureMLTokenAuthentication._get_initial_token_and_expiry()
  
              url = os.environ["AZUREML_SERVICE_ENDPOINT"]
              location_from_url_regex_match = re.compile(r"//(.*?)\.").search(url)
              location = location_from_url_regex_match.group(1) if location_from_url_regex_match else None
  
              from azureml._base_sdk_common.service_discovery import DISCOVERY_END_POINT
              workspace_discovery_url = os.environ.get(DISCOVERY_END_POINT, url + "/discovery")
          except KeyError as key_error:
              raise_from(RunEnvironmentException(), key_error)
          else:
              auth = cls._load_auth(subscription_id=subscription_id,
                                    resource_group=resource_group,
                                    workspace_name=workspace_name,
                                    experiment_name=experiment_name,
                                    run_id=run_id,
                                    url=url,
                                    token=token,
                                    token_expiry_time=token_expiry_time)
              # Disabling service check, as this is in remote context and we don't have an arm token
              # to check arm if the workspace exists or not.
              from azureml._restclient.service_context import ServiceContext
              service_context = ServiceContext(subscription_id,
                                               resource_group,
                                               workspace_name,
                                               workspace_id,
                                               workspace_discovery_url,
                                               auth)
              workspace = Workspace._from_service_context(service_context, location)
  
              # The experiment already exists so disabling the creation call
              experiment = Experiment(workspace,
                                      experiment_name,
                                      _id=experiment_id,
                                      _skip_name_validation=True,
                                      _create_in_cloud=False)
  
              return experiment, run_id
  
      @classmethod
      def _load_auth(cls,
                     subscription_id,
                     resource_group,
                     workspace_name,
                     experiment_name,
                     run_id,
                     url,
                     token,
                     token_expiry_time):
          identity_type = os.environ.get("AZUREML_IDENTITY_IN_USE", None)
          if identity_type in [None, "AMLToken"]:
              if identity_type is None:
                  module_logger.debug("Identity in use is not set. Falling back to using AMLToken")
              return cls._get_aml_token_auth(subscription_id=subscription_id,
                                             resource_group=resource_group,
                                             workspace_name=workspace_name,
                                             experiment_name=experiment_name,
                                             run_id=run_id,
                                             host=url,
                                             token=token,
                                             token_expiry_time=token_expiry_time)
  
          if identity_type == "Managed":
              return cls._get_msi_auth()
  
          if identity_type == "ServicePrincipal":
              raise_from(RunEnvironmentException(),
                         ValueError("ServicePrincipal Auth to be used for remote runs is not supported yet."))
  
          raise ValueError("IdentityType {0} is not supported".format(identity_type))
  
      @classmethod
      def _get_aml_token_auth(cls,
                              subscription_id,
                              resource_group,
                              workspace_name,
                              experiment_name,
                              run_id,
                              host,
                              token_expiry_time,
                              token):
          module_logger.debug("Using AMLToken auth for remote run")
          from .authentication import AzureMLTokenAuthentication
          return AzureMLTokenAuthentication.create(token,
                                                   AzureMLTokenAuthentication._convert_to_datetime(
                                                       token_expiry_time),
                                                   host,
                                                   subscription_id,
                                                   resource_group,
                                                   workspace_name,
                                                   experiment_name,
                                                   run_id)
  
      @classmethod
      def _get_msi_auth(cls):
          module_logger.debug("Using Managed Identity for remote run")
          msi_conf = {}
          for item in ["client_id", "object_id", "msi_res_id"]:
              value = os.environ.get("AZUREML_IDENTITY_" + item.upper(), None)
              if value:
                  msi_conf.update({item: value})
          from .authentication import MsiAuthentication
          return MsiAuthentication(identity_config=msi_conf)
  
      @classmethod
      def get_context(cls, allow_offline=True, used_for_context_manager=False, **kwargs):
          """Return current service context.
  
          Use this method to retrieve the current service context for logging metrics and uploading files. If
          ``allow_offline`` is True (the default), actions against the Run object will be printed to standard
          out.
  
          .. remarks::
  
              This function is commonly used to retrieve the authenticated Run object
              inside of a script to be submitted for execution via experiment.submit(). This run object is both
              an authenticated context to communicate with Azure Machine Learning services and a conceptual container
              within which metrics, files (artifacts), and models are contained.
  
              .. code-block:: python
  
                  run = Run.get_context() # allow_offline=True by default, so can be run locally as well
                  ...
                  run.log("Accuracy", 0.98)
                  run.log_row("Performance", epoch=e, error=err)
  
          :param cls: Indicates class method.
          :param allow_offline: Allow the service context to fall back to offline mode so that the training script
              can be tested locally without submitting a job with the SDK. True by default.
          :type allow_offline: bool
          :param kwargs: A dictionary of additional parameters.
          :type kwargs: dict
          :return: The submitted run.
          :rtype: azureml.core.run.Run
          """
          try:
              experiment, run_id = cls._load_scope()
  
              # Querying for the run instead of initializing to load current state
              if used_for_context_manager:
                  return _SubmittedRun(experiment, run_id, **kwargs)
              return _SubmittedRun._get_instance(experiment, run_id, **kwargs)
          except RunEnvironmentException as ex:
              module_logger.debug("Could not load run context %s, switching offline: %s", ex, allow_offline)
              if allow_offline:
                  module_logger.info("Could not load the run context. Logging offline")
                  return _OfflineRun(**kwargs)
              else:
                  module_logger.debug("Could not load the run context and allow_offline set to False")
                  raise RunEnvironmentException(inner_exception=ex)
  
      @classmethod
      def get_submitted_run(cls, **kwargs):
          """DEPRECATED. Use :func:`get_context`.
  
          Get the submitted run for this experiment.
  
          :return: The submitted run.
          :rtype: azureml.core.run.Run
          """
          module_logger.warning("Run.get_submitted_run() is deprecated, use Run.get_context() instead")
          module_logger.warning("Run.get_submitted_run() will be removed shortly")
          return cls.get_context(allow_offline=False, **kwargs)
  
      def get_secret(self, name):
          """Get the secret value from the context of a run.
  
          Get the secret value for the name provided. The secret name references a value stored in
          Azure Key Vault associated with your workspace. For an example of working with secrets,
          see `Use secrets in training
          runs <https://docs.microsoft.com/azure/machine-learning/how-to-use-secrets-in-runs>`_.
  
          :param name: The secret name for which to return a secret.
          :type name: str
          :return: The secret value.
          :rtype: str
          """
          module_logger.info("Run.get_secret() is called from run context")
          return Keyvault(self.experiment.workspace).get_secret(name)
  
      def get_secrets(self, secrets):
          """Get the secret values for a given list of secret names.
  
          Get a dictionary of found and not found secrets for the list of names provided.
          Each secret name references a value stored in
          Azure Key Vault associated with your workspace. For an example of working with secrets,
          see `Use secrets in training
          runs <https://docs.microsoft.com/azure/machine-learning/how-to-use-secrets-in-runs>`_.
  
          :param secrets: A list of secret names for which to return secret values.
          :type secrets: builtin.list[str]
          :return: Returns a dictionary of found and not found secrets.
          :rtype: dict[str: str]
          """
          module_logger.info("Batch Run.get_secrets() is called from run context")
          return Keyvault(self.experiment.workspace).get_secrets(secrets)
  
      @staticmethod
      def add_type_provider(runtype, run_factory):
          """Extensibility hook for custom Run types stored in Run History.
  
          :param runtype: The value of Run.type for which the factory will be invoked. Examples include
              'hyperdrive' or 'azureml.scriptrun', but can be extended with custom types.
          :type runtype: str
          :param run_factory: A function with signature (Experiment, RunDto) -> Run to be invoked when
              listing runs.
          :type run_factory: function
          """
          if Run._run_source_initializers.get(runtype, None) is not None:
              raise ValueError("RunType {0} already has a factory, can't add another".format(runtype))
  
          # TODO: Validate signature of the factory if possible via the code object
          module_logger.debug("Adding new factory {0} for run source {1}".format(run_factory, runtype))
          Run._run_source_initializers[runtype] = run_factory
  
      @staticmethod
      def _rehydrate_runs(experiment, run_dtos):
          """Rehydrate runs.
  
          :param experiment: The containing experiment.
          :param run_dtos:
          :type run_dtos: azureml._restclient.models.run_dto.RunDto
          """
          module_logger.debug("Available factories for run types {0}".format(Run._run_source_initializers))
          for run_dto in run_dtos:
              run_id = run_dto.run_id
  
              # TODO: Run source is around for backward compatibility. Delete after PuP
              run_properties = getattr(run_dto, "properties", {})
              run_source = run_properties.get(Run._RUNSOURCE_PROPERTY, 'None')
              runtype = run_dto.run_type if run_dto.run_type is not None else run_source
  
              module_logger.debug("Initializing Run {0} from type {1}".format(run_id, runtype))
  
              factory = Run._run_source_initializers.get(runtype, None)
              if factory is None and (runtype and not runtype.strip()):
                  warnmsg = "Run {0} has type {1} but has no client type registered. " \
                            "Entire functionality might not be available."
                  module_logger.warning(warnmsg.format(run_id, runtype))
              if factory is None:
                  factory = Run._dto_to_run
  
              yield factory(experiment, run_dto)
  
      @staticmethod
      def list(
              experiment,
              type=None,
              tags=None,
              properties=None,
              status=None,
              include_children=False,
              _rehydrate_runs=True):
          """Get a list of runs in an experiment specified by optional filters.
  
          .. remarks::
  
              The following code example shows some uses of the ``list`` method.
  
              .. code-block:: python
  
                  favorite_completed_runs = Run.list(experiment, status='Completed', tags='favorite')
  
                  all_distinct_runs = Run.list(experiment)
                  and_their_children = Run.list(experiment, include_children=True)
  
                  only_script_runs = Run.list(experiment, type=ScriptRun.RUN_TYPE)
  
          :param experiment: The containing experiment.
          :type experiment: azureml.core.Experiment
          :param type: If specified, returns runs matching specified type.
          :type type: str
          :param tags:  If specified, returns runs matching specified *"tag"* or {*"tag"*: *"value"*}.
          :type tags: str or dict
          :param properties: If specified, returns runs matching specified *"property"* or {*"property"*: *"value"*}.
          :type properties: str or dict
          :param status: If specified, returns runs with status specified *"status"*.
          :type status: str
          :param include_children: If set to true, fetch all the runs, not only top-level ones.
          :type include_children: bool
          :param _rehydrate_runs: If set to True (by default), will use the registered provider to reinstantiate
              an object for that type instead of the base Run.
          :type _rehydrate_runs: bool
          :return: A list of runs.
          :rtype: builtin.list[azureml.core.run.Run]
          """
          from azureml._run_impl.run_history_facade import RunHistoryFacade
  
          run_dtos = RunHistoryFacade.get_runs(
              experiment,
              runtype=type,
              tags=tags,
              properties=properties,
              status=status,
              include_children=include_children)
  
          module_logger.debug("Rehydrating runs on enumerate: {0}".format(_rehydrate_runs))
          if _rehydrate_runs:
              return Run._rehydrate_runs(experiment, run_dtos)
          else:
              return (Run._dto_to_run(experiment, run_dto) for run_dto in run_dtos)
  
      @staticmethod
      def list_by_compute(
              compute,
              type=None,
              tags=None,
              properties=None,
              status=None):
          """Get a list of runs in a compute specified by optional filters.
  
          :param compute: The containing compute.
          :type compute: azureml.core.ComputeTarget
          :param type: If specified, returns runs matching specified type.
          :type type: str
          :param tags:  If specified, returns runs matching specified *"tag"* or {*"tag"*: *"value"*}.
          :type tags: str or dict
          :param properties: If specified, returns runs matching specified *"property"* or {*"property"*: *"value"*}.
          :type properties: str or dict
          :param status: If specified, returns runs with status specified *"status"*.
              Only allowed values are "Running" and "Queued".
          :type status: str
          :return: a generator of ~_restclient.models.RunDto
          :rtype: builtin.generator
          """
          from azureml._run_impl.run_history_facade import RunHistoryFacade
  
          run_dtos = RunHistoryFacade.get_runs_by_compute(
              compute.workspace,
              compute.name,
              runtype=type,
              tags=tags,
              properties=properties,
              status=status)
  
          # note we don't have an experiment to populate the run with, hence we are returning generator of run_dtos,
          # not the run objects
          return run_dtos
  
      @staticmethod
      def _start_logging(experiment, name=None, run_id=None,
                         outputs=None, snapshot_directory=".", **kwargs):
          """Create and start a new run for the workspace and experiment name.
  
          :param experiment: The containing experiment.
          :type experiment: azureml.core.Experiment
          :param name: Optional name for the run.
          :type name: str
          :param run_id: Optional run_id for the run, otherwise uses default.
          :type run_id: str
          :param outputs: Optional outputs directory to track.
          :type outputs: str
          :param snapshot_directory: Optional directory to take snapshot of. Setting to None will take no snapshot.
          :type snapshot_directory: str
          :return: Return a run.
          :rtype: azureml.core.run.Run
          :param kwargs:
          """
          outputs = outputs if outputs else OUTPUTS_DIR
          properties = global_tracking_info_registry.gather_all(snapshot_directory)
          typev2 = RunTypeV2(orchestrator="External", traits=['unspecified'])
          run = Run._create(experiment, name=name, run_id=run_id, outputs=outputs,
                            properties=properties, typev2=typev2, **kwargs)
          run._client.start()
          if snapshot_directory:
              try:
                  run.take_snapshot(snapshot_directory)
              except SnapshotException as snapshot_ex:
                  error_msg = "Failed to take a snapshot of {}. Pass snapshot_directory=None to start_logging " \
                              "to skip taking a snapshot of the given directory.".format(snapshot_directory)
                  raise_from(SnapshotException(error_msg), snapshot_ex)
          return run
  
      @staticmethod
      def _wait_before_polling(current_seconds):
          if current_seconds < 0:
              raise ValueError("current_seconds must be positive")
          import math
          # Sigmoid that tapers off near the max at ~ 3 min
          duration = Run._WAIT_COMPLETION_POLLING_INTERVAL_MAX / (1.0 + 100 * math.exp(-current_seconds / 20.0))
          return max(Run._WAIT_COMPLETION_POLLING_INTERVAL_MIN, duration)
  
      @staticmethod
      def get(workspace, run_id):
          """Get the run for this workspace with its run ID.
  
          :param workspace: The containing workspace.
          :type workspace: azureml.core.Workspace
          :param run_id: The run ID.
          :type run_id: string
          :return: The submitted run.
          :rtype: azureml.core.run.Run
          """
          from azureml.core.experiment import Experiment
          from azureml._restclient.workspace_client import WorkspaceClient
          client = WorkspaceClient(workspace.service_context)
          run_dto = client.get_run(run_id)
  
          experiment_dto = client.get_experiment_by_id(run_dto.experiment_id)
          experiment = Experiment(
              workspace,
              experiment_dto.name,
              _id=experiment_dto.experiment_id,
              _archived_time=experiment_dto.archived_time,
              _create_in_cloud=False,
              _experiment_dto=experiment_dto)
  
          runs = Run._rehydrate_runs(experiment, [run_dto])
          return next(runs)
  
      def submit_child(self, config, tags=None, **kwargs):
          """Submit an experiment and return the active child run.
  
          .. remarks::
  
              Submit is an asynchronous call to the Azure Machine Learning platform to execute a trial on local
              or remote hardware.  Depending on the configuration, submit will automatically prepare
              your execution environments, execute your code, and capture your source code and results
              into the experiment's run history.
  
              To submit an experiment you first need to create a configuration object describing
              how the experiment is to be run.  The configuration depends on the type of trial required.
  
              An example of how to submit a child experiment from your local machine using
              :class:`azureml.core.script_run_config.ScriptRunConfig` is as follows:
  
              .. code-block:: python
  
                  from azureml.core import ScriptRunConfig
  
                  # run a trial from the train.py code in your current directory
                  config = ScriptRunConfig(source_directory='.', script='train.py',
                      run_config=RunConfiguration())
                  run = parent_run.submit_child(config)
  
                  # get the url to view the progress of the experiment and then wait
                  # until the trial is complete
                  print(run.get_portal_url())
                  run.wait_for_completion()
  
              For details on how to configure a run, see :func:`azureml.core.Experiment.submit`.
  
          :param config: The config to be submitted.
          :type config: object
          :param tags: Tags to be added to the submitted run, e.g., {"tag": "value"}.
          :type tags: dict
          :param kwargs: Additional parameters used in submit function for configurations.
          :type kwargs: dict
          :return: A run object.
          :rtype: azureml.core.Run
          """
          return self.experiment.submit(config, tags=tags, _parent_run_id=self.id, **kwargs)
  
      @staticmethod
      def _create(experiment, name=None, run_id=None, outputs=None,
                  properties=None, tags=None, typev2=None, **kwargs):
          """Create a new run for the workspace and experiment name.
  
          :param experiment: The containing experiment.
          :type experiment: azureml.core.Experiment
          :param name: Optional name for the run.
          :type name: str
          :param run_id: Optional run_id for the run, otherwise uses default.
          :type run_id: str
          :param outputs: Optional outputs directory to track.
          :type outputs: str
          :param properties: Optional initial properties of a run
          :type properties: dict[str]
          :param tags: Optional initial tags on a run
          :type tags: dict[str]
          :param typev2: Optional RunTypeV2 for the run
          :type typev2: RunTypeV2
          :return: Return a run.
          :rtype: azureml.core.run.Run
          :param kwargs:
          """
          from azureml._run_impl.run_history_facade import RunHistoryFacade
          run_dto = RunHistoryFacade.create_run(experiment, name=name, run_id=run_id,
                                                properties=properties, tags=tags, typev2=typev2)
          return Run._dto_to_run(experiment, run_dto, outputs=outputs, **kwargs)
  
      @property
      def id(self):
          """Get run ID.
  
          The ID of the run is an identifier unique across the containing experiment.
  
          :return: The run ID.
          :rtype: str
          """
          return self._run_id
  
      @property
      def name(self):
          """Return the run name.
  
          The optional name of the run is a user-specified string useful for later identification of the run.
  
          :return: The run ID.
          :rtype: str
          """
          return self._run_name
  
      @property
      def type(self):
          """Get run type.
  
          Indicates how the run was created or configured.
  
          :return: The run type.
          :rtype: str
          """
          return self._runtype
  
      @property
      def number(self):
          """Get run number.
  
          A monotonically increasing number representing the order of runs within an experiment.
  
          :return: The run number.
          :rtype: int
          """
          self._logger.warning("Run Number's semantics may be modified in the future")
          return self._run_number
  
      @property
      def parent(self):
          """Fetch the parent run for this run from the service.
  
          Runs can have an optional parent, resulting in a potential tree hierarchy of runs. To log metrics to
          a parent run, use the :meth:`log` method of the parent object, for example, ``run.parent.log()``.
  
          :return: The parent run, or None if one is not set.
          :rtype: azureml.core.run.Run
          """
          if self._parent_run is None:
              parent_run_id = self._run_dto.get('parent_run_id', None)
  
              if parent_run_id is not None:
                  parent_run_obj = get_run(self.experiment, parent_run_id)
                  self._parent_run = parent_run_obj
  
          return self._parent_run
  
      @property
      def status(self):
          """Return the run object's status."""
          return self._client.run_dto.status
  
      def wait_for_completion(self, show_output=False, wait_post_processing=False, raise_on_error=True):
          """Wait for the completion of this run. Returns the status object after the wait.
  
          :param show_output: Indicates whether to show the run output on sys.stdout.
          :type show_output: bool
          :param wait_post_processing: Indicates whether to wait for the post processing to
              complete after the run completes.
          :type wait_post_processing: bool
          :param raise_on_error: Indicates whether an Error is raised when the Run is in a failed state.
          :type raise_on_error: bool
          :return: The status object.
          :rtype: dict
          """
          if show_output:
              try:
                  self._stream_run_output(
                      file_handle=sys.stdout,
                      wait_post_processing=wait_post_processing,
                      raise_on_error=raise_on_error)
                  return self.get_details()
              except KeyboardInterrupt:
                  error_message = "The output streaming for the run interrupted.\n" \
                                  "But the run is still executing on the compute target. \n" \
                                  "Details for canceling the run can be found here: " \
                                  "https://aka.ms/aml-docs-cancel-run"
  
                  raise ExperimentExecutionException(error_message)
          else:
              running_states = RUNNING_STATES
              if wait_post_processing:
                  running_states.extend(POST_PROCESSING_STATES)
  
              current_status = None
              poll_start_time = time.time()
              while current_status is None or current_status in running_states:
                  time.sleep(Run._wait_before_polling(time.time() - poll_start_time))
                  current_status = self.get_status()
                  # add unknown status as running status
                  if current_status not in RunStatus.list():
                      running_states.append(current_status)
  
              final_details = self.get_details()
              if current_status == RunStatus.FAILED:
                  error = final_details.get("error")
                  if not error:
                      error = "Detailed error not set on the Run. Please check the logs for details."
  
                  if raise_on_error:
                      raise ActivityFailedException(error_details=json.dumps(error, indent=4))
  
              return final_details
  
      def clean(self):
          """Remove the files corresponding to the current run on the target specified in the run configuration.
  
          :return: A list of files deleted.
          :rtype: builtin.list
          """
          from azureml._execution import _commands
          return _commands.clean(self._project_object, self._run_config, run_id=self._run_id)
  
      def get_all_logs(self, destination=None):
          """Download all logs for the run to a directory.
  
          :param destination: The destination path to store logs. If unspecified, a directory named as the run ID
              is created in the project directory.
          :type destination: str
          :return: A list of names of logs downloaded.
          :rtype: builtin.list
          """
          if destination:
              if os.path.exists(destination) and not os.path.isdir(destination):
                  raise UserErrorException("{} exists and is not a directory.".format(destination))
          else:
              destination = os.path.join(self._project_object.project_directory, "assets", self.id)
  
          os.makedirs(destination, exist_ok=True)
  
          details = self.get_details_with_logs()
          log_files = details["logFiles"]
          downloaded_logs = []
          for log_name in log_files:
              target_path = os.path.join(destination, log_name)
              os.makedirs(os.path.dirname(target_path), exist_ok=True)
              with open(target_path, 'w') as file:
                  file.write(log_files[log_name])
              downloaded_logs.append(target_path)
  
          return downloaded_logs
  
      def get_details_with_logs(self):
          """Return run status including log file content.
  
          :return: Returns the status for the run with log file contents.
          :rtype: dict
          """
          details = self.get_details()
          log_files = details["logFiles"]
          session = create_session_with_retry()
  
          for log_name in log_files:
              content = download_file_stream(log_files[log_name], session=session)
              log_files[log_name] = content
          return details
  
      # For distributed jobs, the names will be like 60-control_log_rank_0.txt
      # We want to get the highest number in front (60-...) - the number in front is the priority.
      # but also the lowest number in back (rank_0) - lower ranks are usually more "primary"
      # (and if there aren't any ranks in the filenames, just return the latest one)
      @staticmethod
      def _get_last_log_primary_instance(logs):
          """Return last log for primary instance.
  
          :param logs:
          :type logs: builtin.list
          :return: Returns the last log primary instance.
          :rtype:
          """
          primary_ranks = ["rank_0", "worker_0"]
          rank_match_re = re.compile("(.*)_(.*?_.*?)\.txt")
          last_log_name = logs[-1]
  
          last_log_match = rank_match_re.match(last_log_name)
          if not last_log_match:
              return last_log_name
  
          last_log_prefix = last_log_match.group(1)
          matching_logs = sorted(filter(lambda x: x.startswith(last_log_prefix), logs))
  
          # we have some specific ranks that denote the primary, use those if found
          for log_name in matching_logs:
              match = rank_match_re.match(log_name)
              if not match:
                  continue
              if match.group(2) in primary_ranks:
                  return log_name
  
          # no definitively primary instance, just return the highest sorted
          return matching_logs[0]
  
      def _get_logs(self, status):
          """Return logs.
  
          :param status:
          :type status: dict
          :return:
          :rtype: builtin.list
          """
          logs = [x for x in status["logFiles"] if re.match(self._output_logs_pattern, x)]
          logs.sort()
          return logs
  
      def _stream_run_output(self, file_handle=sys.stdout, wait_post_processing=False, raise_on_error=True):
          """Stream the experiment run output to the specified file handle.
  
          By default the the file handle points to stdout.
  
          :param file_handle: A file handle to stream the output to.
          :type file_handle: file
          :param wait_post_processing:
          :type wait_post_processing: bool
          :return:
          :rtype: builtin.list
          """
          from azureml._execution import _commands
  
          def incremental_print(log, printed, fileout):
              """Incremental print.
  
              :param log:
              :type log: dict
              :param printed:
              :type printed: int
              :param fileout:
              :type fileout: TestIOWrapper
              :return:
              :rtype: int
              """
              count = 0
              for line in log.splitlines():
                  if count >= printed:
                      fileout.write(line + "\n")
                      printed += 1
                  count += 1
  
              return printed
  
          web_uri = self.get_portal_url()
  
          file_handle.write("RunId: {}\n".format(self._run_id))
          file_handle.write("Web View: {}\n".format(web_uri))
  
          printed = 0
          current_log = None
          running_states = RUNNING_STATES
          if wait_post_processing:
              running_states.extend(POST_PROCESSING_STATES)
  
          self._current_details = self.get_details()
          session = create_session_with_retry()
  
          # TODO: Temporary solution to wait for all the logs to be printed in the finalizing state.
          poll_start_time = time.time()
          while (self._current_details["status"] in running_states or
                 self._current_details["status"] == RunStatus.FINALIZING):
              file_handle.flush()
              time.sleep(Run._wait_before_polling(time.time() - poll_start_time))
              self._current_details = self.get_details()  # TODO use FileWatcher
  
              # Check whether there is a higher priority log than the one we are currently streaming (current_log)
              available_logs = self._get_logs(self._current_details)
              # next_log is the log we should be following now, based on the available logs we just got
              next_log = Run._get_last_log_primary_instance(available_logs) if available_logs else None
              # if next_log != current_log, we need to switch to streaming next_log
              if available_logs and current_log != next_log:
                  printed = 0
                  current_log = next_log
                  file_handle.write("\n")
                  file_handle.write("Streaming " + current_log + "\n")
                  file_handle.write("=" * (len(current_log) + 10) + "\n")
                  file_handle.write("\n")
  
              if current_log:
                  content = _commands._get_content_from_uri(self._current_details["logFiles"][current_log],
                                                            session,
                                                            timeout=Run._DEFAULT_GET_CONTENT_TIMEOUT)
                  printed = incremental_print(content, printed, file_handle)
  
                  # TODO: Temporary solution to wait for all the logs to be printed in the finalizing state.
                  if (self._current_details["status"] not in running_states and
                          self._current_details["status"] == RunStatus.FINALIZING and
                          "The activity completed successfully. Finalizing run..." in content):
                      break
  
          file_handle.write("\n")
          file_handle.write("Execution Summary\n")
          file_handle.write("=================\n")
          file_handle.write("RunId: {}\n".format(self.id))
          file_handle.write("Web View: {}\n".format(web_uri))
  
          warnings = self._current_details.get("warnings")
          if warnings:
              messages = [x.get("message") for x in warnings if x.get("message")]
              if len(messages) > 0:
                  file_handle.write("\nWarnings:\n")
                  for message in messages:
                      file_handle.write(message + "\n")
                  file_handle.write("\n")
  
          error = self._current_details.get("error")
          if not error:
              error = "Detailed error not set on the Run. Please check the logs for details."
  
          if self._current_details["status"] == RunStatus.FAILED:
              # If we are raising the error later on, so we don't double print.
              if not raise_on_error:
                  file_handle.write("\nError:\n")
                  file_handle.write(json.dumps(error, indent=4))
                  file_handle.write("\n")
              else:
                  raise ActivityFailedException(error_details=json.dumps(error, indent=4))
  
          file_handle.write("\n")
          file_handle.flush()
  
      def get_status(self):
          """Fetch the latest status of the run.
  
          Common values returned include "Running", "Completed", and "Failed".
  
          .. remarks::
  
              * NotStarted - This is a temporary state client-side Run objects are in before cloud submission.
              * Starting - The Run has started being processed in the cloud. The caller has a run ID at this point.
              * Provisioning - Returned when on-demand compute is being created for a given job submission.
              * Preparing - The run environment is being prepared:
                  * docker image build
                  * conda environment setup
              * Queued - The job is queued in the compute target. For example, in BatchAI the job is in queued state
                   while waiting for all the requested nodes to be ready.
              * Running - The job started to run in the compute target.
              * Finalizing - User code has completed and the run is in post-processing stages.
              * CancelRequested - Cancellation has been requested for the job.
              * Completed - The run completed successfully. This includes both the user code and run
                  post-processing stages.
              * Failed - The run failed. Usually the Error property on a run will provide details as to why.
              * Canceled - Follows a cancellation request and indicates that the run is now successfully cancelled.
              * NotResponding - For runs that have Heartbeats enabled, no heartbeat has been recently sent.
  
              .. code-block:: python
  
                  run = experiment.submit(config)
                  while run.get_status() not in ['Completed', 'Failed']: # For example purposes only, not exhaustive
                      print('Run {} not in terminal state'.format(run.id))
                      time.sleep(10)
  
          :return: The latest status.
          :rtype: str
          """
          if self.status not in ["Completed", "Failed", "Canceled"]:
              self._client.refresh_run_dto()
          return self.status
  
      def get_detailed_status(self):
          """Fetch the latest status of the run. If the status of the run is "Queued", it will show the details.
  
          .. remarks::
  
              * `status`: The run's current status. Same value as that returned from `get_status()`.
              * `details`: The detailed information for current status.
  
              .. code-block:: python
  
                  run = experiment.submit(config)
                  details = run.get_detailed_status()
                  # details = {
                  #     'status': 'Queued',
                  #     'details': 'Run requested 1 node(s). Run is in pending status.',
                  # }
  
          :return: The latest status and details
          :rtype: dict[str, str]
          """
          status = self.get_status()
          queueing_info = self._client.run_dto.queueing_info
          details = None if not queueing_info else queueing_info.message
          return {
              "status": status,
              "details": details
          }
  
      def get_environment(self):
          """Get the environment definition that was used by this run.
  
          :return: Return the environment object.
          :rtype: azureml.core.environment.Environment
          """
          run_details = self.get_details()
  
          # For parent run there can be a case when there is no run definition
          run_def = run_details.get("runDefinition")
          if run_def is None:
              raise UserErrorException("There is no run definition or environment specified for the run.")
  
          # For parent run there can be a case when there is no environment
          env = run_def.get("environment")
          if env:
              env_name = env["name"]
              env_version = env["version"]
          else:
              raise Exception("There is no environment specified for the run")
  
          workspace = self._client._experiment.workspace
          environment = Environment.get(workspace, env_name, env_version)
  
          return environment
  
      def get_details(self):
          """Get the definition, status information, current log files, and other details of the run.
  
          .. remarks::
  
              The returned dictionary contains the following key-value pairs:
  
              * `runId`: ID of this run.
              * `target`
              * `status`: The run's current status. Same value as that returned from `get_status()`.
              * `startTimeUtc`: UTC time of when this run was started, in ISO8601.
              * `endTimeUtc`: UTC time of when this run was finished (either Completed or Failed), in ISO8601.
                  This key does not exist if the run is still in progress.
              * `properties`: Immutable key-value pairs associated with the run. Default properties include
                the run's snapshot ID and information about the git repository from which the run was created (if any).
                Additional properties can be added to a run using :func:`add_properties`.
              * `inputDatasets`: Input datasets associated with the run.
              * `outputDatasets`: Output datasets associated with the run.
              * `logFiles`
              * `submittedBy`
  
              .. code-block:: python
  
                  run = experiment.start_logging()
  
                  details = run.get_details()
                  # details = {
                  #     'runId': '5c24aa28-6e4a-4572-96a0-fb522d26fe2d',
                  #     'target': 'sdk',
                  #     'status': 'Running',
                  #     'startTimeUtc': '2019-01-01T13:08:01.713777Z',
                  #     'endTimeUtc': '2019-01-01T17:15:65.986253Z',
                  #     'properties': {
                  #         'azureml.git.repository_uri': 'https://example.com/my/git/repo',
                  #         'azureml.git.branch': 'master',
                  #         'azureml.git.commit': '7dc972657c2168927a02c3bc2b161e0f370365d7',
                  #         'azureml.git.dirty': 'True',
                  #         'mlflow.source.git.repoURL': 'https://example.com/my/git/repo',
                  #         'mlflow.source.git.branch': 'master',
                  #         'mlflow.source.git.commit': '7dc972657c2168927a02c3bc2b161e0f370365d7',
                  #         'ContentSnapshotId': 'b4689489-ce2f-4db5-b6d7-6ad11e77079c'
                  #     },
                  #     'inputDatasets': [{
                  #         'dataset': {'id': 'cdebf245-701d-4a68-8055-41f9cf44f298'},
                  #         'consumptionDetails': {
                  #             'type': 'RunInput',
                  #             'inputName': 'training-data',
                  #             'mechanism': 'Mount',
                  #             'pathOnCompute': '/mnt/datasets/train'
                  #         }
                  #     }],
                  #     'outputDatasets': [{
                  #         'dataset': {'id': 'd04e8a19-1caa-4b1f-b318-4cbff9af9615'},
                  #         'outputType': 'RunOutput',
                  #         'outputDetails': {
                  #             'outputName': 'training-result'
                  #         }
                  #     }],
                  #     'runDefinition': {},
                  #     'logFiles': {},
                  #     'submittedBy': 'Alan Turing'
                  # }
  
          :return: Return the details for the run
          :rtype: dict[str, str]
          """
          details = self._client.get_runstatus().as_dict(key_transformer=camel_case_transformer)
          try:
              details['submittedBy'] = self._client.run_dto.created_by.user_name
          except AttributeError:
              logging.warning("user_name does not exist.")
              details['submittedBy'] = None
          input_datasets = details.get('inputDatasets')
          output_datasets = details.get('outputDatasets')
          if input_datasets is not None:
              from azureml.data._dataset_lineage import _InputDatasetsLineage
              details['inputDatasets'] = _InputDatasetsLineage(self.experiment.workspace, input_datasets)
          if output_datasets:
              from azureml.data._dataset_lineage import update_output_lineage
              update_output_lineage(self.experiment.workspace, output_datasets)
  
          return details
  
      def get_children(self, recursive=False, tags=None, properties=None, type=None, status=None, _rehydrate_runs=True):
          """Get all children for the current run selected by specified filters.
  
          :param recursive: Indicates whether to recurse through all descendants.
          :type recursive: bool
          :param tags:  If specified, returns runs matching specified *"tag"* or {*"tag"*: *"value"*}.
          :type tags: str or dict
          :param properties: If specified, returns runs matching specified *"property"* or {*"property"*: *"value"*}.
          :type properties: str or dict
          :param type: If specified, returns runs matching this type.
          :type type: str
          :param status: If specified, returns runs with status specified *"status"*.
          :type status: str
          :param _rehydrate_runs: Indicates whether to instantiate a run of the original type
              or the base Run.
          :type _rehydrate_runs: bool
          :return: A list of :class:`azureml.core.run.Run` objects.
          :rtype: builtin.list
          """
          children = self._client.get_descendants(
              root_run_id=self._root_run_id,
              recursive=recursive,
              tags=tags,
              properties=properties,
              runtype=type,
              status=status)
  
          self._logger.debug("Rehydrating runs on enumerate: {0}".format(_rehydrate_runs))
          if _rehydrate_runs:
              return Run._rehydrate_runs(self.experiment, children)
          else:
              return (Run._dto_to_run(self.experiment, child) for child in children)
  
      def get_metrics(self, name=None, recursive=False, run_type=None, populate=False):
          """Retrieve the metrics logged to the run.
  
          If ``recursive`` is True (False by default), then fetch metrics for runs in the given run's subtree.
  
          .. remarks::
  
              .. code-block:: python
  
                  run = experiment.start_logging() # run id: 123
                  run.log("A", 1)
                  with run.child_run() as child: # run id: 456
                      child.log("A", 2)
  
                  metrics = run.get_metrics()
                  # metrics = { 'A': 1 }
  
                  metrics = run.get_metrics(recursive=True)
                  # metrics = { '123': { 'A': 1 }, '456': { 'A': 2 } } note key is runId
  
          :param name: The name of the metric.
          :type name: str
          :param recursive: Indicates whether to recurse through all descendants.
          :type recursive: bool
          :param run_type:
          :type run_type: str
          :param populate: Indicates whether to fetch the contents of external data linked to the metric.
          :type populate: bool
          :return: A dictionary containing the users metrics.
          :rtype: dict
          """
          return self._client.get_metrics(name=name, recursive=recursive, run_type=run_type,
                                          populate=populate, root_run_id=self._root_run_id)
  
      def _get_outputs_datapath(self):
          """Get the DataPath to the outputs location of the run.
  
          :return DataPath to the outputs location of the run.
          :rtype azureml.data.datapath.DataPath
          """
          outputs_dir = self._outputs if self._outputs else "outputs"
  
          datastore = self._get_blob_datastore_from_run()
  
          return DataPath(datastore=datastore,
                          path_on_datastore="{}/{}/{}".format(RUN_ORIGIN, self._container, outputs_dir))
  
      def _get_blob_datastore_from_run(self):
          """Get a AzureBlobDataStore used by the run.
  
          :return AzureBlobDataStore pointing to a container used by the run.
          :rtype azureml.data.azure_storage_datastore.AzureBlobDatastore
          """
          workspace = self.experiment.workspace
  
          artifacts = [artifact for artifact in
                       self._client.artifacts.get_artifact_by_container(RUN_ORIGIN, self._container)]
  
          if artifacts and artifacts[0].data_path and artifacts[0].data_path.datastore_name:
              datastore = Datastore.get(workspace, artifacts[0].data_path.datastore_name)
          else:
              ws_datastore = Datastore.get(workspace, datastore_name='workspaceblobstore')
  
              datastore = Datastore.register_azure_blob_container(
                  workspace=workspace, datastore_name="blobstorage_datastore", container_name="azureml",
                  account_name=ws_datastore.account_name, account_key=ws_datastore.account_key,
                  create_if_not_exists=False)
  
          return datastore
  
      @property
      def experiment(self):
          """Get experiment containing the run.
  
          :return: Retrieves the experiment corresponding to the run.
          :rtype: azureml.core.Experiment
          """
          return self._experiment
  
      def start(self):
          """Mark the run as started.
  
          This is typically used in advanced scenarios when the run has been created by another actor.
          """
          self._client.start()
  
      def complete(self, _set_status=True):
          """Wait for task queue to be processed.
  
          Then run is marked as completed. This is typically used in interactive notebook scenarios.
  
          :param _set_status: Indicates whether to send the status event for tracking.
          :type _set_status: bool
          """
          self._client.complete(_set_status=_set_status)
  
      def fail(self, error_details=None, error_code=None, _set_status=True):
          """Mark the run as failed.
  
          Optionally set the Error property of the run with a message or exception passed to ``error_details``.
  
          :param error_details: Optional details of the error.
          :type error_details: str or BaseException
          :param error_code: Optional error code of the error for the error classification.
          :type error_code: str
          :param _set_status: Indicates whether to send the status event for tracking.
          :type _set_status: bool
          """
          self._client.fail(error_details=error_details, error_code=error_code, _set_status=_set_status)
  
      def cancel(self):
          """Mark the run as canceled.
  
          If there is an associated job with a set cancel_uri field, terminate that job as well.
          """
          uri = self._run_dto.get('cancel_uri', None)
          self._client.cancel(uri=uri)
  
      def flush(self, timeout_seconds=300):
          """Wait for task queue to be processed.
  
          :param timeout_seconds: How long to wait (in seconds) for task queue to be processed.
          :type timeout_seconds: int
          """
          self._client.flush(timeout_seconds=timeout_seconds)
  
      def __enter__(self):
          """Initialize features related to outputs tracking and status updates for the run.".
  
          :return:
          :rtype: Run
          """
          super(Run, self).__enter__()
          return self
  
      def __exit__(self, exit_type, value, traceback):
          """Close tracking and collect outputs.
  
          :param exit_type:
          :type exit_type:
          :param value:
          :type value:
          :param traceback:
          :type traceback:
          :return:
          :rtype: Run
          """
          return super(Run, self).__exit__(exit_type, value, traceback)
  
      def child_run(self, name=None, run_id=None, outputs=None):
          """Create a child run.
  
          .. remarks::
  
              This is used to isolate part of a run into a subsection. This can be done for
              identifiable "parts" of a run that are interesting to separate, or to capture
              independent metrics across an interation of a subprocess.
  
              If an output directory is set for the child run, the contents of that directory will be
              uploaded to the child run record when the child is completed.
  
          :param name: An optional name for the child run, typically specified for a "part".
          :type name: str
          :param run_id: An optional run ID for the child, otherwise it is auto-generated. Typically this
              parameter is not set.
          :type run_id: str
          :param outputs: Optional outputs directory to track for the child.
          :type outputs: str
          :return: The child run.
          :rtype: azureml.core.run.Run
          """
          run_dto = self._run_dto
          child_dto = self._client.create_child_run(run_dto.get("name"),
                                                    run_dto.get("target", SDK_TARGET),
                                                    child_name=name,
                                                    run_id=run_id)
          child = Run._dto_to_run(self.experiment, child_dto, outputs=outputs,
                                  _worker_pool=self._client.worker_pool, _parent_logger=self._logger)
          child._client.start()
          return child
  
      def create_children(self, count=None, tag_key=None, tag_values=None):
          """Create one or many child runs.
  
          .. remarks::
  
              Either parameter ``count`` OR  parameters ``tag_key`` AND ``tag_values`` must be specified.
  
          :param count: An optional number of children to create.
          :type count: int
          :param tag_key: An optional key to populate the Tags entry in all created children.
          :type tag_key: str
          :param tag_Values: An optional list of values that will map onto Tags[tag_key] for the list of runs created.
          :type tag_values: [str]
          :return: The list of child runs.
          :rtype: builtin.list[azureml.core.run.Run]
          """
          if tag_key is None and tag_values is None and count is None:
              raise UserErrorException("Either count or tag_key and tag_values must be set")
          if tag_key is not None or tag_values is not None:
              if tag_key is None:
                  raise UserErrorException("tag_key must not be null when tag_values is set")
              if tag_values is None:
                  raise UserErrorException("tag_values must not be null when tag_key is set")
              if count is not None:
                  raise UserErrorException("count can not be set when tag_key and tag_values are set")
              if len(tag_values) > 100:
                  raise UserErrorException("create_children can not create more than 100 runs")
  
          if count is not None:
              if count > 100:
                  raise UserErrorException("create_children can not create more than 100 runs")
              tag_key = "azureml.batch_child_number"
              tag_values = list(range(count))
  
          child_dtos = self._client.create_children(tag_key, tag_values)
          children = []
          for child_dto in child_dtos:
              child = Run._dto_to_run(self.experiment, child_dto,
                                      _worker_pool=self._client.worker_pool, _parent_logger=self._logger)
              children.append(child)
          return children
  
      def tag(self, key, value=None):
          """Tag the run with a string key and optional string value.
  
          .. remarks::
  
              Tags and properties on a run are both dictionaries of string -> string. The difference between them
              is mutability: Tags can be set, updated, and deleted while Properties can only be added. This makes
              Properties more appropriate for system/workflow related behavior triggers, while Tags are generally
              user-facing and meaningful for the consumers of the experiment.
  
              .. code-block:: python
  
                  run = experiment.start_logging()
                  run.tag('DeploymentCandidate')
                  run.tag('modifiedBy', 'Master CI')
                  run.tag('modifiedBy', 'release pipeline') # Careful, tags are mutable
  
                  run.add_properties({'BuildId': os.environ.get('VSTS_BUILD_ID')}) # Properties are not
  
                  tags = run.get_tags()
                  # tags = { 'DeploymentCandidate': None, 'modifiedBy': 'release pipeline' }
  
  
          :param key: The tag key
          :type key: str
          :param value: An optional value for the tag
          :type value: str
          """
          self.set_tags({key: value})
  
      def set_tags(self, tags):
          """Add or modify a set of tags on the run. Tags not passed in the dictionary are left untouched.
  
          You can also add simple string tags. When these tags appear in the tag dictionary as keys,
          they have a value of None. For more information, see `Tag and find runs
          <https://docs.microsoft.com/azure/machine-learning/how-to-manage-runs#tag-and-find-runs>`_.
  
          :param tags: The tags stored in the run object.
          :type tags: dict[str] or str
          """
          self._client.set_tags(tags)
  
      @property
      def tags(self):
          """Return the set of mutable tags on this run.
  
          :return: The tags stored on the run object.
          :rtype: dict
          """
          return self._run_dto.get('tags', {})
  
      def get_tags(self):
          """Fetch the latest set of mutable tags on the run from the service.
  
          :return: The tags stored on the run object.
          :rtype: dict
          """
          return self._client.get_tags()
  
      def remove_tags(self, tags):
          """Delete the list of mutable tags on this run.
  
          :param tags: A list of tags to remove.
          :type tags: builtin.list
          :return: The tags stored on the run object
          """
          self._client.delete_tags(tags)
  
      def add_properties(self, properties):
          """Add immutable properties to the run.
  
          Tags and properties (both dict[str, str]) differ in their mutability.
          Properties are immutable, so they create a permanent record for auditing purposes.
          Tags are mutable. For more information about working with tags and properties, see `Tag
          and find runs <https://docs.microsoft.com/azure/machine-learning/how-to-manage-runs#tag-and-find-runs>`_.
  
          :param properties: The hidden properties stored in the run object.
          :type properties: dict
          """
          if properties:  # Empty dict is a no-op.
              self._client.add_properties(properties)
  
      @property
      def properties(self):
          """Return the immutable properties of this run.
  
          .. remarks::
  
              Properties include immutable system-generated information such as
              duration, date of execution, user, etc.
  
          :return: The locally cached properties of the run.
          :rtype: dict[str] or str
          """
          return self._run_dto.get('properties', {})
  
      def get_properties(self):
          """Fetch the latest properties of the run from the service.
  
          .. remarks::
  
              Properties are immutable system-generated information such as
              duration, date of execution, user, and custom properties added with the
              :meth:`azureml.core.Run.add_properties` method. For more information, see `Tag and find runs
              <https://docs.microsoft.com/azure/machine-learning/how-to-manage-runs#tag-and-find-runs>`_.
  
              When submitting a job to Azure Machine Learning, if source files are stored in a local git
              repository then information about the repo is stored as properties.
              These git properties are added when creating a run or calling Experiment.submit.
              For more information about git properties see `Git integration for Azure Machine
              Learning <https://docs.microsoft.com/azure/machine-learning/concept-train-model-git-integration>`_.
  
          :return: The properties of the run.
          :rtype: dict
          """
          return self._client.get_properties()
  
      def log(self, name, value, description=""):
          """Log a metric value to the run with the given name.
  
          .. remarks::
  
              Logging a metric to a run causes that metric to be stored in
              the run record in the experiment.  You can log the same metric
              multiple times within a run, the result being considered a vector
              of that metric.
  
              .. code-block: python
  
                  run = experiment.start_logging()
                  run.log('Accuracy', (tp + tn) / (p + n))
                  run.log('Precision', tp / (tp + fp))
                  run.log('Recall', tp / (tp + fn))
  
          :param name: The name of metric.
          :type name: str
          :param value: The value to be posted to the service.
          :type value:
          :param description: An optional metric description.
          :type description: str
          """
          # python2/3 compatible check for range, range is not a python2 type
          is_range = isinstance(value, type(range(0)))
          if any([isinstance(value, list),
                  is_range,
                  isgeneratorfunction(value),
                  isinstance(value, GeneratorType)]):
              raise UserErrorException("run.log does not support "
                                       "list type. Use run.log_list instead")
          else:
              self._client.log_scalar(name, value, description=description)
  
      def log_list(self, name, value, description=""):
          """Log a list of metric values to the run with the given name.
  
          :param name: The name of metric.
          :type name: str
          :param value: The values of the metric.
          :type value: builtin.list
          :param description: An optional metric description.
          :type description: str
          """
          self._client.log_list(name, value, description=description)
  
      def log_row(self, name, description=None, **kwargs):
          """Log a row metric to the run with the given name.
  
          .. remarks::
  
              Using ``log_row`` creates a table metric with columns as described in kwargs.  Each named
              parameter generates a column with the value specified. ``log_row`` can be
              be called once to log an arbitrary tuple, or multiple times in a loop to generate
              a complete table.
  
              .. code-block:: python
  
                  citrus = ['orange', 'lemon', 'lime']
                  sizes = [ 10, 7, 3]
                  for index in range(len(citrus)):
                      run.log_row("citrus", fruit = citrus[index], size=sizes[index])
  
          :param name: The name of metric.
          :type name: str
          :param description: An optional metric description.
          :type description: str
          :param kwargs: A dictionary of additional parameters. In this case, the columns of the metric.
          :type kwargs: dict
          """
          self._client.log_row(name, kwargs, description=description)
  
      def log_table(self, name, value, description=""):
          """Log a table metric to the run with the given name.
  
          :param name: The name of metric.
          :type name: str
          :param value: The table value of the metric, a dictionary where keys are columns to be posted to the service.
          :type value: dict
          :param description: An optional metric description.
          :type description: str
          """
          self._client.log_table(name, value, description=description)
  
      @_check_for_data_container_id
      def log_confusion_matrix(self, name, value, description=""):
          """Log a confusion matrix to the artifact store.
  
          This logs a wrapper around the sklearn confusion matrix.
          The metric data contains the class labels and a 2D list
          for the matrix itself.
          See the following link for more details on how the metric is computed:
          `https://scikit-learn.org/stable/modules/generated/sklearn.metrics.confusion_matrix.html`
  
          .. remarks::
  
              Example of a valid JSON value:
  
              .. code-block:: json
  
                  {
                      "schema_type": "confusion_matrix",
                      "schema_version": "1.0.0",
                      "data": {
                          "class_labels": ["0", "1", "2", "3"],
                          "matrix": [
                              [3, 0, 1, 0],
                              [0, 1, 0, 1],
                              [0, 0, 1, 0],
                              [0, 0, 0, 1]
                          ]
                      }
                  }
  
          :param name: The name of the confusion matrix.
          :type name: str
          :param value: JSON containing name, version, and data properties.
          :type value: str or dict
          :param description: An optional metric description.
          :type description: str
          :return:
          """
          self._client.log_confusion_matrix(name, value, description=description)
  
      @_check_for_data_container_id
      def log_accuracy_table(self, name, value, description=""):
          """Log an accuracy table to the artifact store.
  
          The accuracy table metric is a multi-use, non-scalar metric
          that can be used to produce multiple types of line charts
          that vary continuously over the space of predicted probabilities.
          Examples of these charts are ROC, precision-recall, and lift curves.
  
          The calculation of the accuracy table is similar to the calculation
          of an ROC curve. An ROC curve stores true positive rates and
          false positive rates at many different probability thresholds.
          The accuracy table stores the raw number of
          true positives, false positives, true negatives, and false negatives
          at many probability thresholds.
  
          There are two methods used for selecting thresholds: "probability" and
          "percentile." They differ in how they sample from the space of
          predicted probabilities.
  
          Probability thresholds are uniformly spaced thresholds between 0 and 1.
          If NUM_POINTS is 5 the probability thresholds would be
          [0.0, 0.25, 0.5, 0.75, 1.0].
  
          Percentile thresholds are spaced according to the distribution of
          predicted probabilities. Each threshold corresponds to the percentile
          of the data at a probability threshold.
          For example, if NUM_POINTS is 5, then the first threshold would be at
          the 0th percentile, the second at the 25th percentile, the
          third at the 50th, and so on.
  
          The probability tables and percentile tables are both 3D lists where
          the first dimension represents the class label, the second dimension
          represents the sample at one threshold (scales with NUM_POINTS),
          and the third dimension always has 4 values: TP, FP, TN, FN, and
          always in that order.
  
          The confusion values (TP, FP, TN, FN) are computed with the
          one vs. rest strategy. See the following link for more details:
          `https://en.wikipedia.org/wiki/Multiclass_classification`
  
          .. remarks::
  
              Example of a valid JSON value:
  
              .. code-block:: json
  
                  {
                      "schema_type": "accuracy_table",
                      "schema_version": "1.0.1",
                      "data": {
                          "probability_tables": [
                              [
                                  [82, 118, 0, 0],
                                  [75, 31, 87, 7],
                                  [66, 9, 109, 16],
                                  [46, 2, 116, 36],
                                  [0, 0, 118, 82]
                              ],
                              [
                                  [60, 140, 0, 0],
                                  [56, 20, 120, 4],
                                  [47, 4, 136, 13],
                                  [28, 0, 140, 32],
                                  [0, 0, 140, 60]
                              ],
                              [
                                  [58, 142, 0, 0],
                                  [53, 29, 113, 5],
                                  [40, 10, 132, 18],
                                  [24, 1, 141, 34],
                                  [0, 0, 142, 58]
                              ]
                          ],
                          "percentile_tables": [
                              [
                                  [82, 118, 0, 0],
                                  [82, 67, 51, 0],
                                  [75, 26, 92, 7],
                                  [48, 3, 115, 34],
                                  [3, 0, 118, 79]
                              ],
                              [
                                  [60, 140, 0, 0],
                                  [60, 89, 51, 0],
                                  [60, 41, 99, 0],
                                  [46, 5, 135, 14],
                                  [3, 0, 140, 57]
                              ],
                              [
                                  [58, 142, 0, 0],
                                  [56, 93, 49, 2],
                                  [54, 47, 95, 4],
                                  [41, 10, 132, 17],
                                  [3, 0, 142, 55]
                              ]
                          ],
                          "probability_thresholds": [0.0, 0.25, 0.5, 0.75, 1.0],
                          "percentile_thresholds": [0.0, 0.01, 0.24, 0.98, 1.0],
                          "class_labels": ["0", "1", "2"]
                      }
                  }
  
          N = # of samples in validation dataset (200 in example)
          M = # thresholds = # samples taken from the probability space (5 in example)
          C = # classes in full dataset (3 in example)
  
          Some invariants of the accuracy table:
          - TP + FP + TN + FN = N for all thresholds for all classes
          - TP + FN is the same at all thresholds for any class
          - TN + FP is the same at all thresholds for any class
          - Probability tables and percentile tables have shape [C, M, 4]
  
          Note: M can be any value and controls the resolution of the charts
          This is independent of the dataset, is defined when calculating
          metrics, and trades off storage space, computation time, and resolution.
  
          Class labels should be strings, confusion values should be integers, and
          thresholds should be floats.
  
          :param name: The name of the accuracy table.
          :type name: str
          :param value: JSON containing name, version, and data properties.
          :type value: str or dict
          :param description: An optional metric description.
          :type description: str
          :return:
          """
          self._client.log_accuracy_table(name, value, description=description)
  
      @_check_for_data_container_id
      def log_residuals(self, name, value, description=""):
          """Log residuals to the artifact store.
  
          This logs the data needed to display a histogram of
          residuals for a regression task.
          The residuals are predicted - actual.
  
          There should be one more edge than the number of counts.
          Please see the numpy histogram documentation for examples of
          using counts and edges to represent a histogram.
          `https://docs.scipy.org/doc/numpy/reference/generated/numpy.histogram.html`
  
          .. remarks::
  
              Example of a valid JSON value:
  
              .. code-block:: json
  
                  {
                      "schema_type": "residuals",
                      "schema_version": "1.0.0",
                      "data": {
                          "bin_edges": [50, 100, 200, 300, 350],
                          "bin_counts": [0.88, 20, 30, 50.99]
                      }
                  }
  
          :param name: The name of the residuals.
          :type name: str
          :param value: JSON containing name, version, and data properties.
          :type value: str or dict
          :param description: An optional metric description.
          :type description: str
          :return:
          """
          self._client.log_residuals(name, value, description=description)
  
      @_check_for_data_container_id
      def log_predictions(self, name, value, description=""):
          """Log predictions to the artifact store.
  
          This logs a metric score that can be used to compare
          the distributions of true target values to the
          distribution of predicted values for a regression task.
  
          The predictions are binned and standard deviations are calculated
          for error bars on a line chart.
  
          .. remarks::
  
              Example of a valid JSON value:
  
              .. code-block:: json
  
                  {
                      "schema_type": "predictions",
                      "schema_version": "1.0.0",
                      "data": {
                          "bin_averages": [0.25, 0.75],
                          "bin_errors": [0.013, 0.042],
                          "bin_counts": [56, 34],
                          "bin_edges": [0.0, 0.5, 1.0]
                      }
                  }
  
          :param name: The name of the predictions.
          :type name: str
          :param value: JSON containing name, version, and data properties.
          :type value: str or dict
          :param description: An optional metric description.
          :type description: str
          :return:
          """
          self._client.log_predictions(name, value, description=description)
  
      @_check_for_data_container_id
      def log_image(self, name, path=None, plot=None, description=""):
          """Log an image metric to the run record.
  
          .. remarks::
  
              Use this method to log an image file or a `matplotlib plot <https://matplotlib.org/api/pyplot_api.html>`_
              to the run. These images will be visible and comparable in the run record.
  
          :param name: The name of the metric.
          :type name: str
          :param path: The path or stream of the image.
          :type path: str
          :param plot: The plot to log as an image.
          :type plot: matplotlib.pyplot
          :param description: An optional metric description.
          :type description: str
          :return:
          """
          return self._client.log_image(name, path=path, plot=plot, description=description)
  
      @_check_for_data_container_id
      def upload_file(self, name, path_or_stream, datastore_name=None):
          r"""Upload a file to the run record.
  
          .. remarks::
  
              .. code-block:: python
  
                  run = experiment.start_logging()
                  run.upload_file(name='important_file', path_or_stream="path/on/disk/file.txt")
  
              .. note::
  
                  Runs automatically capture file in the specified output directory, which defaults \
                  to "./outputs" for most run types.  Use upload_file only when additional files need \
                  to be uploaded or an output directory is not specified.
  
          :param name: The name of the file to upload.
          :type name: str
          :param path_or_stream: The relative local path or stream to the file to upload.
          :type path_or_stream: str
          :param datastore_name: Optional DataStore name
          :type datastore_name: str
          :return:
          :rtype: azure.storage.blob.models.ResourceProperties
          """
          return self._client.artifacts.upload_artifact(path_or_stream, RUN_ORIGIN, self._container, name,
                                                        datastore_name=datastore_name)
  
      @_check_for_data_container_id
      def upload_folder(self, name, path, datastore_name=None):
          r"""Upload the specified folder to the given prefix name.
  
          .. remarks::
  
              .. code-block:: python
  
                  run = experiment.start_logging()
                  run.upload_folder(name='important_files', path='path/on/disk')
  
                  run.download_file('important_files/existing_file.txt', 'local_file.txt')
  
              .. note::
  
                  Runs automatically capture files in the specified output directory, which defaults \
                  to "./outputs" for most run types.  Use upload_folder only when additional files need \
                  to be uploaded or an output directory is not specified.
  
          :param name: The name of the folder of files to upload.
          :type name: str
          :param folder: The relative local path to the folder to upload.
          :type folder: str
          :param datastore_name: Optional DataStore name
          :type datastore_name: str
          """
          return self._client.artifacts.upload_dir(path, RUN_ORIGIN, self._container,
                                                   lambda fpath: name + fpath[len(path):],
                                                   datastore_name=datastore_name)
  
      @_check_for_data_container_id
      def upload_files(self, names, paths, return_artifacts=False, timeout_seconds=None,
                       datastore_name=None):
          r"""Upload files to the run record.
  
          .. remarks::
  
              ``upload_files`` has the same effect as ``upload_file`` on separate files, however
              there are performance and resource utilization benefits when using ``upload_files``.
  
              .. code-block:: python
  
                  import os
  
                  run = experiment.start_logging()
                  file_name_1 = 'important_file_1'
                  file_name_2 = 'important_file_2'
                  run.upload_files(names=[file_name_1, file_name_2],
                                      paths=['path/on/disk/file_1.txt', 'other/path/on/disk/file_2.txt'])
  
                  run.download_file(file_name_1, 'file_1.txt')
  
                  os.mkdir("path")  # The path must exist
                  run.download_file(file_name_2, 'path/file_2.txt')
  
              .. note::
  
                  Runs automatically capture files in the specified output directory, which defaults \
                  to "./outputs" for most run types.  Use upload_files only when additional files need \
                  to be uploaded or an output directory is not specified.
  
          :param names: The names of the files to upload. If set, paths must also be set.
          :type names: builtin.list
          :param paths: The relative local paths to the files to upload. If set, names is required.
          :type paths: builtin.list
          :param return_artifacts: Indicates that an artifact object should be returned for each file uploaded.
          :type return_artifacts: bool
          :param timeout_seconds: The timeout for uploading files.
          :type timeout_seconds: int
          :param datastore_name: Optional DataStore name
          :type datastore_name: str
          """
          if len(names) != len(paths):
              raise TypeError("names and paths must be lists of equal lengths. "
                              "Names was of length {} and paths of length {}.".format(len(names), len(paths)))
          return self._client.artifacts.upload_files(paths, RUN_ORIGIN, self._container, names,
                                                     return_artifacts=return_artifacts,
                                                     timeout_seconds=timeout_seconds,
                                                     datastore_name=datastore_name)
  
      def get_file_names(self):
          """List the files that are stored in association with the run.
  
          :return: The list of paths for existing artifacts
          :rtype: builtin.list[str]
          """
          if self._container is None:
              return []
          else:
              return list(self._client.artifacts.get_file_paths(RUN_ORIGIN, self._container))
  
      @_check_for_data_container_id
      def download_file(self, name, output_file_path=None, _validate_checksum=False):
          """Download an associated file from storage.
  
          :param name: The name of the artifact to be downloaded.
          :type name: str
          :param output_file_path: The local path where to store the artifact.
          :type output_file_path: str
          """
          if output_file_path is None:
              self._logger.debug("output_file_path for download_file was not set.")
              output_file_path = os.path.basename(name)
  
          self._client.artifacts.download_artifact(RUN_ORIGIN, self._container, name, output_file_path,
                                                   _validate_checksum)
  
      @_check_for_data_container_id
      def download_files(self,
                         prefix=None,
                         output_directory=None,
                         output_paths=None,
                         batch_size=100,
                         append_prefix=True):
          """Download files from a given storage prefix (folder name) or the entire container if prefix is unspecified.
  
          :param prefix: The filepath prefix within the container from which to download all artifacts.
          :type prefix: str
          :param output_directory: An optional directory that all artifact paths use as a prefix.
          :type output_directory: str
          :param output_paths: Optional filepaths in which to store the downloaded artifacts.
              Should be unique and match length of paths.
          :type output_paths: [str]
          :param batch_size: The number of files to download per batch. The default is 100 files.
          :type batch_size: int
          :param append_prefix: An optional flag whether to append the specified prefix from the final output file path.
              If False then the prefix is removed from the output file path.
          :type append_prefix: bool
          """
          self._client.artifacts.download_artifacts_from_prefix(RUN_ORIGIN,
                                                                self._container,
                                                                prefix=prefix,
                                                                output_directory=output_directory,
                                                                output_paths=output_paths,
                                                                batch_size=batch_size,
                                                                append_prefix=append_prefix)
  
      @_check_for_data_container_id
      def _download_artifact_contents_to_string(self, name):
          """Download the contents of an artifact saved to the run."""
          return self._client.artifacts.download_artifact_contents_to_string(
              origin=RUN_ORIGIN, container=self._container, path=name)
  
      def take_snapshot(self, file_or_folder_path):
          r"""Save a snapshot of the input file or folder.
  
          .. remarks::
  
              Snapshots are intended to be the *source code* used to execute the experiment run.
              These are stored with the run so that the run trial can be replicated in the future.
  
              .. note::
  
                  Snapshots are automatically taken when :func:`azureml.core.Experiment.submit` is called. \
                  Typically, this the ``take_snapshot`` method  is only required for interactive (notebook) runs.
  
          :param file_or_folder_path: The file or folder containing the run source code.
          :type file_or_folder_path: str
          :return: Returns the snapshot ID.
          :rtype: str
          """
          return self._client.take_snapshot(file_or_folder_path)
  
      def restore_snapshot(self, snapshot_id=None, path=None):
          """Restore a snapshot as a ZIP file. Returns the path to the ZIP.
  
          :param snapshot_id: The snapshot ID to restore. The latest is used if not specified.
          :type snapshot_id: str
          :param path: The path where the downloaded ZIP is saved.
          :type path: str
          :return: The path.
          :rtype: str
          """
          if snapshot_id is None:
              snapshot_id = self.get_snapshot_id()
  
          return self._client.snapshots.restore_snapshot(snapshot_id, path)
  
      def get_snapshot_id(self):
          """Get the latest snapshot ID.
  
          :return: The most recent snapshot ID.
          :rtype: str
          """
          return self._client.get_snapshot_id()
  
      def register_model(self, model_name, model_path=None, tags=None, properties=None, model_framework=None,
                         model_framework_version=None, description=None, datasets=None,
                         sample_input_dataset=None, sample_output_dataset=None, resource_configuration=None, **kwargs):
          """Register a model for operationalization.
  
          .. remarks::
  
              .. code-block:: python
  
                  model = best_run.register_model(model_name = 'best_model', model_path = 'outputs/model.pkl')
  
          :param model_name: The name of the model.
          :type model_name: str
          :param model_path: The relative cloud path to the model, for example, "outputs/modelname".
              When not specified (None), ``model_name`` is used as the path.
          :type model_path: str
          :param tags: A dictionary of key value tags to assign to the model.
          :type tags: dict[str, str]
          :param properties: A dictionary of key value properties to assign to the model. These properties cannot
              be changed after model creation, however new key value pairs can be added.
          :type properties: dict[str, str]
          :param model_framework: The framework of the model to register.
              Currently supported frameworks: TensorFlow, ScikitLearn, Onnx, Custom, Multi
          :type model_framework: str
          :param model_framework_version: The framework version of the registered model.
          :type model_framework_version: str
          :param description: An optional description of the model.
          :type description: str
          :param datasets: A list of tuples where the first element describes the dataset-model relationship and
              the second element is the dataset.
          :type datasets: builtin.list[(str, azureml.data.abstract_dataset.AbstractDataset)]
          :param sample_input_dataset: Optional. Sample input dataset for the registered model
          :type sample_input_dataset: azureml.data.abstract_dataset.AbstractDataset
          :param sample_output_dataset: Optional. Sample output dataset for the registered model
          :type sample_output_dataset: azureml.data.abstract_dataset.AbstractDataset
          :param resource_configuration: Optional. Resource configuration to run the registered model
          :type resource_configuration: azureml.core.resource_configuration.ResourceConfiguration
          :param kwargs: Optional parameters.
          :type kwargs: dict
          :return: The registered model.
          :rtype: azureml.core.model.Model
          """
          model_name_validation(model_name)
          return self._client.register_model(
              model_name, model_path, tags, properties, model_framework, model_framework_version,
              description=description, datasets=datasets, unpack=False, sample_input_dataset=sample_input_dataset,
              sample_output_dataset=sample_output_dataset, resource_configuration=resource_configuration, **kwargs)
  
      def _update_dataset_lineage(self, datasets):
          self._client._update_dataset_lineage(datasets)
  
      def _update_output_dataset_lineage(self, output_datasets):
          self._client._update_output_dataset_lineage(output_datasets)
  
  
  def get_run(experiment, run_id, rehydrate=True):
      """Get the run for this experiment with its run ID.
  
      :param experiment: The containing experiment.
      :type experiment: azureml.core.Experiment
      :param run_id: The run ID.
      :type run_id: string
      :param rehydrate: Indicates whether the original run object is returned or just a base run object.
          If True, this function returns the original run object type. For example, for an AutoML run,
          an :class:`azureml.train.automl.run.AutoMLRun` object is returned, while for
          a HyperDrive run, a :class:`azureml.train.hyperdrive.HyperDriveRun` object is returned.
  
          If False, the function returns a :class:`azureml.core.run.Run` object.
      :type rehydrate: boolean
      :return: The submitted run.
      :rtype: azureml.core.run.Run
      """
      from azureml._restclient.run_client import RunClient
      client = RunClient(experiment.workspace.service_context, experiment.name, run_id, experiment_id=experiment.id)
      run_dto = client.get_run()
      if rehydrate:
          runs = Run._rehydrate_runs(experiment, [run_dto])
          return next(runs)
      return Run._dto_to_run(experiment, run_dto)
  
  
  class _SubmittedRun(Run):
      """Run class for remote execution."""
  
      __instances = {}
  
      @staticmethod
      def _get_instance(experiment, run_id, **kwargs):
          experiment_name = experiment.name
          workspace_name = experiment.workspace.name
          subscription_id = experiment.workspace.subscription_id
          resource_group = experiment.workspace.resource_group
  
          arm_scope_with_run_id = (subscription_id, resource_group, workspace_name, experiment_name, run_id)
  
          run = _SubmittedRun.__instances.get(arm_scope_with_run_id)
          if run is None:
              run = _SubmittedRun(experiment, run_id, **kwargs)
              _SubmittedRun.__instances[arm_scope_with_run_id] = run
          return run
  
      def __init__(self, *args, **kwargs):
          super(_SubmittedRun, self).__init__(*args, **kwargs)
          self._input_datasets = None
          self._output_datasets = None
  
      def complete(self):
          """Override complete since it is not supported for submitted runs."""
          self._logger.info("complete is not setting status for submitted runs.")
          super(_SubmittedRun, self).complete(_set_status=False)
  
      def fail(self, error_details=None, error_code=None):
          """Override fail since it is not supported for submitted runs."""
          self._logger.info("fail is not setting status for submitted runs.")
          super(_SubmittedRun, self).fail(error_details=error_details, error_code=error_code, _set_status=False)
  
      def cancel(self):
          """Override cancel since it is not supported for submitted runs."""
          self._logger.warning("cancel is not supported for submitted runs.")
  
      def __enter__(self):
          """Override __enter__ since it is not supported for submitted runs."""
          return self
  
      def __exit__(self, *args):
          """Override __exit__ since it is not supported for submitted runs.
  
          :param args:
          """
          self.complete()
  
      @property
      def input_datasets(self):
          """Return the dictionary for input datasets.
  
          :return: A dictionary with the name being the dataset's name and value being the delivered data. If the mode
              is set to mount or download, it will return the base path of the delivered data. If the mode is set to
              direct, it will return the dataset object.
          :rtype: azureml.core.run.InputDatasets
          """
          if self._input_datasets is None:
              self._input_datasets = InputDatasets(self.experiment.workspace)
          return self._input_datasets
  
      @property
      def output_datasets(self):
          """Return the dictionary for output datasets.
  
          :return: A dictionary with the name being the output's name and value being the path that the output should
              be written to.
          :rtype: OutputDatasets
          """
          if self._output_datasets is None:
              self._output_datasets = OutputDatasets(self.experiment.workspace)
          return self._output_datasets
  
  
  class _OfflineRun(ChainedIdentity):
      def __init__(self, parent_logger=None, run_id=None, **kwargs):
          self._run_id = "OfflineRun_{}".format(uuid4()) if run_id is None else run_id
          super(_OfflineRun, self).__init__(
              _ident=self._run_id,
              _parent_logger=parent_logger if parent_logger is not None else module_logger)
  
      @property
      def id(self):
          return self._run_id
  
      def add_properties(self, properties):
          print("Attempted to add properties {0}".format(properties))
  
      def set_tags(self, tags):
          print("Attempted to set tags {0}".format(tags))
  
      def tag(self, name, value=None):
          print("Attempted to add tag {0} with value {1}".format(name, value))
  
      def complete(self):
          self._logger.debug("complete is not supported for offline runs.")
  
      def fail(self, error_details=None, error_code=None):
          self._logger.debug("fail is not supported for offline runs.")
          print("Run failed with error code {}:\n{}".format(str(error_code), str(error_details)))
  
      def cancel(self):
          self._logger.debug("cancel is not supported for offline runs.")
  
      def log(self, name, value, description=""):
          self._emit("scalar", name, value)
  
      def log_list(self, name, value, description=""):
          self._emit("list", name, value)
  
      def log_table(self, name, value, description=""):
          self._emit("table", name, value)
  
      def log_row(self, name, description=None, **kwargs):
          self._emit("row", name, kwargs)
  
      def log_image(self, name, path=None, plot=None, description=""):
          self._emit("image", name, plot or path)
  
      def log_residuals(self, name, value, description=None):
          self._emit("residual", name, value)
  
      def log_predictions(self, name, value, description=None):
          self._emit("predictions", name, value)
  
      def log_accuracy_table(self, name, value, description=None):
          self._emit("accuracy_table", name, value)
  
      def log_confusion_matrix(self, name, value, description=None):
          self._emit("confusion_matrix", name, value)
  
      def upload_folder(self, name, path, datastore_name=None):
          print("Attempted to track folder {0} at {1}".format(name, path))
  
      def upload_files(self, name, paths, return_artifacts=False, timeout_seconds=None, datastore_name=None):
          print("Attempted to track files {0} at {1} with timeout {2}s".format(name, paths, timeout_seconds))
  
      def upload_file(self, name, path_or_stream, datastore_name=None):
          print("Attempted to track file {0} at {1}".format(name, path_or_stream))
  
      def child_run(self, name=None, run_id=None, outputs=None):
          childname = str(uuid4()) if name is None else name
          newrid = "{}.{}".format(self.id, childname) if run_id is None else run_id
          return _OfflineRun(parent_logger=self._logger, run_id=newrid)
  
      def __enter__(self):
          print("[START] Run {} context".format(self.id))
          return self
  
      def __exit__(self, exit_type, value, traceback):
          print("[STOP] Run {} context".format(self.id))
  
      def _emit(self, type, name, value):
          print("Attempted to log {0} metric {1}:\n{2}".format(type, name, value))
  
      def flush(self):
          print("Attempted to flush run with id {0}".format(self.id))
  
  
  class InputDatasets(dict):
      """Defines a container for holding a materialized dataset in a run.
  
      .. remarks::
  
          An InputDatasets object is a dictionary containing the input Datasets in a run.
          The key is the input name of the Dataset in the control plane. The value will be the target path when mode is
          download or mount or the actual Dataset object if mode is direct.
  
      :param workspace: The workspace where the dataset is saved or registered.
      :type workspace: azureml.core.Workspace
      """
  
      def __init__(self, workspace):
          """Initialize the InputDatasets object.
  
          :param workspace: The workspace where the dataset is saved or registered.
          :type workspace: azureml.core.Workspace
          """
          super(InputDatasets, self).__init__()
          self._load_input_datasets(workspace)
  
      def _load_input_datasets(self, workspace):
          """Load all the dataset from environment variables."""
          from azureml.data._dataset import _Dataset
          from azureml.data.constants import DIRECT_MODE
  
          dataset_environment_vars = os.environ.get("AZUREML_DATASET_ENVIRONMENT_VARS")
          if dataset_environment_vars:
              dataset_environment_var_list = \
                  [item for item in dataset_environment_vars.split(',') if item and item != '']
              for pair in dataset_environment_var_list:
                  pair_list = [item for item in pair.split(':') if item and item != '']
                  if len(pair_list) == 2:
                      var_name = pair_list[0]
                      dataset_consumption_type = pair_list[1]
                      var_value = os.environ.get(var_name)
                      if var_value and dataset_consumption_type.lower() == DIRECT_MODE:
                          super(InputDatasets, self).__setitem__(var_name, _Dataset._get_by_id(workspace, var_value))
                      else:
                          super(InputDatasets, self).__setitem__(var_name, var_value)
  
  
  class OutputDatasets(dict):
      """Defines a container for holding a output path in a run.
  
      .. remarks:
          An OutputDatasets object is a dictionary containing the output name to a local path/uri in a run.
          The value type is dependent on the output mode that was specified in the job submission code.
      """
  
      def __init__(self, workspace):
          """Initialize the OutputDatasets object."""
          super(OutputDatasets, self).__init__()
          self.workspace = workspace
          self._initialize()
  
      def _initialize(self):
          output_env_vars = os.environ.get("AZUREML_DATASET_FILE_OUTPUTS")
          link_output_env_vars = os.environ.get("AZUREML_LINK_DATASET_OUTPUTS")
  
          if output_env_vars:
              for output_env in output_env_vars.split(";"):
                  output_env = output_env.rstrip('/\\')
                  self[output_env] = os.environ[output_env]
  
          if link_output_env_vars:
              for output_env in link_output_env_vars.split(";"):
                  output_env = output_env.rstrip('/\\')
                  self[output_env] = LinkOutput(self.workspace, output_env)
  
  
  class LinkOutput:
      """Defines a container for holding a output path in a run.
  
      ... remarks:
          An LinkOutput object is OutputData that will be linked with dataset in data plane..
      """
  
      def __init__(self, workspace, name=None):
          """Initialize the LinkOutput object."""
          self.workspace = workspace
          self.name = name
  
      @experimental
      def link(self, dataset):
          """Link output with dataset.
  
          :param dataset: dataset to link to output.
          :type dataset: azureml.data.abstract_dataset.AbstractDataset
          """
          def save_lineage(dataset):
              from azureml._restclient.models import OutputDatasetLineage, DatasetIdentifier, DatasetOutputType, \
                  DatasetOutputDetails, DatasetOutputMechanism
              from azureml.core import Run
  
              id = dataset._ensure_saved(self.workspace)
              registered_id = dataset._registration and dataset._registration.registered_id
              version = dataset.version
              dataset_id = DatasetIdentifier(id, registered_id, version)
              output_details = DatasetOutputDetails(
                  self.name,
                  DatasetOutputMechanism.link
              )
              output_lineage = OutputDatasetLineage(dataset_id, DatasetOutputType.run_output, output_details)
  
              try:
                  run = Run.get_context()
                  run._update_output_dataset_lineage([output_lineage])
              except Exception:
                  module_logger.error("Failed to update output dataset lineage")
  
          if not dataset:
              raise RuntimeError("Argument dataset cannot be None.")
  
          if not isinstance(dataset, AbstractDataset):
              raise RuntimeError("Argument dataset must an instance of AbstractDataset")
  
          save_lineage(dataset)
  