

from __future__ import annotations

from collections.abc import Iterator
from typing import Any



class        _UpperCAmelCase :
  '''simple docstring'''







  def __init__( self	: int		,						UpperCamelCase__	: Any ):
        A    =				data
        A    =				None







  def __str__( self	: str ):
        return f'''{self.data}'''



class        _UpperCAmelCase :
  '''simple docstring'''







  def __init__( self	: str ):
        A    =				None
        A    =				None







  def __iter__( self	: Any ):
        A    =				self.front
        while node:
              yield node.data
              A    =				node.next







  def __len__( self	: Optional[int] ):
        return len(tuple(iter(self ) ) )







  def __str__( self	: Optional[int] ):
        return " <- ".join(str(UpperCamelCase__ ) for item in self )







  def 	UpperCamelCase       ( self	: Dict ):
        return len(self ) == 0







  def 	UpperCamelCase       ( self	: Any		,						UpperCamelCase__	: Any ):
        A    =				Node(UpperCamelCase__ )
        if self.is_empty():
              A    =				A    =				node
        else:
              assert isinstance(self.rear		,						UpperCamelCase__ )
              A    =				node
              A    =				node







  def 	UpperCamelCase       ( self	: Dict ):
        if self.is_empty():
              raise IndexError('dequeue from empty queue' )
        assert isinstance(self.front		,						UpperCamelCase__ )
        A    =				self.front
        A    =				self.front.next
        if self.front is None:
              A    =				None
        return node.data







  def 	UpperCamelCase       ( self	: Dict ):
        A    =				A    =				None


if __name__ == "__main__":
 from doctest import testmod

 testmod()






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 			__UpperCamelCase    (lowerCAmelCase					:  Any     )     ->  Dict:
      if isinstance(lowerCAmelCase,    torch.Tensor     ):
            return image
      elif isinstance(lowerCAmelCase,    PIL.Image.Image     ):
            A    =				[image]

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

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

      return image







class        _UpperCAmelCase (   __lowercase			):
  '''simple docstring'''







  def __init__( self	: List[Any]		,						UpperCamelCase__	: OnnxRuntimeModel		,						UpperCamelCase__	: OnnxRuntimeModel		,						UpperCamelCase__	: Any		,						UpperCamelCase__	: OnnxRuntimeModel		,						UpperCamelCase__	: DDPMScheduler		,						UpperCamelCase__	: Any		,						UpperCamelCase__	: int = 350		,						):
        super().__init__(
            vae=UpperCamelCase__		,						text_encoder=UpperCamelCase__		,						tokenizer=UpperCamelCase__		,						unet=UpperCamelCase__		,						low_res_scheduler=UpperCamelCase__		,						scheduler=UpperCamelCase__		,						safety_checker=UpperCamelCase__		,						feature_extractor=UpperCamelCase__		,						watermarker=UpperCamelCase__		,						max_noise_level=UpperCamelCase__		,						)







  def __call__( self	: int		,						UpperCamelCase__	: Union[str, List[str]]		,						UpperCamelCase__	: Union[torch.FloatTensor, PIL.Image.Image, List[PIL.Image.Image]]		,						UpperCamelCase__	: int = 75		,						UpperCamelCase__	: float = 9.0		,						UpperCamelCase__	: int = 20		,						UpperCamelCase__	: Optional[Union[str, List[str]]] = None		,						UpperCamelCase__	: Optional[int] = 1		,						UpperCamelCase__	: float = 0.0		,						UpperCamelCase__	: Optional[Union[torch.Generator, List[torch.Generator]]] = None		,						UpperCamelCase__	: Optional[torch.FloatTensor] = None		,						UpperCamelCase__	: Optional[np.ndarray] = None		,						UpperCamelCase__	: Optional[np.ndarray] = None		,						UpperCamelCase__	: Optional[str] = "pil"		,						UpperCamelCase__	: bool = True		,						UpperCamelCase__	: Optional[Callable[[int, int, torch.FloatTensor], None]] = None		,						UpperCamelCase__	: Optional[int] = 1		,						):
        self.check_inputs(UpperCamelCase__		,						UpperCamelCase__		,						UpperCamelCase__		,						UpperCamelCase__ )

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

        A    =				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    =				guidance_scale > 1.0

        # 3. Encode input prompt
        A    =				self._encode_prompt(
            UpperCamelCase__		,						UpperCamelCase__		,						UpperCamelCase__		,						UpperCamelCase__		,						UpperCamelCase__		,						prompt_embeds=UpperCamelCase__		,						negative_prompt_embeds=UpperCamelCase__		,						)

        A    =				ORT_TO_PT_TYPE[str(text_embeddings.dtype )]

        # 4. Preprocess image
        A    =				preprocess(UpperCamelCase__ )
        A    =				image.cpu()

        # 5. set timesteps
        self.scheduler.set_timesteps(UpperCamelCase__		,						device=UpperCamelCase__ )
        A    =				self.scheduler.timesteps

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

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

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

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

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

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

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

                    # timestep to tensor
                    A    =				np.array([t]		,						dtype=UpperCamelCase__ )

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

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

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

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

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

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

        if not return_dict:
              return (image,)

        return ImagePipelineOutput(images=UpperCamelCase__ )







  def 	UpperCamelCase       ( self	: Optional[int]		,						UpperCamelCase__	: Dict ):
        A    =				1 / 0.08_333 * latents
        A    =				self.vae(latent_sample=UpperCamelCase__ )[0]
        A    =				np.clip(image / 2 + 0.5		,						0		,						1 )
        A    =				image.transpose((0, 2, 3, 1) )
        return image







  def 	UpperCamelCase       ( self	: Tuple		,						UpperCamelCase__	: Union[str, List[str]]		,						UpperCamelCase__	: List[Any]		,						UpperCamelCase__	: Optional[int]		,						UpperCamelCase__	: bool		,						UpperCamelCase__	: Optional[str]		,						UpperCamelCase__	: Optional[torch.FloatTensor] = None		,						UpperCamelCase__	: Optional[torch.FloatTensor] = None		,						):
        if prompt is not None and isinstance(UpperCamelCase__		,						UpperCamelCase__ ):
              A    =				1
        elif prompt is not None and isinstance(UpperCamelCase__		,						UpperCamelCase__ ):
              A    =				len(UpperCamelCase__ )
        else:
              A    =				prompt_embeds.shape[0]

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

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

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

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

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

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

              A    =				self.text_encoder(
                  input_ids=uncond_input.input_ids.int().to(UpperCamelCase__ )		,						)
              A    =				uncond_embeddings[0]

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

        return prompt_embeds



