
from typing import List, Optional, Union

import torch
from transformers import (
    XLMRobertaTokenizer,
)

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


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

_A		     =    '''
    Examples:
        ```py
        >>> from diffusers import KandinskyPipeline, KandinskyPriorPipeline
        >>> import torch

        >>> pipe_prior = KandinskyPriorPipeline.from_pretrained("kandinsky-community/Kandinsky-2-1-prior")
        >>> pipe_prior.to("cuda")

        >>> prompt = "red cat, 4k photo"
        >>> out = pipe_prior(prompt)
        >>> image_emb = out.image_embeds
        >>> negative_image_emb = out.negative_image_embeds

        >>> pipe = KandinskyPipeline.from_pretrained("kandinsky-community/kandinsky-2-1")
        >>> pipe.to("cuda")

        >>> image = pipe(
        ...     prompt,
        ...     image_embeds=image_emb,
        ...     negative_image_embeds=negative_image_emb,
        ...     height=768,
        ...     width=768,
        ...     num_inference_steps=100,
        ... ).images

        >>> image[0].save("cat.png")
        ```
'''




def  lowerCamelCase__ (  a__     :     str		,					a__     :     Any		,					a__     :     List[str]=8   )						->		Dict:
       UpperCamelCase_											=				h // scale_factor**2
       if h % scale_factor**2 != 0:
              new_h += 1
       UpperCamelCase_											=				w // scale_factor**2
       if w % scale_factor**2 != 0:
              new_w += 1
       return new_h * scale_factor, new_w * scale_factor







class 						lowercase_				(				__SCREAMING_SNAKE_CASE ):

 def __init__(					self   ,		__UpperCamelCase   ,		__UpperCamelCase   ,		__UpperCamelCase   ,		__UpperCamelCase   ,		__UpperCamelCase   ,		):




        """simple docstring"""


        super().__init__()

        self.register_modules(
            text_encoder=__lowerCAmelCase   ,		tokenizer=__lowerCAmelCase   ,		unet=__lowerCAmelCase   ,		scheduler=__lowerCAmelCase   ,		movq=__lowerCAmelCase   ,		)
        UpperCamelCase_											=				2 ** (len(self.movq.config.block_out_channels   ) - 1)

 def lowerCamelCase_ (					self   ,		__UpperCamelCase   ,		__UpperCamelCase   ,		__UpperCamelCase   ,		__UpperCamelCase   ,		__UpperCamelCase   ,		__UpperCamelCase   ):




        """simple docstring"""


        if latents is None:
               UpperCamelCase_											=				randn_tensor(__lowerCAmelCase   ,		generator=__lowerCAmelCase   ,		device=__lowerCAmelCase   ,		dtype=__lowerCAmelCase   )
        else:
               if latents.shape != shape:
                      raise ValueError(f'''Unexpected latents shape, got {latents.shape}, expected {shape}'''   )
               UpperCamelCase_											=				latents.to(__lowerCAmelCase   )

        UpperCamelCase_											=				latents * scheduler.init_noise_sigma
        return latents

 def lowerCamelCase_ (					self   ,		__UpperCamelCase   ,		__UpperCamelCase   ,		__UpperCamelCase   ,		__UpperCamelCase   ,		__UpperCamelCase=None   ,		):




        """simple docstring"""


        UpperCamelCase_											=				len(__lowerCAmelCase   ) if isinstance(__lowerCAmelCase   ,		__lowerCAmelCase   ) else 1
        # get prompt text embeddings
        UpperCamelCase_											=				self.tokenizer(
            __lowerCAmelCase   ,		padding="""max_length"""   ,		truncation=__lowerCAmelCase   ,		max_length=7_7   ,		return_attention_mask=__lowerCAmelCase   ,		add_special_tokens=__lowerCAmelCase   ,		return_tensors="""pt"""   ,		)

        UpperCamelCase_											=				text_inputs.input_ids
        UpperCamelCase_											=				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   ):
               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}'''   )

        UpperCamelCase_											=				text_input_ids.to(__lowerCAmelCase   )
        UpperCamelCase_											=				text_inputs.attention_mask.to(__lowerCAmelCase   )

        UpperCamelCase_				, UpperCamelCase_											=				self.text_encoder(
            input_ids=__lowerCAmelCase   ,		attention_mask=__lowerCAmelCase   )

        UpperCamelCase_											=				prompt_embeds.repeat_interleave(__lowerCAmelCase   ,		dim=0   )
        UpperCamelCase_											=				text_encoder_hidden_states.repeat_interleave(__lowerCAmelCase   ,		dim=0   )
        UpperCamelCase_											=				text_mask.repeat_interleave(__lowerCAmelCase   ,		dim=0   )

        if do_classifier_free_guidance:
               UpperCamelCase_											=				4_2
               if negative_prompt is None:
                      UpperCamelCase_											=				[""""""] * 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   ):
                      UpperCamelCase_											=				[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:
                      UpperCamelCase_											=				negative_prompt

               UpperCamelCase_											=				self.tokenizer(
                   __lowerCAmelCase   ,		padding="""max_length"""   ,		max_length=7_7   ,		truncation=__lowerCAmelCase   ,		return_attention_mask=__lowerCAmelCase   ,		add_special_tokens=__lowerCAmelCase   ,		return_tensors="""pt"""   ,		)
               UpperCamelCase_											=				uncond_input.input_ids.to(__lowerCAmelCase   )
               UpperCamelCase_											=				uncond_input.attention_mask.to(__lowerCAmelCase   )

               UpperCamelCase_				, UpperCamelCase_											=				self.text_encoder(
                   input_ids=__lowerCAmelCase   ,		attention_mask=__lowerCAmelCase   )

               # duplicate unconditional embeddings for each generation per prompt, using mps friendly method

               UpperCamelCase_											=				negative_prompt_embeds.shape[1]
               UpperCamelCase_											=				negative_prompt_embeds.repeat(1   ,		__lowerCAmelCase   )
               UpperCamelCase_											=				negative_prompt_embeds.view(batch_size * num_images_per_prompt   ,		__lowerCAmelCase   )

               UpperCamelCase_											=				uncond_text_encoder_hidden_states.shape[1]
               UpperCamelCase_											=				uncond_text_encoder_hidden_states.repeat(1   ,		__lowerCAmelCase   ,		1   )
               UpperCamelCase_											=				uncond_text_encoder_hidden_states.view(
                   batch_size * num_images_per_prompt   ,		__lowerCAmelCase   ,		-1   )
               UpperCamelCase_											=				uncond_text_mask.repeat_interleave(__lowerCAmelCase   ,		dim=0   )

               # done duplicates

               # 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_											=				torch.cat([negative_prompt_embeds, prompt_embeds]   )
               UpperCamelCase_											=				torch.cat([uncond_text_encoder_hidden_states, text_encoder_hidden_states]   )

               UpperCamelCase_											=				torch.cat([uncond_text_mask, text_mask]   )

        return prompt_embeds, text_encoder_hidden_states, text_mask

 def lowerCamelCase_ (					self   ,		__UpperCamelCase=0   ):




        """simple docstring"""


        if is_accelerate_available():
               from accelerate import cpu_offload
        else:
               raise ImportError("""Please install accelerate via `pip install accelerate`"""   )

        UpperCamelCase_											=				torch.device(f'''cuda:{gpu_id}'''   )

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

 def lowerCamelCase_ (					self   ,		__UpperCamelCase=0   ):




        """simple docstring"""


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

        UpperCamelCase_											=				torch.device(f'''cuda:{gpu_id}'''   )

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

        UpperCamelCase_											=				None
        for cpu_offloaded_model in [self.text_encoder, self.unet, self.movq]:
               UpperCamelCase_				, UpperCamelCase_											=				cpu_offload_with_hook(__lowerCAmelCase   ,		__lowerCAmelCase   ,		prev_module_hook=__lowerCAmelCase   )

        if self.safety_checker is not None:
               UpperCamelCase_				, UpperCamelCase_											=				cpu_offload_with_hook(self.safety_checker   ,		__lowerCAmelCase   ,		prev_module_hook=__lowerCAmelCase   )

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

 @property
 # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
 def lowerCamelCase_ (					self   ):




        """simple docstring"""


        if not hasattr(self.unet   ,		"""_hf_hook"""   ):
               return self.device
        for module in self.unet.modules():
               if (
                   hasattr(__lowerCAmelCase   ,		"""_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(__lowerCAmelCase   )
 def __call__(					self   ,		__UpperCamelCase   ,		__UpperCamelCase   ,		__UpperCamelCase   ,		__UpperCamelCase = None   ,		__UpperCamelCase = 5_1_2   ,		__UpperCamelCase = 5_1_2   ,		__UpperCamelCase = 1_0_0   ,		__UpperCamelCase = 4.0   ,		__UpperCamelCase = 1   ,		__UpperCamelCase = None   ,		__UpperCamelCase = None   ,		__UpperCamelCase = "pil"   ,		__UpperCamelCase = True   ,		):




        """simple docstring"""


        if isinstance(__lowerCAmelCase   ,		__lowerCAmelCase   ):
               UpperCamelCase_											=				1
        elif isinstance(__lowerCAmelCase   ,		__lowerCAmelCase   ):
               UpperCamelCase_											=				len(__lowerCAmelCase   )
        else:
               raise ValueError(f'''`prompt` has to be of type `str` or `list` but is {type(__lowerCAmelCase   )}'''   )

        UpperCamelCase_											=				self._execution_device

        UpperCamelCase_											=				batch_size * num_images_per_prompt
        UpperCamelCase_											=				guidance_scale > 1.0

        UpperCamelCase_				, UpperCamelCase_				, UpperCamelCase_											=				self._encode_prompt(
            __lowerCAmelCase   ,		__lowerCAmelCase   ,		__lowerCAmelCase   ,		__lowerCAmelCase   ,		__lowerCAmelCase   )

        if isinstance(__lowerCAmelCase   ,		__lowerCAmelCase   ):
               UpperCamelCase_											=				torch.cat(__lowerCAmelCase   ,		dim=0   )
        if isinstance(__lowerCAmelCase   ,		__lowerCAmelCase   ):
               UpperCamelCase_											=				torch.cat(__lowerCAmelCase   ,		dim=0   )

        if do_classifier_free_guidance:
               UpperCamelCase_											=				image_embeds.repeat_interleave(__lowerCAmelCase   ,		dim=0   )
               UpperCamelCase_											=				negative_image_embeds.repeat_interleave(__lowerCAmelCase   ,		dim=0   )

        UpperCamelCase_											=				torch.cat([negative_image_embeds, image_embeds]   ,		dim=0   ).to(
            dtype=prompt_embeds.dtype   ,		device=__lowerCAmelCase   )

        self.scheduler.set_timesteps(__lowerCAmelCase   ,		device=__lowerCAmelCase   )
        UpperCamelCase_											=				self.scheduler.timesteps

        UpperCamelCase_											=				self.unet.config.in_channels

        UpperCamelCase_				, UpperCamelCase_											=				get_new_h_w(__lowerCAmelCase   ,		__lowerCAmelCase   ,		self.movq_scale_factor   )

        # create initial latent
        UpperCamelCase_											=				self.prepare_latents(
            (batch_size, num_channels_latents, height, width)   ,		text_encoder_hidden_states.dtype   ,		__lowerCAmelCase   ,		__lowerCAmelCase   ,		__lowerCAmelCase   ,		self.scheduler   ,		)

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

               UpperCamelCase_											=				{"""text_embeds""": prompt_embeds, """image_embeds""": image_embeds}
               UpperCamelCase_											=				self.unet(
                   sample=__lowerCAmelCase   ,		timestep=__lowerCAmelCase   ,		encoder_hidden_states=__lowerCAmelCase   ,		added_cond_kwargs=__lowerCAmelCase   ,		return_dict=__lowerCAmelCase   ,		)[0]

               if do_classifier_free_guidance:
                      UpperCamelCase_				, UpperCamelCase_											=				noise_pred.split(latents.shape[1]   ,		dim=1   )
                      UpperCamelCase_				, UpperCamelCase_											=				noise_pred.chunk(2   )
                      UpperCamelCase_				, UpperCamelCase_											=				variance_pred.chunk(2   )
                      UpperCamelCase_											=				noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
                      UpperCamelCase_											=				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"]
               ):
                      UpperCamelCase_				, UpperCamelCase_											=				noise_pred.split(latents.shape[1]   ,		dim=1   )

               # compute the previous noisy sample x_t -> x_t-1
               UpperCamelCase_											=				self.scheduler.step(
                   __lowerCAmelCase   ,		__lowerCAmelCase   ,		__lowerCAmelCase   ,		generator=__lowerCAmelCase   ,		).prev_sample
        # post-processing
        UpperCamelCase_											=				self.movq.decode(__lowerCAmelCase   ,		force_not_quantize=__lowerCAmelCase   )["""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"]:
               UpperCamelCase_											=				image * 0.5 + 0.5
               UpperCamelCase_											=				image.clamp(0   ,		1   )
               UpperCamelCase_											=				image.cpu().permute(0   ,		2   ,		3   ,		1   ).float().numpy()

        if output_type == "pil":
               UpperCamelCase_											=				self.numpy_to_pil(__lowerCAmelCase   )

        if not return_dict:
               return (image,)

        return ImagePipelineOutput(images=__lowerCAmelCase   )










import dataclasses
import json
import warnings
from dataclasses import dataclass, field
from time import time
from typing import List

from ..utils import logging


A    :					Dict =      logging.get_logger(__name__)







def    __lowerCamelCase					(							__a   :int=None    ,     __a   :Optional[Any]=None    )		->				int:






		"""simple docstring"""

		return field(default_factory=lambda: default    ,     metadata=__a    )






@dataclass
class   A			:
			'''simple docstring'''




			__lowerCamelCase	:       List[str]							=    list_field(
			    default=[]   ,     metadata={
			        '''help''': (
			            '''Model checkpoints to be provided to the AutoModel classes. Leave blank to benchmark the base version'''
			            ''' of all available models'''
			        )
			    }   ,     )

			__lowerCamelCase	:       List[int]							=    list_field(
			    default=[8]   ,     metadata={'''help''': '''List of batch sizes for which memory and time performance will be evaluated'''}    )

			__lowerCamelCase	:       List[int]							=    list_field(
			    default=[8, 32, 128, 512]   ,     metadata={'''help''': '''List of sequence lengths for which memory and time performance will be evaluated'''}   ,     )

			__lowerCamelCase	:       bool							=    field(
			    default=SCREAMING_SNAKE_CASE   ,     metadata={'''help''': '''Whether to benchmark inference of model. Inference can be disabled via --no-inference.'''}   ,     )
			__lowerCamelCase	:       bool							=    field(
			    default=SCREAMING_SNAKE_CASE   ,     metadata={'''help''': '''Whether to run on available cuda devices. Cuda can be disabled via --no-cuda.'''}   ,     )
			__lowerCamelCase	:       bool							=    field(
			    default=SCREAMING_SNAKE_CASE   ,     metadata={'''help''': '''Whether to run on available tpu devices. TPU can be disabled via --no-tpu.'''}    )
			__lowerCamelCase	:       bool							=    field(default=SCREAMING_SNAKE_CASE   ,     metadata={'''help''': '''Use FP16 to accelerate inference.'''}    )
			__lowerCamelCase	:       bool							=    field(default=SCREAMING_SNAKE_CASE   ,     metadata={'''help''': '''Benchmark training of model'''}    )
			__lowerCamelCase	:       bool							=    field(default=SCREAMING_SNAKE_CASE   ,     metadata={'''help''': '''Verbose memory tracing'''}    )
			__lowerCamelCase	:       bool							=    field(
			    default=SCREAMING_SNAKE_CASE   ,     metadata={'''help''': '''Whether to perform speed measurements. Speed measurements can be disabled via --no-speed.'''}   ,     )
			__lowerCamelCase	:       bool							=    field(
			    default=SCREAMING_SNAKE_CASE   ,     metadata={
			        '''help''': '''Whether to perform memory measurements. Memory measurements can be disabled via --no-memory'''
			    }   ,     )
			__lowerCamelCase	:       bool							=    field(default=SCREAMING_SNAKE_CASE   ,     metadata={'''help''': '''Trace memory line by line'''}    )
			__lowerCamelCase	:       bool							=    field(default=SCREAMING_SNAKE_CASE   ,     metadata={'''help''': '''Save result to a CSV file'''}    )
			__lowerCamelCase	:       bool							=    field(default=SCREAMING_SNAKE_CASE   ,     metadata={'''help''': '''Save all print statements in a log file'''}    )
			__lowerCamelCase	:       bool							=    field(default=SCREAMING_SNAKE_CASE   ,     metadata={'''help''': '''Whether to print environment information'''}    )
			__lowerCamelCase	:       bool							=    field(
			    default=SCREAMING_SNAKE_CASE   ,     metadata={
			        '''help''': (
			            '''Whether to use multiprocessing for memory and speed measurement. It is highly recommended to use'''
			            ''' multiprocessing for accurate CPU and GPU memory measurements. This option should only be disabled'''
			            ''' for debugging / testing and on TPU.'''
			        )
			    }   ,     )
			__lowerCamelCase	:       str							=    field(
			    default=F'''inference_time_{round(time()    )}.csv'''   ,     metadata={'''help''': '''CSV filename used if saving time results to csv.'''}   ,     )
			__lowerCamelCase	:       str							=    field(
			    default=F'''inference_memory_{round(time()    )}.csv'''   ,     metadata={'''help''': '''CSV filename used if saving memory results to csv.'''}   ,     )
			__lowerCamelCase	:       str							=    field(
			    default=F'''train_time_{round(time()    )}.csv'''   ,     metadata={'''help''': '''CSV filename used if saving time results to csv for training.'''}   ,     )
			__lowerCamelCase	:       str							=    field(
			    default=F'''train_memory_{round(time()    )}.csv'''   ,     metadata={'''help''': '''CSV filename used if saving memory results to csv for training.'''}   ,     )
			__lowerCamelCase	:       str							=    field(
			    default=F'''env_info_{round(time()    )}.csv'''   ,     metadata={'''help''': '''CSV filename used if saving environment information.'''}   ,     )
			__lowerCamelCase	:       str							=    field(
			    default=F'''log_{round(time()    )}.csv'''   ,     metadata={'''help''': '''Log filename used if print statements are saved in log.'''}   ,     )
			__lowerCamelCase	:       int							=    field(default=3   ,     metadata={'''help''': '''Times an experiment will be run.'''}    )
			__lowerCamelCase	:       bool							=    field(
			    default=SCREAMING_SNAKE_CASE   ,     metadata={
			        '''help''': (
			            '''Instead of loading the model as defined in `config.architectures` if exists, just load the pretrain'''
			            ''' model weights.'''
			        )
			    }   ,     )






			def    a_       (			self   :    Dict   )	->	Union[str, Any]:







					"""simple docstring"""






					warnings.warn(
					    f'The class {self.__class__} is deprecated. Hugging Face Benchmarking utils'
					    """ are deprecated in general and it is advised to use external Benchmarking libraries """
					    """ to benchmark Transformer models."""							,	__lowerCAmelCase							,	)






			def    a_       (			self   :    Union[str, Any]   )	->	List[str]:







					"""simple docstring"""






					return json.dumps(dataclasses.asdict(self   )							,	indent=2   )






			@property
			def    a_       (			self   :    Tuple   )	->	List[str]:







					"""simple docstring"""






					if len(self.models   ) <= 0:
							raise ValueError(
							    """Please make sure you provide at least one model name / model identifier, *e.g.* `--models"""
							    """ bert-base-cased` or `args.models = ['bert-base-cased']."""   )
					return self.models







			@property
			def    a_       (			self   :    Union[str, Any]   )	->	Optional[Any]:







					"""simple docstring"""






					if not self.multi_process:
							return False
					elif self.is_tpu:
							logger.info("""Multiprocessing is currently not possible on TPU."""   )
							return False
					else:
							return True



