


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


UpperCamelCase__							:Dict       =  getLogger(__name__)


UpperCamelCase__							:Union[str, Any]       =  4
UpperCamelCase__							:Optional[Any]       =  7

UpperCamelCase__							:str       =  {
    """float16""": torch.floataa,
    """float32""": torch.floataa,
}





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

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

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

			return image






class     A(     lowerCamelCase__		):

		"""simple docstring"""





		def __init__(   self				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__ = 3_50				,							)			->  List[str]:

					"""simple docstring"""



					super().__init__(
					    vae=SCREAMING_SNAKE_CASE__				,							text_encoder=SCREAMING_SNAKE_CASE__				,							tokenizer=SCREAMING_SNAKE_CASE__				,							unet=SCREAMING_SNAKE_CASE__				,							low_res_scheduler=SCREAMING_SNAKE_CASE__				,							scheduler=SCREAMING_SNAKE_CASE__				,							safety_checker=SCREAMING_SNAKE_CASE__				,							feature_extractor=SCREAMING_SNAKE_CASE__				,							watermarker=SCREAMING_SNAKE_CASE__				,							max_noise_level=SCREAMING_SNAKE_CASE__				,							)
		def __call__(   self				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__ = 75				,							SCREAMING_SNAKE_CASE__ = 9.0				,							SCREAMING_SNAKE_CASE__ = 20				,							SCREAMING_SNAKE_CASE__ = None				,							SCREAMING_SNAKE_CASE__ = 1				,							SCREAMING_SNAKE_CASE__ = 0.0				,							SCREAMING_SNAKE_CASE__ = None				,							SCREAMING_SNAKE_CASE__ = None				,							SCREAMING_SNAKE_CASE__ = None				,							SCREAMING_SNAKE_CASE__ = None				,							SCREAMING_SNAKE_CASE__ = "pil"				,							SCREAMING_SNAKE_CASE__ = True				,							SCREAMING_SNAKE_CASE__ = None				,							SCREAMING_SNAKE_CASE__ = 1				,							)			->  Union[str, Any]:

					"""simple docstring"""



					self.check_inputs(SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__      )

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

					_UpperCamelCase				:int				    =  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.
					_UpperCamelCase				:List[Any]				    =  guidance_scale > 1.0

					# 3. Encode input prompt
					_UpperCamelCase				:List[str]				    =  self._encode_prompt(
					    SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__				,							prompt_embeds=SCREAMING_SNAKE_CASE__				,							negative_prompt_embeds=SCREAMING_SNAKE_CASE__				,							)

					_UpperCamelCase				:str				    =  ORT_TO_PT_TYPE[str(text_embeddings.dtype      )]

					# 4. Preprocess image
					_UpperCamelCase				:Optional[Any]				    =  preprocess(SCREAMING_SNAKE_CASE__      )
					_UpperCamelCase				:Union[str, Any]				    =  image.cpu()

					# 5. set timesteps
					self.scheduler.set_timesteps(SCREAMING_SNAKE_CASE__				,							device=SCREAMING_SNAKE_CASE__      )
					_UpperCamelCase				:Dict				    =  self.scheduler.timesteps

					# 5. Add noise to image
					_UpperCamelCase				:str				    =  torch.tensor([noise_level]				,							dtype=torch.long				,							device=SCREAMING_SNAKE_CASE__      )
					_UpperCamelCase				:str				    =  torch.randn(image.shape				,							generator=SCREAMING_SNAKE_CASE__				,							device=SCREAMING_SNAKE_CASE__				,							dtype=SCREAMING_SNAKE_CASE__      )
					_UpperCamelCase				:Dict				    =  self.low_res_scheduler.add_noise(SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__      )

					_UpperCamelCase				:str				    =  2 if do_classifier_free_guidance else 1
					_UpperCamelCase				:Tuple				    =  np.concatenate([image] * batch_multiplier * num_images_per_prompt      )
					_UpperCamelCase				:List[Any]				    =  np.concatenate([noise_level] * image.shape[0]      )

					# 6. Prepare latent variables
					_UpperCamelCase    ,			_UpperCamelCase				:List[str]				    =  image.shape[2:]
					_UpperCamelCase				:Dict				    =  self.prepare_latents(
					    batch_size * num_images_per_prompt				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__				,							)

					# 7. Check that sizes of image and latents match
					_UpperCamelCase				: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
					_UpperCamelCase				:Dict				    =  self.prepare_extra_step_kwargs(SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__      )

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

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

											# concat latents, mask, masked_image_latents in the channel dimension
											_UpperCamelCase				:Dict				    =  self.scheduler.scale_model_input(SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__      )
											_UpperCamelCase				:Optional[int]				    =  np.concatenate([latent_model_input, image]				,							axis=1      )

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

											# predict the noise residual
											_UpperCamelCase				:List[str]				    =  self.unet(
											    sample=SCREAMING_SNAKE_CASE__				,							timestep=SCREAMING_SNAKE_CASE__				,							encoder_hidden_states=SCREAMING_SNAKE_CASE__				,							class_labels=noise_level.astype(np.intaa      )				,							)[0]

											# perform guidance
											if do_classifier_free_guidance:
														_UpperCamelCase    ,			_UpperCamelCase				:Optional[Any]				    =  np.split(SCREAMING_SNAKE_CASE__				,							2      )
											_UpperCamelCase				:Union[str, Any]				    =  noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

											# compute the previous noisy sample x_t -> x_t-1
											_UpperCamelCase				:Dict				    =  self.scheduler.step(
											    torch.from_numpy(SCREAMING_SNAKE_CASE__      )				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__				,							**SCREAMING_SNAKE_CASE__      ).prev_sample

											# call the callback, if provided
											if i == len(SCREAMING_SNAKE_CASE__      ) - 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(SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__      )

        # 10. Post-processing
					_UpperCamelCase				:str				    =  self.decode_latents(latents.float()      )

					# 11. Convert to PIL
					if output_type == "pil":
								_UpperCamelCase				:str				    =  self.numpy_to_pil(SCREAMING_SNAKE_CASE__      )

					if not return_dict:
								return (image,)

					return ImagePipelineOutput(images=SCREAMING_SNAKE_CASE__      )
		def      _UpperCamelCase(   self				,							SCREAMING_SNAKE_CASE__      )			->  str:

					"""simple docstring"""



					_UpperCamelCase				:Optional[int]				    =  1 / 0.0_8_3_3_3 * latents
					_UpperCamelCase				:Optional[Any]				    =  self.vae(latent_sample=SCREAMING_SNAKE_CASE__      )[0]
					_UpperCamelCase				:List[Any]				    =  np.clip(image / 2 + 0.5				,							0				,							1      )
					_UpperCamelCase				:List[Any]				    =  image.transpose((0, 2, 3, 1)      )
					return image





		def      _UpperCamelCase(   self				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__ = None				,							SCREAMING_SNAKE_CASE__ = None				,							)			->  List[Any]:

					"""simple docstring"""



					if prompt is not None and isinstance(SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__      ):
								_UpperCamelCase				:Dict				    =  1
					elif prompt is not None and isinstance(SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__      ):
								_UpperCamelCase				:List[Any]				    =  len(SCREAMING_SNAKE_CASE__      )
					else:
								_UpperCamelCase				:List[str]				    =  prompt_embeds.shape[0]

					if prompt_embeds is None:
								_UpperCamelCase				:Union[str, Any]				    =  self.tokenizer(
								    SCREAMING_SNAKE_CASE__				,							padding='''max_length'''				,							max_length=self.tokenizer.model_max_length				,							truncation=SCREAMING_SNAKE_CASE__				,							return_tensors='''pt'''				,							)
								_UpperCamelCase				:int				    =  text_inputs.input_ids
								_UpperCamelCase				:Dict				    =  self.tokenizer(SCREAMING_SNAKE_CASE__				,							padding='''longest'''				,							return_tensors='''pt'''      ).input_ids

								if untruncated_ids.shape[-1] >= text_input_ids.shape[-1] and not torch.equal(
								    SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__      ):
											_UpperCamelCase				:int				    =  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
								_UpperCamelCase				:Optional[int]				    =  self.text_encoder(
								    input_ids=text_input_ids.int().to(SCREAMING_SNAKE_CASE__      )				,							)
								_UpperCamelCase				:List[Any]				    =  prompt_embeds[0]

					_UpperCamelCase    ,			_UpperCamelCase    ,			_UpperCamelCase				:Dict				    =  prompt_embeds.shape
					# duplicate text embeddings for each generation per prompt, using mps friendly method
					_UpperCamelCase				:List[Any]				    =  prompt_embeds.repeat(1				,							SCREAMING_SNAKE_CASE__      )
					_UpperCamelCase				:Tuple				    =  prompt_embeds.reshape(bs_embed * num_images_per_prompt				,							SCREAMING_SNAKE_CASE__				,							-1      )

					# get unconditional embeddings for classifier free guidance
					if do_classifier_free_guidance and negative_prompt_embeds is None:
								_UpperCamelCase				:List[str]
								if negative_prompt is None:
											_UpperCamelCase				:Tuple				    =  [''''''] * batch_size
								elif type(SCREAMING_SNAKE_CASE__      ) is not type(SCREAMING_SNAKE_CASE__      ):
											raise TypeError(
											    f"`negative_prompt` should be the same type to `prompt`, but got {type(SCREAMING_SNAKE_CASE__      )} !="
											    f" {type(SCREAMING_SNAKE_CASE__      )}."      )
								elif isinstance(SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__      ):
											_UpperCamelCase				:List[str]				    =  [negative_prompt]
								elif batch_size != len(SCREAMING_SNAKE_CASE__      ):
											raise ValueError(
											    f"`negative_prompt`: {negative_prompt} has batch size {len(SCREAMING_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:
											_UpperCamelCase				:List[str]				    =  negative_prompt

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

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

								_UpperCamelCase				:Dict				    =  self.text_encoder(
								    input_ids=uncond_input.input_ids.int().to(SCREAMING_SNAKE_CASE__      )				,							)
								_UpperCamelCase				:Union[str, Any]				    =  uncond_embeddings[0]

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

					return prompt_embeds





"""simple docstring"""






from __future__ import annotations

from enum import Enum






class     A(     lowerCamelCase__		):

		"""simple docstring"""





		A         =       24
		A         =       21
		A         =       18
		A         =       15
		A         =       12
		A         =       9
		A         =       6
		A         =       3
		A         =       2
		A         =       1
		A         =       -1
		A         =       -2
		A         =       -3
		A         =       -6
		A         =       -9
		A         =       -12
		A         =       -15
		A         =       -18
		A         =       -21
		A         =       -24






class     A(     lowerCamelCase__		):

		"""simple docstring"""





		A         =       8
		A         =       7
		A         =       6
		A         =       5
		A         =       4
		A         =       3
		A         =       2
		A         =       1





def 							A_					(   snake_case__					,				snake_case__					,				snake_case__					,				)				->     float:
			if isinstance(snake_case__					,				snake_case__		):
						_UpperCamelCase				:Tuple				    =  SIUnit[known_prefix.lower()]
			if isinstance(snake_case__					,				snake_case__		):
						_UpperCamelCase				:Optional[int]				    =  SIUnit[unknown_prefix.lower()]
			_UpperCamelCase				:float				    =  known_amount * (
			    10 ** (known_prefix.value - unknown_prefix.value)
			)
			return unknown_amount





def 							A_					(   snake_case__					,				snake_case__					,				snake_case__					,				)				->     float:
			if isinstance(snake_case__					,				snake_case__		):
						_UpperCamelCase				:Any				    =  BinaryUnit[known_prefix.lower()]
			if isinstance(snake_case__					,				snake_case__		):
						_UpperCamelCase				:Dict				    =  BinaryUnit[unknown_prefix.lower()]
			_UpperCamelCase				:float				    =  known_amount * (
			    2 ** ((known_prefix.value - unknown_prefix.value) * 10)
			)
			return unknown_amount


if __name__ == "__main__":
					import doctest

					doctest.testmod()

