
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 			lowercase_		(      A       ):


       def __init__(    self						,       *__A						,       __A=None						,       __A=None						,       __A=None						,       **__A				)						->    Optional[int]:
              super().__init__(*__A						,       **__A				)
              SCREAMING_SNAKE_CASE_							:				Optional[Any]		=eval_examples
              SCREAMING_SNAKE_CASE_							:				List[Any]		=post_process_function
              SCREAMING_SNAKE_CASE_							:				List[Any]		=quant_trainer_args
              SCREAMING_SNAKE_CASE_							:				Tuple		=128  # default number of calibration samples


       def 	_snake_case				(    self						,       __A=None				)						->    str:







              if calib_dataset is None and self.calib_dataset is None:
                     raise ValueError('''Trainer: calibration requires an calib_dataset.'''				)
              SCREAMING_SNAKE_CASE_							:				List[str]		=calib_dataset if calib_dataset is not None else self.calib_dataset

              SCREAMING_SNAKE_CASE_							:				Tuple		=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_							:				Optional[int]		=self.train_dataset if calib_dataset is None else calib_dataset
              SCREAMING_SNAKE_CASE_							:				Any		=self.get_calib_dataloader(__A				)

              SCREAMING_SNAKE_CASE_							:				List[Any]		=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_							:				List[str]		=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_							:				List[Any]		=model


       def 	_snake_case				(    self						,       __A=None						,       __A=None						,       __A=None						,       __A = "eval"				)						->    str:
              SCREAMING_SNAKE_CASE_							:				Optional[int]		=self.eval_dataset if eval_dataset is None else eval_dataset
              SCREAMING_SNAKE_CASE_							:				Union[str, Any]		=self.get_eval_dataloader(__A				)
              SCREAMING_SNAKE_CASE_							:				Dict		=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_							:				int		=self.compute_metrics
              SCREAMING_SNAKE_CASE_							:				Optional[int]		=None
              SCREAMING_SNAKE_CASE_							:				List[Any]		=self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
              try:
                     SCREAMING_SNAKE_CASE_							:				List[str]		=eval_loop(
                         __A						,       description='''Evaluation'''						,       prediction_loss_only=True if compute_metrics is None else None						,       ignore_keys=__A						,       )
              finally:
                     SCREAMING_SNAKE_CASE_							:				List[str]		=compute_metrics

              if self.post_process_function is not None and self.compute_metrics is not None:
                     SCREAMING_SNAKE_CASE_							:				Tuple		=self.post_process_function(__A						,       __A						,       output.predictions				)
                     SCREAMING_SNAKE_CASE_							:				Optional[int]		=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_							:				List[Any]		=metrics.pop(__A				)

                     self.log(__A				)
              else:
                     SCREAMING_SNAKE_CASE_							:				Optional[int]		={}

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


       def 	_snake_case				(    self						,       __A						,       __A						,       __A=None						,       __A = "test"				)						->    int:
              SCREAMING_SNAKE_CASE_							:				str		=self.get_test_dataloader(__A				)

              # Temporarily disable metric computation, we will do it in the loop here.
              SCREAMING_SNAKE_CASE_							:				List[str]		=self.compute_metrics
              SCREAMING_SNAKE_CASE_							:				str		=None
              SCREAMING_SNAKE_CASE_							:				Dict		=self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
              try:
                     SCREAMING_SNAKE_CASE_							:				int		=eval_loop(
                         __A						,       description='''Prediction'''						,       prediction_loss_only=True if compute_metrics is None else None						,       ignore_keys=__A						,       )
              finally:
                     SCREAMING_SNAKE_CASE_							:				Tuple		=compute_metrics

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

              SCREAMING_SNAKE_CASE_							:				int		=self.post_process_function(__A						,       __A						,       output.predictions						,       '''predict'''				)
              SCREAMING_SNAKE_CASE_							:				Union[str, Any]		=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_							:				Optional[int]		=metrics.pop(__A				)

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





       def 	_snake_case				(    self						,       __A="./"				)						->    Optional[int]:
              SCREAMING_SNAKE_CASE_							:				Tuple		=self.eval_dataset
              SCREAMING_SNAKE_CASE_							:				int		=self.get_eval_dataloader(__A				)

              SCREAMING_SNAKE_CASE_							:				Union[str, Any]		=next(iter(__A				)				)

              # saving device - to make it consistent
              SCREAMING_SNAKE_CASE_							:				Optional[Any]		=torch.device('''cuda''' if torch.cuda.is_available() else '''cpu'''				)

              # convert to tuple
              SCREAMING_SNAKE_CASE_							:				Optional[int]		=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_							:				List[str]		=True

              SCREAMING_SNAKE_CASE_							:				Dict		=self.model.to(__A				)

              model.eval()
              model.float()

              SCREAMING_SNAKE_CASE_							:				int		=model.module if hasattr(__A						,       '''module'''				) else model
              quant_trainer.configure_model(__A						,       self.quant_trainer_args				)

              SCREAMING_SNAKE_CASE_							:				Any		=os.path.join(__A						,       '''model.onnx'''				)
              logger.info(F'exporting model to {output_model_file}'				)

              SCREAMING_SNAKE_CASE_							:				List[str]		={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'''				)






'''simple docstring'''



def   lowercase__(			__UpperCamelCase:					int    ):

      """simple docstring"""






      if not isinstance(__UpperCamelCase      ,__UpperCamelCase    ):
            raise TypeError('Input value must be an \'int\' type'    )

      SCREAMING_SNAKE_CASE   :      int    					=						0
      while number:
            position += 1
            number >>= 1

      return position


if __name__ == "__main__":
      import doctest

      doctest.testmod()


