
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   						=     getLogger(__name__)


__UpperCAmelCase   						=     4
__UpperCAmelCase   						=     7

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


def  A__ (       __lowerCamelCase						):
   if isinstance(__lowerCamelCase,  torch.Tensor						):
      return image
   elif isinstance(__lowerCamelCase,  PIL.Image.Image						):
      SCREAMING_SNAKE_CASE_	=      [image]

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

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

   return image





class 				UpperCamelCase__		(	__SCREAMING_SNAKE_CASE	):






       """simple docstring"""


       def __init__(							self ,     _A ,     _A ,     _A ,     _A ,     _A ,     _A ,     _A = 350 ,     )				->				str:
          super().__init__(
              vae=_A ,     text_encoder=_A ,     tokenizer=_A ,     unet=_A ,     low_res_scheduler=_A ,     scheduler=_A ,     safety_checker=_A ,     feature_extractor=_A ,     watermarker=_A ,     max_noise_level=_A ,     )



       def __call__(							self ,     _A ,     _A ,     _A = 75 ,     _A = 9.0 ,     _A = 20 ,     _A = None ,     _A = 1 ,     _A = 0.0 ,     _A = None ,     _A = None ,     _A = None ,     _A = None ,     _A = "pil" ,     _A = True ,     _A = None ,     _A = 1 ,     )				->				Optional[Any]:
          self.check_inputs(_A ,     _A ,     _A ,     _A	)

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

          SCREAMING_SNAKE_CASE_	=      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_	=      guidance_scale > 1.0

          # 3. Encode input prompt
          SCREAMING_SNAKE_CASE_	=      self._encode_prompt(
              _A ,     _A ,     _A ,     _A ,     _A ,     prompt_embeds=_A ,     negative_prompt_embeds=_A ,     )

          SCREAMING_SNAKE_CASE_	=      ORT_TO_PT_TYPE[str(text_embeddings.dtype	)]

          # 4. Preprocess image
          SCREAMING_SNAKE_CASE_	=      preprocess(_A	)
          SCREAMING_SNAKE_CASE_	=      image.cpu()

          # 5. set timesteps
          self.scheduler.set_timesteps(_A ,     device=_A	)
          SCREAMING_SNAKE_CASE_	=      self.scheduler.timesteps

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

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

          # 6. Prepare latent variables
          SCREAMING_SNAKE_CASE_	, SCREAMING_SNAKE_CASE_	=      image.shape[2:]
          SCREAMING_SNAKE_CASE_	=      self.prepare_latents(
              batch_size * num_images_per_prompt ,     _A ,     _A ,     _A ,     _A ,     _A ,     _A ,     _A ,     )

          # 7. Check that sizes of image and latents match
          SCREAMING_SNAKE_CASE_	=      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_	=      self.prepare_extra_step_kwargs(_A ,     _A	)

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

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

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

                # timestep to tensor
                SCREAMING_SNAKE_CASE_	=      np.array([t] ,     dtype=_A	)

                # predict the noise residual
                SCREAMING_SNAKE_CASE_	=      self.unet(
                    sample=_A ,     timestep=_A ,     encoder_hidden_states=_A ,     class_labels=noise_level.astype(np.intaa	) ,     )[0]

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

                # compute the previous noisy sample x_t -> x_t-1
                SCREAMING_SNAKE_CASE_	=      self.scheduler.step(
                    torch.from_numpy(_A	) ,     _A ,     _A ,     **_A	).prev_sample

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

        # 10. Post-processing
          SCREAMING_SNAKE_CASE_	=      self.decode_latents(latents.float()	)

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

          if not return_dict:
             return (image,)

          return ImagePipelineOutput(images=_A	)



       def     _UpperCamelCase	(							self ,     _A	)				->				List[Any]:
          SCREAMING_SNAKE_CASE_	=      1 / 0.0_8333 * latents
          SCREAMING_SNAKE_CASE_	=      self.vae(latent_sample=_A	)[0]
          SCREAMING_SNAKE_CASE_	=      np.clip(image / 2 + 0.5 ,     0 ,     1	)
          SCREAMING_SNAKE_CASE_	=      image.transpose((0, 2, 3, 1)	)
          return image




       def     _UpperCamelCase	(							self ,     _A ,     _A ,     _A ,     _A ,     _A ,     _A = None ,     _A = None ,     )				->				Tuple:
          if prompt is not None and isinstance(_A ,     _A	):
             SCREAMING_SNAKE_CASE_	=      1
          elif prompt is not None and isinstance(_A ,     _A	):
             SCREAMING_SNAKE_CASE_	=      len(_A	)
          else:
             SCREAMING_SNAKE_CASE_	=      prompt_embeds.shape[0]

          if prompt_embeds is None:
             SCREAMING_SNAKE_CASE_	=      self.tokenizer(
                 _A ,     padding='''max_length''' ,     max_length=self.tokenizer.model_max_length ,     truncation=_A ,     return_tensors='''pt''' ,     )
             SCREAMING_SNAKE_CASE_	=      text_inputs.input_ids
             SCREAMING_SNAKE_CASE_	=      self.tokenizer(_A ,     padding='''longest''' ,     return_tensors='''pt'''	).input_ids

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

          SCREAMING_SNAKE_CASE_	, SCREAMING_SNAKE_CASE_	, SCREAMING_SNAKE_CASE_	=      prompt_embeds.shape
          # duplicate text embeddings for each generation per prompt, using mps friendly method
          SCREAMING_SNAKE_CASE_	=      prompt_embeds.repeat(1 ,     _A	)
          SCREAMING_SNAKE_CASE_	=      prompt_embeds.reshape(bs_embed * num_images_per_prompt ,     _A ,     -1	)

          # get unconditional embeddings for classifier free guidance
          if do_classifier_free_guidance and negative_prompt_embeds is None:
             SCREAMING_SNAKE_CASE_	=      42
             if negative_prompt is None:
                SCREAMING_SNAKE_CASE_	=      [''''''] * batch_size
             elif type(_A	) is not type(_A	):
                raise TypeError(
                    F'''`negative_prompt` should be the same type to `prompt`, but got {type(_A	)} !='''
                    F''' {type(_A	)}.'''	)
             elif isinstance(_A ,     _A	):
                SCREAMING_SNAKE_CASE_	=      [negative_prompt]
             elif batch_size != len(_A	):
                raise ValueError(
                    F'''`negative_prompt`: {negative_prompt} has batch size {len(_A	)}, 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_	=      negative_prompt

             SCREAMING_SNAKE_CASE_	=      text_input_ids.shape[-1]
             SCREAMING_SNAKE_CASE_	=      self.tokenizer(
                 _A ,     padding='''max_length''' ,     max_length=_A ,     truncation=_A ,     return_tensors='''pt''' ,     )

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

             SCREAMING_SNAKE_CASE_	=      self.text_encoder(
                 input_ids=uncond_input.input_ids.int().to(_A	) ,     )
             SCREAMING_SNAKE_CASE_	=      uncond_embeddings[0]

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

          return prompt_embeds




import copy

from ...configuration_utils import PretrainedConfig
from ...utils import logging
from ..auto.configuration_auto import AutoConfig


__UpperCAmelCase   						=     logging.get_logger(__name__)





class 				UpperCamelCase__		(	__SCREAMING_SNAKE_CASE	):






       """simple docstring"""


       UpperCAmelCase_				       ="speech-encoder-decoder"
       UpperCAmelCase_				       =True



       def __init__(							self ,     **_A	)				->				Dict:
          super().__init__(**_A	)
          if "encoder" not in kwargs or "decoder" not in kwargs:
             raise ValueError(
                 F'''A configuraton of type {self.model_type} cannot be instantiated because not both `encoder` and'''
                 F''' `decoder` sub-configurations are passed, but only {kwargs}'''	)

          SCREAMING_SNAKE_CASE_	=      kwargs.pop('''encoder'''	)
          SCREAMING_SNAKE_CASE_	=      encoder_config.pop('''model_type'''	)
          SCREAMING_SNAKE_CASE_	=      kwargs.pop('''decoder'''	)
          SCREAMING_SNAKE_CASE_	=      decoder_config.pop('''model_type'''	)

          SCREAMING_SNAKE_CASE_	=      AutoConfig.for_model(_A ,     **_A	)
          SCREAMING_SNAKE_CASE_	=      AutoConfig.for_model(_A ,     **_A	)
          SCREAMING_SNAKE_CASE_	=      True



       @classmethod
       def     _UpperCamelCase	(							cls ,     _A ,     _A ,     **_A	)				->				PretrainedConfig:
          logger.info('''Setting `config.is_decoder=True` and `config.add_cross_attention=True` for decoder_config'''	)
          SCREAMING_SNAKE_CASE_	=      True
          SCREAMING_SNAKE_CASE_	=      True

          return cls(encoder=encoder_config.to_dict() ,     decoder=decoder_config.to_dict() ,     **_A	)




       def     _UpperCamelCase	(							self	)				->				Union[str, Any]:
          SCREAMING_SNAKE_CASE_	=      copy.deepcopy(self.__dict__	)
          SCREAMING_SNAKE_CASE_	=      self.encoder.to_dict()
          SCREAMING_SNAKE_CASE_	=      self.decoder.to_dict()
          SCREAMING_SNAKE_CASE_	=      self.__class__.model_type
          return output


