


'''simple docstring'''






import math
import time
from typing import Dict, List, Optional

from torch.utils.data import Dataset

from transformers import SeqaSeqTrainer, 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__							(				a__    ):





							'''simple docstring'''






							def __init__(						self   ,       *lowerCamelCase_   ,       lowerCamelCase_=None   ,       lowerCamelCase_=None   ,       **lowerCamelCase_  )			->				List[Any]:
														super().__init__(*lowerCamelCase_   ,       **lowerCamelCase_  )
														lowerCAmelCase__			  =      eval_examples
														lowerCAmelCase__			  =      post_process_function







							def     __SCREAMING_SNAKE_CASE				(						self   ,       lowerCamelCase_ = None   ,       lowerCamelCase_=None   ,       lowerCamelCase_ = None   ,       lowerCamelCase_ = "eval"   ,       **lowerCamelCase_   ,       )			->				Dict[str, float]:
														lowerCAmelCase__			  =      gen_kwargs.copy()
														lowerCAmelCase__			  =      (
														    gen_kwargs['''max_length'''] if gen_kwargs.get('''max_length'''  ) is not None else self.args.generation_max_length
														)
														lowerCAmelCase__			  =      (
														    gen_kwargs['''num_beams'''] if gen_kwargs.get('''num_beams'''  ) is not None else self.args.generation_num_beams
														)
														lowerCAmelCase__			  =      gen_kwargs

														lowerCAmelCase__			  =      self.eval_dataset if eval_dataset is None else eval_dataset
														lowerCAmelCase__			  =      self.get_eval_dataloader(lowerCamelCase_  )
														lowerCAmelCase__			  =      self.eval_examples if eval_examples is None else eval_examples

														# Temporarily disable metric computation, we will do it in the loop here.
														lowerCAmelCase__			  =      self.compute_metrics
														lowerCAmelCase__			  =      None
														lowerCAmelCase__			  =      time.time()
														lowerCAmelCase__			  =      self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
														try:
																					lowerCAmelCase__			  =      eval_loop(
																					    lowerCamelCase_   ,       description='''Evaluation'''   ,       prediction_loss_only=True if compute_metrics is None else None   ,       ignore_keys=lowerCamelCase_   ,       metric_key_prefix=lowerCamelCase_   ,       )
														finally:
																					lowerCAmelCase__			  =      compute_metrics
														lowerCAmelCase__			  =      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
																					lowerCAmelCase__			  =      self.post_process_function(lowerCamelCase_   ,       lowerCamelCase_   ,       lowerCamelCase_  )
																					lowerCAmelCase__			  =      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}_"""  ):
																																			lowerCAmelCase__			  =      metrics.pop(lowerCamelCase_  )

																					metrics.update(output.metrics  )
														else:
																					lowerCAmelCase__			  =      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()  )

														lowerCAmelCase__			  =      self.callback_handler.on_evaluate(self.args   ,       self.state   ,       self.control   ,       lowerCamelCase_  )
														return metrics







							def     __SCREAMING_SNAKE_CASE				(						self   ,       lowerCamelCase_   ,       lowerCamelCase_   ,       lowerCamelCase_=None   ,       lowerCamelCase_ = "test"   ,       **lowerCamelCase_  )			->				int:
														lowerCAmelCase__			  =      gen_kwargs.copy()

														lowerCAmelCase__			  =      self.get_test_dataloader(lowerCamelCase_  )

														# Temporarily disable metric computation, we will do it in the loop here.
														lowerCAmelCase__			  =      self.compute_metrics
														lowerCAmelCase__			  =      None
														lowerCAmelCase__			  =      time.time()
														lowerCAmelCase__			  =      self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
														try:
																					lowerCAmelCase__			  =      eval_loop(
																					    lowerCamelCase_   ,       description='''Prediction'''   ,       prediction_loss_only=True if compute_metrics is None else None   ,       ignore_keys=lowerCamelCase_   ,       metric_key_prefix=lowerCamelCase_   ,       )
														finally:
																					lowerCAmelCase__			  =      compute_metrics

														lowerCAmelCase__			  =      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

														lowerCAmelCase__			  =      self.post_process_function(lowerCamelCase_   ,       lowerCamelCase_   ,       lowerCamelCase_   ,       '''predict'''  )
														lowerCAmelCase__			  =      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}_"""  ):
																												lowerCAmelCase__			  =      metrics.pop(lowerCamelCase_  )
														metrics.update(output.metrics  )
														return PredictionOutput(predictions=predictions.predictions   ,       label_ids=predictions.label_ids   ,       metrics=lowerCamelCase_  )







# We ignore warnings about stepping the scheduler since we step it ourselves during gradient accumulation

import warnings

from .state import AcceleratorState, GradientState


warnings.filterwarnings("ignore", category=UserWarning, module="torch.optim.lr_scheduler")






class 						lowercase_   :






      '''simple docstring'''




      def __init__( self :		Optional[int]					,   __UpperCAmelCase :		str					,   __UpperCAmelCase :		str					,   __UpperCAmelCase :		bool = True					,   __UpperCAmelCase :		bool = False    )       ->Tuple:




       """simple docstring"""


       a						    = scheduler
       a						    = optimizers if isinstance(__UpperCAmelCase					,   (list, tuple)    ) else [optimizers]
       a						    = split_batches
       a						    = step_with_optimizer
       a						    = GradientState()







      def  __lowerCAmelCase   ( self :		Tuple					,   *__UpperCAmelCase :		List[Any]					,   **__UpperCAmelCase :		Optional[Any]    )       ->int:




       """simple docstring"""


       if not self.step_with_optimizer:
        # No link between scheduler and optimizer -> just step
        self.scheduler.step(*__UpperCAmelCase					,   **__UpperCAmelCase    )
        return

       # Otherwise, first make sure the optimizer was stepped.
       if not self.gradient_state.sync_gradients:
        if self.gradient_state.adjust_scheduler:
         self.scheduler._step_count += 1
        return

       for opt in self.optimizers:
        if opt.step_was_skipped:
         return
       if self.split_batches:
        # Split batches -> the training dataloader batch size is not changed so one step per training step
        self.scheduler.step(*__UpperCAmelCase					,   **__UpperCAmelCase    )
       else:
        # Otherwise the training dataloader batch size was multiplied by `num_processes`, so we need to do
        # num_processes steps per training step
        a						    = AcceleratorState().num_processes
        for _ in range(__UpperCAmelCase    ):
         # Special case when using OneCycle and `drop_last` was not used
         if hasattr(self.scheduler					,   '''total_steps'''    ):
          if self.scheduler._step_count <= self.scheduler.total_steps:
           self.scheduler.step(*__UpperCAmelCase					,   **__UpperCAmelCase    )
         else:
          self.scheduler.step(*__UpperCAmelCase					,   **__UpperCAmelCase    )







      def  __lowerCAmelCase   ( self :		Optional[int]    )       ->List[str]:




       """simple docstring"""


       return self.scheduler.get_last_lr()







      def  __lowerCAmelCase   ( self :		Optional[Any]    )       ->Any:




       """simple docstring"""


       return self.scheduler.state_dict()







      def  __lowerCAmelCase   ( self :		Union[str, Any]					,   __UpperCAmelCase :		Any    )       ->Tuple:




       """simple docstring"""


       self.scheduler.load_state_dict(__UpperCAmelCase    )







      def  __lowerCAmelCase   ( self :		Dict    )       ->Optional[int]:




       """simple docstring"""


       return self.scheduler.get_lr()







      def  __lowerCAmelCase   ( self :		int					,   *__UpperCAmelCase :		List[str]					,   **__UpperCAmelCase :		Optional[Any]    )       ->List[str]:




       """simple docstring"""


       return self.scheduler.print_lr(*__UpperCAmelCase					,   **__UpperCAmelCase    )


