

'''simple docstring'''


import inspect
from typing import Callable, List, Optional, Union

import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import DiffusionPipeline
from diffusers.models import AutoencoderKL, UNetaDConditionModel
from diffusers.pipelines.stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.schedulers import DDIMScheduler, LMSDiscreteScheduler, PNDMScheduler
from diffusers.utils import logging


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




class     a			(				snake_case_       ):






						"""simple docstring"""




						def __init__(							self					:      List[Any]			,			snake_case_					:      Tuple			,			snake_case_					:      int			,			snake_case_					:      List[str]			,			snake_case_					:      List[str]			,			snake_case_					:      str			,			snake_case_					:      Tuple			,			snake_case_					:      List[Any]			,			):



								'''simple docstring'''



								super().__init__()
								self.register_modules(
								    vae=snake_case_			,			text_encoder=snake_case_			,			tokenizer=snake_case_			,			unet=snake_case_			,			scheduler=snake_case_			,			safety_checker=snake_case_			,			feature_extractor=snake_case_			,			)







						def       __magic_name__							(							self					:      Optional[Any]			,			snake_case_					:      Union[str, Any] = "auto"  ):



								'''simple docstring'''



								if slice_size == "auto":
										# half the attention head size is usually a good trade-off between
										# speed and memory
										snake_case__						:       Any       						=  self.unet.config.attention_head_dim // 2
								self.unet.set_attention_slice(snake_case_  )







						def       __magic_name__							(							self					:      Union[str, Any]  ):



								'''simple docstring'''



								self.enable_attention_slicing(snake_case_  )







						@torch.no_grad()
						def __call__(							self					:      Tuple			,			snake_case_					:      Optional[int]			,			snake_case_					:      List[Any] = 5_1_2			,			snake_case_					:      str = 5_1_2			,			snake_case_					:      Any = 5_0			,			snake_case_					:      Tuple = 7.5			,			snake_case_					:      str = None			,			snake_case_					:      str = 1			,			snake_case_					:      Optional[int] = 0.0			,			snake_case_					:      List[str] = None			,			snake_case_					:      Dict = None			,			snake_case_					:      Union[str, Any] = "pil"			,			snake_case_					:      List[Any] = True			,			snake_case_					:      Tuple = None			,			snake_case_					:      Tuple = 1			,			snake_case_					:      List[str] = None			,			**snake_case_					:      Union[str, Any]			,			):



								'''simple docstring'''



								if isinstance(snake_case_			,			snake_case_  ):
										snake_case__						:       int       						=  1
								elif isinstance(snake_case_			,			snake_case_  ):
										snake_case__						:       List[Any]       						=  len(snake_case_  )
								else:
										raise ValueError(F"""`prompt` has to be of type `str` or `list` but is {type(snake_case_  )}"""  )

								if height % 8 != 0 or width % 8 != 0:
										raise ValueError(F"""`height` and `width` have to be divisible by 8 but are {height} and {width}."""  )

								if (callback_steps is None) or (
								    callback_steps is not None and (not isinstance(snake_case_			,			snake_case_  ) or callback_steps <= 0)
								):
										raise ValueError(
										    F"""`callback_steps` has to be a positive integer but is {callback_steps} of type"""
										    F""" {type(snake_case_  )}."""  )

								# get prompt text embeddings
								snake_case__						:       List[Any]       						=  self.tokenizer(
								    snake_case_			,			padding='''max_length'''			,			max_length=self.tokenizer.model_max_length			,			return_tensors='''pt'''			,			)
								snake_case__						:       int       						=  text_inputs.input_ids

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

								if text_embeddings is None:
										snake_case__						:       int       						=  self.text_encoder(text_input_ids.to(self.device  )  )[0]

								# duplicate text embeddings for each generation per prompt, using mps friendly method
								snake_case__						,       snake_case__						,       snake_case__						:       int       						=  text_embeddings.shape
								snake_case__						:       Optional[int]       						=  text_embeddings.repeat(1			,			snake_case_			,			1  )
								snake_case__						:       Optional[int]       						=  text_embeddings.view(bs_embed * num_images_per_prompt			,			snake_case_			,			-1  )

								# here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
								# of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
								# corresponds to doing no classifier free guidance.
								snake_case__						:       int       						=  guidance_scale > 1.0
								# get unconditional embeddings for classifier free guidance
								if do_classifier_free_guidance:
										snake_case__						:       List[str]       						=  4_2
										if negative_prompt is None:
												snake_case__						:       Optional[Any]       						=  ['''''']
										elif type(snake_case_  ) is not type(snake_case_  ):
												raise TypeError(
												    F"""`negative_prompt` should be the same type to `prompt`, but got {type(snake_case_  )} !="""
												    F""" {type(snake_case_  )}."""  )
										elif isinstance(snake_case_			,			snake_case_  ):
												snake_case__						:       Tuple       						=  [negative_prompt]
										elif batch_size != len(snake_case_  ):
												raise ValueError(
												    F"""`negative_prompt`: {negative_prompt} has batch size {len(snake_case_  )}, but `prompt`:"""
												    F""" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"""
												    ''' the batch size of `prompt`.'''  )
										else:
												snake_case__						:       Optional[Any]       						=  negative_prompt

										snake_case__						:       Optional[int]       						=  text_input_ids.shape[-1]
										snake_case__						:       int       						=  self.tokenizer(
										    snake_case_			,			padding='''max_length'''			,			max_length=snake_case_			,			truncation=snake_case_			,			return_tensors='''pt'''			,			)
										snake_case__						:       Dict       						=  self.text_encoder(uncond_input.input_ids.to(self.device  )  )[0]

										# duplicate unconditional embeddings for each generation per prompt, using mps friendly method
										snake_case__						:       Optional[int]       						=  uncond_embeddings.shape[1]
										snake_case__						:       Optional[Any]       						=  uncond_embeddings.repeat(snake_case_			,			snake_case_			,			1  )
										snake_case__						:       str       						=  uncond_embeddings.view(batch_size * num_images_per_prompt			,			snake_case_			,			-1  )

										# 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
										snake_case__						:       int       						=  torch.cat([uncond_embeddings, text_embeddings]  )

								# get the initial random noise unless the user supplied it

								# Unlike in other pipelines, latents need to be generated in the target device
								# for 1-to-1 results reproducibility with the CompVis implementation.
								# However this currently doesn't work in `mps`.
								snake_case__						:       Optional[int]       						=  (batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
								snake_case__						:       Any       						=  (batch_size * num_images_per_prompt, self.unet.config.in_channels, 6_4, 6_4)
								snake_case__						:       List[Any]       						=  text_embeddings.dtype
								if latents is None:
										if self.device.type == "mps":
												# randn does not exist on mps
												snake_case__						:       Dict       						=  torch.randn(
												    snake_case_			,			generator=snake_case_			,			device='''cpu'''			,			dtype=snake_case_  ).to(self.device  )
												snake_case__						:       Optional[int]       						=  torch.randn(snake_case_			,			generator=snake_case_			,			device='''cpu'''			,			dtype=snake_case_  ).to(
												    self.device  )
										else:
												snake_case__						:       int       						=  torch.randn(
												    snake_case_			,			generator=snake_case_			,			device=self.device			,			dtype=snake_case_  )
												snake_case__						:       Tuple       						=  torch.randn(snake_case_			,			generator=snake_case_			,			device=self.device			,			dtype=snake_case_  )
								else:
										if latents_reference.shape != latents_shape:
												raise ValueError(F"""Unexpected latents shape, got {latents.shape}, expected {latents_shape}"""  )
										snake_case__						:       Optional[int]       						=  latents_reference.to(self.device  )
										snake_case__						:       Union[str, Any]       						=  latents.to(self.device  )

								# This is the key part of the pipeline where we
								# try to ensure that the generated images w/ the same seed
								# but different sizes actually result in similar images
								snake_case__						:       Optional[Any]       						=  (latents_shape[3] - latents_shape_reference[3]) // 2
								snake_case__						:       Any       						=  (latents_shape[2] - latents_shape_reference[2]) // 2
								snake_case__						:       Optional[int]       						=  latents_shape_reference[3] if dx >= 0 else latents_shape_reference[3] + 2 * dx
								snake_case__						:       Optional[Any]       						=  latents_shape_reference[2] if dy >= 0 else latents_shape_reference[2] + 2 * dy
								snake_case__						:       Tuple       						=  0 if dx < 0 else dx
								snake_case__						:       List[Any]       						=  0 if dy < 0 else dy
								snake_case__						:       int       						=  max(-dx			,			0  )
								snake_case__						:       Optional[int]       						=  max(-dy			,			0  )
								# import pdb
								# pdb.set_trace()
								snake_case__						:       Optional[Any]       						=  latents_reference[:, :, dy : dy + h, dx : dx + w]

								# set timesteps
								self.scheduler.set_timesteps(snake_case_  )

								# Some schedulers like PNDM have timesteps as arrays
								# It's more optimized to move all timesteps to correct device beforehand
								snake_case__						:       List[str]       						=  self.scheduler.timesteps.to(self.device  )

								# scale the initial noise by the standard deviation required by the scheduler
								snake_case__						:       Optional[Any]       						=  latents * self.scheduler.init_noise_sigma

								# prepare extra kwargs for the scheduler step, since not all schedulers have the same signature
								# eta (η) is only used with the DDIMScheduler, it will be ignored for other schedulers.
								# eta corresponds to η in DDIM paper: https://arxiv.org/abs/2010.02502
								# and should be between [0, 1]
								snake_case__						:       Optional[int]       						=  '''eta''' in set(inspect.signature(self.scheduler.step  ).parameters.keys()  )
								snake_case__						:       Union[str, Any]       						=  {}
								if accepts_eta:
										snake_case__						:       Union[str, Any]       						=  eta

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

										# predict the noise residual
										snake_case__						:       Any       						=  self.unet(snake_case_			,			snake_case_			,			encoder_hidden_states=snake_case_  ).sample

										# perform guidance
										if do_classifier_free_guidance:
												snake_case__						,       snake_case__						:       Optional[Any]       						=  noise_pred.chunk(2  )
												snake_case__						:       Optional[Any]       						=  noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

										# compute the previous noisy sample x_t -> x_t-1
										snake_case__						:       Tuple       						=  self.scheduler.step(snake_case_			,			snake_case_			,			snake_case_			,			**snake_case_  ).prev_sample

										# call the callback, if provided
										if callback is not None and i % callback_steps == 0:
												callback(snake_case_			,			snake_case_			,			snake_case_  )

								snake_case__						:       int       						=  1 / 0.1_8_2_1_5 * latents
								snake_case__						:       Dict       						=  self.vae.decode(snake_case_  ).sample

								snake_case__						:       Tuple       						=  (image / 2 + 0.5).clamp(0			,			1  )

								# we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
								snake_case__						:       Optional[Any]       						=  image.cpu().permute(0			,			2			,			3			,			1  ).float().numpy()

								if self.safety_checker is not None:
										snake_case__						:       Any       						=  self.feature_extractor(self.numpy_to_pil(snake_case_  )			,			return_tensors='''pt'''  ).to(
										    self.device  )
										snake_case__						,       snake_case__						:       str       						=  self.safety_checker(
										    images=snake_case_			,			clip_input=safety_checker_input.pixel_values.to(text_embeddings.dtype  )  )
								else:
										snake_case__						:       List[Any]       						=  None

								if output_type == "pil":
										snake_case__						:       Dict       						=  self.numpy_to_pil(snake_case_  )

								if not return_dict:
										return (image, has_nsfw_concept)

								return StableDiffusionPipelineOutput(images=snake_case_			,			nsfw_content_detected=snake_case_  )






'''simple docstring'''


import warnings

from diffusers import StableDiffusionImgaImgPipeline  # noqa F401


warnings.warn(
    """The `image_to_image.py` script is outdated. Please use directly `from diffusers import"""
    """ StableDiffusionImg2ImgPipeline` instead."""
)



