


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_   (				__snake_case		)	->     List[str]:







 """simple docstring"""


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

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

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

 return image







class  lowerCamelCase__  (			lowerCAmelCase):





     def __init__(self						,	UpperCAmelCase						,	UpperCAmelCase						,	UpperCAmelCase						,	UpperCAmelCase						,	UpperCAmelCase						,	UpperCAmelCase						,	UpperCAmelCase = 3_5_0						,	)	->  List[Any]:
      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						,	UpperCAmelCase						,	UpperCAmelCase						,	UpperCAmelCase = 7_5						,	UpperCAmelCase = 9.0						,	UpperCAmelCase = 2_0						,	UpperCAmelCase = None						,	UpperCAmelCase = 1						,	UpperCAmelCase = 0.0						,	UpperCAmelCase = None						,	UpperCAmelCase = None						,	UpperCAmelCase = None						,	UpperCAmelCase = None						,	UpperCAmelCase = "pil"						,	UpperCAmelCase = True						,	UpperCAmelCase = None						,	UpperCAmelCase = 1						,	)	->  Union[str, Any]:
      self.check_inputs(UpperCAmelCase						,	UpperCAmelCase						,	UpperCAmelCase						,	UpperCAmelCase			)

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

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

      # 3. Encode input prompt
      _lowercase				 =self._encode_prompt(
          UpperCAmelCase						,	UpperCAmelCase						,	UpperCAmelCase						,	UpperCAmelCase						,	UpperCAmelCase						,	prompt_embeds=UpperCAmelCase						,	negative_prompt_embeds=UpperCAmelCase						,	)

      _lowercase				 =ORT_TO_PT_TYPE[str(text_embeddings.dtype			)]

      # 4. Preprocess image
      _lowercase				 =preprocess(UpperCAmelCase			)
      _lowercase				 =image.cpu()

      # 5. set timesteps
      self.scheduler.set_timesteps(UpperCAmelCase						,	device=UpperCAmelCase			)
      _lowercase				 =self.scheduler.timesteps

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

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

      # 6. Prepare latent variables
      _lowercase ,       _lowercase				 =image.shape[2:]
      _lowercase				 =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
      _lowercase				 =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
      _lowercase				 =self.prepare_extra_step_kwargs(UpperCAmelCase						,	UpperCAmelCase			)

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

      # 9. Denoising loop
      _lowercase				 =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
        _lowercase				 =np.concatenate([latents] * 2			) if do_classifier_free_guidance else latents

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

        # timestep to tensor
        _lowercase				 =np.array([t]						,	dtype=UpperCAmelCase			)

        # predict the noise residual
        _lowercase				 =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:
         _lowercase ,       _lowercase				 =np.split(UpperCAmelCase						,	2			)
        _lowercase				 =noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

        # compute the previous noisy sample x_t -> x_t-1
        _lowercase				 =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
      _lowercase				 =self.decode_latents(latents.float()			)

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

      if not return_dict:
       return (image,)

      return ImagePipelineOutput(images=UpperCAmelCase			)





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





     def 	__A   (self						,	UpperCAmelCase						,	UpperCAmelCase						,	UpperCAmelCase						,	UpperCAmelCase						,	UpperCAmelCase						,	UpperCAmelCase = None						,	UpperCAmelCase = None						,	)	->  List[Any]:
      if prompt is not None and isinstance(UpperCAmelCase						,	UpperCAmelCase			):
       _lowercase				 =1
      elif prompt is not None and isinstance(UpperCAmelCase						,	UpperCAmelCase			):
       _lowercase				 =len(UpperCAmelCase			)
      else:
       _lowercase				 =prompt_embeds.shape[0]

      if prompt_embeds is None:
       _lowercase				 =self.tokenizer(
           UpperCAmelCase						,	padding='''max_length'''						,	max_length=self.tokenizer.model_max_length						,	truncation=UpperCAmelCase						,	return_tensors='''pt'''						,	)
       _lowercase				 =text_inputs.input_ids
       _lowercase				 =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			):
        _lowercase				 =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
       _lowercase				 =self.text_encoder(
           input_ids=text_input_ids.int().to(UpperCAmelCase			)						,	)
       _lowercase				 =prompt_embeds[0]

      _lowercase ,       _lowercase ,       _lowercase				 =prompt_embeds.shape
      # duplicate text embeddings for each generation per prompt, using mps friendly method
      _lowercase				 =prompt_embeds.repeat(1						,	UpperCAmelCase			)
      _lowercase				 =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:
       _lowercase				 =42
       if negative_prompt is None:
        _lowercase				 =[''''''] * 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			):
        _lowercase				 =[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:
        _lowercase				 =negative_prompt

       _lowercase				 =text_input_ids.shape[-1]
       _lowercase				 =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

       _lowercase				 =self.text_encoder(
           input_ids=uncond_input.input_ids.int().to(UpperCAmelCase			)						,	)
       _lowercase				 =uncond_embeddings[0]

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

      return prompt_embeds





import json
from typing import List, Optional, Tuple

from tokenizers import normalizers

from ...tokenization_utils import AddedToken
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import logging
from .tokenization_mpnet import MPNetTokenizer


UpperCAmelCase__        =				logging.get_logger(__name__)

UpperCAmelCase__        =				{'''vocab_file''': '''vocab.txt''', '''tokenizer_file''': '''tokenizer.json'''}

UpperCAmelCase__        =				{
    '''vocab_file''': {
        '''microsoft/mpnet-base''': '''https://huggingface.co/microsoft/mpnet-base/resolve/main/vocab.txt''',
    },
    '''tokenizer_file''': {
        '''microsoft/mpnet-base''': '''https://huggingface.co/microsoft/mpnet-base/resolve/main/tokenizer.json''',
    },
}

UpperCAmelCase__        =				{
    '''microsoft/mpnet-base''': 512,
}

UpperCAmelCase__        =				{
    '''microsoft/mpnet-base''': {'''do_lower_case''': True},
}







class  lowerCamelCase__  (			lowerCAmelCase):
     SCREAMING_SNAKE_CASE__      =     VOCAB_FILES_NAMES
     SCREAMING_SNAKE_CASE__      =     PRETRAINED_VOCAB_FILES_MAP
     SCREAMING_SNAKE_CASE__      =     PRETRAINED_INIT_CONFIGURATION
     SCREAMING_SNAKE_CASE__      =     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     SCREAMING_SNAKE_CASE__      =     MPNetTokenizer
     SCREAMING_SNAKE_CASE__      =     ['''input_ids''', '''attention_mask''']





     def __init__(self						,	UpperCAmelCase=None						,	UpperCAmelCase=None						,	UpperCAmelCase=True						,	UpperCAmelCase="<s>"						,	UpperCAmelCase="</s>"						,	UpperCAmelCase="</s>"						,	UpperCAmelCase="<s>"						,	UpperCAmelCase="[UNK]"						,	UpperCAmelCase="<pad>"						,	UpperCAmelCase="<mask>"						,	UpperCAmelCase=True						,	UpperCAmelCase=None						,	**UpperCAmelCase						,	)	->  Optional[int]:
      super().__init__(
          UpperCAmelCase						,	tokenizer_file=UpperCAmelCase						,	do_lower_case=UpperCAmelCase						,	bos_token=UpperCAmelCase						,	eos_token=UpperCAmelCase						,	sep_token=UpperCAmelCase						,	cls_token=UpperCAmelCase						,	unk_token=UpperCAmelCase						,	pad_token=UpperCAmelCase						,	mask_token=UpperCAmelCase						,	tokenize_chinese_chars=UpperCAmelCase						,	strip_accents=UpperCAmelCase						,	**UpperCAmelCase						,	)

      _lowercase				 =json.loads(self.backend_tokenizer.normalizer.__getstate__()			)
      if (
          pre_tok_state.get('''lowercase'''						,	UpperCAmelCase			) != do_lower_case
          or pre_tok_state.get('''strip_accents'''						,	UpperCAmelCase			) != strip_accents
      ):
       _lowercase				 =getattr(UpperCAmelCase						,	pre_tok_state.pop('''type'''			)			)
       _lowercase				 =do_lower_case
       _lowercase				 =strip_accents
       _lowercase				 =pre_tok_class(**UpperCAmelCase			)

      _lowercase				 =do_lower_case





     @property
     def 	__A   (self			)	->  str:
      if self._mask_token is None:
       if self.verbose:
        logger.error('''Using mask_token, but it is not set yet.'''			)
       return None
      return str(self._mask_token			)





     @mask_token.setter
     def 	__A   (self						,	UpperCAmelCase			)	->  Optional[int]:
      _lowercase				 =AddedToken(UpperCAmelCase						,	lstrip=UpperCAmelCase						,	rstrip=UpperCAmelCase			) if isinstance(UpperCAmelCase						,	UpperCAmelCase			) else value
      _lowercase				 =value





     def 	__A   (self						,	UpperCAmelCase						,	UpperCAmelCase=None			)	->  List[str]:
      _lowercase				 =[self.bos_token_id] + token_ids_a + [self.eos_token_id]
      if token_ids_a is None:
       return output

      return output + [self.eos_token_id] + token_ids_a + [self.eos_token_id]





     def 	__A   (self						,	UpperCAmelCase						,	UpperCAmelCase = None			)	->  List[int]:
      _lowercase				 =[self.sep_token_id]
      _lowercase				 =[self.cls_token_id]

      if token_ids_a is None:
       return len(cls + token_ids_a + sep			) * [0]
      return len(cls + token_ids_a + sep + sep + token_ids_a + sep			) * [0]





     def 	__A   (self						,	UpperCAmelCase						,	UpperCAmelCase = None			)	->  Tuple[str]:
      _lowercase				 =self._tokenizer.model.save(UpperCAmelCase						,	name=UpperCAmelCase			)
      return tuple(UpperCAmelCase			)

