






'''simple docstring'''







import types
from typing import List, Optional, Tuple, Union

import torch
from transformers import CLIPTextModelWithProjection, CLIPTokenizer
from transformers.models.clip.modeling_clip import CLIPTextModelOutput

from diffusers.models import PriorTransformer
from diffusers.pipelines import DiffusionPipeline, StableDiffusionImageVariationPipeline
from diffusers.schedulers import UnCLIPScheduler
from diffusers.utils import logging, randn_tensor


__A      :       int      =	logging.get_logger(__name__)  # pylint: disable=invalid-name


def 						_UpperCamelCase							(     self							,			snake_case__							,			snake_case__							,			snake_case__							,			snake_case__							):
      __A		   =      image.to(device=snake_case__							)
      __A		   =      image  # take image as image_embeddings
      __A		   =      image_embeddings.unsqueeze(1							)

      # duplicate image embeddings for each generation per prompt, using mps friendly method
      __A   ,				__A   ,				__A		   =      image_embeddings.shape
      __A		   =      image_embeddings.repeat(1							,			snake_case__							,			1							)
      __A		   =      image_embeddings.view(bs_embed * num_images_per_prompt							,			snake_case__							,			-1							)

      if do_classifier_free_guidance:
            __A		   =      torch.zeros_like(snake_case__							)

            # For classifier free guidance, we need to do two forward passes.
            # Here we concatenate the unconditional and text embeddings into a single batch
            # to avoid doing two forward passes
            __A		   =      torch.cat([uncond_embeddings, image_embeddings]							)

      return image_embeddings






class 					a__    (			lowerCamelCase						):
     '''simple docstring'''







     def __init__(self    :	Any				,				lowercase__    :	PriorTransformer				,				lowercase__    :	CLIPTokenizer				,				lowercase__    :	CLIPTextModelWithProjection				,				lowercase__    :	UnCLIPScheduler				,				lowercase__    :	Optional[dict] = None				,				)  ->       Tuple:
           super().__init__()

           __A		   =      {"image_encoder": None} if decoder_pipe_kwargs is None else decoder_pipe_kwargs

           __A		   =      decoder_pipe_kwargs.get("torch_dtype"				,				lowercase__					) or prior.dtype

           __A		   =      StableDiffusionImageVariationPipeline.from_pretrained(
               "lambdalabs/sd-image-variations-diffusers"				,				**lowercase__					)

           # replace `_encode_image` method
           __A		   =      types.MethodType(_encode_image				,				self.decoder_pipe					)

           self.register_modules(
               prior=lowercase__				,				tokenizer=lowercase__				,				text_encoder=lowercase__				,				prior_scheduler=lowercase__				,				)






     def     _UpperCamelCase   (self    :	Dict				,				lowercase__    :	Optional[Any]				,				lowercase__    :	Optional[int]				,				lowercase__    :	List[str]				,				lowercase__    :	Optional[Any]				,				lowercase__    :	Optional[Union[CLIPTextModelOutput, Tuple]] = None				,				lowercase__    :	Optional[torch.Tensor] = None				,				)  ->       Any:
           if text_model_output is None:
                 __A		   =      len(lowercase__					) if isinstance(lowercase__				,				lowercase__					) else 1
                 # get prompt text embeddings
                 __A		   =      self.tokenizer(
                     lowercase__				,				padding="max_length"				,				max_length=self.tokenizer.model_max_length				,				return_tensors="pt"				,				)
                 __A		   =      text_inputs.input_ids
                 __A		   =      text_inputs.attention_mask.bool().to(lowercase__					)

                 if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
                       __A		   =      self.tokenizer.batch_decode(text_input_ids[:, self.tokenizer.model_max_length :]					)
                       logger.warning(
                           "The following part of your input was truncated because CLIP can only handle sequences up to"
                           f" {self.tokenizer.model_max_length} tokens: {removed_text}"					)
                       __A		   =      text_input_ids[:, : self.tokenizer.model_max_length]

                 __A		   =      self.text_encoder(text_input_ids.to(lowercase__					)					)

                 __A		   =      text_encoder_output.text_embeds
                 __A		   =      text_encoder_output.last_hidden_state

           else:
                 __A		   =      text_model_output[0].shape[0]
                 __A   ,				__A		   =      text_model_output[0], text_model_output[1]
                 __A		   =      text_attention_mask

           __A		   =      text_embeddings.repeat_interleave(lowercase__				,				dim=0					)
           __A		   =      text_encoder_hidden_states.repeat_interleave(lowercase__				,				dim=0					)
           __A		   =      text_mask.repeat_interleave(lowercase__				,				dim=0					)

           if do_classifier_free_guidance:
                 __A		   =      [""] * batch_size

                 __A		   =      self.tokenizer(
                     lowercase__				,				padding="max_length"				,				max_length=self.tokenizer.model_max_length				,				truncation=lowercase__				,				return_tensors="pt"				,				)
                 __A		   =      uncond_input.attention_mask.bool().to(lowercase__					)
                 __A		   =      self.text_encoder(uncond_input.input_ids.to(lowercase__					)					)

                 __A		   =      uncond_embeddings_text_encoder_output.text_embeds
                 __A		   =      uncond_embeddings_text_encoder_output.last_hidden_state

                 # duplicate unconditional embeddings for each generation per prompt, using mps friendly method

                 __A		   =      uncond_embeddings.shape[1]
                 __A		   =      uncond_embeddings.repeat(1				,				lowercase__					)
                 __A		   =      uncond_embeddings.view(batch_size * num_images_per_prompt				,				lowercase__					)

                 __A		   =      uncond_text_encoder_hidden_states.shape[1]
                 __A		   =      uncond_text_encoder_hidden_states.repeat(1				,				lowercase__				,				1					)
                 __A		   =      uncond_text_encoder_hidden_states.view(
                     batch_size * num_images_per_prompt				,				lowercase__				,				-1					)
                 __A		   =      uncond_text_mask.repeat_interleave(lowercase__				,				dim=0					)

                 # done duplicates

                 # For classifier free guidance, we need to do two forward passes.
                 # Here we concatenate the unconditional and text embeddings into a single batch
                 # to avoid doing two forward passes
                 __A		   =      torch.cat([uncond_embeddings, text_embeddings]					)
                 __A		   =      torch.cat([uncond_text_encoder_hidden_states, text_encoder_hidden_states]					)

                 __A		   =      torch.cat([uncond_text_mask, text_mask]					)

           return text_embeddings, text_encoder_hidden_states, text_mask






     @property
     def     _UpperCamelCase   (self    :	Optional[Any]					)  ->       Any:
           if self.device != torch.device("meta"					) or not hasattr(self.prior				,				"_hf_hook"					):
                 return self.device
           for module in self.prior.modules():
                 if (
                     hasattr(lowercase__				,				"_hf_hook"					)
                     and hasattr(module._hf_hook				,				"execution_device"					)
                     and module._hf_hook.execution_device is not None
                 ):
                       return torch.device(module._hf_hook.execution_device					)
           return self.device






     def     _UpperCamelCase   (self    :	Optional[int]				,				lowercase__    :	Tuple				,				lowercase__    :	Any				,				lowercase__    :	List[Any]				,				lowercase__    :	Union[str, Any]				,				lowercase__    :	List[Any]				,				lowercase__    :	str					)  ->       List[str]:
           if latents is None:
                 __A		   =      randn_tensor(lowercase__				,				generator=lowercase__				,				device=lowercase__				,				dtype=lowercase__					)
           else:
                 if latents.shape != shape:
                       raise ValueError(f"Unexpected latents shape, got {latents.shape}, expected {shape}"					)
                 __A		   =      latents.to(lowercase__					)

           __A		   =      latents * scheduler.init_noise_sigma
           return latents






     def     _UpperCamelCase   (self    :	Any				,				lowercase__    :	Optional[Union[str, torch.device]] = None					)  ->       Union[str, Any]:
           self.decoder_pipe.to(lowercase__					)
           super().to(lowercase__					)






     @torch.no_grad()
     def __call__(self    :	Tuple				,				lowercase__    :	Optional[Union[str, List[str]]] = None				,				lowercase__    :	Optional[int] = None				,				lowercase__    :	Optional[int] = None				,				lowercase__    :	int = 1				,				lowercase__    :	int = 25				,				lowercase__    :	Optional[torch.Generator] = None				,				lowercase__    :	Optional[torch.FloatTensor] = None				,				lowercase__    :	Optional[Union[CLIPTextModelOutput, Tuple]] = None				,				lowercase__    :	Optional[torch.Tensor] = None				,				lowercase__    :	float = 4.0				,				lowercase__    :	float = 8.0				,				lowercase__    :	int = 50				,				lowercase__    :	Optional[int] = 1				,				lowercase__    :	float = 0.0				,				lowercase__    :	Optional[str] = "pil"				,				lowercase__    :	bool = True				,				)  ->       List[Any]:
           if prompt is not None:
                 if isinstance(lowercase__				,				lowercase__					):
                       __A		   =      1
                 elif isinstance(lowercase__				,				lowercase__					):
                       __A		   =      len(lowercase__					)
                 else:
                       raise ValueError(f"`prompt` has to be of type `str` or `list` but is {type(lowercase__					)}"					)
           else:
                 __A		   =      text_model_output[0].shape[0]

           __A		   =      self._execution_device

           __A		   =      batch_size * num_images_per_prompt

           __A		   =      prior_guidance_scale > 1.0 or decoder_guidance_scale > 1.0

           __A   ,				__A   ,				__A		   =      self._encode_prompt(
               lowercase__				,				lowercase__				,				lowercase__				,				lowercase__				,				lowercase__				,				lowercase__					)

           # prior

           self.prior_scheduler.set_timesteps(lowercase__				,				device=lowercase__					)
           __A		   =      self.prior_scheduler.timesteps

           __A		   =      self.prior.config.embedding_dim

           __A		   =      self.prepare_latents(
               (batch_size, embedding_dim)				,				text_embeddings.dtype				,				lowercase__				,				lowercase__				,				lowercase__				,				self.prior_scheduler				,				)

           for i, t in enumerate(self.progress_bar(lowercase__					)					):
                 # expand the latents if we are doing classifier free guidance
                 __A		   =      torch.cat([prior_latents] * 2					) if do_classifier_free_guidance else prior_latents

                 __A		   =      self.prior(
                     lowercase__				,				timestep=lowercase__				,				proj_embedding=lowercase__				,				encoder_hidden_states=lowercase__				,				attention_mask=lowercase__				,				).predicted_image_embedding

                 if do_classifier_free_guidance:
                       __A   ,				__A		   =      predicted_image_embedding.chunk(2					)
                       __A		   =      predicted_image_embedding_uncond + prior_guidance_scale * (
                           predicted_image_embedding_text - predicted_image_embedding_uncond
                       )

                 if i + 1 == prior_timesteps_tensor.shape[0]:
                       __A		   =      None
                 else:
                       __A		   =      prior_timesteps_tensor[i + 1]

                 __A		   =      self.prior_scheduler.step(
                     lowercase__				,				timestep=lowercase__				,				sample=lowercase__				,				generator=lowercase__				,				prev_timestep=lowercase__				,				).prev_sample

           __A		   =      self.prior.post_process_latents(lowercase__					)

           __A		   =      prior_latents

           __A		   =      self.decoder_pipe(
               image=lowercase__				,				height=lowercase__				,				width=lowercase__				,				num_inference_steps=lowercase__				,				guidance_scale=lowercase__				,				generator=lowercase__				,				output_type=lowercase__				,				return_dict=lowercase__				,				num_images_per_prompt=lowercase__				,				eta=lowercase__				,				)
           return output










'''simple docstring'''







from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


__A      :       str      =	{'configuration_umt5': ['UMT5Config', 'UMT5OnnxConfig']}


try:
       if not is_torch_available():
              raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
       pass
else:
       __A      :       str      =	[
           'UMT5EncoderModel',
           'UMT5ForConditionalGeneration',
           'UMT5ForQuestionAnswering',
           'UMT5Model',
           'UMT5PreTrainedModel',
       ]

if TYPE_CHECKING:
       from .configuration_umta import UMTaConfig, UMTaOnnxConfig

       try:
              if not is_torch_available():
                     raise OptionalDependencyNotAvailable()
       except OptionalDependencyNotAvailable:
              pass
       else:
              from .modeling_umta import (
                  UMTaEncoderModel,
                  UMTaForConditionalGeneration,
                  UMTaForQuestionAnswering,
                  UMTaModel,
                  UMTaPreTrainedModel,
              )
else:
       import sys

       __A      :       Optional[int]      =	_LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)


