class_docstr: Builds ensembles out of runs that exist in the Backend
  
  This is used by EnsembleBuilderManager and created in a dask-client
  every time a run finishes and there is currently no EnsembleBuilder active.
docstr: Run the main loop of ensemble building
  
  The process is:
  * Load all available runs + previous candidates (if any)
  * Update the loss of those that require
  * From these runs, get a list of candidates
  * Save candidates
  * Delete models that are not candidates
  * Build an ensemble from the candidates if there are new candidates
  
  Parameters
  ----------
  time_left : float | None = None
      How much time is left for this run
  
  iteration : int = 0
      The iteration of this run
  
  Returns
  -------
  (ensemble_history: list[dict[str, Any]], nbest: int | float)
docstr_tok: ["Run","the","main","loop","of","ensemble","building","The","process","is",":","*","Load","all","available","runs","+","previous","candidates","(","if","any",")","*","Update","the","loss","of","those","that","require","*","From","these","runs",",","get","a","list","of","candidates","*","Save","candidates","*","Delete","models","that","are","not","candidates","*","Build","an","ensemble","from","the","candidates","if","there","are","new","candidates","Parameters","--","--","--","--","--","time_left",":","float","|","None","=","None","How","much","time","is","left","for","this","run","iteration",":","int","=","0","The","iteration","of","this","run","Returns","--","--","--","-","(","ensemble_history",":","list","[","dict","[","str",",","Any","]","]",",","nbest",":","int","|","float",")"]
code: def main(
      self,
      time_left: float | None = None,
      iteration: int = 0,
  ) -> tuple[list[dict[str, Any]], int | float]:
      """Run the main loop of ensemble building
  
      The process is:
      * Load all available runs + previous candidates (if any)
      * Update the loss of those that require
      * From these runs, get a list of candidates
      * Save candidates
      * Delete models that are not candidates
      * Build an ensemble from the candidates if there are new candidates
  
      Parameters
      ----------
      time_left : float | None = None
          How much time is left for this run
  
      iteration : int = 0
          The iteration of this run
  
      Returns
      -------
      (ensemble_history: list[dict[str, Any]], nbest: int | float)
      """
      # Pynisher jobs inside dask 'forget' the logger configuration.
      # So we have to set it up accordingly
      self.logger = get_named_client_logger(
          name="EnsembleBuilder",
          port=self.logger_port,
      )
  
      if time_left is not None:
          self.start_time = time.time()
          used_time = time.time() - self.start_time
          left_for_iter = time_left - used_time
          itr = iteration if str(iteration) is not None else ""
          self.logger.debug(
              f"Starting iteration {itr}, time left: {left_for_iter}"
          )
  
      # Can't load data, exit early
      if not os.path.exists(
          self.backend._get_targets_ensemble_filename()
      ):
          self.logger.debug(
              f"No targets for ensemble: {traceback.format_exc()}"
          )
          raise RuntimeError("No targets for ensemble")
  
      # We will delete runs once we are complete
      deletable_runs: set[Run] = set()
  
      # Load in information from previous candidates and also runs
      available_runs = self.available_runs()
  
      # Update runs with information of available previous candidates
      previous_candidates = self.previous_candidates()
      available_runs.update(previous_candidates)
  
      # We just need the values now, not the key value pairs {run.id: Run}
      runs = list(available_runs.values())
  
      if len(runs) == 0:
          self.logger.debug("Found no runs")
          raise RuntimeError("Found no runs")
  
      # We load in `X_data` if we need it
      if any(m._needs_X for m in self.metrics):
          ensemble_X_data = self.X_data("ensemble")
  
          if ensemble_X_data is None:
              msg = "No `X_data` for 'ensemble' which was required by metrics"
              self.logger.debug(msg)
              raise RuntimeError(msg)
      else:
          ensemble_X_data = None
  
      # Calculate the loss for those that require it
      requires_update = self.requires_loss_update(runs)
      if self.read_at_most is not None:
          requires_update = requires_update[: self.read_at_most]
  
      for run in requires_update:
          run.record_modified_times()  # So we don't count as modified next time
          run.losses = {
              metric.name: self.loss(
                  run, metric=metric, X_data=ensemble_X_data
              )
              for metric in self.metrics
          }
  
      # Get the dummy and real runs
      dummies, candidates = split(runs, by=lambda r: r.is_dummy())
  
      # We see if we need to delete any of the real runs before we waste compute
      # on evaluating their candidacy for ensemble building
      if any(candidates):
          candidates, to_delete = self.requires_deletion(
              candidates,
              max_models=self.max_models_on_disc,
              memory_limit=self.model_memory_limit,
          )
  
          # If there are no candidates left, we just keep the best one, ordered by
          # objectives and finally num_run
          if not any(candidates):
              best = min(
                  to_delete,
                  key=lambda r: (
                      *[
                          r.losses.get(m.name, np.inf)
                          for m in self.metrics
                      ],
                      r.num_run,
                  ),
              )
              candidates = [best]
              to_delete.remove(best)
  
          if any(to_delete):
              self.logger.info(
                  f"Deleting runs {to_delete} due to"
                  f" max_models={self.max_models_on_disc} and/or"
                  f" memory_limit={self.model_memory_limit}"
              )
              deletable_runs.update(to_delete)
  
      # If there are any candidates, perform candidates selection
      if any(candidates):
          candidates, to_delete = self.candidate_selection(
              runs=candidates,
              dummies=dummies,
              better_than_dummy=True,
              nbest=self.ensemble_nbest,
          )
          if any(to_delete):
              self.logger.info(
                  f"Deleting runs {to_delete} due to"
                  f" nbest={self.ensemble_nbest} or worse than dummies"
              )
              deletable_runs.update(to_delete)
      else:
          candidates = dummies
          self.logger.warning(
              "No runs were available to build an ensemble from"
          )
  
      # In case we record test predictions and not every model has test predictions,
      # only use the subset of models that has predictions for both the test set and
      # the ensemble optimization set.
      candidates_set = set(candidates)
      test_subset = {
          r for r in candidates if r.pred_path("test").exists()
      }
  
      if len(test_subset) > 0:
          candidates = sorted(test_subset, key=lambda r: r.id)
          test_models = candidates
  
          to_delete = candidates_set - test_subset
          if any(to_delete):
              self.logger.info(
                  f"Deleting runs {to_delete} due to runs not"
                  ' having "test_predictions" while others do not:'
                  f"\nHave test_predictions = {test_subset}"
                  f"\nNo test_predictions = {to_delete}"
              )
              deletable_runs.update(to_delete)
  
      else:
          candidates = sorted(candidates_set, key=lambda r: r.id)
          test_models = []
  
      # Save the candidates for the next round
      with self.previous_candidates_path.open("wb") as f:
          pickle.dump({run.id: run for run in candidates}, f)
  
      # If there was any change from the previous run, either in terms of
      # runs or one of those runs had its loss updated, then we need to
      # fit the ensemble builder
      previous_candidate_ids = set(previous_candidates)
      current_candidate_ids = set(run.id for run in candidates)
      difference = previous_candidate_ids ^ current_candidate_ids
  
      was_updated_candidates = list(
          run in candidates for run in requires_update
      )
  
      if not any(difference) and not any(was_updated_candidates):
          self.logger.info(
              "All ensemble candidates the same, no update required"
          )
          return self.ensemble_history, self.ensemble_nbest
  
      targets = cast(
          np.ndarray, self.targets("ensemble")
      )  # Sure they exist
  
      ensemble = self.fit_ensemble(
          candidates=candidates,
          targets=targets,
          runs=runs,
          ensemble_class=self.ensemble_class,
          ensemble_kwargs=self.ensemble_kwargs,
          X_data=ensemble_X_data,
          task=self.task_type,
          metrics=self.metrics,
          precision=self.precision,
          random_state=self.random_state,
      )
  
      self.logger.info(str(ensemble))
      ens_perf = ensemble.get_validation_performance()
      self.validation_performance_ = min(
          self.validation_performance_, ens_perf
      )
      self.backend.save_ensemble(
          ensemble=ensemble, idx=iteration, seed=self.seed  # type: ignore
      )
  
      performance_stamp = {"Timestamp": pd.Timestamp.now()}
  
      for kind, score_name, models in [
          ("ensemble", "optimization", candidates),
          ("test", "test", test_models),
      ]:
          if len(models) == 0:
              continue
  
          pred_targets = self.targets(kind)
          if pred_targets is None:
              self.logger.warning(f"No ensemble targets for {kind}")
              continue
  
          run_preds = [
              r.predictions(kind, precision=self.precision)
              for r in models
          ]
          pred = ensemble.predict(run_preds)
  
          if any(m._needs_X for m in self.metrics):
              X_data = self.X_data(kind)
              if X_data is None:
                  msg = f"No `X` data for '{kind}' which was required by metrics"
                  self.logger.debug(msg)
                  raise RuntimeError(msg)
          else:
              X_data = None
  
          scores = calculate_scores(
              solution=pred_targets,
              prediction=pred,
              task_type=self.task_type,
              metrics=self.metrics,
              X_data=X_data,
              scoring_functions=None,
          )
  
          # TODO only one metric in history
          #
          #   We should probably return for all metrics but this makes
          #   automl::performance_history a lot more complicated, will
          #   tackle in a future PR
          first_metric = self.metrics[0]
          performance_stamp[f"ensemble_{score_name}_score"] = scores[
              first_metric.name
          ]
  
      # Add the performance stamp to the history
      self.ensemble_history.append(performance_stamp)
  
      # Lastly, delete any runs that need to be deleted. We save this as the last step
      # so that we have an ensemble saved that is up to date. If we do not do so,
      # there could be runs deleted that are in th previous ensemble and we do not
      # manage to update the ensemble due to a crash or the process being killed
      # before it could be updated
      self.delete_runs(deletable_runs)
  
      return self.ensemble_history, self.ensemble_nbest
  
imports: from __future__ import annotations
  from typing import Any, Iterable, Mapping, Sequence, Type, cast
  import logging.handlers
  import multiprocessing
  import numbers
  import os
  import pickle
  import time
  import traceback
  from itertools import accumulate
  from pathlib import Path
  import numpy
  import pandas
  import pynisher
  from sklearn.utils.validation import check_random_state
  from autosklearn.automl_common.common.utils.backend import Backend
  from autosklearn.data.validation import SUPPORTED_FEAT_TYPES
  from autosklearn.data.xy_data_manager import XYDataManager
  from autosklearn.ensemble_building.run import Run, RunID
  from autosklearn.ensembles.abstract_ensemble import AbstractEnsemble
  from autosklearn.ensembles.ensemble_selection import EnsembleSelection
  from autosklearn.metrics import Scorer, calculate_loss, calculate_scores
  from autosklearn.util.disk import rmtree
  from autosklearn.util.functional import cut, findwhere, roundrobin, split
  from autosklearn.util.logging_ import get_named_client_logger
  from autosklearn.util.parallel import preload_modules
label_desc: Use image node_id 8 for calling the EnsembleBuilder obj's underlying member method code with example usage: obj.main(time_left, iteration) and returns: self,  self, self,  self