





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


snake_case__      :							Dict									=  getLogger(__name__)


snake_case__      :							str									=  4
snake_case__      :							Optional[int]									=  7

snake_case__      :							int									=  {
    'float16': torch.floataa,
    'float32': torch.floataa,
}



def _a       (			lowerCamelCase:     Any				)			->		Tuple:







      '''simple docstring'''






      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 A_      (    _lowerCamelCase     ):




   def __init__(self			:Optional[int]					,     _UpperCamelCase			:OnnxRuntimeModel					,     _UpperCamelCase			:OnnxRuntimeModel					,     _UpperCamelCase			:Any					,     _UpperCamelCase			:OnnxRuntimeModel					,     _UpperCamelCase			:DDPMScheduler					,     _UpperCamelCase			:Any					,     _UpperCamelCase			:int = 350					,     )->      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			:List[str]					,     _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					,     )->      Union[str, Any]:
         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 _lowerCAmelCase   (self			:Dict					,     _UpperCamelCase			:List[Any]				)->      Dict:
         __A      =       1 / 0.0_8_3_3_3 * 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 _lowerCAmelCase   (self			:Optional[Any]					,     _UpperCamelCase			:Union[str, List[str]]					,     _UpperCamelCase			:int					,     _UpperCamelCase			:Optional[int]					,     _UpperCamelCase			:bool					,     _UpperCamelCase			:Optional[str]					,     _UpperCamelCase			:Optional[torch.FloatTensor] = None					,     _UpperCamelCase			:Optional[torch.FloatTensor] = None					,     )->      Optional[Any]:
         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












from unittest.mock import patch

import datasets
from datasets import Dataset



def _a       (			)			->		Optional[Any]:







      '''simple docstring'''






      __A      =       Dataset.from_dict({'''col_1''': [3, 2, 0, 1]}				)

      with patch('''tqdm.auto.tqdm'''				) as mock_tqdm:
            datasets.disable_progress_bar()
            dset.map(lambda lowerCamelCase				: {"col_2": x["col_1"] + 1}				)
            mock_tqdm.assert_not_called()

            mock_tqdm.reset_mock()

            datasets.enable_progress_bar()
            dset.map(lambda lowerCamelCase				: {"col_2": x["col_1"] + 1}				)
            mock_tqdm.assert_called()





