code: def _ComputeMetricsAndPlots(  # pylint: disable=invalid-name
      extracts: beam.pvalue.PCollection,
      eval_shared_model: types.EvalSharedModel,
      desired_batch_size: Optional[int] = None,
      compute_confidence_intervals: Optional[bool] = False,
      random_seed_for_testing: Optional[int] = None,
  ) -> Tuple[beam.pvalue.DoOutputsTuple, beam.pvalue.PCollection]:
      """Computes metrics and plots using the EvalSavedModel.
  
      Args:
        extracts: PCollection of Extracts. The extracts MUST contain a
          FeaturesPredictionsLabels extract keyed by
          tfma.FEATURE_PREDICTIONS_LABELS_KEY and a list of SliceKeyType extracts
          keyed by tfma.SLICE_KEY_TYPES_KEY. Typically these will be added by
          calling the default_extractors function.
        eval_shared_model: Shared model parameters for EvalSavedModel including any
          additional metrics (see EvalSharedModel for more information on how to
          configure additional metrics).
        desired_batch_size: Optional batch size for batching in Aggregate.
        compute_confidence_intervals: Set to True to run metrics analysis over
          multiple bootstrap samples and compute uncertainty intervals.
        random_seed_for_testing: Provide for deterministic tests only.
  
      Returns:
        Tuple of Tuple[PCollection of (slice key, metrics),
        PCollection of (slice key, plot metrics)] and
        PCollection of (slice_key and its example count).
      """
      # pylint: disable=no-value-for-parameter
  
      slices = (
          extracts
          # Downstream computation only cares about FPLs, so we prune before fanout.
          # Note that fanout itself will prune the slice keys.
          # TODO(b/130032676, b/111353165): Prune FPLs to contain only the necessary
          # set for the calculation of post_export_metrics if possible.
          | "PruneExtracts"
          >> extractor.Filter(
              include=[
                  constants.FEATURES_PREDICTIONS_LABELS_KEY,
                  constants.SLICE_KEY_TYPES_KEY,
                  constants.INPUT_KEY,
              ]
          )
          # Input: one example at a time, with slice keys in extracts.
          # Output: one fpl example per slice key (notice that the example turns
          #         into n logical examples, references to which are replicated once
          #         per applicable slice key).
          | "FanoutSlices" >> slicer.FanoutSlices()
      )
  
      slices_count = (
          slices
          | "ExtractSliceKeys" >> beam.Keys()
          | "CountPerSliceKey" >> beam.combiners.Count.PerElement()
      )
  
      _ = (
          extracts.pipeline
          | "IncrementMetricsCallbacksCounters"
          >> counter_util.IncrementMetricsCallbacksCounters(
              eval_shared_model.add_metrics_callbacks,
              eval_shared_model.model_type,
          ),
          slices_count
          | "IncreamentSliceSpecCounters"
          >> counter_util.IncrementSliceSpecCounters(),
      )
  
      aggregated_metrics = (
          slices
          # Metrics are computed per slice key.
          # Output: Multi-outputs, a dict of slice key to computed metrics, and
          # plots if applicable.
          | "ComputePerSliceMetrics"
          >> poisson_bootstrap.ComputeWithConfidenceIntervals(
              legacy_aggregate.ComputePerSliceMetrics,
              num_bootstrap_samples=(
                  poisson_bootstrap.DEFAULT_NUM_BOOTSTRAP_SAMPLES
                  if compute_confidence_intervals
                  else 1
              ),
              random_seed_for_testing=random_seed_for_testing,
              eval_shared_model=eval_shared_model,
              desired_batch_size=desired_batch_size,
          )
          | "SeparateMetricsAndPlots"
          >> beam.ParDo(_SeparateMetricsAndPlotsFn()).with_outputs(
              _SeparateMetricsAndPlotsFn.OUTPUT_TAG_PLOTS,
              main=_SeparateMetricsAndPlotsFn.OUTPUT_TAG_METRICS,
          )
      )
  
      return (aggregated_metrics, slices_count)
  
file_path: model-analysis/tensorflow_model_analysis/evaluators/legacy_metrics_and_plots_evaluator.py
imports: from typing import Any, Dict, Optional, Tuple
  import apache_beam
  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 evaluator
  from tensorflow_model_analysis.evaluators import legacy_aggregate
  from tensorflow_model_analysis.evaluators import legacy_poisson_bootstrap
  from tensorflow_model_analysis.extractors import extractor
  from tensorflow_model_analysis.extractors import slice_key_extractor
  from tensorflow_model_analysis.post_export_metrics import metric_keys
  from tensorflow_model_analysis.slicer import slicer_lib
  from tensorflow_model_analysis.writers import metrics_plots_and_validations_writer
label_desc: Use image node_id 2 for calling a global function with example usage: _ComputeMetricsAndPlots(extracts, eval_shared_model, desired_batch_size, compute_confidence_intervals, random_seed_for_testing) and returns: aggregated_metrics,  slices_count