docstr: A short function to fit and create an ensemble. It is just a wrapper to easily
  send a request to dask to create an ensemble and clean the memory when finished
  
  Parameters
  ----------
  iteration: int
      The current iteration
  
  end_at: float
      At what time the job must finish. Needs to be the endtime and not the
      time left because we do not know when dask schedules the job.
  
  backend: Backend
      Backend to write and read files
  
  dataset_name: str
      name of dataset
  
  task_type: int
      type of ML task
  
  metrics: Sequence[Scorer]
      Metrics to optimize the ensemble for.
  
  pynisher_context: "fork" | "spawn" | "forkserver" = "fork"
      Context to use for multiprocessing, can be either fork, spawn or forkserver.
  
  ensemble_class : Type[AbstractEnsemble] (default=EnsembleSelection)
      Class implementing the post-hoc ensemble algorithm. Set to
      ``None`` to disable ensemble building or use ``SingleBest``
      to obtain only use the single best model instead of an
      ensemble.
  
  ensemble_kwargs : Dict, optional
      Keyword arguments that are passed to the ensemble class upon
      initialization.
  
  ensemble_nbest: int | float = 50
      If int: consider only the n best prediction
      If float: consider only this fraction of the best models
  
  max_models_on_disc: int | float | None = 100
     Defines the maximum number of models that are kept in the disc.
  
     If int, it must be greater or equal than 1, and dictates the max number of
     models to keep.
  
     If float, it will be interpreted as the max megabytes allowed of disc space.
     That is, if the number of ensemble candidates require more disc space than
     this float value, the worst models will be deleted to keep within this
     budget. Models and predictions of the worst-performing models will be
     deleted then.
  
     If None, the feature is disabled.
  
  seed: int = 1
      Seed used for training the models in the backend
  
  precision: 16 | 32 | 64 | 128 = 32
      Precision of floats to read the predictions
  
  memory_limit: int | None = None
      Memory limit in mb. If ``None``, no memory limit is enforced.
  
  read_at_most: int | None = None
      read at most n new prediction files in each iteration. If `None`, will read
      the predictions and calculate losses for all runs that require it.
  
  logger_port: int = DEFAULT_TCP_LOGGING_PORT
      The port where the logging server is listening to.
  
  random_state: int | RandomState | None = None
      A random state used for the ensemble selection process.
  
  Returns
  -------
  (ensemble_history: list[dict[str, Any]], nbest: int | float)
      The ensemble history and the nbest chosen members
docstr_tok: ["A","short","function","to","fit","and","create","an","ensemble",".","It","is","just","a","wrapper","to","easily","send","a","request","to","dask","to","create","an","ensemble","and","clean","the","memory","when","finished","Parameters","--","--","--","--","--","iteration",":","int","The","current","iteration","end_at",":","float","At","what","time","the","job","must","finish",".","Needs","to","be","the","endtime","and","not","the","time","left","because","we","do","not","know","when","dask","schedules","the","job",".","backend",":","Backend","Backend","to","write","and","read","files","dataset_name",":","str","name","of","dataset","task_type",":","int","type","of","ML","task","metrics",":","Sequence","[","Scorer","]","Metrics","to","optimize","the","ensemble","for",".","pynisher_context",":","``","fork","''","|","``","spawn","''","|","``","forkserver","''","=","``","fork","''","Context","to","use","for","multiprocessing",",","can","be","either","fork",",","spawn","or","forkserver",".","ensemble_class",":","Type","[","AbstractEnsemble","]","(","default=EnsembleSelection",")","Class","implementing","the","post-hoc","ensemble","algorithm",".","Set","to","``","None","``","to","disable","ensemble","building","or","use","``","SingleBest","``","to","obtain","only","use","the","single","best","model","instead","of","an","ensemble",".","ensemble_kwargs",":","Dict",",","optional","Keyword","arguments","that","are","passed","to","the","ensemble","class","upon","initialization",".","ensemble_nbest",":","int","|","float","=","50","If","int",":","consider","only","the","n","best","prediction","If","float",":","consider","only","this","fraction","of","the","best","models","max_models_on_disc",":","int","|","float","|","None","=","100","Defines","the","maximum","number","of","models","that","are","kept","in","the","disc",".","If","int",",","it","must","be","greater","or","equal","than","1",",","and","dictates","the","max","number","of","models","to","keep",".","If","float",",","it","will","be","interpreted","as","the","max","megabytes","allowed","of","disc","space",".","That","is",",","if","the","number","of","ensemble","candidates","require","more","disc","space","than","this","float","value",",","the","worst","models","will","be","deleted","to","keep","within","this","budget",".","Models","and","predictions","of","the","worst-performing","models","will","be","deleted","then",".","If","None",",","the","feature","is","disabled",".","seed",":","int","=","1","Seed","used","for","training","the","models","in","the","backend","precision",":","16","|","32","|","64","|","128","=","32","Precision","of","floats","to","read","the","predictions","memory_limit",":","int","|","None","=","None","Memory","limit","in","mb",".","If","``","None","``",",","no","memory","limit","is","enforced",".","read_at_most",":","int","|","None","=","None","read","at","most","n","new","prediction","files","in","each","iteration",".","If","`","None","`",",","will","read","the","predictions","and","calculate","losses","for","all","runs","that","require","it",".","logger_port",":","int","=","DEFAULT_TCP_LOGGING_PORT","The","port","where","the","logging","server","is","listening","to",".","random_state",":","int","|","RandomState","|","None","=","None","A","random","state","used","for","the","ensemble","selection","process",".","Returns","--","--","--","-","(","ensemble_history",":","list","[","dict","[","str",",","Any","]","]",",","nbest",":","int","|","float",")","The","ensemble","history","and","the","nbest","chosen","members"]
code: def fit_and_return_ensemble(
      iteration: int,
      end_at: float,
      backend: Backend,
      dataset_name: str,
      task_type: int,
      metrics: Sequence[Scorer],
      pynisher_context: str,
      ensemble_class: Type[AbstractEnsemble] = EnsembleSelection,
      ensemble_kwargs: Dict[str, Any] | None = None,
      ensemble_nbest: int | float = 50,
      max_models_on_disc: int | float | None = None,
      seed: int = 1,
      precision: int = 32,
      memory_limit: int | None = None,
      read_at_most: int | None = None,
      logger_port: int = logging.handlers.DEFAULT_TCP_LOGGING_PORT,
      random_state: int | np.random.RandomState | None = None,
  ) -> tuple[list[dict[str, Any]], int | float]:
      """
      A short function to fit and create an ensemble. It is just a wrapper to easily
      send a request to dask to create an ensemble and clean the memory when finished
  
      Parameters
      ----------
      iteration: int
          The current iteration
  
      end_at: float
          At what time the job must finish. Needs to be the endtime and not the
          time left because we do not know when dask schedules the job.
  
      backend: Backend
          Backend to write and read files
  
      dataset_name: str
          name of dataset
  
      task_type: int
          type of ML task
  
      metrics: Sequence[Scorer]
          Metrics to optimize the ensemble for.
  
      pynisher_context: "fork" | "spawn" | "forkserver" = "fork"
          Context to use for multiprocessing, can be either fork, spawn or forkserver.
  
      ensemble_class : Type[AbstractEnsemble] (default=EnsembleSelection)
          Class implementing the post-hoc ensemble algorithm. Set to
          ``None`` to disable ensemble building or use ``SingleBest``
          to obtain only use the single best model instead of an
          ensemble.
  
      ensemble_kwargs : Dict, optional
          Keyword arguments that are passed to the ensemble class upon
          initialization.
  
      ensemble_nbest: int | float = 50
          If int: consider only the n best prediction
          If float: consider only this fraction of the best models
  
      max_models_on_disc: int | float | None = 100
         Defines the maximum number of models that are kept in the disc.
  
         If int, it must be greater or equal than 1, and dictates the max number of
         models to keep.
  
         If float, it will be interpreted as the max megabytes allowed of disc space.
         That is, if the number of ensemble candidates require more disc space than
         this float value, the worst models will be deleted to keep within this
         budget. Models and predictions of the worst-performing models will be
         deleted then.
  
         If None, the feature is disabled.
  
      seed: int = 1
          Seed used for training the models in the backend
  
      precision: 16 | 32 | 64 | 128 = 32
          Precision of floats to read the predictions
  
      memory_limit: int | None = None
          Memory limit in mb. If ``None``, no memory limit is enforced.
  
      read_at_most: int | None = None
          read at most n new prediction files in each iteration. If `None`, will read
          the predictions and calculate losses for all runs that require it.
  
      logger_port: int = DEFAULT_TCP_LOGGING_PORT
          The port where the logging server is listening to.
  
      random_state: int | RandomState | None = None
          A random state used for the ensemble selection process.
  
      Returns
      -------
      (ensemble_history: list[dict[str, Any]], nbest: int | float)
          The ensemble history and the nbest chosen members
      """
      random_state = check_random_state(random_state)
      result = EnsembleBuilder(
          backend=backend,
          dataset_name=dataset_name,
          task_type=task_type,
          metrics=metrics,
          ensemble_class=ensemble_class,
          ensemble_kwargs=ensemble_kwargs,
          ensemble_nbest=ensemble_nbest,
          max_models_on_disc=max_models_on_disc,
          seed=seed,
          precision=precision,
          memory_limit=memory_limit,
          read_at_most=read_at_most,
          random_state=random_state.randint(10000000),
          logger_port=logger_port,
      ).run(
          end_at=end_at,
          iteration=iteration,
          pynisher_context=pynisher_context,
      )
      return result
  
imports: from __future__ import annotations
  from typing import Any, Dict, Sequence, Type
  import logging.handlers
  import time
  import traceback
  import dask.distributed
  import numpy
  from sklearn.utils.validation import check_random_state
  from smac.callbacks import IncorporateRunResultCallback
  from smac.optimizer.smbo import SMBO
  from smac.runhistory.runhistory import RunInfo, RunValue
  from smac.tae.base import StatusType
  from autosklearn.automl_common.common.utils.backend import Backend
  from autosklearn.ensemble_building.builder import EnsembleBuilder
  from autosklearn.ensembles.abstract_ensemble import AbstractEnsemble
  from autosklearn.ensembles.ensemble_selection import EnsembleSelection
  from autosklearn.metrics import Scorer
  from autosklearn.util.logging_ import get_named_client_logger
label_desc: Use image node_id 4 for calling the EnsembleBuilderManager obj's underlying member method code with example usage: obj.fit_and_return_ensemble(iteration, end_at, backend, dataset_name, task_type, metrics, pynisher_context, ensemble_class, ensemble_kwargs, ensemble_nbest, max_models_on_disc, seed, precision, memory_limit, read_at_most, logger_port, random_state) and returns: result