



"""simple docstring"""




from typing import Callable, List, Optional, Tuple, Union

import torch
from transformers import CLIPTextModel, CLIPTokenizer

from ...configuration_utils import ConfigMixin, register_to_config
from ...models import ModelMixin, TransformeraDModel, VQModel
from ...schedulers import VQDiffusionScheduler
from ...utils import logging
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput


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






class      UpperCAmelCase_  (    _UpperCamelCase	,      _UpperCamelCase						):



      @register_to_config
      def __init__(					self					:     Any					,     A					:     bool					,     A					:     Optional[int] = None					,     A					:     Optional[int] = None						):
           super().__init__()

           _UpperCAmelCase      :     List[str]        =   learnable

           if self.learnable:
                assert hidden_size is not None, "learnable=True requires `hidden_size` to be set"
                assert length is not None, "learnable=True requires `length` to be set"

                _UpperCAmelCase      :     int        =   torch.zeros(A					,     A						)
           else:
                _UpperCAmelCase      :     Optional[Any]        =   None

           _UpperCAmelCase      :     Optional[Any]        =   torch.nn.Parameter(A						)







class      UpperCAmelCase_  (    _UpperCamelCase						):
      __SCREAMING_SNAKE_CASE						:       VQModel
      __SCREAMING_SNAKE_CASE						:       CLIPTextModel
      __SCREAMING_SNAKE_CASE						:       CLIPTokenizer
      __SCREAMING_SNAKE_CASE						:       TransformeraDModel
      __SCREAMING_SNAKE_CASE						:       LearnedClassifierFreeSamplingEmbeddings
      __SCREAMING_SNAKE_CASE						:       VQDiffusionScheduler



      def __init__(					self					:     Any					,     A					:     VQModel					,     A					:     CLIPTextModel					,     A					:     CLIPTokenizer					,     A					:     TransformeraDModel					,     A					:     VQDiffusionScheduler					,     A					:     LearnedClassifierFreeSamplingEmbeddings					,     ):
           super().__init__()

           self.register_modules(
               vqvae=A					,     transformer=A					,     text_encoder=A					,     tokenizer=A					,     scheduler=A					,     learned_classifier_free_sampling_embeddings=A					,     )



      def 						snake_case_ (					self					:     Union[str, Any]					,     A					:     Any					,     A					:     Optional[int]					,     A					:     List[Any]						):
           _UpperCAmelCase      :     int        =   len(A						) if isinstance(A					,     A						) else 1

           # get prompt text embeddings
           _UpperCAmelCase      :     str        =   self.tokenizer(
               A					,     padding="max_length"					,     max_length=self.tokenizer.model_max_length					,     return_tensors="pt"					,     )
           _UpperCAmelCase      :     int        =   text_inputs.input_ids

           if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
                _UpperCAmelCase      :     int        =   self.tokenizer.batch_decode(text_input_ids[:, self.tokenizer.model_max_length :]						)
                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      :     Union[str, Any]        =   text_input_ids[:, : self.tokenizer.model_max_length]
           _UpperCAmelCase      :     int        =   self.text_encoder(text_input_ids.to(self.device						)						)[0]

           # NOTE: This additional step of normalizing the text embeddings is from VQ-Diffusion.
           # While CLIP does normalize the pooled output of the text transformer when combining
           # the image and text embeddings, CLIP does not directly normalize the last hidden state.
           #
           # CLIP normalizing the pooled output.
           # https://github.com/huggingface/transformers/blob/d92e22d1f28324f513f3080e5c47c071a3916721/src/transformers/models/clip/modeling_clip.py#L1052-L1053
           _UpperCAmelCase      :     Tuple        =   prompt_embeds / prompt_embeds.norm(dim=-1					,     keepdim=A						)

           # duplicate text embeddings for each generation per prompt
           _UpperCAmelCase      :     int        =   prompt_embeds.repeat_interleave(A					,     dim=0						)

           if do_classifier_free_guidance:
                if self.learned_classifier_free_sampling_embeddings.learnable:
                     _UpperCAmelCase      :     Optional[int]        =   self.learned_classifier_free_sampling_embeddings.embeddings
                     _UpperCAmelCase      :     int        =   negative_prompt_embeds.unsqueeze(0						).repeat(A					,     1					,     1						)
                else:
                     _UpperCAmelCase      :     Optional[Any]        =   [""] * batch_size

                     _UpperCAmelCase      :     Optional[Any]        =   text_input_ids.shape[-1]
                     _UpperCAmelCase      :     Union[str, Any]        =   self.tokenizer(
                         A					,     padding="max_length"					,     max_length=A					,     truncation=A					,     return_tensors="pt"					,     )
                     _UpperCAmelCase      :     Tuple        =   self.text_encoder(uncond_input.input_ids.to(self.device						)						)[0]
                     # See comment for normalizing text embeddings
                     _UpperCAmelCase      :     Optional[Any]        =   negative_prompt_embeds / negative_prompt_embeds.norm(dim=-1					,     keepdim=A						)

                # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
                _UpperCAmelCase      :     int        =   negative_prompt_embeds.shape[1]
                _UpperCAmelCase      :     Dict        =   negative_prompt_embeds.repeat(1					,     A					,     1						)
                _UpperCAmelCase      :     Union[str, Any]        =   negative_prompt_embeds.view(batch_size * num_images_per_prompt					,     A					,     -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
                _UpperCAmelCase      :     Optional[Any]        =   torch.cat([negative_prompt_embeds, prompt_embeds]						)

           return prompt_embeds



      @torch.no_grad()
      def __call__(					self					:     Tuple					,     A					:     Union[str, List[str]]					,     A					:     int = 1_0_0					,     A					:     float = 5.0					,     A					:     float = 1.0					,     A					:     int = 1					,     A					:     Optional[Union[torch.Generator, List[torch.Generator]]] = None					,     A					:     Optional[torch.FloatTensor] = None					,     A					:     Optional[str] = "pil"					,     A					:     bool = True					,     A					:     Optional[Callable[[int, int, torch.FloatTensor], None]] = None					,     A					:     int = 1					,     ):
           if isinstance(A					,     A						):
                _UpperCAmelCase      :     List[Any]        =   1
           elif isinstance(A					,     A						):
                _UpperCAmelCase      :     Union[str, Any]        =   len(A						)
           else:
                raise ValueError(f'`prompt` has to be of type `str` or `list` but is {type(A						)}'						)

           _UpperCAmelCase      :     Tuple        =   batch_size * num_images_per_prompt

           _UpperCAmelCase      :     Optional[Any]        =   guidance_scale > 1.0

           _UpperCAmelCase      :     Optional[int]        =   self._encode_prompt(A					,     A					,     A						)

           if (callback_steps is None) or (
               callback_steps is not None and (not isinstance(A					,     A						) or callback_steps <= 0)
           ):
                raise ValueError(
                    f'`callback_steps` has to be a positive integer but is {callback_steps} of type'
                    f' {type(A						)}.'						)

           # get the initial completely masked latents unless the user supplied it

           _UpperCAmelCase      :     Optional[Any]        =   (batch_size, self.transformer.num_latent_pixels)
           if latents is None:
                _UpperCAmelCase      :     Union[str, Any]        =   self.transformer.num_vector_embeds - 1
                _UpperCAmelCase      :     Union[str, Any]        =   torch.full(A					,     A						).to(self.device						)
           else:
                if latents.shape != latents_shape:
                     raise ValueError(f'Unexpected latents shape, got {latents.shape}, expected {latents_shape}'						)
                if (latents < 0).any() or (latents >= self.transformer.num_vector_embeds).any():
                     raise ValueError(
                         "Unexpected latents value(s). All latents be valid embedding indices i.e. in the range 0,"
                         f' {self.transformer.num_vector_embeds - 1} (inclusive).'						)
                _UpperCAmelCase      :     Any        =   latents.to(self.device						)

           # set timesteps
           self.scheduler.set_timesteps(A					,     device=self.device						)

           _UpperCAmelCase      :     Optional[Any]        =   self.scheduler.timesteps.to(self.device						)

           _UpperCAmelCase      :     Dict        =   latents

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

                # predict the un-noised image
                # model_output == `log_p_x_0`
                _UpperCAmelCase      :     Optional[int]        =   self.transformer(A					,     encoder_hidden_states=A					,     timestep=A						).sample

                if do_classifier_free_guidance:
                     _UpperCAmelCase    ,	_UpperCAmelCase      :     Union[str, Any]        =   model_output.chunk(2						)
                     _UpperCAmelCase      :     List[Any]        =   model_output_uncond + guidance_scale * (model_output_text - model_output_uncond)
                     model_output -= torch.logsumexp(A					,     dim=1					,     keepdim=A						)

                _UpperCAmelCase      :     Dict        =   self.truncate(A					,     A						)

                # remove `log(0)`'s (`-inf`s)
                _UpperCAmelCase      :     int        =   model_output.clamp(-7_0						)

                # compute the previous noisy sample x_t -> x_t-1
                _UpperCAmelCase      :     Dict        =   self.scheduler.step(A					,     timestep=A					,     sample=A					,     generator=A						).prev_sample

                # call the callback, if provided
                if callback is not None and i % callback_steps == 0:
                     callback(A					,     A					,     A						)

           _UpperCAmelCase      :     int        =   self.vqvae.config.vq_embed_dim
           _UpperCAmelCase      :     Union[str, Any]        =   (batch_size, self.transformer.height, self.transformer.width, embedding_channels)
           _UpperCAmelCase      :     List[Any]        =   self.vqvae.quantize.get_codebook_entry(A					,     shape=A						)
           _UpperCAmelCase      :     Dict        =   self.vqvae.decode(A					,     force_not_quantize=A						).sample

           _UpperCAmelCase      :     Optional[Any]        =   (image / 2 + 0.5).clamp(0					,     1						)
           _UpperCAmelCase      :     Optional[int]        =   image.cpu().permute(0					,     2					,     3					,     1						).numpy()

           if output_type == "pil":
                _UpperCAmelCase      :     Optional[Any]        =   self.numpy_to_pil(A						)

           if not return_dict:
                return (image,)

           return ImagePipelineOutput(images=A						)



      def 						snake_case_ (					self					:     Dict					,     A					:     torch.FloatTensor					,     A					:     float						):
           _UpperCAmelCase    ,	_UpperCAmelCase      :     Optional[int]        =   torch.sort(A					,     1					,     descending=A						)
           _UpperCAmelCase      :     Union[str, Any]        =   torch.exp(A						)
           _UpperCAmelCase      :     Any        =   sorted_p_x_0.cumsum(dim=1						) < truncation_rate

           # Ensure that at least the largest probability is not zeroed out
           _UpperCAmelCase      :     Optional[Any]        =   torch.full_like(keep_mask[:, 0:1, :]					,     A						)
           _UpperCAmelCase      :     Optional[int]        =   torch.cat((all_true, keep_mask)					,     dim=1						)
           _UpperCAmelCase      :     Optional[int]        =   keep_mask[:, :-1, :]

           _UpperCAmelCase      :     List[Any]        =   keep_mask.gather(1					,     indices.argsort(1						)						)

           _UpperCAmelCase      :     Dict        =   log_p_x_0.clone()

           _UpperCAmelCase      :     Optional[Any]        =   -torch.inf  # -inf = log(0)

           return rv





"""simple docstring"""




import json
from typing import List, Optional, Tuple

from tokenizers import normalizers

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from .tokenization_electra import ElectraTokenizer


_lowerCAmelCase    : List[str]							=							{"vocab_file": "vocab.txt", "tokenizer_file": "tokenizer.json"}

_lowerCAmelCase    : int							=							{
    "vocab_file": {
        "google/electra-small-generator": (
            "https://huggingface.co/google/electra-small-generator/resolve/main/vocab.txt"
        ),
        "google/electra-base-generator": "https://huggingface.co/google/electra-base-generator/resolve/main/vocab.txt",
        "google/electra-large-generator": (
            "https://huggingface.co/google/electra-large-generator/resolve/main/vocab.txt"
        ),
        "google/electra-small-discriminator": (
            "https://huggingface.co/google/electra-small-discriminator/resolve/main/vocab.txt"
        ),
        "google/electra-base-discriminator": (
            "https://huggingface.co/google/electra-base-discriminator/resolve/main/vocab.txt"
        ),
        "google/electra-large-discriminator": (
            "https://huggingface.co/google/electra-large-discriminator/resolve/main/vocab.txt"
        ),
    },
    "tokenizer_file": {
        "google/electra-small-generator": (
            "https://huggingface.co/google/electra-small-generator/resolve/main/tokenizer.json"
        ),
        "google/electra-base-generator": (
            "https://huggingface.co/google/electra-base-generator/resolve/main/tokenizer.json"
        ),
        "google/electra-large-generator": (
            "https://huggingface.co/google/electra-large-generator/resolve/main/tokenizer.json"
        ),
        "google/electra-small-discriminator": (
            "https://huggingface.co/google/electra-small-discriminator/resolve/main/tokenizer.json"
        ),
        "google/electra-base-discriminator": (
            "https://huggingface.co/google/electra-base-discriminator/resolve/main/tokenizer.json"
        ),
        "google/electra-large-discriminator": (
            "https://huggingface.co/google/electra-large-discriminator/resolve/main/tokenizer.json"
        ),
    },
}

_lowerCAmelCase    : List[Any]							=							{
    "google/electra-small-generator": 5_12,
    "google/electra-base-generator": 5_12,
    "google/electra-large-generator": 5_12,
    "google/electra-small-discriminator": 5_12,
    "google/electra-base-discriminator": 5_12,
    "google/electra-large-discriminator": 5_12,
}

_lowerCAmelCase    : Optional[Any]							=							{
    "google/electra-small-generator": {"do_lower_case": True},
    "google/electra-base-generator": {"do_lower_case": True},
    "google/electra-large-generator": {"do_lower_case": True},
    "google/electra-small-discriminator": {"do_lower_case": True},
    "google/electra-base-discriminator": {"do_lower_case": True},
    "google/electra-large-discriminator": {"do_lower_case": True},
}






class      UpperCAmelCase_  (    _UpperCamelCase						):
      __SCREAMING_SNAKE_CASE						:       Optional[Any]								=			VOCAB_FILES_NAMES
      __SCREAMING_SNAKE_CASE						:       Any								=			PRETRAINED_VOCAB_FILES_MAP
      __SCREAMING_SNAKE_CASE						:       List[str]								=			PRETRAINED_INIT_CONFIGURATION
      __SCREAMING_SNAKE_CASE						:       List[str]								=			PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      __SCREAMING_SNAKE_CASE						:       Tuple								=			ElectraTokenizer



      def __init__(					self					:     Dict					,     A					:     Dict=None					,     A					:     Optional[int]=None					,     A					:     Dict=True					,     A					:     Optional[Any]="[UNK]"					,     A					:     Any="[SEP]"					,     A					:     str="[PAD]"					,     A					:     Tuple="[CLS]"					,     A					:     Optional[Any]="[MASK]"					,     A					:     Any=True					,     A					:     Tuple=None					,     **A					:     Any					,     ):
           super().__init__(
               A					,     tokenizer_file=A					,     do_lower_case=A					,     unk_token=A					,     sep_token=A					,     pad_token=A					,     cls_token=A					,     mask_token=A					,     tokenize_chinese_chars=A					,     strip_accents=A					,     **A					,     )

           _UpperCAmelCase      :     List[str]        =   json.loads(self.backend_tokenizer.normalizer.__getstate__()						)
           if (
               normalizer_state.get("lowercase"					,     A						) != do_lower_case
               or normalizer_state.get("strip_accents"					,     A						) != strip_accents
               or normalizer_state.get("handle_chinese_chars"					,     A						) != tokenize_chinese_chars
           ):
                _UpperCAmelCase      :     Union[str, Any]        =   getattr(A					,     normalizer_state.pop("type"						)						)
                _UpperCAmelCase      :     Dict        =   do_lower_case
                _UpperCAmelCase      :     Optional[int]        =   strip_accents
                _UpperCAmelCase      :     Any        =   tokenize_chinese_chars
                _UpperCAmelCase      :     Optional[Any]        =   normalizer_class(**A						)

           _UpperCAmelCase      :     int        =   do_lower_case



      def 						snake_case_ (					self					:     Tuple					,     A					:     str					,     A					:     int=None						):
           _UpperCAmelCase      :     Dict        =   [self.cls_token_id] + token_ids_a + [self.sep_token_id]

           if token_ids_a:
                output += token_ids_a + [self.sep_token_id]

           return output



      def 						snake_case_ (					self					:     Any					,     A					:     List[int]					,     A					:     Optional[List[int]] = None						):
           _UpperCAmelCase      :     Any        =   [self.sep_token_id]
           _UpperCAmelCase      :     Dict        =   [self.cls_token_id]
           if token_ids_a is None:
                return len(cls + token_ids_a + sep						) * [0]
           return len(cls + token_ids_a + sep						) * [0] + len(token_ids_a + sep						) * [1]



      def 						snake_case_ (					self					:     Any					,     A					:     str					,     A					:     Optional[str] = None						):
           _UpperCAmelCase      :     List[Any]        =   self._tokenizer.model.save(A					,     name=A						)
           return tuple(A						)
