docstr: Internal version of ``create_model`` with private arguments.
docstr_tok: ["Internal","version","of","``","create_model","``","with","private","arguments","."]
code: def _create_model(
      self,
      estimator,
      num_clusters: int = 4,
      fraction: float = 0.05,
      ground_truth: Optional[str] = None,
      round: int = 4,
      fit_kwargs: Optional[dict] = None,
      experiment_custom_tags: Optional[Dict[str, Any]] = None,
      verbose: bool = True,
      system: bool = True,
      add_to_model_list: bool = True,
      raise_num_clusters: bool = False,
      X_data: Optional[pd.DataFrame] = None,  # added in pycaret==2.2.0
      display: Optional[
          CommonDisplay
      ] = None,  # added in pycaret==2.2.0
      **kwargs,
  ) -> Any:
      """
      Internal version of ``create_model`` with private arguments.
      """
      function_params_str = ", ".join(
          [
              f"{k}={v}"
              for k, v in locals().items()
              if k not in ("X_data")
          ]
      )
  
      self.logger.info("Initializing create_model()")
      self.logger.info(f"create_model({function_params_str})")
  
      self.logger.info("Checking exceptions")
  
      # run_time
      runtime_start = time.time()
  
      available_estimators = set(self._all_models_internal.keys())
  
      if not fit_kwargs:
          fit_kwargs = {}
  
      # only raise exception of estimator is of type string.
      if isinstance(estimator, str):
          if estimator not in available_estimators:
              raise ValueError(
                  f"Estimator {estimator} not available. Please see docstring for list of available estimators."
              )
      elif not hasattr(estimator, "fit"):
          raise ValueError(
              f"Estimator {estimator} does not have the required fit() method."
          )
  
      # checking round parameter
      if type(round) is not int:
          raise TypeError("Round parameter only accepts integer value.")
  
      # checking verbose parameter
      if type(verbose) is not bool:
          raise TypeError(
              "Verbose parameter can only take argument as True or False."
          )
  
      # checking system parameter
      if type(system) is not bool:
          raise TypeError(
              "System parameter can only take argument as True or False."
          )
  
      # checking fraction type:
      if fraction <= 0 or fraction >= 1:
          raise TypeError(
              "Fraction parameter can only take value as float between 0 to 1."
          )
  
      # checking num_clusters type:
      if num_clusters <= 1:
          raise TypeError(
              "num_clusters parameter can only take value integer value greater than 1."
          )
  
      # check ground truth exist in data_
      if ground_truth is not None:
          if ground_truth not in self.dataset.columns:
              raise ValueError(
                  f"ground_truth {ground_truth} doesn't exist in the dataset."
              )
  
      """
  
          ERROR HANDLING ENDS HERE
  
          """
  
      if not display:
          progress_args = {"max": 3}
          timestampStr = datetime.datetime.now().strftime("%H:%M:%S")
          monitor_rows = [
              [
                  "Initiated",
                  ". . . . . . . . . . . . . . . . . .",
                  timestampStr,
              ],
              [
                  "Status",
                  ". . . . . . . . . . . . . . . . . .",
                  "Loading Dependencies",
              ],
              [
                  "Estimator",
                  ". . . . . . . . . . . . . . . . . .",
                  "Compiling Library",
              ],
          ]
          display = CommonDisplay(
              verbose=verbose,
              html_param=self.html_param,
              progress_args=progress_args,
              monitor_rows=monitor_rows,
          )
  
      np.random.seed(self.seed)
  
      # Storing X_train and y_train in data_X and data_y parameter
      data_X = self.X if X_data is None else X_data
      transformed_data = (
          self.X_transformed
          if X_data is None
          else self.pipeline.transform(X_data)
      )
  
      """
          MONITOR UPDATE STARTS
          """
      display.update_monitor(1, "Selecting Estimator")
      """
          MONITOR UPDATE ENDS
          """
  
      self.logger.info("Importing untrained model")
  
      is_cblof = False
  
      if (
          isinstance(estimator, str)
          and estimator in available_estimators
      ):
          is_cblof = estimator == "cluster"
          model_definition = self._all_models_internal[estimator]
          model_args = model_definition.args
          model_args = {**model_args, **kwargs}
          model = model_definition.class_def(**model_args)
          full_name = model_definition.name
      else:
          self.logger.info("Declaring custom model")
  
          model = clone(estimator)
          model.set_params(**kwargs)
  
          full_name = self._get_model_name(model)
  
      display.update_monitor(2, full_name)
  
      if self._ml_usecase == MLUsecase.CLUSTERING:
          if raise_num_clusters:
              model.set_params(n_clusters=num_clusters)
          else:
              try:
                  model.set_params(n_clusters=num_clusters)
              except Exception:
                  pass
      else:
          model.set_params(contamination=fraction)
  
      # workaround for an issue with set_params in cuML
      try:
          model = clone(model)
      except Exception:
          self.logger.warning(
              f"create_model() for {model} raised an exception when cloning:"
          )
          self.logger.warning(traceback.format_exc())
  
      self.logger.info(f"{full_name} Imported successfully")
  
      display.move_progress()
  
      """
          MONITOR UPDATE STARTS
          """
  
      if self._ml_usecase == MLUsecase.CLUSTERING:
          display.update_monitor(1, f"Fitting {num_clusters} Clusters")
      else:
          display.update_monitor(1, f"Fitting {fraction} Fraction")
  
      """
          MONITOR UPDATE ENDS
          """
  
      with estimator_pipeline(
          self.pipeline, model
      ) as pipeline_with_model:
          fit_kwargs = get_pipeline_fit_kwargs(
              pipeline_with_model, fit_kwargs
          )
  
          self.logger.info("Fitting Model")
          model_fit_start = time.time()
          with redirect_output(self.logger):
              if is_cblof and "n_clusters" not in kwargs:
                  try:
                      pipeline_with_model.fit(data_X, **fit_kwargs)
                  except Exception:
                      try:
                          pipeline_with_model.set_params(
                              actual_estimator__n_clusters=12
                          )
                          model_fit_start = time.time()
                          pipeline_with_model.fit(data_X, **fit_kwargs)
                      except Exception as e:
                          raise RuntimeError(
                              "Could not form valid cluster separation. Try a different dataset or model."
                          ) from e
              else:
                  pipeline_with_model.fit(data_X, **fit_kwargs)
          model_fit_end = time.time()
  
          model_fit_time = np.array(
              model_fit_end - model_fit_start
          ).round(2)
  
      display.move_progress()
  
      if ground_truth is not None:
          self.logger.info(
              f"ground_truth parameter set to {ground_truth}"
          )
  
          gt = np.array(self.dataset[ground_truth])
      else:
          gt = None
  
      if self._ml_usecase == MLUsecase.CLUSTERING:
          with redirect_output(self.logger):
              metrics = self._calculate_metrics(
                  transformed_data, model.labels_, ground_truth=gt
              )
      else:
          metrics = {}
  
      self.logger.info(str(model))
      self.logger.info(
          "create_models() successfully completed......................................"
      )
  
      runtime = time.time() - runtime_start
  
      # mlflow logging
      if self.logging_param and system:
          metrics_log = {k: v for k, v in metrics.items()}
  
          self._log_model(
              model=model,
              model_results=None,
              score_dict=metrics_log,
              source="create_model",
              runtime=runtime,
              model_fit_time=model_fit_time,
              pipeline=self.pipeline,
              log_plots=self.log_plots_param,
              experiment_custom_tags=experiment_custom_tags,
              display=display,
          )
  
      display.move_progress()
  
      self.logger.info("Uploading results into container")
  
      if metrics:
          model_results = pd.DataFrame(metrics, index=[0])
          model_results = model_results.round(round)
          self._display_container.append(model_results)
      else:
          model_results = None
  
      if add_to_model_list:
          # storing results in _master_model_container
          self.logger.info("Uploading model into container now")
          self._master_model_container.append(
              {"model": model, "scores": model_results, "cv": None}
          )
  
      if model_results is not None and system:
          display.display(model_results.style.format(precision=round))
      else:
          display.close()
  
      self.logger.info(
          f"_master_model_container: {len(self._master_model_container)}"
      )
      self.logger.info(
          f"_display_container: {len(self._display_container)}"
      )
  
      self.logger.info(str(model))
      self.logger.info(
          "create_model() successfully completed......................................"
      )
      gc.collect()
  
      if not system:
          return (model, model_fit_time)
  
      return model
  
imports: import datetime
  import gc
  import logging
  import time
  import traceback
  from typing import Any, Callable, Dict, List, Optional, Union
  import numpy
  import pandas
  from joblib.memory import Memory
  from sklearn.base import clone
  from pycaret.containers.metrics.clustering import get_all_metric_containers
  from pycaret.containers.models.clustering import ALL_ALLOWED_ENGINES, get_container_default_engines
  from pycaret.internal.display import CommonDisplay
  from pycaret.internal.logging import get_logger, redirect_output
  from pycaret.internal.pipeline import Pipeline
  from pycaret.internal.pipeline import estimator_pipeline, get_pipeline_fit_kwargs
  from pycaret.internal.preprocess.preprocessor import Preprocessor
  from pycaret.internal.pycaret_experiment.tabular_experiment import _TabularExperiment
  from pycaret.internal.validation import is_sklearn_pipeline
  from pycaret.loggers.base_logger import BaseLogger
  from pycaret.utils.constants import DATAFRAME_LIKE, SEQUENCE_LIKE
  from pycaret.utils.generic import MLUsecase, highlight_setup
label_desc: Use image node_id 8 for calling the _UnsupervisedExperiment obj's underlying member method code with example usage: obj._create_model(estimator, num_clusters, fraction, ground_truth, round, fit_kwargs, experiment_custom_tags, verbose, system, add_to_model_list, raise_num_clusters, X_data, display) and returns: model, model,  model_fit_time