



'''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__	:     Optional[int]    =			getLogger(__name__)


UpperCamelCase__	:     Optional[int]    =			4
UpperCamelCase__	:     Any    =			7

UpperCamelCase__	:     Tuple    =			{
    '''float16''': torch.floataa,
    '''float32''': torch.floataa,
}


def     lowerCAmelCase_       (     _lowerCamelCase:       Tuple ):
		if isinstance(_lowerCamelCase	,				torch.Tensor ):
				return image
		elif isinstance(_lowerCamelCase	,				PIL.Image.Image ):
				__SCREAMING_SNAKE_CASE	:		Any   	=		[image]

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

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

		return image






class 						_UpperCamelCase  (							lowerCamelCase__   ):


			'''simple docstring'''

			def __init__(							self					:			List[Any]							,					lowerCAmelCase__					:			OnnxRuntimeModel							,					lowerCAmelCase__					:			OnnxRuntimeModel							,					lowerCAmelCase__					:			Any							,					lowerCAmelCase__					:			OnnxRuntimeModel							,					lowerCAmelCase__					:			DDPMScheduler							,					lowerCAmelCase__					:			Any							,					lowerCAmelCase__					:			int = 3_5_0							,					):





					"""simple docstring"""






					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					:			Optional[Any]							,					lowerCAmelCase__					:			Union[str, List[str]]							,					lowerCAmelCase__					:			Union[torch.FloatTensor, PIL.Image.Image, List[PIL.Image.Image]]							,					lowerCAmelCase__					:			int = 7_5							,					lowerCAmelCase__					:			float = 9.0							,					lowerCAmelCase__					:			int = 2_0							,					lowerCAmelCase__					:			Optional[Union[str, List[str]]] = None							,					lowerCAmelCase__					:			Optional[int] = 1							,					lowerCAmelCase__					:			float = 0.0							,					lowerCAmelCase__					:			Optional[Union[torch.Generator, List[torch.Generator]]] = None							,					lowerCAmelCase__					:			Optional[torch.FloatTensor] = None							,					lowerCAmelCase__					:			Optional[np.ndarray] = None							,					lowerCAmelCase__					:			Optional[np.ndarray] = None							,					lowerCAmelCase__					:			Optional[str] = "pil"							,					lowerCAmelCase__					:			bool = True							,					lowerCAmelCase__					:			Optional[Callable[[int, int, torch.FloatTensor], None]] = None							,					lowerCAmelCase__					:			Optional[int] = 1							,					):





					"""simple docstring"""






					self.check_inputs(lowerCAmelCase__							,					lowerCAmelCase__							,					lowerCAmelCase__							,					lowerCAmelCase__						)

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

					__SCREAMING_SNAKE_CASE	:		List[Any]   	=		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	:		Union[str, Any]   	=		guidance_scale > 1.0

					# 3. Encode input prompt
					__SCREAMING_SNAKE_CASE	:		Tuple   	=		self._encode_prompt(
					    lowerCAmelCase__							,					lowerCAmelCase__							,					lowerCAmelCase__							,					lowerCAmelCase__							,					lowerCAmelCase__							,					prompt_embeds=lowerCAmelCase__							,					negative_prompt_embeds=lowerCAmelCase__							,					)

					__SCREAMING_SNAKE_CASE	:		Optional[int]   	=		ORT_TO_PT_TYPE[str(text_embeddings.dtype						)]

					# 4. Preprocess image
					__SCREAMING_SNAKE_CASE	:		Tuple   	=		preprocess(lowerCAmelCase__						)
					__SCREAMING_SNAKE_CASE	:		Dict   	=		image.cpu()

					# 5. set timesteps
					self.scheduler.set_timesteps(lowerCAmelCase__							,					device=lowerCAmelCase__						)
					__SCREAMING_SNAKE_CASE	:		Any   	=		self.scheduler.timesteps

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

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

					# 6. Prepare latent variables
					__SCREAMING_SNAKE_CASE	,						__SCREAMING_SNAKE_CASE	:		List[Any]   	=		image.shape[2:]
					__SCREAMING_SNAKE_CASE	:		List[Any]   	=		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	:		List[str]   	=		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	:		List[str]   	=		next(
					    (input.type for input in self.unet.model.get_inputs() if input.name == """timestep""")							,					"""tensor(float)"""						)
					__SCREAMING_SNAKE_CASE	:		Tuple   	=		ORT_TO_NP_TYPE[timestep_dtype]

					# 9. Denoising loop
					__SCREAMING_SNAKE_CASE	:		List[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	:		List[Any]   	=		np.concatenate([latents] * 2						) if do_classifier_free_guidance else latents

									# concat latents, mask, masked_image_latents in the channel dimension
									__SCREAMING_SNAKE_CASE	:		Dict   	=		self.scheduler.scale_model_input(lowerCAmelCase__							,					lowerCAmelCase__						)
									__SCREAMING_SNAKE_CASE	:		str   	=		np.concatenate([latent_model_input, image]							,					axis=1						)

									# timestep to tensor
									__SCREAMING_SNAKE_CASE	:		int   	=		np.array([t]							,					dtype=lowerCAmelCase__						)

									# predict the noise residual
									__SCREAMING_SNAKE_CASE	:		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	,						__SCREAMING_SNAKE_CASE	:		str   	=		np.split(lowerCAmelCase__							,					2						)
									__SCREAMING_SNAKE_CASE	:		Union[str, Any]   	=		noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

									# compute the previous noisy sample x_t -> x_t-1
									__SCREAMING_SNAKE_CASE	:		Optional[int]   	=		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	:		List[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 UpperCamelCase__ (							self					:			List[str]							,					lowerCAmelCase__					:			Any						):





					"""simple docstring"""






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



			def UpperCamelCase__ (							self					:			Optional[Any]							,					lowerCAmelCase__					:			Union[str, List[str]]							,					lowerCAmelCase__					:			str							,					lowerCAmelCase__					:			Optional[int]							,					lowerCAmelCase__					:			bool							,					lowerCAmelCase__					:			Optional[str]							,					lowerCAmelCase__					:			Optional[torch.FloatTensor] = None							,					lowerCAmelCase__					:			Optional[torch.FloatTensor] = None							,					):





					"""simple docstring"""






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

					if prompt_embeds is None:
							__SCREAMING_SNAKE_CASE	:		Dict   	=		self.tokenizer(
							    lowerCAmelCase__							,					padding="""max_length"""							,					max_length=self.tokenizer.model_max_length							,					truncation=lowerCAmelCase__							,					return_tensors="""pt"""							,					)
							__SCREAMING_SNAKE_CASE	:		Any   	=		text_inputs.input_ids
							__SCREAMING_SNAKE_CASE	:		Tuple   	=		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	:		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	:		List[str]   	=		self.text_encoder(
							    input_ids=text_input_ids.int().to(lowerCAmelCase__						)							,					)
							__SCREAMING_SNAKE_CASE	:		List[Any]   	=		prompt_embeds[0]

					__SCREAMING_SNAKE_CASE	,						__SCREAMING_SNAKE_CASE	,						__SCREAMING_SNAKE_CASE	:		Any   	=		prompt_embeds.shape
					# duplicate text embeddings for each generation per prompt, using mps friendly method
					__SCREAMING_SNAKE_CASE	:		Union[str, Any]   	=		prompt_embeds.repeat(1							,					lowerCAmelCase__						)
					__SCREAMING_SNAKE_CASE	:		str   	=		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	:		Any   	=		[""""""] * 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	:		Union[str, 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	:		str   	=		text_input_ids.shape[-1]
							__SCREAMING_SNAKE_CASE	:		List[Any]   	=		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	:		List[Any]   	=		self.text_encoder(
							    input_ids=uncond_input.input_ids.int().to(lowerCAmelCase__						)							,					)
							__SCREAMING_SNAKE_CASE	:		int   	=		uncond_embeddings[0]

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

					return prompt_embeds




'''simple docstring'''




import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin






class 						_UpperCamelCase  (							lowerCamelCase__   ):


			'''simple docstring'''

			_A				:		int         =		['''image_processor''', '''tokenizer''']
			_A				:		Union[str, Any]         =		'''AutoImageProcessor'''
			_A				:		Optional[Any]         =		'''AutoTokenizer'''

			def __init__(							self					:			Tuple							,					lowerCAmelCase__					:			Tuple=None							,					lowerCAmelCase__					:			Union[str, Any]=None							,					**lowerCAmelCase__					:			Any						):





					"""simple docstring"""






					__SCREAMING_SNAKE_CASE	:		Tuple   	=		None
					if "feature_extractor" in kwargs:
							warnings.warn(
							    """The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"""
							    """ instead."""							,					lowerCAmelCase__							,					)
							__SCREAMING_SNAKE_CASE	:		str   	=		kwargs.pop("""feature_extractor"""						)

					__SCREAMING_SNAKE_CASE	:		Any   	=		image_processor if image_processor is not None else feature_extractor
					if image_processor is None:
							raise ValueError("""You need to specify an `image_processor`."""						)
					if tokenizer is None:
							raise ValueError("""You need to specify a `tokenizer`."""						)

					super().__init__(lowerCAmelCase__							,					lowerCAmelCase__						)
					__SCREAMING_SNAKE_CASE	:		List[str]   	=		self.image_processor
					__SCREAMING_SNAKE_CASE	:		Dict   	=		False

			def __call__(							self					:			List[str]							,					*lowerCAmelCase__					:			List[str]							,					**lowerCAmelCase__					:			Dict						):





					"""simple docstring"""






					if self._in_target_context_manager:
							return self.current_processor(*lowerCAmelCase__							,					**lowerCAmelCase__						)

					__SCREAMING_SNAKE_CASE	:		Optional[Any]   	=		kwargs.pop("""images"""							,					lowerCAmelCase__						)
					__SCREAMING_SNAKE_CASE	:		Union[str, Any]   	=		kwargs.pop("""text"""							,					lowerCAmelCase__						)
					if len(lowerCAmelCase__						) > 0:
							__SCREAMING_SNAKE_CASE	:		Tuple   	=		args[0]
							__SCREAMING_SNAKE_CASE	:		int   	=		args[1:]

					if images is None and text is None:
							raise ValueError("""You need to specify either an `images` or `text` input to process."""						)

					if images is not None:
							__SCREAMING_SNAKE_CASE	:		List[str]   	=		self.image_processor(lowerCAmelCase__							,					*lowerCAmelCase__							,					**lowerCAmelCase__						)
					if text is not None:
							__SCREAMING_SNAKE_CASE	:		Optional[int]   	=		self.tokenizer(lowerCAmelCase__							,					**lowerCAmelCase__						)

					if text is None:
							return inputs
					elif images is None:
							return encodings
					else:
							__SCREAMING_SNAKE_CASE	:		int   	=		encodings["""input_ids"""]
							return inputs

			def UpperCamelCase__ (							self					:			Any							,					*lowerCAmelCase__					:			Any							,					**lowerCAmelCase__					:			List[str]						):





					"""simple docstring"""






					return self.tokenizer.batch_decode(*lowerCAmelCase__							,					**lowerCAmelCase__						)

			def UpperCamelCase__ (							self					:			int							,					*lowerCAmelCase__					:			str							,					**lowerCAmelCase__					:			Dict						):





					"""simple docstring"""






					return self.tokenizer.decode(*lowerCAmelCase__							,					**lowerCAmelCase__						)

			@contextmanager
			def UpperCamelCase__ (							self					:			Union[str, Any]						):





					"""simple docstring"""






					warnings.warn(
					    """`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your """
					    """labels by using the argument `text` of the regular `__call__` method (either in the same call as """
					    """your images inputs, or in a separate call."""						)
					__SCREAMING_SNAKE_CASE	:		List[Any]   	=		True
					__SCREAMING_SNAKE_CASE	:		Optional[Any]   	=		self.tokenizer
					yield
					__SCREAMING_SNAKE_CASE	:		Union[str, Any]   	=		self.image_processor
					__SCREAMING_SNAKE_CASE	:		List[str]   	=		False

			@property
			def UpperCamelCase__ (							self					:			Optional[Any]						):





					"""simple docstring"""






					warnings.warn(
					    """`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."""							,					lowerCAmelCase__							,					)
					return self.image_processor_class



			@property
			def UpperCamelCase__ (							self					:			List[Any]						):





					"""simple docstring"""






					warnings.warn(
					    """`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead."""							,					lowerCAmelCase__							,					)
					return self.image_processor