


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 							a						(_lowerCAmelCase	,							_lowerCAmelCase	,							unittest.TestCase ):




							"""simple docstring"""


							__UpperCAmelCase       :				str	    =    StableDiffusionSAGPipeline
							__UpperCAmelCase       :				List[str]	    =    TEXT_TO_IMAGE_PARAMS
							__UpperCAmelCase       :				Dict	    =    TEXT_TO_IMAGE_BATCH_PARAMS
							__UpperCAmelCase       :				Union[str, Any]	    =    TEXT_TO_IMAGE_IMAGE_PARAMS
							__UpperCAmelCase       :				Optional[Any]	    =    TEXT_TO_IMAGE_IMAGE_PARAMS
							__UpperCAmelCase       :				List[str]	    =    False

							def        __snake_case   (				self			:    List[Any]							)      ->       List[str]:
														torch.manual_seed(0							)
														__snake_case						:     Dict									=  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						:     Optional[int]									=  DDIMScheduler(
														    beta_start=0.0_00_85       ,      beta_end=0.0_12       ,      beta_schedule="scaled_linear"       ,      clip_sample=lowerCamelCase       ,      set_alpha_to_one=lowerCamelCase       ,      )
														torch.manual_seed(0							)
														__snake_case						:     Any									=  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						:     Optional[int]									=  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=1000       ,      )
														__snake_case						:     Optional[Any]									=  CLIPTextModel(lowerCamelCase							)
														__snake_case						:     List[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        __snake_case   (				self			:    Tuple       ,      lowerCamelCase			:    List[Any]       ,      lowerCamelCase			:    Union[str, Any]=0							)      ->       List[Any]:
														if str(lowerCamelCase							).startswith("mps"							):
																					__snake_case						:     Dict									=  torch.manual_seed(lowerCamelCase							)
														else:
																					__snake_case						:     Optional[Any]									=  torch.Generator(device=lowerCamelCase							).manual_seed(lowerCamelCase							)
														__snake_case						:     int									=  {
														    "prompt": ".",
														    "generator": generator,
														    "num_inference_steps": 2,
														    "guidance_scale": 1.0,
														    "sag_scale": 1.0,
														    "output_type": "numpy",
														}
														return inputs






							def        __snake_case   (				self			:    int							)      ->       str:
														super().test_inference_batch_single_identical(expected_max_diff=3E-3							)

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




							"""simple docstring"""


							def        __snake_case   (				self			:    List[str]							)      ->       Union[str, Any]:
														# clean up the VRAM after each test
														super().tearDown()
														gc.collect()
														torch.cuda.empty_cache()

							def        __snake_case   (				self			:    Optional[int]							)      ->       Optional[int]:
														__snake_case						:     Tuple									=  StableDiffusionSAGPipeline.from_pretrained("CompVis/stable-diffusion-v1-4"							)
														__snake_case						:     Tuple									=  sag_pipe.to(lowerCamelCase							)
														sag_pipe.set_progress_bar_config(disable=lowerCamelCase							)

														__snake_case						:     int									=  "."
														__snake_case						:     int									=  torch.manual_seed(0							)
														__snake_case						:     int									=  sag_pipe(
														    [prompt]       ,      generator=lowerCamelCase       ,      guidance_scale=7.5       ,      sag_scale=1.0       ,      num_inference_steps=20       ,      output_type="np"							)

														__snake_case						:     int									=  output.images

														__snake_case						:     List[Any]									=  image[0, -3:, -3:, -1]

														assert image.shape == (1, 512, 512, 3)
														__snake_case						:     List[Any]									=  np.array([0.15_68, 0.17_38, 0.16_95, 0.16_93, 0.15_07, 0.17_05, 0.15_47, 0.17_51, 0.19_49]							)

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

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

														__snake_case						:     Optional[Any]									=  "."
														__snake_case						:     List[str]									=  torch.manual_seed(0							)
														__snake_case						:     str									=  sag_pipe(
														    [prompt]       ,      generator=lowerCamelCase       ,      guidance_scale=7.5       ,      sag_scale=1.0       ,      num_inference_steps=20       ,      output_type="np"							)

														__snake_case						:     Tuple									=  output.images

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

														assert image.shape == (1, 512, 512, 3)
														__snake_case						:     str									=  np.array([0.34_59, 0.28_76, 0.25_37, 0.30_02, 0.26_71, 0.21_60, 0.30_26, 0.22_62, 0.23_71]							)

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






							def        __snake_case   (				self			:    Tuple							)      ->       List[str]:
														__snake_case						:     str									=  StableDiffusionSAGPipeline.from_pretrained("stabilityai/stable-diffusion-2-1-base"							)
														__snake_case						:     List[Any]									=  sag_pipe.to(lowerCamelCase							)
														sag_pipe.set_progress_bar_config(disable=lowerCamelCase							)

														__snake_case						:     int									=  "."
														__snake_case						:     Union[str, Any]									=  torch.manual_seed(0							)
														__snake_case						:     Any									=  sag_pipe(
														    [prompt]       ,      width=768       ,      height=512       ,      generator=lowerCamelCase       ,      guidance_scale=7.5       ,      sag_scale=1.0       ,      num_inference_steps=20       ,      output_type="np"       ,      )

														__snake_case						:     Tuple									=  output.images

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









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

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


if is_sentencepiece_available():
							from .tokenization_pegasus import PegasusTokenizer
else:
							A__      :     Optional[Any]       =					None


A__      :     str       =					logging.get_logger(__name__)


A__      :     Optional[int]       =					'▁'

A__      :     List[str]       =					{'vocab_file': 'spiece.model', 'tokenizer_file': 'tokenizer.json'}

A__      :     Dict       =					{
    'vocab_file': {'google/pegasus-xsum': 'https://huggingface.co/google/pegasus-xsum/resolve/main/spiece.model'},
    'tokenizer_file': {
        'google/pegasus-xsum': 'https://huggingface.co/google/pegasus-xsum/resolve/main/tokenizer.json'
    },
}

A__      :     int       =					{
    'google/pegasus-xsum': 5_12,
}
class  _UpperCAmelCase   (			A__      ):

				"""simple docstring"""




				lowercase__											=			VOCAB_FILES_NAMES
				lowercase__											=			PRETRAINED_VOCAB_FILES_MAP
				lowercase__											=			PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
				lowercase__											=			PegasusTokenizer
				lowercase__											=			["""input_ids""", """attention_mask"""]




				def __init__(	self		:       Dict,    lowerCamelCase		:       str=None,    lowerCamelCase		:       List[str]=None,    lowerCamelCase		:       int="<pad>",    lowerCamelCase		:       str="</s>",    lowerCamelCase		:       Optional[Any]="<unk>",    lowerCamelCase		:       Any="<mask_2>",    lowerCamelCase		:       Optional[Any]="<mask_1>",    lowerCamelCase		:       int=None,    lowerCamelCase		:       Dict=103,    **lowerCamelCase		:       List[Any],    ):

								'''simple docstring'''
								lowercase__  					=     offset

								if additional_special_tokens is not None:
												if not isinstance(lowerCamelCase,    lowerCamelCase    ):
																raise TypeError(
																    F"""additional_special_tokens should be of type {type(lowerCamelCase    )}, but is"""
																    F""" {type(lowerCamelCase    )}"""    )

												lowercase__  					=     (
												    ([mask_token_sent] + additional_special_tokens)
												    if mask_token_sent not in additional_special_tokens and mask_token_sent is not None
												    else additional_special_tokens
												)
												# fill additional tokens with ..., <unk_token_102> in case not all additional tokens are already taken
												additional_special_tokens_extended += [
												    F"""<unk_{i}>""" for i in range(len(lowerCamelCase    ),    self.offset - 1    )
												]

												if len(set(lowerCamelCase    )    ) != len(lowerCamelCase    ):
																raise ValueError(
																    '''Please make sure that the provided additional_special_tokens do not contain an incorrectly'''
																    F""" shifted list of <unk_x> tokens. Found {additional_special_tokens_extended}."""    )
												lowercase__  					=     additional_special_tokens_extended
								else:
												lowercase__  					=     [mask_token_sent] if mask_token_sent is not None else []
												additional_special_tokens += [F"""<unk_{i}>""" for i in range(2,    self.offset    )]

								super().__init__(
								    lowerCamelCase,    tokenizer_file=lowerCamelCase,    pad_token=lowerCamelCase,    eos_token=lowerCamelCase,    unk_token=lowerCamelCase,    mask_token=lowerCamelCase,    mask_token_sent=lowerCamelCase,    offset=lowerCamelCase,    additional_special_tokens=lowerCamelCase,    **lowerCamelCase,    )
								lowercase__  					=     vocab_file
								lowercase__  					=     False if not self.vocab_file else True




				def 							lowercase__  (	self		:       Optional[Any],    lowerCamelCase		:       int    ):

								'''simple docstring'''
								lowercase__  					=     set(self.all_special_ids    )  # call it once instead of inside list comp
								all_special_ids.remove(self.unk_token_id    )  # <unk> is only sometimes special

								if all_special_ids != set(range(len(self.additional_special_tokens    ) + 3    )    ):
												raise ValueError(
												    '''There should be 3 special tokens: mask_token, pad_token, and eos_token +'''
												    F""" {len(self.additional_special_tokens    )} additional_special_tokens, but got {all_special_ids}"""    )

								return [1 if x in all_special_ids else 0 for x in seq]




				def 							lowercase__  (	self		:       Optional[Any],    lowerCamelCase		:       List,    lowerCamelCase		:       Optional[List] = None,    lowerCamelCase		:       bool = False    ):

								'''simple docstring'''
								if already_has_special_tokens:
												return self._special_token_mask(lowerCamelCase    )
								elif token_ids_a is None:
												return self._special_token_mask(lowerCamelCase    ) + [1]
								else:
												return self._special_token_mask(token_ids_a + token_ids_a    ) + [1]




				def 							lowercase__  (	self		:       Optional[Any],    lowerCamelCase		:       Optional[int],    lowerCamelCase		:       Optional[Any]=None    ):

								'''simple docstring'''
								if token_ids_a is None:
												return token_ids_a + [self.eos_token_id]
								# We don't expect to process pairs, but leave the pair logic for API consistency
								return token_ids_a + token_ids_a + [self.eos_token_id]




				def 							lowercase__  (	self		:       Optional[Any],    lowerCamelCase		:       str,    lowerCamelCase		:       Optional[str] = None    ):

								'''simple docstring'''
								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(lowerCamelCase    ):
												logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""    )
												return
								lowercase__  					=     os.path.join(
								    lowerCamelCase,    (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']    )

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

								return (out_vocab_file,)




