code: def ot_distance(
      ground_truth: pd.Series,
      classifier: Union[pd.Series, pd.DataFrame],
      prot_attr: pd.Series = None,
      favorable_value: Union[str, float] = None,
      scoring: str = "Wasserstein1",
      num_iters: int = 1e5,
      penalty: float = 1e-17,
      mode: str = "binary",
      cost_matrix: np.ndarray = None,
      **kwargs,
  ):
      """Normalize and calculate Wasserstein distance between groups defined by `prot_attr` in `ground_truth` and `classifier`.
  
      Args:
          ground_truth (pd.Series, str): ground truth (correct) target values.
          classifier (pd.Series, pd.DataFrame, str): estimated target values.
              If `mode` is nominal, must be a dataframe with columns containing predictions for each nominal class.
              If `None`, model is assumed to be a dummy model that predicts the mean of the targets
                  or 1/(number of categories) for nominal mode.
          prot_attr (pd.Series, str): sensitive attribute values.
              If `None`, assume all samples belong to the same protected group.
          favorable_value(str, float, optional): Either "high", "low" or a float value if the mode in [binary, ordinal, or continuous].
                  If float, value has to be the minimum or the maximum in the ground_truth column.
                  Defaults to high if None for these modes.
                  Support for float left in to keep the intuition clear in binary classification tasks.
                  If `mode` is nominal, favorable values should be one of the unique categories in the ground_truth.
                  Defaults to a one-vs-all scan if None for nominal mode.
          scoring (str or class): only 'Wasserstein1'
          num_iters (int, optional): number of iterations (random restarts) for EMD. Should be positive.
          penalty (float, optional): penalty term. Should be positive. The penalty term as with any regularization parameter
              may need to be tuned for a particular use case. The higher the penalty, the higher the influence of entropy regualizer.
          mode: one of ['binary', 'continuous', 'nominal', 'ordinal']. Defaults to binary.
                  In nominal mode, up to 10 categories are supported by default.
                  To increase this, pass in keyword argument max_nominal = integer value.
          cost_matrix (np.ndarray): cost matrix for the Wasserstein distance. Defaults to absolute difference between samples.
  
      Returns:
          ot.emd2 (float, dict): Earth mover's distance or dictionary of optimal transports for each of option of classifier
  
      Raises:
          ValueError: if `mode` is 'binary' but `ground_truth` contains less than 1 or more than 2 unique values.
      """
  
      # Assert correct mode passed
      if mode not in ["binary", "continuous", "nominal", "ordinal"]:
          raise ValueError(
              f"Expected one of {['binary', 'continuous', 'nominal', 'ordinal']}, got {mode}."
          )
  
      # Assert correct types passed to ground_truth, classifier and prot_attr
      if not isinstance(ground_truth, (pd.Series, str)):
          raise TypeError(
              f"ground_truth: expected pd.Series or str, got {type(ground_truth)}"
          )
      if classifier is not None:
          if mode in ["binary", "continuous"] and not isinstance(
              classifier, pd.Series
          ):
              raise TypeError(
                  f"classifier: expected pd.Series for {mode} mode, got {type(classifier)}"
              )
          if mode in ["nominal", "ordinal"] and not isinstance(
              classifier, pd.DataFrame
          ):
              raise TypeError(
                  f"classifier: expected pd.DataFrame for {mode} mode, got {type(classifier)}"
              )
      if prot_attr is not None and not isinstance(
          prot_attr, (pd.Series, str)
      ):
          raise TypeError(
              f"prot_attr: expected pd.Series or str, got {type(prot_attr)}"
          )
  
      # Assert correct type passed to cost_matrix
      if cost_matrix is not None and not isinstance(
          cost_matrix, np.ndarray
      ):
          raise TypeError(
              f"cost_matrix: expected numpy.ndarray, got {type(cost_matrix)}"
          )
  
      # Assert scoring is "Wasserstein1"
      if not scoring == "Wasserstein1":
          raise ValueError(
              f'Scoring mode can only be "Wasserstein1", got {scoring}'
          )
  
      grt = ground_truth.copy()
  
      if classifier is not None:
          cls = classifier.copy()
          if prot_attr is not None:
              cls.index = grt.index
      else:
          cls = None
  
      if prot_attr is not None:
          sat = prot_attr.copy()
          sat.index = grt.index
      else:
          sat = None
  
      uniques = list(grt.unique())
      if mode == "binary":
          if len(uniques) > 2:
              raise ValueError(
                  f"Only 2 unique values allowed in ground_truth for binary mode, got {uniques}"
              )
  
      # Encode variables
      if not np.issubdtype(grt.dtype, np.number):
          grt_encoder = LabelEncoder().fit(grt)
          grt = pd.Series(grt_encoder.transform(grt))
  
      # Set correct favorable value (this tells us if higher or lower is better)
      min_val, max_val = grt.min(), grt.max()
  
      if favorable_value == "high":
          favorable_value = max_val
      elif favorable_value == "low":
          favorable_value = min_val
      elif favorable_value is None:
          if mode in ["binary", "ordinal", "continuous"]:
              favorable_value = max_val  # Default to higher is better
          elif mode == "nominal":
              favorable_value = (
                  "flag-all"  # Default to scan through all categories
              )
  
      if favorable_value not in [
          min_val,
          max_val,
          "flag-all",
          *uniques,
      ]:
          raise ValueError(
              f"Favorable_value should be high, low, or one of categories {uniques}, got {favorable_value}."
          )
  
      if (
          mode == "binary"
      ):  # Flip ground truth if favorable_value is 0 in binary mode.
          grt = pd.Series(grt == favorable_value, dtype=int)
          if cls is None:
              cls = pd.Series(grt.mean(), index=grt.index)
          emds = _evaluate(
              grt, cls, sat, num_iters, cost_matrix, **kwargs
          )
  
      elif mode == "continuous":
          if cls is None:
              cls = pd.Series(grt.mean(), index=grt.index)
          emds = _evaluate(
              grt, cls, sat, num_iters, cost_matrix, **kwargs
          )
  
      ## TODO: rework ordinal mode to take into account distance between pred and true
      elif mode in ["nominal", "ordinal"]:
          if (
              cls is None
          ):  # Set classifier to 1/(num of categories) for nominal mode
              cls = pd.DataFrame(
                  [pd.Series(1 / grt.nunique(), index=grt.index)]
                  * grt.nunique()
              )
          if grt.nunique() != cls.shape[-1]:
              raise ValueError(
                  f"classifier must have  a column for each class. Expected shape [:, {grt.nunique()}], got {cls.shape}"
              )
          emds = {}
          for class_label in uniques:
              grt_cl = grt.map({class_label: 1}).fillna(0)
              cls_cl = cls[class_label]
              emds[class_label] = _evaluate(
                  grt_cl, cls_cl, sat, num_iters, cost_matrix, **kwargs
              )
  
      return emds
  
file_path: AIF360/aif360/metrics/ot_metric.py
imports: from typing import Union
  import pandas
  import numpy
  import ot
  from sklearn.preprocessing import LabelEncoder
label_desc: Use image node_id 4 for calling a global function with example usage: ot_distance(ground_truth, classifier, prot_attr, favorable_value, scoring, num_iters, penalty, mode, cost_matrix) and returns: emds