



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







							def __init__(							self					:       Optional[Any]	,*_UpperCAmelCase					:       Dict	,_UpperCAmelCase					:       Dict=None	,_UpperCAmelCase					:       Any=None	,**_UpperCAmelCase					:       List[str]		):
									super().__init__(*_UpperCAmelCase	,**_UpperCAmelCase		)
									_a     :      Optional[Any]        =							eval_examples
									_a     :      Dict        =							post_process_function







							def 							__lowercase			(							self					:       List[str]	,_UpperCAmelCase					:       Optional[int]=None	,_UpperCAmelCase					:       List[Any]=None	,_UpperCAmelCase					:       int=None	,_UpperCAmelCase					:       str = "eval"		):
									_a     :      Tuple        =							self.eval_dataset if eval_dataset is None else eval_dataset
									_a     :      Any        =							self.get_eval_dataloader(_UpperCAmelCase		)
									_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     :      int        =							self.compute_metrics
									_a     :      Tuple        =							None
									_a     :      Tuple        =							self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
									_a     :      List[Any]        =							time.time()
									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     :      Optional[int]        =							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 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(_UpperCAmelCase	,_UpperCAmelCase	,output.predictions		)
											_a     :      int        =							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[Any]        =							metrics.pop(_UpperCAmelCase		)
											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(_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     :      Union[str, Any]        =							self.callback_handler.on_evaluate(self.args	,self.state	,self.control	,_UpperCAmelCase		)
									return metrics







							def 							__lowercase			(							self					:       Optional[int]	,_UpperCAmelCase					:       Tuple	,_UpperCAmelCase					:       Dict	,_UpperCAmelCase					:       Dict=None	,_UpperCAmelCase					:       str = "test"		):
									_a     :      List[Any]        =							self.get_test_dataloader(_UpperCAmelCase		)

									# Temporarily disable metric computation, we will do it in the loop here.
									_a     :      Dict        =							self.compute_metrics
									_a     :      Tuple        =							None
									_a     :      Union[str, Any]        =							self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
									_a     :      Optional[int]        =							time.time()
									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[str]        =							compute_metrics
									_a     :      List[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 None or self.compute_metrics is None:
											return output

									_a     :      Optional[int]        =							self.post_process_function(_UpperCAmelCase	,_UpperCAmelCase	,output.predictions	,'predict'		)
									_a     :      Tuple        =							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     :      Optional[int]        =							metrics.pop(_UpperCAmelCase		)
									metrics.update(output.metrics		)
									return PredictionOutput(predictions=predictions.predictions	,label_ids=predictions.label_ids	,metrics=_UpperCAmelCase		)





'''simple docstring'''

from abc import ABC, abstractmethod
from argparse import ArgumentParser







class 				_lowercase     (				UpperCAmelCase__			):


 '''simple docstring'''






 @staticmethod
 @abstractmethod
 def a					(      SCREAMING_SNAKE_CASE__   :	ArgumentParser							)  ->    Tuple:
  raise NotImplementedError()





 @abstractmethod
 def a					(      self   :	int							)  ->    Union[str, Any]:
  raise NotImplementedError()






