





'''simple docstring'''

import logging
import os

import quant_trainer
import torch
from torch.utils.data import DataLoader

from transformers import Trainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput


lowercase_					= logging.getLogger(__name__)

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 ,		*A ,		A=None ,		A=None ,		A=None ,		**A       )		->   Any:
        super().__init__(*_A ,		**_A       )
        _SCREAMING_SNAKE_CASE									=			eval_examples
        _SCREAMING_SNAKE_CASE									=			post_process_function
        _SCREAMING_SNAKE_CASE									=			quant_trainer_args
        _SCREAMING_SNAKE_CASE									=			128  # default number of calibration samples


    def 			snake_case_(		self ,		A=None       )		->   Dict:






        if calib_dataset is None and self.calib_dataset is None:
            raise ValueError("""Trainer: calibration requires an calib_dataset."""       )
        _SCREAMING_SNAKE_CASE									=			calib_dataset if calib_dataset is not None else self.calib_dataset

        _SCREAMING_SNAKE_CASE									=			self._remove_unused_columns(_A ,		description="""Calibration"""       )

        return DataLoader(
            _A ,		batch_size=self.args.eval_batch_size ,		collate_fn=self.data_collator ,		drop_last=self.args.dataloader_drop_last ,		num_workers=self.args.dataloader_num_workers ,		pin_memory=self.args.dataloader_pin_memory ,		shuffle=_A ,		)


    def 			snake_case_(		self ,		A=None       )		->   Tuple:
        _SCREAMING_SNAKE_CASE									=			self.train_dataset if calib_dataset is None else calib_dataset
        _SCREAMING_SNAKE_CASE									=			self.get_calib_dataloader(_A       )

        _SCREAMING_SNAKE_CASE									=			self.model
        quant_trainer.configure_model(_A ,		self.quant_trainer_args ,		calib=_A       )
        model.eval()
        quant_trainer.enable_calibration(_A       )

        logger.info("""***** Running calibration *****"""       )
        logger.info(f'  Num examples = {self.calib_num}'       )
        logger.info(f'  Batch size = {calib_dataloader.batch_size}'       )

        for step, inputs in enumerate(_A       ):
            # Prediction step
            _SCREAMING_SNAKE_CASE									=			self.prediction_step(_A ,		_A ,		prediction_loss_only=_A       )
            if (step + 1) * calib_dataloader.batch_size >= self.calib_num:
                break

        quant_trainer.finish_calibration(_A ,		self.quant_trainer_args       )
        _SCREAMING_SNAKE_CASE									=			model


    def 			snake_case_(		self ,		A=None ,		A=None ,		A=None ,		A = "eval"       )		->   Any:
        _SCREAMING_SNAKE_CASE									=			self.eval_dataset if eval_dataset is None else eval_dataset
        _SCREAMING_SNAKE_CASE									=			self.get_eval_dataloader(_A       )
        _SCREAMING_SNAKE_CASE									=			self.eval_examples if eval_examples is None else eval_examples

        # Temporarily disable metric computation, we will do it in the loop here.
        _SCREAMING_SNAKE_CASE									=			self.compute_metrics
        _SCREAMING_SNAKE_CASE									=			None
        _SCREAMING_SNAKE_CASE									=			self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
        try:
            _SCREAMING_SNAKE_CASE									=			eval_loop(
                _A ,		description="""Evaluation""" ,		prediction_loss_only=True if compute_metrics is None else None ,		ignore_keys=_A ,		)
        finally:
            _SCREAMING_SNAKE_CASE									=			compute_metrics

        if self.post_process_function is not None and self.compute_metrics is not None:
            _SCREAMING_SNAKE_CASE									=			self.post_process_function(_A ,		_A ,		output.predictions       )
            _SCREAMING_SNAKE_CASE									=			self.compute_metrics(_A       )

            # Prefix all keys with metric_key_prefix + '_'
            for key in list(metrics.keys()       ):
                if not key.startswith(f'{metric_key_prefix}_'       ):
                    _SCREAMING_SNAKE_CASE									=			metrics.pop(_A       )

            self.log(_A       )
        else:
            _SCREAMING_SNAKE_CASE									=			{}

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

        _SCREAMING_SNAKE_CASE									=			self.callback_handler.on_evaluate(self.args ,		self.state ,		self.control ,		_A       )
        return metrics


    def 			snake_case_(		self ,		A ,		A ,		A=None ,		A = "test"       )		->   Optional[int]:
        _SCREAMING_SNAKE_CASE									=			self.get_test_dataloader(_A       )

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

        if self.post_process_function is None or self.compute_metrics is None:
            return output

        _SCREAMING_SNAKE_CASE									=			self.post_process_function(_A ,		_A ,		output.predictions ,		"""predict"""       )
        _SCREAMING_SNAKE_CASE									=			self.compute_metrics(_A       )

        # Prefix all keys with metric_key_prefix + '_'
        for key in list(metrics.keys()       ):
            if not key.startswith(f'{metric_key_prefix}_'       ):
                _SCREAMING_SNAKE_CASE									=			metrics.pop(_A       )

        return PredictionOutput(predictions=predictions.predictions ,		label_ids=predictions.label_ids ,		metrics=_A       )





    def 			snake_case_(		self ,		A="./"       )		->   Union[str, Any]:
        _SCREAMING_SNAKE_CASE									=			self.eval_dataset
        _SCREAMING_SNAKE_CASE									=			self.get_eval_dataloader(_A       )

        _SCREAMING_SNAKE_CASE									=			next(iter(_A       )       )

        # saving device - to make it consistent
        _SCREAMING_SNAKE_CASE									=			torch.device("""cuda""" if torch.cuda.is_available() else """cpu"""       )

        # convert to tuple
        _SCREAMING_SNAKE_CASE									=			tuple(v.to(_A       ) for k, v in batch.items()       )

        logger.info("""Converting model to be onnx compatible"""       )
        from pytorch_quantization.nn import TensorQuantizer

        _SCREAMING_SNAKE_CASE									=			True

        _SCREAMING_SNAKE_CASE									=			self.model.to(_A       )

        model.eval()
        model.float()

        _SCREAMING_SNAKE_CASE									=			model.module if hasattr(_A ,		"""module"""       ) else model
        quant_trainer.configure_model(_A ,		self.quant_trainer_args       )

        _SCREAMING_SNAKE_CASE									=			os.path.join(_A ,		"""model.onnx"""       )
        logger.info(f'exporting model to {output_model_file}'       )

        _SCREAMING_SNAKE_CASE									=			{0: "batch_size", 1: "seq_len"}

        torch.onnx.export(
            _A ,		_A ,		_A ,		export_params=_A ,		opset_version=13 ,		do_constant_folding=_A ,		input_names=["""input_ids""", """attention_mask""", """token_type_ids"""] ,		output_names=["""output_start_logits""", """output_end_logits"""] ,		dynamic_axes={
                """input_ids""": axes,
                """attention_mask""": axes,
                """token_type_ids""": axes,
                """output_start_logits""": axes,
                """output_end_logits""": axes,
            } ,		verbose=_A ,		)
        logger.info("""onnx export finished"""       )




from math import factorial, radians




def       _snake_case	(   lowerCAmelCase							:   float   , lowerCAmelCase							:   int = 1_8   , lowerCAmelCase							:   int = 1_0	):
		"""simple docstring"""







		SCREAMING_SNAKE_CASE_  :		List[str]				      =    angle_in_degrees - ((angle_in_degrees // 360.0) * 360.0)

		# Converting from degrees to radians
		SCREAMING_SNAKE_CASE_  :		Tuple				      =    radians(lowerCAmelCase	)

		SCREAMING_SNAKE_CASE_  :		Optional[Any]				      =    angle_in_radians
		SCREAMING_SNAKE_CASE_  :		List[str]				      =    3
		SCREAMING_SNAKE_CASE_  :		str				      =    -1

		for _ in range(lowerCAmelCase	):
				result += (b * (angle_in_radians**a)) / factorial(lowerCAmelCase	)

				SCREAMING_SNAKE_CASE_  :		Dict				      =    -b  # One positive term and the next will be negative and so on...
				a += 2  # Increased by 2 for every term.

		return round(lowerCAmelCase   , lowerCAmelCase	)


if __name__ == "__main__":
		__import__('''doctest''').testmod()
