"""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


__UpperCamelCase								=    getLogger(__name__)


__UpperCamelCase								=    4
__UpperCamelCase								=    7

__UpperCamelCase								=    {
    "float16": torch.floataa,
    "float32": torch.floataa,
}







def       UpperCAmelCase					(	UpperCAmelCase      )       ->   Optional[int]:







     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 % 64 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 				UpperCamelCase      (       SCREAMING_SNAKE_CASE__     ):







       def __init__(				self,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__ = 350,							)   ->   int:
            super().__init__(
                vae=lowerCAmelCase__,							text_encoder=lowerCAmelCase__,							tokenizer=lowerCAmelCase__,							unet=lowerCAmelCase__,							low_res_scheduler=lowerCAmelCase__,							scheduler=lowerCAmelCase__,							safety_checker=lowerCAmelCase__,							feature_extractor=lowerCAmelCase__,							watermarker=lowerCAmelCase__,							max_noise_level=lowerCAmelCase__,							)







       def __call__(				self,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__ = 75,							lowerCAmelCase__ = 9.0,							lowerCAmelCase__ = 20,							lowerCAmelCase__ = None,							lowerCAmelCase__ = 1,							lowerCAmelCase__ = 0.0,							lowerCAmelCase__ = None,							lowerCAmelCase__ = None,							lowerCAmelCase__ = None,							lowerCAmelCase__ = None,							lowerCAmelCase__ = "pil",							lowerCAmelCase__ = True,							lowerCAmelCase__ = None,							lowerCAmelCase__ = 1,							)   ->   List[Any]:
            self.check_inputs(lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__)

            # 2. Define call parameters
            if prompt is not None and isinstance(lowerCAmelCase__,							lowerCAmelCase__):
                 snake_case_       =      1
            elif prompt is not None and isinstance(lowerCAmelCase__,							lowerCAmelCase__):
                 snake_case_       =      len(lowerCAmelCase__)
            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(
                lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							prompt_embeds=lowerCAmelCase__,							negative_prompt_embeds=lowerCAmelCase__,							)

            snake_case_       =      ORT_TO_PT_TYPE[str(text_embeddings.dtype)]

            # 4. Preprocess image
            snake_case_       =      preprocess(lowerCAmelCase__)
            snake_case_       =      image.cpu()

            # 5. set timesteps
            self.scheduler.set_timesteps(lowerCAmelCase__,							device=lowerCAmelCase__)
            snake_case_       =      self.scheduler.timesteps

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

            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,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							)

            # 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(lowerCAmelCase__,							lowerCAmelCase__)

            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(lowerCAmelCase__) - num_inference_steps * self.scheduler.order
            with self.progress_bar(total=lowerCAmelCase__) as progress_bar:
                 for i, t in enumerate(lowerCAmelCase__):
                      # 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(lowerCAmelCase__,							lowerCAmelCase__)
                      snake_case_       =      np.concatenate([latent_model_input, image],							axis=1)

                      # timestep to tensor
                      snake_case_       =      np.array([t],							dtype=lowerCAmelCase__)

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

                      # perform guidance
                      if do_classifier_free_guidance:
                           snake_case_ ,			snake_case_       =      np.split(lowerCAmelCase__,							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(lowerCAmelCase__),							lowerCAmelCase__,							lowerCAmelCase__,							**lowerCAmelCase__).prev_sample

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

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

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

            if not return_dict:
                 return (image,)

            return ImagePipelineOutput(images=lowerCAmelCase__)







       def 	a_		(				self,							lowerCAmelCase__)   ->   Any:
            snake_case_       =      1 / 0.08333 * latents
            snake_case_       =      self.vae(latent_sample=lowerCAmelCase__)[0]
            snake_case_       =      np.clip(image / 2 + 0.5,							0,							1)
            snake_case_       =      image.transpose((0, 2, 3, 1))
            return image







       def 	a_		(				self,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__ = None,							lowerCAmelCase__ = None,							)   ->   Optional[int]:



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

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

                 if untruncated_ids.shape[-1] >= text_input_ids.shape[-1] and not torch.equal(
                     lowerCAmelCase__,							lowerCAmelCase__):
                      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(lowerCAmelCase__),							)
                 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,							lowerCAmelCase__)
            snake_case_       =      prompt_embeds.reshape(bs_embed * num_images_per_prompt,							lowerCAmelCase__,							-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(lowerCAmelCase__) is not type(lowerCAmelCase__):
                      raise TypeError(
                          f'`negative_prompt` should be the same type to `prompt`, but got {type(lowerCAmelCase__)} !='
                          f' {type(lowerCAmelCase__)}.')
                 elif isinstance(lowerCAmelCase__,							lowerCAmelCase__):
                      snake_case_       =      [negative_prompt]
                 elif batch_size != len(lowerCAmelCase__):
                      raise ValueError(
                          f'`negative_prompt`: {negative_prompt} has batch size {len(lowerCAmelCase__)}, 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(
                     lowerCAmelCase__,							padding='max_length',							max_length=lowerCAmelCase__,							truncation=lowerCAmelCase__,							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(lowerCAmelCase__),							)
                 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,							lowerCAmelCase__)
                 snake_case_       =      uncond_embeddings.reshape(batch_size * num_images_per_prompt,							lowerCAmelCase__,							-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"""


import copy

from ...configuration_utils import PretrainedConfig
from ...utils import logging


__UpperCamelCase								=    logging.get_logger(__name__)

__UpperCamelCase								=    {
    '''facebook/sam-vit-huge''': '''https://huggingface.co/facebook/sam-vit-huge/resolve/main/config.json''',
    '''facebook/sam-vit-large''': '''https://huggingface.co/facebook/sam-vit-large/resolve/main/config.json''',
    '''facebook/sam-vit-base''': '''https://huggingface.co/facebook/sam-vit-base/resolve/main/config.json''',
}




class 				UpperCamelCase      (       lowerCAmelCase__     ):







       def __init__(				self,							lowerCAmelCase__=256,							lowerCAmelCase__=1024,							lowerCAmelCase__=16,							lowerCAmelCase__=16,							lowerCAmelCase__=4,							lowerCAmelCase__="gelu",							lowerCAmelCase__=1e-6,							**lowerCAmelCase__,							)   ->   int:
            super().__init__(**lowerCAmelCase__)
            snake_case_       =      hidden_size
            snake_case_       =      image_size
            snake_case_       =      patch_size
            snake_case_       =      image_size // patch_size
            snake_case_       =      mask_input_channels
            snake_case_       =      num_point_embeddings
            snake_case_       =      hidden_act
            snake_case_       =      layer_norm_eps




class 				UpperCamelCase      (       lowerCAmelCase__     ):







       def __init__(				self,							lowerCAmelCase__=256,							lowerCAmelCase__="relu",							lowerCAmelCase__=2048,							lowerCAmelCase__=2,							lowerCAmelCase__=8,							lowerCAmelCase__=2,							lowerCAmelCase__=3,							lowerCAmelCase__=3,							lowerCAmelCase__=256,							lowerCAmelCase__=1e-6,							**lowerCAmelCase__,							)   ->   List[str]:
            super().__init__(**lowerCAmelCase__)
            snake_case_       =      hidden_size
            snake_case_       =      hidden_act
            snake_case_       =      mlp_dim
            snake_case_       =      num_hidden_layers
            snake_case_       =      num_attention_heads
            snake_case_       =      attention_downsample_rate
            snake_case_       =      num_multimask_outputs
            snake_case_       =      iou_head_depth
            snake_case_       =      iou_head_hidden_dim
            snake_case_       =      layer_norm_eps




class 				UpperCamelCase      (       lowerCAmelCase__     ):







       def __init__(				self,							lowerCAmelCase__=768,							lowerCAmelCase__=256,							lowerCAmelCase__=12,							lowerCAmelCase__=12,							lowerCAmelCase__=3,							lowerCAmelCase__=1024,							lowerCAmelCase__=16,							lowerCAmelCase__="gelu",							lowerCAmelCase__=1e-06,							lowerCAmelCase__=0.0,							lowerCAmelCase__=1e-10,							lowerCAmelCase__=True,							lowerCAmelCase__=4.0,							lowerCAmelCase__=True,							lowerCAmelCase__=True,							lowerCAmelCase__=14,							lowerCAmelCase__=[2, 5, 8, 11],							lowerCAmelCase__=128,							lowerCAmelCase__=None,							**lowerCAmelCase__,							)   ->   List[str]:
            super().__init__(**lowerCAmelCase__)

            snake_case_       =      hidden_size
            snake_case_       =      output_channels
            snake_case_       =      num_hidden_layers
            snake_case_       =      num_attention_heads
            snake_case_       =      num_channels
            snake_case_       =      image_size
            snake_case_       =      patch_size
            snake_case_       =      hidden_act
            snake_case_       =      layer_norm_eps
            snake_case_       =      attention_dropout
            snake_case_       =      initializer_range
            snake_case_       =      qkv_bias
            snake_case_       =      mlp_ratio
            snake_case_       =      use_abs_pos
            snake_case_       =      use_rel_pos
            snake_case_       =      window_size
            snake_case_       =      global_attn_indexes
            snake_case_       =      num_pos_feats
            snake_case_       =      int(hidden_size * mlp_ratio) if mlp_dim is None else mlp_dim




class 				UpperCamelCase      (       lowerCAmelCase__     ):
       SCREAMING_SNAKE_CASE_				=   "sam"
       SCREAMING_SNAKE_CASE_				=   True







       def __init__(				self,							lowerCAmelCase__=None,							lowerCAmelCase__=None,							lowerCAmelCase__=None,							lowerCAmelCase__=0.02,							**lowerCAmelCase__,							)   ->   Any:
            super().__init__(**lowerCAmelCase__)
            snake_case_       =      vision_config if vision_config is not None else {}
            snake_case_       =      prompt_encoder_config if prompt_encoder_config is not None else {}
            snake_case_       =      mask_decoder_config if mask_decoder_config is not None else {}

            if isinstance(lowerCAmelCase__,							lowerCAmelCase__):
                 snake_case_       =      vision_config.to_dict()
            if isinstance(lowerCAmelCase__,							lowerCAmelCase__):
                 snake_case_       =      prompt_encoder_config.to_dict()
            if isinstance(lowerCAmelCase__,							lowerCAmelCase__):
                 snake_case_       =      mask_decoder_config.to_dict()

            snake_case_       =      SamVisionConfig(**lowerCAmelCase__)
            snake_case_       =      SamPromptEncoderConfig(**lowerCAmelCase__)
            snake_case_       =      SamMaskDecoderConfig(**lowerCAmelCase__)
            snake_case_       =      initializer_range







       def 	a_		(				self)   ->   Dict:
            snake_case_       =      copy.deepcopy(self.__dict__)
            snake_case_       =      self.vision_config.to_dict()
            snake_case_       =      self.prompt_encoder_config.to_dict()
            snake_case_       =      self.mask_decoder_config.to_dict()
            snake_case_       =      self.__class__.model_type
            return output

