






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  _lowercase   :
      def __init__(     self				,			a				,			a=1_3				,			a=7				,			a=True				,			a=True				,			a=True				,			a=True				,			a=9_9				,			a=3_2				,			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=None				,			):
       snake_case__  :    Dict							       =parent
       snake_case__  :    str							       =1_3
       snake_case__  :    Union[str, Any]							       =7
       snake_case__  :    Any							       =True
       snake_case__  :    Tuple							       =True
       snake_case__  :    Dict							       =True
       snake_case__  :    Dict							       =True
       snake_case__  :    Union[str, Any]							       =9_9
       snake_case__  :    List[Any]							       =3_8_4
       snake_case__  :    Any							       =2
       snake_case__  :    Optional[Any]							       =4
       snake_case__  :    Tuple							       =3_7
       snake_case__  :    int							       ="""gelu"""
       snake_case__  :    Optional[int]							       =0.1
       snake_case__  :    List[str]							       =0.1
       snake_case__  :    int							       =5_1_2
       snake_case__  :    int							       =1_6
       snake_case__  :    Tuple							       =2
       snake_case__  :    str							       =0.02
       snake_case__  :    Dict							       =3
       snake_case__  :    List[Any]							       =4
       snake_case__  :    List[str]							       =1_2_8
       snake_case__  :    str							       =2
       snake_case__  :    List[str]							       =9
       snake_case__  :    Any							       =1
       snake_case__  :    Optional[int]							       =None
      def 						lowercase__					(     self   ):
       snake_case__  :    Optional[Any]							       =ids_tensor([self.batch_size, self.seq_length]				,			self.vocab_size   )

       snake_case__  :    int							       =None
       if self.use_input_mask:
        snake_case__  :    List[str]							       =random_attention_mask([self.batch_size, self.seq_length]   )

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

       snake_case__  :    Dict							       =None
       snake_case__  :    Tuple							       =None
       snake_case__  :    int							       =None
       if self.use_labels:
        snake_case__  :    Optional[Any]							       =ids_tensor([self.batch_size]				,			self.type_sequence_label_size   )
        snake_case__  :    List[str]							       =ids_tensor([self.batch_size, self.seq_length]				,			self.num_labels   )
        snake_case__  :    str							       =ids_tensor([self.batch_size]				,			self.num_choices   )

       snake_case__  :    Optional[Any]							       =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=a				,			)

       return config, input_ids, token_type_ids, input_mask, sequence_labels, token_labels, choice_labels
      def 						lowercase__					(     self				,			a				,			a				,			a				,			a				,			a				,			a				,			a   ):
       snake_case__  :    Any							       =TFConvBertModel(config=a   )
       snake_case__  :    Optional[int]							       ={"""input_ids""": input_ids, """attention_mask""": input_mask, """token_type_ids""": token_type_ids}

       snake_case__  :    List[Any]							       =[input_ids, input_mask]
       snake_case__  :    List[str]							       =model(a   )

       snake_case__  :    Optional[int]							       =model(a   )

       self.parent.assertEqual(result.last_hidden_state.shape				,			(self.batch_size, self.seq_length, self.hidden_size)   )
      def 						lowercase__					(     self				,			a				,			a				,			a				,			a				,			a				,			a				,			a   ):
       snake_case__  :    List[str]							       =TFConvBertForMaskedLM(config=a   )
       snake_case__  :    str							       ={
           """input_ids""": input_ids,
           """attention_mask""": input_mask,
           """token_type_ids""": token_type_ids,
       }
       snake_case__  :    Union[str, Any]							       =model(a   )
       self.parent.assertEqual(result.logits.shape				,			(self.batch_size, self.seq_length, self.vocab_size)   )
      def 						lowercase__					(     self				,			a				,			a				,			a				,			a				,			a				,			a				,			a   ):
       snake_case__  :    List[Any]							       =self.num_labels
       snake_case__  :    Optional[int]							       =TFConvBertForSequenceClassification(config=a   )
       snake_case__  :    Union[str, Any]							       ={
           """input_ids""": input_ids,
           """attention_mask""": input_mask,
           """token_type_ids""": token_type_ids,
       }

       snake_case__  :    Optional[int]							       =model(a   )
       self.parent.assertEqual(result.logits.shape				,			(self.batch_size, self.num_labels)   )
      def 						lowercase__					(     self				,			a				,			a				,			a				,			a				,			a				,			a				,			a   ):
       snake_case__  :    Dict							       =self.num_choices
       snake_case__  :    Union[str, Any]							       =TFConvBertForMultipleChoice(config=a   )
       snake_case__  :    str							       =tf.tile(tf.expand_dims(a				,			1   )				,			(1, self.num_choices, 1)   )
       snake_case__  :    Dict							       =tf.tile(tf.expand_dims(a				,			1   )				,			(1, self.num_choices, 1)   )
       snake_case__  :    Dict							       =tf.tile(tf.expand_dims(a				,			1   )				,			(1, self.num_choices, 1)   )
       snake_case__  :    Union[str, Any]							       ={
           """input_ids""": multiple_choice_inputs_ids,
           """attention_mask""": multiple_choice_input_mask,
           """token_type_ids""": multiple_choice_token_type_ids,
       }
       snake_case__  :    Dict							       =model(a   )
       self.parent.assertEqual(result.logits.shape				,			(self.batch_size, self.num_choices)   )
      def 						lowercase__					(     self				,			a				,			a				,			a				,			a				,			a				,			a				,			a   ):
       snake_case__  :    Tuple							       =self.num_labels
       snake_case__  :    Union[str, Any]							       =TFConvBertForTokenClassification(config=a   )
       snake_case__  :    List[Any]							       ={
           """input_ids""": input_ids,
           """attention_mask""": input_mask,
           """token_type_ids""": token_type_ids,
       }
       snake_case__  :    Any							       =model(a   )
       self.parent.assertEqual(result.logits.shape				,			(self.batch_size, self.seq_length, self.num_labels)   )
      def 						lowercase__					(     self				,			a				,			a				,			a				,			a				,			a				,			a				,			a   ):
       snake_case__  :    Dict							       =TFConvBertForQuestionAnswering(config=a   )
       snake_case__  :    Optional[Any]							       ={
           """input_ids""": input_ids,
           """attention_mask""": input_mask,
           """token_type_ids""": token_type_ids,
       }

       snake_case__  :    Tuple							       =model(a   )
       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 						lowercase__					(     self   ):
       snake_case__  :    int							       =self.prepare_config_and_inputs()
       (
           snake_case__   
       )  :    List[str]							       =config_and_inputs
       snake_case__  :    Union[str, Any]							       ={"""input_ids""": input_ids, """token_type_ids""": token_type_ids, """attention_mask""": input_mask}
       return config, inputs_dict







@require_tf
class  _lowercase   (   _A						,				_A						,				unittest.TestCase						):
      _a							:						List[str]										=							(
          (
              TFConvBertModel,
              TFConvBertForMaskedLM,
              TFConvBertForQuestionAnswering,
              TFConvBertForSequenceClassification,
              TFConvBertForTokenClassification,
              TFConvBertForMultipleChoice,
          )
          if is_tf_available()
          else ()
      )
      _a							:						Optional[int]										=							(
          {
              'feature-extraction': TFConvBertModel,
              'fill-mask': TFConvBertForMaskedLM,
              'question-answering': TFConvBertForQuestionAnswering,
              'text-classification': TFConvBertForSequenceClassification,
              'token-classification': TFConvBertForTokenClassification,
              'zero-shot': TFConvBertForSequenceClassification,
          }
          if is_tf_available()
          else {}
      )
      _a							:						List[str]										=							False
      _a							:						Tuple										=							False
      _a							:						Optional[Any]										=							False
      def 						lowercase__					(     self   ):
       snake_case__  :    Optional[Any]							       =TFConvBertModelTester(self   )
       snake_case__  :    Tuple							       =ConfigTester(self				,			config_class=a				,			hidden_size=3_7   )
      def 						lowercase__					(     self   ):
       self.config_tester.run_common_tests()
      def 						lowercase__					(     self   ):
       snake_case__  :    List[str]							       =self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_model(*a   )
      def 						lowercase__					(     self   ):
       snake_case__  :    Optional[Any]							       =self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_for_masked_lm(*a   )
      def 						lowercase__					(     self   ):
       snake_case__  :    List[str]							       =self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_for_multiple_choice(*a   )
      def 						lowercase__					(     self   ):
       snake_case__  :    Tuple							       =self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_for_question_answering(*a   )
      def 						lowercase__					(     self   ):
       snake_case__  :    Optional[Any]							       =self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_for_sequence_classification(*a   )
      def 						lowercase__					(     self   ):
       snake_case__  :    Optional[int]							       =self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_for_token_classification(*a   )
      @slow
      def 						lowercase__					(     self   ):
       snake_case__  :    Union[str, Any]							       =self.model_tester.prepare_config_and_inputs_for_common()
       snake_case__  :    int							       =True
       snake_case__  :    Tuple							       =True

       if hasattr(a				,			"""use_cache"""   ):
        snake_case__  :    Optional[Any]							       =True

       snake_case__  :    Optional[Any]							       =getattr(self.model_tester				,			"""encoder_seq_length"""				,			self.model_tester.seq_length   )
       snake_case__  :    Union[str, Any]							       =getattr(self.model_tester				,			"""key_length"""				,			a   )

       for model_class in self.all_model_classes:
        snake_case__  :    Tuple							       =self._prepare_for_class(a				,			a   )
        snake_case__  :    Any							       =model_class(a   )
        snake_case__  :    Optional[int]							       =len(model(a   )   )

        with tempfile.TemporaryDirectory() as tmpdirname:
         model.save_pretrained(a				,			saved_model=a   )
         snake_case__  :    Tuple							       =os.path.join(a				,			"""saved_model"""				,			"""1"""   )
         snake_case__  :    Optional[int]							       =tf.keras.models.load_model(a   )
         snake_case__  :    Optional[Any]							       =model(a   )

         if self.is_encoder_decoder:
          snake_case__  :    List[Any]							       =outputs["""encoder_hidden_states"""]
          snake_case__  :    Tuple							       =outputs["""encoder_attentions"""]
         else:
          snake_case__  :    int							       =outputs["""hidden_states"""]
          snake_case__  :    str							       =outputs["""attentions"""]

         self.assertEqual(len(a   )				,			a   )

         snake_case__  :    List[Any]							       =getattr(
             self.model_tester				,			"""expected_num_hidden_layers"""				,			self.model_tester.num_hidden_layers + 1   )

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

         self.assertEqual(len(a   )				,			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 						lowercase__					(     self   ):
       snake_case__  :    Optional[Any]							       =TFConvBertModel.from_pretrained("""YituTech/conv-bert-base"""   )
       self.assertIsNotNone(a   )



      def 						lowercase__					(     self   ):
       snake_case__  :    Optional[Any]							       =self.model_tester.prepare_config_and_inputs_for_common()
       snake_case__  :    int							       =True
       snake_case__  :    Optional[Any]							       =getattr(self.model_tester				,			"""decoder_seq_length"""				,			self.model_tester.seq_length   )
       snake_case__  :    Any							       =getattr(self.model_tester				,			"""encoder_seq_length"""				,			self.model_tester.seq_length   )
       snake_case__  :    Union[str, Any]							       =getattr(self.model_tester				,			"""key_length"""				,			a   )
       snake_case__  :    int							       =getattr(self.model_tester				,			"""key_length"""				,			a   )

       def check_decoder_attentions_output(a   ):
        snake_case__  :    Any							       =len(a   )
        self.assertEqual(out_len % 2				,			0   )
        snake_case__  :    Any							       =outputs.decoder_attentions
        self.assertEqual(len(a   )				,			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(a   ):
        snake_case__  :    Optional[Any]							       =[
            t.numpy() for t in (outputs.encoder_attentions if config.is_encoder_decoder else outputs.attentions)
        ]
        self.assertEqual(len(a   )				,			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:
        snake_case__  :    List[Any]							       =True
        snake_case__  :    Any							       =False
        snake_case__  :    Any							       =model_class(a   )
        snake_case__  :    Dict							       =model(self._prepare_for_class(a				,			a   )   )
        snake_case__  :    str							       =len(a   )
        self.assertEqual(config.output_hidden_states				,			a   )
        check_encoder_attentions_output(a   )

        if self.is_encoder_decoder:
         snake_case__  :    Optional[int]							       =model_class(a   )
         snake_case__  :    List[Any]							       =model(self._prepare_for_class(a				,			a   )   )
         self.assertEqual(config.output_hidden_states				,			a   )
         check_decoder_attentions_output(a   )

        # Check that output attentions can also be changed via the config
        del inputs_dict["output_attentions"]
        snake_case__  :    List[Any]							       =True
        snake_case__  :    str							       =model_class(a   )
        snake_case__  :    Tuple							       =model(self._prepare_for_class(a				,			a   )   )
        self.assertEqual(config.output_hidden_states				,			a   )
        check_encoder_attentions_output(a   )

        # Check attention is always last and order is fine
        snake_case__  :    Tuple							       =True
        snake_case__  :    Optional[int]							       =True
        snake_case__  :    Optional[int]							       =model_class(a   )
        snake_case__  :    List[str]							       =model(self._prepare_for_class(a				,			a   )   )

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







@require_tf
class  _lowercase   (   unittest.TestCase						):
      @slow
      def 						lowercase__					(     self   ):
       snake_case__  :    Optional[Any]							       =TFConvBertModel.from_pretrained("""YituTech/conv-bert-base"""   )
       snake_case__  :    Tuple							       =tf.constant([[0, 1, 2, 3, 4, 5]]   )
       snake_case__  :    Dict							       =model(a   )[0]

       snake_case__  :    Any							       =[1, 6, 7_6_8]
       self.assertEqual(output.shape				,			a   )

       snake_case__  :    str							       =tf.constant(
           [
               [
                   [-0.03475493, -0.4686034, -0.30638832],
                   [0.22637248, -0.26988646, -0.7423424],
                   [0.10324868, -0.45013508, -0.58280784],
               ]
           ]   )
       tf.debugging.assert_near(output[:, :3, :3]				,			a				,			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(_A						)
class  _lowercase   (   _A						):
      def __init__(     self				,			**a   ):
       super().__init__(**a   )

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

       requires_backends(self				,			"""vision"""   )
       self.check_model_type(a   )
      def __call__(     self				,			a				,			a = None				,			**a				,			):
       if "text_queries" in kwargs:
        snake_case__  :    Any							       =kwargs.pop("""text_queries"""   )

       if isinstance(a				,			(str, Image.Image)   ):
        snake_case__  :    Union[str, Any]							       ={"""image""": image, """candidate_labels""": candidate_labels}
       else:
        snake_case__  :    str							       =image
       snake_case__  :    Union[str, Any]							       =super().__call__(a				,			**a   )
       return results
      def 						lowercase__					(     self				,			**a   ):
       snake_case__  :    Union[str, Any]							       ={}
       if "threshold" in kwargs:
        snake_case__  :    Optional[Any]							       =kwargs["""threshold"""]
       if "top_k" in kwargs:
        snake_case__  :    Dict							       =kwargs["""top_k"""]
       return {}, {}, postprocess_params
      def 						lowercase__					(     self				,			a   ):
       snake_case__  :    List[Any]							       =load_image(inputs["""image"""]   )
       snake_case__  :    Dict							       =inputs["""candidate_labels"""]
       if isinstance(a				,			a   ):
        snake_case__  :    str							       =candidate_labels.split(""","""   )

       snake_case__  :    Optional[Any]							       =torch.tensor([[image.height, image.width]]				,			dtype=torch.intaa   )
       for i, candidate_label in enumerate(a   ):
        snake_case__  :    Tuple							       =self.tokenizer(a				,			return_tensors=self.framework   )
        snake_case__  :    int							       =self.image_processor(a				,			return_tensors=self.framework   )
        yield {
            "is_last": i == len(a   ) - 1,
            "target_size": target_size,
            "candidate_label": candidate_label,
            **text_inputs,
            **image_features,
        }
      def 						lowercase__					(     self				,			a   ):
       snake_case__  :    Tuple							       =model_inputs.pop("""target_size"""   )
       snake_case__  :    Tuple							       =model_inputs.pop("""candidate_label"""   )
       snake_case__  :    Dict							       =model_inputs.pop("""is_last"""   )

       snake_case__  :    Any							       =self.model(**a   )

       snake_case__  :    List[Any]							       ={"""target_size""": target_size, """candidate_label""": candidate_label, """is_last""": is_last, **outputs}
       return model_outputs
      def 						lowercase__					(     self				,			a				,			a=0.1				,			a=None   ):
       snake_case__  :    Optional[int]							       =[]
       for model_output in model_outputs:
        snake_case__  :    Tuple							       =model_output["""candidate_label"""]
        snake_case__  :    Optional[int]							       =BaseModelOutput(a   )
        snake_case__  :    int							       =self.image_processor.post_process_object_detection(
            outputs=a				,			threshold=a				,			target_sizes=model_output["""target_size"""]   )[0]

        for index in outputs["scores"].nonzero():
         snake_case__  :    int							       =outputs["""scores"""][index].item()
         snake_case__  :    Tuple							       =self._get_bounding_box(outputs["""boxes"""][index][0]   )

         snake_case__  :    Tuple							       ={"""score""": score, """label""": label, """box""": box}
         results.append(a   )

       snake_case__  :    Tuple							       =sorted(a				,			key=lambda a   : x["score"]				,			reverse=a   )
       if top_k:
        snake_case__  :    Union[str, Any]							       =results[:top_k]

       return results



      def 						lowercase__					(     self				,			a   ):
       if self.framework != "pt":
        raise ValueError("""The ZeroShotObjectDetectionPipeline is only available in PyTorch."""   )
       snake_case__       ,   snake_case__       ,   snake_case__       ,   snake_case__  :    Dict							       =box.int().tolist()
       snake_case__  :    int							       ={
           """xmin""": xmin,
           """ymin""": ymin,
           """xmax""": xmax,
           """ymax""": ymax,
       }
       return bbox





