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


UpperCamelCase__											=					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_     ):


   def __init__(self     :						Any       ,  *__UpperCAmelCase     :						List[str]       ,  __UpperCAmelCase     :						Dict=None       ,  __UpperCAmelCase     :						Any=None       ,  __UpperCAmelCase     :						Optional[int]=None       ,  **__UpperCAmelCase     :						Optional[Any]		)						->				Union[str, Any]:
         """simple docstring"""



         super().__init__(*__UpperCAmelCase       ,  **__UpperCAmelCase		)
         UpperCAmelCase__								=  eval_examples
         UpperCAmelCase__								=  post_process_function
         UpperCAmelCase__								=  quant_trainer_args
         UpperCAmelCase__								=  1_2_8  # default number of calibration samples


   def 			lowercase_      (self     :						int       ,  __UpperCAmelCase     :						str=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."		)
         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 			lowercase_      (self     :						int       ,  __UpperCAmelCase     :						int=None		)						->				Optional[Any]:
         """simple docstring"""



         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__		,      UpperCAmelCase__		,      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 			lowercase_      (self     :						List[Any]       ,  __UpperCAmelCase     :						str=None       ,  __UpperCAmelCase     :						str=None       ,  __UpperCAmelCase     :						str=None       ,  __UpperCAmelCase     :						str = "eval"		)						->				int:
         """simple docstring"""



         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 			lowercase_      (self     :						Union[str, Any]       ,  __UpperCAmelCase     :						Tuple       ,  __UpperCAmelCase     :						str       ,  __UpperCAmelCase     :						Optional[int]=None       ,  __UpperCAmelCase     :						str = "test"		)						->				Dict:
         """simple docstring"""



         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 			lowercase_      (self     :						Optional[int]       ,  __UpperCAmelCase     :						int="./"		)						->				Any:
         """simple docstring"""



         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=1_3       ,  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'''




from typing import List, Optional, Union

from ...image_utils import ImageInput
from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding, PaddingStrategy, PreTokenizedInput, TextInput, TruncationStrategy
from ...utils import TensorType







class 	_lowerCAmelCase							(			__UpperCAmelCase ):
	__SCREAMING_SNAKE_CASE   :		Any					=					['image_processor', 'tokenizer']
	__SCREAMING_SNAKE_CASE   :		Union[str, Any]					=					'BlipImageProcessor'
	__SCREAMING_SNAKE_CASE   :		List[Any]					=					('BertTokenizer', 'BertTokenizerFast')


	def __init__(self       ,      lowercase       ,      lowercase							):
				A_	:						List[Any]	=     False
				super().__init__(lowercase       ,      lowercase							)
				A_	:						Tuple	=     self.image_processor


	def __call__(self       ,      lowercase = None       ,      lowercase = None       ,      lowercase = True       ,      lowercase = False       ,      lowercase = None       ,      lowercase = None       ,      lowercase = 0       ,      lowercase = None       ,      lowercase = None       ,      lowercase = False       ,      lowercase = False       ,      lowercase = False       ,      lowercase = False       ,      lowercase = False       ,      lowercase = True       ,      lowercase = None       ,      **lowercase       ,      ):
				if images is None and text is None:
							raise ValueError("""You have to specify either images or text."""							)

				# Get only text
				if images is None:
							A_	:						Optional[Any]	=     self.tokenizer
							A_	:						Tuple	=     self.tokenizer(
							    text=lowercase       ,      add_special_tokens=lowercase       ,      padding=lowercase       ,      truncation=lowercase       ,      max_length=lowercase       ,      stride=lowercase       ,      pad_to_multiple_of=lowercase       ,      return_attention_mask=lowercase       ,      return_overflowing_tokens=lowercase       ,      return_special_tokens_mask=lowercase       ,      return_offsets_mapping=lowercase       ,      return_token_type_ids=lowercase       ,      return_length=lowercase       ,      verbose=lowercase       ,      return_tensors=lowercase       ,      **lowercase       ,      )
							return text_encoding

				# add pixel_values
				A_	:						int	=     self.image_processor(lowercase       ,      return_tensors=lowercase							)

				if text is not None:
							A_	:						Optional[Any]	=     self.tokenizer(
							    text=lowercase       ,      add_special_tokens=lowercase       ,      padding=lowercase       ,      truncation=lowercase       ,      max_length=lowercase       ,      stride=lowercase       ,      pad_to_multiple_of=lowercase       ,      return_attention_mask=lowercase       ,      return_overflowing_tokens=lowercase       ,      return_special_tokens_mask=lowercase       ,      return_offsets_mapping=lowercase       ,      return_token_type_ids=lowercase       ,      return_length=lowercase       ,      verbose=lowercase       ,      return_tensors=lowercase       ,      **lowercase       ,      )
				else:
							A_	:						List[str]	=     None

				if text_encoding is not None:
							encoding_image_processor.update(lowercase							)

				return encoding_image_processor


	def     _a	(self       ,      *lowercase       ,      **lowercase							):
				return self.tokenizer.batch_decode(*lowercase       ,      **lowercase							)


	def     _a	(self       ,      *lowercase       ,      **lowercase							):
				return self.tokenizer.decode(*lowercase       ,      **lowercase							)


	@property
	def     _a	(self							):
				A_	:						int	=     self.tokenizer.model_input_names
				A_	:						Any	=     self.image_processor.model_input_names
				return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names							)							)