



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


__snake_case       :	Union[str, Any]            =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  lowerCamelCase__    (							snake_case__):




  '''simple docstring'''



  def __init__(self	,*__lowerCamelCase	,__lowerCamelCase=None	,__lowerCamelCase=None	,__lowerCamelCase=None	,**__lowerCamelCase				)  ->    Tuple:

      """simple docstring"""






      super().__init__(*lowercase_	,**lowercase_				)
      lowerCAmelCase__  :       List[str]				=  eval_examples
      lowerCAmelCase__  :       str				=  post_process_function
      lowerCAmelCase__  :       int				=  quant_trainer_args
      lowerCAmelCase__  :       Optional[int]				=  1_28  # default number of calibration samples



  def 	lowerCAmelCase__   (self	,__lowerCamelCase=None				)  ->    Union[str, Any]:

      """simple docstring"""






      if calib_dataset is None and self.calib_dataset is None:
          raise ValueError('''Trainer: calibration requires an calib_dataset.'''				)
      lowerCAmelCase__  :       Optional[Any]				=  calib_dataset if calib_dataset is not None else self.calib_dataset

      lowerCAmelCase__  :       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 	lowerCAmelCase__   (self	,__lowerCamelCase=None				)  ->    Optional[Any]:

      """simple docstring"""






      lowerCAmelCase__  :       Dict				=  self.train_dataset if calib_dataset is None else calib_dataset
      lowerCAmelCase__  :       str				=  self.get_calib_dataloader(lowercase_				)

      lowerCAmelCase__  :       Any				=  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
          lowerCAmelCase__  :       Optional[Any]				=  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				)
      lowerCAmelCase__  :       Optional[int]				=  model



  def 	lowerCAmelCase__   (self	,__lowerCamelCase=None	,__lowerCamelCase=None	,__lowerCamelCase=None	,__lowerCamelCase = "eval"				)  ->    str:

      """simple docstring"""






      lowerCAmelCase__  :       List[str]				=  self.eval_dataset if eval_dataset is None else eval_dataset
      lowerCAmelCase__  :       Any				=  self.get_eval_dataloader(lowercase_				)
      lowerCAmelCase__  :       int				=  self.eval_examples if eval_examples is None else eval_examples

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

      if self.post_process_function is not None and self.compute_metrics is not None:
          lowerCAmelCase__  :       Tuple				=  self.post_process_function(lowercase_	,lowercase_	,output.predictions				)
          lowerCAmelCase__  :       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}_"""				):
                  lowerCAmelCase__  :       Dict				=  metrics.pop(lowercase_				)

          self.log(lowercase_				)
      else:
          lowerCAmelCase__  :       List[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()				)

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



  def 	lowerCAmelCase__   (self	,__lowerCamelCase	,__lowerCamelCase	,__lowerCamelCase=None	,__lowerCamelCase = "test"				)  ->    Tuple:

      """simple docstring"""






      lowerCAmelCase__  :       Dict				=  self.get_test_dataloader(lowercase_				)

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

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

      lowerCAmelCase__  :       List[str]				=  self.post_process_function(lowercase_	,lowercase_	,output.predictions	,'''predict'''				)
      lowerCAmelCase__  :       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}_"""				):
              lowerCAmelCase__  :       int				=  metrics.pop(lowercase_				)

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



  def 	lowerCAmelCase__   (self	,__lowerCamelCase="./"				)  ->    Dict:

      """simple docstring"""






      lowerCAmelCase__  :       int				=  self.eval_dataset
      lowerCAmelCase__  :       Any				=  self.get_eval_dataloader(lowercase_				)

      lowerCAmelCase__  :       int				=  next(iter(lowercase_				)				)

      # saving device - to make it consistent
      lowerCAmelCase__  :       Union[str, Any]				=  torch.device('''cuda''' if torch.cuda.is_available() else '''cpu'''				)

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

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

      lowerCAmelCase__  :       Union[str, Any]				=  True

      lowerCAmelCase__  :       Optional[Any]				=  self.model.to(lowercase_				)

      model.eval()
      model.float()

      lowerCAmelCase__  :       Optional[Any]				=  model.module if hasattr(lowercase_	,'''module'''				) else model
      quant_trainer.configure_model(lowercase_	,self.quant_trainer_args				)

      lowerCAmelCase__  :       List[Any]				=  os.path.join(lowercase_	,'''model.onnx'''				)
      logger.info(f"""exporting model to {output_model_file}"""				)

      lowerCAmelCase__  :       Dict				=  {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'''				)






import copy

from ...configuration_utils import PretrainedConfig
from ...utils import logging


__snake_case       :	Any            =logging.get_logger(__name__)



class  lowerCamelCase__    (							lowerCamelCase__):




		'''simple docstring'''



		snake_case_			="""encoder-decoder"""
		snake_case_			=True



		def __init__(self	,**__lowerCamelCase				)  ->    Union[str, Any]:

						"""simple docstring"""






						super().__init__(**__lowerCamelCase				)
						assert (
						    "encoder" in kwargs and "decoder" in kwargs
						), "Config has to be initialized with encoder and decoder config"
						lowerCAmelCase__  :       Optional[Any]				=  kwargs.pop('''encoder'''				)
						lowerCAmelCase__  :       Any				=  encoder_config.pop('''model_type'''				)
						lowerCAmelCase__  :       str				=  kwargs.pop('''decoder'''				)
						lowerCAmelCase__  :       Tuple				=  decoder_config.pop('''model_type'''				)

						from ..auto.configuration_auto import AutoConfig

						lowerCAmelCase__  :       Tuple				=  AutoConfig.for_model(__lowerCamelCase	,**__lowerCamelCase				)
						lowerCAmelCase__  :       Any				=  AutoConfig.for_model(__lowerCamelCase	,**__lowerCamelCase				)
						lowerCAmelCase__  :       str				=  True



		@classmethod
		def 	lowerCAmelCase__   (cls	,__lowerCamelCase	,__lowerCamelCase	,**__lowerCamelCase				)  ->    PretrainedConfig:

						"""simple docstring"""






						logger.info('''Set `config.is_decoder=True` and `config.add_cross_attention=True` for decoder_config'''				)
						lowerCAmelCase__  :       Optional[int]				=  True
						lowerCAmelCase__  :       List[str]				=  True

						return cls(encoder=encoder_config.to_dict()	,decoder=decoder_config.to_dict()	,**__lowerCamelCase				)



		def 	lowerCAmelCase__   (self				)  ->    Tuple:

						"""simple docstring"""






						lowerCAmelCase__  :       int				=  copy.deepcopy(self.__dict__				)
						lowerCAmelCase__  :       Optional[Any]				=  self.encoder.to_dict()
						lowerCAmelCase__  :       str				=  self.decoder.to_dict()
						lowerCAmelCase__  :       Optional[int]				=  self.__class__.model_type
						return output

