
"""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


_A			:       List[str]   		=       False

@skip_mps
class 		a__						(      a_,			a_,			a_,			unittest.TestCase						):
   __lowerCAmelCase            =				StableDiffusionAttendAndExcitePipeline
   __lowerCAmelCase            =				False
   __lowerCAmelCase            =				TEXT_TO_IMAGE_PARAMS
   __lowerCAmelCase            =				TEXT_TO_IMAGE_BATCH_PARAMS.union({"""token_indices"""}						)
   __lowerCAmelCase            =				TEXT_TO_IMAGE_IMAGE_PARAMS
   __lowerCAmelCase            =				TEXT_TO_IMAGE_IMAGE_PARAMS
   @classmethod
   def 			__magic_name__			(			cls				):
          super().setUpClass()
          torch.use_deterministic_algorithms(_a				)
   @classmethod
   def 			__magic_name__			(			cls				):
          super().tearDownClass()
          torch.use_deterministic_algorithms(_a				)
   def 			__magic_name__			(			self				):
          torch.manual_seed(0				)
          lowercase    :				List[Any]       		=							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=_a	,    )
          lowercase    :				List[Any]       		=							DDIMScheduler(
              beta_start=0.0_0_0_8_5	,    beta_end=0.0_1_2	,    beta_schedule="scaled_linear"	,    clip_sample=_a	,    set_alpha_to_one=_a	,    )
          torch.manual_seed(0				)
          lowercase    :				Optional[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	,    sample_size=128	,    )
          torch.manual_seed(0				)
          lowercase    :				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=1_000	,    hidden_act="gelu"	,    projection_dim=512	,    )
          lowercase    :				Tuple       		=							CLIPTextModel(_a				)
          lowercase    :				Dict       		=							CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"				)

          lowercase    :				Optional[Any]       		=							{
              "unet": unet,
              "scheduler": scheduler,
              "vae": vae,
              "text_encoder": text_encoder,
              "tokenizer": tokenizer,
              "safety_checker": None,
              "feature_extractor": None,
          }

          return components
   def 			__magic_name__			(			self	,    _a	,    _a=0				):
          if str(_a				).startswith("mps"				):
                 lowercase    :				Optional[int]       		=							torch.manual_seed(_a				)
          else:
                 lowercase    :				Dict       		=							torch.Generator(device=_a				).manual_seed(_a				)
          lowercase    :				int       		=							{
              "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 			__magic_name__			(			self				):
          lowercase    :				Tuple       		=							"cpu"

          lowercase    :				str       		=							self.get_dummy_components()
          lowercase    :				Any       		=							self.pipeline_class(**_a				)
          pipe.to(_a				)
          pipe.set_progress_bar_config(disable=_a				)

          lowercase    :				Tuple       		=							self.get_dummy_inputs(_a				)
          lowercase    :				Optional[int]       		=							pipe(**_a				).images
          lowercase    :				Dict       		=							image[0, -3:, -3:, -1]

          self.assertEqual(image.shape	,    (1, 64, 64, 3)				)
          lowercase    :				int       		=							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]				)
          lowercase    :				List[str]       		=							np.abs(image_slice.flatten() - expected_slice				).max()
          self.assertLessEqual(_a	,    1E-3				)
   def 			__magic_name__			(			self				):
          super().test_cpu_offload_forward_pass(expected_max_diff=5E-4				)
   def 			__magic_name__			(			self				):
          # NOTE: Larger batch sizes cause this test to timeout, only test on smaller batches
          self._test_inference_batch_consistent(batch_sizes=[1, 2]				)
   def 			__magic_name__			(			self				):
          self._test_inference_batch_single_identical(batch_size=2	,    expected_max_diff=7E-4				)
   def 			__magic_name__			(			self				):
          super().test_dict_tuple_outputs_equivalent(expected_max_difference=3E-3				)
   def 			__magic_name__			(			self				):
          super().test_pt_np_pil_outputs_equivalent(expected_max_diff=5E-4				)
   def 			__magic_name__			(			self				):
          super().test_save_load_local(expected_max_difference=5E-4				)

   def 			__magic_name__			(			self				):
          super().test_save_load_optional_components(expected_max_difference=4E-4				)


@require_torch_gpu
@slow
class 		a__						(      unittest.TestCase						):
   @classmethod
   def 			__magic_name__			(			cls				):
          super().setUpClass()
          torch.use_deterministic_algorithms(_a				)
   @classmethod
   def 			__magic_name__			(			cls				):
          super().tearDownClass()
          torch.use_deterministic_algorithms(_a				)
   def 			__magic_name__			(			self				):
          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()

   def 			__magic_name__			(			self				):
          lowercase    :				int       		=							torch.manual_seed(51				)

          lowercase    :				List[Any]       		=							StableDiffusionAttendAndExcitePipeline.from_pretrained(
              "CompVis/stable-diffusion-v1-4"	,    safety_checker=_a	,    torch_dtype=torch.floataa				)
          pipe.to("cuda"				)

          lowercase    :				Tuple       		=							"a painting of an elephant with glasses"
          lowercase    :				Optional[int]       		=							[5, 7]

          lowercase    :				List[str]       		=							pipe(
              prompt=_a	,    token_indices=_a	,    guidance_scale=7.5	,    generator=_a	,    num_inference_steps=5	,    max_iter_to_alter=5	,    output_type="numpy"	,    ).images[0]

          lowercase    :				List[Any]       		=							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
import re
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


_A			:       List[str]   		=       logging.get_logger(__name__)

_A			:       List[str]   		=       {"""vocab_file""": """spiece.model"""}

_A			:       List[Any]   		=       {
    """vocab_file""": {
        """google/bigbird-roberta-base""": """https://huggingface.co/google/bigbird-roberta-base/resolve/main/spiece.model""",
        """google/bigbird-roberta-large""": (
            """https://huggingface.co/google/bigbird-roberta-large/resolve/main/spiece.model"""
        ),
        """google/bigbird-base-trivia-itc""": (
            """https://huggingface.co/google/bigbird-base-trivia-itc/resolve/main/spiece.model"""
        ),
    }
}

_A			:       str   		=       {
    """google/bigbird-roberta-base""": 40_96,
    """google/bigbird-roberta-large""": 40_96,
    """google/bigbird-base-trivia-itc""": 40_96,
}

class 		a__						(      a_						):
   __lowerCAmelCase            =				VOCAB_FILES_NAMES
   __lowerCAmelCase            =				PRETRAINED_VOCAB_FILES_MAP
   __lowerCAmelCase            =				PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
   __lowerCAmelCase            =				["""input_ids""", """attention_mask"""]
   __lowerCAmelCase            =				[]
   def __init__(			self	,    _a	,    _a="<unk>"	,    _a="<s>"	,    _a="</s>"	,    _a="<pad>"	,    _a="[SEP]"	,    _a="[MASK]"	,    _a="[CLS]"	,    _a = None	,    **_a	,    ):
          lowercase    :				Dict       		=							AddedToken(_a	,    lstrip=_a	,    rstrip=_a				) if isinstance(_a	,    _a				) else bos_token
          lowercase    :				Any       		=							AddedToken(_a	,    lstrip=_a	,    rstrip=_a				) if isinstance(_a	,    _a				) else eos_token
          lowercase    :				Any       		=							AddedToken(_a	,    lstrip=_a	,    rstrip=_a				) if isinstance(_a	,    _a				) else unk_token
          lowercase    :				List[Any]       		=							AddedToken(_a	,    lstrip=_a	,    rstrip=_a				) if isinstance(_a	,    _a				) else pad_token
          lowercase    :				Any       		=							AddedToken(_a	,    lstrip=_a	,    rstrip=_a				) if isinstance(_a	,    _a				) else cls_token
          lowercase    :				Any       		=							AddedToken(_a	,    lstrip=_a	,    rstrip=_a				) if isinstance(_a	,    _a				) else sep_token

          # Mask token behave like a normal word, i.e. include the space before it
          lowercase    :				List[Any]       		=							AddedToken(_a	,    lstrip=_a	,    rstrip=_a				) if isinstance(_a	,    _a				) else mask_token

          lowercase    :				Optional[int]       		=							{} if sp_model_kwargs is None else sp_model_kwargs

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

          lowercase    :				Optional[Any]       		=							vocab_file

          lowercase    :				List[str]       		=							spm.SentencePieceProcessor(**self.sp_model_kwargs				)
          self.sp_model.Load(_a				)
   @property
   def 			__magic_name__			(			self				):
          return self.sp_model.get_piece_size()
   def 			__magic_name__			(			self				):
          lowercase    :				str       		=							{self.convert_ids_to_tokens(_a				): i for i in range(self.vocab_size				)}
          vocab.update(self.added_tokens_encoder				)
          return vocab
   def __getstate__(			self				):
          lowercase    :				Union[str, Any]       		=							self.__dict__.copy()
          lowercase    :				Dict       		=							None
          return state
   def __setstate__(			self	,    _a				):
          lowercase    :				str       		=							d

          # for backward compatibility
          if not hasattr(self	,    "sp_model_kwargs"				):
                 lowercase    :				List[str]       		=							{}

          lowercase    :				Any       		=							spm.SentencePieceProcessor(**self.sp_model_kwargs				)
          self.sp_model.Load(self.vocab_file				)
   def 			__magic_name__			(			self	,    _a				):
          return self.sp_model.encode(_a	,    out_type=_a				)
   def 			__magic_name__			(			self	,    _a				):
          return self.sp_model.piece_to_id(_a				)
   def 			__magic_name__			(			self	,    _a				):
          lowercase    :				Union[str, Any]       		=							self.sp_model.IdToPiece(_a				)
          return token
   def 			__magic_name__			(			self	,    _a				):
          lowercase    :				List[Any]       		=							[]
          lowercase    :				List[Any]       		=							""
          lowercase    :				List[Any]       		=							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
                        lowercase    :				int       		=							True
                        lowercase    :				Union[str, Any]       		=							[]
                 else:
                        current_sub_tokens.append(_a				)
                        lowercase    :				int       		=							False
          out_string += self.sp_model.decode(_a				)
          return out_string.strip()
   def 			__magic_name__			(			self	,    _a	,    _a = False	,    _a = None	,    _a = True	,    **_a	,    ):
          lowercase    :				int       		=							kwargs.pop("use_source_tokenizer"	,    _a				)

          lowercase    :				Union[str, Any]       		=							self.convert_ids_to_tokens(_a	,    skip_special_tokens=_a				)

          # To avoid mixing byte-level and unicode for byte-level BPT
          # we need to build string separately for added tokens and byte-level tokens
          # cf. https://github.com/huggingface/transformers/issues/1133
          lowercase    :				Any       		=							[]
          lowercase    :				Dict       		=							[]
          for token in filtered_tokens:
                 if skip_special_tokens and token in self.all_special_ids:
                        continue
                 if token in self.added_tokens_encoder:
                        if current_sub_text:
                               sub_texts.append(self.convert_tokens_to_string(_a				)				)
                               lowercase    :				int       		=							[]
                        sub_texts.append(_a				)
                 else:
                        current_sub_text.append(_a				)
          if current_sub_text:
                 sub_texts.append(self.convert_tokens_to_string(_a				)				)

          # Mimic the behavior of the Rust tokenizer:
          # No space before [MASK] and [SEP]
          if spaces_between_special_tokens:
                 lowercase    :				Dict       		=							re.sub(R" (\[(MASK|SEP)\])"	,    R"\1"	,    " ".join(_a				)				)
          else:
                 lowercase    :				Union[str, Any]       		=							"".join(_a				)

          lowercase    :				int       		=							(
              clean_up_tokenization_spaces
              if clean_up_tokenization_spaces is not None
              else self.clean_up_tokenization_spaces
          )
          if clean_up_tokenization_spaces:
                 lowercase    :				Union[str, Any]       		=							self.clean_up_tokenization(_a				)
                 return clean_text
          else:
                 return text
   def 			__magic_name__			(			self	,    _a	,    _a = None				):
          if not os.path.isdir(_a				):
                 logger.error(f"""Vocabulary path ({save_directory}) should be a directory"""				)
                 return
          lowercase    :				List[Any]       		=							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:
                        lowercase    :				int       		=							self.sp_model.serialized_model_proto()
                        fi.write(_a				)

          return (out_vocab_file,)
   def 			__magic_name__			(			self	,    _a	,    _a = None				):
          if token_ids_a is None:
                 return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
          lowercase    :				List[str]       		=							[self.cls_token_id]
          lowercase    :				Tuple       		=							[self.sep_token_id]
          return cls + token_ids_a + sep + token_ids_a + sep
   def 			__magic_name__			(			self	,    _a	,    _a = None	,    _a = False				):
          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] + ([0] * len(_a				)) + [1]

   def 			__magic_name__			(			self	,    _a	,    _a = None				):
          lowercase    :				Tuple       		=							[self.sep_token_id]
          lowercase    :				Tuple       		=							[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]


