





"""simple docstring"""


import math
import time

from transformers import Trainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput, speed_metrics


if is_torch_tpu_available(check_device=False):
  import torch_xla.core.xla_model as xm
  import torch_xla.debug.metrics as met



class      __SCREAMING_SNAKE_CASE    (						_UpperCamelCase   ):







  '''simple docstring'''


  def __init__(		self   :       Optional[Any]					,   *__a   :       Tuple					,   __a   :       Union[str, Any]=None					,   __a   :       str=None					,   **__a   :       Optional[Any]	)		->				Tuple:
        super().__init__(*__a					,   **__a	)
        _UpperCamelCase      :	List[Any]						=       eval_examples
        _UpperCamelCase      :	Tuple						=       post_process_function

  def    __SCREAMING_SNAKE_CASE				(		self   :       List[Any]					,   __a   :       Optional[Any]=None					,   __a   :       str=None					,   __a   :       Optional[int]=None					,   __a   :       str = "eval"	)		->				Tuple:
        _UpperCamelCase      :	List[Any]						=       self.eval_dataset if eval_dataset is None else eval_dataset
        _UpperCamelCase      :	Tuple						=       self.get_eval_dataloader(__a	)
        _UpperCamelCase      :	Optional[Any]						=       self.eval_examples if eval_examples is None else eval_examples

        # Temporarily disable metric computation, we will do it in the loop here.
        _UpperCamelCase      :	Dict						=       self.compute_metrics
        _UpperCamelCase      :	Optional[Any]						=       None
        _UpperCamelCase      :	int						=       self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
        _UpperCamelCase      :	List[str]						=       time.time()
        try:
              _UpperCamelCase      :	Dict						=       eval_loop(
                  __a					,   description="Evaluation"					,   prediction_loss_only=True if compute_metrics is None else None					,   ignore_keys=__a					,   metric_key_prefix=__a					,   )
        finally:
              _UpperCamelCase      :	Any						=       compute_metrics
        _UpperCamelCase      :	List[Any]						=       self.args.eval_batch_size * self.args.world_size
        if F'''{metric_key_prefix}_jit_compilation_time''' in output.metrics:
              start_time += output.metrics[F'''{metric_key_prefix}_jit_compilation_time''']
        output.metrics.update(
            speed_metrics(
                __a					,   __a					,   num_samples=output.num_samples					,   num_steps=math.ceil(output.num_samples / total_batch_size	)					,   )	)
        if self.post_process_function is not None and self.compute_metrics is not None and self.args.should_save:
              # Only the main node write the results by default
              _UpperCamelCase      :	Union[str, Any]						=       self.post_process_function(__a					,   __a					,   output.predictions	)
              _UpperCamelCase      :	Optional[int]						=       self.compute_metrics(__a	)

              # Prefix all keys with metric_key_prefix + '_'
              for key in list(metrics.keys()	):
                    if not key.startswith(F'''{metric_key_prefix}_'''	):
                          _UpperCamelCase      :	str						=       metrics.pop(__a	)
              metrics.update(output.metrics	)
        else:
              _UpperCamelCase      :	List[str]						=       output.metrics

        if self.args.should_log:
              # Only the main node log the results by default
              self.log(__a	)

        if self.args.tpu_metrics_debug or self.args.debug:
              # tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
              xm.master_print(met.metrics_report()	)

        _UpperCamelCase      :	Any						=       self.callback_handler.on_evaluate(self.args					,   self.state					,   self.control					,   __a	)
        return metrics


  def    __SCREAMING_SNAKE_CASE				(		self   :       Any					,   __a   :       int					,   __a   :       Dict					,   __a   :       List[Any]=None					,   __a   :       str = "test"	)		->				str:
        _UpperCamelCase      :	List[Any]						=       self.get_test_dataloader(__a	)

        # Temporarily disable metric computation, we will do it in the loop here.
        _UpperCamelCase      :	int						=       self.compute_metrics
        _UpperCamelCase      :	Optional[int]						=       None
        _UpperCamelCase      :	Tuple						=       self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
        _UpperCamelCase      :	List[str]						=       time.time()
        try:
              _UpperCamelCase      :	str						=       eval_loop(
                  __a					,   description="Prediction"					,   prediction_loss_only=True if compute_metrics is None else None					,   ignore_keys=__a					,   metric_key_prefix=__a					,   )
        finally:
              _UpperCamelCase      :	Optional[Any]						=       compute_metrics
        _UpperCamelCase      :	Any						=       self.args.eval_batch_size * self.args.world_size
        if F'''{metric_key_prefix}_jit_compilation_time''' in output.metrics:
              start_time += output.metrics[F'''{metric_key_prefix}_jit_compilation_time''']
        output.metrics.update(
            speed_metrics(
                __a					,   __a					,   num_samples=output.num_samples					,   num_steps=math.ceil(output.num_samples / total_batch_size	)					,   )	)

        if self.post_process_function is None or self.compute_metrics is None:
              return output

        _UpperCamelCase      :	Tuple						=       self.post_process_function(__a					,   __a					,   output.predictions					,   "predict"	)
        _UpperCamelCase      :	Dict						=       self.compute_metrics(__a	)

        # Prefix all keys with metric_key_prefix + '_'
        for key in list(metrics.keys()	):
              if not key.startswith(F'''{metric_key_prefix}_'''	):
                    _UpperCamelCase      :	Tuple						=       metrics.pop(__a	)
        metrics.update(output.metrics	)
        return PredictionOutput(predictions=predictions.predictions					,   label_ids=predictions.label_ids					,   metrics=__a	)







"""simple docstring"""


from dataclasses import dataclass, field
from typing import ClassVar, Dict

from ..features import Features, Value
from .base import TaskTemplate



@dataclass(frozen=_UpperCamelCase   )
class      __SCREAMING_SNAKE_CASE    (						_UpperCamelCase   ):







  '''simple docstring'''


  SCREAMING_SNAKE_CASE__							:str	     =					field(default="summarization"		,  metadata={"include_in_asdict_even_if_is_default": True}   )
  SCREAMING_SNAKE_CASE__							:ClassVar[Features]	     =					Features({"text": Value("string"   )}   )
  SCREAMING_SNAKE_CASE__							:ClassVar[Features]	     =					Features({"summary": Value("string"   )}   )
  SCREAMING_SNAKE_CASE__							:str	     =					"text"
  SCREAMING_SNAKE_CASE__							:str	     =					"summary"

  @property
  def    __SCREAMING_SNAKE_CASE				(		self   :       List[Any]	)		->				Dict[str, str]:
        return {self.text_column: "text", self.summary_column: "summary"}
