






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


lowerCAmelCase__							= getLogger(__name__)


lowerCAmelCase__							= 4
lowerCAmelCase__							= 7

lowerCAmelCase__							= {
    'float16': torch.floataa,
    'float32': torch.floataa,
}





def 							lowercase__		(				lowerCamelCase					):
				if isinstance(lowerCamelCase,	torch.Tensor					):
								return image
				elif isinstance(lowerCamelCase,	PIL.Image.Image					):
								_SCREAMING_SNAKE_CASE      :		int							=						[image]

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

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

				return image






class _lowerCAmelCase	(						__UpperCAmelCase					):







						def __init__(						self      ,							lowerCAmelCase_      ,							lowerCAmelCase_      ,							lowerCAmelCase_      ,							lowerCAmelCase_      ,							lowerCAmelCase_      ,							lowerCAmelCase_      ,							lowerCAmelCase_ = 3_5_0      ,							)		->       Dict:
										super().__init__(
										    vae=lowerCAmelCase_      ,							text_encoder=lowerCAmelCase_      ,							tokenizer=lowerCAmelCase_      ,							unet=lowerCAmelCase_      ,							low_res_scheduler=lowerCAmelCase_      ,							scheduler=lowerCAmelCase_      ,							safety_checker=lowerCAmelCase_      ,							feature_extractor=lowerCAmelCase_      ,							watermarker=lowerCAmelCase_      ,							max_noise_level=lowerCAmelCase_      ,							)







						def __call__(						self      ,							lowerCAmelCase_      ,							lowerCAmelCase_      ,							lowerCAmelCase_ = 7_5      ,							lowerCAmelCase_ = 9.0      ,							lowerCAmelCase_ = 2_0      ,							lowerCAmelCase_ = None      ,							lowerCAmelCase_ = 1      ,							lowerCAmelCase_ = 0.0      ,							lowerCAmelCase_ = None      ,							lowerCAmelCase_ = None      ,							lowerCAmelCase_ = None      ,							lowerCAmelCase_ = None      ,							lowerCAmelCase_ = "pil"      ,							lowerCAmelCase_ = True      ,							lowerCAmelCase_ = None      ,							lowerCAmelCase_ = 1      ,							)		->       List[Any]:
										self.check_inputs(lowerCAmelCase_      ,							lowerCAmelCase_      ,							lowerCAmelCase_      ,							lowerCAmelCase_       )

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

										_SCREAMING_SNAKE_CASE      :		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.
										_SCREAMING_SNAKE_CASE      :		Optional[int]							=						guidance_scale > 1.0

										# 3. Encode input prompt
										_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						self._encode_prompt(
										    lowerCAmelCase_      ,							lowerCAmelCase_      ,							lowerCAmelCase_      ,							lowerCAmelCase_      ,							lowerCAmelCase_      ,							prompt_embeds=lowerCAmelCase_      ,							negative_prompt_embeds=lowerCAmelCase_      ,							)

										_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						ORT_TO_PT_TYPE[str(text_embeddings.dtype       )]

										# 4. Preprocess image
										_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						preprocess(lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		List[Any]							=						image.cpu()

										# 5. set timesteps
										self.scheduler.set_timesteps(lowerCAmelCase_      ,							device=lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		Tuple							=						self.scheduler.timesteps

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

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

										# 6. Prepare latent variables
										_SCREAMING_SNAKE_CASE      :		Tuple							=						image.shape[2:]
										_SCREAMING_SNAKE_CASE      :		Dict							=						self.prepare_latents(
										    batch_size * num_images_per_prompt      ,							lowerCAmelCase_      ,							lowerCAmelCase_      ,							lowerCAmelCase_      ,							lowerCAmelCase_      ,							lowerCAmelCase_      ,							lowerCAmelCase_      ,							lowerCAmelCase_      ,							)

										# 7. Check that sizes of image and latents match
										_SCREAMING_SNAKE_CASE      :		Dict							=						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
										_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						self.prepare_extra_step_kwargs(lowerCAmelCase_      ,							lowerCAmelCase_       )

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

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

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

																		# timestep to tensor
																		_SCREAMING_SNAKE_CASE      :		Dict							=						np.array([t]      ,							dtype=lowerCAmelCase_       )

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

																		# perform guidance
																		if do_classifier_free_guidance:
																						_SCREAMING_SNAKE_CASE      :		Tuple							=						np.split(lowerCAmelCase_      ,							2       )
																		_SCREAMING_SNAKE_CASE      :		Tuple							=						noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

																		# compute the previous noisy sample x_t -> x_t-1
																		_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						self.scheduler.step(
																		    torch.from_numpy(lowerCAmelCase_       )      ,							lowerCAmelCase_      ,							lowerCAmelCase_      ,							**lowerCAmelCase_       ).prev_sample

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

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

										# 11. Convert to PIL
										if output_type == "pil":
														_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						self.numpy_to_pil(lowerCAmelCase_       )

										if not return_dict:
														return (image,)

										return ImagePipelineOutput(images=lowerCAmelCase_       )







						def  A			(						self      ,							lowerCAmelCase_       )		->       Optional[int]:
										_SCREAMING_SNAKE_CASE      :		Optional[int]							=						1 / 0.08_333 * latents
										_SCREAMING_SNAKE_CASE      :		int							=						self.vae(latent_sample=lowerCAmelCase_       )[0]
										_SCREAMING_SNAKE_CASE      :		List[str]							=						np.clip(image / 2 + 0.5      ,							0      ,							1       )
										_SCREAMING_SNAKE_CASE      :		Dict							=						image.transpose((0, 2, 3, 1)       )
										return image







						def  A			(						self      ,							lowerCAmelCase_      ,							lowerCAmelCase_      ,							lowerCAmelCase_      ,							lowerCAmelCase_      ,							lowerCAmelCase_      ,							lowerCAmelCase_ = None      ,							lowerCAmelCase_ = None      ,							)		->       Optional[int]:
										if prompt is not None and isinstance(lowerCAmelCase_      ,							lowerCAmelCase_       ):
														_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						1
										elif prompt is not None and isinstance(lowerCAmelCase_      ,							lowerCAmelCase_       ):
														_SCREAMING_SNAKE_CASE      :		List[str]							=						len(lowerCAmelCase_       )
										else:
														_SCREAMING_SNAKE_CASE      :		Any							=						prompt_embeds.shape[0]

										if prompt_embeds is None:
														_SCREAMING_SNAKE_CASE      :		Any							=						self.tokenizer(
														    lowerCAmelCase_      ,							padding='max_length'      ,							max_length=self.tokenizer.model_max_length      ,							truncation=lowerCAmelCase_      ,							return_tensors='pt'      ,							)
														_SCREAMING_SNAKE_CASE      :		Dict							=						text_inputs.input_ids
														_SCREAMING_SNAKE_CASE      :		Dict							=						self.tokenizer(lowerCAmelCase_      ,							padding='longest'      ,							return_tensors='pt'       ).input_ids

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

										_SCREAMING_SNAKE_CASE      :		str							=						prompt_embeds.shape
										# duplicate text embeddings for each generation per prompt, using mps friendly method
										_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						prompt_embeds.repeat(1      ,							lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		int							=						prompt_embeds.reshape(bs_embed * num_images_per_prompt      ,							lowerCAmelCase_      ,							-1       )

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

														_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						text_input_ids.shape[-1]
														_SCREAMING_SNAKE_CASE      :		Dict							=						self.tokenizer(
														    lowerCAmelCase_      ,							padding='max_length'      ,							max_length=lowerCAmelCase_      ,							truncation=lowerCAmelCase_      ,							return_tensors='pt'      ,							)

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

														_SCREAMING_SNAKE_CASE      :		int							=						self.text_encoder(
														    input_ids=uncond_input.input_ids.int().to(lowerCAmelCase_       )      ,							)
														_SCREAMING_SNAKE_CASE      :		List[Any]							=						uncond_embeddings[0]

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

										return prompt_embeds









"""simple docstring"""
def 							lowercase__		(				lowerCamelCase					):
				return -num if num < 0 else num





def 							lowercase__		(				lowerCamelCase					):
				if len(lowerCamelCase					) == 0:
								raise ValueError('abs_min() arg is an empty sequence'					)
				_SCREAMING_SNAKE_CASE      :		List[Any]							=						x[0]
				for i in x:
								if abs_val(lowerCamelCase					) < abs_val(lowerCamelCase					):
												_SCREAMING_SNAKE_CASE      :		List[Any]							=						i
				return j





def 							lowercase__		(				lowerCamelCase					):
				if len(lowerCamelCase					) == 0:
								raise ValueError('abs_max() arg is an empty sequence'					)
				_SCREAMING_SNAKE_CASE      :		Tuple							=						x[0]
				for i in x:
								if abs(lowerCamelCase					) > abs(lowerCamelCase					):
												_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						i
				return j





def 							lowercase__		(				lowerCamelCase					):
				if len(lowerCamelCase					) == 0:
								raise ValueError('abs_max_sort() arg is an empty sequence'					)
				return sorted(lowerCamelCase,	key=lowerCamelCase					)[-1]





def 							lowercase__		(				):
				assert abs_val(0					) == 0
				assert abs_val(34					) == 34
				assert abs_val(-100_000_000_000					) == 100_000_000_000

				_SCREAMING_SNAKE_CASE      :		Tuple							=						[-3, -1, 2, -11]
				assert abs_max(lowerCamelCase					) == -11
				assert abs_max_sort(lowerCamelCase					) == -11
				assert abs_min(lowerCamelCase					) == -1


if __name__ == "__main__":
			import doctest

			doctest.testmod()

			test_abs_val()
			print(abs_val(-34))  # --> 34

