


'''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 		a__    (							a__  ):





							'''simple docstring'''






							def __init__(		self ,   *lowerCamelCase_ ,   lowerCamelCase_=None ,   lowerCamelCase_=None ,   **lowerCamelCase_       )	->							int:
														super().__init__(*lowerCamelCase_ ,   **lowerCamelCase_       )
														lowerCAmelCase__	       =     eval_examples
														lowerCAmelCase__	       =     post_process_function







							def   __SCREAMING_SNAKE_CASE							(		self ,   lowerCamelCase_=None ,   lowerCamelCase_=None ,   lowerCamelCase_=None ,   lowerCamelCase_ = "eval"       )	->							str:
														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__	       =     self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
														lowerCAmelCase__	       =     time.time()
														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_ ,   output.predictions       )
																					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"       )	->							Any:
														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__	       =     self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
														lowerCAmelCase__	       =     time.time()
														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_ ,   output.predictions ,   '''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_       )



'''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 		a__    (							a__  ):





							'''simple docstring'''






							lowercase__      :       int     						=				"linear"
							lowercase__      :       Any     						=				"cosine"
							lowercase__      :       Optional[int]     						=				"cosine_with_restarts"
							lowercase__      :       Optional[Any]     						=				"polynomial"
							lowercase__      :       Tuple     						=				"constant"
							lowercase__      :       Optional[int]     						=				"constant_with_warmup"
							lowercase__      :       Optional[int]     						=				"piecewise_constant"


def  _snake_case       (					A			,   A = -1	)						->   Any:
							return LambdaLR(A			,   lambda A	: 1			,   last_epoch=A	)


def  _snake_case       (					A			,   A			,   A = -1	)						->   Optional[Any]:

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

							return LambdaLR(A			,   A			,   last_epoch=A	)


def  _snake_case       (					A			,   A			,   A = -1	)						->   Union[str, Any]:
							lowerCAmelCase__	       =     {}
							lowerCAmelCase__	       =     step_rules.split(''','''	)
							for rule_str in rule_list[:-1]:
														lowerCAmelCase__       , lowerCAmelCase__	       =     rule_str.split(''':'''	)
														lowerCAmelCase__	       =     int(A	)
														lowerCAmelCase__	       =     float(A	)
														lowerCAmelCase__	       =     value
							lowerCAmelCase__	       =     float(rule_list[-1]	)

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

														return rule_func

							lowerCAmelCase__	       =     create_rules_function(A			,   A	)

							return LambdaLR(A			,   A			,   last_epoch=A	)


def  _snake_case       (					A			,   A			,   A			,   A=-1	)						->   Optional[int]:

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

							return LambdaLR(A			,   A			,   A	)


def  _snake_case       (					A			,   A			,   A			,   A = 0.5			,   A = -1	)						->   List[str]:

							def lr_lambda(A	):
														if current_step < num_warmup_steps:
																					return float(A	) / float(max(1			,   A	)	)
														lowerCAmelCase__	       =     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(A	) * 2.0 * progress	))	)

							return LambdaLR(A			,   A			,   A	)


def  _snake_case       (					A			,   A			,   A			,   A = 1			,   A = -1	)						->   Union[str, Any]:

							def lr_lambda(A	):
														if current_step < num_warmup_steps:
																					return float(A	) / float(max(1			,   A	)	)
														lowerCAmelCase__	       =     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(A	) * progress) % 1.0)	))	)

							return LambdaLR(A			,   A			,   A	)


def  _snake_case       (					A			,   A			,   A			,   A=1E-7			,   A=1.0			,   A=-1	)						->   Union[str, Any]:
							lowerCAmelCase__	       =     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(A	):
														if current_step < num_warmup_steps:
																					return float(A	) / float(max(1			,   A	)	)
														elif current_step > num_training_steps:
																					return lr_end / lr_init  # as LambdaLR multiplies by lr_init
														else:
																					lowerCAmelCase__	       =     lr_init - lr_end
																					lowerCAmelCase__	       =     num_training_steps - num_warmup_steps
																					lowerCAmelCase__	       =     1 - (current_step - num_warmup_steps) / decay_steps
																					lowerCAmelCase__	       =     lr_range * pct_remaining**power + lr_end
																					return decay / lr_init  # as LambdaLR multiplies by lr_init

							return LambdaLR(A			,   A			,   A	)


__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  _snake_case       (					A			,   A			,   A = None			,   A = None			,   A = None			,   A = 1			,   A = 1.0			,   A = -1			,   )						->   int:
							lowerCAmelCase__	       =     SchedulerType(A	)
							lowerCAmelCase__	       =     TYPE_TO_SCHEDULER_FUNCTION[name]
							if name == SchedulerType.CONSTANT:
														return schedule_func(A			,   last_epoch=A	)

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

							# 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(A			,   num_warmup_steps=A			,   last_epoch=A	)

							# 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(
														    A			,   num_warmup_steps=A			,   num_training_steps=A			,   num_cycles=A			,   last_epoch=A			,   )

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

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