code: def _classify_featureset(
      args: Dict[str, Any]
  ) -> List[Dict[str, Any]]:
      """
      Run classification job.
  
      Parameters
      ----------
      args : Dict[str, Any]
          A dictionary with arguments for classifying the ``FeatureSet`` instance.
  
      Returns
      -------
      res : List[Dict[str, Any]]
          The results of the classification, in the format of a list of dictionaries.
  
      Raises
      ------
      ValueError
          If extra unknown arguments are passed to the function.
      """
      # Extract all the arguments.
      # (There doesn't seem to be a better way to do this since one can't specify
      # required keyword arguments.)
  
      experiment_name = args.pop("experiment_name")
      task = args.pop("task")
      sampler = args.pop("sampler")
      feature_hasher = args.pop("feature_hasher")
      hasher_features = args.pop("hasher_features")
      job_name = args.pop("job_name")
      featureset = args.pop("featureset")
      featureset_name = args.pop("featureset_name")
      learner_name = args.pop("learner_name")
      train_path = args.pop("train_path")
      test_path = args.pop("test_path")
      train_set_name = args.pop("train_set_name")
      test_set_name = args.pop("test_set_name")
      shuffle = args.pop("shuffle")
      model_path = args.pop("model_path")
      prediction_prefix = args.pop("prediction_prefix")
      grid_search = args.pop("grid_search")
      grid_objective = args.pop("grid_objective")
      output_metrics = args.pop("output_metrics")
      suffix = args.pop("suffix")
      job_log_file = args.pop("log_file")
      job_log_level = args.pop("log_level")
      probability = args.pop("probability")
      pipeline = args.pop("pipeline")
      results_path = args.pop("results_path")
      fixed_parameters = args.pop("fixed_parameters")
      sampler_parameters = args.pop("sampler_parameters")
      param_grid = args.pop("param_grid")
      pos_label = args.pop("pos_label")
      overwrite = args.pop("overwrite")
      feature_scaling = args.pop("feature_scaling")
      min_feature_count = args.pop("min_feature_count")
      folds_file = args.pop("folds_file")
      grid_search_jobs = args.pop("grid_search_jobs")
      grid_search_folds = args.pop("grid_search_folds")
      cv_folds = args.pop("cv_folds")
      cv_seed = args.pop("cv_seed")
      save_cv_folds = args.pop("save_cv_folds")
      save_cv_models = args.pop("save_cv_models")
      use_folds_file_for_grid_search = args.pop(
          "use_folds_file_for_grid_search"
      )
      stratified_folds = args.pop("do_stratified_folds")
      label_col = args.pop("label_col")
      id_col = args.pop("id_col")
      ids_to_floats = args.pop("ids_to_floats")
      class_map = args.pop("class_map")
      custom_learner_path = args.pop("custom_learner_path")
      custom_metric_path = args.pop("custom_metric_path")
      quiet = args.pop("quiet", False)
      learning_curve_cv_folds = args.pop("learning_curve_cv_folds")
      learning_curve_train_sizes = args.pop(
          "learning_curve_train_sizes"
      )
      save_votes = args.pop("save_votes")
  
      if args:
          raise ValueError(
              "Extra arguments passed to _classify_featureset: "
              f"{args.keys()}"
          )
      start_timestamp = datetime.datetime.now()
  
      # create a new SKLL logger for this specific job and
      # use the given log level
      logger = get_skll_logger(
          job_name, job_log_file, log_level=job_log_level
      )
  
      try:
          # log messages
          logger.info(f"Task: {task}")
  
          # check if we have any possible custom metrics
          possible_custom_metric_names = []
          for metric_name in output_metrics + [grid_objective]:
              # metrics that are not in `SCORERS` or `None` are candidates
              # (the `None` is a by-product of how jobs with single tuning
              # objectives are created)
              if (
                  metric_name not in get_scorer_names()
                  and metric_name not in _CUSTOM_METRICS
                  and metric_name is not None
              ):
                  possible_custom_metric_names.append(metric_name)
  
          # initialize list that will hold any invalid metrics
          # that we could not register as custom metrics
          invalid_metric_names = []
  
          # if we have possible custom metrics
          if possible_custom_metric_names:
              # check that we have a file to load them from
              if not custom_metric_path:
                  raise ValueError(
                      f"invalid metrics specified: {possible_custom_metric_names}"
                  )
              else:
                  # try to register each possible custom metric
                  # raise an exception if we fail, if we don't then
                  # add the custom metric function to `globals()` so
                  # that it serializes properly for gridmap
                  for (
                      custom_metric_name
                  ) in possible_custom_metric_names:
                      try:
                          custom_metric_func = register_custom_metric(
                              custom_metric_path, custom_metric_name
                          )
                      except (AttributeError, NameError, ValueError):
                          invalid_metric_names.append(
                              custom_metric_name
                          )
                      else:
                          logger.info(
                              f"registered '{custom_metric_name}' as a "
                              f"custom metric"
                          )
                          globals()[
                              custom_metric_name
                          ] = custom_metric_func
  
          # raise an error if we have any invalid metrics
          if invalid_metric_names:
              raise ValueError(
                  f"invalid metrics specified: {invalid_metric_names}. "
                  f"If these are custom metrics, check the function "
                  f"names."
              )
  
          if task == "cross_validate":
              if isinstance(cv_folds, int):
                  num_folds = cv_folds
              else:  # folds_file was used, so count the unique fold ids.
                  num_folds = len(set(cv_folds.values()))
              logger.info(
                  f"Cross-validating ({num_folds} folds, seed={cv_seed}) on "
                  f"{train_set_name}, feature set {featureset} ..."
              )
          elif task == "evaluate":
              logger.info(
                  f"Training on {train_set_name}, Test on "
                  f"{test_set_name}, feature set {featureset} ..."
              )
          elif task == "train":
              logger.info(
                  f"Training on {train_set_name}, feature set "
                  f"{featureset} ..."
              )
          elif task == "learning_curve":
              logger.info(
                  "Generating learning curve "
                  f"({learning_curve_cv_folds} 80/20 folds, "
                  f"sizes={learning_curve_train_sizes}, "
                  f"objective={grid_objective}) on {train_set_name}, "
                  f"feature set {featureset} ..."
              )
          else:  # predict
              logger.info(
                  f"Training on {train_set_name}, Making predictions on"
                  f" {test_set_name}, feature set {featureset} ..."
              )
  
          # check whether a trained model on the same data with the same
          # featureset already exists if so, load it and then use it on test data
          modelfile = Path(model_path) / f"{job_name}.model"
          train_set_size: Union[int, str] = "unknown"
          if (
              task in ["cross_validate", "learning_curve"]
              or not modelfile.exists()
              or overwrite
          ):
              train_examples = load_featureset(
                  train_path,
                  featureset,
                  suffix,
                  label_col=label_col,
                  id_col=id_col,
                  ids_to_floats=ids_to_floats,
                  quiet=quiet,
                  class_map=class_map,
                  feature_hasher=feature_hasher,
                  num_features=hasher_features,
                  logger=logger,
              )
  
              train_set_size = len(train_examples.ids)
              if not train_examples.has_labels:
                  raise ValueError(
                      "Training examples do not have labels"
                  )
  
              # set up some keyword arguments for instantiating the learner
              # object; note that these are shared by all types of learners
              # supported by SKLL (regular and voting)
              common_learner_kwargs = {
                  "custom_learner_path": custom_learner_path,
                  "feature_scaling": feature_scaling,
                  "pos_label": pos_label,
                  "min_feature_count": min_feature_count,
                  "logger": logger,
              }
  
              # instantiate the right type of learner object
              learner: Union[Learner, VotingLearner]
              if learner_name in [
                  "VotingClassifier",
                  "VotingRegressor",
              ]:
                  # the fixed parameters dictionary must at least
                  # contains the estimator names for the voting;
                  # the rest can be set to default values
                  try:
                      learner_names = fixed_parameters[
                          "estimator_names"
                      ]
                  except KeyError:
                      raise ValueError(
                          "'estimator_names' must be specified as "
                          "fixed parameters for voting classifiers "
                          "and/or regressors."
                      ) from None
                  else:
                      voting_type = fixed_parameters.get(
                          "voting_type", "hard"
                      )
                      model_kwargs_list = fixed_parameters.get(
                          "estimator_fixed_parameters"
                      )
                      sampler_list = fixed_parameters.get(
                          "estimator_samplers"
                      )
                      sampler_kwargs_list = fixed_parameters.get(
                          "estimator_sampler_parameters"
                      )
                      param_grids_list = fixed_parameters.get(
                          "estimator_param_grids"
                      )
                      learner = VotingLearner(
                          learner_names,
                          voting=voting_type,
                          model_kwargs_list=model_kwargs_list,
                          sampler_list=sampler_list,
                          sampler_kwargs_list=sampler_kwargs_list,
                          **common_learner_kwargs,
                      )
              else:
                  learner = Learner(
                      learner_name,
                      model_kwargs=fixed_parameters,
                      pipeline=pipeline,
                      sampler=sampler,
                      sampler_kwargs=sampler_parameters,
                      probability=probability,
                      **common_learner_kwargs,
                  )
  
          # load the model if it already exists
          else:
              # import custom learner into global namespace if we are reusing
              # a saved model
              if custom_learner_path:
                  globals()[learner_name] = load_custom_learner(
                      custom_learner_path, learner_name
                  )
  
              # load the non-custom learner from disk
              if modelfile.exists() and not overwrite:
                  logger.info(
                      f"Loading pre-existing {learner_name} model: "
                      f"{modelfile}"
                  )
              if learner_name in [
                  "VotingClassifier",
                  "VotingRegressor",
              ]:
                  learner = VotingLearner.from_file(
                      modelfile, logger=logger
                  )
              else:
                  learner = Learner.from_file(modelfile, logger=logger)
  
          # Load test set if there is one
          test_set_size: Union[int, str] = "n/a"
          if task == "evaluate" or task == "predict":
              test_examples = load_featureset(
                  test_path,
                  featureset,
                  suffix,
                  label_col=label_col,
                  id_col=id_col,
                  ids_to_floats=ids_to_floats,
                  quiet=quiet,
                  class_map=class_map,
                  feature_hasher=feature_hasher,
                  num_features=hasher_features,
              )
              test_set_size = len(test_examples.ids)
  
          # compute information about xval and grid folds that can be put in results
          # in readable form
          if isinstance(cv_folds, dict):
              cv_folds_to_print = (
                  f"{len(set(cv_folds.values()))} via folds file"
              )
          else:
              cv_folds_to_print = str(cv_folds)
  
          if isinstance(grid_search_folds, dict):
              grid_search_folds_to_print = f"{len(set(grid_search_folds.values()))} via folds file"
          else:
              grid_search_folds_to_print = str(grid_search_folds)
  
          # create a list of dictionaries of the results information
          learner_result_dict_base = {
              "experiment_name": experiment_name,
              "train_set_name": train_set_name,
              "train_set_size": train_set_size,
              "test_set_name": test_set_name,
              "test_set_size": test_set_size,
              "featureset": json.dumps(featureset),
              "featureset_name": featureset_name,
              "shuffle": shuffle,
              "learner_name": learner_name,
              "task": task,
              "start_timestamp": start_timestamp.strftime(
                  "%d %b %Y %H:%M:" "%S.%f"
              ),
              "version": __version__,
              "feature_scaling": feature_scaling,
              "folds_file": folds_file,
              "grid_search": grid_search,
              "grid_objective": grid_objective,
              "grid_search_folds": grid_search_folds_to_print,
              "min_feature_count": min_feature_count,
              "cv_folds": cv_folds_to_print,
              "using_folds_file": isinstance(cv_folds, dict)
              or isinstance(grid_search_folds, dict),
              "save_cv_folds": save_cv_folds,
              "save_cv_models": save_cv_models,
              "use_folds_file_for_grid_search": use_folds_file_for_grid_search,
              "stratified_folds": stratified_folds,
              "scikit_learn_version": SCIKIT_VERSION,
          }
  
          # check if we're doing cross-validation, because we only load/save
          # models when we're not.
          task_results: List[EvaluateTaskResults]
          grid_scores: Union[List[None], List[float]] = [None]
          grid_search_cv_results_dicts: Union[
              List[None], List[Dict[str, Any]]
          ] = [None]
  
          if task == "cross_validate":
              logger.info("Cross-validating")
  
              # set up the keyword arguments for learner cross-validation;
              # note that most are shared by all types of learners
              # supported by SKLL (regular and voting)
              xval_kwargs = {
                  "shuffle": shuffle,
                  "stratified": stratified_folds,
                  "prediction_prefix": prediction_prefix,
                  "grid_search": grid_search,
                  "grid_search_folds": grid_search_folds,
                  "cv_folds": cv_folds,
                  "cv_seed": cv_seed,
                  "grid_objective": grid_objective,
                  "output_metrics": output_metrics,
                  "grid_jobs": grid_search_jobs,
                  "save_cv_folds": save_cv_folds,
                  "save_cv_models": save_cv_models,
                  "use_custom_folds_for_grid_search": use_folds_file_for_grid_search,
              }
  
              # voting learners require an optional list of parameter grids
              # passed as fixed parameters whereas regular learners only need
              # a single parameter grid
              if isinstance(learner, VotingLearner):
                  xval_kwargs["param_grid_list"] = param_grids_list
                  xval_kwargs["individual_predictions"] = save_votes
              else:
                  xval_kwargs["param_grid"] = param_grid
  
              # cross-validate the learner
              # voting learners return only a subset of the results
              # for cross-validation (no grid search results)
              models: Union[
                  Optional[List[Learner]], Optional[List[VotingLearner]]
              ]
              if isinstance(learner, Learner):
                  learner_cv_results: CrossValidateTaskResults = (
                      learner.cross_validate(
                          train_examples, **xval_kwargs
                      )
                  )
                  (
                      task_results,
                      grid_scores,
                      grid_search_cv_results_dicts,
                      skll_fold_ids,
                      models,
                  ) = learner_cv_results
              else:
                  voting_learner_cv_results: VotingCrossValidateTaskResults = learner.cross_validate(
                      train_examples, **xval_kwargs
                  )
                  (
                      task_results,
                      skll_fold_ids,
                      models,
                  ) = voting_learner_cv_results
                  grid_scores = [None] * cv_folds
                  grid_search_cv_results_dicts = [None] * cv_folds
  
              if models and len(models) > 0:
                  for index, model in enumerate(models, start=1):
                      assert isinstance(model, (Learner, VotingLearner))
                      modelfile = (
                          Path(model_path)
                          / f"{job_name}_fold{index}.model"
                      )
                      model.save(modelfile)
  
          elif task == "learning_curve":
              logger.info("Generating learning curve(s)")
              (
                  curve_train_scores,
                  curve_test_scores,
                  curve_fit_times,
                  computed_curve_train_sizes,
              ) = learner.learning_curve(
                  train_examples,
                  grid_objective,
                  cv_folds=learning_curve_cv_folds,
                  train_sizes=learning_curve_train_sizes,
              )
          else:
              # if we do not have a saved model, we need to train one
              if not modelfile.exists() or overwrite:
                  logger.info(
                      f"Featurizing and training new {learner_name} model"
                  )
  
                  # set up the keyword arguments for learner training;
                  # note that most are shared by all types of learners
                  # supported by SKLL (regular and voting)
                  train_kwargs = {
                      "grid_search": grid_search,
                      "grid_search_folds": grid_search_folds,
                      "grid_objective": grid_objective,
                      "grid_jobs": grid_search_jobs,
                      "shuffle": shuffle,
                  }
  
                  # voting learners require an optional list of parameter grids
                  # passed as fixed parameters whereas regular learners only need
                  # a single parameter grid
                  if isinstance(learner, VotingLearner):
                      train_kwargs["param_grid_list"] = param_grids_list
                  else:
                      train_kwargs["param_grid"] = param_grid
  
                  # train the model
  
                  # regular learners return a grid score and results
                  if isinstance(learner, Learner):
                      results = learner.train(
                          train_examples, **train_kwargs
                      )
                      grid_scores = [results[0]]
                      grid_search_cv_results_dicts = [results[1]]
                      if grid_search:
                          logger.info(
                              f"Best {grid_objective} grid search score: "
                              f"{round(results[0], 3)}"
                          )
                  else:
                      learner.train(train_examples, **train_kwargs)
  
                  # save model, if asked
                  if model_path:
                      learner.save(modelfile)
  
              # print out the model parameters; note that for
              # voting learners, we exclude the parameters for
              # the underlying estimators
              params = learner.model.get_params(
                  deep=isinstance(learner, Learner)
              )
              param_out = (
                  f"{param_name}: {param_value}"
                  for param_name, param_value in params.items()
              )
              logger.info(f"Hyperparameters: {', '.join(param_out)}")
  
              # evaluate the model on the test set;
              if task == "evaluate":
                  logger.info("Evaluating predictions")
                  # voting learners have an extra keyword argument indicating
                  # whether we want to save the individual predictions (or votes)
                  extra_kwargs = {}
                  if isinstance(learner, VotingLearner):
                      extra_kwargs[
                          "individual_predictions"
                      ] = save_votes
                  task_results = [
                      learner.evaluate(
                          test_examples,
                          prediction_prefix=prediction_prefix,
                          grid_objective=grid_objective,
                          output_metrics=output_metrics,
                          **extra_kwargs,
                      )
                  ]
              elif task == "predict":
                  logger.info("Writing predictions")
                  # we set `class_labels` to `False` so that if the learner is
                  # probabilistic, probabilities are written instead of labels;
                  # voting learners have an extra keyword argument indicating
                  # whether we want to save the individual predictions (or votes)
                  extra_kwargs = {}
                  if isinstance(learner, VotingLearner):
                      extra_kwargs[
                          "individual_predictions"
                      ] = save_votes
                  learner.predict(
                      test_examples,
                      prediction_prefix=prediction_prefix,
                      class_labels=False,
                      **extra_kwargs,
                  )
  
          end_timestamp = datetime.datetime.now()
          learner_result_dict_base[
              "end_timestamp"
          ] = end_timestamp.strftime("%d %b %Y %H:%M:%S.%f")
          total_time = end_timestamp - start_timestamp
          learner_result_dict_base["total_time"] = str(total_time)
  
          if task == "cross_validate" or task == "evaluate":
              results_json_path = (
                  Path(results_path) / f"{job_name}.results.json"
              )
  
              res = _create_learner_result_dicts(
                  task_results,
                  grid_scores,
                  grid_search_cv_results_dicts,
                  learner_result_dict_base,
              )
  
              # write out the result dictionary to a json file
              with open(results_json_path, "w") as json_file:
                  json.dump(res, json_file, cls=NumpyTypeEncoder)
  
              with open(
                  Path(results_path) / f"{job_name}.results", "w"
              ) as output_file:
                  _print_fancy_output(res, output_file)
  
          elif task == "learning_curve":
              results_json_path = (
                  Path(results_path) / f"{job_name}.results.json"
              )
              result_dict = {}
              result_dict.update(learner_result_dict_base)
              result_dict.update(
                  {
                      "learning_curve_cv_folds": learning_curve_cv_folds,
                      "given_curve_train_sizes": learning_curve_train_sizes,
                      "learning_curve_train_scores_means": np.mean(
                          curve_train_scores, axis=1
                      ),
                      "learning_curve_test_scores_means": np.mean(
                          curve_test_scores, axis=1
                      ),
                      "learning_curve_fit_times_means": np.mean(
                          curve_fit_times, axis=1
                      ),
                      "learning_curve_train_scores_stds": np.std(
                          curve_train_scores, axis=1, ddof=1
                      ),
                      "learning_curve_test_scores_stds": np.std(
                          curve_test_scores, axis=1, ddof=1
                      ),
                      "learning_curve_fit_times_stds": np.std(
                          curve_fit_times, axis=1
                      ),
                      "computed_curve_train_sizes": computed_curve_train_sizes,
                  }
              )
  
              # we need to return and write out a list of dictionaries
              res = [result_dict]
  
              # write out the result dictionary to a json file
              with open(results_json_path, "w") as json_file:
                  json.dump(res, json_file, cls=NumpyTypeEncoder)
  
          # For all other tasks, i.e. train or predict
          else:
              if results_path:
                  results_json_path = (
                      Path(results_path) / f"{job_name}.results.json"
                  )
  
                  assert len(grid_scores) == 1
                  assert len(grid_search_cv_results_dicts) == 1
                  grid_search_cv_results_dict: Dict[str, Any] = {
                      "grid_score": grid_scores[0]
                  }
                  grid_search_cv_results_dict[
                      "grid_search_cv_results"
                  ] = grid_search_cv_results_dicts[0]
                  grid_search_cv_results_dict.update(
                      learner_result_dict_base
                  )
                  # write out the result dictionary to a json file
                  with open(results_json_path, "w") as json_file:
                      json.dump(
                          grid_search_cv_results_dict,
                          json_file,
                          cls=NumpyTypeEncoder,
                      )
              res = [learner_result_dict_base]
  
          # write out the cv folds if required
          if (
              task == "cross_validate"
              and save_cv_folds
              and skll_fold_ids is not None
          ):
              skll_fold_ids_file = (
                  f"{experiment_name}_skll_fold_ids.csv"
              )
              with open(
                  Path(results_path) / skll_fold_ids_file, "w"
              ) as output_file:
                  _write_skll_folds(skll_fold_ids, output_file)
  
      finally:
          close_and_remove_logger_handlers(logger)
  
      return res
  
imports: from __future__ import annotations
  import datetime
  import json
  import logging
  from itertools import combinations
  from pathlib import Path
  from typing import Any, Dict, List, Optional, Union
  import matplotlib.pyplot
  import numpy
  from sklearn import __version__
  from sklearn.metrics import get_scorer_names
  from skll.config import parse_config_file
  from skll.config.utils import _munge_featureset_name
  from skll.learner import MAX_CONCURRENT_PROCESSES, Learner, load_custom_learner
  from skll.learner.voting import VotingLearner
  from skll.metrics import _CUSTOM_METRICS, register_custom_metric
  from skll.types import CrossValidateTaskResults, EvaluateTaskResults, PathOrStr, VotingCrossValidateTaskResults
  from skll.utils.logging import close_and_remove_logger_handlers, get_skll_logger
  from skll.version import __version__
  from ..utils.wandb import WandbLogger
  from .input import load_featureset
  from .output import _print_fancy_output, _write_learning_curve_file, _write_skll_folds, _write_summary_file, generate_learning_curve_plots
  from .utils import NumpyTypeEncoder, _check_job_results, _create_learner_result_dicts
label_desc: Use image node_id 1 for calling a global function with example usage: _classify_featureset(args) and returns: res