





"""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__					:     Union[str, Any]							=				getLogger(__name__)


a__					:     Union[str, Any]							=				4
a__					:     str							=				7

a__					:     Optional[int]							=				{
    """float16""": torch.floataa,
    """float32""": torch.floataa,
}





def 							a__			(  _UpperCAmelCase	)    ->  int:

 '''simple docstring'''






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

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

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

 return image



class      lowercase_				(     __lowerCAmelCase							):

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






        '''simple docstring'''

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

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






        '''simple docstring'''

        self.check_inputs(__SCREAMING_SNAKE_CASE   ,					__SCREAMING_SNAKE_CASE   ,					__SCREAMING_SNAKE_CASE   ,					__SCREAMING_SNAKE_CASE     )

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

        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.
        snake_case__    							=				guidance_scale > 1.0

        # 3. Encode input prompt
        snake_case__    							=				self._encode_prompt(
            __SCREAMING_SNAKE_CASE   ,					__SCREAMING_SNAKE_CASE   ,					__SCREAMING_SNAKE_CASE   ,					__SCREAMING_SNAKE_CASE   ,					__SCREAMING_SNAKE_CASE   ,					prompt_embeds=__SCREAMING_SNAKE_CASE   ,					negative_prompt_embeds=__SCREAMING_SNAKE_CASE   ,					)

        snake_case__    							=				ORT_TO_PT_TYPE[str(text_embeddings.dtype     )]

        # 4. Preprocess image
        snake_case__    							=				preprocess(__SCREAMING_SNAKE_CASE     )
        snake_case__    							=				image.cpu()

        # 5. set timesteps
        self.scheduler.set_timesteps(__SCREAMING_SNAKE_CASE   ,					device=__SCREAMING_SNAKE_CASE     )
        snake_case__    							=				self.scheduler.timesteps

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

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

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

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

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

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

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

          # timestep to tensor
          snake_case__    							=				np.array([t]   ,					dtype=__SCREAMING_SNAKE_CASE     )

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

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

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

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

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

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

        if not return_dict:
         return (image,)

        return ImagePipelineOutput(images=__SCREAMING_SNAKE_CASE     )

       def 		_SCREAMING_SNAKE_CASE(			self   ,					__SCREAMING_SNAKE_CASE     ):






        '''simple docstring'''

        snake_case__    							=				1 / 0.08333 * latents
        snake_case__    							=				self.vae(latent_sample=__SCREAMING_SNAKE_CASE     )[0]
        snake_case__    							=				np.clip(image / 2 + 0.5   ,					0   ,					1     )
        snake_case__    							=				image.transpose((0, 2, 3, 1)     )
        return image

       def 		_SCREAMING_SNAKE_CASE(			self   ,					__SCREAMING_SNAKE_CASE   ,					__SCREAMING_SNAKE_CASE   ,					__SCREAMING_SNAKE_CASE   ,					__SCREAMING_SNAKE_CASE   ,					__SCREAMING_SNAKE_CASE   ,					__SCREAMING_SNAKE_CASE = None   ,					__SCREAMING_SNAKE_CASE = None   ,					):






        '''simple docstring'''

        if prompt is not None and isinstance(__SCREAMING_SNAKE_CASE   ,					__SCREAMING_SNAKE_CASE     ):
         snake_case__    							=				1
        elif prompt is not None and isinstance(__SCREAMING_SNAKE_CASE   ,					__SCREAMING_SNAKE_CASE     ):
         snake_case__    							=				len(__SCREAMING_SNAKE_CASE     )
        else:
         snake_case__    							=				prompt_embeds.shape[0]

        if prompt_embeds is None:
         snake_case__    							=				self.tokenizer(
             __SCREAMING_SNAKE_CASE   ,					padding="max_length"   ,					max_length=self.tokenizer.model_max_length   ,					truncation=__SCREAMING_SNAKE_CASE   ,					return_tensors="pt"   ,					)
         snake_case__    							=				text_inputs.input_ids
         snake_case__    							=				self.tokenizer(__SCREAMING_SNAKE_CASE   ,					padding="longest"   ,					return_tensors="pt"     ).input_ids

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

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

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

         snake_case__    							=				text_input_ids.shape[-1]
         snake_case__    							=				self.tokenizer(
             __SCREAMING_SNAKE_CASE   ,					padding="max_length"   ,					max_length=__SCREAMING_SNAKE_CASE   ,					truncation=__SCREAMING_SNAKE_CASE   ,					return_tensors="pt"   ,					)

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

         snake_case__    							=				self.text_encoder(
             input_ids=uncond_input.input_ids.int().to(__SCREAMING_SNAKE_CASE     )   ,					)
         snake_case__    							=				uncond_embeddings[0]

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

        return prompt_embeds








"""simple docstring"""


def 							a__			(  _UpperCAmelCase	)    ->  int:

 '''simple docstring'''






 return int(_UpperCAmelCase	) if x - int(_UpperCAmelCase	) <= 0 else int(_UpperCAmelCase	) + 1


if __name__ == "__main__":
   import doctest

   doctest.testmod()

