

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






     '''simple docstring'''



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


     def        UpperCAmelCase_	(		self			:					Any					,      lowercase_			:					Tuple=None	)						->     List[str]:
      if calib_dataset is None and self.calib_dataset is None:
       raise ValueError('Trainer: calibration requires an calib_dataset.'	)
      UpperCAmelCase	:     List[Any]						 =			calib_dataset if calib_dataset is not None else self.calib_dataset

      UpperCAmelCase	:     str						 =			self._remove_unused_columns(lowercase_					,      description='Calibration'	)

      return DataLoader(
          lowercase_					,      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=lowercase_					,      )


     def        UpperCAmelCase_	(		self			:					Optional[int]					,      lowercase_			:					Any=None	)						->     Optional[int]:
      UpperCAmelCase	:     Any						 =			self.train_dataset if calib_dataset is None else calib_dataset
      UpperCAmelCase	:     Dict						 =			self.get_calib_dataloader(lowercase_	)

      UpperCAmelCase	:     str						 =			self.model
      quant_trainer.configure_model(lowercase_					,      self.quant_trainer_args					,      calib=lowercase_	)
      model.eval()
      quant_trainer.enable_calibration(lowercase_	)

      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(lowercase_	):
       # Prediction step
       UpperCAmelCase ,			UpperCAmelCase ,			UpperCAmelCase	:     Dict						 =			self.prediction_step(lowercase_					,      lowercase_					,      prediction_loss_only=lowercase_	)
       if (step + 1) * calib_dataloader.batch_size >= self.calib_num:
        break

      quant_trainer.finish_calibration(lowercase_					,      self.quant_trainer_args	)
      UpperCAmelCase	:     Optional[Any]						 =			model


     def        UpperCAmelCase_	(		self			:					List[Any]					,      lowercase_			:					Optional[Any]=None					,      lowercase_			:					Tuple=None					,      lowercase_			:					str=None					,      lowercase_			:					str = "eval"	)						->     Union[str, Any]:
      UpperCAmelCase	:     str						 =			self.eval_dataset if eval_dataset is None else eval_dataset
      UpperCAmelCase	:     int						 =			self.get_eval_dataloader(lowercase_	)
      UpperCAmelCase	:     Dict						 =			self.eval_examples if eval_examples is None else eval_examples

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

      if self.post_process_function is not None and self.compute_metrics is not None:
       UpperCAmelCase	:     int						 =			self.post_process_function(lowercase_					,      lowercase_					,      output.predictions	)
       UpperCAmelCase	:     Tuple						 =			self.compute_metrics(lowercase_	)

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

       self.log(lowercase_	)
      else:
       UpperCAmelCase	:     Any						 =			{}

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

      UpperCAmelCase	:     int						 =			self.callback_handler.on_evaluate(self.args					,      self.state					,      self.control					,      lowercase_	)
      return metrics


     def        UpperCAmelCase_	(		self			:					Optional[int]					,      lowercase_			:					List[Any]					,      lowercase_			:					int					,      lowercase_			:					Dict=None					,      lowercase_			:					str = "test"	)						->     Dict:
      UpperCAmelCase	:     Optional[Any]						 =			self.get_test_dataloader(lowercase_	)

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

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

      UpperCAmelCase	:     Dict						 =			self.post_process_function(lowercase_					,      lowercase_					,      output.predictions					,      'predict'	)
      UpperCAmelCase	:     Any						 =			self.compute_metrics(lowercase_	)

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

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




     def        UpperCAmelCase_	(		self			:					List[str]					,      lowercase_			:					int="./"	)						->     Tuple:
      UpperCAmelCase	:     Dict						 =			self.eval_dataset
      UpperCAmelCase	:     List[str]						 =			self.get_eval_dataloader(lowercase_	)

      UpperCAmelCase	:     List[Any]						 =			next(iter(lowercase_	)	)

      # saving device - to make it consistent
      UpperCAmelCase	:     Tuple						 =			torch.device('cuda' if torch.cuda.is_available() else 'cpu'	)

      # convert to tuple
      UpperCAmelCase	:     Tuple						 =			tuple(v.to(lowercase_	) for k, v in batch.items()	)

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

      UpperCAmelCase	:     Optional[Any]						 =			True

      UpperCAmelCase	:     str						 =			self.model.to(lowercase_	)

      model.eval()
      model.float()

      UpperCAmelCase	:     List[str]						 =			model.module if hasattr(lowercase_					,      'module'	) else model
      quant_trainer.configure_model(lowercase_					,      self.quant_trainer_args	)

      UpperCAmelCase	:     Dict						 =			os.path.join(lowercase_					,      'model.onnx'	)
      logger.info(f"""exporting model to {output_model_file}"""	)

      UpperCAmelCase	:     int						 =			{0: 'batch_size', 1: 'seq_len'}

      torch.onnx.export(
          lowercase_					,      lowercase_					,      lowercase_					,      export_params=lowercase_					,      opset_version=13					,      do_constant_folding=lowercase_					,      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=lowercase_					,      )
      logger.info('onnx export finished'	)







'''simple docstring'''

def 				UpperCamelCase(					UpperCAmelCase_  ,	UpperCAmelCase_     ):
 while a != 0:
  UpperCAmelCase ,			UpperCAmelCase	:     Tuple						 =			b % a, a
 return b




def 				UpperCamelCase(					UpperCAmelCase_  ,	UpperCAmelCase_     ):
 if gcd(UpperCAmelCase_  ,	UpperCAmelCase_     ) != 1:
  UpperCAmelCase	:     List[str]						 =			F"""mod inverse of {a!r} and {m!r} does not exist"""
  raise ValueError(UpperCAmelCase_     )
 UpperCAmelCase ,			UpperCAmelCase ,			UpperCAmelCase	:     Any						 =			1, 0, a
 UpperCAmelCase ,			UpperCAmelCase ,			UpperCAmelCase	:     Dict						 =			0, 1, m
 while va != 0:
  UpperCAmelCase	:     Tuple						 =			ua // va
  UpperCAmelCase ,			UpperCAmelCase ,			UpperCAmelCase ,			UpperCAmelCase ,			UpperCAmelCase ,			UpperCAmelCase	:     List[Any]						 =			(ua - q * va), (ua - q * va), (ua - q * va), va, va, va
 return ua % m




