
import gc
import random
import unittest

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

import diffusers
from diffusers import (
    AutoencoderKL,
    EulerDiscreteScheduler,
    StableDiffusionLatentUpscalePipeline,
    StableDiffusionPipeline,
    UNetaDConditionModel,
)
from diffusers.schedulers import KarrasDiffusionSchedulers
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS, TEXT_GUIDED_IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()






def      lowerCamelCase__       (       __lowerCAmelCase						:						str       ):




			"""simple docstring"""


			lowerCAmelCase_      			=			[tensor.shape for tensor in tensor_list]
			return all(shape == shapes[0] for shape in shapes[1:]       )





class 			_lowerCAmelCase      (			__a      ,   __a      ,   __a      ,   unittest.TestCase ):
				_lowercase    	=StableDiffusionLatentUpscalePipeline
				_lowercase    	=TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
				    '''height''',
				    '''width''',
				    '''cross_attention_kwargs''',
				    '''negative_prompt_embeds''',
				    '''prompt_embeds''',
				}
				_lowercase    	=PipelineTesterMixin.required_optional_params - {'''num_images_per_prompt'''}
				_lowercase    	=TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
				_lowercase    	=frozenset(
				    [] )  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
				_lowercase    	=frozenset([] )

				_lowercase    	=True




				@property
				def   __a		(      self  )       ->   int:
							lowerCAmelCase_      			=			1
							lowerCAmelCase_      			=			4
							lowerCAmelCase_      			=			(16, 16)

							lowerCAmelCase_      			=			floats_tensor((batch_size, num_channels) + sizes       ,       rng=random.Random(0  )  ).to(_UpperCamelCase  )
							return image




				def   __a		(      self  )       ->   Union[str, Any]:
							torch.manual_seed(0  )
							lowerCAmelCase_      			=			UNetaDConditionModel(
							    act_fn="gelu"       ,       attention_head_dim=8       ,       norm_num_groups=_UpperCamelCase       ,       block_out_channels=[32, 32, 64, 64]       ,       time_cond_proj_dim=160       ,       conv_in_kernel=1       ,       conv_out_kernel=1       ,       cross_attention_dim=32       ,       down_block_types=(
							        "KDownBlock2D",
							        "KCrossAttnDownBlock2D",
							        "KCrossAttnDownBlock2D",
							        "KCrossAttnDownBlock2D",
							    )       ,       in_channels=8       ,       mid_block_type=_UpperCamelCase       ,       only_cross_attention=_UpperCamelCase       ,       out_channels=5       ,       resnet_time_scale_shift="scale_shift"       ,       time_embedding_type="fourier"       ,       timestep_post_act="gelu"       ,       up_block_types=("KCrossAttnUpBlock2D", "KCrossAttnUpBlock2D", "KCrossAttnUpBlock2D", "KUpBlock2D")       ,       )
							lowerCAmelCase_      			=			AutoencoderKL(
							    block_out_channels=[32, 32, 64, 64]       ,       in_channels=3       ,       out_channels=3       ,       down_block_types=[
							        "DownEncoderBlock2D",
							        "DownEncoderBlock2D",
							        "DownEncoderBlock2D",
							        "DownEncoderBlock2D",
							    ]       ,       up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D", "UpDecoderBlock2D", "UpDecoderBlock2D"]       ,       latent_channels=4       ,       )
							lowerCAmelCase_      			=			EulerDiscreteScheduler(prediction_type="sample"  )
							lowerCAmelCase_      			=			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       ,       hidden_act="quick_gelu"       ,       projection_dim=512       ,       )
							lowerCAmelCase_      			=			CLIPTextModel(_UpperCamelCase  )
							lowerCAmelCase_      			=			CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"  )

							lowerCAmelCase_      			=			{
							    "unet": model.eval(),
							    "vae": vae.eval(),
							    "scheduler": scheduler,
							    "text_encoder": text_encoder,
							    "tokenizer": tokenizer,
							}

							return components




				def   __a		(      self       ,       _UpperCamelCase       ,       _UpperCamelCase=0  )       ->   Dict:
							if str(_UpperCamelCase  ).startswith("mps"  ):
										lowerCAmelCase_      			=			torch.manual_seed(_UpperCamelCase  )
							else:
										lowerCAmelCase_      			=			torch.Generator(device=_UpperCamelCase  ).manual_seed(_UpperCamelCase  )
							lowerCAmelCase_      			=			{
							    "prompt": "A painting of a squirrel eating a burger",
							    "image": self.dummy_image.cpu(),
							    "generator": generator,
							    "num_inference_steps": 2,
							    "output_type": "numpy",
							}
							return inputs




				def   __a		(      self  )       ->   Tuple:
							lowerCAmelCase_      			=			"cpu"

							lowerCAmelCase_      			=			self.get_dummy_components()
							lowerCAmelCase_      			=			self.pipeline_class(**_UpperCamelCase  )
							pipe.to(_UpperCamelCase  )
							pipe.set_progress_bar_config(disable=_UpperCamelCase  )

							lowerCAmelCase_      			=			self.get_dummy_inputs(_UpperCamelCase  )
							lowerCAmelCase_      			=			pipe(**_UpperCamelCase  ).images
							lowerCAmelCase_      			=			image[0, -3:, -3:, -1]

							self.assertEqual(image.shape       ,       (1, 256, 256, 3)  )
							lowerCAmelCase_      			=			np.array(
							    [0.47222412, 0.41921633, 0.44717434, 0.46874192, 0.42588258, 0.46150726, 0.4677534, 0.45583832, 0.48579055]  )
							lowerCAmelCase_      			=			np.abs(image_slice.flatten() - expected_slice  ).max()
							self.assertLessEqual(_UpperCamelCase       ,       1e-3  )




				def   __a		(      self  )       ->   Optional[Any]:
							super().test_attention_slicing_forward_pass(expected_max_diff=7e-3  )




				def   __a		(      self  )       ->   Union[str, Any]:
							super().test_cpu_offload_forward_pass(expected_max_diff=3e-3  )




				def   __a		(      self  )       ->   Tuple:
							super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3  )




				def   __a		(      self  )       ->   str:
							super().test_inference_batch_single_identical(expected_max_diff=7e-3  )




				def   __a		(      self  )       ->   str:
							super().test_pt_np_pil_outputs_equivalent(expected_max_diff=3e-3  )




				def   __a		(      self  )       ->   Union[str, Any]:
							super().test_save_load_local(expected_max_difference=3e-3  )




				def   __a		(      self  )       ->   str:
							super().test_save_load_optional_components(expected_max_difference=3e-3  )




				def   __a		(      self  )       ->   str:
							lowerCAmelCase_      			=			[
							    "DDIMScheduler",
							    "DDPMScheduler",
							    "PNDMScheduler",
							    "HeunDiscreteScheduler",
							    "EulerAncestralDiscreteScheduler",
							    "KDPM2DiscreteScheduler",
							    "KDPM2AncestralDiscreteScheduler",
							    "DPMSolverSDEScheduler",
							]
							lowerCAmelCase_      			=			self.get_dummy_components()
							lowerCAmelCase_      			=			self.pipeline_class(**_UpperCamelCase  )

							# make sure that PNDM does not need warm-up
							pipe.scheduler.register_to_config(skip_prk_steps=_UpperCamelCase  )

							pipe.to(_UpperCamelCase  )
							pipe.set_progress_bar_config(disable=_UpperCamelCase  )
							lowerCAmelCase_      			=			self.get_dummy_inputs(_UpperCamelCase  )
							lowerCAmelCase_      			=			2

							lowerCAmelCase_      			=			[]
							for scheduler_enum in KarrasDiffusionSchedulers:
										if scheduler_enum.name in skip_schedulers:
													# no sigma schedulers are not supported
													# no schedulers
													continue

										lowerCAmelCase_      			=			getattr(_UpperCamelCase       ,       scheduler_enum.name  )
										lowerCAmelCase_      			=			scheduler_cls.from_config(pipe.scheduler.config  )
										lowerCAmelCase_      			=			pipe(**_UpperCamelCase  )[0]
										outputs.append(_UpperCamelCase  )

							assert check_same_shape(_UpperCamelCase  )





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




				def   __a		(      self  )       ->   Optional[int]:
							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()




				def   __a		(      self  )       ->   Tuple:
							lowerCAmelCase_      			=			torch.manual_seed(33  )

							lowerCAmelCase_      			=			StableDiffusionPipeline.from_pretrained("CompVis/stable-diffusion-v1-4"       ,       torch_dtype=torch.floataa  )
							pipe.to("cuda"  )

							lowerCAmelCase_      			=			StableDiffusionLatentUpscalePipeline.from_pretrained(
							    "stabilityai/sd-x2-latent-upscaler"       ,       torch_dtype=torch.floataa  )
							upscaler.to("cuda"  )

							lowerCAmelCase_      			=			"a photo of an astronaut high resolution, unreal engine, ultra realistic"

							lowerCAmelCase_      			=			pipe(_UpperCamelCase       ,       generator=_UpperCamelCase       ,       output_type="latent"  ).images

							lowerCAmelCase_      			=			upscaler(
							    prompt=_UpperCamelCase       ,       image=_UpperCamelCase       ,       num_inference_steps=20       ,       guidance_scale=0       ,       generator=_UpperCamelCase       ,       output_type="np"       ,       ).images[0]

							lowerCAmelCase_      			=			load_numpy(
							    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/astronaut_1024.npy"  )
							assert np.abs((expected_image - image).mean()  ) < 5e-2




				def   __a		(      self  )       ->   Union[str, Any]:
							lowerCAmelCase_      			=			torch.manual_seed(33  )

							lowerCAmelCase_      			=			StableDiffusionLatentUpscalePipeline.from_pretrained(
							    "stabilityai/sd-x2-latent-upscaler"       ,       torch_dtype=torch.floataa  )
							upscaler.to("cuda"  )

							lowerCAmelCase_      			=			"the temple of fire by Ross Tran and Gerardo Dottori, oil on canvas"

							lowerCAmelCase_      			=			load_image(
							    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/fire_temple_512.png"  )

							lowerCAmelCase_      			=			upscaler(
							    prompt=_UpperCamelCase       ,       image=_UpperCamelCase       ,       num_inference_steps=20       ,       guidance_scale=0       ,       generator=_UpperCamelCase       ,       output_type="np"       ,       ).images[0]

							lowerCAmelCase_      			=			load_numpy(
							    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/fire_temple_1024.npy"  )
							assert np.abs((expected_image - image).max()  ) < 5e-2





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

import sentencepiece as spm

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


_A     =  "▁"

_A     =  {"vocab_file": "spiece.model"}

_A     =  {
    "vocab_file": {"google/pegasus-xsum": "https://huggingface.co/google/pegasus-xsum/resolve/main/spiece.model"}
}

_A     =  {
    "google/pegasus-xsum": 5_12,
}


_A     =  logging.get_logger(__name__)





class 			_lowerCAmelCase      (			__a ):
				_lowercase    	=VOCAB_FILES_NAMES

				_lowercase    	=VOCAB_FILES_NAMES
				_lowercase    	=PRETRAINED_VOCAB_FILES_MAP
				_lowercase    	=PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
				_lowercase    	=['''input_ids''', '''attention_mask''']




				def __init__(      self       ,       _UpperCamelCase       ,       _UpperCamelCase="<pad>"       ,       _UpperCamelCase="</s>"       ,       _UpperCamelCase="<unk>"       ,       _UpperCamelCase="<mask_2>"       ,       _UpperCamelCase="<mask_1>"       ,       _UpperCamelCase=None       ,       _UpperCamelCase=103       ,       _UpperCamelCase = None       ,       **_UpperCamelCase       ,       )       ->   None:
							lowerCAmelCase_      			=			offset
							if additional_special_tokens is not None:
										if not isinstance(_UpperCamelCase       ,       _UpperCamelCase  ):
													raise TypeError(
													    f"""additional_special_tokens should be of type {type(_UpperCamelCase  )}, but is"""
													    f""" {type(_UpperCamelCase  )}"""  )

										lowerCAmelCase_      			=			(
										    ([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(_UpperCamelCase  )       ,       self.offset - 1  )
										]

										if len(set(_UpperCamelCase  )  ) != len(_UpperCamelCase  ):
													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}."""  )
										lowerCAmelCase_      			=			additional_special_tokens_extended
							else:
										lowerCAmelCase_      			=			[mask_token_sent] if mask_token_sent is not None else []
										additional_special_tokens += [f"""<unk_{i}>""" for i in range(2       ,       self.offset  )]

							lowerCAmelCase_      			=			{} if sp_model_kwargs is None else sp_model_kwargs

							super().__init__(
							    eos_token=_UpperCamelCase       ,       unk_token=_UpperCamelCase       ,       mask_token=_UpperCamelCase       ,       pad_token=_UpperCamelCase       ,       mask_token_sent=_UpperCamelCase       ,       offset=_UpperCamelCase       ,       additional_special_tokens=_UpperCamelCase       ,       sp_model_kwargs=self.sp_model_kwargs       ,       **_UpperCamelCase       ,       )
							lowerCAmelCase_      			=			mask_token_sent
							lowerCAmelCase_      			=			vocab_file
							lowerCAmelCase_      			=			spm.SentencePieceProcessor(**self.sp_model_kwargs  )
							self.sp_model.Load(_UpperCamelCase  )

							# add special tokens to encoder dict
							lowerCAmelCase_      			=			{
							    0: self.pad_token,
							    1: self.eos_token,
							}

							if self.mask_token_sent is not None:
										self.encoder.update(
										    {
										        2: self.mask_token_sent,
										        3: self.mask_token,
										    }  )

							if self.offset > 0:
										# entries 2-104 are only used for pretraining and called <mask_1>, <mask_2>, unk_2, ...unk_102
										# mask_token_sent is already added to list -> so start at 1
										self.encoder.update({i + 3: additional_special_tokens[i] for i in range(1       ,       self.offset - 1  )}  )

							lowerCAmelCase_      			=			{v: k for k, v in self.encoder.items()}




				@property
				def   __a		(      self  )       ->   int:
							return len(self.sp_model  ) + self.offset




				def   __a		(      self  )       ->   Dict[str, int]:
							lowerCAmelCase_      			=			{self.convert_ids_to_tokens(_UpperCamelCase  ): i for i in range(self.vocab_size  )}
							vocab.update(self.added_tokens_encoder  )
							return vocab




				def __getstate__(      self  )       ->   List[Any]:
							lowerCAmelCase_      			=			self.__dict__.copy()
							lowerCAmelCase_      			=			None
							return state




				def __setstate__(      self       ,       _UpperCamelCase  )       ->   Optional[int]:
							lowerCAmelCase_      			=			d

							# for backward compatibility
							if not hasattr(self       ,       "sp_model_kwargs"  ):
										lowerCAmelCase_      			=			{}

							lowerCAmelCase_      			=			spm.SentencePieceProcessor(**self.sp_model_kwargs  )
							self.sp_model.Load(self.vocab_file  )




				def   __a		(      self       ,       _UpperCamelCase  )       ->   List[str]:
							return self.sp_model.encode(_UpperCamelCase       ,       out_type=_UpperCamelCase  )




				def   __a		(      self       ,       _UpperCamelCase  )       ->   int:
							if token in self.decoder:
										return self.decoder[token]
							elif token in self.added_tokens_decoder:
										return self.added_tokens_decoder[token]
							lowerCAmelCase_      			=			self.sp_model.piece_to_id(_UpperCamelCase  )
							return sp_id + self.offset




				def   __a		(      self       ,       _UpperCamelCase  )       ->   str:
							if index in self.encoder:
										return self.encoder[index]
							elif index in self.added_tokens_encoder:
										return self.added_tokens_encoder[index]
							else:
										lowerCAmelCase_      			=			self.sp_model.IdToPiece(index - self.offset  )
							return token




				def   __a		(      self       ,       _UpperCamelCase  )       ->   Optional[Any]:
							lowerCAmelCase_      			=			[]
							lowerCAmelCase_      			=			""
							for token in tokens:
										# make sure that special tokens are not decoded using sentencepiece model
										if token in self.all_special_tokens:
													out_string += self.sp_model.decode(_UpperCamelCase  ) + token
													lowerCAmelCase_      			=			[]
										else:
													current_sub_tokens.append(_UpperCamelCase  )
							out_string += self.sp_model.decode(_UpperCamelCase  )
							return out_string.strip()




				def   __a		(      self       ,       _UpperCamelCase=False  )       ->   Optional[int]:
							return 1




				def   __a		(      self       ,       _UpperCamelCase  )       ->   int:
							lowerCAmelCase_      			=			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

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




				def   __a		(      self       ,       _UpperCamelCase       ,       _UpperCamelCase = None       ,       _UpperCamelCase = False  )       ->   List[int]:
							if already_has_special_tokens:
										return self._special_token_mask(_UpperCamelCase  )
							elif token_ids_a is None:
										return self._special_token_mask(_UpperCamelCase  ) + [1]
							else:
										return self._special_token_mask(token_ids_a + token_ids_a  ) + [1]




				def   __a		(      self       ,       _UpperCamelCase       ,       _UpperCamelCase=None  )       ->   List[int]:
							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   __a		(      self       ,       _UpperCamelCase       ,       _UpperCamelCase = None  )       ->   Tuple[str]:
							if not os.path.isdir(_UpperCamelCase  ):
										logger.error(f"""Vocabulary path ({save_directory}) should be a directory"""  )
										return
							lowerCAmelCase_      			=			os.path.join(
							    _UpperCamelCase       ,       (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]  )

							if os.path.abspath(self.vocab_file  ) != os.path.abspath(_UpperCamelCase  ) and os.path.isfile(self.vocab_file  ):
										copyfile(self.vocab_file       ,       _UpperCamelCase  )
							elif not os.path.isfile(self.vocab_file  ):
										with open(_UpperCamelCase       ,       "wb"  ) as fi:
													lowerCAmelCase_      			=			self.sp_model.serialized_model_proto()
													fi.write(_UpperCamelCase  )

							return (out_vocab_file,)



