




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 					SCREAMING_SNAKE_CASE	(       __UpperCamelCase			:						str      )  ->							List[Any]:
    if isinstance(__UpperCamelCase       ,     torch.Tensor      ):
        return image
    elif isinstance(__UpperCamelCase       ,     PIL.Image.Image      ):
        UpperCAmelCase_							 =     [image]

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

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

    return image



class 			a     (   _A			):

     '''simple docstring'''





     def __init__(	self   :			Dict					,    __snake_case   :			OnnxRuntimeModel					,    __snake_case   :			OnnxRuntimeModel					,    __snake_case   :			Any					,    __snake_case   :			OnnxRuntimeModel					,    __snake_case   :			DDPMScheduler					,    __snake_case   :			Any					,    __snake_case   :			int = 3_50					,    ):
         super().__init__(
             vae=__snake_case					,    text_encoder=__snake_case					,    tokenizer=__snake_case					,    unet=__snake_case					,    low_res_scheduler=__snake_case					,    scheduler=__snake_case					,    safety_checker=__snake_case					,    feature_extractor=__snake_case					,    watermarker=__snake_case					,    max_noise_level=__snake_case					,    )
     def __call__(	self   :			Optional[int]					,    __snake_case   :			Union[str, List[str]]					,    __snake_case   :			Union[torch.FloatTensor, PIL.Image.Image, List[PIL.Image.Image]]					,    __snake_case   :			int = 75					,    __snake_case   :			float = 9.0					,    __snake_case   :			int = 20					,    __snake_case   :			Optional[Union[str, List[str]]] = None					,    __snake_case   :			Optional[int] = 1					,    __snake_case   :			float = 0.0					,    __snake_case   :			Optional[Union[torch.Generator, List[torch.Generator]]] = None					,    __snake_case   :			Optional[torch.FloatTensor] = None					,    __snake_case   :			Optional[np.ndarray] = None					,    __snake_case   :			Optional[np.ndarray] = None					,    __snake_case   :			Optional[str] = "pil"					,    __snake_case   :			bool = True					,    __snake_case   :			Optional[Callable[[int, int, torch.FloatTensor], None]] = None					,    __snake_case   :			Optional[int] = 1					,    ):
         self.check_inputs(__snake_case					,    __snake_case					,    __snake_case					,    __snake_case				)

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

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

         # 3. Encode input prompt
         UpperCAmelCase_							 =     self._encode_prompt(
             __snake_case					,    __snake_case					,    __snake_case					,    __snake_case					,    __snake_case					,    prompt_embeds=__snake_case					,    negative_prompt_embeds=__snake_case					,    )

         UpperCAmelCase_							 =     ORT_TO_PT_TYPE[str(text_embeddings.dtype				)]

         # 4. Preprocess image
         UpperCAmelCase_							 =     preprocess(__snake_case				)
         UpperCAmelCase_							 =     image.cpu()

         # 5. set timesteps
         self.scheduler.set_timesteps(__snake_case					,    device=__snake_case				)
         UpperCAmelCase_							 =     self.scheduler.timesteps

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

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

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

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

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

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

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

                 # timestep to tensor
                 UpperCAmelCase_							 =     np.array([t]					,    dtype=__snake_case				)

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

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

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

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

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

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

         if not return_dict:
             return (image,)

         return ImagePipelineOutput(images=__snake_case				)
     def   lowerCamelCase_   (	self   :			Any					,    __snake_case   :			str				):
         UpperCAmelCase_							 =     1 / 0.08_333 * latents
         UpperCAmelCase_							 =     self.vae(latent_sample=__snake_case				)[0]
         UpperCAmelCase_							 =     np.clip(image / 2 + 0.5					,    0					,    1				)
         UpperCAmelCase_							 =     image.transpose((0, 2, 3, 1)				)
         return image




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

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

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

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

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

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

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

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

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

         return prompt_embeds






'''simple docstring'''


from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tokenizers_available, is_torch_available


snake_case__     :			int      			=  {
    '''configuration_markuplm''': ['''MARKUPLM_PRETRAINED_CONFIG_ARCHIVE_MAP''', '''MarkupLMConfig'''],
    '''feature_extraction_markuplm''': ['''MarkupLMFeatureExtractor'''],
    '''processing_markuplm''': ['''MarkupLMProcessor'''],
    '''tokenization_markuplm''': ['''MarkupLMTokenizer'''],
}

try:
 if not is_tokenizers_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 snake_case__     :			List[str]      			=  ['''MarkupLMTokenizerFast''']

try:
 if not is_torch_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 snake_case__     :			Dict      			=  [
     '''MARKUPLM_PRETRAINED_MODEL_ARCHIVE_LIST''',
     '''MarkupLMForQuestionAnswering''',
     '''MarkupLMForSequenceClassification''',
     '''MarkupLMForTokenClassification''',
     '''MarkupLMModel''',
     '''MarkupLMPreTrainedModel''',
 ]


if TYPE_CHECKING:
 from .configuration_markuplm import MARKUPLM_PRETRAINED_CONFIG_ARCHIVE_MAP, MarkupLMConfig
 from .feature_extraction_markuplm import MarkupLMFeatureExtractor
 from .processing_markuplm import MarkupLMProcessor
 from .tokenization_markuplm import MarkupLMTokenizer

 try:
  if not is_tokenizers_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .tokenization_markuplm_fast import MarkupLMTokenizerFast

 try:
  if not is_torch_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .modeling_markuplm import (
      MARKUPLM_PRETRAINED_MODEL_ARCHIVE_LIST,
      MarkupLMForQuestionAnswering,
      MarkupLMForSequenceClassification,
      MarkupLMForTokenClassification,
      MarkupLMModel,
      MarkupLMPreTrainedModel,
  )


else:
 import sys

 snake_case__     :			Optional[Any]      			=  _LazyModule(__name__, globals()['''__file__'''], _import_structure)






