'''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


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




class    a__				(       UpperCAmelCase__							,       UpperCAmelCase__			):






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





         """simple docstring"""



         super().__init__()

         __lowerCamelCase          =  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"

             __lowerCamelCase          =  torch.zeros(a  ,    a   )
         else:
             __lowerCamelCase          =  None

         __lowerCamelCase          =  torch.nn.Parameter(a   )




class    a__				(       UpperCAmelCase__			):
     lowerCamelCase  :					VQModel
     lowerCamelCase  :					CLIPTextModel
     lowerCamelCase  :					CLIPTokenizer
     lowerCamelCase  :					TransformeraDModel
     lowerCamelCase  :					LearnedClassifierFreeSamplingEmbeddings
     lowerCamelCase  :					VQDiffusionScheduler






     def __init__(			self		:		List[str]  ,    a		:		VQModel  ,    a		:		CLIPTextModel  ,    a		:		CLIPTokenizer  ,    a		:		TransformeraDModel  ,    a		:		VQDiffusionScheduler  ,    a		:		LearnedClassifierFreeSamplingEmbeddings  ,    ):





         """simple docstring"""



         super().__init__()

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






     def        SCREAMING_SNAKE_CASE__       (			self		:		Any  ,    a		:		List[str]  ,    a		:		Union[str, Any]  ,    a		:		List[str]   ):





         """simple docstring"""



         __lowerCamelCase          =  len(a   ) if isinstance(a  ,    a   ) else 1

         # get prompt text embeddings
         __lowerCamelCase          =  self.tokenizer(
             a  ,    padding='''max_length'''  ,    max_length=self.tokenizer.model_max_length  ,    return_tensors='''pt'''  ,    )
         __lowerCamelCase          =  text_inputs.input_ids

         if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
             __lowerCamelCase          =  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}"""   )
             __lowerCamelCase          =  text_input_ids[:, : self.tokenizer.model_max_length]
         __lowerCamelCase          =  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
         __lowerCamelCase          =  prompt_embeds / prompt_embeds.norm(dim=-1  ,    keepdim=a   )

         # duplicate text embeddings for each generation per prompt
         __lowerCamelCase          =  prompt_embeds.repeat_interleave(a  ,    dim=0   )

         if do_classifier_free_guidance:
             if self.learned_classifier_free_sampling_embeddings.learnable:
                 __lowerCamelCase          =  self.learned_classifier_free_sampling_embeddings.embeddings
                 __lowerCamelCase          =  negative_prompt_embeds.unsqueeze(0   ).repeat(a  ,    1  ,    1   )
             else:
                 __lowerCamelCase          =  [''''''] * batch_size

                 __lowerCamelCase          =  text_input_ids.shape[-1]
                 __lowerCamelCase          =  self.tokenizer(
                     a  ,    padding='''max_length'''  ,    max_length=a  ,    truncation=a  ,    return_tensors='''pt'''  ,    )
                 __lowerCamelCase          =  self.text_encoder(uncond_input.input_ids.to(self.device   )   )[0]
                 # See comment for normalizing text embeddings
                 __lowerCamelCase          =  negative_prompt_embeds / negative_prompt_embeds.norm(dim=-1  ,    keepdim=a   )

             # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
             __lowerCamelCase          =  negative_prompt_embeds.shape[1]
             __lowerCamelCase          =  negative_prompt_embeds.repeat(1  ,    a  ,    1   )
             __lowerCamelCase          =  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
             __lowerCamelCase          =  torch.cat([negative_prompt_embeds, prompt_embeds]   )

         return prompt_embeds






     @torch.no_grad()
     def __call__(			self		:		Optional[int]  ,    a		:		Union[str, List[str]]  ,    a		:		int = 1_00  ,    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  ,    ):





         """simple docstring"""



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

         __lowerCamelCase          =  batch_size * num_images_per_prompt

         __lowerCamelCase          =  guidance_scale > 1.0

         __lowerCamelCase          =  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

         __lowerCamelCase          =  (batch_size, self.transformer.num_latent_pixels)
         if latents is None:
             __lowerCamelCase          =  self.transformer.num_vector_embeds - 1
             __lowerCamelCase          =  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)."""   )
             __lowerCamelCase          =  latents.to(self.device   )

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

         __lowerCamelCase          =  self.scheduler.timesteps.to(self.device   )

         __lowerCamelCase          =  latents

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

             # predict the un-noised image
             # model_output == `log_p_x_0`
             __lowerCamelCase          =  self.transformer(a  ,    encoder_hidden_states=a  ,    timestep=a   ).sample

             if do_classifier_free_guidance:
                 __lowerCamelCase     ,  __lowerCamelCase          =  model_output.chunk(2   )
                 __lowerCamelCase          =  model_output_uncond + guidance_scale * (model_output_text - model_output_uncond)
                 model_output -= torch.logsumexp(a  ,    dim=1  ,    keepdim=a   )

             __lowerCamelCase          =  self.truncate(a  ,    a   )

             # remove `log(0)`'s (`-inf`s)
             __lowerCamelCase          =  model_output.clamp(-70   )

             # compute the previous noisy sample x_t -> x_t-1
             __lowerCamelCase          =  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   )

         __lowerCamelCase          =  self.vqvae.config.vq_embed_dim
         __lowerCamelCase          =  (batch_size, self.transformer.height, self.transformer.width, embedding_channels)
         __lowerCamelCase          =  self.vqvae.quantize.get_codebook_entry(a  ,    shape=a   )
         __lowerCamelCase          =  self.vqvae.decode(a  ,    force_not_quantize=a   ).sample

         __lowerCamelCase          =  (image / 2 + 0.5).clamp(0  ,    1   )
         __lowerCamelCase          =  image.cpu().permute(0  ,    2  ,    3  ,    1   ).numpy()

         if output_type == "pil":
             __lowerCamelCase          =  self.numpy_to_pil(a   )

         if not return_dict:
             return (image,)

         return ImagePipelineOutput(images=a   )






     def        SCREAMING_SNAKE_CASE__       (			self		:		Union[str, Any]  ,    a		:		torch.FloatTensor  ,    a		:		float   ):





         """simple docstring"""



         __lowerCamelCase     ,  __lowerCamelCase          =  torch.sort(a  ,    1  ,    descending=a   )
         __lowerCamelCase          =  torch.exp(a   )
         __lowerCamelCase          =  sorted_p_x_0.cumsum(dim=1   ) < truncation_rate

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

         __lowerCamelCase          =  keep_mask.gather(1  ,    indices.argsort(1   )   )

         __lowerCamelCase          =  log_p_x_0.clone()

         __lowerCamelCase          =  -torch.inf  # -inf = log(0)

         return rv

'''simple docstring'''




import os
import shutil
import tempfile
from unittest import TestCase
from unittest.mock import patch

import numpy as np
from datasets import Dataset

from transformers.models.realm.configuration_realm import RealmConfig
from transformers.models.realm.retrieval_realm import _REALM_BLOCK_RECORDS_FILENAME, RealmRetriever
from transformers.models.realm.tokenization_realm import VOCAB_FILES_NAMES, RealmTokenizer




class    a__				(       UpperCAmelCase__			):






     def        SCREAMING_SNAKE_CASE__       (			self		:		Optional[Any]   ):





         """simple docstring"""



         __lowerCamelCase          =  tempfile.mkdtemp()
         __lowerCamelCase          =  5

         # Realm tok
         __lowerCamelCase          =  [
             '''[UNK]''',
             '''[CLS]''',
             '''[SEP]''',
             '''[PAD]''',
             '''[MASK]''',
             '''test''',
             '''question''',
             '''this''',
             '''is''',
             '''the''',
             '''first''',
             '''second''',
             '''third''',
             '''fourth''',
             '''fifth''',
             '''record''',
             '''want''',
             '''##want''',
             '''##ed''',
             '''wa''',
             '''un''',
             '''runn''',
             '''##ing''',
             ''',''',
             '''low''',
             '''lowest''',
         ]
         __lowerCamelCase          =  os.path.join(self.tmpdirname  ,    '''realm_tokenizer'''   )
         os.makedirs(a  ,    exist_ok=a   )
         __lowerCamelCase          =  os.path.join(a  ,    VOCAB_FILES_NAMES['''vocab_file''']   )
         with open(self.vocab_file  ,    '''w'''  ,    encoding='''utf-8'''   ) as vocab_writer:
             vocab_writer.write(''''''.join([x + '''\n''' for x in vocab_tokens]   )   )

         __lowerCamelCase          =  os.path.join(self.tmpdirname  ,    '''realm_block_records'''   )
         os.makedirs(a  ,    exist_ok=a   )






     def        SCREAMING_SNAKE_CASE__       (			self		:		Optional[Any]   ):





         """simple docstring"""



         return RealmTokenizer.from_pretrained(os.path.join(self.tmpdirname  ,    '''realm_tokenizer'''   )   )






     def        SCREAMING_SNAKE_CASE__       (			self		:		Any   ):





         """simple docstring"""



         shutil.rmtree(self.tmpdirname   )






     def        SCREAMING_SNAKE_CASE__       (			self		:		Optional[Any]   ):





         """simple docstring"""



         __lowerCamelCase          =  RealmConfig(num_block_records=self.num_block_records   )
         return config






     def        SCREAMING_SNAKE_CASE__       (			self		:		int   ):





         """simple docstring"""



         __lowerCamelCase          =  Dataset.from_dict(
             {
                 '''id''': ['''0''', '''1'''],
                 '''question''': ['''foo''', '''bar'''],
                 '''answers''': [['''Foo''', '''Bar'''], ['''Bar''']],
             }   )
         return dataset






     def        SCREAMING_SNAKE_CASE__       (			self		:		Optional[Any]   ):





         """simple docstring"""



         __lowerCamelCase          =  np.array(
             [
                 b'''This is the first record''',
                 b'''This is the second record''',
                 b'''This is the third record''',
                 b'''This is the fourth record''',
                 b'''This is the fifth record''',
                 b'''This is a longer longer longer record''',
             ]  ,    dtype=a  ,    )
         return block_records






     def        SCREAMING_SNAKE_CASE__       (			self		:		str   ):





         """simple docstring"""



         __lowerCamelCase          =  RealmRetriever(
             block_records=self.get_dummy_block_records()  ,    tokenizer=self.get_tokenizer()  ,    )
         return retriever






     def        SCREAMING_SNAKE_CASE__       (			self		:		Any   ):





         """simple docstring"""



         __lowerCamelCase          =  self.get_config()
         __lowerCamelCase          =  self.get_dummy_retriever()
         __lowerCamelCase          =  retriever.tokenizer

         __lowerCamelCase          =  np.array([0, 3]  ,    dtype='''long'''   )
         __lowerCamelCase          =  tokenizer(['''Test question''']   ).input_ids
         __lowerCamelCase          =  tokenizer(
             ['''the fourth''']  ,    add_special_tokens=a  ,    return_token_type_ids=a  ,    return_attention_mask=a  ,    ).input_ids
         __lowerCamelCase          =  config.reader_seq_len

         __lowerCamelCase     ,  __lowerCamelCase     ,  __lowerCamelCase     ,  __lowerCamelCase          =  retriever(
             a  ,    a  ,    answer_ids=a  ,    max_length=a  ,    return_tensors='''np'''   )

         self.assertEqual(len(a   )  ,    2   )
         self.assertEqual(len(a   )  ,    2   )
         self.assertEqual(len(a   )  ,    2   )
         self.assertEqual(concat_inputs.input_ids.shape  ,    (2, 10)   )
         self.assertEqual(concat_inputs.attention_mask.shape  ,    (2, 10)   )
         self.assertEqual(concat_inputs.token_type_ids.shape  ,    (2, 10)   )
         self.assertEqual(concat_inputs.special_tokens_mask.shape  ,    (2, 10)   )
         self.assertEqual(
             tokenizer.convert_ids_to_tokens(concat_inputs.input_ids[0]   )  ,    ['''[CLS]''', '''test''', '''question''', '''[SEP]''', '''this''', '''is''', '''the''', '''first''', '''record''', '''[SEP]''']  ,    )
         self.assertEqual(
             tokenizer.convert_ids_to_tokens(concat_inputs.input_ids[1]   )  ,    ['''[CLS]''', '''test''', '''question''', '''[SEP]''', '''this''', '''is''', '''the''', '''fourth''', '''record''', '''[SEP]''']  ,    )






     def        SCREAMING_SNAKE_CASE__       (			self		:		Optional[Any]   ):





         """simple docstring"""



         __lowerCamelCase          =  self.get_config()
         __lowerCamelCase          =  self.get_dummy_retriever()
         __lowerCamelCase          =  retriever.tokenizer

         __lowerCamelCase          =  np.array([0, 3, 5]  ,    dtype='''long'''   )
         __lowerCamelCase          =  tokenizer(['''Test question''']   ).input_ids
         __lowerCamelCase          =  tokenizer(
             ['''the fourth''', '''longer longer''']  ,    add_special_tokens=a  ,    return_token_type_ids=a  ,    return_attention_mask=a  ,    ).input_ids
         __lowerCamelCase          =  config.reader_seq_len

         __lowerCamelCase     ,  __lowerCamelCase     ,  __lowerCamelCase     ,  __lowerCamelCase          =  retriever(
             a  ,    a  ,    answer_ids=a  ,    max_length=a  ,    return_tensors='''np'''   )

         self.assertEqual([False, True, True]  ,    a   )
         self.assertEqual([[-1, -1, -1], [6, -1, -1], [6, 7, 8]]  ,    a   )
         self.assertEqual([[-1, -1, -1], [7, -1, -1], [7, 8, 9]]  ,    a   )






     def        SCREAMING_SNAKE_CASE__       (			self		:		int   ):





         """simple docstring"""



         __lowerCamelCase          =  self.get_dummy_retriever()
         retriever.save_pretrained(os.path.join(self.tmpdirname  ,    '''realm_block_records'''   )   )

         # Test local path
         __lowerCamelCase          =  retriever.from_pretrained(os.path.join(self.tmpdirname  ,    '''realm_block_records'''   )   )
         self.assertEqual(retriever.block_records[0]  ,    b'''This is the first record'''   )

         # Test mocked remote path
         with patch('''transformers.models.realm.retrieval_realm.hf_hub_download'''   ) as mock_hf_hub_download:
             __lowerCamelCase          =  os.path.join(
                 os.path.join(self.tmpdirname  ,    '''realm_block_records'''   )  ,    _REALM_BLOCK_RECORDS_FILENAME   )
             __lowerCamelCase          =  RealmRetriever.from_pretrained('''google/realm-cc-news-pretrained-openqa'''   )

         self.assertEqual(retriever.block_records[0]  ,    b'''This is the first record'''   )
