


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 typing import List, Optional, Union

import numpy as np

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


_snake_case	:						List[str]					=					logging.get_logger(__name__)






class    _UpperCAmelCase  (      lowercase_					):
					UpperCamelCase						=				['''input_values''', '''attention_mask''']


					def __init__(   self     :str	,			__UpperCamelCase     :List[str]=1	,			__UpperCamelCase     :Tuple=1_60_00	,			__UpperCamelCase     :Optional[int]=0.0	,			__UpperCamelCase     :Optional[int]=False	,			__UpperCamelCase     :Optional[Any]=True	,			**__UpperCamelCase     :Optional[Any]	,			):
												super().__init__(feature_size=__UpperCamelCase	,			sampling_rate=__UpperCamelCase	,			padding_value=__UpperCamelCase	,			**__UpperCamelCase						)
												A						=    return_attention_mask
												A						=    do_normalize


					@staticmethod
					def       lowerCamelCase			(   __UpperCamelCase     :List[np.ndarray]	,			__UpperCamelCase     :List[np.ndarray]	,			__UpperCamelCase     :float = 0.0						):
												if attention_mask is not None:
																			A						=    np.array(__UpperCamelCase	,			np.intaa						)
																			A						=    []

																			for vector, length in zip(__UpperCamelCase	,			attention_mask.sum(-1						)						):
																										A						=    (vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1e-7						)
																										if length < normed_slice.shape[0]:
																																	A						=    padding_value

																										normed_input_values.append(__UpperCamelCase						)
												else:
																			A						=    [(x - x.mean()) / np.sqrt(x.var() + 1e-7						) for x in input_values]

												return normed_input_values


					def __call__(   self     :Tuple	,			__UpperCamelCase     :Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]	,			__UpperCamelCase     :Union[bool, str, PaddingStrategy] = False	,			__UpperCamelCase     :Optional[int] = None	,			__UpperCamelCase     :bool = False	,			__UpperCamelCase     :Optional[int] = None	,			__UpperCamelCase     :Optional[bool] = None	,			__UpperCamelCase     :Optional[Union[str, TensorType]] = None	,			__UpperCamelCase     :Optional[int] = None	,			**__UpperCamelCase     :List[str]	,			):

												if sampling_rate is not None:
																			if sampling_rate != self.sampling_rate:
																										raise ValueError(
																										    f"The model corresponding to this feature extractor: {self} was trained using a sampling rate of"
																										    f" {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled with"
																										    f" {self.sampling_rate} and not {sampling_rate}."						)
												else:
																			logger.warning(
																			    "It is strongly recommended to pass the ``sampling_rate`` argument to this function. "
																			    "Failing to do so can result in silent errors that might be hard to debug."						)

												A						=    isinstance(__UpperCamelCase	,			np.ndarray						) and len(raw_speech.shape						) > 1
												if is_batched_numpy and len(raw_speech.shape						) > 2:
																			raise ValueError(f"Only mono-channel audio is supported for input to {self}"						)
												A						=    is_batched_numpy or (
												    isinstance(__UpperCamelCase	,			(list, tuple)						) and (isinstance(raw_speech[0]	,			(np.ndarray, tuple, list)						))
												)

												# always return batch
												if not is_batched:
																			A						=    [raw_speech]

												# convert into correct format for padding
												A						=    BatchFeature({"input_values": raw_speech}						)

												A						=    self.pad(
												    __UpperCamelCase	,			padding=__UpperCamelCase	,			max_length=__UpperCamelCase	,			truncation=__UpperCamelCase	,			pad_to_multiple_of=__UpperCamelCase	,			return_attention_mask=__UpperCamelCase	,			)

												# convert input values to correct format
												A						=    padded_inputs["input_values"]
												if not isinstance(input_values[0]	,			np.ndarray						):
																			A						=    [np.asarray(__UpperCamelCase	,			dtype=np.floataa						) for array in input_values]
												elif (
												    not isinstance(__UpperCamelCase	,			np.ndarray						)
												    and isinstance(input_values[0]	,			np.ndarray						)
												    and input_values[0].dtype is np.dtype(np.floataa						)
												):
																			A						=    [array.astype(np.floataa						) for array in input_values]
												elif isinstance(__UpperCamelCase	,			np.ndarray						) and input_values.dtype is np.dtype(np.floataa						):
																			A						=    input_values.astype(np.floataa						)

												# convert attention_mask to correct format
												A						=    padded_inputs.get("attention_mask"						)
												if attention_mask is not None:
																			A						=    [np.asarray(__UpperCamelCase	,			dtype=np.intaa						) for array in attention_mask]

												# zero-mean and unit-variance normalization
												if self.do_normalize:
																			A						=    (
																			    attention_mask
																			    if self._get_padding_strategies(__UpperCamelCase	,			max_length=__UpperCamelCase						) is not PaddingStrategy.DO_NOT_PAD
																			    else None
																			)
																			A						=    self.zero_mean_unit_var_norm(
																			    padded_inputs["input_values"]	,			attention_mask=__UpperCamelCase	,			padding_value=self.padding_value						)

												if return_tensors is not None:
																			A						=    padded_inputs.convert_to_tensors(__UpperCamelCase						)

												return padded_inputs


