





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 					A__   (       __snake_case			):





      def __init__( self       ,							*A_       ,							A_=None       ,							A_=None       ,							**A_   ):
       '''simple docstring'''




       super().__init__(*A_       ,							**A_   )
       UpperCamelCase	:					str       	=   eval_examples
       UpperCamelCase	:					int       	=   post_process_function





      def        __UpperCamelCase( self       ,							A_=None       ,							A_=None       ,							A_=None       ,							A_ = "eval"   ):
       '''simple docstring'''




       UpperCamelCase	:					Union[str, Any]       	=   self.eval_dataset if eval_dataset is None else eval_dataset
       UpperCamelCase	:					str       	=   self.get_eval_dataloader(A_   )
       UpperCamelCase	:					List[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	:					Dict       	=   None
       UpperCamelCase	:					Optional[Any]       	=   self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
       UpperCamelCase	:					List[str]       	=   time.time()
       try:
        UpperCamelCase	:					Optional[int]       	=   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	:					Optional[Any]       	=   compute_metrics
       UpperCamelCase	:					Tuple       	=   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	:					Tuple       	=   self.post_process_function(A_       ,							A_       ,							output.predictions   )
        UpperCamelCase	:					Tuple       	=   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   )
       else:
        UpperCamelCase	:					Union[str, Any]       	=   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	:					Tuple       	=   self.callback_handler.on_evaluate(self.args       ,							self.state       ,							self.control       ,							A_   )
       return metrics





      def        __UpperCamelCase( self       ,							A_       ,							A_       ,							A_=None       ,							A_ = "test"   ):
       '''simple docstring'''




       UpperCamelCase	:					Any       	=   self.get_test_dataloader(A_   )

       # Temporarily disable metric computation, we will do it in the loop here.
       UpperCamelCase	:					Any       	=   self.compute_metrics
       UpperCamelCase	:					List[Any]       	=   None
       UpperCamelCase	:					str       	=   self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
       UpperCamelCase	:					Tuple       	=   time.time()
       try:
        UpperCamelCase	:					List[Any]       	=   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	:					List[str]       	=   compute_metrics
       UpperCamelCase	:					Union[str, 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	:					List[str]       	=   self.post_process_function(A_       ,							A_       ,							output.predictions       ,							"predict"   )
       UpperCamelCase	:					List[str]       	=   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   )
       return PredictionOutput(predictions=predictions.predictions       ,							label_ids=predictions.label_ids       ,							metrics=A_   )

"""simple docstring"""
import os
from typing import Dict, List, Tuple, TypeVar, Union


lowercase__					       =	TypeVar('T')

lowercase__					       =	Union[List[T], Tuple[T, ...]]
lowercase__					       =	Union[T, List[T], Dict[str, T]]
lowercase__					       =	Union[str, bytes, os.PathLike]


