class_docstr: Wrapper that makes the algo "multi-task" by adding a task id to the inputs.
class_docstr_tok: ["Wrapper","that","makes","the","algo","``","multi-task","''","by","adding","a","task","id","to","the","inputs","."]
docstr: Observe some of the given trials to warm-start the HPO algorithm.
  
  These experiments and their trials were fetched from some storage, and
  are believed to be somewhat similar to the current on-going experiment.
  By default, this wrapper only considers the trials which are compatible
  with the current space.
  
  Parameters
  ----------
  warm_start_trials:
      Dictionary mapping from ExperimentConfig objects (containing the experiment config) to
      the list of Trials associated with that experiment.
docstr_tok: ["Observe","some","of","the","given","trials","to","warm-start","the","HPO","algorithm",".","These","experiments","and","their","trials","were","fetched","from","some","storage",",","and","are","believed","to","be","somewhat","similar","to","the","current","on-going","experiment",".","By","default",",","this","wrapper","only","considers","the","trials","which","are","compatible","with","the","current","space",".","Parameters","--","--","--","--","--","warm_start_trials",":","Dictionary","mapping","from","ExperimentConfig","objects","(","containing","the","experiment","config",")","to","the","list","of","Trials","associated","with","that","experiment","."]
code: def warm_start(
      self,
      warm_start_trials: list[tuple[ExperimentConfig, list[Trial]]],
  ) -> None:
      """Observe some of the given trials to warm-start the HPO algorithm.
  
      These experiments and their trials were fetched from some storage, and
      are believed to be somewhat similar to the current on-going experiment.
      By default, this wrapper only considers the trials which are compatible
      with the current space.
  
      Parameters
      ----------
      warm_start_trials:
          Dictionary mapping from ExperimentConfig objects (containing the experiment config) to
          the list of Trials associated with that experiment.
      """
      logger.info(
          "Warm-starting using only multi-task training with compatible trials, since the algo "
          "isn't warm-starteable."
      )
  
      # Dict mapping from task ID to list of compatible trials.
      compatible_trials: dict[int, list[Trial]] = defaultdict(list)
      task_ids_to_experiments: dict[int, ExperimentConfig] = {}
      for i, (experiment_info, trials) in enumerate(warm_start_trials):
          # Start the task ids at 1, so the current experiment has task id 0.
          task_id = i + 1
          for trial in trials:
              # Drop the point if it doesn't fit inside the current space.
              # IDEA: Possibly add some 'translation' logic in an eventual Knowledge Base
              # implementation.
              if self._is_compatible(trial):
                  compatible_trials[task_id].append(trial)
                  task_ids_to_experiments[task_id] = experiment_info
  
          n_compatible_trials = len(compatible_trials[task_id])
          logger.info(
              f"Experiment {experiment_info} has {len(trials)} trials in "
              f"total, out of which {n_compatible_trials} were found to be "
              f"compatible with the target experiment."
          )
  
      if not compatible_trials:
          logger.info("No compatible trials found!")
          return
  
      # Note: If there are no compatible trials found, this would allow multiple attempts at
      # warm-starting. This isn't used though.
      if self._total_warm_start_trials:
          raise RuntimeError(
              "The algorithm can only be warm-started once!"
          )
      self._total_warm_start_trials = sum(
          map(len, compatible_trials.values())
      )
  
      logger.info(
          "Algo will observe a total of %s trials from other experiments.",
          self._total_warm_start_trials,
      )
  
      for task_id, trials in compatible_trials.items():
          logger.debug(
              "Observing %s new trials from task %s",
              len(trials),
              task_id,
          )
          assert task_id != TARGET_TASK_ID
          with self.in_task(task_id):
              # NOTE: self.observe saves those trials in our registry, and adds the task ids
              # to the trials and passes that calls `self.transform`.
              # This also reuses the collision handling logic from the Transform wrapper.
              self.observe(trials)
  
      if self._max_trials is not None:
          wrapped_algo_max_trials = (
              self._max_trials + self._total_warm_start_trials
          )
          logger.debug(
              "Setting the max_trials of the wrapped algo to %s rather than %s, to account for "
              "the trials observed during warm-starting.",
              wrapped_algo_max_trials,
              self._max_trials,
          )
          self.algorithm.max_trials = wrapped_algo_max_trials
  
imports: from __future__ import annotations
  import copy
  from collections import defaultdict
  from contextlib import contextmanager
  from logging import getLogger
  from orion.algo.space import Categorical, Space
  from orion.core.utils.format_trials import dict_to_trial
  from orion.core.worker.algo_wrappers.algo_wrapper import AlgoT
  from orion.core.worker.algo_wrappers.transform_wrapper import TransformWrapper, _copy_status_and_results
  from orion.core.worker.experiment_config import ExperimentConfig
  from orion.core.worker.trial import Trial
  from orion.core.worker.warm_start.knowledge_base import KnowledgeBase
  from orion.core.worker.warm_start.warm_starteable import WarmStarteable
label_desc: Use image node_id 6 for calling the MultiTaskWrapper obj's underlying member method code with example usage: obj.warm_start(warm_start_trials) without return types