code: def init(
      *,
      distributed: Optional[core.DistributedContext] = None,
      checkpoint_storage: Optional[Union[str, Dict[str, Any]]] = None,
      preempt_mode: core.PreemptMode = core.PreemptMode.WorkersAskChief,
      tensorboard_mode: core.TensorboardMode = core.TensorboardMode.AUTO,
  ) -> Context:
      """
      ``core.init()`` builds a :class:`core.Context <determined.core.Context>` for use with the Core
      API.
  
      Always use ``with core.init() as context`` instead of instantiating a ``core.Context`` directly.
      Certain components of the Core API may be configured by passing arguments to ``core.init()``.
      The only arg that is required is a ``DistributedContext``, and even that is only required for
      multi-slot tasks.
  
      All of your training must occur within the scope of the ``with core.init() as core_context``, as
      there are resources necessary for training which start in the ``core.Context``'s ``__enter__``
      method and must be cleaned up in its ``__exit__()`` method.
  
      Arguments:
          distributed (``core.DistributedContext``, optional): Passing a ``DistributedContext`` is
              required for multi-slot training, but unnecessary for single-slot training.  Defaults to
              ``None``.
          preempt_mode (``core.PreemptMode``, optional): Configure the calling pattern for the
              ``core_context.preempt.should_preempt()`` method.  See
              :class:`~determined.core.PreemptMode` for more detail.  Defaults to ``WorkersAskChief``.
          checkpoint_storage (``Union[str, dict]``, optional): A directory path or a cloud storage URI
              of the form ``s3://<bucket>[/<prefix>]`` (AWS) or ``gs://<bucket>[/<prefix>]`` (GCP).
              This should only be used when IAM permissions can be assumed. You may also pass a
              dictionary matching the ``checkpoint_storage`` field of the experiment config, with the
              exception that ``type: shared_fs`` configs are not allowed.
          tensorboard_mode (``core.TensorboardMode``, optional): Define how Tensorboard
              metrics and profiling data are retained. See
              :class:`~determined.core.TensorboardMode`` for more detail. Defaults to ``AUTO``.
      """
      info = det.get_cluster_info()
      if info is None:
          return _dummy_init(
              distributed=distributed,
              checkpoint_storage=checkpoint_storage,
          )
  
      # We are on the cluster.
      cert = certs.default_load(info.master_url)
      session = api.Session(
          info.master_url,
          None,
          None,
          cert,
          max_retries=util.get_max_retries_config(),
      )
  
      if distributed is None:
          if len(info.container_addrs) > 1 or len(info.slot_ids) > 1:
              raise ValueError(
                  "you must provide a valid DistributedContext for a multi-slot task"
              )
  
      distributed = distributed or core.DummyDistributedContext()
  
      # At present, we only support tensorboards in Trial tasks.
      tbd_writer = None
  
      train = None
      searcher = None
      tensorboard_manager = None
      experimental = None
  
      storage_manager = _get_storage_manager(checkpoint_storage)
  
      if info.task_type == "TRIAL":
          # Prepare the tensorboard hooks.
          tensorboard_manager = tensorboard.build(
              info.cluster_id,
              str(info.trial.experiment_id),
              str(info.trial.trial_id),
              info.trial._config["checkpoint_storage"],
              container_path=constants.SHARED_FS_CONTAINER_PATH,
              async_upload=True,
              sync_on_close=(
                  tensorboard_mode == core.TensorboardMode.AUTO
              ),
          )
          if tensorboard_mode == core.TensorboardMode.AUTO:
              tbd_writer = tensorboard.get_metric_writer()
  
          train = core.TrainContext(
              session,
              info.trial.trial_id,
              info.trial._trial_run_id,
              info.trial.experiment_id,
              distributed,
              tensorboard_mode,
              tensorboard_manager,
              tbd_writer,
          )
          units = core._parse_searcher_units(info.trial._config)
          searcher = core.SearcherContext(
              session,
              distributed,
              info.trial.trial_id,
              info.trial._trial_run_id,
              info.allocation_id,
              units,
          )
  
          if storage_manager is None:
              storage_manager = storage.build(
                  info.trial._config["checkpoint_storage"],
                  container_path=constants.SHARED_FS_CONTAINER_PATH,
              )
  
          checkpoint = core.CheckpointContext(
              distributed,
              storage_manager,
              session,
              info.task_id,
              info.allocation_id,
              tensorboard_mode,
              tensorboard_manager,
          )
  
          preempt = core.PreemptContext(
              session, info.allocation_id, distributed, preempt_mode
          )
          experimental = core.ExperimentalCoreContext(
              session, info.trial.trial_id
          )
  
      else:
          # TODO: support checkpointing for non-trial tasks.
          if storage_manager is None:
              base_path = appdirs.user_data_dir("determined")
              logger.info(
                  f"no storage_manager provided; storing checkpoints in {base_path}"
              )
              storage_manager = storage.SharedFSStorageManager(
                  base_path
              )
          checkpoint = core.DummyCheckpointContext(
              distributed, storage_manager
          )
          preempt = core.DummyPreemptContext(distributed, preempt_mode)
  
      _install_stacktrace_on_sigusr1()
  
      return Context(
          distributed=distributed,
          checkpoint=checkpoint,
          preempt=preempt,
          train=train,
          searcher=searcher,
          experimental=experimental,
          _tensorboard_manager=tensorboard_manager,
      )
  
imports: import logging
  import pathlib
  import signal
  import sys
  import threading
  import traceback
  import types
  from typing import Any, Dict, Optional, Union
  import appdirs
  import determined
  from determined import core, tensorboard
  from determined.common import api, constants, storage, util
  from determined.common.api import certs
label_desc: Use image node_id 4 for calling a global function with example usage: init() and returns: Context, _dummy_init