



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  _snake_case			(							__snake_case     )  ->  Union[str, Any]:






      '''simple docstring'''







      if isinstance(__snake_case       ,	torch.Tensor     ):
            return image
      elif isinstance(__snake_case       ,	PIL.Image.Image     ):
            UpperCAmelCase_	:						List[Any]					   =		[image]

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

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

      return image





class 	snake_case_			(lowercase__	):




       """simple docstring"""






       def __init__(  self  ,lowercase  ,lowercase  ,lowercase  ,lowercase  ,lowercase  ,lowercase  ,lowercase = 350  ,):

             """simple docstring"""





             super().__init__(
                 vae=lowercase  ,text_encoder=lowercase  ,tokenizer=lowercase  ,unet=lowercase  ,low_res_scheduler=lowercase  ,scheduler=lowercase  ,safety_checker=lowercase  ,feature_extractor=lowercase  ,watermarker=lowercase  ,max_noise_level=lowercase  ,)



       def __call__(  self  ,lowercase  ,lowercase  ,lowercase = 75  ,lowercase = 9.0  ,lowercase = 20  ,lowercase = None  ,lowercase = 1  ,lowercase = 0.0  ,lowercase = None  ,lowercase = None  ,lowercase = None  ,lowercase = None  ,lowercase = "pil"  ,lowercase = True  ,lowercase = None  ,lowercase = 1  ,):

             """simple docstring"""





             self.check_inputs(lowercase  ,lowercase  ,lowercase  ,lowercase)

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

             UpperCAmelCase_	:						Union[str, 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.
             UpperCAmelCase_	:						int					   =		guidance_scale > 1.0

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

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

             # 4. Preprocess image
             UpperCAmelCase_	:						Any					   =		preprocess(lowercase)
             UpperCAmelCase_	:						int					   =		image.cpu()

             # 5. set timesteps
             self.scheduler.set_timesteps(lowercase  ,device=lowercase)
             UpperCAmelCase_	:						Tuple					   =		self.scheduler.timesteps

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

             UpperCAmelCase_	:						Optional[Any]					   =		2 if do_classifier_free_guidance else 1
             UpperCAmelCase_	:						Union[str, Any]					   =		np.concatenate([image] * batch_multiplier * num_images_per_prompt)
             UpperCAmelCase_	:						Tuple					   =		np.concatenate([noise_level] * image.shape[0])

             # 6. Prepare latent variables
             UpperCAmelCase_   ,   UpperCAmelCase_	:						str					   =		image.shape[2:]
             UpperCAmelCase_	:						Optional[int]					   =		self.prepare_latents(
                 batch_size * num_images_per_prompt  ,lowercase  ,lowercase  ,lowercase  ,lowercase  ,lowercase  ,lowercase  ,lowercase  ,)

             # 7. Check that sizes of image and latents match
             UpperCAmelCase_	:						Optional[int]					   =		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_	:						int					   =		self.prepare_extra_step_kwargs(lowercase  ,lowercase)

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

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

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

                         # timestep to tensor
                         UpperCAmelCase_	:						List[str]					   =		np.array([t]  ,dtype=lowercase)

                         # predict the noise residual
                         UpperCAmelCase_	:						List[Any]					   =		self.unet(
                             sample=lowercase  ,timestep=lowercase  ,encoder_hidden_states=lowercase  ,class_labels=noise_level.astype(np.intaa)  ,)[0]

                         # perform guidance
                         if do_classifier_free_guidance:
                               UpperCAmelCase_   ,   UpperCAmelCase_	:						int					   =		np.split(lowercase  ,2)
                         UpperCAmelCase_	:						List[str]					   =		noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

                         # compute the previous noisy sample x_t -> x_t-1
                         UpperCAmelCase_	:						Optional[Any]					   =		self.scheduler.step(
                             torch.from_numpy(lowercase)  ,lowercase  ,lowercase  ,**lowercase).prev_sample

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

        # 10. Post-processing
             UpperCAmelCase_	:						Tuple					   =		self.decode_latents(latents.float())

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

             if not return_dict:
                   return (image,)

             return ImagePipelineOutput(images=lowercase)



       def 			A_ (  self  ,lowercase):

             """simple docstring"""





             UpperCAmelCase_	:						Union[str, Any]					   =		1 / 0.0_8333 * latents
             UpperCAmelCase_	:						List[str]					   =		self.vae(latent_sample=lowercase)[0]
             UpperCAmelCase_	:						Union[str, Any]					   =		np.clip(image / 2 + 0.5  ,0  ,1)
             UpperCAmelCase_	:						Optional[Any]					   =		image.transpose((0, 2, 3, 1))
             return image



       def 			A_ (  self  ,lowercase  ,lowercase  ,lowercase  ,lowercase  ,lowercase  ,lowercase = None  ,lowercase = None  ,):

             """simple docstring"""





             if prompt is not None and isinstance(lowercase  ,lowercase):
                   UpperCAmelCase_	:						Any					   =		1
             elif prompt is not None and isinstance(lowercase  ,lowercase):
                   UpperCAmelCase_	:						Tuple					   =		len(lowercase)
             else:
                   UpperCAmelCase_	:						Tuple					   =		prompt_embeds.shape[0]

             if prompt_embeds is None:
                   UpperCAmelCase_	:						Dict					   =		self.tokenizer(
                       lowercase  ,padding="max_length"  ,max_length=self.tokenizer.model_max_length  ,truncation=lowercase  ,return_tensors="pt"  ,)
                   UpperCAmelCase_	:						Any					   =		text_inputs.input_ids
                   UpperCAmelCase_	:						Union[str, Any]					   =		self.tokenizer(lowercase  ,padding="longest"  ,return_tensors="pt").input_ids

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

             UpperCAmelCase_   ,   UpperCAmelCase_   ,   UpperCAmelCase_	:						str					   =		prompt_embeds.shape
             # duplicate text embeddings for each generation per prompt, using mps friendly method
             UpperCAmelCase_	:						Optional[Any]					   =		prompt_embeds.repeat(1  ,lowercase)
             UpperCAmelCase_	:						int					   =		prompt_embeds.reshape(bs_embed * num_images_per_prompt  ,lowercase  ,-1)

             # get unconditional embeddings for classifier free guidance
             if do_classifier_free_guidance and negative_prompt_embeds is None:
                   UpperCAmelCase_	:						List[str]
                   if negative_prompt is None:
                         UpperCAmelCase_	:						Union[str, Any]					   =		[""] * batch_size
                   elif type(lowercase) is not type(lowercase):
                         raise TypeError(
                             F"""`negative_prompt` should be the same type to `prompt`, but got {type(lowercase)} !="""
                             F""" {type(lowercase)}.""")
                   elif isinstance(lowercase  ,lowercase):
                         UpperCAmelCase_	:						int					   =		[negative_prompt]
                   elif batch_size != len(lowercase):
                         raise ValueError(
                             F"""`negative_prompt`: {negative_prompt} has batch size {len(lowercase)}, but `prompt`:"""
                             F""" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"""
                             " the batch size of `prompt`.")
                   else:
                         UpperCAmelCase_	:						Dict					   =		negative_prompt

                   UpperCAmelCase_	:						List[Any]					   =		text_input_ids.shape[-1]
                   UpperCAmelCase_	:						Dict					   =		self.tokenizer(
                       lowercase  ,padding="max_length"  ,max_length=lowercase  ,truncation=lowercase  ,return_tensors="pt"  ,)

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

                   UpperCAmelCase_	:						str					   =		self.text_encoder(
                       input_ids=uncond_input.input_ids.int().to(lowercase)  ,)
                   UpperCAmelCase_	:						int					   =		uncond_embeddings[0]

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

             return prompt_embeds





import os
import unittest

import pytest
from parameterized import parameterized

from transformers.testing_utils import require_torch, require_torch_gpu, slow, torch_device


# skipping in unittest tests

__lowerCamelCase								=      [(1,)]







def  _snake_case			(							)  ->  str:






      '''simple docstring'''







      UpperCAmelCase_	:						Optional[Any]					   =		bool(os.getenv("RUN_SLOW"       ,	0     )     )
      if run_slow:
            assert True
      else:
            assert False, "should have been skipped"







def  _snake_case			(							)  ->  Tuple:






      '''simple docstring'''







      UpperCAmelCase_	:						Union[str, Any]					   =		bool(os.getenv("RUN_SLOW"       ,	0     )     )
      if run_slow and torch_device == "cuda":
            assert True
      else:
            assert False, "should have been skipped"


@require_torch
class 	snake_case_			(unittest.TestCase	):




       """simple docstring"""






       @slow
       @require_torch_gpu
       def 			A_ (  self):

             """simple docstring"""





             check_slow_torch_cuda()



       @require_torch_gpu
       @slow
       def 			A_ (  self):

             """simple docstring"""





             check_slow_torch_cuda()



       @parameterized.expand(lowercase)
       @slow
       def 			A_ (  self  ,lowercase=None):

             """simple docstring"""





             check_slow()







@slow
@require_torch_gpu
def  _snake_case			(							)  ->  str:






      '''simple docstring'''







      check_slow_torch_cuda()







@require_torch_gpu
@slow
def  _snake_case			(							)  ->  Union[str, Any]:






      '''simple docstring'''







      check_slow_torch_cuda()







@slow
@pytest.mark.parametrize("param"       ,	[1]     )
def  _snake_case			(							__snake_case     )  ->  Optional[Any]:






      '''simple docstring'''







      check_slow()







@pytest.mark.parametrize("param"       ,	[1]     )
@slow
def  _snake_case			(							__snake_case     )  ->  Any:






      '''simple docstring'''







      check_slow()
