


"""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


lowerCamelCase_       :       Tuple      						=  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    (	_SCREAMING_SNAKE_CASE      ):






 """simple docstring"""




 def __init__(		self			,   *__A			,   __A=None			,   __A=None			,   __A=None			,   **__A						) ->					str:
     super().__init__(*__A			,   **__A						)
     a					=eval_examples
     a					=post_process_function
     a					=quant_trainer_args
     a					=128  # default number of calibration samples


 def 				SCREAMING_SNAKE_CASE							(		self			,   __A=None						) ->					int:
     if calib_dataset is None and self.calib_dataset is None:
         raise ValueError('''Trainer: calibration requires an calib_dataset.'''						)
     a					=calib_dataset if calib_dataset is not None else self.calib_dataset

     a					=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 				SCREAMING_SNAKE_CASE							(		self			,   __A=None						) ->					Optional[Any]:
     a					=self.train_dataset if calib_dataset is None else calib_dataset
     a					=self.get_calib_dataloader(__A						)

     a					=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
         a		,				a		,				a					=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						)
     a					=model


 def 				SCREAMING_SNAKE_CASE							(		self			,   __A=None			,   __A=None			,   __A=None			,   __A = "eval"						) ->					List[str]:
     a					=self.eval_dataset if eval_dataset is None else eval_dataset
     a					=self.get_eval_dataloader(__A						)
     a					=self.eval_examples if eval_examples is None else eval_examples

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

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

         self.log(__A						)
     else:
         a					={}

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

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


 def 				SCREAMING_SNAKE_CASE							(		self			,   __A			,   __A			,   __A=None			,   __A = "test"						) ->					Optional[Any]:
     a					=self.get_test_dataloader(__A						)

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

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

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

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





 def 				SCREAMING_SNAKE_CASE							(		self			,   __A="./"						) ->					Union[str, Any]:
     a					=self.eval_dataset
     a					=self.get_eval_dataloader(__A						)

     a					=next(iter(__A						)						)

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

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

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

     a					=True

     a					=self.model.to(__A						)

     model.eval()
     model.float()

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

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

     a					={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"""







import os
import sys
import unittest


lowerCamelCase_       :       Tuple      						=  os.path.abspath(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
sys.path.append(os.path.join(git_repo_path, """utils"""))

import get_test_info  # noqa: E402
from get_test_info import (  # noqa: E402
    get_model_to_test_mapping,
    get_model_to_tester_mapping,
    get_test_to_tester_mapping,
)


lowerCamelCase_       :       Dict      						=  os.path.join("""tests""", """models""", """bert""", """test_modeling_bert.py""")
lowerCamelCase_       :       Dict      						=  os.path.join("""tests""", """models""", """blip""", """test_modeling_blip.py""")






class 				__A    (	unittest.TestCase      ):






 """simple docstring"""




 def 				SCREAMING_SNAKE_CASE							(		self						) ->					Any:
     a					=get_test_to_tester_mapping(__A						)
     a					=get_test_to_tester_mapping(__A						)

     a					={'''BertModelTest''': '''BertModelTester'''}

     a					={
         '''BlipModelTest''': '''BlipModelTester''',
         '''BlipTextImageModelTest''': '''BlipTextImageModelsModelTester''',
         '''BlipTextModelTest''': '''BlipTextModelTester''',
         '''BlipTextRetrievalModelTest''': '''BlipTextRetrievalModelTester''',
         '''BlipVQAModelTest''': '''BlipVQAModelTester''',
         '''BlipVisionModelTest''': '''BlipVisionModelTester''',
     }

     self.assertEqual(get_test_info.to_json(__A						)			,   __A						)
     self.assertEqual(get_test_info.to_json(__A						)			,   __A						)


 def 				SCREAMING_SNAKE_CASE							(		self						) ->					List[str]:
     a					=get_model_to_test_mapping(__A						)
     a					=get_model_to_test_mapping(__A						)

     a					={
         '''BertForMaskedLM''': ['''BertModelTest'''],
         '''BertForMultipleChoice''': ['''BertModelTest'''],
         '''BertForNextSentencePrediction''': ['''BertModelTest'''],
         '''BertForPreTraining''': ['''BertModelTest'''],
         '''BertForQuestionAnswering''': ['''BertModelTest'''],
         '''BertForSequenceClassification''': ['''BertModelTest'''],
         '''BertForTokenClassification''': ['''BertModelTest'''],
         '''BertLMHeadModel''': ['''BertModelTest'''],
         '''BertModel''': ['''BertModelTest'''],
     }

     a					={
         '''BlipForConditionalGeneration''': ['''BlipTextImageModelTest'''],
         '''BlipForImageTextRetrieval''': ['''BlipTextRetrievalModelTest'''],
         '''BlipForQuestionAnswering''': ['''BlipVQAModelTest'''],
         '''BlipModel''': ['''BlipModelTest'''],
         '''BlipTextModel''': ['''BlipTextModelTest'''],
         '''BlipVisionModel''': ['''BlipVisionModelTest'''],
     }

     self.assertEqual(get_test_info.to_json(__A						)			,   __A						)
     self.assertEqual(get_test_info.to_json(__A						)			,   __A						)





 def 				SCREAMING_SNAKE_CASE							(		self						) ->					List[Any]:
     a					=get_model_to_tester_mapping(__A						)
     a					=get_model_to_tester_mapping(__A						)

     a					={
         '''BertForMaskedLM''': ['''BertModelTester'''],
         '''BertForMultipleChoice''': ['''BertModelTester'''],
         '''BertForNextSentencePrediction''': ['''BertModelTester'''],
         '''BertForPreTraining''': ['''BertModelTester'''],
         '''BertForQuestionAnswering''': ['''BertModelTester'''],
         '''BertForSequenceClassification''': ['''BertModelTester'''],
         '''BertForTokenClassification''': ['''BertModelTester'''],
         '''BertLMHeadModel''': ['''BertModelTester'''],
         '''BertModel''': ['''BertModelTester'''],
     }

     a					={
         '''BlipForConditionalGeneration''': ['''BlipTextImageModelsModelTester'''],
         '''BlipForImageTextRetrieval''': ['''BlipTextRetrievalModelTester'''],
         '''BlipForQuestionAnswering''': ['''BlipVQAModelTester'''],
         '''BlipModel''': ['''BlipModelTester'''],
         '''BlipTextModel''': ['''BlipTextModelTester'''],
         '''BlipVisionModel''': ['''BlipVisionModelTester'''],
     }

     self.assertEqual(get_test_info.to_json(__A						)			,   __A						)
     self.assertEqual(get_test_info.to_json(__A						)			,   __A						)