


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      snake_case__	(   _A    ,							unittest.TestCase	):
				__lowerCAmelCase		=   PreTrainedTokenizerFast
				__lowerCAmelCase		=   False
				__lowerCAmelCase		=   True
				__lowerCAmelCase		=   '''tokenizer_file'''
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										__magic_name__					:				Optional[int]				=   False  # because we don't have pretrained_vocab_files_map
										super().setUp()
										__magic_name__					:				Optional[int]				=   True

										__magic_name__					:				str				=   ['robot-test/dummy-tokenizer-fast', 'robot-test/dummy-tokenizer-wordlevel']
										__magic_name__					:				List[str]				=   'SaulLu/dummy-tokenizer-bytelevel-bpe'

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

										__magic_name__					:				Any				=   PreTrainedTokenizerFast.from_pretrained(model_paths[0]   )
										tokenizer.save_pretrained(self.tmpdirname   )
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										pass
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										pass
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										pass
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										pass
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										__magic_name__					:				Optional[int]				=   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:
																												__magic_name__					:				Tuple				=   tempfile.mkdtemp()
																												__magic_name__					:				int				=   self.rust_tokenizer_class.from_pretrained(_UpperCAmelCase  ,   **_UpperCAmelCase   )

																												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   )
																												__magic_name__					:				int				=   tmpdirname_orig
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										__magic_name__					:				Any				=   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:
																												__magic_name__					:				Union[str, Any]				=   tempfile.mkdtemp()
																												__magic_name__					:				Optional[int]				=   self.rust_tokenizer_class.from_pretrained(_UpperCAmelCase  ,   **_UpperCAmelCase   )

																												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   )
																												__magic_name__					:				List[str]				=   tmpdirname_orig





				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										__magic_name__					:				List[str]				=   self.rust_tokenizer_class.from_pretrained(self.bytelevel_bpe_model_name   )

										__magic_name__					:				int				=   ('a' for _ in range(1000   ))
										__magic_name__					:				Optional[int]				=   tokenizer.train_new_from_iterator(text_iterator=_UpperCAmelCase  ,   length=1000  ,   vocab_size=50   )

										__magic_name__					:				str				=   new_tokenizer.encode('a🤗'   )
										self.assertEqual(_UpperCAmelCase  ,   [64, 172, 253, 97, 245]   )






@require_tokenizers
class      snake_case__	(   unittest.TestCase	):
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										__magic_name__					:				int				=   AutoTokenizer.from_pretrained('bert-base-cased'   )
										__magic_name__					:				Dict				=   json.loads(tokenizer._tokenizer.to_str()   )
										__magic_name__					:				str				=   len(_UpperCAmelCase   )

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

																# This should pick the new tokenizer file as the version of Transformers is > 4.0.0
																__magic_name__					:				str				=   AutoTokenizer.from_pretrained(_UpperCAmelCase   )
																self.assertEqual(len(_UpperCAmelCase   )  ,   len(_UpperCAmelCase   ) + 1   )
																__magic_name__					:				Dict				=   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(_UpperCAmelCase  ,   'tokenizer.4.0.0.json'   )  ,   os.path.join(_UpperCAmelCase  ,   'tokenizer.42.0.0.json'   )   )
																__magic_name__					:				Any				=   ['tokenizer.42.0.0.json']
																tokenizer.save_pretrained(_UpperCAmelCase   )
																__magic_name__					:				Tuple				=   AutoTokenizer.from_pretrained(_UpperCAmelCase   )
																self.assertEqual(len(_UpperCAmelCase   )  ,   len(_UpperCAmelCase   )   )
																__magic_name__					:				List[str]				=   json.loads(new_tokenizer._tokenizer.to_str()   )
																self.assertNotIn('huggingface'  ,   json_tokenizer['model']['vocab']   )





				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										__magic_name__					:				Union[str, Any]				=   'hf-internal-testing/test-two-tokenizers'

										# This should pick the new tokenizer file as the version of Transformers is > 4.0.0
										__magic_name__					:				Optional[int]				=   AutoTokenizer.from_pretrained(_UpperCAmelCase   )
										self.assertEqual(len(_UpperCAmelCase   )  ,   2_8997   )
										__magic_name__					:				Tuple				=   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

										__magic_name__					:				List[str]				=   '3.0.0'
										__magic_name__					:				List[Any]				=   old_transformers.models.auto.AutoTokenizer.from_pretrained(_UpperCAmelCase   )
										self.assertEqual(len(_UpperCAmelCase   )  ,   2_8996   )
										__magic_name__					:				str				=   json.loads(old_tokenizer._tokenizer.to_str()   )
										self.assertNotIn('huggingface'  ,   json_tokenizer['model']['vocab']   )






@require_tokenizers
class      snake_case__	(   unittest.TestCase	):
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										__magic_name__					:				Union[str, Any]				=   PreTrainedTokenizerFast.from_pretrained('robot-test/dummy-tokenizer-wordlevel'   )
										__magic_name__					:				Optional[Any]				=   'The Matrix is a 1999 science fiction action film.'

										with concurrent.futures.ThreadPoolExecutor() as executor:
																__magic_name__					:				Union[str, Any]				=   [executor.submit(self.fetch  ,   _UpperCAmelCase  ,   _UpperCAmelCase   ) for i in range(10   )]
																__magic_name__					:				List[Any]				=   [future.result() for future in futures]
																self.assertEqual(_UpperCAmelCase  ,   [[1, 10, 0, 8, 0, 18, 0, 0, 0, 2] for i in range(10   )]   )





				def 			UpperCamelCase_    (	self  ,   _UpperCAmelCase  ,   _UpperCAmelCase   ):




										"""simple docstring"""




										return tokenizer.encode(_UpperCAmelCase  ,   truncation='longest_first'  ,   padding='longest'   )






import tempfile
import unittest

import torch

from diffusers import UNetaDConditionModel
from diffusers.training_utils import EMAModel
from diffusers.utils.testing_utils import enable_full_determinism, skip_mps, torch_device


enable_full_determinism()



class      snake_case__	(   unittest.TestCase	):
				__lowerCAmelCase		=   '''hf-internal-testing/tiny-stable-diffusion-pipe'''
				__lowerCAmelCase		=   1
				__lowerCAmelCase		=   77
				__lowerCAmelCase		=   32
				__lowerCAmelCase		=   4
				__lowerCAmelCase		=   __lowerCAmelCase		=   64
				__lowerCAmelCase		=   torch.manual_seed(0	)
				def 			UpperCamelCase_    (	self  ,   _UpperCAmelCase=0.9999   ):




										"""simple docstring"""




										__magic_name__					:				List[Any]				=   UNetaDConditionModel.from_pretrained(self.model_id  ,   subfolder='unet'   )
										__magic_name__					:				int				=   unet.to(_UpperCAmelCase   )
										__magic_name__					:				Optional[int]				=   EMAModel(unet.parameters()  ,   decay=_UpperCAmelCase  ,   model_cls=_UpperCAmelCase  ,   model_config=unet.config   )
										return unet, ema_unet
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										__magic_name__					:				List[str]				=   torch.randn(
										    self.batch_size  ,   self.num_in_channels  ,   self.latent_height  ,   self.latent_width  ,   generator=self.generator   ).to(_UpperCAmelCase   )
										__magic_name__					:				List[str]				=   torch.randint(0  ,   1000  ,   size=(self.batch_size,)  ,   generator=self.generator   ).to(_UpperCAmelCase   )
										__magic_name__					:				Any				=   torch.randn(
										    self.batch_size  ,   self.prompt_length  ,   self.text_encoder_hidden_dim  ,   generator=self.generator   ).to(_UpperCAmelCase   )
										return noisy_latents, timesteps, encoder_hidden_states
				def 			UpperCamelCase_    (	self  ,   _UpperCAmelCase   ):




										"""simple docstring"""




										__magic_name__					:				Any				=   {}
										for k, param in unet.state_dict().items():
																__magic_name__					:				str				=   torch.randn_like(_UpperCAmelCase   ) + (param * torch.randn_like(_UpperCAmelCase   ))
																updated_state_dict.update({k: updated_param}   )
										unet.load_state_dict(_UpperCAmelCase   )
										return unet
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										__magic_name__		,			__magic_name__					:				int				=   self.get_models()
										# Take the first (hypothetical) EMA step.
										ema_unet.step(unet.parameters()   )
										assert ema_unet.optimization_step == 1

										# Take two more.
										for _ in range(2   ):
																ema_unet.step(unet.parameters()   )
										assert ema_unet.optimization_step == 3
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										__magic_name__		,			__magic_name__					:				Union[str, Any]				=   self.get_models()
										# Since the `unet` is not being updated (i.e., backprop'd)
										# there won't be any difference between the `params` of `unet`
										# and `ema_unet` even if we call `ema_unet.step(unet.parameters())`.
										ema_unet.step(unet.parameters()   )
										__magic_name__					:				Any				=   list(unet.parameters()   )
										for s_param, param in zip(ema_unet.shadow_params  ,   _UpperCAmelCase   ):
																assert torch.allclose(_UpperCAmelCase  ,   _UpperCAmelCase   )

										# The above holds true even if we call `ema.step()` multiple times since
										# `unet` params are still not being updated.
										for _ in range(4   ):
																ema_unet.step(unet.parameters()   )
										for s_param, param in zip(ema_unet.shadow_params  ,   _UpperCAmelCase   ):
																assert torch.allclose(_UpperCAmelCase  ,   _UpperCAmelCase   )
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										__magic_name__		,			__magic_name__					:				Optional[Any]				=   self.get_models()
										# Here we simulate the parameter updates for `unet`. Since there might
										# be some parameters which are initialized to zero we take extra care to
										# initialize their values to something non-zero before the multiplication.
										__magic_name__					:				Any				=   self.simulate_backprop(_UpperCAmelCase   )

										# Take the EMA step.
										ema_unet.step(unet_pseudo_updated_step_one.parameters()   )

										# Now the EMA'd parameters won't be equal to the original model parameters.
										__magic_name__					:				Any				=   list(unet_pseudo_updated_step_one.parameters()   )
										for s_param, param in zip(ema_unet.shadow_params  ,   _UpperCAmelCase   ):
																assert ~torch.allclose(_UpperCAmelCase  ,   _UpperCAmelCase   )

										# Ensure this is the case when we take multiple EMA steps.
										for _ in range(4   ):
																ema_unet.step(unet.parameters()   )
										for s_param, param in zip(ema_unet.shadow_params  ,   _UpperCAmelCase   ):
																assert ~torch.allclose(_UpperCAmelCase  ,   _UpperCAmelCase   )
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										__magic_name__		,			__magic_name__					:				List[str]				=   self.get_models()

										# First backprop + EMA
										__magic_name__					:				int				=   self.simulate_backprop(_UpperCAmelCase   )
										ema_unet.step(unet_step_one.parameters()   )
										__magic_name__					:				Any				=   ema_unet.shadow_params

										# Second backprop + EMA
										__magic_name__					:				Any				=   self.simulate_backprop(_UpperCAmelCase   )
										ema_unet.step(unet_step_two.parameters()   )
										__magic_name__					:				Union[str, Any]				=   ema_unet.shadow_params

										for step_one, step_two in zip(_UpperCAmelCase  ,   _UpperCAmelCase   ):
																assert ~torch.allclose(_UpperCAmelCase  ,   _UpperCAmelCase   )
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										__magic_name__		,			__magic_name__					:				Optional[Any]				=   self.get_models(decay=0.0   )
										__magic_name__					:				str				=   self.simulate_backprop(_UpperCAmelCase   )
										ema_unet.step(unet_step_one.parameters()   )
										__magic_name__					:				List[Any]				=   ema_unet.shadow_params

										__magic_name__					:				Optional[Any]				=   self.simulate_backprop(_UpperCAmelCase   )
										ema_unet.step(unet_step_two.parameters()   )
										__magic_name__					:				Optional[Any]				=   ema_unet.shadow_params

										for step_one, step_two in zip(_UpperCAmelCase  ,   _UpperCAmelCase   ):
																assert torch.allclose(_UpperCAmelCase  ,   _UpperCAmelCase   )





				@skip_mps
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										__magic_name__		,			__magic_name__					:				Optional[int]				=   self.get_models()
										__magic_name__		,			__magic_name__		,			__magic_name__					:				str				=   self.get_dummy_inputs()

										with tempfile.TemporaryDirectory() as tmpdir:
																ema_unet.save_pretrained(_UpperCAmelCase   )
																__magic_name__					:				Optional[int]				=   UNetaDConditionModel.from_pretrained(_UpperCAmelCase  ,   model_cls=_UpperCAmelCase   )
																__magic_name__					:				str				=   loaded_unet.to(unet.device   )

										# Since no EMA step has been performed the outputs should match.
										__magic_name__					:				Union[str, Any]				=   unet(_UpperCAmelCase  ,   _UpperCAmelCase  ,   _UpperCAmelCase   ).sample
										__magic_name__					:				List[Any]				=   loaded_unet(_UpperCAmelCase  ,   _UpperCAmelCase  ,   _UpperCAmelCase   ).sample

										assert torch.allclose(_UpperCAmelCase  ,   _UpperCAmelCase  ,   atol=1E-4   )


