


"""simple docstring"""







from dataclasses import dataclass
from typing import Callable, Optional

import datasets
@dataclass
class 				lowerCamelCase							(      datasets.BuilderConfig			):


							"""simple docstring"""
							SCREAMING_SNAKE_CASE              =   None
							SCREAMING_SNAKE_CASE              =   None
							SCREAMING_SNAKE_CASE              =   None






							def        lowerCAmelCase  (    self			:  Optional[int]					):
													assert self.generator is not None, "generator must be specified"

													if self.gen_kwargs is None:
																			UpperCAmelCase__												=     {}




class 				lowerCamelCase							(      datasets.GeneratorBasedBuilder			):


							"""simple docstring"""
							SCREAMING_SNAKE_CASE              =   GeneratorConfig






							def        lowerCAmelCase  (    self			:  Optional[Any]					):
													return datasets.DatasetInfo(features=self.config.features					)






							def        lowerCAmelCase  (    self			:  Tuple		,			lowerCamelCase_			:  str					):
													return [datasets.SplitGenerator(name=datasets.Split.TRAIN		,			gen_kwargs=self.config.gen_kwargs					)]






							def        lowerCAmelCase  (    self			:  int		,			**lowerCamelCase_			:  List[str]					):
													for idx, ex in enumerate(self.config.generator(**lowerCamelCase_					)					):
																			yield idx, ex



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


__a      			= getLogger(__name__)


__a      			= 4
__a      			= 7

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






def        a__			(							_UpperCamelCase			):





						"""simple docstring"""

						if isinstance(_UpperCamelCase						,      torch.Tensor			):
												return image
						elif isinstance(_UpperCamelCase						,      PIL.Image.Image			):
												UpperCAmelCase__												=     [image]

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

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

						return image


class 				lowerCamelCase							(      __SCREAMING_SNAKE_CASE			):


							"""simple docstring"""






							def __init__(    self			:  Optional[int]		,			lowerCamelCase_			:  OnnxRuntimeModel		,			lowerCamelCase_			:  OnnxRuntimeModel		,			lowerCamelCase_			:  Any		,			lowerCamelCase_			:  OnnxRuntimeModel		,			lowerCamelCase_			:  DDPMScheduler		,			lowerCamelCase_			:  Any		,			lowerCamelCase_			:  int = 350		,			):
													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[int]		,			lowerCamelCase_			:  Union[str, List[str]]		,			lowerCamelCase_			:  Union[torch.FloatTensor, PIL.Image.Image, List[PIL.Image.Image]]		,			lowerCamelCase_			:  int = 75		,			lowerCamelCase_			:  float = 9.0		,			lowerCamelCase_			:  int = 20		,			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		,			):
													self.check_inputs(lowerCamelCase_		,			lowerCamelCase_		,			lowerCamelCase_		,			lowerCamelCase_					)

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

													UpperCAmelCase__												=     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__												=     guidance_scale > 1.0

													# 3. Encode input prompt
													UpperCAmelCase__												=     self._encode_prompt(
													    lowerCamelCase_		,			lowerCamelCase_		,			lowerCamelCase_		,			lowerCamelCase_		,			lowerCamelCase_		,			prompt_embeds=lowerCamelCase_		,			negative_prompt_embeds=lowerCamelCase_		,			)

													UpperCAmelCase__												=     ORT_TO_PT_TYPE[str(text_embeddings.dtype					)]

													# 4. Preprocess image
													UpperCAmelCase__												=     preprocess(lowerCamelCase_					)
													UpperCAmelCase__												=     image.cpu()

													# 5. set timesteps
													self.scheduler.set_timesteps(lowerCamelCase_		,			device=lowerCamelCase_					)
													UpperCAmelCase__												=     self.scheduler.timesteps

													# 5. Add noise to image
													UpperCAmelCase__												=     torch.tensor([noise_level]		,			dtype=torch.long		,			device=lowerCamelCase_					)
													UpperCAmelCase__												=     torch.randn(image.shape		,			generator=lowerCamelCase_		,			device=lowerCamelCase_		,			dtype=lowerCamelCase_					)
													UpperCAmelCase__												=     self.low_res_scheduler.add_noise(lowerCamelCase_		,			lowerCamelCase_		,			lowerCamelCase_					)

													UpperCAmelCase__												=     2 if do_classifier_free_guidance else 1
													UpperCAmelCase__												=     np.concatenate([image] * batch_multiplier * num_images_per_prompt					)
													UpperCAmelCase__												=     np.concatenate([noise_level] * image.shape[0]					)

													# 6. Prepare latent variables
													UpperCAmelCase__						,					UpperCAmelCase__												=     image.shape[2:]
													UpperCAmelCase__												=     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
													UpperCAmelCase__												=     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__												=     self.prepare_extra_step_kwargs(lowerCamelCase_		,			lowerCamelCase_					)

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

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

																									# concat latents, mask, masked_image_latents in the channel dimension
																									UpperCAmelCase__												=     self.scheduler.scale_model_input(lowerCamelCase_		,			lowerCamelCase_					)
																									UpperCAmelCase__												=     np.concatenate([latent_model_input, image]		,			axis=1					)

																									# timestep to tensor
																									UpperCAmelCase__												=     np.array([t]		,			dtype=lowerCamelCase_					)

																									# predict the noise residual
																									UpperCAmelCase__												=     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:
																															UpperCAmelCase__						,					UpperCAmelCase__												=     np.split(lowerCamelCase_		,			2					)
																									UpperCAmelCase__												=     noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

																									# compute the previous noisy sample x_t -> x_t-1
																									UpperCAmelCase__												=     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
													UpperCAmelCase__												=     self.decode_latents(latents.float()					)

													# 11. Convert to PIL
													if output_type == "pil":
																			UpperCAmelCase__												=     self.numpy_to_pil(lowerCamelCase_					)

													if not return_dict:
																			return (image,)

													return ImagePipelineOutput(images=lowerCamelCase_					)






							def        lowerCAmelCase  (    self			:  Tuple		,			lowerCamelCase_			:  List[Any]					):
													UpperCAmelCase__												=     1 / 0.08333 * latents
													UpperCAmelCase__												=     self.vae(latent_sample=lowerCamelCase_					)[0]
													UpperCAmelCase__												=     np.clip(image / 2 + 0.5		,			0		,			1					)
													UpperCAmelCase__												=     image.transpose((0, 2, 3, 1)					)
													return image






							def        lowerCAmelCase  (    self			:  Optional[Any]		,			lowerCamelCase_			:  Union[str, List[str]]		,			lowerCamelCase_			:  Union[str, Any]		,			lowerCamelCase_			:  Optional[int]		,			lowerCamelCase_			:  bool		,			lowerCamelCase_			:  Optional[str]		,			lowerCamelCase_			:  Optional[torch.FloatTensor] = None		,			lowerCamelCase_			:  Optional[torch.FloatTensor] = None		,			):
													if prompt is not None and isinstance(lowerCamelCase_		,			lowerCamelCase_					):
																			UpperCAmelCase__												=     1
													elif prompt is not None and isinstance(lowerCamelCase_		,			lowerCamelCase_					):
																			UpperCAmelCase__												=     len(lowerCamelCase_					)
													else:
																			UpperCAmelCase__												=     prompt_embeds.shape[0]

													if prompt_embeds is None:
																			UpperCAmelCase__												=     self.tokenizer(
																			    lowerCamelCase_		,			padding='''max_length'''		,			max_length=self.tokenizer.model_max_length		,			truncation=lowerCamelCase_		,			return_tensors='''pt'''		,			)
																			UpperCAmelCase__												=     text_inputs.input_ids
																			UpperCAmelCase__												=     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_					):
																									UpperCAmelCase__												=     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__												=     self.text_encoder(
																			    input_ids=text_input_ids.int().to(lowerCamelCase_					)		,			)
																			UpperCAmelCase__												=     prompt_embeds[0]

													UpperCAmelCase__						,					UpperCAmelCase__						,					UpperCAmelCase__												=     prompt_embeds.shape
													# duplicate text embeddings for each generation per prompt, using mps friendly method
													UpperCAmelCase__												=     prompt_embeds.repeat(1		,			lowerCamelCase_					)
													UpperCAmelCase__												=     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:
																			UpperCAmelCase__												=     42
																			if negative_prompt is None:
																									UpperCAmelCase__												=     [''''''] * 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_					):
																									UpperCAmelCase__												=     [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:
																									UpperCAmelCase__												=     negative_prompt

																			UpperCAmelCase__												=     text_input_ids.shape[-1]
																			UpperCAmelCase__												=     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

																			UpperCAmelCase__												=     self.text_encoder(
																			    input_ids=uncond_input.input_ids.int().to(lowerCamelCase_					)		,			)
																			UpperCAmelCase__												=     uncond_embeddings[0]

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

													return prompt_embeds