




'''simple docstring'''





import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionAttendAndExcitePipeline,
    UNetaDConditionModel,
)
from diffusers.utils import load_numpy, skip_mps, slow
from diffusers.utils.testing_utils import 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 PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


_SCREAMING_SNAKE_CASE							:       Optional[Any]						=			False




@skip_mps
class 			_snake_case		(     __lowerCAmelCase						,						__lowerCAmelCase						,						__lowerCAmelCase						,						unittest.TestCase							):
   lowerCAmelCase_					:				Dict							=   StableDiffusionAttendAndExcitePipeline
   lowerCAmelCase_					:				List[str]							=   False
   lowerCAmelCase_					:				int							=   TEXT_TO_IMAGE_PARAMS
   lowerCAmelCase_					:				Tuple							=   TEXT_TO_IMAGE_BATCH_PARAMS.union({"token_indices"}							)
   lowerCAmelCase_					:				int							=   TEXT_TO_IMAGE_IMAGE_PARAMS
   lowerCAmelCase_					:				Any							=   TEXT_TO_IMAGE_IMAGE_PARAMS




   @classmethod
   def        lowerCAmelCase__    ( cls							)    ->    Union[str, Any]:

          '''simple docstring'''




          super().setUpClass()
          torch.use_deterministic_algorithms(lowerCamelCase__							)




   @classmethod
   def        lowerCAmelCase__    ( cls							)    ->    Dict:

          '''simple docstring'''




          super().tearDownClass()
          torch.use_deterministic_algorithms(lowerCamelCase__							)




   def        lowerCAmelCase__    ( self							)    ->    str:

          '''simple docstring'''




          torch.manual_seed(0							)
          snake_case_      =				UNetaDConditionModel(
              block_out_channels=(32, 64)    ,							layers_per_block=1    ,							sample_size=32    ,							in_channels=4    ,							out_channels=4    ,							down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")    ,							up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")    ,							cross_attention_dim=32    ,							attention_head_dim=(2, 4)    ,							use_linear_projection=lowerCamelCase__    ,							)
          snake_case_      =				DDIMScheduler(
              beta_start=0.0_0_0_8_5    ,							beta_end=0.0_1_2    ,							beta_schedule="scaled_linear"    ,							clip_sample=lowerCamelCase__    ,							set_alpha_to_one=lowerCamelCase__    ,							)
          torch.manual_seed(0							)
          snake_case_      =				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    ,							sample_size=128    ,							)
          torch.manual_seed(0							)
          snake_case_      =				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    ,							hidden_act="gelu"    ,							projection_dim=512    ,							)
          snake_case_      =				CLIPTextModel(lowerCamelCase__							)
          snake_case_      =				CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"							)

          snake_case_      =				{
              '''unet''': unet,
              '''scheduler''': scheduler,
              '''vae''': vae,
              '''text_encoder''': text_encoder,
              '''tokenizer''': tokenizer,
              '''safety_checker''': None,
              '''feature_extractor''': None,
          }

          return components




   def        lowerCAmelCase__    ( self    ,							a__    ,							a__=0							)    ->    List[str]:

          '''simple docstring'''




          if str(lowerCamelCase__							).startswith("mps"							):
                 snake_case_      =				torch.manual_seed(lowerCamelCase__							)
          else:
                 snake_case_      =				torch.Generator(device=lowerCamelCase__							).manual_seed(lowerCamelCase__							)
          snake_case_      =				{
              '''prompt''': '''a cat and a frog''',
              '''token_indices''': [2, 5],
              '''generator''': generator,
              '''num_inference_steps''': 1,
              '''guidance_scale''': 6.0,
              '''output_type''': '''numpy''',
              '''max_iter_to_alter''': 2,
              '''thresholds''': {0: 0.7},
          }
          return inputs




   def        lowerCAmelCase__    ( self							)    ->    Tuple:

          '''simple docstring'''




          snake_case_      =				'''cpu'''

          snake_case_      =				self.get_dummy_components()
          snake_case_      =				self.pipeline_class(**lowerCamelCase__							)
          pipe.to(lowerCamelCase__							)
          pipe.set_progress_bar_config(disable=lowerCamelCase__							)

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

          self.assertEqual(image.shape    ,							(1, 64, 64, 3)							)
          snake_case_      =				np.array(
              [0.6_3_9_0_5_3_6_4, 0.6_2_8_9_7_3_0_7, 0.4_8_5_9_9_0_1_7, 0.5_1_3_3_6_2_4, 0.5_5_5_0_0_4_8, 0.4_5_7_6_9_5_1_6, 0.5_0_3_2_6_9_7_3, 0.5_0_2_3_1_3_9, 0.4_5_3_8_4_4_9_6]							)
          snake_case_      =				np.abs(image_slice.flatten() - expected_slice							).max()
          self.assertLessEqual(lowerCamelCase__    ,							1e-3							)




   def        lowerCAmelCase__    ( self							)    ->    str:

          '''simple docstring'''




          super().test_cpu_offload_forward_pass(expected_max_diff=5e-4							)




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

          '''simple docstring'''




          self._test_inference_batch_consistent(batch_sizes=[1, 2]							)




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

          '''simple docstring'''




          self._test_inference_batch_single_identical(batch_size=2    ,							expected_max_diff=7e-4							)




   def        lowerCAmelCase__    ( self							)    ->    Dict:

          '''simple docstring'''




          super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3							)




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

          '''simple docstring'''




          super().test_pt_np_pil_outputs_equivalent(expected_max_diff=5e-4							)




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

          '''simple docstring'''




          super().test_save_load_local(expected_max_difference=5e-4							)




   def        lowerCAmelCase__    ( self							)    ->    Tuple:

          '''simple docstring'''




          super().test_save_load_optional_components(expected_max_difference=4e-4							)




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




   @classmethod
   def        lowerCAmelCase__    ( cls							)    ->    Any:

          '''simple docstring'''




          super().setUpClass()
          torch.use_deterministic_algorithms(lowerCamelCase__							)




   @classmethod
   def        lowerCAmelCase__    ( cls							)    ->    Dict:

          '''simple docstring'''




          super().tearDownClass()
          torch.use_deterministic_algorithms(lowerCamelCase__							)




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

          '''simple docstring'''




          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()




   def        lowerCAmelCase__    ( self							)    ->    str:

          '''simple docstring'''




          snake_case_      =				torch.manual_seed(51							)

          snake_case_      =				StableDiffusionAttendAndExcitePipeline.from_pretrained(
              "CompVis/stable-diffusion-v1-4"    ,							safety_checker=lowerCamelCase__    ,							torch_dtype=torch.floataa							)
          pipe.to("cuda"							)

          snake_case_      =				'''a painting of an elephant with glasses'''
          snake_case_      =				[5, 7]

          snake_case_      =				pipe(
              prompt=lowerCamelCase__    ,							token_indices=lowerCamelCase__    ,							guidance_scale=7.5    ,							generator=lowerCamelCase__    ,							num_inference_steps=5    ,							max_iter_to_alter=5    ,							output_type="numpy"    ,							).images[0]

          snake_case_      =				load_numpy(
              "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/attend-and-excite/elephant_glasses.npy"							)
          assert np.abs((expected_image - image).max()							) < 5e-1










'''simple docstring'''





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

import sentencepiece as spm

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


_SCREAMING_SNAKE_CASE							:       Tuple						=			logging.get_logger(__name__)

_SCREAMING_SNAKE_CASE							:       List[Any]						=			{"vocab_file": "sentencepiece.bpe.model"}

_SCREAMING_SNAKE_CASE							:       Tuple						=			{
    "vocab_file": {
        "moussaKam/mbarthez": "https://huggingface.co/moussaKam/mbarthez/resolve/main/sentencepiece.bpe.model",
        "moussaKam/barthez": "https://huggingface.co/moussaKam/barthez/resolve/main/sentencepiece.bpe.model",
        "moussaKam/barthez-orangesum-title": (
            "https://huggingface.co/moussaKam/barthez-orangesum-title/resolve/main/sentencepiece.bpe.model"
        ),
    },
}

_SCREAMING_SNAKE_CASE							:       List[Any]						=			{
    "moussaKam/mbarthez": 1024,
    "moussaKam/barthez": 1024,
    "moussaKam/barthez-orangesum-title": 1024,
}

_SCREAMING_SNAKE_CASE							:       Any						=			"▁"




class 			_snake_case		(     lowercase_							):
			lowerCAmelCase_					:				List[Any]							=   VOCAB_FILES_NAMES
			lowerCAmelCase_					:				Optional[Any]							=   PRETRAINED_VOCAB_FILES_MAP
			lowerCAmelCase_					:				int							=   PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
			lowerCAmelCase_					:				Tuple							=   ["input_ids", "attention_mask"]




			def __init__( self    ,							a__    ,							a__="<s>"    ,							a__="</s>"    ,							a__="</s>"    ,							a__="<s>"    ,							a__="<unk>"    ,							a__="<pad>"    ,							a__="<mask>"    ,							a__ = None    ,							**a__    ,							)    ->    None:

										'''simple docstring'''




										snake_case_      =				AddedToken(a__    ,							lstrip=a__    ,							rstrip=a__							) if isinstance(a__    ,							a__							) else mask_token

										snake_case_      =				{} if sp_model_kwargs is None else sp_model_kwargs

										super().__init__(
										    bos_token=a__    ,							eos_token=a__    ,							unk_token=a__    ,							sep_token=a__    ,							cls_token=a__    ,							pad_token=a__    ,							mask_token=a__    ,							sp_model_kwargs=self.sp_model_kwargs    ,							**a__    ,							)

										snake_case_      =				vocab_file
										snake_case_      =				spm.SentencePieceProcessor(**self.sp_model_kwargs							)
										self.sp_model.Load(str(a__							)							)

										snake_case_      =				{"<s>": 0, "<pad>": 1, "</s>": 2, "<unk>": 3}

										snake_case_      =				len(self.sp_model							) - 1
										snake_case_      =				{v: k for k, v in self.fairseq_tokens_to_ids.items()}




			def        lowerCAmelCase__    ( self    ,							a__    ,							a__ = None							)    ->    List[int]:

										'''simple docstring'''




										if token_ids_a is None:
																	return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
										snake_case_      =				[self.cls_token_id]
										snake_case_      =				[self.sep_token_id]
										return cls + token_ids_a + sep + sep + token_ids_a + sep




			def        lowerCAmelCase__    ( self    ,							a__    ,							a__ = None    ,							a__ = False							)    ->    List[int]:

										'''simple docstring'''




										if already_has_special_tokens:
																	return super().get_special_tokens_mask(
																	    token_ids_a=a__    ,							token_ids_a=a__    ,							already_has_special_tokens=a__							)

										if token_ids_a is None:
																	return [1] + ([0] * len(a__							)) + [1]
										return [1] + ([0] * len(a__							)) + [1, 1] + ([0] * len(a__							)) + [1]




			def        lowerCAmelCase__    ( self    ,							a__    ,							a__ = None							)    ->    List[int]:

										'''simple docstring'''




										snake_case_      =				[self.sep_token_id]
										snake_case_      =				[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 + sep + token_ids_a + sep							) * [0]




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

										'''simple docstring'''




										return len(self.sp_model							)




			def        lowerCAmelCase__    ( self							)    ->    int:

										'''simple docstring'''




										snake_case_      =				{self.convert_ids_to_tokens(a__							): i for i in range(self.vocab_size							)}
										vocab.update(self.added_tokens_encoder							)
										return vocab




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

										'''simple docstring'''




										return self.sp_model.encode(a__    ,							out_type=a__							)




			def        lowerCAmelCase__    ( self    ,							a__							)    ->    int:

										'''simple docstring'''




										if token in self.fairseq_tokens_to_ids:
																	return self.fairseq_tokens_to_ids[token]
										snake_case_      =				self.sp_model.PieceToId(a__							)

										return spm_id if spm_id else self.unk_token_id




			def        lowerCAmelCase__    ( self    ,							a__							)    ->    Tuple:

										'''simple docstring'''




										if index in self.fairseq_ids_to_tokens:
																	return self.fairseq_ids_to_tokens[index]
										return self.sp_model.IdToPiece(a__							)




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

										'''simple docstring'''




										snake_case_      =				[]
										snake_case_      =				""
										snake_case_      =				False
										for token in tokens:
																	# make sure that special tokens are not decoded using sentencepiece model
																	if token in self.all_special_tokens:
																								if not prev_is_special:
																															out_string += " "
																								out_string += self.sp_model.decode(a__							) + token
																								snake_case_      =				True
																								snake_case_      =				[]
																	else:
																								current_sub_tokens.append(a__							)
																								snake_case_      =				False
										out_string += self.sp_model.decode(a__							)
										return out_string.strip()




			def __getstate__( self							)    ->    Dict:

										'''simple docstring'''




										snake_case_      =				self.__dict__.copy()
										snake_case_      =				None
										return state




			def __setstate__( self    ,							a__							)    ->    str:

										'''simple docstring'''




										snake_case_      =				d

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

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




			def        lowerCAmelCase__    ( self    ,							a__    ,							a__ = None							)    ->    Tuple[str]:

										'''simple docstring'''




										if not os.path.isdir(a__							):
																	logger.error(F'Vocabulary path ({save_directory}) should be a directory'							)
																	return
										snake_case_      =				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__							) and os.path.isfile(self.vocab_file							):
																	copyfile(self.vocab_file    ,							a__							)
										elif not os.path.isfile(self.vocab_file							):
																	with open(a__    ,							"wb"							) as fi:
																								snake_case_      =				self.sp_model.serialized_model_proto()
																								fi.write(a__							)

										return (out_vocab_file,)




