



"""simple docstring"""
import os
import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from huggingface_hub.file_download import http_get
from requests.exceptions import HTTPError

from transformers import (
    AlbertTokenizer,
    AutoTokenizer,
    BertTokenizer,
    BertTokenizerFast,
    GPTaTokenizerFast,
    is_tokenizers_available,
)
from transformers.testing_utils import TOKEN, USER, is_staging_test, require_tokenizers
from transformers.tokenization_utils import Trie


sys.path.append(str(Path(__file__).parent.parent / "utils"))

from test_module.custom_tokenization import CustomTokenizer  # noqa E402


if is_tokenizers_available():
						from test_module.custom_tokenization_fast import CustomTokenizerFast
class 				lowerCamelCase  (			unittest.TestCase				):






					'''simple docstring'''





					def    lowerCAmelCase_				(       self:							Optional[Any]						)   ->	Union[str, Any]:
											# A mock response for an HTTP head request to emulate server down
											snake_case_ :Any						=			mock.Mock()
											snake_case_ :int						=			500
											snake_case_ :Optional[Any]						=			{}
											snake_case_ :Optional[Any]						=			HTTPError
											snake_case_ :Union[str, Any]						=			{}

											# Download this model to make sure it's in the cache.
											snake_case_ :List[str]						=			BertTokenizer.from_pretrained("""hf-internal-testing/tiny-random-bert"""						)

											# Under the mock environment we get a 500 error when trying to reach the tokenizer.
											with mock.patch("""requests.Session.request"""						,     return_value=snake_case						) as mock_head:
																	snake_case_ :str						=			BertTokenizer.from_pretrained("""hf-internal-testing/tiny-random-bert"""						)
																	# This check we did call the fake head request
																	mock_head.assert_called()






					@require_tokenizers
					def    lowerCAmelCase_				(       self:							Optional[Any]						)   ->	Optional[Any]:
											# A mock response for an HTTP head request to emulate server down
											snake_case_ :Any						=			mock.Mock()
											snake_case_ :Dict						=			500
											snake_case_ :Any						=			{}
											snake_case_ :List[Any]						=			HTTPError
											snake_case_ :Optional[Any]						=			{}

											# Download this model to make sure it's in the cache.
											snake_case_ :Any						=			GPTaTokenizerFast.from_pretrained("""gpt2"""						)

											# Under the mock environment we get a 500 error when trying to reach the tokenizer.
											with mock.patch("""requests.Session.request"""						,     return_value=snake_case						) as mock_head:
																	snake_case_ :Union[str, Any]						=			GPTaTokenizerFast.from_pretrained("""gpt2"""						)
																	# This check we did call the fake head request
																	mock_head.assert_called()






					def    lowerCAmelCase_				(       self:							Optional[int]						)   ->	Any:
											# This test is for deprecated behavior and can be removed in v5
											try:
																	snake_case_ :List[str]						=			tempfile.mktemp()
																	with open(snake_case						,     """wb"""						) as f:
																							http_get("""https://huggingface.co/albert-base-v1/resolve/main/spiece.model"""						,     snake_case						)

																	snake_case_ :List[str]						=			AlbertTokenizer.from_pretrained(snake_case						)
											finally:
																	os.remove(snake_case						)

											# Supporting this legacy load introduced a weird bug where the tokenizer would load local files if they are in
											# the current folder and have the right name.
											if os.path.isfile("""tokenizer.json"""						):
																	# We skip the test if the user has a `tokenizer.json` in this folder to avoid deleting it.
																	return
											try:
																	with open("""tokenizer.json"""						,     """wb"""						) as f:
																							http_get("""https://huggingface.co/hf-internal-testing/tiny-random-bert/blob/main/tokenizer.json"""						,     snake_case						)
																	snake_case_ :List[str]						=			AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-gpt2"""						)
																	# The tiny random BERT has a vocab size of 1024, tiny gpt2 as a vocab size of 1000
																	self.assertEqual(tokenizer.vocab_size						,     1_000						)
																	# Tokenizer should depend on the remote checkpoint, not the local tokenizer.json file.

											finally:
																	os.remove("""tokenizer.json"""						)






					def    lowerCAmelCase_				(       self:							int						)   ->	Tuple:
											# This test is for deprecated behavior and can be removed in v5
											snake_case_ :str						=			AlbertTokenizer.from_pretrained("""https://huggingface.co/albert-base-v1/resolve/main/spiece.model"""						)

@is_staging_test
class 				lowerCamelCase  (			unittest.TestCase				):






					'''simple docstring'''





					_A		:			str						=     ["""[UNK]""", """[CLS]""", """[SEP]""", """[PAD]""", """[MASK]""", """bla""", """blou"""]






					@classmethod
					def    lowerCAmelCase_				(       cls:							str						)   ->	Tuple:
											snake_case_ :int						=			TOKEN
											HfFolder.save_token(snake_case						)






					@classmethod
					def    lowerCAmelCase_				(       cls:							Any						)   ->	List[str]:
											try:
																	delete_repo(token=cls._token						,     repo_id="""test-tokenizer"""						)
											except HTTPError:
																	pass

											try:
																	delete_repo(token=cls._token						,     repo_id="""valid_org/test-tokenizer-org"""						)
											except HTTPError:
																	pass

											try:
																	delete_repo(token=cls._token						,     repo_id="""test-dynamic-tokenizer"""						)
											except HTTPError:
																	pass






					def    lowerCAmelCase_				(       self:							Tuple						)   ->	List[str]:
											with tempfile.TemporaryDirectory() as tmp_dir:
																	snake_case_ :Tuple						=			os.path.join(snake_case						,     """vocab.txt"""						)
																	with open(snake_case						,     """w"""						,     encoding="""utf-8"""						) as vocab_writer:
																							vocab_writer.write("""""".join([x + """\n""" for x in self.vocab_tokens]						)						)
																	snake_case_ :Any						=			BertTokenizer(snake_case						)

											tokenizer.push_to_hub("""test-tokenizer"""						,     use_auth_token=self._token						)
											snake_case_ :Optional[int]						=			BertTokenizer.from_pretrained(f"""{USER}/test-tokenizer"""						)
											self.assertDictEqual(new_tokenizer.vocab						,     tokenizer.vocab						)

											# Reset repo
											delete_repo(token=self._token						,     repo_id="""test-tokenizer"""						)

											# Push to hub via save_pretrained
											with tempfile.TemporaryDirectory() as tmp_dir:
																	tokenizer.save_pretrained(snake_case						,     repo_id="""test-tokenizer"""						,     push_to_hub=snake_case						,     use_auth_token=self._token						)

											snake_case_ :Optional[int]						=			BertTokenizer.from_pretrained(f"""{USER}/test-tokenizer"""						)
											self.assertDictEqual(new_tokenizer.vocab						,     tokenizer.vocab						)






					def    lowerCAmelCase_				(       self:							Union[str, Any]						)   ->	List[str]:
											with tempfile.TemporaryDirectory() as tmp_dir:
																	snake_case_ :Union[str, Any]						=			os.path.join(snake_case						,     """vocab.txt"""						)
																	with open(snake_case						,     """w"""						,     encoding="""utf-8"""						) as vocab_writer:
																							vocab_writer.write("""""".join([x + """\n""" for x in self.vocab_tokens]						)						)
																	snake_case_ :Dict						=			BertTokenizer(snake_case						)

											tokenizer.push_to_hub("""valid_org/test-tokenizer-org"""						,     use_auth_token=self._token						)
											snake_case_ :List[str]						=			BertTokenizer.from_pretrained("""valid_org/test-tokenizer-org"""						)
											self.assertDictEqual(new_tokenizer.vocab						,     tokenizer.vocab						)

											# Reset repo
											delete_repo(token=self._token						,     repo_id="""valid_org/test-tokenizer-org"""						)

											# Push to hub via save_pretrained
											with tempfile.TemporaryDirectory() as tmp_dir:
																	tokenizer.save_pretrained(
																	    snake_case						,     repo_id="""valid_org/test-tokenizer-org"""						,     push_to_hub=snake_case						,     use_auth_token=self._token						)

											snake_case_ :Dict						=			BertTokenizer.from_pretrained("""valid_org/test-tokenizer-org"""						)
											self.assertDictEqual(new_tokenizer.vocab						,     tokenizer.vocab						)






					@require_tokenizers
					def    lowerCAmelCase_				(       self:							List[Any]						)   ->	Optional[Any]:
											CustomTokenizer.register_for_auto_class()
											with tempfile.TemporaryDirectory() as tmp_dir:
																	snake_case_ :int						=			os.path.join(snake_case						,     """vocab.txt"""						)
																	with open(snake_case						,     """w"""						,     encoding="""utf-8"""						) as vocab_writer:
																							vocab_writer.write("""""".join([x + """\n""" for x in self.vocab_tokens]						)						)
																	snake_case_ :Union[str, Any]						=			CustomTokenizer(snake_case						)

											# No fast custom tokenizer
											tokenizer.push_to_hub("""test-dynamic-tokenizer"""						,     use_auth_token=self._token						)

											snake_case_ :Optional[int]						=			AutoTokenizer.from_pretrained(f"""{USER}/test-dynamic-tokenizer"""						,     trust_remote_code=snake_case						)
											# Can't make an isinstance check because the new_model.config is from the CustomTokenizer class of a dynamic module
											self.assertEqual(tokenizer.__class__.__name__						,     """CustomTokenizer"""						)

											# Fast and slow custom tokenizer
											CustomTokenizerFast.register_for_auto_class()
											with tempfile.TemporaryDirectory() as tmp_dir:
																	snake_case_ :int						=			os.path.join(snake_case						,     """vocab.txt"""						)
																	with open(snake_case						,     """w"""						,     encoding="""utf-8"""						) as vocab_writer:
																							vocab_writer.write("""""".join([x + """\n""" for x in self.vocab_tokens]						)						)

																	snake_case_ :Optional[int]						=			BertTokenizerFast.from_pretrained(snake_case						)
																	bert_tokenizer.save_pretrained(snake_case						)
																	snake_case_ :List[Any]						=			CustomTokenizerFast.from_pretrained(snake_case						)

											tokenizer.push_to_hub("""test-dynamic-tokenizer"""						,     use_auth_token=self._token						)

											snake_case_ :List[Any]						=			AutoTokenizer.from_pretrained(f"""{USER}/test-dynamic-tokenizer"""						,     trust_remote_code=snake_case						)
											# Can't make an isinstance check because the new_model.config is from the FakeConfig class of a dynamic module
											self.assertEqual(tokenizer.__class__.__name__						,     """CustomTokenizerFast"""						)
											snake_case_ :List[Any]						=			AutoTokenizer.from_pretrained(
											    f"""{USER}/test-dynamic-tokenizer"""						,     use_fast=snake_case						,     trust_remote_code=snake_case						)
											# Can't make an isinstance check because the new_model.config is from the FakeConfig class of a dynamic module
											self.assertEqual(tokenizer.__class__.__name__						,     """CustomTokenizer"""						)

class 				lowerCamelCase  (			unittest.TestCase				):






					'''simple docstring'''





					def    lowerCAmelCase_				(       self:							Optional[Any]						)   ->	str:
											snake_case_ :Tuple						=			Trie()
											trie.add("""Hello 友達"""						)
											self.assertEqual(trie.data						,     {"""H""": {"""e""": {"""l""": {"""l""": {"""o""": {""" """: {"""友""": {"""達""": {"""""": 1}}}}}}}}}						)
											trie.add("""Hello"""						)
											trie.data
											self.assertEqual(trie.data						,     {"""H""": {"""e""": {"""l""": {"""l""": {"""o""": {"""""": 1, """ """: {"""友""": {"""達""": {"""""": 1}}}}}}}}}						)






					def    lowerCAmelCase_				(       self:							Union[str, Any]						)   ->	int:
											snake_case_ :List[Any]						=			Trie()
											self.assertEqual(trie.split("""[CLS] This is a extra_id_100"""						)						,     ["""[CLS] This is a extra_id_100"""]						)
											trie.add("""[CLS]"""						)
											trie.add("""extra_id_1"""						)
											trie.add("""extra_id_100"""						)
											self.assertEqual(trie.split("""[CLS] This is a extra_id_100"""						)						,     ["""[CLS]""", """ This is a """, """extra_id_100"""]						)






					def    lowerCAmelCase_				(       self:							Optional[int]						)   ->	str:
											snake_case_ :Union[str, Any]						=			Trie()
											trie.add("""A"""						)
											self.assertEqual(trie.split("""ABC"""						)						,     ["""A""", """BC"""]						)
											self.assertEqual(trie.split("""BCA"""						)						,     ["""BC""", """A"""]						)






					def    lowerCAmelCase_				(       self:							List[str]						)   ->	Optional[int]:
											snake_case_ :Any						=			Trie()
											trie.add("""TOKEN]"""						)
											trie.add("""[SPECIAL_TOKEN]"""						)
											self.assertEqual(trie.split("""This is something [SPECIAL_TOKEN]"""						)						,     ["""This is something """, """[SPECIAL_TOKEN]"""]						)






					def    lowerCAmelCase_				(       self:							Any						)   ->	int:
											snake_case_ :Optional[int]						=			Trie()
											trie.add("""A"""						)
											trie.add("""P"""						)
											trie.add("""[SPECIAL_TOKEN]"""						)
											self.assertEqual(trie.split("""This is something [SPECIAL_TOKEN]"""						)						,     ["""This is something """, """[SPECIAL_TOKEN]"""]						)






					def    lowerCAmelCase_				(       self:							Tuple						)   ->	str:
											snake_case_ :int						=			Trie()
											trie.add("""AB"""						)
											trie.add("""B"""						)
											trie.add("""C"""						)
											self.assertEqual(trie.split("""ABC"""						)						,     ["""AB""", """C"""]						)






					def    lowerCAmelCase_				(       self:							List[str]						)   ->	Optional[int]:
											snake_case_ :Dict						=			Trie()
											trie.add("""ABC"""						)
											trie.add("""B"""						)
											trie.add("""CD"""						)
											self.assertEqual(trie.split("""ABCD"""						)						,     ["""ABC""", """D"""]						)






					def    lowerCAmelCase_				(       self:							Tuple						)   ->	Any:
											# Even if the offsets are wrong, we necessarily output correct string
											# parts.
											snake_case_ :List[Any]						=			Trie()
											snake_case_ :Tuple						=			trie.cut_text("""ABC"""						,     [0, 0, 2, 1, 2, 3]						)
											self.assertEqual(snake_case						,     ["""AB""", """C"""]						)





"""simple docstring"""
import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionSAGPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()
class 				lowerCamelCase  (			_lowerCAmelCase    ,						_lowerCAmelCase    ,						unittest.TestCase				):






					'''simple docstring'''





					_A		:			str						=     StableDiffusionSAGPipeline
					_A		:			Optional[Any]						=     TEXT_TO_IMAGE_PARAMS
					_A		:			Any						=     TEXT_TO_IMAGE_BATCH_PARAMS
					_A		:			Tuple						=     TEXT_TO_IMAGE_IMAGE_PARAMS
					_A		:			Tuple						=     TEXT_TO_IMAGE_IMAGE_PARAMS
					_A		:			List[str]						=     False






					def    lowerCAmelCase_				(       self:							Optional[Any]						)   ->	str:
											torch.manual_seed(0						)
											snake_case_ :Any						=			UNetaDConditionModel(
											    block_out_channels=(32, 64)						,     layers_per_block=2						,     sample_size=32						,     in_channels=4						,     out_channels=4						,     down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")						,     up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")						,     cross_attention_dim=32						,     )
											snake_case_ :Any						=			DDIMScheduler(
											    beta_start=0.0_0_0_8_5						,     beta_end=0.0_1_2						,     beta_schedule="""scaled_linear"""						,     clip_sample=snake_case						,     set_alpha_to_one=snake_case						,     )
											torch.manual_seed(0						)
											snake_case_ :Optional[int]						=			AutoencoderKL(
											    block_out_channels=[32, 64]						,     in_channels=3						,     out_channels=3						,     down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]						,     up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]						,     latent_channels=4						,     )
											torch.manual_seed(0						)
											snake_case_ :Union[str, Any]						=			CLIPTextConfig(
											    bos_token_id=0						,     eos_token_id=2						,     hidden_size=32						,     intermediate_size=37						,     layer_norm_eps=1E-05						,     num_attention_heads=4						,     num_hidden_layers=5						,     pad_token_id=1						,     vocab_size=1_000						,     )
											snake_case_ :Tuple						=			CLIPTextModel(snake_case						)
											snake_case_ :str						=			CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""						)

											snake_case_ :Dict						=			{
											    """unet""": unet,
											    """scheduler""": scheduler,
											    """vae""": vae,
											    """text_encoder""": text_encoder,
											    """tokenizer""": tokenizer,
											    """safety_checker""": None,
											    """feature_extractor""": None,
											}
											return components






					def    lowerCAmelCase_				(       self:							List[str]						,     snake_case:							Tuple						,     snake_case:							List[str]=0						)   ->	str:
											if str(snake_case						).startswith("""mps"""						):
																	snake_case_ :Tuple						=			torch.manual_seed(snake_case						)
											else:
																	snake_case_ :Optional[int]						=			torch.Generator(device=snake_case						).manual_seed(snake_case						)
											snake_case_ :Any						=			{
											    """prompt""": """.""",
											    """generator""": generator,
											    """num_inference_steps""": 2,
											    """guidance_scale""": 1.0,
											    """sag_scale""": 1.0,
											    """output_type""": """numpy""",
											}
											return inputs






					def    lowerCAmelCase_				(       self:							Optional[int]						)   ->	str:
											super().test_inference_batch_single_identical(expected_max_diff=3E-3						)

@slow
@require_torch_gpu
class 				lowerCamelCase  (			unittest.TestCase				):






					'''simple docstring'''





					def    lowerCAmelCase_				(       self:							int						)   ->	str:
											# clean up the VRAM after each test
											super().tearDown()
											gc.collect()
											torch.cuda.empty_cache()






					def    lowerCAmelCase_				(       self:							int						)   ->	List[str]:
											snake_case_ :Any						=			StableDiffusionSAGPipeline.from_pretrained("""CompVis/stable-diffusion-v1-4"""						)
											snake_case_ :int						=			sag_pipe.to(snake_case						)
											sag_pipe.set_progress_bar_config(disable=snake_case						)

											snake_case_ :Union[str, Any]						=			"""."""
											snake_case_ :str						=			torch.manual_seed(0						)
											snake_case_ :str						=			sag_pipe(
											    [prompt]						,     generator=snake_case						,     guidance_scale=7.5						,     sag_scale=1.0						,     num_inference_steps=20						,     output_type="""np"""						)

											snake_case_ :List[Any]						=			output.images

											snake_case_ :Tuple						=			image[0, -3:, -3:, -1]

											assert image.shape == (1, 512, 512, 3)
											snake_case_ :List[Any]						=			np.array([0.1_5_6_8, 0.1_7_3_8, 0.1_6_9_5, 0.1_6_9_3, 0.1_5_0_7, 0.1_7_0_5, 0.1_5_4_7, 0.1_7_5_1, 0.1_9_4_9]						)

											assert np.abs(image_slice.flatten() - expected_slice						).max() < 5E-2






					def    lowerCAmelCase_				(       self:							Dict						)   ->	str:
											snake_case_ :Tuple						=			StableDiffusionSAGPipeline.from_pretrained("""stabilityai/stable-diffusion-2-1-base"""						)
											snake_case_ :Optional[int]						=			sag_pipe.to(snake_case						)
											sag_pipe.set_progress_bar_config(disable=snake_case						)

											snake_case_ :Tuple						=			"""."""
											snake_case_ :Union[str, Any]						=			torch.manual_seed(0						)
											snake_case_ :Tuple						=			sag_pipe(
											    [prompt]						,     generator=snake_case						,     guidance_scale=7.5						,     sag_scale=1.0						,     num_inference_steps=20						,     output_type="""np"""						)

											snake_case_ :Optional[int]						=			output.images

											snake_case_ :Optional[Any]						=			image[0, -3:, -3:, -1]

											assert image.shape == (1, 512, 512, 3)
											snake_case_ :Tuple						=			np.array([0.3_4_5_9, 0.2_8_7_6, 0.2_5_3_7, 0.3_0_0_2, 0.2_6_7_1, 0.2_1_6_0, 0.3_0_2_6, 0.2_2_6_2, 0.2_3_7_1]						)

											assert np.abs(image_slice.flatten() - expected_slice						).max() < 5E-2






					def    lowerCAmelCase_				(       self:							List[str]						)   ->	List[str]:
											snake_case_ :Optional[int]						=			StableDiffusionSAGPipeline.from_pretrained("""stabilityai/stable-diffusion-2-1-base"""						)
											snake_case_ :int						=			sag_pipe.to(snake_case						)
											sag_pipe.set_progress_bar_config(disable=snake_case						)

											snake_case_ :Tuple						=			"""."""
											snake_case_ :Optional[int]						=			torch.manual_seed(0						)
											snake_case_ :List[str]						=			sag_pipe(
											    [prompt]						,     width=768						,     height=512						,     generator=snake_case						,     guidance_scale=7.5						,     sag_scale=1.0						,     num_inference_steps=20						,     output_type="""np"""						,     )

											snake_case_ :Optional[Any]						=			output.images

											assert image.shape == (1, 512, 768, 3)
