code: def _ComputeMetricsAndPlots(  # pylint: disable=invalid-name
      extracts: beam.pvalue.PCollection,
      eval_config: config_pb2.EvalConfig,
      metrics_specs: List[config_pb2.MetricsSpec],
      eval_shared_models: Optional[
          Dict[str, types.EvalSharedModel]
      ] = None,
      metrics_key: str = constants.METRICS_KEY,
      plots_key: str = constants.PLOTS_KEY,
      attributions_key: str = constants.ATTRIBUTIONS_KEY,
      schema: Optional[schema_pb2.Schema] = None,
      random_seed_for_testing: Optional[int] = None,
  ) -> evaluator.Evaluation:
      """Computes metrics and plots.
  
      Args:
        extracts: PCollection of Extracts. If a query_key was used then the
          PCollection will contain a list of extracts.
        eval_config: Eval config.
        metrics_specs: Subset of the metric specs to compute metrics for. If a
          query_key was used all of the metric specs will be for the same query_key.
        eval_shared_models: Optional dict of shared models keyed by model name. Only
          required if there are metrics to be computed in-graph using the model.
        metrics_key: Name to use for metrics key in Evaluation output.
        plots_key: Name to use for plots key in Evaluation output.
        attributions_key: Name to use for attributions key in Evaluation output.
        schema: A schema to use for customizing metrics and plots.
        random_seed_for_testing: Seed to use for unit testing.
  
      Returns:
        Evaluation containing dict of PCollections of (slice_key, results_dict)
        tuples where the dict is keyed by either the metrics_key (e.g. 'metrics'),
        plots_key (e.g. 'plots'), or attributions_key (e.g. 'attributions')
        depending on what the results_dict contains.
      """
      computations = []
      # Add default metric computations
      if eval_shared_models:
          # Note that there is the possibility for metric naming collisions here
          # (e.g. 'auc' calculated within the model as well as by AUC metric
          # computation performed outside the model). Currently all the overlapping
          # metrics such as AUC that are computed outside the model are all derived
          # metrics so they will override the metrics calculated by the model which is
          # the desired behavior.
          for (
              model_name,
              eval_shared_model,
          ) in eval_shared_models.items():
              if not eval_shared_model.include_default_metrics:
                  continue
              if eval_shared_model.model_type == constants.TF_KERAS:
                  computations.extend(
                      keras_util.metric_computations_using_keras_saved_model(
                          model_name,
                          eval_shared_model.model_loader,
                          eval_config,
                      )
                  )
              elif eval_shared_model.model_type == constants.TFMA_EVAL:
                  computations.extend(
                      eval_saved_model_util.metric_computations_using_eval_saved_model(
                          model_name, eval_shared_model.model_loader
                      )
                  )
      # Add metric computations from specs
      metric_computations = _filter_and_separate_computations(
          metric_specs.to_computations(
              metrics_specs, eval_config=eval_config, schema=schema
          )
      )
      computations.extend(metric_computations.non_derived_computations)
  
      # Find out which model is baseline.
      baseline_spec = model_util.get_baseline_model_spec(eval_config)
      baseline_model_name = (
          baseline_spec.name if baseline_spec else None
      )
  
      # pylint: disable=no-value-for-parameter
  
      # Input: Single extract per example (or list of extracts if query_key used)
      #        where each item contains slice keys and other extracts from upstream
      #        extractors (e.g. labels, predictions, etc).
      # Output: Single extract (per example) containing slice keys and initial
      #         combiner state returned from preprocessor. Note that even if a
      #         query_key was used the output is still only a single extract
      #         (though, that extract may contain lists of values (predictions,
      #         labels, etc) in its keys).
      #
      # Note that the output of this step is extracts instead of just a tuple of
      # computation outputs because FanoutSlices takes extracts as input (and in
      # many cases a subset of the extracts themselves are what is fanned out).
      extracts = extracts | "Preprocesss" >> beam.ParDo(
          _PreprocessorDoFn(computations)
      )
  
      # Input: Single extract containing slice keys and initial combiner inputs. If
      #        query_key is used the extract represents multiple examples with the
      #        same query_key, otherwise the extract represents a single example.
      # Output: Tuple (slice key, combiner inputs extracts). Notice that the per
      #         example (or list or examples if query_key used) input extract turns
      #         into n logical extracts, references to which are replicated once per
      #         applicable slice key.
      slices = extracts | "FanoutSlices" >> slicer.FanoutSlices()
  
      slices_count = (
          slices
          | "ExtractSliceKeys" >> beam.Keys()
          | "CountPerSliceKey" >> beam.combiners.Count.PerElement()
      )
  
      model_types = _get_model_types_for_logging(eval_shared_models)
  
      _ = (
          extracts.pipeline
          | "IncrementMetricsSpecsCounters"
          >> counter_util.IncrementMetricsSpecsCounters(
              metrics_specs, model_types
          ),
          slices_count
          | "IncrementSliceSpecCounters"
          >> counter_util.IncrementSliceSpecCounters(),
      )
  
      ci_params = _get_confidence_interval_params(
          eval_config, metrics_specs
      )
  
      cross_slice_specs = []
      if eval_config.cross_slicing_specs:
          cross_slice_specs = eval_config.cross_slicing_specs
  
      computations_combine_fn = _ComputationsCombineFn(
          computations=computations
      )
      derived_metrics_ptransform = _AddDerivedCrossSliceAndDiffMetrics(
          metric_computations.derived_computations,
          metric_computations.cross_slice_computations,
          cross_slice_specs,
          baseline_model_name,
      )
  
      # Input: Tuple of (slice key, combiner input extracts).
      # Output: Tuple of (slice key, dict of computed metrics/plots/attributions).
      #         The dicts will be keyed by MetricKey/PlotKey/AttributionsKey and the
      #         values will be the result of the associated computations. A given
      #         MetricComputation can perform computations for multiple keys, but
      #         the keys should be unique across computations.
      if ci_params.num_bootstrap_samples:
          sliced_metrics_plots_and_attributions = (
              slices
              | "PoissonBootstrapConfidenceIntervals"
              >> poisson_bootstrap.ComputeWithConfidenceIntervals(
                  computations_combine_fn=computations_combine_fn,
                  derived_metrics_ptransform=derived_metrics_ptransform,
                  num_bootstrap_samples=ci_params.num_bootstrap_samples,
                  hot_key_fanout=_COMBINE_PER_SLICE_KEY_HOT_KEY_FANOUT,
                  skip_ci_metric_keys=ci_params.skip_ci_metric_keys,
                  random_seed_for_testing=random_seed_for_testing,
              )
          )
      elif ci_params.num_jackknife_samples:
          sliced_metrics_plots_and_attributions = (
              slices
              | "JackknifeConfidenceIntervals"
              >> jackknife.ComputeWithConfidenceIntervals(
                  computations_combine_fn=computations_combine_fn,
                  derived_metrics_ptransform=derived_metrics_ptransform,
                  num_jackknife_samples=ci_params.num_jackknife_samples,
                  skip_ci_metric_keys=ci_params.skip_ci_metric_keys,
                  random_seed_for_testing=random_seed_for_testing,
              )
          )
      else:
          sliced_metrics_plots_and_attributions = (
              slices
              | "CombineMetricsPerSlice"
              >> beam.CombinePerKey(
                  computations_combine_fn
              ).with_hot_key_fanout(
                  _COMBINE_PER_SLICE_KEY_HOT_KEY_FANOUT
              )
              | "AddDerivedCrossSliceAndDiffMetrics"
              >> derived_metrics_ptransform
          )
  
      sliced_metrics_plots_and_attributions = (
          sliced_metrics_plots_and_attributions
          | "AddCIDerivedMetrics"
          >> beam.Map(
              _add_ci_derived_metrics,
              metric_computations.ci_derived_computations,
          )
          | "RemovePrivateMetrics"
          >> beam.MapTuple(_remove_private_metrics)
      )
  
      if eval_config.options.min_slice_size.value > 1:
          sliced_metrics_plots_and_attributions = (
              sliced_metrics_plots_and_attributions
              | "FilterSmallSlices"
              >> slicer.FilterOutSlices(
                  slices_count, eval_config.options.min_slice_size.value
              )
          )
  
      sliced_metrics = (
          sliced_metrics_plots_and_attributions
          | "FilterByMetrics"
          >> beam.Map(_filter_by_key_type, metric_types.MetricKey)
      )
      sliced_plots = (
          sliced_metrics_plots_and_attributions
          | "FilterByPlots"
          >> beam.Map(_filter_by_key_type, metric_types.PlotKey)
      )
  
      sliced_attributions = (
          sliced_metrics_plots_and_attributions
          | "FilterByAttributions"
          >> beam.Map(_filter_by_key_type, metric_types.AttributionsKey)
      )
  
      # pylint: enable=no-value-for-parameter
  
      return {
          metrics_key: sliced_metrics,
          plots_key: sliced_plots,
          attributions_key: sliced_attributions,
      }
  
imports: import copy
  import datetime
  import itertools
  import numbers
  from typing import Any, Dict, Iterable, Iterator, List, NamedTuple, Optional, Tuple, Type, TypeVar, Union
  import apache_beam
  import numpy
  from tensorflow_model_analysis import constants
  from tensorflow_model_analysis.api import types
  from tensorflow_model_analysis.evaluators import counter_util
  from tensorflow_model_analysis.evaluators import eval_saved_model_util
  from tensorflow_model_analysis.evaluators import evaluator
  from tensorflow_model_analysis.evaluators import jackknife
  from tensorflow_model_analysis.evaluators import keras_util
  from tensorflow_model_analysis.evaluators import metrics_validator
  from tensorflow_model_analysis.evaluators import poisson_bootstrap
  from tensorflow_model_analysis.extractors import slice_key_extractor
  from tensorflow_model_analysis.metrics import metric_specs
  from tensorflow_model_analysis.metrics import metric_types
  from tensorflow_model_analysis.metrics import metric_util
  from tensorflow_model_analysis.proto import config_pb2
  from tensorflow_model_analysis.slicer import slicer_lib
  from tensorflow_model_analysis.utils import model_util
  from tensorflow_model_analysis.utils import util
  from tensorflow_metadata.proto.v0 import schema_pb2
label_desc: Use image node_id 12 for calling a global function with example usage: _ComputeMetricsAndPlots(extracts, eval_config, metrics_specs, eval_shared_models, metrics_key, plots_key, attributions_key, schema, random_seed_for_testing) and returns: dict