




'''simple docstring'''





from typing import List, Optional, Union

import torch

from ...models import UNetaDConditionModel, VQModel
from ...pipelines import DiffusionPipeline
from ...pipelines.pipeline_utils import ImagePipelineOutput
from ...schedulers import DDPMScheduler
from ...utils import (
    is_accelerate_available,
    is_accelerate_version,
    logging,
    randn_tensor,
    replace_example_docstring,
)


SCREAMING_SNAKE_CASE__      					=     logging.get_logger(__name__)  # pylint: disable=invalid-name

SCREAMING_SNAKE_CASE__      					=     "\n    Examples:\n        ```py\n        >>> from diffusers import KandinskyV22Pipeline, KandinskyV22PriorPipeline\n        >>> import torch\n\n        >>> pipe_prior = KandinskyV22PriorPipeline.from_pretrained(\"kandinsky-community/kandinsky-2-2-prior\")\n        >>> pipe_prior.to(\"cuda\")\n        >>> prompt = \"red cat, 4k photo\"\n        >>> out = pipe_prior(prompt)\n        >>> image_emb = out.image_embeds\n        >>> zero_image_emb = out.negative_image_embeds\n        >>> pipe = KandinskyV22Pipeline.from_pretrained(\"kandinsky-community/kandinsky-2-2-decoder\")\n        >>> pipe.to(\"cuda\")\n        >>> image = pipe(\n        ...     image_embeds=image_emb,\n        ...     negative_image_embeds=zero_image_emb,\n        ...     height=768,\n        ...     width=768,\n        ...     num_inference_steps=50,\n        ... ).images\n        >>> image[0].save(\"cat.png\")\n        ```\n"




def        lowerCamelCase  (				_snake_case  :						Dict     ,_snake_case  :						str     ,_snake_case  :						Dict=8       ):





    '''simple docstring'''


    lowercase__		       =						height // scale_factor**2
    if height % scale_factor**2 != 0:
        new_height += 1
    lowercase__		       =						width // scale_factor**2
    if width % scale_factor**2 != 0:
        new_width += 1
    return new_height * scale_factor, new_width * scale_factor



class 				snake_case (UpperCamelCase   ):



       def __init__(				self     ,UpperCAmelCase_     ,UpperCAmelCase_     ,UpperCAmelCase_     ,)  ->							Optional[Any]:
           super().__init__()

           self.register_modules(
               unet=UpperCAmelCase_     ,scheduler=UpperCAmelCase_     ,movq=UpperCAmelCase_     ,)
           lowercase__		       =						2 ** (len(self.movq.config.block_out_channels      ) - 1)



       def 							_a      (				self     ,UpperCAmelCase_     ,UpperCAmelCase_     ,UpperCAmelCase_     ,UpperCAmelCase_     ,UpperCAmelCase_     ,UpperCAmelCase_      )  ->							int:
           if latents is None:
               lowercase__		       =						randn_tensor(UpperCAmelCase_     ,generator=UpperCAmelCase_     ,device=UpperCAmelCase_     ,dtype=UpperCAmelCase_      )
           else:
               if latents.shape != shape:
                   raise ValueError(F'''Unexpected latents shape, got {latents.shape}, expected {shape}'''      )
               lowercase__		       =						latents.to(UpperCAmelCase_      )

           lowercase__		       =						latents * scheduler.init_noise_sigma
           return latents



       def 							_a      (				self     ,UpperCAmelCase_=0      )  ->							Dict:
           if is_accelerate_available():
               from accelerate import cpu_offload
           else:
               raise ImportError("Please install accelerate via `pip install accelerate`"      )

           lowercase__		       =						torch.device(F'''cuda:{gpu_id}'''      )

           lowercase__		       =						[
               self.unet,
               self.movq,
           ]
           for cpu_offloaded_model in models:
               if cpu_offloaded_model is not None:
                   cpu_offload(UpperCAmelCase_     ,UpperCAmelCase_      )



       def 							_a      (				self     ,UpperCAmelCase_=0      )  ->							List[Any]:
           if is_accelerate_available() and is_accelerate_version(">="     ,"0.17.0.dev0"      ):
               from accelerate import cpu_offload_with_hook
           else:
               raise ImportError("`enable_model_cpu_offload` requires `accelerate v0.17.0` or higher."      )

           lowercase__		       =						torch.device(F'''cuda:{gpu_id}'''      )

           if self.device.type != "cpu":
               self.to("cpu"     ,silence_dtype_warnings=UpperCAmelCase_      )
               torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

           lowercase__		       =						None
           for cpu_offloaded_model in [self.unet, self.movq]:
               lowercase__       ,       lowercase__		       =						cpu_offload_with_hook(UpperCAmelCase_     ,UpperCAmelCase_     ,prev_module_hook=UpperCAmelCase_      )

           # We'll offload the last model manually.
           lowercase__		       =						hook



       @property
       # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
       def 							_a      (				self      )  ->							Tuple:
           if not hasattr(self.unet     ,"_hf_hook"      ):
               return self.device
           for module in self.unet.modules():
               if (
                   hasattr(UpperCAmelCase_     ,"_hf_hook"      )
                   and hasattr(module._hf_hook     ,"execution_device"      )
                   and module._hf_hook.execution_device is not None
               ):
                   return torch.device(module._hf_hook.execution_device      )
           return self.device







       @torch.no_grad()
       @replace_example_docstring(UpperCAmelCase_      )
       def __call__(				self     ,UpperCAmelCase_     ,UpperCAmelCase_     ,UpperCAmelCase_ = 512     ,UpperCAmelCase_ = 512     ,UpperCAmelCase_ = 100     ,UpperCAmelCase_ = 4.0     ,UpperCAmelCase_ = 1     ,UpperCAmelCase_ = None     ,UpperCAmelCase_ = None     ,UpperCAmelCase_ = "pil"     ,UpperCAmelCase_ = True     ,)  ->							Union[str, Any]:
           lowercase__		       =						self._execution_device

           lowercase__		       =						guidance_scale > 1.0

           if isinstance(UpperCAmelCase_     ,UpperCAmelCase_      ):
               lowercase__		       =						torch.cat(UpperCAmelCase_     ,dim=0      )
           lowercase__		       =						image_embeds.shape[0] * num_images_per_prompt
           if isinstance(UpperCAmelCase_     ,UpperCAmelCase_      ):
               lowercase__		       =						torch.cat(UpperCAmelCase_     ,dim=0      )

           if do_classifier_free_guidance:
               lowercase__		       =						image_embeds.repeat_interleave(UpperCAmelCase_     ,dim=0      )
               lowercase__		       =						negative_image_embeds.repeat_interleave(UpperCAmelCase_     ,dim=0      )

           lowercase__		       =						torch.cat([negative_image_embeds, image_embeds]     ,dim=0      ).to(dtype=self.unet.dtype     ,device=UpperCAmelCase_      )

           self.scheduler.set_timesteps(UpperCAmelCase_     ,device=UpperCAmelCase_      )
           lowercase__		       =						self.scheduler.timesteps

           lowercase__		       =						self.unet.config.in_channels

           lowercase__       ,       lowercase__		       =						downscale_height_and_width(UpperCAmelCase_     ,UpperCAmelCase_     ,self.movq_scale_factor      )

           # create initial latent
           lowercase__		       =						self.prepare_latents(
               (batch_size, num_channels_latents, height, width)     ,image_embeds.dtype     ,UpperCAmelCase_     ,UpperCAmelCase_     ,UpperCAmelCase_     ,self.scheduler     ,)

           for i, t in enumerate(self.progress_bar(UpperCAmelCase_      )      ):
               # expand the latents if we are doing classifier free guidance
               lowercase__		       =						torch.cat([latents] * 2      ) if do_classifier_free_guidance else latents

               lowercase__		       =						{"image_embeds": image_embeds}
               lowercase__		       =						self.unet(
                   sample=UpperCAmelCase_     ,timestep=UpperCAmelCase_     ,encoder_hidden_states=UpperCAmelCase_     ,added_cond_kwargs=UpperCAmelCase_     ,return_dict=UpperCAmelCase_     ,)[0]

               if do_classifier_free_guidance:
                   lowercase__       ,       lowercase__		       =						noise_pred.split(latents.shape[1]     ,dim=1      )
                   lowercase__       ,       lowercase__		       =						noise_pred.chunk(2      )
                   lowercase__       ,       lowercase__		       =						variance_pred.chunk(2      )
                   lowercase__		       =						noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
                   lowercase__		       =						torch.cat([noise_pred, variance_pred_text]     ,dim=1      )

               if not (
                   hasattr(self.scheduler.config     ,"variance_type"      )
                   and self.scheduler.config.variance_type in ["learned", "learned_range"]
               ):
                   lowercase__       ,       lowercase__		       =						noise_pred.split(latents.shape[1]     ,dim=1      )

               # compute the previous noisy sample x_t -> x_t-1
               lowercase__		       =						self.scheduler.step(
                   UpperCAmelCase_     ,UpperCAmelCase_     ,UpperCAmelCase_     ,generator=UpperCAmelCase_     ,)[0]
           # post-processing
           lowercase__		       =						self.movq.decode(UpperCAmelCase_     ,force_not_quantize=UpperCAmelCase_      )["sample"]

           if output_type not in ["pt", "np", "pil"]:
               raise ValueError(F'''Only the output types `pt`, `pil` and `np` are supported not output_type={output_type}'''      )

           if output_type in ["np", "pil"]:
               lowercase__		       =						image * 0.5 + 0.5
               lowercase__		       =						image.clamp(0     ,1      )
               lowercase__		       =						image.cpu().permute(0     ,2     ,3     ,1      ).float().numpy()

           if output_type == "pil":
               lowercase__		       =						self.numpy_to_pil(UpperCAmelCase_      )

           if not return_dict:
               return (image,)

           return ImagePipelineOutput(images=UpperCAmelCase_      )









'''simple docstring'''





import os
import time

import numpy as np
import onnxruntime as ort


SCREAMING_SNAKE_CASE__      					=     "1"
SCREAMING_SNAKE_CASE__      					=     "0"
SCREAMING_SNAKE_CASE__      					=     "1"

SCREAMING_SNAKE_CASE__      					=     ort.SessionOptions()
SCREAMING_SNAKE_CASE__      					=     ort.GraphOptimizationLevel.ORT_DISABLE_ALL
print("Create inference session...")
SCREAMING_SNAKE_CASE__      					=     ["TensorrtExecutionProvider", "CUDAExecutionProvider"]
SCREAMING_SNAKE_CASE__      					=     ort.InferenceSession("model.onnx", sess_options=sess_opt, providers=execution_provider)
SCREAMING_SNAKE_CASE__      					=     ort.RunOptions()

SCREAMING_SNAKE_CASE__      					=     128
SCREAMING_SNAKE_CASE__      					=     1
SCREAMING_SNAKE_CASE__      					=     np.ones((batch, sequence), dtype=np.intaa)
SCREAMING_SNAKE_CASE__      					=     np.ones((batch, sequence), dtype=np.intaa)
SCREAMING_SNAKE_CASE__      					=     np.ones((batch, sequence), dtype=np.intaa)

print("Warm up phase...")
sess.run(
    None,
    {
        sess.get_inputs()[0].name: input_ids,
        sess.get_inputs()[1].name: attention_mask,
        sess.get_inputs()[2].name: token_type_ids,
    },
    run_options=run_opt,
)

print("Start inference...")
SCREAMING_SNAKE_CASE__      					=     time.time()
SCREAMING_SNAKE_CASE__      					=     2000
SCREAMING_SNAKE_CASE__      					=     {}
for iter in range(max_iters):
 SCREAMING_SNAKE_CASE__      					=     sess.run(
     None,
     {
         sess.get_inputs()[0].name: input_ids,
         sess.get_inputs()[1].name: attention_mask,
         sess.get_inputs()[2].name: token_type_ids,
     },
     run_options=run_opt,
 )
print("Average Inference Time = {:.3f} ms".format((time.time() - start_time) * 1000 / max_iters))




