


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      lowerCAmelCase__       (				a):




     '''simple docstring'''


     def __init__(						self ,						*__lowerCamelCase ,						__lowerCamelCase=None ,						__lowerCamelCase=None ,						**__lowerCamelCase)			->				List[str]:
            super().__init__(*__lowerCamelCase ,						**__lowerCamelCase)
            _A      :       List[Any]				    = eval_examples
            _A      :       Optional[Any]				    = post_process_function


     def  _lowerCamelCase (						self ,						__lowerCamelCase=None ,						__lowerCamelCase=None ,						__lowerCamelCase=None ,						__lowerCamelCase = "eval")			->				str:
            _A      :       Union[str, Any]				    = self.eval_dataset if eval_dataset is None else eval_dataset
            _A      :       List[Any]				    = self.get_eval_dataloader(__lowerCamelCase)
            _A      :       str				    = self.eval_examples if eval_examples is None else eval_examples

            # Temporarily disable metric computation, we will do it in the loop here.
            _A      :       Optional[Any]				    = self.compute_metrics
            _A      :       Optional[Any]				    = None
            _A      :       Tuple				    = self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
            _A      :       List[str]				    = time.time()
            try:
                   _A      :       Optional[Any]				    = eval_loop(
                       __lowerCamelCase ,						description="Evaluation" ,						prediction_loss_only=True if compute_metrics is None else None ,						ignore_keys=__lowerCamelCase ,						metric_key_prefix=__lowerCamelCase ,						)
            finally:
                   _A      :       List[str]				    = compute_metrics
            _A      :       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(
                    __lowerCamelCase ,						__lowerCamelCase ,						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
                   _A      :       Tuple				    = self.post_process_function(__lowerCamelCase ,						__lowerCamelCase ,						output.predictions)
                   _A      :       List[str]				    = self.compute_metrics(__lowerCamelCase)

                   # Prefix all keys with metric_key_prefix + '_'
                   for key in list(metrics.keys()):
                          if not key.startswith(F"{metric_key_prefix}_"):
                                 _A      :       Optional[int]				    = metrics.pop(__lowerCamelCase)
                   metrics.update(output.metrics)
            else:
                   _A      :       Dict				    = output.metrics

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

            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())

            _A      :       Optional[Any]				    = self.callback_handler.on_evaluate(self.args ,						self.state ,						self.control ,						__lowerCamelCase)
            return metrics


     def  _lowerCamelCase (						self ,						__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase=None ,						__lowerCamelCase = "test")			->				Tuple:
            _A      :       Any				    = self.get_test_dataloader(__lowerCamelCase)

            # Temporarily disable metric computation, we will do it in the loop here.
            _A      :       List[Any]				    = self.compute_metrics
            _A      :       str				    = None
            _A      :       int				    = self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
            _A      :       Dict				    = time.time()
            try:
                   _A      :       Union[str, Any]				    = eval_loop(
                       __lowerCamelCase ,						description="Prediction" ,						prediction_loss_only=True if compute_metrics is None else None ,						ignore_keys=__lowerCamelCase ,						metric_key_prefix=__lowerCamelCase ,						)
            finally:
                   _A      :       Union[str, Any]				    = compute_metrics
            _A      :       str				    = 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(
                    __lowerCamelCase ,						__lowerCamelCase ,						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

            _A      :       List[str]				    = self.post_process_function(__lowerCamelCase ,						__lowerCamelCase ,						output.predictions ,						"predict")
            _A      :       Tuple				    = self.compute_metrics(__lowerCamelCase)

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




"""simple docstring"""






import os
import sys
from contextlib import contextmanager


# Windows only
if os.name == "nt":
							import ctypes
							import msvcrt  # noqa





							class      lowerCamelCase__   (     ctypes.Structure				):







												"""simple docstring"""


												__a	       =							[("""size""", ctypes.c_int), ("""visible""", ctypes.c_byte)]
def      lowerCamelCase       (     )		->       Optional[int]:

							'''simple docstring'''



							if os.name == "nt":
														__UpperCAmelCase   :						Dict      							=					CursorInfo()
														__UpperCAmelCase   :						Any      							=					ctypes.windll.kernelaa.GetStdHandle(-1_1							)
														ctypes.windll.kernelaa.GetConsoleCursorInfo(_UpperCamelCase       ,			ctypes.byref(_UpperCamelCase							)							)
														__UpperCAmelCase   :						Tuple      							=					False
														ctypes.windll.kernelaa.SetConsoleCursorInfo(_UpperCamelCase       ,			ctypes.byref(_UpperCamelCase							)							)
							elif os.name == "posix":
														sys.stdout.write("""\033[?25l"""							)
														sys.stdout.flush()
def      lowerCamelCase       (     )		->       Optional[int]:

							'''simple docstring'''



							if os.name == "nt":
														__UpperCAmelCase   :						str      							=					CursorInfo()
														__UpperCAmelCase   :						int      							=					ctypes.windll.kernelaa.GetStdHandle(-1_1							)
														ctypes.windll.kernelaa.GetConsoleCursorInfo(_UpperCamelCase       ,			ctypes.byref(_UpperCamelCase							)							)
														__UpperCAmelCase   :						Union[str, Any]      							=					True
														ctypes.windll.kernelaa.SetConsoleCursorInfo(_UpperCamelCase       ,			ctypes.byref(_UpperCamelCase							)							)
							elif os.name == "posix":
														sys.stdout.write("""\033[?25h"""							)
														sys.stdout.flush()
@contextmanager
def      lowerCamelCase       (     )		->       str:

							'''simple docstring'''



							try:
														hide_cursor()
														yield
							finally:
														show_cursor()
