


'''simple docstring'''
from __future__ import annotations

import unittest

import numpy as np

from transformers import BlipTextConfig
from transformers.testing_utils import require_tf, slow
from transformers.utils import is_tf_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor, random_attention_mask


if is_tf_available():
					import tensorflow as tf

					from transformers import TFBlipTextModel
					from transformers.models.blip.modeling_tf_blip import TF_BLIP_PRETRAINED_MODEL_ARCHIVE_LIST




class        _UpperCamelCase	:


				'''simple docstring'''


				def __init__(     self							,      a_							,      a_=1_2							,      a_=7							,      a_=True							,      a_=True							,      a_=True							,      a_=9_9							,      a_=3_2							,      a_=3_2							,      a_=2							,      a_=4							,      a_=3_7							,      a_=0.1							,      a_=0.1							,      a_=5_1_2							,      a_=0.02							,      a_=0							,      a_=None							,      )    ->  Dict:
					lowercase						:		Dict   					=		parent
					lowercase						:		Optional[Any]   					=		batch_size
					lowercase						:		Optional[Any]   					=		seq_length
					lowercase						:		Dict   					=		is_training
					lowercase						:		str   					=		use_input_mask
					lowercase						:		Optional[Any]   					=		use_labels
					lowercase						:		int   					=		vocab_size
					lowercase						:		Any   					=		hidden_size
					lowercase						:		Tuple   					=		projection_dim
					lowercase						:		Union[str, Any]   					=		num_hidden_layers
					lowercase						:		List[Any]   					=		num_attention_heads
					lowercase						:		int   					=		intermediate_size
					lowercase						:		Optional[Any]   					=		dropout
					lowercase						:		str   					=		attention_dropout
					lowercase						:		int   					=		max_position_embeddings
					lowercase						:		Optional[int]   					=		initializer_range
					lowercase						:		Optional[int]   					=		scope
					lowercase						:		int   					=		bos_token_id

				def 	a__		(     self							)    ->  Any:
					lowercase						:		str   					=		ids_tensor([self.batch_size, self.seq_length]							,      self.vocab_size							)

					lowercase						:		Dict   					=		None
					if self.use_input_mask:
						lowercase						:		Optional[Any]   					=		random_attention_mask([self.batch_size, self.seq_length]							)

					if input_mask is not None:
						lowercase						:		int   					=		input_mask.numpy()
						lowercase					,		lowercase						:		Union[str, Any]   					=		input_mask.shape
						lowercase						:		Tuple   					=		np.random.randint(1							,      seq_length - 1							,      size=(batch_size,)							)
						for batch_idx, start_index in enumerate(a_							):
							lowercase						:		List[Any]   					=		1
							lowercase						:		Dict   					=		0

					lowercase						:		int   					=		self.get_config()

					return config, input_ids, tf.convert_to_tensor(a_							)

				def 	a__		(     self							)    ->  Tuple:
					return BlipTextConfig(
					    vocab_size=self.vocab_size							,      hidden_size=self.hidden_size							,      projection_dim=self.projection_dim							,      num_hidden_layers=self.num_hidden_layers							,      num_attention_heads=self.num_attention_heads							,      intermediate_size=self.intermediate_size							,      dropout=self.dropout							,      attention_dropout=self.attention_dropout							,      max_position_embeddings=self.max_position_embeddings							,      initializer_range=self.initializer_range							,      bos_token_id=self.bos_token_id							,      )

				def 	a__		(     self							,      a_							,      a_							,      a_							)    ->  List[Any]:
					lowercase						:		str   					=		TFBlipTextModel(config=a_							)
					lowercase						:		Dict   					=		model(a_							,      attention_mask=a_							,      training=a_							)
					lowercase						:		List[Any]   					=		model(a_							,      training=a_							)
					self.parent.assertEqual(result.last_hidden_state.shape							,      (self.batch_size, self.seq_length, self.hidden_size)							)
					self.parent.assertEqual(result.pooler_output.shape							,      (self.batch_size, self.hidden_size)							)


				def 	a__		(     self							)    ->  str:
					lowercase						:		List[Any]   					=		self.prepare_config_and_inputs()
					lowercase					,		lowercase					,		lowercase						:		List[Any]   					=		config_and_inputs
					lowercase						:		Dict   					=		{"input_ids": input_ids, "attention_mask": input_mask}
					return config, inputs_dict




@require_tf
class        _UpperCamelCase	(    SCREAMING_SNAKE_CASE						,			unittest.TestCase):


				'''simple docstring'''


				_snake_case						       =				(TFBlipTextModel,) if is_tf_available() else ()
				_snake_case						       =				False
				_snake_case						       =				False
				_snake_case						       =				False

				def 	a__		(     self							)    ->  Optional[int]:
					lowercase						:		Optional[int]   					=		BlipTextModelTester(self							)
					lowercase						:		List[str]   					=		ConfigTester(self							,      config_class=a_							,      hidden_size=3_7							)

				def 	a__		(     self							)    ->  Optional[int]:
					self.config_tester.run_common_tests()

				def 	a__		(     self							)    ->  Dict:
					lowercase						:		List[str]   					=		self.model_tester.prepare_config_and_inputs()
					self.model_tester.create_and_check_model(*a_							)

				def 	a__		(     self							)    ->  List[Any]:
					pass

				def 	a__		(     self							)    ->  Tuple:
					pass

				@unittest.skip(reason="Blip does not use inputs_embeds"							)
				def 	a__		(     self							)    ->  Optional[Any]:
					pass

				@unittest.skip(reason="BlipTextModel has no base class and is not available in MODEL_MAPPING"							)
				def 	a__		(     self							)    ->  List[str]:
					pass

				@unittest.skip(reason="BlipTextModel has no base class and is not available in MODEL_MAPPING"							)
				def 	a__		(     self							)    ->  str:
					pass

				@slow
				def 	a__		(     self							)    ->  Tuple:
					for model_name in TF_BLIP_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
						lowercase						:		Any   					=		TFBlipTextModel.from_pretrained(a_							)
						self.assertIsNotNone(a_							)


				def 	a__		(     self							,      a_=True							)    ->  List[str]:
					super().test_pt_tf_model_equivalence(allow_missing_keys=a_							)






'''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   : Dict             =				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        _UpperCamelCase	(    SCREAMING_SNAKE_CASE):


				'''simple docstring'''


				def __init__(     self							,      *a_							,      a_=None							,      a_=None							,      a_=None							,      **a_							)    ->  Any:
					super().__init__(*a_							,      **a_							)
					lowercase						:		Optional[Any]   					=		eval_examples
					lowercase						:		Any   					=		post_process_function
					lowercase						:		Dict   					=		quant_trainer_args
					lowercase						:		List[Any]   					=		1_2_8  # default number of calibration samples

				def 	a__		(     self							,      a_=None							)    ->  List[str]:
					if calib_dataset is None and self.calib_dataset is None:
						raise ValueError("Trainer: calibration requires an calib_dataset."							)
					lowercase						:		Dict   					=		calib_dataset if calib_dataset is not None else self.calib_dataset

					lowercase						:		Optional[int]   					=		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 	a__		(     self							,      a_=None							)    ->  str:
					lowercase						:		Optional[Any]   					=		self.train_dataset if calib_dataset is None else calib_dataset
					lowercase						:		Optional[Any]   					=		self.get_calib_dataloader(a_							)

					lowercase						:		Union[str, Any]   					=		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
						lowercase					,		lowercase					,		lowercase						:		Union[str, Any]   					=		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							)
					lowercase						:		List[Any]   					=		model

				def 	a__		(     self							,      a_=None							,      a_=None							,      a_=None							,      a_ = "eval"							)    ->  int:
					lowercase						:		List[str]   					=		self.eval_dataset if eval_dataset is None else eval_dataset
					lowercase						:		Union[str, Any]   					=		self.get_eval_dataloader(a_							)
					lowercase						:		Tuple   					=		self.eval_examples if eval_examples is None else eval_examples

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

					if self.post_process_function is not None and self.compute_metrics is not None:
						lowercase						:		Union[str, Any]   					=		self.post_process_function(a_							,      a_							,      output.predictions							)
						lowercase						:		Union[str, Any]   					=		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}_'''							):
								lowercase						:		Dict   					=		metrics.pop(a_							)

						self.log(a_							)
					else:
						lowercase						:		Tuple   					=		{}

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

					lowercase						:		int   					=		self.callback_handler.on_evaluate(self.args							,      self.state							,      self.control							,      a_							)
					return metrics

				def 	a__		(     self							,      a_							,      a_							,      a_=None							,      a_ = "test"							)    ->  Optional[int]:
					lowercase						:		Dict   					=		self.get_test_dataloader(a_							)

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

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

					lowercase						:		List[str]   					=		self.post_process_function(a_							,      a_							,      output.predictions							,      "predict"							)
					lowercase						:		Dict   					=		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}_'''							):
							lowercase						:		Union[str, Any]   					=		metrics.pop(a_							)

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


				def 	a__		(     self							,      a_="./"							)    ->  Optional[Any]:
					lowercase						:		str   					=		self.eval_dataset
					lowercase						:		Dict   					=		self.get_eval_dataloader(a_							)

					lowercase						:		Any   					=		next(iter(a_							)							)

					# saving device - to make it consistent
					lowercase						:		List[Any]   					=		torch.device("cuda" if torch.cuda.is_available() else "cpu"							)

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

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

					lowercase						:		Optional[Any]   					=		True

					lowercase						:		int   					=		self.model.to(a_							)

					model.eval()
					model.float()

					lowercase						:		Union[str, Any]   					=		model.module if hasattr(a_							,      "module"							) else model
					quant_trainer.configure_model(a_							,      self.quant_trainer_args							)

					lowercase						:		List[str]   					=		os.path.join(a_							,      "model.onnx"							)
					logger.info(F'''exporting model to {output_model_file}'''							)

					lowercase						:		Any   					=		{0: "batch_size", 1: "seq_len"}

					torch.onnx.export(
					    a_							,      a_							,      a_							,      export_params=a_							,      opset_version=1_3							,      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"							)


