





'''simple docstring'''




from logging import getLogger
from typing import Any, Callable, List, Optional, Union

import numpy as np
import PIL
import torch

from ...schedulers import DDPMScheduler
from ..onnx_utils import ORT_TO_NP_TYPE, OnnxRuntimeModel
from ..pipeline_utils import ImagePipelineOutput
from . import StableDiffusionUpscalePipeline


snake_case__          =						getLogger(__name__)


snake_case__          =						4
snake_case__          =						7

snake_case__          =						{
    """float16""": torch.floataa,
    """float32""": torch.floataa,
}




def 						snake_case__    ( lowerCamelCase__				:   int			)   ->   Any:
					if isinstance(lowerCamelCase__			,		torch.Tensor			):
										return image
					elif isinstance(lowerCamelCase__			,		PIL.Image.Image			):
										A_ :     int =							[image]

					if isinstance(image[0]			,		PIL.Image.Image			):
										A_ ,A_ :     List[Any] =							image[0].size
										A_ ,A_ :     List[str] =							(x - x % 6_4 for x in (w, h))  # resize to integer multiple of 32

										A_ :     List[str] =							[np.array(i.resize((w, h)			)			)[None, :] for i in image]
										A_ :     Optional[Any] =							np.concatenate(lowerCamelCase__			,		axis=0			)
										A_ :     Any =							np.array(lowerCamelCase__			).astype(np.floataa			) / 255.0
										A_ :     Tuple =							image.transpose(0			,		3			,		1			,		2			)
										A_ :     Optional[Any] =							2.0 * image - 1.0
										A_ :     Optional[int] =							torch.from_numpy(lowerCamelCase__			)
					elif isinstance(image[0]			,		torch.Tensor			):
										A_ :     Dict =							torch.cat(lowerCamelCase__			,		dim=0			)

					return image






class     UpperCamelCase_     (_SCREAMING_SNAKE_CASE     ):







						"""simple docstring"""


						def __init__(     self      :       Optional[Any]							, _lowerCamelCase      :       Optional[Any]							, _lowerCamelCase      :       Dict							, _lowerCamelCase      :       int							, _lowerCamelCase      :       List[str]							, _lowerCamelCase      :       str							, _lowerCamelCase      :       List[str]							, _lowerCamelCase      :       str = 350							, ):





											"""simple docstring"""







											super().__init__(
											    vae=__A							, text_encoder=__A							, tokenizer=__A							, unet=__A							, low_res_scheduler=__A							, scheduler=__A							, safety_checker=__A							, feature_extractor=__A							, watermarker=__A							, max_noise_level=__A							, )







						def __call__(     self      :       List[str]							, _lowerCamelCase      :       str							, _lowerCamelCase      :       str							, _lowerCamelCase      :       str = 75							, _lowerCamelCase      :       Any = 9.0							, _lowerCamelCase      :       Dict = 20							, _lowerCamelCase      :       Optional[Any] = None							, _lowerCamelCase      :       Optional[Any] = 1							, _lowerCamelCase      :       Tuple = 0.0							, _lowerCamelCase      :       Any = None							, _lowerCamelCase      :       List[str] = None							, _lowerCamelCase      :       List[str] = None							, _lowerCamelCase      :       int = None							, _lowerCamelCase      :       Optional[Any] = "pil"							, _lowerCamelCase      :       str = True							, _lowerCamelCase      :       Tuple = None							, _lowerCamelCase      :       Tuple = 1							, ):





											"""simple docstring"""







											self.check_inputs(__A							, __A							, __A							, __A    )

											# 2. Define call parameters
											if prompt is not None and isinstance(__A							, __A    ):
																A_ :     Union[str, Any] =							1
											elif prompt is not None and isinstance(__A							, __A    ):
																A_ :     List[str] =							len(__A    )
											else:
																A_ :     Optional[Any] =							prompt_embeds.shape[0]

											A_ :     str =							self._execution_device
											# 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.
											A_ :     int =							guidance_scale > 1.0

											# 3. Encode input prompt
											A_ :     Dict =							self._encode_prompt(
											    __A							, __A							, __A							, __A							, __A							, prompt_embeds=__A							, negative_prompt_embeds=__A							, )

											A_ :     Union[str, Any] =							ORT_TO_PT_TYPE[str(text_embeddings.dtype    )]

											# 4. Preprocess image
											A_ :     str =							preprocess(__A    )
											A_ :     Optional[Any] =							image.cpu()

											# 5. set timesteps
											self.scheduler.set_timesteps(__A							, device=__A    )
											A_ :     Tuple =							self.scheduler.timesteps

											# 5. Add noise to image
											A_ :     Optional[int] =							torch.tensor([noise_level]							, dtype=torch.long							, device=__A    )
											A_ :     str =							torch.randn(image.shape							, generator=__A							, device=__A							, dtype=__A    )
											A_ :     Optional[Any] =							self.low_res_scheduler.add_noise(__A							, __A							, __A    )

											A_ :     Optional[int] =							2 if do_classifier_free_guidance else 1
											A_ :     List[str] =							np.concatenate([image] * batch_multiplier * num_images_per_prompt    )
											A_ :     Union[str, Any] =							np.concatenate([noise_level] * image.shape[0]    )

											# 6. Prepare latent variables
											A_ ,A_ :     Optional[Any] =							image.shape[2:]
											A_ :     Dict =							self.prepare_latents(
											    batch_size * num_images_per_prompt							, __A							, __A							, __A							, __A							, __A							, __A							, __A							, )

											# 7. Check that sizes of image and latents match
											A_ :     List[Any] =							image.shape[1]
											if NUM_LATENT_CHANNELS + num_channels_image != NUM_UNET_INPUT_CHANNELS:
																raise ValueError(
																    '''Incorrect configuration settings! The config of `pipeline.unet` expects'''
																    f' {NUM_UNET_INPUT_CHANNELS} but received `num_channels_latents`: {NUM_LATENT_CHANNELS} +'
																    f' `num_channels_image`: {num_channels_image} '
																    f' = {NUM_LATENT_CHANNELS+num_channels_image}. Please verify the config of'
																    ''' `pipeline.unet` or your `image` input.'''    )

											# 8. Prepare extra step kwargs. TODO: Logic should ideally just be moved out of the pipeline
											A_ :     Optional[Any] =							self.prepare_extra_step_kwargs(__A							, __A    )

											A_ :     str =							next(
											    (input.type for input in self.unet.model.get_inputs() if input.name == '''timestep''')							, '''tensor(float)'''    )
											A_ :     str =							ORT_TO_NP_TYPE[timestep_dtype]

											# 9. Denoising loop
											A_ :     Union[str, Any] =							len(__A    ) - num_inference_steps * self.scheduler.order
											with self.progress_bar(total=__A    ) as progress_bar:
																for i, t in enumerate(__A    ):
																					# expand the latents if we are doing classifier free guidance
																					A_ :     Optional[int] =							np.concatenate([latents] * 2    ) if do_classifier_free_guidance else latents

																					# concat latents, mask, masked_image_latents in the channel dimension
																					A_ :     str =							self.scheduler.scale_model_input(__A							, __A    )
																					A_ :     Optional[Any] =							np.concatenate([latent_model_input, image]							, axis=1    )

																					# timestep to tensor
																					A_ :     Any =							np.array([t]							, dtype=__A    )

																					# predict the noise residual
																					A_ :     Optional[Any] =							self.unet(
																					    sample=__A							, timestep=__A							, encoder_hidden_states=__A							, class_labels=noise_level.astype(np.intaa    )							, )[0]

																					# perform guidance
																					if do_classifier_free_guidance:
																										A_ ,A_ :     Any =							np.split(__A							, 2    )
																					A_ :     Dict =							noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

																					# compute the previous noisy sample x_t -> x_t-1
																					A_ :     List[Any] =							self.scheduler.step(
																					    torch.from_numpy(__A    )							, __A							, __A							, **__A    ).prev_sample

																					# call the callback, if provided
																					if i == len(__A    ) - 1 or ((i + 1) > num_warmup_steps and (i + 1) % self.scheduler.order == 0):
																										progress_bar.update()
																										if callback is not None and i % callback_steps == 0:
																															callback(__A							, __A							, __A    )

        # 10. Post-processing
											A_ :     Union[str, Any] =							self.decode_latents(latents.float()    )

											# 11. Convert to PIL
											if output_type == "pil":
																A_ :     Tuple =							self.numpy_to_pil(__A    )

											if not return_dict:
																return (image,)

											return ImagePipelineOutput(images=__A    )







						def       _a    (     self      :       Tuple							, _lowerCamelCase      :       Optional[Any]    ):





											"""simple docstring"""







											A_ :     List[str] =							1 / 0.0_83_33 * latents
											A_ :     List[Any] =							self.vae(latent_sample=__A    )[0]
											A_ :     int =							np.clip(image / 2 + 0.5							, 0							, 1    )
											A_ :     Union[str, Any] =							image.transpose((0, 2, 3, 1)    )
											return image







						def       _a    (     self      :       int							, _lowerCamelCase      :       Optional[int]							, _lowerCamelCase      :       int							, _lowerCamelCase      :       str							, _lowerCamelCase      :       Optional[int]							, _lowerCamelCase      :       str							, _lowerCamelCase      :       Optional[Any] = None							, _lowerCamelCase      :       int = None							, ):





											"""simple docstring"""







											if prompt is not None and isinstance(__A							, __A    ):
																A_ :     Dict =							1
											elif prompt is not None and isinstance(__A							, __A    ):
																A_ :     int =							len(__A    )
											else:
																A_ :     str =							prompt_embeds.shape[0]

											if prompt_embeds is None:
																A_ :     int =							self.tokenizer(
																    __A							, padding='''max_length'''							, max_length=self.tokenizer.model_max_length							, truncation=__A							, return_tensors='''pt'''							, )
																A_ :     str =							text_inputs.input_ids
																A_ :     int =							self.tokenizer(__A							, padding='''longest'''							, return_tensors='''pt'''    ).input_ids

																if untruncated_ids.shape[-1] >= text_input_ids.shape[-1] and not torch.equal(
																    __A							, __A    ):
																					A_ :     Union[str, Any] =							self.tokenizer.batch_decode(
																					    untruncated_ids[:, self.tokenizer.model_max_length - 1 : -1]    )
																					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}'    )

																# no positional arguments to text_encoder
																A_ :     str =							self.text_encoder(
																    input_ids=text_input_ids.int().to(__A    )							, )
																A_ :     Optional[int] =							prompt_embeds[0]

											A_ ,A_ ,A_ :     Dict =							prompt_embeds.shape
											# duplicate text embeddings for each generation per prompt, using mps friendly method
											A_ :     Tuple =							prompt_embeds.repeat(1							, __A    )
											A_ :     int =							prompt_embeds.reshape(bs_embed * num_images_per_prompt							, __A							, -1    )

											# get unconditional embeddings for classifier free guidance
											if do_classifier_free_guidance and negative_prompt_embeds is None:
																A_ :     Optional[int] =							42
																if negative_prompt is None:
																					A_ :     Optional[Any] =							[''''''] * batch_size
																elif type(__A    ) is not type(__A    ):
																					raise TypeError(
																					    f'`negative_prompt` should be the same type to `prompt`, but got {type(__A    )} !='
																					    f' {type(__A    )}.'    )
																elif isinstance(__A							, __A    ):
																					A_ :     Dict =							[negative_prompt]
																elif batch_size != len(__A    ):
																					raise ValueError(
																					    f'`negative_prompt`: {negative_prompt} has batch size {len(__A    )}, but `prompt`:'
																					    f' {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches'
																					    ''' the batch size of `prompt`.'''    )
																else:
																					A_ :     Tuple =							negative_prompt

																A_ :     List[Any] =							text_input_ids.shape[-1]
																A_ :     List[str] =							self.tokenizer(
																    __A							, padding='''max_length'''							, max_length=__A							, truncation=__A							, return_tensors='''pt'''							, )

																# if hasattr(uncond_input, "attention_mask"):
																#     attention_mask = uncond_input.attention_mask.to(device)
																# else:
																#     attention_mask = None

																A_ :     Dict =							self.text_encoder(
																    input_ids=uncond_input.input_ids.int().to(__A    )							, )
																A_ :     Any =							uncond_embeddings[0]

											if do_classifier_free_guidance:
																A_ :     Dict =							uncond_embeddings.shape[1]
																# duplicate unconditional embeddings for each generation per prompt, using mps friendly method
																A_ :     Optional[Any] =							uncond_embeddings.repeat(1							, __A    )
																A_ :     List[Any] =							uncond_embeddings.reshape(batch_size * num_images_per_prompt							, __A							, -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
																A_ :     Dict =							np.concatenate([uncond_embeddings, prompt_embeds]    )

											return prompt_embeds









'''simple docstring'''




def 						snake_case__    ( lowerCamelCase__				:   int			)   ->   bool:
					if not isinstance(lowerCamelCase__			,		lowerCamelCase__			):
										A_ :     int =							f'Input value of [number={number}] must be an integer'
										raise TypeError(lowerCamelCase__			)
					if number < 0 or number % 2 == 1:
										return False
					A_ :     Dict =							int(number**0.5			)
					return number == number_sqrt * (number_sqrt + 1)


if __name__ == "__main__":
							import doctest

							doctest.testmod()



