






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


__A					=	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			(     UpperCamelCase				):
    """simple docstring"""

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



    def _a							(				self      :     Dict ,     __SCREAMING_SNAKE_CASE      :     List[str]=None      )      ->							str:

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

           __UpperCAmelCase     				=self._remove_unused_columns(__UpperCAmelCase ,     description="""Calibration"""      )

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



    def _a							(				self      :     Any ,     __SCREAMING_SNAKE_CASE      :     Tuple=None      )      ->							Dict:
           __UpperCAmelCase     				=self.train_dataset if calib_dataset is None else calib_dataset
           __UpperCAmelCase     				=self.get_calib_dataloader(__UpperCAmelCase      )

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

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

           quant_trainer.finish_calibration(__UpperCAmelCase ,     self.quant_trainer_args      )
           __UpperCAmelCase     				=model



    def _a							(				self      :     List[Any] ,     __SCREAMING_SNAKE_CASE      :     str=None ,     __SCREAMING_SNAKE_CASE      :     List[Any]=None ,     __SCREAMING_SNAKE_CASE      :     Optional[int]=None ,     __SCREAMING_SNAKE_CASE      :     Optional[int] = "eval"      )      ->							Dict:
           __UpperCAmelCase     				=self.eval_dataset if eval_dataset is None else eval_dataset
           __UpperCAmelCase     				=self.get_eval_dataloader(__UpperCAmelCase      )
           __UpperCAmelCase     				=self.eval_examples if eval_examples is None else eval_examples

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

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

                  self.log(__UpperCAmelCase      )
           else:
                  __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()      )

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



    def _a							(				self      :     str ,     __SCREAMING_SNAKE_CASE      :     Optional[int] ,     __SCREAMING_SNAKE_CASE      :     List[str] ,     __SCREAMING_SNAKE_CASE      :     Dict=None ,     __SCREAMING_SNAKE_CASE      :     Dict = "test"      )      ->							Union[str, Any]:
           __UpperCAmelCase     				=self.get_test_dataloader(__UpperCAmelCase      )

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

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

           __UpperCAmelCase     				=self.post_process_function(__UpperCAmelCase ,     __UpperCAmelCase ,     output.predictions ,     """predict"""      )
           __UpperCAmelCase     				=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}_'''      ):
                         __UpperCAmelCase     				=metrics.pop(__UpperCAmelCase      )

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



    def _a							(				self      :     Dict ,     __SCREAMING_SNAKE_CASE      :     Union[str, Any]="./"      )      ->							List[Any]:
           __UpperCAmelCase     				=self.eval_dataset
           __UpperCAmelCase     				=self.get_eval_dataloader(__UpperCAmelCase      )

           __UpperCAmelCase     				=next(iter(__UpperCAmelCase      )      )

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

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

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

           __UpperCAmelCase     				=True

           __UpperCAmelCase     				=self.model.to(__UpperCAmelCase      )

           model.eval()
           model.float()

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

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

           __UpperCAmelCase     				={0: 'batch_size', 1: 'seq_len'}

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











"""simple docstring"""





import tempfile
import unittest

from make_student import create_student_by_copying_alternating_layers

from transformers import AutoConfig
from transformers.file_utils import cached_property
from transformers.testing_utils import require_torch


__A											=						"""sshleifer/bart-tiny-random"""
__A											=						"""patrickvonplaten/t5-tiny-random"""






@require_torch
class    _lowerCAmelCase  (	unittest.TestCase			):
       """simple docstring"""




       @cached_property
       def 	snake_case		(       self ):



         '''simple docstring'''

         return AutoConfig.from_pretrained(__UpperCAmelCase )






       def 	snake_case		(       self ):



         '''simple docstring'''

         lowerCAmelCase__				,							*lowerCAmelCase__     :List[str]											=				create_student_by_copying_alternating_layers(__UpperCAmelCase       ,					tempfile.mkdtemp()       ,					e=1       ,					d=1 )
         self.assertEqual(student.config.num_hidden_layers       ,					1 )






       def 	snake_case		(       self ):



         '''simple docstring'''

         lowerCAmelCase__				,							*lowerCAmelCase__     :Optional[int]											=				create_student_by_copying_alternating_layers(__UpperCAmelCase       ,					tempfile.mkdtemp()       ,					e=1       ,					d=__UpperCAmelCase )






       def 	snake_case		(       self ):



         '''simple docstring'''

         lowerCAmelCase__				,							*lowerCAmelCase__     :List[Any]											=				create_student_by_copying_alternating_layers(__UpperCAmelCase       ,					tempfile.mkdtemp()       ,					e=1       ,					d=__UpperCAmelCase )
         self.assertEqual(student.config.encoder_layers       ,					1 )
         self.assertEqual(student.config.decoder_layers       ,					self.teacher_config.encoder_layers )






       def 	snake_case		(       self ):



         '''simple docstring'''

         lowerCAmelCase__				,							*lowerCAmelCase__     :Optional[int]											=				create_student_by_copying_alternating_layers(__UpperCAmelCase       ,					tempfile.mkdtemp()       ,					e=1       ,					d=1 )
         self.assertEqual(student.config.encoder_layers       ,					1 )
         self.assertEqual(student.config.decoder_layers       ,					1 )






       def 	snake_case		(       self ):



         '''simple docstring'''

         with self.assertRaises(__UpperCAmelCase ):
           create_student_by_copying_alternating_layers(__UpperCAmelCase       ,					tempfile.mkdtemp()       ,					e=__UpperCAmelCase       ,					d=__UpperCAmelCase )
