'''simple docstring'''


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


snake_case__					:      Optional[Any]				      =  logging.get_logger(__name__)  # pylint: disable=invalid-name

snake_case__					:      Tuple				      =  '''
    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   (				lowerCamelCase_ :							Optional[int]     ,					lowerCamelCase_ :							str     ,					lowerCamelCase_ :							Any=8       ):



     """simple docstring"""





     UpperCAmelCase_							:       Dict       =				h // scale_factor**2
     if h % scale_factor**2 != 0:
          new_h += 1
     UpperCAmelCase_							:       str       =				w // scale_factor**2
     if w % scale_factor**2 != 0:
          new_w += 1
     return new_h * scale_factor, new_w * scale_factor






class 					__SCREAMING_SNAKE_CASE						(	lowerCamelCase_	):







  '''simple docstring'''






  def __init__(			self				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		):




       '''simple docstring'''

       super().__init__()

       self.register_modules(
           text_encoder=snake_case_				,		tokenizer=snake_case_				,		unet=snake_case_				,		scheduler=snake_case_				,		movq=snake_case_				,		)
       UpperCAmelCase_							:       List[str]       =				2 ** (len(self.movq.config.block_out_channels						) - 1)


  def _UpperCamelCase					(			self				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_						):




       '''simple docstring'''

       if latents is None:
            UpperCAmelCase_							:       Tuple       =				randn_tensor(snake_case_				,		generator=snake_case_				,		device=snake_case_				,		dtype=snake_case_						)
       else:
            if latents.shape != shape:
                 raise ValueError(F'''Unexpected latents shape, got {latents.shape}, expected {shape}'''						)
            UpperCAmelCase_							:       Tuple       =				latents.to(snake_case_						)

       UpperCAmelCase_							:       Dict       =				latents * scheduler.init_noise_sigma
       return latents


  def _UpperCamelCase					(			self				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_=None				,		):




       '''simple docstring'''

       UpperCAmelCase_							:       Optional[int]       =				len(snake_case_						) if isinstance(snake_case_				,		snake_case_						) else 1
       # get prompt text embeddings
       UpperCAmelCase_							:       Optional[Any]       =				self.tokenizer(
           snake_case_				,		padding='max_length'				,		truncation=snake_case_				,		max_length=7_7				,		return_attention_mask=snake_case_				,		add_special_tokens=snake_case_				,		return_tensors='pt'				,		)

       UpperCAmelCase_							:       List[str]       =				text_inputs.input_ids
       UpperCAmelCase_							:       Optional[int]       =				self.tokenizer(snake_case_				,		padding='longest'				,		return_tensors='pt'						).input_ids

       if untruncated_ids.shape[-1] >= text_input_ids.shape[-1] and not torch.equal(snake_case_				,		snake_case_						):
            UpperCAmelCase_							:       List[Any]       =				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_							:       Optional[Any]       =				text_input_ids.to(snake_case_						)
       UpperCAmelCase_							:       Dict       =				text_inputs.attention_mask.to(snake_case_						)

       UpperCAmelCase_       ,      UpperCAmelCase_							:       int       =				self.text_encoder(
           input_ids=snake_case_				,		attention_mask=snake_case_						)

       UpperCAmelCase_							:       str       =				prompt_embeds.repeat_interleave(snake_case_				,		dim=0						)
       UpperCAmelCase_							:       Dict       =				text_encoder_hidden_states.repeat_interleave(snake_case_				,		dim=0						)
       UpperCAmelCase_							:       int       =				text_mask.repeat_interleave(snake_case_				,		dim=0						)

       if do_classifier_free_guidance:
            UpperCAmelCase_							:       List[str]
            if negative_prompt is None:
                 UpperCAmelCase_							:       str       =				[''] * batch_size
            elif type(snake_case_						) is not type(snake_case_						):
                 raise TypeError(
                     F'''`negative_prompt` should be the same type to `prompt`, but got {type(snake_case_						)} !='''
                     F''' {type(snake_case_						)}.'''						)
            elif isinstance(snake_case_				,		snake_case_						):
                 UpperCAmelCase_							:       Tuple       =				[negative_prompt]
            elif batch_size != len(snake_case_						):
                 raise ValueError(
                     F'''`negative_prompt`: {negative_prompt} has batch size {len(snake_case_						)}, but `prompt`:'''
                     F''' {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches'''
                     ' the batch size of `prompt`.'						)
            else:
                 UpperCAmelCase_							:       Union[str, Any]       =				negative_prompt

            UpperCAmelCase_							:       Optional[int]       =				self.tokenizer(
                snake_case_				,		padding='max_length'				,		max_length=7_7				,		truncation=snake_case_				,		return_attention_mask=snake_case_				,		add_special_tokens=snake_case_				,		return_tensors='pt'				,		)
            UpperCAmelCase_							:       str       =				uncond_input.input_ids.to(snake_case_						)
            UpperCAmelCase_							:       List[str]       =				uncond_input.attention_mask.to(snake_case_						)

            UpperCAmelCase_       ,      UpperCAmelCase_							:       Optional[int]       =				self.text_encoder(
                input_ids=snake_case_				,		attention_mask=snake_case_						)

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

            UpperCAmelCase_							:       Union[str, Any]       =				negative_prompt_embeds.shape[1]
            UpperCAmelCase_							:       str       =				negative_prompt_embeds.repeat(1				,		snake_case_						)
            UpperCAmelCase_							:       Any       =				negative_prompt_embeds.view(batch_size * num_images_per_prompt				,		snake_case_						)

            UpperCAmelCase_							:       Tuple       =				uncond_text_encoder_hidden_states.shape[1]
            UpperCAmelCase_							:       int       =				uncond_text_encoder_hidden_states.repeat(1				,		snake_case_				,		1						)
            UpperCAmelCase_							:       Tuple       =				uncond_text_encoder_hidden_states.view(
                batch_size * num_images_per_prompt				,		snake_case_				,		-1						)
            UpperCAmelCase_							:       Any       =				uncond_text_mask.repeat_interleave(snake_case_				,		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_							:       Optional[int]       =				torch.cat([negative_prompt_embeds, prompt_embeds]						)
            UpperCAmelCase_							:       List[Any]       =				torch.cat([uncond_text_encoder_hidden_states, text_encoder_hidden_states]						)

            UpperCAmelCase_							:       Tuple       =				torch.cat([uncond_text_mask, text_mask]						)

       return prompt_embeds, text_encoder_hidden_states, text_mask


  def _UpperCamelCase					(			self				,		snake_case_=0						):




       '''simple docstring'''

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

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

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


  def _UpperCamelCase					(			self				,		snake_case_=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_							:       Any       =				torch.device(F'''cuda:{gpu_id}'''						)

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

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

       if self.safety_checker is not None:
            UpperCAmelCase_       ,      UpperCAmelCase_							:       List[str]       =				cpu_offload_with_hook(self.safety_checker				,		snake_case_				,		prev_module_hook=snake_case_						)

       # We'll offload the last model manually.
       UpperCAmelCase_							:       List[Any]       =				hook


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




       '''simple docstring'''

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




       '''simple docstring'''

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

       UpperCAmelCase_							:       Union[str, Any]       =				self._execution_device

       UpperCAmelCase_							:       Any       =				batch_size * num_images_per_prompt
       UpperCAmelCase_							:       Tuple       =				guidance_scale > 1.0

       UpperCAmelCase_       ,      UpperCAmelCase_       ,      UpperCAmelCase_							:       Optional[int]       =				self._encode_prompt(
           snake_case_				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_						)

       if isinstance(snake_case_				,		snake_case_						):
            UpperCAmelCase_							:       Optional[Any]       =				torch.cat(snake_case_				,		dim=0						)
       if isinstance(snake_case_				,		snake_case_						):
            UpperCAmelCase_							:       str       =				torch.cat(snake_case_				,		dim=0						)

       if do_classifier_free_guidance:
            UpperCAmelCase_							:       str       =				image_embeds.repeat_interleave(snake_case_				,		dim=0						)
            UpperCAmelCase_							:       Any       =				negative_image_embeds.repeat_interleave(snake_case_				,		dim=0						)

       UpperCAmelCase_							:       List[str]       =				torch.cat([negative_image_embeds, image_embeds]				,		dim=0						).to(
           dtype=prompt_embeds.dtype				,		device=snake_case_						)

       self.scheduler.set_timesteps(snake_case_				,		device=snake_case_						)
       UpperCAmelCase_							:       List[str]       =				self.scheduler.timesteps

       UpperCAmelCase_							:       List[Any]       =				self.unet.config.in_channels

       UpperCAmelCase_       ,      UpperCAmelCase_							:       str       =				get_new_h_w(snake_case_				,		snake_case_				,		self.movq_scale_factor						)

       # create initial latent
       UpperCAmelCase_							:       Union[str, Any]       =				self.prepare_latents(
           (batch_size, num_channels_latents, height, width)				,		text_encoder_hidden_states.dtype				,		snake_case_				,		snake_case_				,		snake_case_				,		self.scheduler				,		)

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

            UpperCAmelCase_							:       Dict       =				{'text_embeds': prompt_embeds, 'image_embeds': image_embeds}
            UpperCAmelCase_							:       Union[str, Any]       =				self.unet(
                sample=snake_case_				,		timestep=snake_case_				,		encoder_hidden_states=snake_case_				,		added_cond_kwargs=snake_case_				,		return_dict=snake_case_				,		)[0]

            if do_classifier_free_guidance:
                 UpperCAmelCase_       ,      UpperCAmelCase_							:       int       =				noise_pred.split(latents.shape[1]				,		dim=1						)
                 UpperCAmelCase_       ,      UpperCAmelCase_							:       Optional[Any]       =				noise_pred.chunk(2						)
                 UpperCAmelCase_       ,      UpperCAmelCase_							:       Union[str, Any]       =				variance_pred.chunk(2						)
                 UpperCAmelCase_							:       Union[str, Any]       =				noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
                 UpperCAmelCase_							:       Union[str, Any]       =				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_							:       Tuple       =				noise_pred.split(latents.shape[1]				,		dim=1						)

            # compute the previous noisy sample x_t -> x_t-1
            UpperCAmelCase_							:       Tuple       =				self.scheduler.step(
                snake_case_				,		snake_case_				,		snake_case_				,		generator=snake_case_				,		).prev_sample
       # post-processing
       UpperCAmelCase_							:       str       =				self.movq.decode(snake_case_				,		force_not_quantize=snake_case_						)['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_							:       Union[str, Any]       =				image * 0.5 + 0.5
            UpperCAmelCase_							:       str       =				image.clamp(0				,		1						)
            UpperCAmelCase_							:       List[str]       =				image.cpu().permute(0				,		2				,		3				,		1						).float().numpy()

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

       if not return_dict:
            return (image,)

       return ImagePipelineOutput(images=snake_case_						)







'''simple docstring'''


snake_case__					:      str				      =  '''Tobias Carryer'''

from time import time






class 					__SCREAMING_SNAKE_CASE						:







  '''simple docstring'''






  def __init__(			self				,		snake_case_				,		snake_case_				,		snake_case_				,		snake_case_=int(time()						)						):  # noqa: B008




       '''simple docstring'''

       UpperCAmelCase_							:       str       =				multiplier
       UpperCAmelCase_							:       Dict       =				increment
       UpperCAmelCase_							:       Tuple       =				modulo
       UpperCAmelCase_							:       Dict       =				seed






  def _UpperCamelCase					(			self						):




       '''simple docstring'''

       UpperCAmelCase_							:       Union[str, Any]       =				(self.multiplier * self.seed + self.increment) % self.modulo
       return self.seed


if __name__ == "__main__":
 # Show the LCG in action.
 snake_case__					:      Any				      =  LinearCongruentialGenerator(166_4525, 10_1390_4223, 2 << 31)
 while True:
  print(lcg.next_number())






