




'''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 				_snake_case	(			__UpperCAmelCase		,		unittest.TestCase							):
			lowerCAmelCase_  :   List[str]		 =					PreTrainedTokenizerFast
			lowerCAmelCase_  :   Tuple		 =					False
			lowerCAmelCase_  :   Dict		 =					True
			lowerCAmelCase_  :   Any		 =					"""tokenizer_file"""




			def      lowerCAmelCase__       (      self							)       ->       Union[str, Any]:

										'''simple docstring'''




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

										'''simple docstring'''




										pass




			def      lowerCAmelCase__       (      self							)       ->       int:

										'''simple docstring'''




										pass




			def      lowerCAmelCase__       (      self							)       ->       Optional[int]:

										'''simple docstring'''




										pass




			def      lowerCAmelCase__       (      self							)       ->       str:

										'''simple docstring'''




										pass




			def      lowerCAmelCase__       (      self							)       ->       str:

										'''simple docstring'''




										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							)       ->       int:

										'''simple docstring'''




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

										'''simple docstring'''




										snake_case_			      =   self.rust_tokenizer_class.from_pretrained(self.bytelevel_bpe_model_name							)

										snake_case_			      =   ("a" for _ in range(1_000							))
										snake_case_			      =   tokenizer.train_new_from_iterator(text_iterator=lowerCamelCase__       ,							length=1_000       ,							vocab_size=50							)

										snake_case_			      =   new_tokenizer.encode("a🤗"							)
										self.assertEqual(lowerCamelCase__       ,							[64, 172, 253, 97, 245]							)




@require_tokenizers
class 				_snake_case	(			unittest.TestCase							):




			def      lowerCAmelCase__       (      self							)       ->       str:

										'''simple docstring'''




										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							)       ->       Any:

										'''simple docstring'''




										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__							)       ,							28_997							)
										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__							)       ,							28_996							)
										snake_case_			      =   json.loads(old_tokenizer._tokenizer.to_str()							)
										self.assertNotIn("huggingface"       ,							json_tokenizer["model"]["vocab"]							)




@require_tokenizers
class 				_snake_case	(			unittest.TestCase							):




			def      lowerCAmelCase__       (      self							)       ->       List[Any]:

										'''simple docstring'''




										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       ,							a__       ,							a__							)       ->       Any:

										'''simple docstring'''




										return tokenizer.encode(lowerCamelCase__       ,							truncation="longest_first"       ,							padding="longest"							)










'''simple docstring'''





import tempfile

import torch

from diffusers import (
    DEISMultistepScheduler,
    DPMSolverMultistepScheduler,
    DPMSolverSinglestepScheduler,
    UniPCMultistepScheduler,
)

from .test_schedulers import SchedulerCommonTest




class 				_snake_case	(			lowercase_							):
			lowerCAmelCase_  :   Optional[int]		 =					(DPMSolverMultistepScheduler,)
			lowerCAmelCase_  :   int		 =					(("num_inference_steps", 25),)




			def      lowerCAmelCase__       (      self       ,							**a__							)       ->       Optional[int]:

										'''simple docstring'''




										snake_case_			      =   {
										    "num_train_timesteps": 1_000,
										    "beta_start": 0.0_0_0_1,
										    "beta_end": 0.0_2,
										    "beta_schedule": "linear",
										    "solver_order": 2,
										    "prediction_type": "epsilon",
										    "thresholding": False,
										    "sample_max_value": 1.0,
										    "algorithm_type": "dpmsolver++",
										    "solver_type": "midpoint",
										    "lower_order_final": False,
										    "lambda_min_clipped": -float("inf"							),
										    "variance_type": None,
										}

										config.update(**a__							)
										return config




			def      lowerCAmelCase__       (      self       ,							a__=0       ,							**a__							)       ->       Dict:

										'''simple docstring'''




										snake_case_			      =   dict(self.forward_default_kwargs							)
										snake_case_			      =   kwargs.pop("num_inference_steps"       ,							a__							)
										snake_case_			      =   self.dummy_sample
										snake_case_			      =   0.1 * sample
										snake_case_			      =   [residual + 0.2, residual + 0.1_5, residual + 0.1_0]

										for scheduler_class in self.scheduler_classes:
																	snake_case_			      =   self.get_scheduler_config(**a__							)
																	snake_case_			      =   scheduler_class(**a__							)
																	scheduler.set_timesteps(a__							)
																	# copy over dummy past residuals
																	snake_case_			      =   dummy_past_residuals[: scheduler.config.solver_order]

																	with tempfile.TemporaryDirectory() as tmpdirname:
																								scheduler.save_config(a__							)
																								snake_case_			      =   scheduler_class.from_pretrained(a__							)
																								new_scheduler.set_timesteps(a__							)
																								# copy over dummy past residuals
																								snake_case_			      =   dummy_past_residuals[: new_scheduler.config.solver_order]

																	snake_case_      ,   snake_case_			      =   sample, sample
																	for t in range(a__       ,							time_step + scheduler.config.solver_order + 1							):
																								snake_case_			      =   scheduler.step(a__       ,							a__       ,							a__       ,							**a__							).prev_sample
																								snake_case_			      =   new_scheduler.step(a__       ,							a__       ,							a__       ,							**a__							).prev_sample

																								assert torch.sum(torch.abs(output - new_output							)							) < 1e-5, "Scheduler outputs are not identical"




			def      lowerCAmelCase__       (      self							)       ->       Dict:

										'''simple docstring'''




										pass




			def      lowerCAmelCase__       (      self       ,							a__=0       ,							**a__							)       ->       Optional[int]:

										'''simple docstring'''




										snake_case_			      =   dict(self.forward_default_kwargs							)
										snake_case_			      =   kwargs.pop("num_inference_steps"       ,							a__							)
										snake_case_			      =   self.dummy_sample
										snake_case_			      =   0.1 * sample
										snake_case_			      =   [residual + 0.2, residual + 0.1_5, residual + 0.1_0]

										for scheduler_class in self.scheduler_classes:
																	snake_case_			      =   self.get_scheduler_config()
																	snake_case_			      =   scheduler_class(**a__							)
																	scheduler.set_timesteps(a__							)

																	# copy over dummy past residuals (must be after setting timesteps)
																	snake_case_			      =   dummy_past_residuals[: scheduler.config.solver_order]

																	with tempfile.TemporaryDirectory() as tmpdirname:
																								scheduler.save_config(a__							)
																								snake_case_			      =   scheduler_class.from_pretrained(a__							)
																								# copy over dummy past residuals
																								new_scheduler.set_timesteps(a__							)

																								# copy over dummy past residual (must be after setting timesteps)
																								snake_case_			      =   dummy_past_residuals[: new_scheduler.config.solver_order]

																	snake_case_			      =   scheduler.step(a__       ,							a__       ,							a__       ,							**a__							).prev_sample
																	snake_case_			      =   new_scheduler.step(a__       ,							a__       ,							a__       ,							**a__							).prev_sample

																	assert torch.sum(torch.abs(output - new_output							)							) < 1e-5, "Scheduler outputs are not identical"




			def      lowerCAmelCase__       (      self       ,							a__=None       ,							**a__							)       ->       Tuple:

										'''simple docstring'''




										if scheduler is None:
																	snake_case_			      =   self.scheduler_classes[0]
																	snake_case_			      =   self.get_scheduler_config(**a__							)
																	snake_case_			      =   scheduler_class(**a__							)

										snake_case_			      =   10
										snake_case_			      =   self.dummy_model()
										snake_case_			      =   self.dummy_sample_deter
										scheduler.set_timesteps(a__							)

										for i, t in enumerate(scheduler.timesteps							):
																	snake_case_			      =   model(a__       ,							a__							)
																	snake_case_			      =   scheduler.step(a__       ,							a__       ,							a__							).prev_sample

										return sample




			def      lowerCAmelCase__       (      self							)       ->       Dict:

										'''simple docstring'''




										snake_case_			      =   dict(self.forward_default_kwargs							)

										snake_case_			      =   kwargs.pop("num_inference_steps"       ,							a__							)

										for scheduler_class in self.scheduler_classes:
																	snake_case_			      =   self.get_scheduler_config()
																	snake_case_			      =   scheduler_class(**a__							)

																	snake_case_			      =   self.dummy_sample
																	snake_case_			      =   0.1 * sample

																	if num_inference_steps is not None and hasattr(a__       ,							"set_timesteps"							):
																								scheduler.set_timesteps(a__							)
																	elif num_inference_steps is not None and not hasattr(a__       ,							"set_timesteps"							):
																								snake_case_			      =   num_inference_steps

																	# copy over dummy past residuals (must be done after set_timesteps)
																	snake_case_			      =   [residual + 0.2, residual + 0.1_5, residual + 0.1_0]
																	snake_case_			      =   dummy_past_residuals[: scheduler.config.solver_order]

																	snake_case_			      =   scheduler.timesteps[5]
																	snake_case_			      =   scheduler.timesteps[6]

																	snake_case_			      =   scheduler.step(a__       ,							a__       ,							a__       ,							**a__							).prev_sample
																	snake_case_			      =   scheduler.step(a__       ,							a__       ,							a__       ,							**a__							).prev_sample

																	self.assertEqual(output_a.shape       ,							sample.shape							)
																	self.assertEqual(output_a.shape       ,							output_a.shape							)




			def      lowerCAmelCase__       (      self							)       ->       List[str]:

										'''simple docstring'''




										for timesteps in [25, 50, 100, 999, 1_000]:
																	self.check_over_configs(num_train_timesteps=a__							)




			def      lowerCAmelCase__       (      self							)       ->       str:

										'''simple docstring'''




										self.check_over_configs(thresholding=a__							)
										for order in [1, 2, 3]:
																	for solver_type in ["midpoint", "heun"]:
																								for threshold in [0.5, 1.0, 2.0]:
																															for prediction_type in ["epsilon", "sample"]:
																																						self.check_over_configs(
																																						    thresholding=a__       ,							prediction_type=a__       ,							sample_max_value=a__       ,							algorithm_type="dpmsolver++"       ,							solver_order=a__       ,							solver_type=a__       ,							)




			def      lowerCAmelCase__       (      self							)       ->       int:

										'''simple docstring'''




										for prediction_type in ["epsilon", "v_prediction"]:
																	self.check_over_configs(prediction_type=a__							)




			def      lowerCAmelCase__       (      self							)       ->       Optional[int]:

										'''simple docstring'''




										for algorithm_type in ["dpmsolver", "dpmsolver++", "sde-dpmsolver", "sde-dpmsolver++"]:
																	for solver_type in ["midpoint", "heun"]:
																								for order in [1, 2, 3]:
																															for prediction_type in ["epsilon", "sample"]:
																																						if algorithm_type in ["sde-dpmsolver", "sde-dpmsolver++"]:
																																													if order == 3:
																																																				continue
																																						else:
																																													self.check_over_configs(
																																													    solver_order=a__       ,							solver_type=a__       ,							prediction_type=a__       ,							algorithm_type=a__       ,							)
																																						snake_case_			      =   self.full_loop(
																																						    solver_order=a__       ,							solver_type=a__       ,							prediction_type=a__       ,							algorithm_type=a__       ,							)
																																						assert not torch.isnan(a__							).any(), "Samples have nan numbers"




			def      lowerCAmelCase__       (      self							)       ->       Union[str, Any]:

										'''simple docstring'''




										self.check_over_configs(lower_order_final=a__							)
										self.check_over_configs(lower_order_final=a__							)




			def      lowerCAmelCase__       (      self							)       ->       Optional[Any]:

										'''simple docstring'''




										self.check_over_configs(lambda_min_clipped=-float("inf"							)							)
										self.check_over_configs(lambda_min_clipped=-5.1							)




			def      lowerCAmelCase__       (      self							)       ->       Any:

										'''simple docstring'''




										self.check_over_configs(variance_type=a__							)
										self.check_over_configs(variance_type="learned_range"							)




			def      lowerCAmelCase__       (      self							)       ->       Union[str, Any]:

										'''simple docstring'''




										for num_inference_steps in [1, 2, 3, 5, 10, 50, 100, 999, 1_000]:
																	self.check_over_forward(num_inference_steps=a__       ,							time_step=0							)




			def      lowerCAmelCase__       (      self							)       ->       List[str]:

										'''simple docstring'''




										snake_case_			      =   self.full_loop()
										snake_case_			      =   torch.mean(torch.abs(a__							)							)

										assert abs(result_mean.item() - 0.3_3_0_1							) < 1e-3




			def      lowerCAmelCase__       (      self							)       ->       Union[str, Any]:

										'''simple docstring'''




										snake_case_			      =   self.full_loop(thresholding=a__       ,							dynamic_thresholding_ratio=0.8_7       ,							sample_max_value=0.5							)
										snake_case_			      =   torch.mean(torch.abs(a__							)							)

										assert abs(result_mean.item() - 1.1_3_6_4							) < 1e-3




			def      lowerCAmelCase__       (      self							)       ->       List[str]:

										'''simple docstring'''




										snake_case_			      =   self.full_loop(prediction_type="v_prediction"							)
										snake_case_			      =   torch.mean(torch.abs(a__							)							)

										assert abs(result_mean.item() - 0.2_2_5_1							) < 1e-3




			def      lowerCAmelCase__       (      self							)       ->       List[str]:

										'''simple docstring'''




										snake_case_			      =   self.full_loop(prediction_type="v_prediction"       ,							use_karras_sigmas=a__							)
										snake_case_			      =   torch.mean(torch.abs(a__							)							)

										assert abs(result_mean.item() - 0.2_0_9_6							) < 1e-3




			def      lowerCAmelCase__       (      self							)       ->       Union[str, Any]:

										'''simple docstring'''




										snake_case_			      =   DPMSolverMultistepScheduler(**self.get_scheduler_config()							)
										snake_case_			      =   self.full_loop(scheduler=a__							)
										snake_case_			      =   torch.mean(torch.abs(a__							)							)

										assert abs(result_mean.item() - 0.3_3_0_1							) < 1e-3

										snake_case_			      =   DPMSolverSinglestepScheduler.from_config(scheduler.config							)
										snake_case_			      =   UniPCMultistepScheduler.from_config(scheduler.config							)
										snake_case_			      =   DEISMultistepScheduler.from_config(scheduler.config							)
										snake_case_			      =   DPMSolverMultistepScheduler.from_config(scheduler.config							)

										snake_case_			      =   self.full_loop(scheduler=a__							)
										snake_case_			      =   torch.mean(torch.abs(a__							)							)

										assert abs(result_mean.item() - 0.3_3_0_1							) < 1e-3




			def      lowerCAmelCase__       (      self							)       ->       int:

										'''simple docstring'''




										snake_case_			      =   self.scheduler_classes[0]
										snake_case_			      =   self.get_scheduler_config(thresholding=a__       ,							dynamic_thresholding_ratio=0							)
										snake_case_			      =   scheduler_class(**a__							)

										snake_case_			      =   10
										snake_case_			      =   self.dummy_model()
										snake_case_			      =   self.dummy_sample_deter.half()
										scheduler.set_timesteps(a__							)

										for i, t in enumerate(scheduler.timesteps							):
																	snake_case_			      =   model(a__       ,							a__							)
																	snake_case_			      =   scheduler.step(a__       ,							a__       ,							a__							).prev_sample

										assert sample.dtype == torch.floataa




			def      lowerCAmelCase__       (      self       ,							**a__							)       ->       Dict:

										'''simple docstring'''




										for scheduler_class in self.scheduler_classes:
																	snake_case_			      =   self.get_scheduler_config(**a__							)
																	snake_case_			      =   scheduler_class(**a__							)

																	scheduler.set_timesteps(scheduler.config.num_train_timesteps							)
																	assert len(scheduler.timesteps.unique()							) == scheduler.num_inference_steps




