





import copy
import os
from typing import Union

from ...configuration_utils import PretrainedConfig
from ...utils import logging


_lowerCamelCase      							=   logging.get_logger(__name__)

_lowerCamelCase      							=   {
    'BAAI/AltCLIP': 'https://huggingface.co/BAAI/AltCLIP/resolve/main/config.json',
    # See all AltCLIP models at https://huggingface.co/models?filter=altclip
}







class       __A							(   lowerCamelCase__	):




      """simple docstring"""







      UpperCAmelCase__							=   """altclip_text_model"""


      def __init__(		self   ,				a__=25_0002   ,				a__=1024   ,				a__=24   ,				a__=16   ,				a__=4096   ,				a__="gelu"   ,				a__=0.1   ,				a__=0.1   ,				a__=514   ,				a__=1   ,				a__=0.02   ,				a__=0.02   ,				a__=1e-05   ,				a__=1   ,				a__=0   ,				a__=2   ,				a__="absolute"   ,				a__=True   ,				a__=768   ,				**a__   ,				):



             """simple docstring"""





             super().__init__(pad_token_id=a__   ,				bos_token_id=a__   ,				eos_token_id=a__   ,				**a__)

             _lowerCamelCase				:  Dict						=		vocab_size
             _lowerCamelCase				:  List[Any]						=		hidden_size
             _lowerCamelCase				:  Tuple						=		num_hidden_layers
             _lowerCamelCase				:  str						=		num_attention_heads
             _lowerCamelCase				:  str						=		hidden_act
             _lowerCamelCase				:  List[str]						=		intermediate_size
             _lowerCamelCase				:  Union[str, Any]						=		hidden_dropout_prob
             _lowerCamelCase				:  Any						=		attention_probs_dropout_prob
             _lowerCamelCase				:  List[str]						=		max_position_embeddings
             _lowerCamelCase				:  Optional[int]						=		type_vocab_size
             _lowerCamelCase				:  Any						=		initializer_range
             _lowerCamelCase				:  Union[str, Any]						=		initializer_factor
             _lowerCamelCase				:  Optional[Any]						=		layer_norm_eps
             _lowerCamelCase				:  Dict						=		position_embedding_type
             _lowerCamelCase				:  Optional[Any]						=		use_cache
             _lowerCamelCase				:  Optional[int]						=		project_dim







class       __A							(   lowerCamelCase__	):




      """simple docstring"""







      UpperCAmelCase__							=   """altclip_vision_model"""


      def __init__(		self   ,				a__=768   ,				a__=3072   ,				a__=512   ,				a__=12   ,				a__=12   ,				a__=3   ,				a__=224   ,				a__=32   ,				a__="quick_gelu"   ,				a__=1e-5   ,				a__=0.0   ,				a__=0.02   ,				a__=1.0   ,				**a__   ,				):



             """simple docstring"""





             super().__init__(**a__)

             _lowerCamelCase				:  List[str]						=		hidden_size
             _lowerCamelCase				:  Union[str, Any]						=		intermediate_size
             _lowerCamelCase				:  Optional[Any]						=		projection_dim
             _lowerCamelCase				:  Union[str, Any]						=		num_hidden_layers
             _lowerCamelCase				:  List[str]						=		num_attention_heads
             _lowerCamelCase				:  Optional[Any]						=		num_channels
             _lowerCamelCase				:  str						=		patch_size
             _lowerCamelCase				:  Dict						=		image_size
             _lowerCamelCase				:  int						=		initializer_range
             _lowerCamelCase				:  Dict						=		initializer_factor
             _lowerCamelCase				:  Optional[Any]						=		attention_dropout
             _lowerCamelCase				:  int						=		layer_norm_eps
             _lowerCamelCase				:  List[str]						=		hidden_act



      @classmethod
      def 					__snake_case      (		cls   ,				a__   ,				**a__):



             """simple docstring"""





             cls._set_token_in_kwargs(a__)

             _lowerCamelCase,	_lowerCamelCase				:  List[str]						=		cls.get_config_dict(a__   ,				**a__)

             # get the vision config dict if we are loading from AltCLIPConfig
             if config_dict.get('''model_type''') == "altclip":
                    _lowerCamelCase				:  List[Any]						=		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(a__   ,				**a__)







class       __A							(   lowerCamelCase__	):




      """simple docstring"""







      UpperCAmelCase__							=   """altclip"""
      UpperCAmelCase__							=   True


      def __init__(		self   ,				a__=None   ,				a__=None   ,				a__=768   ,				a__=2.6592   ,				**a__):



             """simple docstring"""





             _lowerCamelCase				:  str						=		kwargs.pop('''text_config_dict'''   ,				a__)
             _lowerCamelCase				:  Optional[int]						=		kwargs.pop('''vision_config_dict'''   ,				a__)

             super().__init__(**a__)

             # Instead of simply assigning `[text|vision]_config_dict` to `[text|vision]_config`, we use the values in
             # `[text|vision]_config_dict` to update the values in `[text|vision]_config`. The values should be same in most
             # cases, but we don't want to break anything regarding `_config_dict` that existed before commit `8827e1b2`.
             if text_config_dict is not None:
                    if text_config is None:
                           _lowerCamelCase				:  Dict						=		{}

                    # This is the complete result when using `text_config_dict`.
                    _lowerCamelCase				:  List[Any]						=		AltCLIPTextConfig(**a__).to_dict()

                    # Give a warning if the values exist in both `_text_config_dict` and `text_config` but being different.
                    for key, value in _text_config_dict.items():
                           if key in text_config and value != text_config[key] and key not in ["transformers_version"]:
                                  # If specified in `text_config_dict`
                                  if key in text_config_dict:
                                         _lowerCamelCase				:  Optional[Any]						=		(
                                             F"""`{key}` is found in both `text_config_dict` and `text_config` but with different values. """
                                             F"""The value `text_config_dict[\"{key}\"]` will be used instead."""
                                         )
                                  # If inferred from default argument values (just to be super careful)
                                  else:
                                         _lowerCamelCase				:  Optional[int]						=		(
                                             F"""`text_config_dict` is provided which will be used to initialize `AltCLIPTextConfig`. The """
                                             F"""value `text_config[\"{key}\"]` will be overriden."""
                                         )
                                  logger.warning(a__)

            # Update all values in `text_config` with the ones in `_text_config_dict`.
                    text_config.update(_text_config_dict)

             if vision_config_dict is not None:
                    if vision_config is None:
                           _lowerCamelCase				:  Tuple						=		{}

                    # This is the complete result when using `vision_config_dict`.
                    _lowerCamelCase				:  Any						=		AltCLIPVisionConfig(**a__).to_dict()
                    # convert keys to string instead of integer
                    if "id2label" in _vision_config_dict:
                           _lowerCamelCase				:  Union[str, Any]						=		{
                               str(a__): value for key, value in _vision_config_dict['''id2label'''].items()
                           }

                    # Give a warning if the values exist in both `_vision_config_dict` and `vision_config` but being different.
                    for key, value in _vision_config_dict.items():
                           if key in vision_config and value != vision_config[key] and key not in ["transformers_version"]:
                                  # If specified in `vision_config_dict`
                                  if key in vision_config_dict:
                                         _lowerCamelCase				:  int						=		(
                                             F"""`{key}` is found in both `vision_config_dict` and `vision_config` but with different """
                                             F"""values. The value `vision_config_dict[\"{key}\"]` will be used instead."""
                                         )
                                  # If inferred from default argument values (just to be super careful)
                                  else:
                                         _lowerCamelCase				:  Optional[int]						=		(
                                             F"""`vision_config_dict` is provided which will be used to initialize `AltCLIPVisionConfig`. """
                                             F"""The value `vision_config[\"{key}\"]` will be overriden."""
                                         )
                                  logger.warning(a__)

            # Update all values in `vision_config` with the ones in `_vision_config_dict`.
                    vision_config.update(_vision_config_dict)

             if text_config is None:
                    _lowerCamelCase				:  str						=		{}
                    logger.info('''`text_config` is `None`. Initializing the `AltCLIPTextConfig` with default values.''')

             if vision_config is None:
                    _lowerCamelCase				:  Any						=		{}
                    logger.info('''`vision_config` is `None`. initializing the `AltCLIPVisionConfig` with default values.''')

             _lowerCamelCase				:  Optional[int]						=		AltCLIPTextConfig(**a__)
             _lowerCamelCase				:  Union[str, Any]						=		AltCLIPVisionConfig(**a__)

             _lowerCamelCase				:  int						=		projection_dim
             _lowerCamelCase				:  Union[str, Any]						=		logit_scale_init_value
             _lowerCamelCase				:  List[str]						=		1.0


      @classmethod
      def 					__snake_case      (		cls   ,				a__   ,				a__   ,				**a__):



             """simple docstring"""





             return cls(text_config=text_config.to_dict()   ,				vision_config=vision_config.to_dict()   ,				**a__)



      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase				:  int						=		copy.deepcopy(self.__dict__)
             _lowerCamelCase				:  Any						=		self.text_config.to_dict()
             _lowerCamelCase				:  List[Any]						=		self.vision_config.to_dict()
             _lowerCamelCase				:  List[Any]						=		self.__class__.model_type
             return output








from typing import List, Optional, Tuple, Union

import PIL
import torch
from torchvision import transforms

from diffusers.pipeline_utils import DiffusionPipeline, ImagePipelineOutput
from diffusers.schedulers import DDIMScheduler
from diffusers.utils import randn_tensor


_lowerCamelCase      							=   transforms.Compose(
    [
        transforms.Resize((256, 256)),
        transforms.ToTensor(),
        transforms.Normalize([0.5], [0.5]),
    ]
)




def  __UpperCAmelCase(   lowercase_   ):
       if isinstance(lowercase_			,     torch.Tensor   ):
              return image
       elif isinstance(lowercase_			,     PIL.Image.Image   ):
              _lowerCamelCase				:  List[str]						=		[image]

       _lowerCamelCase				:  Optional[int]						=		[trans(img.convert('''RGB'''   )   ) for img in image]
       _lowerCamelCase				:  Dict						=		torch.stack(lowercase_   )
       return image







class       __A							(   lowerCamelCase__	):




      """simple docstring"""







      def __init__(		self   ,				a__   ,				a__):



             """simple docstring"""





             super().__init__()

             # make sure scheduler can always be converted to DDIM
             _lowerCamelCase				:  Optional[Any]						=		DDIMScheduler.from_config(scheduler.config)

             self.register_modules(unet=a__   ,				scheduler=a__)


      def 					__snake_case      (		self   ,				a__):



             """simple docstring"""





             if strength < 0 or strength > 1:
                    raise ValueError(F"""The value of strength should in [0.0, 1.0] but is {strength}""")


      def 					__snake_case      (		self   ,				a__   ,				a__   ,				a__):



             """simple docstring"""





             _lowerCamelCase				:  List[str]						=		min(int(num_inference_steps * strength)   ,				a__)

             _lowerCamelCase				:  Tuple						=		max(num_inference_steps - init_timestep   ,				0)
             _lowerCamelCase				:  int						=		self.scheduler.timesteps[t_start:]

             return timesteps, num_inference_steps - t_start


      def 					__snake_case      (		self   ,				a__   ,				a__   ,				a__   ,				a__   ,				a__   ,				a__=None):



             """simple docstring"""





             if not isinstance(a__   ,				(torch.Tensor, PIL.Image.Image, list)):
                    raise ValueError(
                        F"""`image` has to be of type `torch.Tensor`, `PIL.Image.Image` or list but is {type(a__)}""")

             _lowerCamelCase				:  Any						=		image.to(device=a__   ,				dtype=a__)

             if isinstance(a__   ,				a__) and len(a__) != batch_size:
                    raise ValueError(
                        F"""You have passed a list of generators of length {len(a__)}, but requested an effective batch"""
                        F""" size of {batch_size}. Make sure the batch size matches the length of the generators.""")

             _lowerCamelCase				:  Any						=		init_latents.shape
             _lowerCamelCase				:  Tuple						=		randn_tensor(a__   ,				generator=a__   ,				device=a__   ,				dtype=a__)

             # get latents
             print('''add noise to latents at timestep'''   ,				a__)
             _lowerCamelCase				:  List[Any]						=		self.scheduler.add_noise(a__   ,				a__   ,				a__)
             _lowerCamelCase				:  List[Any]						=		init_latents

             return latents



      @torch.no_grad()
      def __call__(		self   ,				a__ = None   ,				a__ = 0.8   ,				a__ = 1   ,				a__ = None   ,				a__ = 0.0   ,				a__ = 50   ,				a__ = None   ,				a__ = "pil"   ,				a__ = True   ,				):



             """simple docstring"""





             self.check_inputs(a__)

             # 2. Preprocess image
             _lowerCamelCase				:  Union[str, Any]						=		preprocess(a__)

             # 3. set timesteps
             self.scheduler.set_timesteps(a__   ,				device=self.device)
             _lowerCamelCase,	_lowerCamelCase				:  Any						=		self.get_timesteps(a__   ,				a__   ,				self.device)
             _lowerCamelCase				:  Union[str, Any]						=		timesteps[:1].repeat(a__)

             # 4. Prepare latent variables
             _lowerCamelCase				:  int						=		self.prepare_latents(a__   ,				a__   ,				a__   ,				self.unet.dtype   ,				self.device   ,				a__)
             _lowerCamelCase				:  int						=		latents

             # 5. Denoising loop
             for t in self.progress_bar(a__):
                    # 1. predict noise model_output
                    _lowerCamelCase				:  Dict						=		self.unet(a__   ,				a__).sample

                    # 2. predict previous mean of image x_t-1 and add variance depending on eta
                    # eta corresponds to η in paper and should be between [0, 1]
                    # do x_t -> x_t-1
                    _lowerCamelCase				:  Tuple						=		self.scheduler.step(
                        a__   ,				a__   ,				a__   ,				eta=a__   ,				use_clipped_model_output=a__   ,				generator=a__   ,				).prev_sample

             _lowerCamelCase				:  str						=		(image / 2 + 0.5).clamp(0   ,				1)
             _lowerCamelCase				:  Dict						=		image.cpu().permute(0   ,				2   ,				3   ,				1).numpy()
             if output_type == "pil":
                    _lowerCamelCase				:  Any						=		self.numpy_to_pil(a__)

             if not return_dict:
                    return (image, latent_timestep.item())

             return ImagePipelineOutput(images=a__)

