from __future__ import annotations

import os
import tempfile
import unittest

from transformers import ConvBertConfig, is_tf_available
from transformers.testing_utils import require_tf, slow

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


if is_tf_available():
	import tensorflow as tf

	from transformers import (
	    TFConvBertForMaskedLM,
	    TFConvBertForMultipleChoice,
	    TFConvBertForQuestionAnswering,
	    TFConvBertForSequenceClassification,
	    TFConvBertForTokenClassification,
	    TFConvBertModel,
	)


class 						a__							:



			def __init__(						self				,						UpperCAmelCase				,						UpperCAmelCase=1_3				,						UpperCAmelCase=7				,						UpperCAmelCase=True				,						UpperCAmelCase=True				,						UpperCAmelCase=True				,						UpperCAmelCase=True				,						UpperCAmelCase=9_9				,						UpperCAmelCase=3_2				,						UpperCAmelCase=2				,						UpperCAmelCase=4				,						UpperCAmelCase=3_7				,						UpperCAmelCase="gelu"				,						UpperCAmelCase=0.1				,						UpperCAmelCase=0.1				,						UpperCAmelCase=5_1_2				,						UpperCAmelCase=1_6				,						UpperCAmelCase=2				,						UpperCAmelCase=0.02				,						UpperCAmelCase=3				,						UpperCAmelCase=4				,						UpperCAmelCase=None				,						)			->   Optional[int]:
								__a		 =       parent
								__a		 =       1_3
								__a		 =       7
								__a		 =       True
								__a		 =       True
								__a		 =       True
								__a		 =       True
								__a		 =       9_9
								__a		 =       3_8_4
								__a		 =       2
								__a		 =       4
								__a		 =       3_7
								__a		 =       'gelu'
								__a		 =       0.1
								__a		 =       0.1
								__a		 =       5_1_2
								__a		 =       1_6
								__a		 =       2
								__a		 =       0.02
								__a		 =       3
								__a		 =       4
								__a		 =       1_2_8
								__a		 =       2
								__a		 =       9
								__a		 =       1
								__a		 =       None



			def __SCREAMING_SNAKE_CASE							(						self   )			->   Optional[int]:
								__a		 =       ids_tensor([self.batch_size, self.seq_length]				,						self.vocab_size   )

								__a		 =       None
								if self.use_input_mask:
													__a		 =       random_attention_mask([self.batch_size, self.seq_length]   )

								__a		 =       None
								if self.use_token_type_ids:
													__a		 =       ids_tensor([self.batch_size, self.seq_length]				,						self.type_vocab_size   )

								__a		 =       None
								__a		 =       None
								__a		 =       None
								if self.use_labels:
													__a		 =       ids_tensor([self.batch_size]				,						self.type_sequence_label_size   )
													__a		 =       ids_tensor([self.batch_size, self.seq_length]				,						self.num_labels   )
													__a		 =       ids_tensor([self.batch_size]				,						self.num_choices   )

								__a		 =       ConvBertConfig(
								    vocab_size=self.vocab_size				,						hidden_size=self.hidden_size				,						num_hidden_layers=self.num_hidden_layers				,						num_attention_heads=self.num_attention_heads				,						intermediate_size=self.intermediate_size				,						hidden_act=self.hidden_act				,						hidden_dropout_prob=self.hidden_dropout_prob				,						attention_probs_dropout_prob=self.attention_probs_dropout_prob				,						max_position_embeddings=self.max_position_embeddings				,						type_vocab_size=self.type_vocab_size				,						initializer_range=self.initializer_range				,						return_dict=UpperCAmelCase				,						)

								return config, input_ids, token_type_ids, input_mask, sequence_labels, token_labels, choice_labels



			def __SCREAMING_SNAKE_CASE							(						self				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase   )			->   Optional[int]:
								__a		 =       TFConvBertModel(config=UpperCAmelCase   )
								__a		 =       {'input_ids': input_ids, 'attention_mask': input_mask, 'token_type_ids': token_type_ids}

								__a		 =       [input_ids, input_mask]
								__a		 =       model(UpperCAmelCase   )

								__a		 =       model(UpperCAmelCase   )

								self.parent.assertEqual(result.last_hidden_state.shape				,						(self.batch_size, self.seq_length, self.hidden_size)   )



			def __SCREAMING_SNAKE_CASE							(						self				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase   )			->   Any:
								__a		 =       TFConvBertForMaskedLM(config=UpperCAmelCase   )
								__a		 =       {
								    'input_ids': input_ids,
								    'attention_mask': input_mask,
								    'token_type_ids': token_type_ids,
								}
								__a		 =       model(UpperCAmelCase   )
								self.parent.assertEqual(result.logits.shape				,						(self.batch_size, self.seq_length, self.vocab_size)   )



			def __SCREAMING_SNAKE_CASE							(						self				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase   )			->   Optional[int]:
								__a		 =       self.num_labels
								__a		 =       TFConvBertForSequenceClassification(config=UpperCAmelCase   )
								__a		 =       {
								    'input_ids': input_ids,
								    'attention_mask': input_mask,
								    'token_type_ids': token_type_ids,
								}

								__a		 =       model(UpperCAmelCase   )
								self.parent.assertEqual(result.logits.shape				,						(self.batch_size, self.num_labels)   )



			def __SCREAMING_SNAKE_CASE							(						self				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase   )			->   Tuple:
								__a		 =       self.num_choices
								__a		 =       TFConvBertForMultipleChoice(config=UpperCAmelCase   )
								__a		 =       tf.tile(tf.expand_dims(UpperCAmelCase				,						1   )				,						(1, self.num_choices, 1)   )
								__a		 =       tf.tile(tf.expand_dims(UpperCAmelCase				,						1   )				,						(1, self.num_choices, 1)   )
								__a		 =       tf.tile(tf.expand_dims(UpperCAmelCase				,						1   )				,						(1, self.num_choices, 1)   )
								__a		 =       {
								    'input_ids': multiple_choice_inputs_ids,
								    'attention_mask': multiple_choice_input_mask,
								    'token_type_ids': multiple_choice_token_type_ids,
								}
								__a		 =       model(UpperCAmelCase   )
								self.parent.assertEqual(result.logits.shape				,						(self.batch_size, self.num_choices)   )



			def __SCREAMING_SNAKE_CASE							(						self				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase   )			->   str:
								__a		 =       self.num_labels
								__a		 =       TFConvBertForTokenClassification(config=UpperCAmelCase   )
								__a		 =       {
								    'input_ids': input_ids,
								    'attention_mask': input_mask,
								    'token_type_ids': token_type_ids,
								}
								__a		 =       model(UpperCAmelCase   )
								self.parent.assertEqual(result.logits.shape				,						(self.batch_size, self.seq_length, self.num_labels)   )



			def __SCREAMING_SNAKE_CASE							(						self				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase   )			->   Optional[int]:
								__a		 =       TFConvBertForQuestionAnswering(config=UpperCAmelCase   )
								__a		 =       {
								    'input_ids': input_ids,
								    'attention_mask': input_mask,
								    'token_type_ids': token_type_ids,
								}

								__a		 =       model(UpperCAmelCase   )
								self.parent.assertEqual(result.start_logits.shape				,						(self.batch_size, self.seq_length)   )
								self.parent.assertEqual(result.end_logits.shape				,						(self.batch_size, self.seq_length)   )







			def __SCREAMING_SNAKE_CASE							(						self   )			->   Optional[int]:
								__a		 =       self.prepare_config_and_inputs()
								(
								    (
								    __a
								) ,				(
								    __a
								) ,				(
								    __a
								) ,				(
								    __a
								) ,				(
								    __a
								) ,				(
								    __a
								) ,				(
								    __a
								) ,				
								)		 =       config_and_inputs
								__a		 =       {'input_ids': input_ids, 'token_type_ids': token_type_ids, 'attention_mask': input_mask}
								return config, inputs_dict






@require_tf
class 						a__							(       __snake_case				,						__snake_case				,						unittest.TestCase		):
			A__    :       List[str]        =      (
			    (
			        TFConvBertModel,
			        TFConvBertForMaskedLM,
			        TFConvBertForQuestionAnswering,
			        TFConvBertForSequenceClassification,
			        TFConvBertForTokenClassification,
			        TFConvBertForMultipleChoice,
			    )
			    if is_tf_available()
			    else ()
			)
			A__    :       Dict        =      (
			    {
			        'feature-extraction': TFConvBertModel,
			        'fill-mask': TFConvBertForMaskedLM,
			        'question-answering': TFConvBertForQuestionAnswering,
			        'text-classification': TFConvBertForSequenceClassification,
			        'token-classification': TFConvBertForTokenClassification,
			        'zero-shot': TFConvBertForSequenceClassification,
			    }
			    if is_tf_available()
			    else {}
			)
			A__    :       Optional[int]        =      False
			A__    :       Dict        =      False
			A__    :       Union[str, Any]        =      False



			def __SCREAMING_SNAKE_CASE							(						self   )			->   Dict:
								__a		 =       TFConvBertModelTester(self   )
								__a		 =       ConfigTester(self				,						config_class=UpperCAmelCase				,						hidden_size=3_7   )



			def __SCREAMING_SNAKE_CASE							(						self   )			->   List[Any]:
								self.config_tester.run_common_tests()



			def __SCREAMING_SNAKE_CASE							(						self   )			->   Optional[Any]:
								__a		 =       self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_model(*UpperCAmelCase   )



			def __SCREAMING_SNAKE_CASE							(						self   )			->   Any:
								__a		 =       self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_masked_lm(*UpperCAmelCase   )



			def __SCREAMING_SNAKE_CASE							(						self   )			->   Union[str, Any]:
								__a		 =       self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_multiple_choice(*UpperCAmelCase   )



			def __SCREAMING_SNAKE_CASE							(						self   )			->   str:
								__a		 =       self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_question_answering(*UpperCAmelCase   )



			def __SCREAMING_SNAKE_CASE							(						self   )			->   int:
								__a		 =       self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_sequence_classification(*UpperCAmelCase   )



			def __SCREAMING_SNAKE_CASE							(						self   )			->   List[str]:
								__a		 =       self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_token_classification(*UpperCAmelCase   )



			@slow
			def __SCREAMING_SNAKE_CASE							(						self   )			->   Tuple:
								__a ,				__a		 =       self.model_tester.prepare_config_and_inputs_for_common()
								__a		 =       True
								__a		 =       True

								if hasattr(UpperCAmelCase				,						'use_cache'   ):
													__a		 =       True

								__a		 =       getattr(self.model_tester				,						'encoder_seq_length'				,						self.model_tester.seq_length   )
								__a		 =       getattr(self.model_tester				,						'key_length'				,						UpperCAmelCase   )

								for model_class in self.all_model_classes:
													__a		 =       self._prepare_for_class(UpperCAmelCase				,						UpperCAmelCase   )
													__a		 =       model_class(UpperCAmelCase   )
													__a		 =       len(model(UpperCAmelCase   )   )

													with tempfile.TemporaryDirectory() as tmpdirname:
																		model.save_pretrained(UpperCAmelCase				,						saved_model=UpperCAmelCase   )
																		__a		 =       os.path.join(UpperCAmelCase				,						'saved_model'				,						'1'   )
																		__a		 =       tf.keras.models.load_model(UpperCAmelCase   )
																		__a		 =       model(UpperCAmelCase   )

																		if self.is_encoder_decoder:
																							__a		 =       outputs['encoder_hidden_states']
																							__a		 =       outputs['encoder_attentions']
																		else:
																							__a		 =       outputs['hidden_states']
																							__a		 =       outputs['attentions']

																		self.assertEqual(len(UpperCAmelCase   )				,						UpperCAmelCase   )

																		__a		 =       getattr(
																		    self.model_tester				,						'expected_num_hidden_layers'				,						self.model_tester.num_hidden_layers + 1   )

																		self.assertEqual(len(UpperCAmelCase   )				,						UpperCAmelCase   )
																		self.assertListEqual(
																		    list(output_hidden_states[0].shape[-2:]   )				,						[self.model_tester.seq_length, self.model_tester.hidden_size]				,						)

																		self.assertEqual(len(UpperCAmelCase   )				,						self.model_tester.num_hidden_layers   )
																		self.assertListEqual(
																		    list(output_attentions[0].shape[-3:]   )				,						[self.model_tester.num_attention_heads / 2, encoder_seq_length, encoder_key_length]				,						)



			@slow
			def __SCREAMING_SNAKE_CASE							(						self   )			->   Tuple:
								__a		 =       TFConvBertModel.from_pretrained('YituTech/conv-bert-base'   )
								self.assertIsNotNone(UpperCAmelCase   )







			def __SCREAMING_SNAKE_CASE							(						self   )			->   Any:
								__a ,				__a		 =       self.model_tester.prepare_config_and_inputs_for_common()
								__a		 =       True
								__a		 =       getattr(self.model_tester				,						'decoder_seq_length'				,						self.model_tester.seq_length   )
								__a		 =       getattr(self.model_tester				,						'encoder_seq_length'				,						self.model_tester.seq_length   )
								__a		 =       getattr(self.model_tester				,						'key_length'				,						UpperCAmelCase   )
								__a		 =       getattr(self.model_tester				,						'key_length'				,						UpperCAmelCase   )

								def check_decoder_attentions_output(UpperCAmelCase   ):
													__a		 =       len(UpperCAmelCase   )
													self.assertEqual(out_len % 2				,						0   )
													__a		 =       outputs.decoder_attentions
													self.assertEqual(len(UpperCAmelCase   )				,						self.model_tester.num_hidden_layers   )
													self.assertListEqual(
													    list(decoder_attentions[0].shape[-3:]   )				,						[self.model_tester.num_attention_heads / 2, decoder_seq_length, decoder_key_length]				,						)

								def check_encoder_attentions_output(UpperCAmelCase   ):
													__a		 =       [
													    t.numpy() for t in (outputs.encoder_attentions if config.is_encoder_decoder else outputs.attentions)
													]
													self.assertEqual(len(UpperCAmelCase   )				,						self.model_tester.num_hidden_layers   )
													self.assertListEqual(
													    list(attentions[0].shape[-3:]   )				,						[self.model_tester.num_attention_heads / 2, encoder_seq_length, encoder_key_length]				,						)

								for model_class in self.all_model_classes:
													__a		 =       True
													__a		 =       False
													__a		 =       model_class(UpperCAmelCase   )
													__a		 =       model(self._prepare_for_class(UpperCAmelCase				,						UpperCAmelCase   )   )
													__a		 =       len(UpperCAmelCase   )
													self.assertEqual(config.output_hidden_states				,						UpperCAmelCase   )
													check_encoder_attentions_output(UpperCAmelCase   )

													if self.is_encoder_decoder:
																		__a		 =       model_class(UpperCAmelCase   )
																		__a		 =       model(self._prepare_for_class(UpperCAmelCase				,						UpperCAmelCase   )   )
																		self.assertEqual(config.output_hidden_states				,						UpperCAmelCase   )
																		check_decoder_attentions_output(UpperCAmelCase   )

													# Check that output attentions can also be changed via the config
													del inputs_dict["output_attentions"]
													__a		 =       True
													__a		 =       model_class(UpperCAmelCase   )
													__a		 =       model(self._prepare_for_class(UpperCAmelCase				,						UpperCAmelCase   )   )
													self.assertEqual(config.output_hidden_states				,						UpperCAmelCase   )
													check_encoder_attentions_output(UpperCAmelCase   )

													# Check attention is always last and order is fine
													__a		 =       True
													__a		 =       True
													__a		 =       model_class(UpperCAmelCase   )
													__a		 =       model(self._prepare_for_class(UpperCAmelCase				,						UpperCAmelCase   )   )

													self.assertEqual(out_len + (2 if self.is_encoder_decoder else 1)				,						len(UpperCAmelCase   )   )
													self.assertEqual(model.config.output_hidden_states				,						UpperCAmelCase   )
													check_encoder_attentions_output(UpperCAmelCase   )






@require_tf
class 						a__							(       unittest.TestCase		):



			@slow
			def __SCREAMING_SNAKE_CASE							(						self   )			->   Optional[int]:
								__a		 =       TFConvBertModel.from_pretrained('YituTech/conv-bert-base'   )
								__a		 =       tf.constant([[0, 1, 2, 3, 4, 5]]   )
								__a		 =       model(UpperCAmelCase   )[0]

								__a		 =       [1, 6, 7_6_8]
								self.assertEqual(output.shape				,						UpperCAmelCase   )

								__a		 =       tf.constant(
								    [
								        [
								            [-0.03_475_493, -0.4_686_034, -0.30_638_832],
								            [0.22_637_248, -0.26_988_646, -0.7_423_424],
								            [0.10_324_868, -0.45_013_508, -0.58_280_784],
								        ]
								    ]   )
								tf.debugging.assert_near(output[:, :3, :3]				,						UpperCAmelCase				,						atol=1e-4   )





from typing import Any, Dict, List, Union

from ..utils import add_end_docstrings, is_torch_available, is_vision_available, logging, requires_backends
from .base import PIPELINE_INIT_ARGS, ChunkPipeline


if is_vision_available():
	from PIL import Image

	from ..image_utils import load_image

if is_torch_available():
	import torch

	from transformers.modeling_outputs import BaseModelOutput

	from ..models.auto.modeling_auto import MODEL_FOR_ZERO_SHOT_OBJECT_DETECTION_MAPPING

lowerCamelCase_    :		Dict							=	logging.get_logger(__name__)


@add_end_docstrings(__snake_case		)
class 						a__							(       __snake_case		):



			def __init__(						self				,						**UpperCAmelCase   )			->   List[str]:
								super().__init__(**UpperCAmelCase   )

								if self.framework == "tf":
													raise ValueError(f'''The {self.__class__} is only available in PyTorch.'''   )

								requires_backends(self				,						'vision'   )
								self.check_model_type(UpperCAmelCase   )



			def __call__(						self				,						UpperCAmelCase				,						UpperCAmelCase = None				,						**UpperCAmelCase				,						)			->   Tuple:
								if "text_queries" in kwargs:
													__a		 =       kwargs.pop('text_queries'   )

								if isinstance(UpperCAmelCase				,						(str, Image.Image)   ):
													__a		 =       {'image': image, 'candidate_labels': candidate_labels}
								else:
													__a		 =       image
								__a		 =       super().__call__(UpperCAmelCase				,						**UpperCAmelCase   )
								return results



			def __SCREAMING_SNAKE_CASE							(						self				,						**UpperCAmelCase   )			->   List[str]:
								__a		 =       {}
								if "threshold" in kwargs:
													__a		 =       kwargs['threshold']
								if "top_k" in kwargs:
													__a		 =       kwargs['top_k']
								return {}, {}, postprocess_params



			def __SCREAMING_SNAKE_CASE							(						self				,						UpperCAmelCase   )			->   Union[str, Any]:
								__a		 =       load_image(inputs['image']   )
								__a		 =       inputs['candidate_labels']
								if isinstance(UpperCAmelCase				,						UpperCAmelCase   ):
													__a		 =       candidate_labels.split(','   )

								__a		 =       torch.tensor([[image.height, image.width]]				,						dtype=torch.intaa   )
								for i, candidate_label in enumerate(UpperCAmelCase   ):
													__a		 =       self.tokenizer(UpperCAmelCase				,						return_tensors=self.framework   )
													__a		 =       self.image_processor(UpperCAmelCase				,						return_tensors=self.framework   )
													yield {
													    "is_last": i == len(UpperCAmelCase   ) - 1,
													    "target_size": target_size,
													    "candidate_label": candidate_label,
													    **text_inputs,
													    **image_features,
													}



			def __SCREAMING_SNAKE_CASE							(						self				,						UpperCAmelCase   )			->   str:
								__a		 =       model_inputs.pop('target_size'   )
								__a		 =       model_inputs.pop('candidate_label'   )
								__a		 =       model_inputs.pop('is_last'   )

								__a		 =       self.model(**UpperCAmelCase   )

								__a		 =       {'target_size': target_size, 'candidate_label': candidate_label, 'is_last': is_last, **outputs}
								return model_outputs



			def __SCREAMING_SNAKE_CASE							(						self				,						UpperCAmelCase				,						UpperCAmelCase=0.1				,						UpperCAmelCase=None   )			->   Tuple:
								__a		 =       []
								for model_output in model_outputs:
													__a		 =       model_output['candidate_label']
													__a		 =       BaseModelOutput(UpperCAmelCase   )
													__a		 =       self.image_processor.post_process_object_detection(
													    outputs=UpperCAmelCase				,						threshold=UpperCAmelCase				,						target_sizes=model_output['target_size']   )[0]

													for index in outputs["scores"].nonzero():
																		__a		 =       outputs['scores'][index].item()
																		__a		 =       self._get_bounding_box(outputs['boxes'][index][0]   )

																		__a		 =       {'score': score, 'label': label, 'box': box}
																		results.append(UpperCAmelCase   )

								__a		 =       sorted(UpperCAmelCase				,						key=lambda UpperCAmelCase   : x["score"]				,						reverse=UpperCAmelCase   )
								if top_k:
													__a		 =       results[:top_k]

								return results







			def __SCREAMING_SNAKE_CASE							(						self				,						UpperCAmelCase   )			->   Dict[str, int]:
								if self.framework != "pt":
													raise ValueError('The ZeroShotObjectDetectionPipeline is only available in PyTorch.'   )
								__a ,				__a ,				__a ,				__a		 =       box.int().tolist()
								__a		 =       {
								    'xmin': xmin,
								    'ymin': ymin,
								    'xmax': xmax,
								    'ymax': ymax,
								}
								return bbox




