
import copy
import os
from collections import OrderedDict
from typing import TYPE_CHECKING, Any, Dict, Mapping, Optional, Union


if TYPE_CHECKING:
 from ...processing_utils import ProcessorMixin
 from ...utils import TensorType

from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig
from ...utils import logging


__lowerCAmelCase					:			Optional[int]			      =		logging.get_logger(__name__)

__lowerCAmelCase					:			List[Any]			      =		{
    "google/owlvit-base-patch32": "https://huggingface.co/google/owlvit-base-patch32/resolve/main/config.json",
    "google/owlvit-base-patch16": "https://huggingface.co/google/owlvit-base-patch16/resolve/main/config.json",
    "google/owlvit-large-patch14": "https://huggingface.co/google/owlvit-large-patch14/resolve/main/config.json",
}


class 						__lowerCAmelCase	(	lowerCAmelCase_      ):



      """simple docstring"""







      A__	:			Dict        =    '''owlvit_text_model'''




      def __init__(      self		:       Optional[int]				,       _snake_case		:       Any=4_9408				,       _snake_case		:       Optional[Any]=512				,       _snake_case		:       Optional[Any]=2048				,       _snake_case		:       str=12				,       _snake_case		:       Dict=8				,       _snake_case		:       Optional[Any]=16				,       _snake_case		:       List[str]="quick_gelu"				,       _snake_case		:       Tuple=1E-5				,       _snake_case		:       int=0.0				,       _snake_case		:       int=0.02				,       _snake_case		:       str=1.0				,       _snake_case		:       str=0				,       _snake_case		:       Union[str, Any]=4_9406				,       _snake_case		:       List[str]=4_9407				,       **_snake_case		:       str				,       ):
          super().__init__(pad_token_id=_snake_case				,       bos_token_id=_snake_case				,       eos_token_id=_snake_case				,       **_snake_case				)

          __lowercase		:			int	    =       vocab_size
          __lowercase		:			List[str]	    =       hidden_size
          __lowercase		:			Any	    =       intermediate_size
          __lowercase		:			int	    =       num_hidden_layers
          __lowercase		:			List[Any]	    =       num_attention_heads
          __lowercase		:			str	    =       max_position_embeddings
          __lowercase		:			Any	    =       hidden_act
          __lowercase		:			Tuple	    =       layer_norm_eps
          __lowercase		:			Optional[Any]	    =       attention_dropout
          __lowercase		:			int	    =       initializer_range
          __lowercase		:			int	    =       initializer_factor




      @classmethod
      def 		snake_case_			(      cls		:       int				,       _snake_case		:       Union[str, os.PathLike]				,       **_snake_case		:       Dict				):
          cls._set_token_in_kwargs(_snake_case				)

          __lowercase    ,      __lowercase		:			int	    =       cls.get_config_dict(_snake_case				,       **_snake_case				)

          # get the text config dict if we are loading from OwlViTConfig
          if config_dict.get('''model_type'''				) == "owlvit":
              __lowercase		:			Tuple	    =       config_dict['''text_config''']

          if "model_type" in config_dict and hasattr(cls				,       '''model_type'''				) and config_dict["model_type"] != cls.model_type:
              logger.warning(
                  F'You are using a model of type {config_dict["model_type"]} to instantiate a model of type '
                  F'{cls.model_type}. This is not supported for all configurations of models and can yield errors.'				)

          return cls.from_dict(_snake_case				,       **_snake_case				)






class 						__lowerCAmelCase	(	lowerCAmelCase_      ):



      """simple docstring"""







      A__	:			int        =    '''owlvit_vision_model'''




      def __init__(      self		:       List[Any]				,       _snake_case		:       Tuple=768				,       _snake_case		:       Dict=3072				,       _snake_case		:       Optional[Any]=12				,       _snake_case		:       List[Any]=12				,       _snake_case		:       Optional[Any]=3				,       _snake_case		:       Tuple=768				,       _snake_case		:       str=32				,       _snake_case		:       List[Any]="quick_gelu"				,       _snake_case		:       Optional[Any]=1E-5				,       _snake_case		:       Tuple=0.0				,       _snake_case		:       Dict=0.02				,       _snake_case		:       List[Any]=1.0				,       **_snake_case		:       Optional[Any]				,       ):
          super().__init__(**_snake_case				)

          __lowercase		:			Optional[int]	    =       hidden_size
          __lowercase		:			List[str]	    =       intermediate_size
          __lowercase		:			List[Any]	    =       num_hidden_layers
          __lowercase		:			Optional[Any]	    =       num_attention_heads
          __lowercase		:			Dict	    =       num_channels
          __lowercase		:			Tuple	    =       image_size
          __lowercase		:			Optional[int]	    =       patch_size
          __lowercase		:			int	    =       hidden_act
          __lowercase		:			Dict	    =       layer_norm_eps
          __lowercase		:			Optional[Any]	    =       attention_dropout
          __lowercase		:			Tuple	    =       initializer_range
          __lowercase		:			Optional[Any]	    =       initializer_factor




      @classmethod
      def 		snake_case_			(      cls		:       List[Any]				,       _snake_case		:       Union[str, os.PathLike]				,       **_snake_case		:       Union[str, Any]				):
          cls._set_token_in_kwargs(_snake_case				)

          __lowercase    ,      __lowercase		:			List[Any]	    =       cls.get_config_dict(_snake_case				,       **_snake_case				)

          # get the vision config dict if we are loading from OwlViTConfig
          if config_dict.get('''model_type'''				) == "owlvit":
              __lowercase		:			Tuple	    =       config_dict['''vision_config''']

          if "model_type" in config_dict and hasattr(cls				,       '''model_type'''				) and config_dict["model_type"] != cls.model_type:
              logger.warning(
                  F'You are using a model of type {config_dict["model_type"]} to instantiate a model of type '
                  F'{cls.model_type}. This is not supported for all configurations of models and can yield errors.'				)

          return cls.from_dict(_snake_case				,       **_snake_case				)






class 						__lowerCAmelCase	(	lowerCAmelCase_      ):



      """simple docstring"""







      A__	:			List[str]        =    '''owlvit'''
      A__	:			List[str]        =    True




      def __init__(      self		:       Union[str, Any]				,       _snake_case		:       int=None				,       _snake_case		:       Any=None				,       _snake_case		:       Dict=512				,       _snake_case		:       int=2.65_92				,       _snake_case		:       Tuple=True				,       **_snake_case		:       List[Any]				,       ):
          super().__init__(**_snake_case				)

          if text_config is None:
              __lowercase		:			Optional[int]	    =       {}
              logger.info('''text_config is None. Initializing the OwlViTTextConfig with default values.'''				)

          if vision_config is None:
              __lowercase		:			Optional[Any]	    =       {}
              logger.info('''vision_config is None. initializing the OwlViTVisionConfig with default values.'''				)

          __lowercase		:			List[str]	    =       OwlViTTextConfig(**_snake_case				)
          __lowercase		:			Any	    =       OwlViTVisionConfig(**_snake_case				)

          __lowercase		:			Optional[Any]	    =       projection_dim
          __lowercase		:			int	    =       logit_scale_init_value
          __lowercase		:			Tuple	    =       return_dict
          __lowercase		:			Optional[Any]	    =       1.0




      @classmethod
      def 		snake_case_			(      cls		:       Any				,       _snake_case		:       Union[str, os.PathLike]				,       **_snake_case		:       int				):
          cls._set_token_in_kwargs(_snake_case				)

          __lowercase    ,      __lowercase		:			List[str]	    =       cls.get_config_dict(_snake_case				,       **_snake_case				)

          if "model_type" in config_dict and hasattr(cls				,       '''model_type'''				) and config_dict["model_type"] != cls.model_type:
              logger.warning(
                  F'You are using a model of type {config_dict["model_type"]} to instantiate a model of type '
                  F'{cls.model_type}. This is not supported for all configurations of models and can yield errors.'				)

          return cls.from_dict(_snake_case				,       **_snake_case				)




      @classmethod
      def 		snake_case_			(      cls		:       Dict				,       _snake_case		:       Dict				,       _snake_case		:       Dict				,       **_snake_case		:       str				):
          __lowercase		:			Union[str, Any]	    =       {}
          __lowercase		:			Dict	    =       text_config
          __lowercase		:			List[str]	    =       vision_config

          return cls.from_dict(_snake_case				,       **_snake_case				)




      def 		snake_case_			(      self		:       Optional[Any]				):
          __lowercase		:			Optional[Any]	    =       copy.deepcopy(self.__dict__				)
          __lowercase		:			List[Any]	    =       self.text_config.to_dict()
          __lowercase		:			List[str]	    =       self.vision_config.to_dict()
          __lowercase		:			Tuple	    =       self.__class__.model_type
          return output






class 						__lowerCAmelCase	(	lowerCAmelCase_      ):



      """simple docstring"""







      @property
      def 		snake_case_			(      self		:       Union[str, Any]				):
          return OrderedDict(
              [
                  ('''input_ids''', {0: '''batch''', 1: '''sequence'''}),
                  ('''pixel_values''', {0: '''batch''', 1: '''num_channels''', 2: '''height''', 3: '''width'''}),
                  ('''attention_mask''', {0: '''batch''', 1: '''sequence'''}),
              ]				)




      @property
      def 		snake_case_			(      self		:       int				):
          return OrderedDict(
              [
                  ('''logits_per_image''', {0: '''batch'''}),
                  ('''logits_per_text''', {0: '''batch'''}),
                  ('''text_embeds''', {0: '''batch'''}),
                  ('''image_embeds''', {0: '''batch'''}),
              ]				)




      @property
      def 		snake_case_			(      self		:       Any				):
          return 1E-4




      def 		snake_case_			(      self		:       Union[str, Any]				,       _snake_case		:       "ProcessorMixin"				,       _snake_case		:       int = -1				,       _snake_case		:       int = -1				,       _snake_case		:       Optional["TensorType"] = None				,       ):
          __lowercase		:			List[str]	    =       super().generate_dummy_inputs(
              processor.tokenizer				,       batch_size=_snake_case				,       seq_length=_snake_case				,       framework=_snake_case				)
          __lowercase		:			List[Any]	    =       super().generate_dummy_inputs(
              processor.image_processor				,       batch_size=_snake_case				,       framework=_snake_case				)
          return {**text_input_dict, **image_input_dict}




      @property
      def 		snake_case_			(      self		:       Dict				):
          return 14




import unittest
from queue import Empty
from threading import Thread

from transformers import AutoTokenizer, TextIteratorStreamer, TextStreamer, is_torch_available
from transformers.testing_utils import CaptureStdout, require_torch, torch_device

from ..test_modeling_common import ids_tensor


if is_torch_available():
 import torch

 from transformers import AutoModelForCausalLM


@require_torch
class 						__lowerCAmelCase	(	unittest.TestCase      ):



      """simple docstring"""







      def 		snake_case_			(      self		:       Union[str, Any]				):
          __lowercase		:			Optional[Any]	    =       AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''				)
          __lowercase		:			Any	    =       AutoModelForCausalLM.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''				).to(_snake_case				)
          __lowercase		:			List[str]	    =       -1

          __lowercase		:			Optional[Any]	    =       ids_tensor((1, 5)				,       vocab_size=model.config.vocab_size				).to(_snake_case				)
          __lowercase		:			Tuple	    =       model.generate(_snake_case				,       max_new_tokens=10				,       do_sample=_snake_case				)
          __lowercase		:			Optional[int]	    =       tokenizer.decode(greedy_ids[0]				)

          with CaptureStdout() as cs:
              __lowercase		:			Optional[Any]	    =       TextStreamer(_snake_case				)
              model.generate(_snake_case				,       max_new_tokens=10				,       do_sample=_snake_case				,       streamer=_snake_case				)
          # The greedy text should be printed to stdout, except for the final "\n" in the streamer
          __lowercase		:			List[Any]	    =       cs.out[:-1]

          self.assertEqual(_snake_case				,       _snake_case				)




      def 		snake_case_			(      self		:       Optional[int]				):
          __lowercase		:			Optional[int]	    =       AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''				)
          __lowercase		:			Optional[Any]	    =       AutoModelForCausalLM.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''				).to(_snake_case				)
          __lowercase		:			List[str]	    =       -1

          __lowercase		:			Tuple	    =       ids_tensor((1, 5)				,       vocab_size=model.config.vocab_size				).to(_snake_case				)
          __lowercase		:			List[Any]	    =       model.generate(_snake_case				,       max_new_tokens=10				,       do_sample=_snake_case				)
          __lowercase		:			Optional[Any]	    =       tokenizer.decode(greedy_ids[0]				)

          __lowercase		:			int	    =       TextIteratorStreamer(_snake_case				)
          __lowercase		:			int	    =       {'''input_ids''': input_ids, '''max_new_tokens''': 10, '''do_sample''': False, '''streamer''': streamer}
          __lowercase		:			Optional[Any]	    =       Thread(target=model.generate				,       kwargs=_snake_case				)
          thread.start()
          __lowercase		:			Optional[int]	    =       ''''''
          for new_text in streamer:
              streamer_text += new_text

          self.assertEqual(_snake_case				,       _snake_case				)




      def 		snake_case_			(      self		:       List[str]				):
          __lowercase		:			Dict	    =       AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''				)
          __lowercase		:			int	    =       AutoModelForCausalLM.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''				).to(_snake_case				)
          __lowercase		:			List[Any]	    =       -1

          __lowercase		:			List[Any]	    =       ids_tensor((1, 5)				,       vocab_size=model.config.vocab_size				).to(_snake_case				)
          __lowercase		:			str	    =       model.generate(_snake_case				,       max_new_tokens=10				,       do_sample=_snake_case				)
          __lowercase		:			List[Any]	    =       greedy_ids[:, input_ids.shape[1] :]
          __lowercase		:			Optional[Any]	    =       tokenizer.decode(new_greedy_ids[0]				)

          with CaptureStdout() as cs:
              __lowercase		:			int	    =       TextStreamer(_snake_case				,       skip_prompt=_snake_case				)
              model.generate(_snake_case				,       max_new_tokens=10				,       do_sample=_snake_case				,       streamer=_snake_case				)
          # The greedy text should be printed to stdout, except for the final "\n" in the streamer
          __lowercase		:			int	    =       cs.out[:-1]

          self.assertEqual(_snake_case				,       _snake_case				)




      def 		snake_case_			(      self		:       Any				):
          # Tests that we can pass `decode_kwargs` to the streamer to control how the tokens are decoded. Must be tested
          # with actual models -- the dummy models' tokenizers are not aligned with their models, and
          # `skip_special_tokens=True` has no effect on them
          __lowercase		:			Dict	    =       AutoTokenizer.from_pretrained('''distilgpt2'''				)
          __lowercase		:			List[str]	    =       AutoModelForCausalLM.from_pretrained('''distilgpt2'''				).to(_snake_case				)
          __lowercase		:			str	    =       -1

          __lowercase		:			Dict	    =       torch.ones((1, 5)				,       device=_snake_case				).long() * model.config.bos_token_id
          with CaptureStdout() as cs:
              __lowercase		:			Optional[Any]	    =       TextStreamer(_snake_case				,       skip_special_tokens=_snake_case				)
              model.generate(_snake_case				,       max_new_tokens=1				,       do_sample=_snake_case				,       streamer=_snake_case				)

          # The prompt contains a special token, so the streamer should not print it. As such, the output text, when
          # re-tokenized, must only contain one token
          __lowercase		:			Dict	    =       cs.out[:-1]  # Remove the final "\n"
          __lowercase		:			Any	    =       tokenizer(_snake_case				,       return_tensors='''pt'''				)
          self.assertEqual(streamer_text_tokenized.input_ids.shape				,       (1, 1)				)




      def 		snake_case_			(      self		:       Any				):
          __lowercase		:			List[str]	    =       AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''				)
          __lowercase		:			Union[str, Any]	    =       AutoModelForCausalLM.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''				).to(_snake_case				)
          __lowercase		:			Dict	    =       -1

          __lowercase		:			List[Any]	    =       ids_tensor((1, 5)				,       vocab_size=model.config.vocab_size				).to(_snake_case				)
          __lowercase		:			Tuple	    =       TextIteratorStreamer(_snake_case				,       timeout=0.0_01				)
          __lowercase		:			Tuple	    =       {'''input_ids''': input_ids, '''max_new_tokens''': 10, '''do_sample''': False, '''streamer''': streamer}
          __lowercase		:			Optional[int]	    =       Thread(target=model.generate				,       kwargs=_snake_case				)
          thread.start()

          # The streamer will timeout after 0.001 seconds, so an exception will be raised
          with self.assertRaises(_snake_case				):
              __lowercase		:			List[Any]	    =       ''''''
              for new_text in streamer:
                  streamer_text += new_text


