




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


lowercase_      :							str    							=					logging.get_logger(__name__)  # pylint: disable=invalid-name


def 				__SCREAMING_SNAKE_CASE    (     self      ,       snake_case_      ,       snake_case_      ,       snake_case_      ,       snake_case_		):

						'''simple docstring'''


						_UpperCAmelCase						=	image.to(device=snake_case_		)
						_UpperCAmelCase						=	image  # take image as image_embeddings
						_UpperCAmelCase						=	image_embeddings.unsqueeze(1		)

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

						if do_classifier_free_guidance:
												_UpperCAmelCase						=	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
												_UpperCAmelCase						=	torch.cat([uncond_embeddings, image_embeddings]		)

						return image_embeddings






class        __lowerCAmelCase					(			UpperCAmelCase__					):







			def __init__(			self       :     Tuple       ,       snake_case__       :     PriorTransformer       ,       snake_case__       :     CLIPTokenizer       ,       snake_case__       :     CLIPTextModelWithProjection       ,       snake_case__       :     UnCLIPScheduler       ,       snake_case__       :     Optional[dict] = None       ,       ):




									"""simple docstring"""


									super().__init__()

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

									_UpperCAmelCase						=	decoder_pipe_kwargs.get("torch_dtype"       ,       snake_case__ ) or prior.dtype

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

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

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







			def UpperCamelCase     (			self       :     List[Any]       ,       snake_case__       :     Dict       ,       snake_case__       :     Optional[int]       ,       snake_case__       :     Tuple       ,       snake_case__       :     Any       ,       snake_case__       :     Optional[Union[CLIPTextModelOutput, Tuple]] = None       ,       snake_case__       :     Optional[torch.Tensor] = None       ,       ):




									"""simple docstring"""


									if text_model_output is None:
															_UpperCAmelCase						=	len(snake_case__ ) if isinstance(snake_case__       ,       snake_case__ ) else 1
															# get prompt text embeddings
															_UpperCAmelCase						=	self.tokenizer(
															    snake_case__       ,       padding="max_length"       ,       max_length=self.tokenizer.model_max_length       ,       return_tensors="pt"       ,       )
															_UpperCAmelCase						=	text_inputs.input_ids
															_UpperCAmelCase						=	text_inputs.attention_mask.bool().to(snake_case__ )

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

															_UpperCAmelCase						=	self.text_encoder(text_input_ids.to(snake_case__ ) )

															_UpperCAmelCase						=	text_encoder_output.text_embeds
															_UpperCAmelCase						=	text_encoder_output.last_hidden_state

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

									_UpperCAmelCase						=	text_embeddings.repeat_interleave(snake_case__       ,       dim=0 )
									_UpperCAmelCase						=	text_encoder_hidden_states.repeat_interleave(snake_case__       ,       dim=0 )
									_UpperCAmelCase						=	text_mask.repeat_interleave(snake_case__       ,       dim=0 )

									if do_classifier_free_guidance:
															_UpperCAmelCase						=	[""] * batch_size

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

															_UpperCAmelCase						=	uncond_embeddings_text_encoder_output.text_embeds
															_UpperCAmelCase						=	uncond_embeddings_text_encoder_output.last_hidden_state

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

															_UpperCAmelCase						=	uncond_embeddings.shape[1]
															_UpperCAmelCase						=	uncond_embeddings.repeat(1       ,       snake_case__ )
															_UpperCAmelCase						=	uncond_embeddings.view(batch_size * num_images_per_prompt       ,       snake_case__ )

															_UpperCAmelCase						=	uncond_text_encoder_hidden_states.shape[1]
															_UpperCAmelCase						=	uncond_text_encoder_hidden_states.repeat(1       ,       snake_case__       ,       1 )
															_UpperCAmelCase						=	uncond_text_encoder_hidden_states.view(
															    batch_size * num_images_per_prompt       ,       snake_case__       ,       -1 )
															_UpperCAmelCase						=	uncond_text_mask.repeat_interleave(snake_case__       ,       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
															_UpperCAmelCase						=	torch.cat([uncond_embeddings, text_embeddings] )
															_UpperCAmelCase						=	torch.cat([uncond_text_encoder_hidden_states, text_encoder_hidden_states] )

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

									return text_embeddings, text_encoder_hidden_states, text_mask







			@property
			def UpperCamelCase     (			self       :     str ):




									"""simple docstring"""


									if self.device != torch.device("meta" ) or not hasattr(self.prior       ,       "_hf_hook" ):
															return self.device
									for module in self.prior.modules():
															if (
															    hasattr(snake_case__       ,       "_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       :     Tuple       ,       snake_case__       :     Optional[int]       ,       snake_case__       :     List[Any]       ,       snake_case__       :     Optional[int]       ,       snake_case__       :     Any       ,       snake_case__       :     int       ,       snake_case__       :     Tuple ):




									"""simple docstring"""


									if latents is None:
															_UpperCAmelCase						=	randn_tensor(snake_case__       ,       generator=snake_case__       ,       device=snake_case__       ,       dtype=snake_case__ )
									else:
															if latents.shape != shape:
																					raise ValueError(F"""Unexpected latents shape, got {latents.shape}, expected {shape}""" )
															_UpperCAmelCase						=	latents.to(snake_case__ )

									_UpperCAmelCase						=	latents * scheduler.init_noise_sigma
									return latents







			def UpperCamelCase     (			self       :     Dict       ,       snake_case__       :     Optional[Union[str, torch.device]] = None ):




									"""simple docstring"""


									self.decoder_pipe.to(snake_case__ )
									super().to(snake_case__ )







			@torch.no_grad()
			def __call__(			self       :     Tuple       ,       snake_case__       :     Optional[Union[str, List[str]]] = None       ,       snake_case__       :     Optional[int] = None       ,       snake_case__       :     Optional[int] = None       ,       snake_case__       :     int = 1       ,       snake_case__       :     int = 25       ,       snake_case__       :     Optional[torch.Generator] = None       ,       snake_case__       :     Optional[torch.FloatTensor] = None       ,       snake_case__       :     Optional[Union[CLIPTextModelOutput, Tuple]] = None       ,       snake_case__       :     Optional[torch.Tensor] = None       ,       snake_case__       :     float = 4.0       ,       snake_case__       :     float = 8.0       ,       snake_case__       :     int = 50       ,       snake_case__       :     Optional[int] = 1       ,       snake_case__       :     float = 0.0       ,       snake_case__       :     Optional[str] = "pil"       ,       snake_case__       :     bool = True       ,       ):




									"""simple docstring"""


									if prompt is not None:
															if isinstance(snake_case__       ,       snake_case__ ):
																					_UpperCAmelCase						=	1
															elif isinstance(snake_case__       ,       snake_case__ ):
																					_UpperCAmelCase						=	len(snake_case__ )
															else:
																					raise ValueError(F"""`prompt` has to be of type `str` or `list` but is {type(snake_case__ )}""" )
									else:
															_UpperCAmelCase						=	text_model_output[0].shape[0]

									_UpperCAmelCase						=	self._execution_device

									_UpperCAmelCase						=	batch_size * num_images_per_prompt

									_UpperCAmelCase						=	prior_guidance_scale > 1.0 or decoder_guidance_scale > 1.0

									_UpperCAmelCase				,					_UpperCAmelCase				,					_UpperCAmelCase						=	self._encode_prompt(
									    snake_case__       ,       snake_case__       ,       snake_case__       ,       snake_case__       ,       snake_case__       ,       snake_case__ )

									# prior

									self.prior_scheduler.set_timesteps(snake_case__       ,       device=snake_case__ )
									_UpperCAmelCase						=	self.prior_scheduler.timesteps

									_UpperCAmelCase						=	self.prior.config.embedding_dim

									_UpperCAmelCase						=	self.prepare_latents(
									    (batch_size, embedding_dim)       ,       text_embeddings.dtype       ,       snake_case__       ,       snake_case__       ,       snake_case__       ,       self.prior_scheduler       ,       )

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

															_UpperCAmelCase						=	self.prior(
															    snake_case__       ,       timestep=snake_case__       ,       proj_embedding=snake_case__       ,       encoder_hidden_states=snake_case__       ,       attention_mask=snake_case__       ,       ).predicted_image_embedding

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

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

															_UpperCAmelCase						=	self.prior_scheduler.step(
															    snake_case__       ,       timestep=snake_case__       ,       sample=snake_case__       ,       generator=snake_case__       ,       prev_timestep=snake_case__       ,       ).prev_sample

									_UpperCAmelCase						=	self.prior.post_process_latents(snake_case__ )

									_UpperCAmelCase						=	prior_latents

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










import enum
import inspect
import warnings
from functools import wraps
from typing import Callable, Optional

from .logging import get_logger


lowercase_      :							Optional[int]    							=					set()
lowercase_      :							List[Any]    							=					get_logger(__name__)


def 				__SCREAMING_SNAKE_CASE    (     snake_case_ = None		):

						'''simple docstring'''


						def decorator(snake_case_		):
												global _emitted_deprecation_warnings

												if inspect.isclass(snake_case_		):
																		_UpperCAmelCase						=	deprecated_class_or_function.__init__
																		_UpperCAmelCase						=	deprecated_class_or_function.__name__
												else:
																		_UpperCAmelCase						=	deprecated_class_or_function
																		_UpperCAmelCase						=	deprecated_function.__name__
																		# Support deprecating __init__ class method: class name instead
																		_UpperCAmelCase						=	name if name != "__init__" else deprecated_function.__qualname__.split("."		)[-2]

												_UpperCAmelCase						=	(
												    f"""{name} is deprecated and will be removed in the next major version of datasets.""" + f""" {help_message}"""
												    if help_message
												    else ""
												)

												@wraps(snake_case_		)
												def wrapper(*snake_case_      ,       **snake_case_		):
																		_UpperCAmelCase						=	hash(snake_case_		)
																		if func_hash not in _emitted_deprecation_warnings:
																								warnings.warn(snake_case_      ,       category=snake_case_      ,       stacklevel=2		)
																								_emitted_deprecation_warnings.add(snake_case_		)
																		return deprecated_function(*snake_case_      ,       **snake_case_		)

												_UpperCAmelCase						=	"deprecated"

												if inspect.isclass(snake_case_		):
																		_UpperCAmelCase						=	wrapper
																		return deprecated_class_or_function
												else:
																		return wrapper

						return decorator






class        __lowerCAmelCase					(			enum.EnumMeta					):







			def __getattribute__(			cls       :     str       ,       snake_case__       :     Optional[Any] ):




									"""simple docstring"""


									_UpperCAmelCase						=	super().__getattribute__(snake_case__ )
									if isinstance(snake_case__       ,       enum.Enum ) and obj._on_access:
															obj._on_access()
									return obj







			def __getitem__(			cls       :     str       ,       snake_case__       :     Union[str, Any] ):




									"""simple docstring"""


									_UpperCAmelCase						=	super().__getitem__(snake_case__ )
									if member._on_access:
															member._on_access()
									return member







			def __call__(			cls       :     Dict       ,       snake_case__       :     Any       ,       snake_case__       :     List[str]=None       ,       *, snake_case__       :     Optional[int]=None       ,       snake_case__       :     List[Any]=None       ,       snake_case__       :     Tuple=None       ,       snake_case__       :     List[Any]=1 ):




									"""simple docstring"""


									_UpperCAmelCase						=	super().__call__(snake_case__       ,       snake_case__       ,       module=snake_case__       ,       qualname=snake_case__       ,       type=snake_case__       ,       start=snake_case__ )
									if isinstance(snake_case__       ,       enum.Enum ) and obj._on_access:
															obj._on_access()
									return obj






class        __lowerCAmelCase					(			enum.Enum   ,						metaclass=UpperCAmelCase__					):







			def __new__(			cls       :     Any       ,       snake_case__       :     Dict ):




									"""simple docstring"""


									_UpperCAmelCase						=	object.__new__(cls )
									_UpperCAmelCase						=	value
									_UpperCAmelCase						=	member.deprecate
									return member







			@property
			def UpperCamelCase     (			self       :     str ):




									"""simple docstring"""


									return ""







			def UpperCamelCase     (			self       :     Optional[int] ):




									"""simple docstring"""


									_UpperCAmelCase						=	F""" {self.help_message}""" if self.help_message else ""
									warnings.warn(
									    F"""'{self.__objclass__.__name__}' is deprecated and will be removed in the next major version of datasets."""
									    + help_message       ,       snake_case__       ,       stacklevel=3       ,       )




