



"""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


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


def  __SCREAMING_SNAKE_CASE (  self   ,				__UpperCAmelCase   ,				__UpperCAmelCase   ,				__UpperCAmelCase   ,				__UpperCAmelCase      ):
		_lowercase				: Optional[Any]     =   image.to(device=__UpperCAmelCase      )
		_lowercase				: List[str]     =   image  # take image as image_embeddings
		_lowercase				: str     =   image_embeddings.unsqueeze(1      )

		# duplicate image embeddings for each generation per prompt, using mps friendly method
		_lowercase     ,  _lowercase     ,  _lowercase				: Any     =   image_embeddings.shape
		_lowercase				: Optional[int]     =   image_embeddings.repeat(1   ,				__UpperCAmelCase   ,				1      )
		_lowercase				: Any     =   image_embeddings.view(bs_embed * num_images_per_prompt   ,				__UpperCAmelCase   ,				-1      )

		if do_classifier_free_guidance:
				_lowercase				: List[Any]     =   torch.zeros_like(__UpperCAmelCase      )

				# 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
				_lowercase				: Tuple     =   torch.cat([uncond_embeddings, image_embeddings]      )

		return image_embeddings





class   UpperCamelCase	(     snake_case       ):




	"""simple docstring"""






	def __init__(     self				,UpperCAmelCase_				,UpperCAmelCase_				,UpperCAmelCase_				,UpperCAmelCase_				,UpperCAmelCase_ = None				,):
			super().__init__()

			_lowercase				: str     =   {"""image_encoder""": None} if decoder_pipe_kwargs is None else decoder_pipe_kwargs

			_lowercase				: List[Any]     =   decoder_pipe_kwargs.get("""torch_dtype"""				,UpperCAmelCase_   ) or prior.dtype

			_lowercase				: List[Any]     =   StableDiffusionImageVariationPipeline.from_pretrained(
			    """lambdalabs/sd-image-variations-diffusers"""				,**UpperCAmelCase_   )

			# replace `_encode_image` method
			_lowercase				: Optional[Any]     =   types.MethodType(_encode_image				,self.decoder_pipe   )

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






	def    lowerCamelCase__					(     self				,UpperCAmelCase_				,UpperCAmelCase_				,UpperCAmelCase_				,UpperCAmelCase_				,UpperCAmelCase_ = None				,UpperCAmelCase_ = None				,):
			if text_model_output is None:
					_lowercase				: Tuple     =   len(UpperCAmelCase_   ) if isinstance(UpperCAmelCase_				,UpperCAmelCase_   ) else 1
					# get prompt text embeddings
					_lowercase				: List[str]     =   self.tokenizer(
					    UpperCAmelCase_				,padding="""max_length"""				,max_length=self.tokenizer.model_max_length				,return_tensors="""pt"""				,)
					_lowercase				: str     =   text_inputs.input_ids
					_lowercase				: int     =   text_inputs.attention_mask.bool().to(UpperCAmelCase_   )

					if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
							_lowercase				: Optional[Any]     =   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}"""   )
							_lowercase				: str     =   text_input_ids[:, : self.tokenizer.model_max_length]

					_lowercase				: int     =   self.text_encoder(text_input_ids.to(UpperCAmelCase_   )   )

					_lowercase				: Optional[Any]     =   text_encoder_output.text_embeds
					_lowercase				: Any     =   text_encoder_output.last_hidden_state

			else:
					_lowercase				: Tuple     =   text_model_output[0].shape[0]
					_lowercase     ,  _lowercase				: Dict     =   text_model_output[0], text_model_output[1]
					_lowercase				: str     =   text_attention_mask

			_lowercase				: Tuple     =   text_embeddings.repeat_interleave(UpperCAmelCase_				,dim=0   )
			_lowercase				: str     =   text_encoder_hidden_states.repeat_interleave(UpperCAmelCase_				,dim=0   )
			_lowercase				: Tuple     =   text_mask.repeat_interleave(UpperCAmelCase_				,dim=0   )

			if do_classifier_free_guidance:
					_lowercase				: Optional[Any]     =   [""""""] * batch_size

					_lowercase				: Dict     =   self.tokenizer(
					    UpperCAmelCase_				,padding="""max_length"""				,max_length=self.tokenizer.model_max_length				,truncation=UpperCAmelCase_				,return_tensors="""pt"""				,)
					_lowercase				: Optional[Any]     =   uncond_input.attention_mask.bool().to(UpperCAmelCase_   )
					_lowercase				: List[str]     =   self.text_encoder(uncond_input.input_ids.to(UpperCAmelCase_   )   )

					_lowercase				: str     =   uncond_embeddings_text_encoder_output.text_embeds
					_lowercase				: Any     =   uncond_embeddings_text_encoder_output.last_hidden_state

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

					_lowercase				: Union[str, Any]     =   uncond_embeddings.shape[1]
					_lowercase				: Dict     =   uncond_embeddings.repeat(1				,UpperCAmelCase_   )
					_lowercase				: int     =   uncond_embeddings.view(batch_size * num_images_per_prompt				,UpperCAmelCase_   )

					_lowercase				: Dict     =   uncond_text_encoder_hidden_states.shape[1]
					_lowercase				: Tuple     =   uncond_text_encoder_hidden_states.repeat(1				,UpperCAmelCase_				,1   )
					_lowercase				: Tuple     =   uncond_text_encoder_hidden_states.view(
					    batch_size * num_images_per_prompt				,UpperCAmelCase_				,-1   )
					_lowercase				: List[Any]     =   uncond_text_mask.repeat_interleave(UpperCAmelCase_				,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
					_lowercase				: Any     =   torch.cat([uncond_embeddings, text_embeddings]   )
					_lowercase				: Tuple     =   torch.cat([uncond_text_encoder_hidden_states, text_encoder_hidden_states]   )

					_lowercase				: Union[str, Any]     =   torch.cat([uncond_text_mask, text_mask]   )

			return text_embeddings, text_encoder_hidden_states, text_mask






	@property
	def    lowerCamelCase__					(     self   ):
			if self.device != torch.device("""meta"""   ) or not hasattr(self.prior				,"""_hf_hook"""   ):
					return self.device
			for module in self.prior.modules():
					if (
					    hasattr(UpperCAmelCase_				,"""_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    lowerCamelCase__					(     self				,UpperCAmelCase_				,UpperCAmelCase_				,UpperCAmelCase_				,UpperCAmelCase_				,UpperCAmelCase_				,UpperCAmelCase_   ):
			if latents is None:
					_lowercase				: Optional[Any]     =   randn_tensor(UpperCAmelCase_				,generator=UpperCAmelCase_				,device=UpperCAmelCase_				,dtype=UpperCAmelCase_   )
			else:
					if latents.shape != shape:
							raise ValueError(f"""Unexpected latents shape, got {latents.shape}, expected {shape}"""   )
					_lowercase				: Optional[Any]     =   latents.to(UpperCAmelCase_   )

			_lowercase				: Optional[Any]     =   latents * scheduler.init_noise_sigma
			return latents






	def    lowerCamelCase__					(     self				,UpperCAmelCase_ = None   ):
			self.decoder_pipe.to(UpperCAmelCase_   )
			super().to(UpperCAmelCase_   )






	@torch.no_grad()
	def __call__(     self				,UpperCAmelCase_ = None				,UpperCAmelCase_ = None				,UpperCAmelCase_ = None				,UpperCAmelCase_ = 1				,UpperCAmelCase_ = 25				,UpperCAmelCase_ = None				,UpperCAmelCase_ = None				,UpperCAmelCase_ = None				,UpperCAmelCase_ = None				,UpperCAmelCase_ = 4.0				,UpperCAmelCase_ = 8.0				,UpperCAmelCase_ = 50				,UpperCAmelCase_ = 1				,UpperCAmelCase_ = 0.0				,UpperCAmelCase_ = "pil"				,UpperCAmelCase_ = True				,):
			if prompt is not None:
					if isinstance(UpperCAmelCase_				,UpperCAmelCase_   ):
							_lowercase				: Optional[Any]     =   1
					elif isinstance(UpperCAmelCase_				,UpperCAmelCase_   ):
							_lowercase				: Union[str, Any]     =   len(UpperCAmelCase_   )
					else:
							raise ValueError(f"""`prompt` has to be of type `str` or `list` but is {type(UpperCAmelCase_   )}"""   )
			else:
					_lowercase				: List[str]     =   text_model_output[0].shape[0]

			_lowercase				: List[str]     =   self._execution_device

			_lowercase				: str     =   batch_size * num_images_per_prompt

			_lowercase				: List[Any]     =   prior_guidance_scale > 1.0 or decoder_guidance_scale > 1.0

			_lowercase     ,  _lowercase     ,  _lowercase				: List[Any]     =   self._encode_prompt(
			    UpperCAmelCase_				,UpperCAmelCase_				,UpperCAmelCase_				,UpperCAmelCase_				,UpperCAmelCase_				,UpperCAmelCase_   )

			# prior

			self.prior_scheduler.set_timesteps(UpperCAmelCase_				,device=UpperCAmelCase_   )
			_lowercase				: List[Any]     =   self.prior_scheduler.timesteps

			_lowercase				: Union[str, Any]     =   self.prior.config.embedding_dim

			_lowercase				: Tuple     =   self.prepare_latents(
			    (batch_size, embedding_dim)				,text_embeddings.dtype				,UpperCAmelCase_				,UpperCAmelCase_				,UpperCAmelCase_				,self.prior_scheduler				,)

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

					_lowercase				: Union[str, Any]     =   self.prior(
					    UpperCAmelCase_				,timestep=UpperCAmelCase_				,proj_embedding=UpperCAmelCase_				,encoder_hidden_states=UpperCAmelCase_				,attention_mask=UpperCAmelCase_				,).predicted_image_embedding

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

					if i + 1 == prior_timesteps_tensor.shape[0]:
							_lowercase				: List[Any]     =   None
					else:
							_lowercase				: str     =   prior_timesteps_tensor[i + 1]

					_lowercase				: Optional[Any]     =   self.prior_scheduler.step(
					    UpperCAmelCase_				,timestep=UpperCAmelCase_				,sample=UpperCAmelCase_				,generator=UpperCAmelCase_				,prev_timestep=UpperCAmelCase_				,).prev_sample

			_lowercase				: int     =   self.prior.post_process_latents(UpperCAmelCase_   )

			_lowercase				: Union[str, Any]     =   prior_latents

			_lowercase				: Union[str, Any]     =   self.decoder_pipe(
			    image=UpperCAmelCase_				,height=UpperCAmelCase_				,width=UpperCAmelCase_				,num_inference_steps=UpperCAmelCase_				,guidance_scale=UpperCAmelCase_				,generator=UpperCAmelCase_				,output_type=UpperCAmelCase_				,return_dict=UpperCAmelCase_				,num_images_per_prompt=UpperCAmelCase_				,eta=UpperCAmelCase_				,)
			return output











'''simple docstring'''






from dataclasses import dataclass
from typing import Optional, Tuple, Union

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .embeddings import GaussianFourierProjection, TimestepEmbedding, Timesteps
from .modeling_utils import ModelMixin
from .unet_ad_blocks import UNetMidBlockaD, get_down_block, get_up_block




@dataclass
class a_     (     UpperCamelCase_				):
   _snake_case	   =				42




class a_     (     UpperCamelCase_		,   UpperCamelCase_				):







   @register_to_config
   def __init__(self       , __a = None       , __a = 3       , __a = 3       , __a = False       , __a = "positional"       , __a = 0       , __a = True       , __a = ("DownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D")       , __a = ("AttnUpBlock2D", "AttnUpBlock2D", "AttnUpBlock2D", "UpBlock2D")       , __a = (2_2_4, 4_4_8, 6_7_2, 8_9_6)       , __a = 2       , __a = 1       , __a = 1       , __a = "conv"       , __a = "conv"       , __a = "silu"       , __a = 8       , __a = 3_2       , __a = 1E-5       , __a = "default"       , __a = True       , __a = None       , __a = None       , )      ->   List[str]:
      """simple docstring"""






      super().__init__()

      __snake_case					:   Optional[int]  				=			sample_size
      __snake_case					:   Union[str, Any]  				=			block_out_channels[0] * 4

      # Check inputs
      if len(__a) != len(__a):
         raise ValueError(
             F"""Must provide the same number of `down_block_types` as `up_block_types`. `down_block_types`: {down_block_types}. `up_block_types`: {up_block_types}.""")

      if len(__a) != len(__a):
         raise ValueError(
             F"""Must provide the same number of `block_out_channels` as `down_block_types`. `block_out_channels`: {block_out_channels}. `down_block_types`: {down_block_types}.""")

      # input
      __snake_case					:   Union[str, Any]  				=			nn.Convad(__a       , block_out_channels[0]       , kernel_size=3       , padding=(1, 1))

      # time
      if time_embedding_type == "fourier":
         __snake_case					:   str  				=			GaussianFourierProjection(embedding_size=block_out_channels[0]       , scale=1_6)
         __snake_case					:   Tuple  				=			2 * block_out_channels[0]
      elif time_embedding_type == "positional":
         __snake_case					:   List[str]  				=			Timesteps(block_out_channels[0]       , __a       , __a)
         __snake_case					:   Optional[Any]  				=			block_out_channels[0]

      __snake_case					:   List[Any]  				=			TimestepEmbedding(__a       , __a)

      # class embedding
      if class_embed_type is None and num_class_embeds is not None:
         __snake_case					:   Optional[Any]  				=			nn.Embedding(__a       , __a)
      elif class_embed_type == "timestep":
         __snake_case					:   int  				=			TimestepEmbedding(__a       , __a)
      elif class_embed_type == "identity":
         __snake_case					:   Any  				=			nn.Identity(__a       , __a)
      else:
         __snake_case					:   Dict  				=			None

      __snake_case					:   Union[str, Any]  				=			nn.ModuleList([])
      __snake_case					:   Dict  				=			None
      __snake_case					:   Union[str, Any]  				=			nn.ModuleList([])

      # down
      __snake_case					:   List[str]  				=			block_out_channels[0]
      for i, down_block_type in enumerate(__a):
         __snake_case					:   List[str]  				=			output_channel
         __snake_case					:   Tuple  				=			block_out_channels[i]
         __snake_case					:   Union[str, Any]  				=			i == len(__a) - 1

         __snake_case					:   Dict  				=			get_down_block(
             __a       , num_layers=__a       , in_channels=__a       , out_channels=__a       , temb_channels=__a       , add_downsample=not is_final_block       , resnet_eps=__a       , resnet_act_fn=__a       , resnet_groups=__a       , attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel       , downsample_padding=__a       , resnet_time_scale_shift=__a       , downsample_type=__a       , )
         self.down_blocks.append(__a)

      # mid
      __snake_case					:   List[str]  				=			UNetMidBlockaD(
          in_channels=block_out_channels[-1]       , temb_channels=__a       , resnet_eps=__a       , resnet_act_fn=__a       , output_scale_factor=__a       , resnet_time_scale_shift=__a       , attention_head_dim=attention_head_dim if attention_head_dim is not None else block_out_channels[-1]       , resnet_groups=__a       , add_attention=__a       , )

      # up
      __snake_case					:   List[Any]  				=			list(reversed(__a))
      __snake_case					:   Any  				=			reversed_block_out_channels[0]
      for i, up_block_type in enumerate(__a):
         __snake_case					:   str  				=			output_channel
         __snake_case					:   List[Any]  				=			reversed_block_out_channels[i]
         __snake_case					:   Optional[Any]  				=			reversed_block_out_channels[min(i + 1       , len(__a) - 1)]

         __snake_case					:   Optional[int]  				=			i == len(__a) - 1

         __snake_case					:   List[Any]  				=			get_up_block(
             __a       , num_layers=layers_per_block + 1       , in_channels=__a       , out_channels=__a       , prev_output_channel=__a       , temb_channels=__a       , add_upsample=not is_final_block       , resnet_eps=__a       , resnet_act_fn=__a       , resnet_groups=__a       , attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel       , resnet_time_scale_shift=__a       , upsample_type=__a       , )
         self.up_blocks.append(__a)
         __snake_case					:   Dict  				=			output_channel

      # out
      __snake_case					:   Tuple  				=			norm_num_groups if norm_num_groups is not None else min(block_out_channels[0] // 4       , 3_2)
      __snake_case					:   Tuple  				=			nn.GroupNorm(num_channels=block_out_channels[0]       , num_groups=__a       , eps=__a)
      __snake_case					:   List[str]  				=			nn.SiLU()
      __snake_case					:   List[Any]  				=			nn.Convad(block_out_channels[0]       , __a       , kernel_size=3       , padding=1)







   def      SCREAMING_SNAKE_CASE__       (self       , __a       , __a       , __a = None       , __a = True       , )      ->   Union[UNetaDOutput, Tuple]:
      """simple docstring"""






      if self.config.center_input_sample:
         __snake_case					:   Optional[Any]  				=			2 * sample - 1.0

      # 1. time
      __snake_case					:   Any  				=			timestep
      if not torch.is_tensor(__a):
         __snake_case					:   Dict  				=			torch.tensor([timesteps]       , dtype=torch.long       , device=sample.device)
      elif torch.is_tensor(__a) and len(timesteps.shape) == 0:
         __snake_case					:   Dict  				=			timesteps[None].to(sample.device)

      # broadcast to batch dimension in a way that's compatible with ONNX/Core ML
      __snake_case					:   Optional[int]  				=			timesteps * torch.ones(sample.shape[0]       , dtype=timesteps.dtype       , device=timesteps.device)

      __snake_case					:   Dict  				=			self.time_proj(__a)

      # timesteps does not contain any weights and will always return f32 tensors
      # but time_embedding might actually be running in fp16. so we need to cast here.
      # there might be better ways to encapsulate this.
      __snake_case					:   Optional[Any]  				=			t_emb.to(dtype=self.dtype)
      __snake_case					:   List[Any]  				=			self.time_embedding(__a)

      if self.class_embedding is not None:
         if class_labels is None:
            raise ValueError('class_labels should be provided when doing class conditioning')

         if self.config.class_embed_type == "timestep":
            __snake_case					:   str  				=			self.time_proj(__a)

         __snake_case					:   List[str]  				=			self.class_embedding(__a).to(dtype=self.dtype)
         __snake_case					:   Optional[Any]  				=			emb + class_emb

      # 2. pre-process
      __snake_case					:   List[str]  				=			sample
      __snake_case					:   Optional[Any]  				=			self.conv_in(__a)

      # 3. down
      __snake_case					:   List[str]  				=			(sample,)
      for downsample_block in self.down_blocks:
         if hasattr(__a       , 'skip_conv'):
            __snake_case				,__snake_case				,__snake_case					:   List[str]  				=			downsample_block(
                hidden_states=__a       , temb=__a       , skip_sample=__a)
         else:
            __snake_case				,__snake_case					:   Dict  				=			downsample_block(hidden_states=__a       , temb=__a)

         down_block_res_samples += res_samples

      # 4. mid
      __snake_case					:   Any  				=			self.mid_block(__a       , __a)

      # 5. up
      __snake_case					:   Dict  				=			None
      for upsample_block in self.up_blocks:
         __snake_case					:   int  				=			down_block_res_samples[-len(upsample_block.resnets) :]
         __snake_case					:   Union[str, Any]  				=			down_block_res_samples[: -len(upsample_block.resnets)]

         if hasattr(__a       , 'skip_conv'):
            __snake_case				,__snake_case					:   List[str]  				=			upsample_block(__a       , __a       , __a       , __a)
         else:
            __snake_case					:   int  				=			upsample_block(__a       , __a       , __a)

        # 6. post-process
      __snake_case					:   List[str]  				=			self.conv_norm_out(__a)
      __snake_case					:   Optional[int]  				=			self.conv_act(__a)
      __snake_case					:   Optional[int]  				=			self.conv_out(__a)

      if skip_sample is not None:
         sample += skip_sample

      if self.config.time_embedding_type == "fourier":
         __snake_case					:   str  				=			timesteps.reshape((sample.shape[0], *([1] * len(sample.shape[1:]))))
         __snake_case					:   Union[str, Any]  				=			sample / timesteps

      if not return_dict:
         return (sample,)

      return UNetaDOutput(sample=__a)