






import unittest

import numpy as np
import torch
from torch import nn
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModelWithProjection,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import KandinskyVaaPriorPipeline, PriorTransformer, UnCLIPScheduler
from diffusers.utils import torch_device
from diffusers.utils.testing_utils import enable_full_determinism, skip_mps

from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()
class        a       (					lowercase__				,			unittest.TestCase  ):



      """simple docstring"""


      a    :					Dict							=      KandinskyVaaPriorPipeline
      a    :					List[Any]							=      ['prompt']
      a    :					int							=      ['prompt', 'negative_prompt']
      a    :					List[str]							=      [
          'num_images_per_prompt',
          'generator',
          'num_inference_steps',
          'latents',
          'negative_prompt',
          'guidance_scale',
          'output_type',
          'return_dict',
      ]
      a    :					Any							=      False
      @property
      def       UpperCAmelCase ( self   :     Optional[Any]   )				-> List[str]:
        return 32
      @property
      def       UpperCAmelCase ( self   :     Tuple   )				-> Any:
        return 32
      @property
      def       UpperCAmelCase ( self   :     int   )				-> List[Any]:
        return self.time_input_dim
      @property
      def       UpperCAmelCase ( self   :     Tuple   )				-> Union[str, Any]:
        return self.time_input_dim * 4
      @property
      def       UpperCAmelCase ( self   :     List[Any]   )				-> int:
        return 100
      @property
      def       UpperCAmelCase ( self   :     Any   )				-> str:
        __UpperCAmelCase				:      Union[str, Any]	 =    CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""   )
        return tokenizer
      @property
      def       UpperCAmelCase ( self   :     Union[str, Any]   )				-> int:
        torch.manual_seed(0   )
        __UpperCAmelCase				:      Optional[Any]	 =    CLIPTextConfig(
            bos_token_id=0						, eos_token_id=2						, hidden_size=self.text_embedder_hidden_size						, projection_dim=self.text_embedder_hidden_size						, intermediate_size=37						, layer_norm_eps=1e-0_5						, num_attention_heads=4						, num_hidden_layers=5						, pad_token_id=1						, vocab_size=1000						, )
        return CLIPTextModelWithProjection(__lowercase   )
      @property
      def       UpperCAmelCase ( self   :     List[str]   )				-> Any:
        torch.manual_seed(0   )

        __UpperCAmelCase				:      List[Any]	 =    {
            """num_attention_heads""": 2,
            """attention_head_dim""": 12,
            """embedding_dim""": self.text_embedder_hidden_size,
            """num_layers""": 1,
        }

        __UpperCAmelCase				:      Any	 =    PriorTransformer(**__lowercase   )
        # clip_std and clip_mean is initialized to be 0 so PriorTransformer.post_process_latents will always return 0 - set clip_std to be 1 so it won't return 0
        __UpperCAmelCase				:      List[Any]	 =    nn.Parameter(torch.ones(model.clip_std.shape   )   )
        return model
      @property
      def       UpperCAmelCase ( self   :     Dict   )				-> Dict:
        torch.manual_seed(0   )
        __UpperCAmelCase				:      List[str]	 =    CLIPVisionConfig(
            hidden_size=self.text_embedder_hidden_size						, image_size=224						, projection_dim=self.text_embedder_hidden_size						, intermediate_size=37						, num_attention_heads=4						, num_channels=3						, num_hidden_layers=5						, patch_size=14						, )

        __UpperCAmelCase				:      Any	 =    CLIPVisionModelWithProjection(__lowercase   )
        return model
      @property
      def       UpperCAmelCase ( self   :     Any   )				-> List[Any]:
        __UpperCAmelCase				:      int	 =    CLIPImageProcessor(
            crop_size=224						, do_center_crop=__lowercase						, do_normalize=__lowercase						, do_resize=__lowercase						, image_mean=[0.48_145_466, 0.4_578_275, 0.40_821_073]						, image_std=[0.26_862_954, 0.26_130_258, 0.27_577_711]						, resample=3						, size=224						, )

        return image_processor
      def       UpperCAmelCase ( self   :     Optional[Any]   )				-> Dict:
        __UpperCAmelCase				:      str	 =    self.dummy_prior
        __UpperCAmelCase				:      List[str]	 =    self.dummy_image_encoder
        __UpperCAmelCase				:      List[str]	 =    self.dummy_text_encoder
        __UpperCAmelCase				:      Optional[int]	 =    self.dummy_tokenizer
        __UpperCAmelCase				:      Any	 =    self.dummy_image_processor

        __UpperCAmelCase				:      Tuple	 =    UnCLIPScheduler(
            variance_type="""fixed_small_log"""						, prediction_type="""sample"""						, num_train_timesteps=1000						, clip_sample=__lowercase						, clip_sample_range=10.0						, )

        __UpperCAmelCase				:      Optional[Any]	 =    {
            """prior""": prior,
            """image_encoder""": image_encoder,
            """text_encoder""": text_encoder,
            """tokenizer""": tokenizer,
            """scheduler""": scheduler,
            """image_processor""": image_processor,
        }

        return components
      def       UpperCAmelCase ( self   :     int						, __lowercase   :     Dict						, __lowercase   :     Tuple=0   )				-> int:
        if str(__lowercase   ).startswith("""mps"""   ):
          __UpperCAmelCase				:      Optional[int]	 =    torch.manual_seed(__lowercase   )
        else:
          __UpperCAmelCase				:      str	 =    torch.Generator(device=__lowercase   ).manual_seed(__lowercase   )
        __UpperCAmelCase				:      str	 =    {
            """prompt""": """horse""",
            """generator""": generator,
            """guidance_scale""": 4.0,
            """num_inference_steps""": 2,
            """output_type""": """np""",
        }
        return inputs
      def       UpperCAmelCase ( self   :     Union[str, Any]   )				-> Tuple:
        __UpperCAmelCase				:      Dict	 =    """cpu"""

        __UpperCAmelCase				:      Any	 =    self.get_dummy_components()

        __UpperCAmelCase				:      Optional[Any]	 =    self.pipeline_class(**__lowercase   )
        __UpperCAmelCase				:      int	 =    pipe.to(__lowercase   )

        pipe.set_progress_bar_config(disable=__lowercase   )

        __UpperCAmelCase				:      int	 =    pipe(**self.get_dummy_inputs(__lowercase   )   )
        __UpperCAmelCase				:      Any	 =    output.image_embeds

        __UpperCAmelCase				:      str	 =    pipe(
            **self.get_dummy_inputs(__lowercase   )						, return_dict=__lowercase						, )[0]

        __UpperCAmelCase				:      int	 =    image[0, -10:]
        __UpperCAmelCase				:      List[Any]	 =    image_from_tuple[0, -10:]

        assert image.shape == (1, 32)

        __UpperCAmelCase				:      str	 =    np.array(
            [-0.0_532, 1.7_120, 0.3_656, -1.0_852, -0.8_946, -1.1_756, 0.4_348, 0.2_482, 0.5_146, -0.1_156]   )

        assert np.abs(image_slice.flatten() - expected_slice   ).max() < 1e-2
        assert np.abs(image_from_tuple_slice.flatten() - expected_slice   ).max() < 1e-2
      @skip_mps
      def       UpperCAmelCase ( self   :     str   )				-> List[Any]:
        __UpperCAmelCase				:      Union[str, Any]	 =    torch_device == """cpu"""
        __UpperCAmelCase				:      List[Any]	 =    True
        __UpperCAmelCase				:      List[str]	 =    False

        self._test_inference_batch_single_identical(
            test_max_difference=__lowercase						, relax_max_difference=__lowercase						, test_mean_pixel_difference=__lowercase						, )




      @skip_mps
      def       UpperCAmelCase ( self   :     Dict   )				-> int:
        __UpperCAmelCase				:      Optional[int]	 =    torch_device == """cpu"""
        __UpperCAmelCase				:      Union[str, Any]	 =    False

        self._test_attention_slicing_forward_pass(
            test_max_difference=__lowercase						, test_mean_pixel_difference=__lowercase						, )






'''simple docstring'''





import warnings
from typing import List

import numpy as np

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding
from ...utils import is_flax_available, is_tf_available, is_torch_available




class 				UpperCAmelCase			(			_snake_case						):
 UpperCAmelCase				=					["image_processor", "tokenizer"]
 UpperCAmelCase				=					"OwlViTImageProcessor"
 UpperCAmelCase				=					("CLIPTokenizer", "CLIPTokenizerFast")


 def __init__(      self							:	List[str]			,       __lowerCamelCase							:	Optional[int]=None			,       __lowerCamelCase							:	Optional[Any]=None			,       **__lowerCamelCase							:	str ):
    UpperCAmelCase__	:int      =      None
    if "feature_extractor" in kwargs:
       warnings.warn(
           '''The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`'''
           ''' instead.'''			,       __lowerCamelCase			,       )
       UpperCAmelCase__	:Union[str, Any]      =      kwargs.pop('''feature_extractor''' )

    UpperCAmelCase__	:List[str]      =      image_processor if image_processor is not None else feature_extractor
    if image_processor is None:
       raise ValueError('''You need to specify an `image_processor`.''' )
    if tokenizer is None:
       raise ValueError('''You need to specify a `tokenizer`.''' )

    super().__init__(__lowerCamelCase			,       __lowerCamelCase )


 def __call__(      self							:	int			,       __lowerCamelCase							:	str=None			,       __lowerCamelCase							:	List[Any]=None			,       __lowerCamelCase							:	Dict=None			,       __lowerCamelCase							:	str="max_length"			,       __lowerCamelCase							:	Any="np"			,       **__lowerCamelCase							:	Tuple ):

    if text is None and query_images is None and images is None:
       raise ValueError(
           '''You have to specify at least one text or query image or image. All three cannot be none.''' )

    if text is not None:
       if isinstance(__lowerCamelCase			,       __lowerCamelCase ) or (isinstance(__lowerCamelCase			,       __lowerCamelCase ) and not isinstance(text[0]			,       __lowerCamelCase )):
          UpperCAmelCase__	:Any      =      [self.tokenizer(__lowerCamelCase			,       padding=__lowerCamelCase			,       return_tensors=__lowerCamelCase			,       **__lowerCamelCase )]

       elif isinstance(__lowerCamelCase			,       __lowerCamelCase ) and isinstance(text[0]			,       __lowerCamelCase ):
          UpperCAmelCase__	:Tuple      =      []

          # Maximum number of queries across batch
          UpperCAmelCase__	:List[str]      =      max([len(__lowerCamelCase ) for t in text] )

          # Pad all batch samples to max number of text queries
          for t in text:
             if len(__lowerCamelCase ) != max_num_queries:
                UpperCAmelCase__	:str      =      t + [''' '''] * (max_num_queries - len(__lowerCamelCase ))

             UpperCAmelCase__	:Tuple      =      self.tokenizer(__lowerCamelCase			,       padding=__lowerCamelCase			,       return_tensors=__lowerCamelCase			,       **__lowerCamelCase )
             encodings.append(__lowerCamelCase )
       else:
          raise TypeError('''Input text should be a string, a list of strings or a nested list of strings''' )

       if return_tensors == "np":
          UpperCAmelCase__	:List[Any]      =      np.concatenate([encoding['''input_ids'''] for encoding in encodings]			,       axis=0 )
          UpperCAmelCase__	:Any      =      np.concatenate([encoding['''attention_mask'''] for encoding in encodings]			,       axis=0 )

       elif return_tensors == "jax" and is_flax_available():
          import jax.numpy as jnp

          UpperCAmelCase__	:List[Any]      =      jnp.concatenate([encoding['''input_ids'''] for encoding in encodings]			,       axis=0 )
          UpperCAmelCase__	:Union[str, Any]      =      jnp.concatenate([encoding['''attention_mask'''] for encoding in encodings]			,       axis=0 )

       elif return_tensors == "pt" and is_torch_available():
          import torch

          UpperCAmelCase__	:Any      =      torch.cat([encoding['''input_ids'''] for encoding in encodings]			,       dim=0 )
          UpperCAmelCase__	:List[str]      =      torch.cat([encoding['''attention_mask'''] for encoding in encodings]			,       dim=0 )

       elif return_tensors == "tf" and is_tf_available():
          import tensorflow as tf

          UpperCAmelCase__	:Optional[Any]      =      tf.stack([encoding['''input_ids'''] for encoding in encodings]			,       axis=0 )
          UpperCAmelCase__	:int      =      tf.stack([encoding['''attention_mask'''] for encoding in encodings]			,       axis=0 )

       else:
          raise ValueError('''Target return tensor type could not be returned''' )

       UpperCAmelCase__	:List[Any]      =      BatchEncoding()
       UpperCAmelCase__	:Union[str, Any]      =      input_ids
       UpperCAmelCase__	:Dict      =      attention_mask

    if query_images is not None:
       UpperCAmelCase__	:Tuple      =      BatchEncoding()
       UpperCAmelCase__	:int      =      self.image_processor(
           __lowerCamelCase			,       return_tensors=__lowerCamelCase			,       **__lowerCamelCase ).pixel_values
       UpperCAmelCase__	:Optional[int]      =      query_pixel_values

    if images is not None:
       UpperCAmelCase__	:str      =      self.image_processor(__lowerCamelCase			,       return_tensors=__lowerCamelCase			,       **__lowerCamelCase )

    if text is not None and images is not None:
       UpperCAmelCase__	:Dict      =      image_features.pixel_values
       return encoding
    elif query_images is not None and images is not None:
       UpperCAmelCase__	:Tuple      =      image_features.pixel_values
       return encoding
    elif text is not None or query_images is not None:
       return encoding
    else:
       return BatchEncoding(data=dict(**__lowerCamelCase )			,       tensor_type=__lowerCamelCase )


 def        __SCREAMING_SNAKE_CASE					(      self							:	int			,       *__lowerCamelCase							:	Optional[int]			,       **__lowerCamelCase							:	int ):
    return self.image_processor.post_process(*__lowerCamelCase			,       **__lowerCamelCase )


 def        __SCREAMING_SNAKE_CASE					(      self							:	Dict			,       *__lowerCamelCase							:	Any			,       **__lowerCamelCase							:	Tuple ):
    return self.image_processor.post_process_object_detection(*__lowerCamelCase			,       **__lowerCamelCase )


 def        __SCREAMING_SNAKE_CASE					(      self							:	Union[str, Any]			,       *__lowerCamelCase							:	Tuple			,       **__lowerCamelCase							:	str ):
    return self.image_processor.post_process_image_guided_detection(*__lowerCamelCase			,       **__lowerCamelCase )


 def        __SCREAMING_SNAKE_CASE					(      self							:	int			,       *__lowerCamelCase							:	Optional[Any]			,       **__lowerCamelCase							:	Union[str, Any] ):
    return self.tokenizer.batch_decode(*__lowerCamelCase			,       **__lowerCamelCase )


 def        __SCREAMING_SNAKE_CASE					(      self							:	Optional[Any]			,       *__lowerCamelCase							:	Union[str, Any]			,       **__lowerCamelCase							:	Dict ):
    return self.tokenizer.decode(*__lowerCamelCase			,       **__lowerCamelCase )


 @property
 def        __SCREAMING_SNAKE_CASE					(      self							:	Union[str, Any] ):
    warnings.warn(
        '''`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead.'''			,       __lowerCamelCase			,       )
    return self.image_processor_class





 @property
 def        __SCREAMING_SNAKE_CASE					(      self							:	Union[str, Any] ):
    warnings.warn(
        '''`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead.'''			,       __lowerCamelCase			,       )
    return self.image_processor



