

"""simple docstring"""




import math
from enum import Enum
from typing import Optional, Union

from torch.optim import Optimizer
from torch.optim.lr_scheduler import LambdaLR

from .utils import logging


__UpperCAmelCase			=					logging.get_logger(__name__)







class   __lowercase      (     __lowerCamelCase       ):
  snake_case_												=							"""linear"""
  snake_case_												=							"""cosine"""
  snake_case_												=							"""cosine_with_restarts"""
  snake_case_												=							"""polynomial"""
  snake_case_												=							"""constant"""
  snake_case_												=							"""constant_with_warmup"""
  snake_case_												=							"""piecewise_constant"""







def 	lowerCAmelCase    (       __UpperCamelCase				,     __UpperCamelCase = -1   ):
 '''simple docstring'''





 return LambdaLR(__UpperCamelCase				,     lambda __UpperCamelCase   : 1				,     last_epoch=__UpperCamelCase   )
def 	lowerCAmelCase    (       __UpperCamelCase				,     __UpperCamelCase				,     __UpperCamelCase = -1   ):
 '''simple docstring'''





 def lr_lambda(__UpperCamelCase   ):
  if current_step < num_warmup_steps:
   return float(__UpperCamelCase   ) / float(max(1.0				,     __UpperCamelCase   )   )
  return 1.0

 return LambdaLR(__UpperCamelCase				,     __UpperCamelCase				,     last_epoch=__UpperCamelCase   )
def 	lowerCAmelCase    (       __UpperCamelCase				,     __UpperCamelCase				,     __UpperCamelCase = -1   ):
 '''simple docstring'''





 UpperCAmelCase__       :	Dict        =       {}
 UpperCAmelCase__       :	Optional[int]        =       step_rules.split(""","""   )
 for rule_str in rule_list[:-1]:
  UpperCAmelCase__ ,   UpperCAmelCase__       :	int        =       rule_str.split(""":"""   )
  UpperCAmelCase__       :	Any        =       int(__UpperCamelCase   )
  UpperCAmelCase__       :	Dict        =       float(__UpperCamelCase   )
  UpperCAmelCase__       :	int        =       value
 UpperCAmelCase__       :	Dict        =       float(rule_list[-1]   )

 def create_rules_function(__UpperCamelCase				,     __UpperCamelCase   ):
  def rule_func(__UpperCamelCase   ) -> float:
   UpperCAmelCase__       :	Optional[Any]        =       sorted(rules_dict.keys()   )
   for i, sorted_step in enumerate(__UpperCamelCase   ):
    if steps < sorted_step:
     return rules_dict[sorted_steps[i]]
   return last_lr_multiple

  return rule_func

 UpperCAmelCase__       :	Dict        =       create_rules_function(__UpperCamelCase				,     __UpperCamelCase   )

 return LambdaLR(__UpperCamelCase				,     __UpperCamelCase				,     last_epoch=__UpperCamelCase   )
def 	lowerCAmelCase    (       __UpperCamelCase				,     __UpperCamelCase				,     __UpperCamelCase				,     __UpperCamelCase=-1   ):
 '''simple docstring'''





 def lr_lambda(__UpperCamelCase   ):
  if current_step < num_warmup_steps:
   return float(__UpperCamelCase   ) / float(max(1				,     __UpperCamelCase   )   )
  return max(
      0.0				,     float(num_training_steps - current_step   ) / float(max(1				,     num_training_steps - num_warmup_steps   )   )   )

 return LambdaLR(__UpperCamelCase				,     __UpperCamelCase				,     __UpperCamelCase   )
def 	lowerCAmelCase    (       __UpperCamelCase				,     __UpperCamelCase				,     __UpperCamelCase				,     __UpperCamelCase = 0.5				,     __UpperCamelCase = -1   ):
 '''simple docstring'''





 def lr_lambda(__UpperCamelCase   ):
  if current_step < num_warmup_steps:
   return float(__UpperCamelCase   ) / float(max(1				,     __UpperCamelCase   )   )
  UpperCAmelCase__       :	Optional[Any]        =       float(current_step - num_warmup_steps   ) / float(max(1				,     num_training_steps - num_warmup_steps   )   )
  return max(0.0				,     0.5 * (1.0 + math.cos(math.pi * float(__UpperCamelCase   ) * 2.0 * progress   ))   )

 return LambdaLR(__UpperCamelCase				,     __UpperCamelCase				,     __UpperCamelCase   )
def 	lowerCAmelCase    (       __UpperCamelCase				,     __UpperCamelCase				,     __UpperCamelCase				,     __UpperCamelCase = 1				,     __UpperCamelCase = -1   ):
 '''simple docstring'''





 def lr_lambda(__UpperCamelCase   ):
  if current_step < num_warmup_steps:
   return float(__UpperCamelCase   ) / float(max(1				,     __UpperCamelCase   )   )
  UpperCAmelCase__       :	Optional[Any]        =       float(current_step - num_warmup_steps   ) / float(max(1				,     num_training_steps - num_warmup_steps   )   )
  if progress >= 1.0:
   return 0.0
  return max(0.0				,     0.5 * (1.0 + math.cos(math.pi * ((float(__UpperCamelCase   ) * progress) % 1.0)   ))   )

 return LambdaLR(__UpperCamelCase				,     __UpperCamelCase				,     __UpperCamelCase   )
def 	lowerCAmelCase    (       __UpperCamelCase				,     __UpperCamelCase				,     __UpperCamelCase				,     __UpperCamelCase=1e-7				,     __UpperCamelCase=1.0				,     __UpperCamelCase=-1   ):
 '''simple docstring'''





 UpperCAmelCase__       :	List[Any]        =       optimizer.defaults["""lr"""]
 if not (lr_init > lr_end):
  raise ValueError(F"lr_end ({lr_end}) must be be smaller than initial lr ({lr_init})"   )

 def lr_lambda(__UpperCamelCase   ):
  if current_step < num_warmup_steps:
   return float(__UpperCamelCase   ) / float(max(1				,     __UpperCamelCase   )   )
  elif current_step > num_training_steps:
   return lr_end / lr_init  # as LambdaLR multiplies by lr_init
  else:
   UpperCAmelCase__       :	List[str]        =       lr_init - lr_end
   UpperCAmelCase__       :	List[Any]        =       num_training_steps - num_warmup_steps
   UpperCAmelCase__       :	Union[str, Any]        =       1 - (current_step - num_warmup_steps) / decay_steps
   UpperCAmelCase__       :	Any        =       lr_range * pct_remaining**power + lr_end
   return decay / lr_init  # as LambdaLR multiplies by lr_init

 return LambdaLR(__UpperCamelCase				,     __UpperCamelCase				,     __UpperCamelCase   )


__UpperCAmelCase			=					{
    SchedulerType.LINEAR: get_linear_schedule_with_warmup,
    SchedulerType.COSINE: get_cosine_schedule_with_warmup,
    SchedulerType.COSINE_WITH_RESTARTS: get_cosine_with_hard_restarts_schedule_with_warmup,
    SchedulerType.POLYNOMIAL: get_polynomial_decay_schedule_with_warmup,
    SchedulerType.CONSTANT: get_constant_schedule,
    SchedulerType.CONSTANT_WITH_WARMUP: get_constant_schedule_with_warmup,
    SchedulerType.PIECEWISE_CONSTANT: get_piecewise_constant_schedule,
}




def 	lowerCAmelCase    (       __UpperCamelCase				,     __UpperCamelCase				,     __UpperCamelCase = None				,     __UpperCamelCase = None				,     __UpperCamelCase = None				,     __UpperCamelCase = 1				,     __UpperCamelCase = 1.0				,     __UpperCamelCase = -1				,     ):
 '''simple docstring'''





 UpperCAmelCase__       :	List[str]        =       SchedulerType(__UpperCamelCase   )
 UpperCAmelCase__       :	Optional[Any]        =       TYPE_TO_SCHEDULER_FUNCTION[name]
 if name == SchedulerType.CONSTANT:
  return schedule_func(__UpperCamelCase				,     last_epoch=__UpperCamelCase   )

 if name == SchedulerType.PIECEWISE_CONSTANT:
  return schedule_func(__UpperCamelCase				,     step_rules=__UpperCamelCase				,     last_epoch=__UpperCamelCase   )

 # All other schedulers require `num_warmup_steps`
 if num_warmup_steps is None:
  raise ValueError(F"{name} requires `num_warmup_steps`, please provide that argument."   )

 if name == SchedulerType.CONSTANT_WITH_WARMUP:
  return schedule_func(__UpperCamelCase				,     num_warmup_steps=__UpperCamelCase				,     last_epoch=__UpperCamelCase   )

 # All other schedulers require `num_training_steps`
 if num_training_steps is None:
  raise ValueError(F"{name} requires `num_training_steps`, please provide that argument."   )

 if name == SchedulerType.COSINE_WITH_RESTARTS:
  return schedule_func(
      __UpperCamelCase				,     num_warmup_steps=__UpperCamelCase				,     num_training_steps=__UpperCamelCase				,     num_cycles=__UpperCamelCase				,     last_epoch=__UpperCamelCase				,     )

 if name == SchedulerType.POLYNOMIAL:
  return schedule_func(
      __UpperCamelCase				,     num_warmup_steps=__UpperCamelCase				,     num_training_steps=__UpperCamelCase				,     power=__UpperCamelCase				,     last_epoch=__UpperCamelCase				,     )

 return schedule_func(
     __UpperCamelCase				,     num_warmup_steps=__UpperCamelCase				,     num_training_steps=__UpperCamelCase				,     last_epoch=__UpperCamelCase   )













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 							snake_case_       (    __UpperCamelCase      ):

						"""simple docstring"""



						def __init__(self:					str ,					*__UpperCAmelCase:					Optional[int] ,					__UpperCAmelCase:					Optional[int]=None ,					__UpperCAmelCase:					Dict=None ,					**__UpperCAmelCase:					Tuple   )    ->     Dict:




											'''simple docstring'''



											super().__init__(*__UpperCAmelCase ,					**__UpperCAmelCase   )
											__a					:						str      =					eval_examples
											__a					:						List[Any]      =					post_process_function



						def        UpperCAmelCase__					(self:					Optional[int] ,					__UpperCAmelCase:					Optional[Dataset] = None ,					__UpperCAmelCase:					Optional[int]=None ,					__UpperCAmelCase:					Optional[List[str]] = None ,					__UpperCAmelCase:					str = "eval" ,					**__UpperCAmelCase:					Any ,					)    ->     Dict[str, float]:




											'''simple docstring'''



											__a					:						Optional[int]      =					gen_kwargs.copy()
											__a					:						Union[str, Any]      =					(
											    gen_kwargs["max_length"] if gen_kwargs.get("max_length"   ) is not None else self.args.generation_max_length
											)
											__a					:						Dict      =					(
											    gen_kwargs["num_beams"] if gen_kwargs.get("num_beams"   ) is not None else self.args.generation_num_beams
											)
											__a					:						Optional[Any]      =					gen_kwargs

											__a					:						Union[str, Any]      =					self.eval_dataset if eval_dataset is None else eval_dataset
											__a					:						Optional[int]      =					self.get_eval_dataloader(__UpperCAmelCase   )
											__a					:						Tuple      =					self.eval_examples if eval_examples is None else eval_examples

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

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

																metrics.update(output.metrics   )
											else:
																__a					:						Optional[int]      =					output.metrics

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

											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					:						Tuple      =					self.callback_handler.on_evaluate(self.args ,					self.state ,					self.control ,					__UpperCAmelCase   )
											return metrics



						def        UpperCAmelCase__					(self:					Dict ,					__UpperCAmelCase:					List[str] ,					__UpperCAmelCase:					List[str] ,					__UpperCAmelCase:					Union[str, Any]=None ,					__UpperCAmelCase:					str = "test" ,					**__UpperCAmelCase:					List[Any]   )    ->     Union[str, Any]:




											'''simple docstring'''



											__a					:						Any      =					gen_kwargs.copy()

											__a					:						Tuple      =					self.get_test_dataloader(__UpperCAmelCase   )

											# Temporarily disable metric computation, we will do it in the loop here.
											__a					:						Optional[Any]      =					self.compute_metrics
											__a					:						int      =					None
											__a					:						List[Any]      =					time.time()
											__a					:						Dict      =					self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
											try:
																__a					:						Optional[Any]      =					eval_loop(
																    __UpperCAmelCase ,					description="Prediction" ,					prediction_loss_only=True if compute_metrics is None else None ,					ignore_keys=__UpperCAmelCase ,					metric_key_prefix=__UpperCAmelCase ,					)
											finally:
																__a					:						List[Any]      =					compute_metrics

											__a					:						Optional[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(
											        __UpperCAmelCase ,					__UpperCAmelCase ,					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					:						Optional[int]      =					self.post_process_function(__UpperCAmelCase ,					__UpperCAmelCase ,					__UpperCAmelCase ,					"predict"   )
											__a					:						Dict      =					self.compute_metrics(__UpperCAmelCase   )

											# Prefix all keys with metric_key_prefix + '_'
											for key in list(metrics.keys()   ):
																if not key.startswith(f'{metric_key_prefix}_'   ):
																					__a					:						List[str]      =					metrics.pop(__UpperCAmelCase   )
											metrics.update(output.metrics   )
											return PredictionOutput(predictions=predictions.predictions ,					label_ids=predictions.label_ids ,					metrics=__UpperCAmelCase   )





