

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


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

a    :Optional[Any]			=       """
    Examples:
        ```py
        >>> from diffusers import KandinskyV22Pipeline, KandinskyV22PriorPipeline
        >>> import torch

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




def        _lowercase    (			__lowerCAmelCase     ,    __lowerCAmelCase     ,    __lowerCAmelCase=8							)				-> List[str]:
      SCREAMING_SNAKE_CASE__      :					Any											=		height // scale_factor**2
      if height % scale_factor**2 != 0:
            new_height += 1
      SCREAMING_SNAKE_CASE__      :					Any											=		width // scale_factor**2
      if width % scale_factor**2 != 0:
            new_width += 1
      return new_height * scale_factor, new_width * scale_factor



class    __a							(UpperCAmelCase__):






     '''simple docstring'''

     def __init__(  self		,			_a		,			_a		,			_a		,			)     ->					str:


           """simple docstring"""


           super().__init__()

           self.register_modules(
               unet=lowercase_		,			scheduler=lowercase_		,			movq=lowercase_		,			)
           SCREAMING_SNAKE_CASE__      :					Tuple											=		2 ** (len(self.movq.config.block_out_channels     ) - 1)

     def     _a	(  self		,			_a		,			_a		,			_a		,			_a		,			_a		,			_a     )     ->					Optional[int]:


           """simple docstring"""


           if latents is None:
                 SCREAMING_SNAKE_CASE__      :					Optional[Any]											=		randn_tensor(lowercase_		,			generator=lowercase_		,			device=lowercase_		,			dtype=lowercase_     )
           else:
                 if latents.shape != shape:
                       raise ValueError(f'''Unexpected latents shape, got {latents.shape}, expected {shape}'''     )
                 SCREAMING_SNAKE_CASE__      :					Optional[int]											=		latents.to(lowercase_     )

           SCREAMING_SNAKE_CASE__      :					Any											=		latents * scheduler.init_noise_sigma
           return latents

     def     _a	(  self		,			_a=0     )     ->					List[str]:


           """simple docstring"""


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

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

           SCREAMING_SNAKE_CASE__      :					Optional[Any]											=		[
               self.unet,
               self.movq,
           ]
           for cpu_offloaded_model in models:
                 if cpu_offloaded_model is not None:
                       cpu_offload(lowercase_		,			lowercase_     )

     def     _a	(  self		,			_a=0     )     ->					Union[str, Any]:


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

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

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

           SCREAMING_SNAKE_CASE__      :					int											=		None
           for cpu_offloaded_model in [self.unet, self.movq]:
                 SCREAMING_SNAKE_CASE__      :					Tuple											=		cpu_offload_with_hook(lowercase_		,			lowercase_		,			prev_module_hook=lowercase_     )

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

     @property
     # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
     def     _a	(  self     )     ->					List[str]:


           """simple docstring"""


           if not hasattr(self.unet		,			"""_hf_hook"""     ):
                 return self.device
           for module in self.unet.modules():
                 if (
                     hasattr(lowercase_		,			"""_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(lowercase_     )
     def __call__(  self		,			_a		,			_a		,			_a = 512		,			_a = 512		,			_a = 100		,			_a = 4.0		,			_a = 1		,			_a = None		,			_a = None		,			_a = "pil"		,			_a = True		,			)     ->					str:


           """simple docstring"""


           SCREAMING_SNAKE_CASE__      :					Dict											=		self._execution_device

           SCREAMING_SNAKE_CASE__      :					Optional[int]											=		guidance_scale > 1.0

           if isinstance(lowercase_		,			lowercase_     ):
                 SCREAMING_SNAKE_CASE__      :					List[str]											=		torch.cat(lowercase_		,			dim=0     )
           SCREAMING_SNAKE_CASE__      :					int											=		image_embeds.shape[0] * num_images_per_prompt
           if isinstance(lowercase_		,			lowercase_     ):
                 SCREAMING_SNAKE_CASE__      :					Union[str, Any]											=		torch.cat(lowercase_		,			dim=0     )

           if do_classifier_free_guidance:
                 SCREAMING_SNAKE_CASE__      :					Optional[int]											=		image_embeds.repeat_interleave(lowercase_		,			dim=0     )
                 SCREAMING_SNAKE_CASE__      :					Optional[Any]											=		negative_image_embeds.repeat_interleave(lowercase_		,			dim=0     )

           SCREAMING_SNAKE_CASE__      :					Any											=		torch.cat([negative_image_embeds, image_embeds]		,			dim=0     ).to(dtype=self.unet.dtype		,			device=lowercase_     )

           self.scheduler.set_timesteps(lowercase_		,			device=lowercase_     )
           SCREAMING_SNAKE_CASE__      :					str											=		self.scheduler.timesteps

           SCREAMING_SNAKE_CASE__      :					Any											=		self.unet.config.in_channels

           SCREAMING_SNAKE_CASE__      :					Optional[int]											=		downscale_height_and_width(lowercase_		,			lowercase_		,			self.movq_scale_factor     )

           # create initial latent
           SCREAMING_SNAKE_CASE__      :					Dict											=		self.prepare_latents(
               (batch_size, num_channels_latents, height, width)		,			image_embeds.dtype		,			lowercase_		,			lowercase_		,			lowercase_		,			self.scheduler		,			)

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

                 SCREAMING_SNAKE_CASE__      :					str											=		{'''image_embeds''': image_embeds}
                 SCREAMING_SNAKE_CASE__      :					Optional[Any]											=		self.unet(
                     sample=lowercase_		,			timestep=lowercase_		,			encoder_hidden_states=lowercase_		,			added_cond_kwargs=lowercase_		,			return_dict=lowercase_		,			)[0]

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

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

           if output_type == "pil":
                 SCREAMING_SNAKE_CASE__      :					int											=		self.numpy_to_pil(lowercase_     )

           if not return_dict:
                 return (image,)

           return ImagePipelineOutput(images=lowercase_     )












"""simple docstring"""






import json
from typing import TYPE_CHECKING, List, Optional, Tuple

from tokenizers import pre_tokenizers

from ...tokenization_utils_base import BatchEncoding
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import logging
from .tokenization_gpta import GPTaTokenizer


if TYPE_CHECKING:
  from transformers.pipelines.conversational import Conversation


UpperCAmelCase_   : Optional[int]   			= logging.get_logger(__name__)

UpperCAmelCase_   : Dict   			= {"""vocab_file""": """vocab.json""", """merges_file""": """merges.txt""", """tokenizer_file""": """tokenizer.json"""}

UpperCAmelCase_   : Dict   			= {
    """vocab_file""": {
        """gpt2""": """https://huggingface.co/gpt2/resolve/main/vocab.json""",
        """gpt2-medium""": """https://huggingface.co/gpt2-medium/resolve/main/vocab.json""",
        """gpt2-large""": """https://huggingface.co/gpt2-large/resolve/main/vocab.json""",
        """gpt2-xl""": """https://huggingface.co/gpt2-xl/resolve/main/vocab.json""",
        """distilgpt2""": """https://huggingface.co/distilgpt2/resolve/main/vocab.json""",
    },
    """merges_file""": {
        """gpt2""": """https://huggingface.co/gpt2/resolve/main/merges.txt""",
        """gpt2-medium""": """https://huggingface.co/gpt2-medium/resolve/main/merges.txt""",
        """gpt2-large""": """https://huggingface.co/gpt2-large/resolve/main/merges.txt""",
        """gpt2-xl""": """https://huggingface.co/gpt2-xl/resolve/main/merges.txt""",
        """distilgpt2""": """https://huggingface.co/distilgpt2/resolve/main/merges.txt""",
    },
    """tokenizer_file""": {
        """gpt2""": """https://huggingface.co/gpt2/resolve/main/tokenizer.json""",
        """gpt2-medium""": """https://huggingface.co/gpt2-medium/resolve/main/tokenizer.json""",
        """gpt2-large""": """https://huggingface.co/gpt2-large/resolve/main/tokenizer.json""",
        """gpt2-xl""": """https://huggingface.co/gpt2-xl/resolve/main/tokenizer.json""",
        """distilgpt2""": """https://huggingface.co/distilgpt2/resolve/main/tokenizer.json""",
    },
}

UpperCAmelCase_   : List[str]   			= {
    """gpt2""": 1024,
    """gpt2-medium""": 1024,
    """gpt2-large""": 1024,
    """gpt2-xl""": 1024,
    """distilgpt2""": 1024,
}




class 							lowerCAmelCase__				(			UpperCAmelCase__ ):






      '''simple docstring'''

      __UpperCamelCase	 =  VOCAB_FILES_NAMES
      __UpperCamelCase	 =  PRETRAINED_VOCAB_FILES_MAP
      __UpperCamelCase	 =  PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      __UpperCamelCase	 =  ["input_ids", "attention_mask"]
      __UpperCamelCase	 =  GPTaTokenizer
      def __init__(						self     :  Optional[int]						,	lowercase_     :  int=None						,	lowercase_     :  List[str]=None						,	lowercase_     :  Union[str, Any]=None						,	lowercase_     :  Tuple="<|endoftext|>"						,	lowercase_     :  str="<|endoftext|>"						,	lowercase_     :  Dict="<|endoftext|>"						,	lowercase_     :  Tuple=False						,	**lowercase_     :  Optional[int]						,	):




          '''simple docstring'''


          super().__init__(
              lowercase_						,	lowercase_						,	tokenizer_file=lowercase_						,	unk_token=lowercase_						,	bos_token=lowercase_						,	eos_token=lowercase_						,	add_prefix_space=lowercase_						,	**lowercase_						,	)

          SCREAMING_SNAKE_CASE_      :  Union[str, Any]    						=		kwargs.pop('''add_bos_token'''						,	lowercase_)

          SCREAMING_SNAKE_CASE_      :  Dict    						=		json.loads(self.backend_tokenizer.pre_tokenizer.__getstate__())
          if pre_tok_state.get('''add_prefix_space'''						,	lowercase_) != add_prefix_space:
              SCREAMING_SNAKE_CASE_      :  int    						=		getattr(lowercase_						,	pre_tok_state.pop('''type'''))
              SCREAMING_SNAKE_CASE_      :  str    						=		add_prefix_space
              SCREAMING_SNAKE_CASE_      :  Dict    						=		pre_tok_class(**lowercase_)

          SCREAMING_SNAKE_CASE_      :  Optional[Any]    						=		add_prefix_space
      def _SCREAMING_SNAKE_CASE (						self     :  str						,	*lowercase_     :  List[Any]						,	**lowercase_     :  Optional[Any]):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_      :  List[Any]    						=		kwargs.get('''is_split_into_words'''						,	lowercase_)
          assert self.add_prefix_space or not is_split_into_words, (
              F'You need to instantiate {self.__class__.__name__} with add_prefix_space=True '
              "to use it with pretokenized inputs."
          )

          return super()._batch_encode_plus(*lowercase_						,	**lowercase_)
      def _SCREAMING_SNAKE_CASE (						self     :  Optional[int]						,	*lowercase_     :  List[str]						,	**lowercase_     :  List[Any]):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_      :  List[str]    						=		kwargs.get('''is_split_into_words'''						,	lowercase_)

          assert self.add_prefix_space or not is_split_into_words, (
              F'You need to instantiate {self.__class__.__name__} with add_prefix_space=True '
              "to use it with pretokenized inputs."
          )

          return super()._encode_plus(*lowercase_						,	**lowercase_)
      def _SCREAMING_SNAKE_CASE (						self     :  int						,	lowercase_     :  str						,	lowercase_     :  Optional[str] = None):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_      :  List[str]    						=		self._tokenizer.model.save(lowercase_						,	name=lowercase_)
          return tuple(lowercase_)



      def _SCREAMING_SNAKE_CASE (						self     :  int						,	lowercase_     :  "Conversation"):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_      :  Dict    						=		[]
          for is_user, text in conversation.iter_texts():
              input_ids.extend(self.encode(lowercase_						,	add_special_tokens=lowercase_) + [self.eos_token_id])

          if len(lowercase_) > self.model_max_length:
              SCREAMING_SNAKE_CASE_      :  Any    						=		input_ids[-self.model_max_length :]
          return input_ids


