code: def extract_robust_digest(
      search_space: SearchSpace, param_names: List[str]
  ) -> Optional[RobustSearchSpaceDigest]:
      """Extracts the `RobustSearchSpaceDigest`.
  
      Args:
          search_space: A `SearchSpace` to digest.
          param_names: A list of names of the parameters that are used in optimization.
              If environmental variables are present, these should be the last entries
              in `param_names`.
  
      Returns:
          If the `search_space` is not a `RobustSearchSpace`, this returns None.
          Otherwise, it returns a `RobustSearchSpaceDigest` with entries populated
          from the properties of the `search_space`. In particular, this constructs
          two optional callables, `sample_param_perturbations` and `sample_environmental`,
          that require no inputs and return a `num_samples x d`-dim array of samples
          from the corresponding parameter distributions, where `d` is the number of
          environmental variables for `environmental_sampler and the number of
          non-environmental parameters in `param_names` for `distribution_sampler`.
      """
      if not isinstance(search_space, RobustSearchSpace):
          return None
      dist_params = search_space._distributional_parameters
      env_vars: Dict[
          str, Parameter
      ] = search_space._environmental_variables
      pert_params = [p for p in dist_params if p not in env_vars]
      # Make sure all distributional parameters are in param_names.
      dist_idcs: Dict[str, int] = {}
      for p_name in dist_params:
          if p_name not in param_names:
              raise RuntimeError(
                  "All distributional parameters must be included in `param_names`."
              )
          dist_idcs[p_name] = param_names.index(p_name)
      num_samples: int = search_space.num_samples
      if len(env_vars) > 0:
          num_non_env_vars: int = len(param_names) - len(env_vars)
          env_idcs = {
              idx for p, idx in dist_idcs.items() if p in env_vars
          }
          if env_idcs != set(range(num_non_env_vars, len(param_names))):
              raise RuntimeError(
                  "Environmental variables must be last entries in `param_names`. "
                  "Otherwise, `AppendFeatures` will not work."
              )
          # NOTE: Extracting it from `param_names` in case the ordering is different.
          environmental_variables = param_names[num_non_env_vars:]
  
          def sample_environmental() -> np.ndarray:
              """Get samples from the environmental distributions.
  
              Samples have the same dimension as the number of environmental variables.
              The samples of an environmental variable appears in the same order it is
              in `param_names`.
              """
              samples = np.zeros((num_samples, len(env_vars)))
              # pyre-ignore [16]
              for dist in search_space._environmental_distributions:
                  dist_samples = dist.distribution.rvs(
                      num_samples
                  ).reshape(num_samples, -1)
                  for i, p_name in enumerate(dist.parameters):
                      target_idx = dist_idcs[p_name] - num_non_env_vars
                      samples[:, target_idx] = dist_samples[:, i]
              return samples
  
      else:
          sample_environmental = None
          environmental_variables = []
  
      if len(pert_params) > 0:
          constructor: Callable[[Tuple[int, int]], np.ndarray] = (
              np.ones if search_space.multiplicative else np.zeros
          )
  
          def sample_param_perturbations() -> np.ndarray:
              """Get samples of the input perturbations.
  
              Samples have the same dimension as the length of `param_names`
              minus the number of environmental variables. The samples of a
              parameter appears in the same order it is in `param_names`. For
              non-distributional parameters, their values are filled as 0 if
              the perturbations are additive and 1 if multiplicative.
              """
              samples = constructor(
                  (num_samples, len(param_names) - len(env_vars))
              )
              # pyre-ignore [16]
              for dist in search_space._perturbation_distributions:
                  dist_samples = dist.distribution.rvs(
                      num_samples
                  ).reshape(num_samples, -1)
                  for i, p_name in enumerate(dist.parameters):
                      samples[:, dist_idcs[p_name]] = dist_samples[:, i]
              return samples
  
      else:
          sample_param_perturbations = None
  
      return RobustSearchSpaceDigest(
          sample_param_perturbations=sample_param_perturbations,
          sample_environmental=sample_environmental,
          environmental_variables=environmental_variables,
          multiplicative=search_space.multiplicative,
      )
  
imports: from __future__ import annotations
  import warnings
  from copy import deepcopy
  from functools import partial
  from logging import Logger
  from typing import Any, Callable, Dict, Iterable, List, Mapping, MutableMapping, Optional, Tuple, Type, TYPE_CHECKING, Union
  import numpy
  import torch
  from ax.core.data import Data
  from ax.core.experiment import Experiment
  from ax.core.objective import MultiObjective, Objective, ScalarizedObjective
  from ax.core.observation import Observation, ObservationData, ObservationFeatures
  from ax.core.optimization_config import MultiObjectiveOptimizationConfig, OptimizationConfig, TRefPoint
  from ax.core.outcome_constraint import ComparisonOp, OutcomeConstraint, ScalarizedOutcomeConstraint
  from ax.core.parameter import ChoiceParameter, Parameter, ParameterType, RangeParameter
  from ax.core.parameter_constraint import ParameterConstraint
  from ax.core.risk_measures import RiskMeasure
  from ax.core.search_space import RobustSearchSpace, RobustSearchSpaceDigest, SearchSpace, SearchSpaceDigest
  from ax.core.types import TBounds, TCandidateMetadata
  from ax.core.utils import get_pending_observation_features, get_pending_observation_features_based_on_trial_status
  from ax.exceptions.core import DataRequiredError, UnsupportedError, UserInputError
  from ax.modelbridge.transforms.base import Transform
  from ax.modelbridge.transforms.utils import derelativize_optimization_config_with_raw_status_quo
  from ax.models.torch.botorch_moo_defaults import pareto_frontier_evaluator
  from ax.models.torch.frontier_utils import get_weighted_mc_objective_and_objective_thresholds
  from ax.utils.common.logger import get_logger
  from ax.utils.common.typeutils import checked_cast, checked_cast_optional, checked_cast_to_tuple, not_none
  from botorch.acquisition.multi_objective.multi_output_risk_measures import IndependentCVaR, IndependentVaR, MARS, MultiOutputExpectation, MVaR
  from botorch.acquisition.risk_measures import CVaR, Expectation, RiskMeasureMCObjective, VaR, WorstCase
  from botorch.utils.datasets import ContextualDataset, SupervisedDataset
  from botorch.utils.multi_objective.box_decompositions.dominated import DominatedPartitioning
  from torch import Tensor
label_desc: Use image node_id 5 for calling a global function with example usage: extract_robust_digest(search_space, param_names) and returns: RobustSearchSpaceDigest, None, samples, samples