


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 			_a      (    __UpperCamelCase			):
     if isinstance(__UpperCamelCase	,	torch.Tensor			):
          return image
     elif isinstance(__UpperCamelCase	,	PIL.Image.Image			):
          a_     :  int				      =     [image]

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

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

     return image


class 	a__						(    lowerCAmelCase_     ):

       def __init__( self   :  Any   ,				lowerCamelCase_   :  OnnxRuntimeModel   ,				lowerCamelCase_   :  OnnxRuntimeModel   ,				lowerCamelCase_   :  Any   ,				lowerCamelCase_   :  OnnxRuntimeModel   ,				lowerCamelCase_   :  DDPMScheduler   ,				lowerCamelCase_   :  Any   ,				lowerCamelCase_   :  int = 3_5_0   ,				):
            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   :  Tuple   ,				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   ,				):
            self.check_inputs(lowerCamelCase_   ,				lowerCamelCase_   ,				lowerCamelCase_   ,				lowerCamelCase_     )

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

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

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

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

            # 4. Preprocess image
            a_     :  Tuple				      =     preprocess(lowerCamelCase_     )
            a_     :  Dict				      =     image.cpu()

            # 5. set timesteps
            self.scheduler.set_timesteps(lowerCamelCase_   ,				device=lowerCamelCase_     )
            a_     :  Union[str, Any]				      =     self.scheduler.timesteps

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

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

            # 6. Prepare latent variables
            a_      ,      a_     :  Union[str, Any]				      =     image.shape[2:]
            a_     :  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
            a_     :  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
            a_     :  Optional[int]				      =     self.prepare_extra_step_kwargs(lowerCamelCase_   ,				lowerCamelCase_     )

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

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

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

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

                      # predict the noise residual
                      a_     :  Tuple				      =     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:
                           a_      ,      a_     :  Dict				      =     np.split(lowerCamelCase_   ,				2     )
                      a_     :  int				      =     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(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
            a_     :  List[str]				      =     self.decode_latents(latents.float()     )

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

            if not return_dict:
                 return (image,)

            return ImagePipelineOutput(images=lowerCamelCase_     )

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

       def     UpperCAmelCase( self   :  Dict   ,				lowerCamelCase_   :  Union[str, List[str]]   ,				lowerCamelCase_   :  Optional[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_     ):
                 a_     :  Optional[int]				      =     1
            elif prompt is not None and isinstance(lowerCamelCase_   ,				lowerCamelCase_     ):
                 a_     :  Optional[int]				      =     len(lowerCamelCase_     )
            else:
                 a_     :  str				      =     prompt_embeds.shape[0]

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

            a_      ,      a_      ,      a_     :  List[str]				      =     prompt_embeds.shape
            # duplicate text embeddings for each generation per prompt, using mps friendly method
            a_     :  List[str]				      =     prompt_embeds.repeat(1   ,				lowerCamelCase_     )
            a_     :  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:
                 a_     :  List[str]
                 if negative_prompt is None:
                      a_     :  List[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_     ):
                      a_     :  Optional[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:
                      a_     :  Union[str, Any]				      =     negative_prompt

                 a_     :  Tuple				      =     text_input_ids.shape[-1]
                 a_     :  Tuple				      =     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

                 a_     :  Union[str, Any]				      =     self.text_encoder(
                     input_ids=uncond_input.input_ids.int().to(lowerCamelCase_     )   ,				)
                 a_     :  str				      =     uncond_embeddings[0]

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

            return prompt_embeds







'''simple docstring'''


from ..utils import deprecate
from .attention_processor import (  # noqa: F401
    Attention,
    AttentionProcessor,
    AttnAddedKVProcessor,
    AttnProcessor2_0,
    LoRAAttnProcessor,
    LoRALinearLayer,
    LoRAXFormersAttnProcessor,
    SlicedAttnAddedKVProcessor,
    SlicedAttnProcessor,
    XFormersAttnProcessor,
)
from .attention_processor import AttnProcessor as AttnProcessorRename  # noqa: F401


deprecate(
    'cross_attention',
    '0.20.0',
    'Importing from cross_attention is deprecated. Please import from diffusers.models.attention_processor instead.',
    standard_warn=False,
)


lowerCamelCase__     						=  AttentionProcessor






class 						lowerCAmelCase__      (      UpperCAmelCase__						):




			def __init__(		self	:  Union[str, Any]						,	*lowerCamelCase__	:  int						,	**lowerCamelCase__	:  List[str]			) ->Any:



					'''simple docstring'''







					_UpperCAmelCase		:							Dict             =   F"""{self.__class__.__name__} is deprecated and will be removed in `0.20.0`. Please use `from diffusers.models.attention_processor import {''.join(self.__class__.__name__.split('Cross'			)			)} instead."""
					deprecate("cross_attention"						,	"0.20.0"						,	lowerCamelCase__						,	standard_warn=lowerCamelCase__			)
					super().__init__(*lowerCamelCase__						,	**lowerCamelCase__			)






class 						lowerCAmelCase__      (      UpperCAmelCase__						):




			def __init__(		self	:  Optional[Any]						,	*lowerCamelCase__	:  Any						,	**lowerCamelCase__	:  Tuple			) ->List[Any]:



					'''simple docstring'''







					_UpperCAmelCase		:							int             =   F"""{self.__class__.__name__} is deprecated and will be removed in `0.20.0`. Please use `from diffusers.models.attention_processor import {''.join(self.__class__.__name__.split('Cross'			)			)} instead."""
					deprecate("cross_attention"						,	"0.20.0"						,	lowerCamelCase__						,	standard_warn=lowerCamelCase__			)
					super().__init__(*lowerCamelCase__						,	**lowerCamelCase__			)






class 						lowerCAmelCase__      (      UpperCAmelCase__						):




			def __init__(		self	:  List[Any]						,	*lowerCamelCase__	:  List[Any]						,	**lowerCamelCase__	:  List[str]			) ->List[Any]:



					'''simple docstring'''







					_UpperCAmelCase		:							List[str]             =   F"""{self.__class__.__name__} is deprecated and will be removed in `0.20.0`. Please use `from diffusers.models.attention_processor import {''.join(self.__class__.__name__.split('Cross'			)			)} instead."""
					deprecate("cross_attention"						,	"0.20.0"						,	lowerCamelCase__						,	standard_warn=lowerCamelCase__			)
					super().__init__(*lowerCamelCase__						,	**lowerCamelCase__			)






class 						lowerCAmelCase__      (      UpperCAmelCase__						):




			def __init__(		self	:  int						,	*lowerCamelCase__	:  Tuple						,	**lowerCamelCase__	:  Union[str, Any]			) ->int:



					'''simple docstring'''







					_UpperCAmelCase		:							Dict             =   F"""{self.__class__.__name__} is deprecated and will be removed in `0.20.0`. Please use `from diffusers.models.attention_processor import {''.join(self.__class__.__name__.split('Cross'			)			)} instead."""
					deprecate("cross_attention"						,	"0.20.0"						,	lowerCamelCase__						,	standard_warn=lowerCamelCase__			)
					super().__init__(*lowerCamelCase__						,	**lowerCamelCase__			)






class 						lowerCAmelCase__      (      UpperCAmelCase__						):




			def __init__(		self	:  int						,	*lowerCamelCase__	:  List[str]						,	**lowerCamelCase__	:  Tuple			) ->Union[str, Any]:



					'''simple docstring'''







					_UpperCAmelCase		:							Optional[Any]             =   F"""{self.__class__.__name__} is deprecated and will be removed in `0.20.0`. Please use `from diffusers.models.attention_processor import {''.join(self.__class__.__name__.split('Cross'			)			)} instead."""
					deprecate("cross_attention"						,	"0.20.0"						,	lowerCamelCase__						,	standard_warn=lowerCamelCase__			)
					super().__init__(*lowerCamelCase__						,	**lowerCamelCase__			)






class 						lowerCAmelCase__      (      UpperCAmelCase__						):




			def __init__(		self	:  List[Any]						,	*lowerCamelCase__	:  int						,	**lowerCamelCase__	:  Optional[int]			) ->Optional[int]:



					'''simple docstring'''







					_UpperCAmelCase		:							List[Any]             =   F"""{self.__class__.__name__} is deprecated and will be removed in `0.20.0`. Please use `from diffusers.models.attention_processor import {''.join(self.__class__.__name__.split('Cross'			)			)} instead."""
					deprecate("cross_attention"						,	"0.20.0"						,	lowerCamelCase__						,	standard_warn=lowerCamelCase__			)
					super().__init__(*lowerCamelCase__						,	**lowerCamelCase__			)






class 						lowerCAmelCase__      (      UpperCAmelCase__						):




			def __init__(		self	:  Any						,	*lowerCamelCase__	:  Optional[Any]						,	**lowerCamelCase__	:  Union[str, Any]			) ->List[str]:



					'''simple docstring'''







					_UpperCAmelCase		:							int             =   F"""{self.__class__.__name__} is deprecated and will be removed in `0.20.0`. Please use `from diffusers.models.attention_processor import {''.join(self.__class__.__name__.split('Cross'			)			)} instead."""
					deprecate("cross_attention"						,	"0.20.0"						,	lowerCamelCase__						,	standard_warn=lowerCamelCase__			)
					super().__init__(*lowerCamelCase__						,	**lowerCamelCase__			)






class 						lowerCAmelCase__      (      UpperCAmelCase__						):




			def __init__(		self	:  Optional[Any]						,	*lowerCamelCase__	:  List[Any]						,	**lowerCamelCase__	:  Dict			) ->Any:



					'''simple docstring'''







					_UpperCAmelCase		:							int             =   F"""{self.__class__.__name__} is deprecated and will be removed in `0.20.0`. Please use `from diffusers.models.attention_processor import {''.join(self.__class__.__name__.split('Cross'			)			)} instead."""
					deprecate("cross_attention"						,	"0.20.0"						,	lowerCamelCase__						,	standard_warn=lowerCamelCase__			)
					super().__init__(*lowerCamelCase__						,	**lowerCamelCase__			)



