



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  __snake_case      (   SCREAMING_SNAKE_CASE			,SCREAMING_SNAKE_CASE			,unittest.TestCase):







							'''simple docstring'''






							UpperCamelCase__				:      Any         =					StableDiffusionSAGPipeline
							UpperCamelCase__				:      Dict         =					TEXT_TO_IMAGE_PARAMS
							UpperCamelCase__				:      Union[str, Any]         =					TEXT_TO_IMAGE_BATCH_PARAMS
							UpperCamelCase__				:      List[Any]         =					TEXT_TO_IMAGE_IMAGE_PARAMS
							UpperCamelCase__				:      Dict         =					TEXT_TO_IMAGE_IMAGE_PARAMS
							UpperCamelCase__				:      Tuple         =					False


							def _a      (							self		):
											torch.manual_seed(0		)
											a__				      =		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  ,  )
											a__				      =		DDIMScheduler(
											    beta_start=0.00_085  ,  beta_end=0.012  ,  beta_schedule="""scaled_linear"""  ,  clip_sample=a_  ,  set_alpha_to_one=a_  ,  )
											torch.manual_seed(0		)
											a__				      =		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		)
											a__				      =		CLIPTextConfig(
											    bos_token_id=0  ,  eos_token_id=2  ,  hidden_size=32  ,  intermediate_size=37  ,  layer_norm_eps=1E-0_5  ,  num_attention_heads=4  ,  num_hidden_layers=5  ,  pad_token_id=1  ,  vocab_size=1_000  ,  )
											a__				      =		CLIPTextModel(a_		)
											a__				      =		CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""		)

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


							def _a      (							self  ,  a_  ,  a_=0		):
											if str(a_		).startswith("""mps"""		):
															a__				      =		torch.manual_seed(a_		)
											else:
															a__				      =		torch.Generator(device=a_		).manual_seed(a_		)
											a__				      =		{
											    """prompt""": """.""",
											    """generator""": generator,
											    """num_inference_steps""": 2,
											    """guidance_scale""": 1.0,
											    """sag_scale""": 1.0,
											    """output_type""": """numpy""",
											}
											return inputs







							def _a      (							self		):
											super().test_inference_batch_single_identical(expected_max_diff=3E-3		)




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







							'''simple docstring'''






							def _a      (							self		):
											# clean up the VRAM after each test
											super().tearDown()
											gc.collect()
											torch.cuda.empty_cache()


							def _a      (							self		):
											a__				      =		StableDiffusionSAGPipeline.from_pretrained("""CompVis/stable-diffusion-v1-4"""		)
											a__				      =		sag_pipe.to(a_		)
											sag_pipe.set_progress_bar_config(disable=a_		)

											a__				      =		"""."""
											a__				      =		torch.manual_seed(0		)
											a__				      =		sag_pipe(
											    [prompt]  ,  generator=a_  ,  guidance_scale=7.5  ,  sag_scale=1.0  ,  num_inference_steps=20  ,  output_type="""np"""		)

											a__				      =		output.images

											a__				      =		image[0, -3:, -3:, -1]

											assert image.shape == (1, 512, 512, 3)
											a__				      =		np.array([0.1_568, 0.1_738, 0.1_695, 0.1_693, 0.1_507, 0.1_705, 0.1_547, 0.1_751, 0.1_949]		)

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


							def _a      (							self		):
											a__				      =		StableDiffusionSAGPipeline.from_pretrained("""stabilityai/stable-diffusion-2-1-base"""		)
											a__				      =		sag_pipe.to(a_		)
											sag_pipe.set_progress_bar_config(disable=a_		)

											a__				      =		"""."""
											a__				      =		torch.manual_seed(0		)
											a__				      =		sag_pipe(
											    [prompt]  ,  generator=a_  ,  guidance_scale=7.5  ,  sag_scale=1.0  ,  num_inference_steps=20  ,  output_type="""np"""		)

											a__				      =		output.images

											a__				      =		image[0, -3:, -3:, -1]

											assert image.shape == (1, 512, 512, 3)
											a__				      =		np.array([0.3_459, 0.2_876, 0.2_537, 0.3_002, 0.2_671, 0.2_160, 0.3_026, 0.2_262, 0.2_371]		)

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







							def _a      (							self		):
											a__				      =		StableDiffusionSAGPipeline.from_pretrained("""stabilityai/stable-diffusion-2-1-base"""		)
											a__				      =		sag_pipe.to(a_		)
											sag_pipe.set_progress_bar_config(disable=a_		)

											a__				      =		"""."""
											a__				      =		torch.manual_seed(0		)
											a__				      =		sag_pipe(
											    [prompt]  ,  width=768  ,  height=512  ,  generator=a_  ,  guidance_scale=7.5  ,  sag_scale=1.0  ,  num_inference_steps=20  ,  output_type="""np"""  ,  )

											a__				      =		output.images

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






import os
from shutil import copyfile
from typing import List, Optional, Tuple

from ...tokenization_utils import AddedToken
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import is_sentencepiece_available, logging


if is_sentencepiece_available():
	from .tokenization_albert import AlbertTokenizer
else:
	UpperCAmelCase							 =					None

UpperCAmelCase							 =					logging.get_logger(__name__)
UpperCAmelCase							 =					{"""vocab_file""": """spiece.model""", """tokenizer_file""": """tokenizer.json"""}

UpperCAmelCase							 =					{
    """vocab_file""": {
        """albert-base-v1""": """https://huggingface.co/albert-base-v1/resolve/main/spiece.model""",
        """albert-large-v1""": """https://huggingface.co/albert-large-v1/resolve/main/spiece.model""",
        """albert-xlarge-v1""": """https://huggingface.co/albert-xlarge-v1/resolve/main/spiece.model""",
        """albert-xxlarge-v1""": """https://huggingface.co/albert-xxlarge-v1/resolve/main/spiece.model""",
        """albert-base-v2""": """https://huggingface.co/albert-base-v2/resolve/main/spiece.model""",
        """albert-large-v2""": """https://huggingface.co/albert-large-v2/resolve/main/spiece.model""",
        """albert-xlarge-v2""": """https://huggingface.co/albert-xlarge-v2/resolve/main/spiece.model""",
        """albert-xxlarge-v2""": """https://huggingface.co/albert-xxlarge-v2/resolve/main/spiece.model""",
    },
    """tokenizer_file""": {
        """albert-base-v1""": """https://huggingface.co/albert-base-v1/resolve/main/tokenizer.json""",
        """albert-large-v1""": """https://huggingface.co/albert-large-v1/resolve/main/tokenizer.json""",
        """albert-xlarge-v1""": """https://huggingface.co/albert-xlarge-v1/resolve/main/tokenizer.json""",
        """albert-xxlarge-v1""": """https://huggingface.co/albert-xxlarge-v1/resolve/main/tokenizer.json""",
        """albert-base-v2""": """https://huggingface.co/albert-base-v2/resolve/main/tokenizer.json""",
        """albert-large-v2""": """https://huggingface.co/albert-large-v2/resolve/main/tokenizer.json""",
        """albert-xlarge-v2""": """https://huggingface.co/albert-xlarge-v2/resolve/main/tokenizer.json""",
        """albert-xxlarge-v2""": """https://huggingface.co/albert-xxlarge-v2/resolve/main/tokenizer.json""",
    },
}

UpperCAmelCase							 =					{
    """albert-base-v1""": 512,
    """albert-large-v1""": 512,
    """albert-xlarge-v1""": 512,
    """albert-xxlarge-v1""": 512,
    """albert-base-v2""": 512,
    """albert-large-v2""": 512,
    """albert-xlarge-v2""": 512,
    """albert-xxlarge-v2""": 512,
}

UpperCAmelCase							 =					"""▁"""

class  __snake_case      (   SCREAMING_SNAKE_CASE):







							'''simple docstring'''






							UpperCamelCase__				:      Tuple         =					VOCAB_FILES_NAMES
							UpperCamelCase__				:      Union[str, Any]         =					PRETRAINED_VOCAB_FILES_MAP
							UpperCamelCase__				:      List[Any]         =					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							UpperCamelCase__				:      List[str]         =					AlbertTokenizer


							def __init__(							self  ,  a_=None  ,  a_=None  ,  a_=True  ,  a_=True  ,  a_=False  ,  a_="[CLS]"  ,  a_="[SEP]"  ,  a_="<unk>"  ,  a_="[SEP]"  ,  a_="<pad>"  ,  a_="[CLS]"  ,  a_="[MASK]"  ,  **a_  ,  ):
											# Mask token behave like a normal word, i.e. include the space before it and
											# is included in the raw text, there should be a match in a non-normalized sentence.
											a__				      =		(
											    AddedToken(a_  ,  lstrip=a_  ,  rstrip=a_  ,  normalized=a_		)
											    if isinstance(a_  ,  a_		)
											    else mask_token
											)

											super().__init__(
											    a_  ,  tokenizer_file=a_  ,  do_lower_case=a_  ,  remove_space=a_  ,  keep_accents=a_  ,  bos_token=a_  ,  eos_token=a_  ,  unk_token=a_  ,  sep_token=a_  ,  pad_token=a_  ,  cls_token=a_  ,  mask_token=a_  ,  **a_  ,  )

											a__				      =		do_lower_case
											a__				      =		remove_space
											a__				      =		keep_accents
											a__				      =		vocab_file
											a__				      =		False if not self.vocab_file else True


							def _a      (							self  ,  a_  ,  a_ = None		):
											a__				      =		[self.sep_token_id]
											a__				      =		[self.cls_token_id]
											if token_ids_a is None:
															return cls + token_ids_a + sep
											return cls + token_ids_a + sep + token_ids_a + sep


							def _a      (							self  ,  a_  ,  a_ = None		):
											a__				      =		[self.sep_token_id]
											a__				      =		[self.cls_token_id]

											if token_ids_a is None:
															return len(cls + token_ids_a + sep		) * [0]
											return len(cls + token_ids_a + sep		) * [0] + len(token_ids_a + sep		) * [1]







							def _a      (							self  ,  a_  ,  a_ = None		):
											if not self.can_save_slow_tokenizer:
															raise ValueError(
															    """Your fast tokenizer does not have the necessary information to save the vocabulary for a slow """
															    """tokenizer."""		)

											if not os.path.isdir(a_		):
															logger.error(F'''Vocabulary path ({save_directory}) should be a directory'''		)
															return
											a__				      =		os.path.join(
											    a_  ,  (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]		)

											if os.path.abspath(self.vocab_file		) != os.path.abspath(a_		):
															copyfile(self.vocab_file  ,  a_		)

											return (out_vocab_file,)

