
'''simple docstring'''





import concurrent.futures
import json
import os
import shutil
import tempfile
import unittest

from transformers import AutoTokenizer, PreTrainedTokenizerFast
from transformers.testing_utils import require_tokenizers

from ..test_tokenization_common import TokenizerTesterMixin



@require_tokenizers
class 			lowerCAmelCase__ (							UpperCamelCase       ,				unittest.TestCase							):



  '''simple docstring'''






  _UpperCamelCase							=			PreTrainedTokenizerFast
  _UpperCamelCase							=			False
  _UpperCamelCase							=			True
  _UpperCamelCase							=			'tokenizer_file'

  def 		_lowerCamelCase						(self	:			Any     ) ->							str:
        snake_case_    							=				False  # because we don't have pretrained_vocab_files_map
        super().setUp()
        snake_case_    							=				True

        snake_case_    							=				["robot-test/dummy-tokenizer-fast", "robot-test/dummy-tokenizer-wordlevel"]
        snake_case_    							=				"SaulLu/dummy-tokenizer-bytelevel-bpe"

        # Inclusion of 2 tokenizers to test different types of models (Unigram and WordLevel for the moment)
        snake_case_    							=				[(PreTrainedTokenizerFast, model_path, {}) for model_path in model_paths]

        snake_case_    							=				PreTrainedTokenizerFast.from_pretrained(model_paths[0]     )
        tokenizer.save_pretrained(self.tmpdirname     )

  def 		_lowerCamelCase						(self	:			Any     ) ->							List[Any]:
        # We disable this test for PreTrainedTokenizerFast because it is the only tokenizer that is not linked to any
        # model
        pass

  def 		_lowerCamelCase						(self	:			Tuple     ) ->							Dict:
        # We disable this test for PreTrainedTokenizerFast because it is the only tokenizer that is not linked to any
        # model
        pass

  def 		_lowerCamelCase						(self	:			Any     ) ->							Dict:
        # We disable this test for PreTrainedTokenizerFast because it is the only tokenizer that is not linked to any
        # model
        pass

  def 		_lowerCamelCase						(self	:			Any     ) ->							Tuple:
        # PreTrainedTokenizerFast doesn't have tokenizer_file in its signature
        pass

  def 		_lowerCamelCase						(self	:			List[str]     ) ->							List[str]:
        snake_case_    							=				self.tmpdirname
        # Here we want to test the 2 available tokenizers that use 2 different types of models: Unigram and WordLevel.
        for tokenizer, pretrained_name, kwargs in self.tokenizers_list:
              with self.subTest(f'''{tokenizer.__class__.__name__} ({pretrained_name})'''     ):
                    try:
                          snake_case_    							=				tempfile.mkdtemp()
                          snake_case_    							=				self.rust_tokenizer_class.from_pretrained(lowerCAmelCase_						,			**lowerCAmelCase_     )

                          tokenizer.save_pretrained(self.tmpdirname     )
                          super().test_training_new_tokenizer()
                    finally:
                          # Even if the test fails, we must be sure that the folder is deleted and that the default tokenizer
                          # is restored
                          shutil.rmtree(self.tmpdirname     )
                          snake_case_    							=				tmpdirname_orig

  def 		_lowerCamelCase						(self	:			Union[str, Any]     ) ->							Dict:
        snake_case_    							=				self.tmpdirname
        # Here we want to test the 2 available tokenizers that use 2 different types of models: Unigram and WordLevel.
        for tokenizer, pretrained_name, kwargs in self.tokenizers_list:
              with self.subTest(f'''{tokenizer.__class__.__name__} ({pretrained_name})'''     ):
                    try:
                          snake_case_    							=				tempfile.mkdtemp()
                          snake_case_    							=				self.rust_tokenizer_class.from_pretrained(lowerCAmelCase_						,			**lowerCAmelCase_     )

                          tokenizer.save_pretrained(self.tmpdirname     )
                          super().test_training_new_tokenizer_with_special_tokens_change()
                    finally:
                          # Even if the test fails, we must be sure that the folder is deleted and that the default tokenizer
                          # is restored
                          shutil.rmtree(self.tmpdirname     )
                          snake_case_    							=				tmpdirname_orig


  def 		_lowerCamelCase						(self	:			Union[str, Any]     ) ->							Union[str, Any]:
        snake_case_    							=				self.rust_tokenizer_class.from_pretrained(self.bytelevel_bpe_model_name     )

        snake_case_    							=				("a" for _ in range(10_00     ))
        snake_case_    							=				tokenizer.train_new_from_iterator(text_iterator=lowerCAmelCase_						,			length=10_00						,			vocab_size=50     )

        snake_case_    							=				new_tokenizer.encode("a🤗"     )
        self.assertEqual(lowerCAmelCase_						,			[64, 1_72, 2_53, 97, 2_45]     )



@require_tokenizers
class 			lowerCAmelCase__ (							unittest.TestCase							):



  '''simple docstring'''






  def 		_lowerCamelCase						(self	:			Tuple     ) ->							Dict:
        snake_case_    							=				AutoTokenizer.from_pretrained("bert-base-cased"     )
        snake_case_    							=				json.loads(tokenizer._tokenizer.to_str()     )
        snake_case_    							=				len(lowerCAmelCase_     )

        with tempfile.TemporaryDirectory() as tmp_dir:
              # Hack to save this in the tokenizer_config.json
              snake_case_    							=				["tokenizer.4.0.0.json"]
              tokenizer.save_pretrained(lowerCAmelCase_     )
              json.dump(lowerCAmelCase_						,			open(os.path.join(lowerCAmelCase_						,			"tokenizer.4.0.0.json"     )						,			"w"     )     )

              # This should pick the new tokenizer file as the version of Transformers is > 4.0.0
              snake_case_    							=				AutoTokenizer.from_pretrained(lowerCAmelCase_     )
              self.assertEqual(len(lowerCAmelCase_     )						,			len(lowerCAmelCase_     ) + 1     )
              snake_case_    							=				json.loads(new_tokenizer._tokenizer.to_str()     )
              self.assertIn("huggingface"						,			json_tokenizer["model"]["vocab"]     )

              # Will need to be adjusted if we reach v42 and this test is still here.
              # Should pick the old tokenizer file as the version of Transformers is < 4.0.0
              shutil.move(os.path.join(lowerCAmelCase_						,			"tokenizer.4.0.0.json"     )						,			os.path.join(lowerCAmelCase_						,			"tokenizer.42.0.0.json"     )     )
              snake_case_    							=				["tokenizer.42.0.0.json"]
              tokenizer.save_pretrained(lowerCAmelCase_     )
              snake_case_    							=				AutoTokenizer.from_pretrained(lowerCAmelCase_     )
              self.assertEqual(len(lowerCAmelCase_     )						,			len(lowerCAmelCase_     )     )
              snake_case_    							=				json.loads(new_tokenizer._tokenizer.to_str()     )
              self.assertNotIn("huggingface"						,			json_tokenizer["model"]["vocab"]     )


  def 		_lowerCamelCase						(self	:			List[str]     ) ->							List[Any]:
        # This repo has two tokenizer files, one for v4.0.0 and above with an added token, one for versions lower.
        snake_case_    							=				"hf-internal-testing/test-two-tokenizers"

        # This should pick the new tokenizer file as the version of Transformers is > 4.0.0
        snake_case_    							=				AutoTokenizer.from_pretrained(lowerCAmelCase_     )
        self.assertEqual(len(lowerCAmelCase_     )						,			2_89_97     )
        snake_case_    							=				json.loads(tokenizer._tokenizer.to_str()     )
        self.assertIn("huggingface"						,			json_tokenizer["model"]["vocab"]     )

        # Testing an older version by monkey-patching the version in the module it's used.
        import transformers as old_transformers

        snake_case_    							=				"3.0.0"
        snake_case_    							=				old_transformers.models.auto.AutoTokenizer.from_pretrained(lowerCAmelCase_     )
        self.assertEqual(len(lowerCAmelCase_     )						,			2_89_96     )
        snake_case_    							=				json.loads(old_tokenizer._tokenizer.to_str()     )
        self.assertNotIn("huggingface"						,			json_tokenizer["model"]["vocab"]     )



@require_tokenizers
class 			lowerCAmelCase__ (							unittest.TestCase							):



  '''simple docstring'''






  def 		_lowerCamelCase						(self	:			List[Any]     ) ->							List[str]:
        # See https://github.com/huggingface/transformers/pull/12550
        # and https://github.com/huggingface/tokenizers/issues/537
        snake_case_    							=				PreTrainedTokenizerFast.from_pretrained("robot-test/dummy-tokenizer-wordlevel"     )
        snake_case_    							=				"The Matrix is a 1999 science fiction action film."

        with concurrent.futures.ThreadPoolExecutor() as executor:
              snake_case_    							=				[executor.submit(self.fetch						,			lowerCAmelCase_						,			lowerCAmelCase_     ) for i in range(10     )]
              snake_case_    							=				[future.result() for future in futures]
              self.assertEqual(lowerCAmelCase_						,			[[1, 10, 0, 8, 0, 18, 0, 0, 0, 2] for i in range(10     )]     )


  def 		_lowerCamelCase						(self	:			Dict						,			lowerCAmelCase_	:			Optional[int]						,			lowerCAmelCase_	:			Optional[Any]     ) ->							Union[str, Any]:
        return tokenizer.encode(lowerCAmelCase_						,			truncation="longest_first"						,			padding="longest"     )


"""simple docstring"""







from dataclasses import dataclass
from typing import Any, Dict, Optional

import torch
import torch.nn.functional as F
from torch import nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..models.embeddings import ImagePositionalEmbeddings
from ..utils import BaseOutput, deprecate
from .attention import BasicTransformerBlock
from .embeddings import PatchEmbed
from .modeling_utils import ModelMixin



@dataclass
class 			_lowerCamelCase      (__lowerCamelCase  ):
				_snake_case  =	42



class 			_lowerCamelCase      (__lowerCamelCase    ,				__lowerCamelCase  ):







				@register_to_config
				def __init__(			self     :							Dict      ,       lowerCamelCase_     :							int = 1_6      ,       lowerCamelCase_     :							int = 8_8      ,       lowerCamelCase_     :							Optional[int] = None      ,       lowerCamelCase_     :							Optional[int] = None      ,       lowerCamelCase_     :							int = 1      ,       lowerCamelCase_     :							float = 0.0      ,       lowerCamelCase_     :							int = 3_2      ,       lowerCamelCase_     :							Optional[int] = None      ,       lowerCamelCase_     :							bool = False      ,       lowerCamelCase_     :							Optional[int] = None      ,       lowerCamelCase_     :							Optional[int] = None      ,       lowerCamelCase_     :							Optional[int] = None      ,       lowerCamelCase_     :							str = "geglu"      ,       lowerCamelCase_     :							Optional[int] = None      ,       lowerCamelCase_     :							bool = False      ,       lowerCamelCase_     :							bool = False      ,       lowerCamelCase_     :							bool = False      ,       lowerCamelCase_     :							str = "layer_norm"      ,       lowerCamelCase_     :							bool = True      ,       ):

					"""simple docstring"""






					super().__init__()
					_lowercase			:     Tuple      					=				use_linear_projection
					_lowercase			:     List[Any]      					=				num_attention_heads
					_lowercase			:     Optional[int]      					=				attention_head_dim
					_lowercase			:     Union[str, Any]      					=				num_attention_heads * attention_head_dim

					# 1. Transformer2DModel can process both standard continuous images of shape `(batch_size, num_channels, width, height)` as well as quantized image embeddings of shape `(batch_size, num_image_vectors)`
					# Define whether input is continuous or discrete depending on configuration
					_lowercase			:     Dict      					=				(in_channels is not None) and (patch_size is None)
					_lowercase			:     Any      					=				num_vector_embeds is not None
					_lowercase			:     List[Any]      					=				in_channels is not None and patch_size is not None

					if norm_type == "layer_norm" and num_embeds_ada_norm is not None:
						_lowercase			:     Tuple      					=				(
						    F'''The configuration file of this model: {self.__class__} is outdated. `norm_type` is either not set or'''
						    ' incorrectly set to `\'layer_norm\'`.Make sure to set `norm_type` to `\'ada_norm\'` in the config.'
						    ' Please make sure to update the config accordingly as leaving `norm_type` might led to incorrect'
						    ' results in future versions. If you have downloaded this checkpoint from the Hugging Face Hub, it'
						    ' would be very nice if you could open a Pull request for the `transformer/config.json` file'
						)
						deprecate('norm_type!=num_embeds_ada_norm'      ,       '1.0.0'      ,       lowerCamelCase_      ,       standard_warn=lowerCamelCase_    )
						_lowercase			:     Dict      					=				'ada_norm'

					if self.is_input_continuous and self.is_input_vectorized:
						raise ValueError(
						    F'''Cannot define both `in_channels`: {in_channels} and `num_vector_embeds`: {num_vector_embeds}. Make'''
						    ' sure that either `in_channels` or `num_vector_embeds` is None.'    )
					elif self.is_input_vectorized and self.is_input_patches:
						raise ValueError(
						    F'''Cannot define both `num_vector_embeds`: {num_vector_embeds} and `patch_size`: {patch_size}. Make'''
						    ' sure that either `num_vector_embeds` or `num_patches` is None.'    )
					elif not self.is_input_continuous and not self.is_input_vectorized and not self.is_input_patches:
						raise ValueError(
						    F'''Has to define `in_channels`: {in_channels}, `num_vector_embeds`: {num_vector_embeds}, or patch_size:'''
						    F''' {patch_size}. Make sure that `in_channels`, `num_vector_embeds` or `num_patches` is not None.'''    )

					# 2. Define input layers
					if self.is_input_continuous:
						_lowercase			:     int      					=				in_channels

						_lowercase			:     List[Any]      					=				torch.nn.GroupNorm(num_groups=lowerCamelCase_      ,       num_channels=lowerCamelCase_      ,       eps=1E-6      ,       affine=lowerCamelCase_    )
						if use_linear_projection:
							_lowercase			:     Any      					=				nn.Linear(lowerCamelCase_      ,       lowerCamelCase_    )
						else:
							_lowercase			:     List[str]      					=				nn.Convad(lowerCamelCase_      ,       lowerCamelCase_      ,       kernel_size=1      ,       stride=1      ,       padding=0    )
					elif self.is_input_vectorized:
						assert sample_size is not None, "Transformer2DModel over discrete input must provide sample_size"
						assert num_vector_embeds is not None, "Transformer2DModel over discrete input must provide num_embed"

						_lowercase			:     int      					=				sample_size
						_lowercase			:     str      					=				sample_size
						_lowercase			:     int      					=				num_vector_embeds
						_lowercase			:     List[str]      					=				self.height * self.width

						_lowercase			:     Optional[int]      					=				ImagePositionalEmbeddings(
						    num_embed=lowerCamelCase_      ,       embed_dim=lowerCamelCase_      ,       height=self.height      ,       width=self.width    )
					elif self.is_input_patches:
						assert sample_size is not None, "Transformer2DModel over patched input must provide sample_size"

						_lowercase			:     Optional[Any]      					=				sample_size
						_lowercase			:     Optional[int]      					=				sample_size

						_lowercase			:     str      					=				patch_size
						_lowercase			:     Union[str, Any]      					=				PatchEmbed(
						    height=lowerCamelCase_      ,       width=lowerCamelCase_      ,       patch_size=lowerCamelCase_      ,       in_channels=lowerCamelCase_      ,       embed_dim=lowerCamelCase_      ,       )

					# 3. Define transformers blocks
					_lowercase			:     Any      					=				nn.ModuleList(
					    [
					        BasicTransformerBlock(
					            lowerCamelCase_      ,       lowerCamelCase_      ,       lowerCamelCase_      ,       dropout=lowerCamelCase_      ,       cross_attention_dim=lowerCamelCase_      ,       activation_fn=lowerCamelCase_      ,       num_embeds_ada_norm=lowerCamelCase_      ,       attention_bias=lowerCamelCase_      ,       only_cross_attention=lowerCamelCase_      ,       upcast_attention=lowerCamelCase_      ,       norm_type=lowerCamelCase_      ,       norm_elementwise_affine=lowerCamelCase_      ,       )
					        for d in range(lowerCamelCase_    )
					    ]    )

					# 4. Define output layers
					_lowercase			:     Dict      					=				in_channels if out_channels is None else out_channels
					if self.is_input_continuous:
						# TODO: should use out_channels for continuous projections
						if use_linear_projection:
							_lowercase			:     str      					=				nn.Linear(lowerCamelCase_      ,       lowerCamelCase_    )
						else:
							_lowercase			:     List[Any]      					=				nn.Convad(lowerCamelCase_      ,       lowerCamelCase_      ,       kernel_size=1      ,       stride=1      ,       padding=0    )
					elif self.is_input_vectorized:
						_lowercase			:     Dict      					=				nn.LayerNorm(lowerCamelCase_    )
						_lowercase			:     Union[str, Any]      					=				nn.Linear(lowerCamelCase_      ,       self.num_vector_embeds - 1    )
					elif self.is_input_patches:
						_lowercase			:     Dict      					=				nn.LayerNorm(lowerCamelCase_      ,       elementwise_affine=lowerCamelCase_      ,       eps=1E-6    )
						_lowercase			:     Union[str, Any]      					=				nn.Linear(lowerCamelCase_      ,       2 * inner_dim    )
						_lowercase			:     List[Any]      					=				nn.Linear(lowerCamelCase_      ,       patch_size * patch_size * self.out_channels    )







				def 				__UpperCAmelCase       (			self     :							Optional[Any]      ,       lowerCamelCase_     :							torch.Tensor      ,       lowerCamelCase_     :							Optional[torch.Tensor] = None      ,       lowerCamelCase_     :							Optional[torch.LongTensor] = None      ,       lowerCamelCase_     :							Optional[torch.LongTensor] = None      ,       lowerCamelCase_     :							Dict[str, Any] = None      ,       lowerCamelCase_     :							Optional[torch.Tensor] = None      ,       lowerCamelCase_     :							Optional[torch.Tensor] = None      ,       lowerCamelCase_     :							bool = True      ,       ):

					"""simple docstring"""






					if attention_mask is not None and attention_mask.ndim == 2:
						# assume that mask is expressed as:
						#   (1 = keep,      0 = discard)
						# convert mask into a bias that can be added to attention scores:
						#       (keep = +0,     discard = -10000.0)
						_lowercase			:     Optional[Any]      					=				(1 - attention_mask.to(hidden_states.dtype    )) * -1_0000.0
						_lowercase			:     Optional[Any]      					=				attention_mask.unsqueeze(1    )

					# convert encoder_attention_mask to a bias the same way we do for attention_mask
					if encoder_attention_mask is not None and encoder_attention_mask.ndim == 2:
						_lowercase			:     List[Any]      					=				(1 - encoder_attention_mask.to(hidden_states.dtype    )) * -1_0000.0
						_lowercase			:     Any      					=				encoder_attention_mask.unsqueeze(1    )

					# 1. Input
					if self.is_input_continuous:
						_lowercase					,   _lowercase					,   _lowercase					,   _lowercase			:     List[Any]      					=				hidden_states.shape
						_lowercase			:     int      					=				hidden_states

						_lowercase			:     Any      					=				self.norm(lowerCamelCase_    )
						if not self.use_linear_projection:
							_lowercase			:     Any      					=				self.proj_in(lowerCamelCase_    )
							_lowercase			:     Any      					=				hidden_states.shape[1]
							_lowercase			:     List[Any]      					=				hidden_states.permute(0      ,       2      ,       3      ,       1    ).reshape(lowerCamelCase_      ,       height * width      ,       lowerCamelCase_    )
						else:
							_lowercase			:     Any      					=				hidden_states.shape[1]
							_lowercase			:     int      					=				hidden_states.permute(0      ,       2      ,       3      ,       1    ).reshape(lowerCamelCase_      ,       height * width      ,       lowerCamelCase_    )
							_lowercase			:     Any      					=				self.proj_in(lowerCamelCase_    )
					elif self.is_input_vectorized:
						_lowercase			:     Optional[Any]      					=				self.latent_image_embedding(lowerCamelCase_    )
					elif self.is_input_patches:
						_lowercase			:     Union[str, Any]      					=				self.pos_embed(lowerCamelCase_    )

					# 2. Blocks
					for block in self.transformer_blocks:
						_lowercase			:     Tuple      					=				block(
						    lowerCamelCase_      ,       attention_mask=lowerCamelCase_      ,       encoder_hidden_states=lowerCamelCase_      ,       encoder_attention_mask=lowerCamelCase_      ,       timestep=lowerCamelCase_      ,       cross_attention_kwargs=lowerCamelCase_      ,       class_labels=lowerCamelCase_      ,       )

					# 3. Output
					if self.is_input_continuous:
						if not self.use_linear_projection:
							_lowercase			:     Union[str, Any]      					=				hidden_states.reshape(lowerCamelCase_      ,       lowerCamelCase_      ,       lowerCamelCase_      ,       lowerCamelCase_    ).permute(0      ,       3      ,       1      ,       2    ).contiguous()
							_lowercase			:     Tuple      					=				self.proj_out(lowerCamelCase_    )
						else:
							_lowercase			:     int      					=				self.proj_out(lowerCamelCase_    )
							_lowercase			:     List[Any]      					=				hidden_states.reshape(lowerCamelCase_      ,       lowerCamelCase_      ,       lowerCamelCase_      ,       lowerCamelCase_    ).permute(0      ,       3      ,       1      ,       2    ).contiguous()

						_lowercase			:     List[str]      					=				hidden_states + residual
					elif self.is_input_vectorized:
						_lowercase			:     Dict      					=				self.norm_out(lowerCamelCase_    )
						_lowercase			:     Dict      					=				self.out(lowerCamelCase_    )
						# (batch, self.num_vector_embeds - 1, self.num_latent_pixels)
						_lowercase			:     Dict      					=				logits.permute(0      ,       2      ,       1    )

						# log(p(x_0))
						_lowercase			:     int      					=				F.log_softmax(logits.double()      ,       dim=1    ).float()
					elif self.is_input_patches:
						# TODO: cleanup!
						_lowercase			:     Tuple      					=				self.transformer_blocks[0].norma.emb(
						    lowerCamelCase_      ,       lowerCamelCase_      ,       hidden_dtype=hidden_states.dtype    )
						_lowercase					,   _lowercase			:     List[Any]      					=				self.proj_out_a(F.silu(lowerCamelCase_    )    ).chunk(2      ,       dim=1    )
						_lowercase			:     Optional[int]      					=				self.norm_out(lowerCamelCase_    ) * (1 + scale[:, None]) + shift[:, None]
						_lowercase			:     Optional[Any]      					=				self.proj_out_a(lowerCamelCase_    )

						# unpatchify
						_lowercase			:     Union[str, Any]      					=				int(hidden_states.shape[1] ** 0.5    )
						_lowercase			:     int      					=				hidden_states.reshape(
						    shape=(-1, height, width, self.patch_size, self.patch_size, self.out_channels)    )
						_lowercase			:     str      					=				torch.einsum('nhwpqc->nchpwq'      ,       lowerCamelCase_    )
						_lowercase			:     List[str]      					=				hidden_states.reshape(
						    shape=(-1, self.out_channels, height * self.patch_size, width * self.patch_size)    )

					if not return_dict:
						return (output,)

					return TransformeraDModelOutput(sample=lowerCamelCase_    )



