





import gc
import random
import unittest

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

from diffusers import AutoencoderKL, DDIMScheduler, DDPMScheduler, StableDiffusionUpscalePipeline, UNetaDConditionModel
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


enable_full_determinism()






class        snake_case_    (		unittest.TestCase						):






  def 				UpperCAmelCase__						(							self			:					Any      )->Optional[int]:






   '''simple docstring'''

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






  @property
  def 				UpperCAmelCase__						(							self			:					List[str]      )->str:






   '''simple docstring'''

   __lowerCAmelCase					:	Dict      =       1
   __lowerCAmelCase					:	Union[str, Any]      =       3
   __lowerCAmelCase					:	str      =       (32, 32)

   __lowerCAmelCase					:	List[Any]      =       floats_tensor((batch_size, num_channels) + sizes       ,  rng=random.Random(0      )      ).to(_snake_case      )
   return image






  @property
  def 				UpperCAmelCase__						(							self			:					Optional[Any]      )->Optional[Any]:






   '''simple docstring'''

   torch.manual_seed(0      )
   __lowerCAmelCase					:	Any      =       UNetaDConditionModel(
       block_out_channels=(32, 32, 64)       ,  layers_per_block=2       ,  sample_size=32       ,  in_channels=7       ,  out_channels=4       ,  down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""", """CrossAttnDownBlock2D""")       ,  up_block_types=("""CrossAttnUpBlock2D""", """CrossAttnUpBlock2D""", """UpBlock2D""")       ,  cross_attention_dim=32       ,  attention_head_dim=8       ,  use_linear_projection=_snake_case       ,  only_cross_attention=(True, True, False)       ,  num_class_embeds=100       ,  )
   return model






  @property
  def 				UpperCAmelCase__						(							self			:					List[str]      )->List[str]:






   '''simple docstring'''

   torch.manual_seed(0      )
   __lowerCAmelCase					:	List[Any]      =       AutoencoderKL(
       block_out_channels=[32, 32, 64]       ,  in_channels=3       ,  out_channels=3       ,  down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D""", """DownEncoderBlock2D"""]       ,  up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D""", """UpDecoderBlock2D"""]       ,  latent_channels=4       ,  )
   return model






  @property
  def 				UpperCAmelCase__						(							self			:					List[Any]      )->Union[str, Any]:






   '''simple docstring'''

   torch.manual_seed(0      )
   __lowerCAmelCase					:	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       ,  hidden_act="""gelu"""       ,  projection_dim=512       ,  )
   return CLIPTextModel(_snake_case      )






  def 				UpperCAmelCase__						(							self			:					Optional[int]      )->List[str]:






   '''simple docstring'''

   __lowerCAmelCase					:	Optional[int]      =       """cpu"""  # ensure determinism for the device-dependent torch.Generator
   __lowerCAmelCase					:	Dict      =       self.dummy_cond_unet_upscale
   __lowerCAmelCase					:	Dict      =       DDPMScheduler()
   __lowerCAmelCase					:	Union[str, Any]      =       DDIMScheduler(prediction_type="""v_prediction"""      )
   __lowerCAmelCase					:	Dict      =       self.dummy_vae
   __lowerCAmelCase					:	Union[str, Any]      =       self.dummy_text_encoder
   __lowerCAmelCase					:	Union[str, Any]      =       CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""      )

   __lowerCAmelCase					:	List[Any]      =       self.dummy_image.cpu().permute(0       ,  2       ,  3       ,  1      )[0]
   __lowerCAmelCase					:	str      =       Image.fromarray(np.uinta(_snake_case      )      ).convert("""RGB"""      ).resize((64, 64)      )

   # make sure here that pndm scheduler skips prk
   __lowerCAmelCase					:	Dict      =       StableDiffusionUpscalePipeline(
       unet=_snake_case       ,  low_res_scheduler=_snake_case       ,  scheduler=_snake_case       ,  vae=_snake_case       ,  text_encoder=_snake_case       ,  tokenizer=_snake_case       ,  max_noise_level=350       ,  )
   __lowerCAmelCase					:	Dict      =       sd_pipe.to(_snake_case      )
   sd_pipe.set_progress_bar_config(disable=_snake_case      )

   __lowerCAmelCase					:	Dict      =       """A painting of a squirrel eating a burger"""
   __lowerCAmelCase					:	Any      =       torch.Generator(device=_snake_case      ).manual_seed(0      )
   __lowerCAmelCase					:	Union[str, Any]      =       sd_pipe(
       [prompt]       ,  image=_snake_case       ,  generator=_snake_case       ,  guidance_scale=6.0       ,  noise_level=20       ,  num_inference_steps=2       ,  output_type="""np"""       ,  )

   __lowerCAmelCase					:	Optional[int]      =       output.images

   __lowerCAmelCase					:	int      =       torch.Generator(device=_snake_case      ).manual_seed(0      )
   __lowerCAmelCase					:	List[str]      =       sd_pipe(
       [prompt]       ,  image=_snake_case       ,  generator=_snake_case       ,  guidance_scale=6.0       ,  noise_level=20       ,  num_inference_steps=2       ,  output_type="""np"""       ,  return_dict=_snake_case       ,  )[0]

   __lowerCAmelCase					:	List[Any]      =       image[0, -3:, -3:, -1]
   __lowerCAmelCase					:	Dict      =       image_from_tuple[0, -3:, -3:, -1]

   __lowerCAmelCase					:	Tuple      =       low_res_image.size[0] * 4
   assert image.shape == (1, expected_height_width, expected_height_width, 3)
   __lowerCAmelCase					:	Tuple      =       np.array([0.3_113, 0.3_910, 0.4_272, 0.4_859, 0.5_061, 0.4_652, 0.5_362, 0.5_715, 0.5_661]      )

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






  def 				UpperCAmelCase__						(							self			:					Optional[Any]      )->List[str]:






   '''simple docstring'''

   __lowerCAmelCase					:	Tuple      =       """cpu"""  # ensure determinism for the device-dependent torch.Generator
   __lowerCAmelCase					:	str      =       self.dummy_cond_unet_upscale
   __lowerCAmelCase					:	str      =       DDPMScheduler()
   __lowerCAmelCase					:	Any      =       DDIMScheduler(prediction_type="""v_prediction"""      )
   __lowerCAmelCase					:	Any      =       self.dummy_vae
   __lowerCAmelCase					:	List[Any]      =       self.dummy_text_encoder
   __lowerCAmelCase					:	Optional[int]      =       CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""      )

   __lowerCAmelCase					:	Optional[int]      =       self.dummy_image.cpu().permute(0       ,  2       ,  3       ,  1      )[0]
   __lowerCAmelCase					:	Any      =       Image.fromarray(np.uinta(_snake_case      )      ).convert("""RGB"""      ).resize((64, 64)      )

   # make sure here that pndm scheduler skips prk
   __lowerCAmelCase					:	str      =       StableDiffusionUpscalePipeline(
       unet=_snake_case       ,  low_res_scheduler=_snake_case       ,  scheduler=_snake_case       ,  vae=_snake_case       ,  text_encoder=_snake_case       ,  tokenizer=_snake_case       ,  max_noise_level=350       ,  )
   __lowerCAmelCase					:	Optional[int]      =       sd_pipe.to(_snake_case      )
   sd_pipe.set_progress_bar_config(disable=_snake_case      )

   __lowerCAmelCase					:	int      =       """A painting of a squirrel eating a burger"""
   __lowerCAmelCase					:	str      =       sd_pipe(
       2 * [prompt]       ,  image=2 * [low_res_image]       ,  guidance_scale=6.0       ,  noise_level=20       ,  num_inference_steps=2       ,  output_type="""np"""       ,  )
   __lowerCAmelCase					:	Optional[Any]      =       output.images
   assert image.shape[0] == 2

   __lowerCAmelCase					:	Any      =       torch.Generator(device=_snake_case      ).manual_seed(0      )
   __lowerCAmelCase					:	Tuple      =       sd_pipe(
       [prompt]       ,  image=_snake_case       ,  generator=_snake_case       ,  num_images_per_prompt=2       ,  guidance_scale=6.0       ,  noise_level=20       ,  num_inference_steps=2       ,  output_type="""np"""       ,  )
   __lowerCAmelCase					:	Union[str, Any]      =       output.images
   assert image.shape[0] == 2






  @unittest.skipIf(torch_device != """cuda"""       ,  """This test requires a GPU"""      )
  def 				UpperCAmelCase__						(							self			:					Any      )->Any:






   '''simple docstring'''

   __lowerCAmelCase					:	Dict      =       self.dummy_cond_unet_upscale
   __lowerCAmelCase					:	Dict      =       DDPMScheduler()
   __lowerCAmelCase					:	int      =       DDIMScheduler(prediction_type="""v_prediction"""      )
   __lowerCAmelCase					:	Dict      =       self.dummy_vae
   __lowerCAmelCase					:	List[str]      =       self.dummy_text_encoder
   __lowerCAmelCase					:	List[Any]      =       CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""      )

   __lowerCAmelCase					:	Optional[int]      =       self.dummy_image.cpu().permute(0       ,  2       ,  3       ,  1      )[0]
   __lowerCAmelCase					:	List[str]      =       Image.fromarray(np.uinta(_snake_case      )      ).convert("""RGB"""      ).resize((64, 64)      )

   # put models in fp16, except vae as it overflows in fp16
   __lowerCAmelCase					:	str      =       unet.half()
   __lowerCAmelCase					:	Dict      =       text_encoder.half()

   # make sure here that pndm scheduler skips prk
   __lowerCAmelCase					:	Tuple      =       StableDiffusionUpscalePipeline(
       unet=_snake_case       ,  low_res_scheduler=_snake_case       ,  scheduler=_snake_case       ,  vae=_snake_case       ,  text_encoder=_snake_case       ,  tokenizer=_snake_case       ,  max_noise_level=350       ,  )
   __lowerCAmelCase					:	List[str]      =       sd_pipe.to(_snake_case      )
   sd_pipe.set_progress_bar_config(disable=_snake_case      )

   __lowerCAmelCase					:	Union[str, Any]      =       """A painting of a squirrel eating a burger"""
   __lowerCAmelCase					:	Any      =       torch.manual_seed(0      )
   __lowerCAmelCase					:	Union[str, Any]      =       sd_pipe(
       [prompt]       ,  image=_snake_case       ,  generator=_snake_case       ,  num_inference_steps=2       ,  output_type="""np"""       ,  ).images

   __lowerCAmelCase					:	Tuple      =       low_res_image.size[0] * 4
   assert image.shape == (1, expected_height_width, expected_height_width, 3)






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






  def 				UpperCAmelCase__						(							self			:					int      )->Any:






   '''simple docstring'''

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






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






   '''simple docstring'''

   __lowerCAmelCase					:	Union[str, Any]      =       load_image(
       """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
       """/sd2-upscale/low_res_cat.png"""      )
   __lowerCAmelCase					:	List[Any]      =       load_numpy(
       """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-upscale"""
       """/upsampled_cat.npy"""      )

   __lowerCAmelCase					:	Union[str, Any]      =       """stabilityai/stable-diffusion-x4-upscaler"""
   __lowerCAmelCase					:	Optional[int]      =       StableDiffusionUpscalePipeline.from_pretrained(_snake_case      )
   pipe.to(_snake_case      )
   pipe.set_progress_bar_config(disable=_snake_case      )
   pipe.enable_attention_slicing()

   __lowerCAmelCase					:	Optional[int]      =       """a cat sitting on a park bench"""

   __lowerCAmelCase					:	Optional[int]      =       torch.manual_seed(0      )
   __lowerCAmelCase					:	List[str]      =       pipe(
       prompt=_snake_case       ,  image=_snake_case       ,  generator=_snake_case       ,  output_type="""np"""       ,  )
   __lowerCAmelCase					:	str      =       output.images[0]

   assert image.shape == (512, 512, 3)
   assert np.abs(expected_image - image      ).max() < 1E-3






  def 				UpperCAmelCase__						(							self			:					Optional[int]      )->Tuple:






   '''simple docstring'''

   __lowerCAmelCase					:	str      =       load_image(
       """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
       """/sd2-upscale/low_res_cat.png"""      )
   __lowerCAmelCase					:	List[str]      =       load_numpy(
       """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-upscale"""
       """/upsampled_cat_fp16.npy"""      )

   __lowerCAmelCase					:	Union[str, Any]      =       """stabilityai/stable-diffusion-x4-upscaler"""
   __lowerCAmelCase					:	int      =       StableDiffusionUpscalePipeline.from_pretrained(
       _snake_case       ,  torch_dtype=torch.floataa       ,  )
   pipe.to(_snake_case      )
   pipe.set_progress_bar_config(disable=_snake_case      )
   pipe.enable_attention_slicing()

   __lowerCAmelCase					:	Optional[int]      =       """a cat sitting on a park bench"""

   __lowerCAmelCase					:	Tuple      =       torch.manual_seed(0      )
   __lowerCAmelCase					:	Optional[int]      =       pipe(
       prompt=_snake_case       ,  image=_snake_case       ,  generator=_snake_case       ,  output_type="""np"""       ,  )
   __lowerCAmelCase					:	Optional[int]      =       output.images[0]

   assert image.shape == (512, 512, 3)
   assert np.abs(expected_image - image      ).max() < 5E-1






  def 				UpperCAmelCase__						(							self			:					int      )->Any:






   '''simple docstring'''

   torch.cuda.empty_cache()
   torch.cuda.reset_max_memory_allocated()
   torch.cuda.reset_peak_memory_stats()

   __lowerCAmelCase					:	Tuple      =       load_image(
       """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
       """/sd2-upscale/low_res_cat.png"""      )

   __lowerCAmelCase					:	List[Any]      =       """stabilityai/stable-diffusion-x4-upscaler"""
   __lowerCAmelCase					:	int      =       StableDiffusionUpscalePipeline.from_pretrained(
       _snake_case       ,  torch_dtype=torch.floataa       ,  )
   pipe.to(_snake_case      )
   pipe.set_progress_bar_config(disable=_snake_case      )
   pipe.enable_attention_slicing(1      )
   pipe.enable_sequential_cpu_offload()

   __lowerCAmelCase					:	Optional[int]      =       """a cat sitting on a park bench"""

   __lowerCAmelCase					:	List[str]      =       torch.manual_seed(0      )
   __lowerCAmelCase					:	Any      =       pipe(
       prompt=_snake_case       ,  image=_snake_case       ,  generator=_snake_case       ,  num_inference_steps=5       ,  output_type="""np"""       ,  )

   __lowerCAmelCase					:	str      =       torch.cuda.max_memory_allocated()
   # make sure that less than 2.9 GB is allocated
   assert mem_bytes < 2.9 * 10**9






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


_UpperCAmelCase             =							'▁'

_UpperCAmelCase             =							{'vocab_file': 'spiece.model'}

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

_UpperCAmelCase             =							{
    'google/pegasus-xsum': 512,
}


_UpperCAmelCase             =							logging.get_logger(__name__)






class        snake_case_    (		__lowercase						):
  A_      							=   VOCAB_FILES_NAMES

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






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






   '''simple docstring'''

   __lowerCAmelCase					:	Union[str, Any]      =       offset
   if additional_special_tokens is not None:
    if not isinstance(_snake_case       ,  _snake_case      ):
     raise TypeError(
         F'''additional_special_tokens should be of type {type(_snake_case      )}, but is'''
         F''' {type(_snake_case      )}'''      )

    __lowerCAmelCase					:	List[str]      =       (
        ([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(_snake_case      )       ,  self.offset - 1      )
    ]

    if len(set(_snake_case      )      ) != len(_snake_case      ):
     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					:	Dict      =       additional_special_tokens_extended
   else:
    __lowerCAmelCase					:	Tuple      =       [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					:	Optional[int]      =       {} if sp_model_kwargs is None else sp_model_kwargs

   super().__init__(
       eos_token=_snake_case       ,  unk_token=_snake_case       ,  mask_token=_snake_case       ,  pad_token=_snake_case       ,  mask_token_sent=_snake_case       ,  offset=_snake_case       ,  additional_special_tokens=_snake_case       ,  sp_model_kwargs=self.sp_model_kwargs       ,  **_snake_case       ,  )
   __lowerCAmelCase					:	Optional[Any]      =       mask_token_sent
   __lowerCAmelCase					:	Any      =       vocab_file
   __lowerCAmelCase					:	Any      =       spm.SentencePieceProcessor(**self.sp_model_kwargs      )
   self.sp_model.Load(_snake_case      )

   # add special tokens to encoder dict
   __lowerCAmelCase					:	Dict[int, str]      =       {
       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					:	Dict[str, int]      =       {v: k for k, v in self.encoder.items()}






  @property
  def 				UpperCAmelCase__						(							self			:					str      )->int:






   '''simple docstring'''

   return len(self.sp_model      ) + self.offset






  def 				UpperCAmelCase__						(							self			:					Dict      )->Dict[str, int]:






   '''simple docstring'''

   __lowerCAmelCase					:	Tuple      =       {self.convert_ids_to_tokens(_snake_case      ): i for i in range(self.vocab_size      )}
   vocab.update(self.added_tokens_encoder      )
   return vocab






  def __getstate__(							self			:					Optional[int]      )->str:






   '''simple docstring'''

   __lowerCAmelCase					:	Optional[Any]      =       self.__dict__.copy()
   __lowerCAmelCase					:	Union[str, Any]      =       None
   return state






  def __setstate__(							self			:					Any       ,  _snake_case			:					str      )->Any:






   '''simple docstring'''

   __lowerCAmelCase					:	str      =       d

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

   __lowerCAmelCase					:	Optional[int]      =       spm.SentencePieceProcessor(**self.sp_model_kwargs      )
   self.sp_model.Load(self.vocab_file      )






  def 				UpperCAmelCase__						(							self			:					Dict       ,  _snake_case			:					str      )->List[str]:






   '''simple docstring'''

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






  def 				UpperCAmelCase__						(							self			:					Tuple       ,  _snake_case			:					str      )->int:






   '''simple docstring'''

   if token in self.decoder:
    return self.decoder[token]
   elif token in self.added_tokens_decoder:
    return self.added_tokens_decoder[token]
   __lowerCAmelCase					:	Any      =       self.sp_model.piece_to_id(_snake_case      )
   return sp_id + self.offset






  def 				UpperCAmelCase__						(							self			:					List[Any]       ,  _snake_case			:					int      )->str:






   '''simple docstring'''

   if index in self.encoder:
    return self.encoder[index]
   elif index in self.added_tokens_encoder:
    return self.added_tokens_encoder[index]
   else:
    __lowerCAmelCase					:	Optional[int]      =       self.sp_model.IdToPiece(index - self.offset      )
   return token






  def 				UpperCAmelCase__						(							self			:					Union[str, Any]       ,  _snake_case			:					Optional[int]      )->List[str]:






   '''simple docstring'''

   __lowerCAmelCase					:	Any      =       []
   __lowerCAmelCase					:	Dict      =       """"""
   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(_snake_case      ) + token
     __lowerCAmelCase					:	int      =       []
    else:
     current_sub_tokens.append(_snake_case      )
   out_string += self.sp_model.decode(_snake_case      )
   return out_string.strip()






  def 				UpperCAmelCase__						(							self			:					Optional[int]       ,  _snake_case			:					Dict=False      )->int:






   '''simple docstring'''

   return 1






  def 				UpperCAmelCase__						(							self			:					Tuple       ,  _snake_case			:					Tuple      )->str:






   '''simple docstring'''

   __lowerCAmelCase					:	List[Any]      =       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 				UpperCAmelCase__						(							self			:					List[str]       ,  _snake_case			:					List       ,  _snake_case			:					Optional[List] = None       ,  _snake_case			:					bool = False      )->List[int]:






   '''simple docstring'''

   if already_has_special_tokens:
    return self._special_token_mask(_snake_case      )
   elif token_ids_a is None:
    return self._special_token_mask(_snake_case      ) + [1]
   else:
    return self._special_token_mask(token_ids_a + token_ids_a      ) + [1]






  def 				UpperCAmelCase__						(							self			:					Any       ,  _snake_case			:					Union[str, Any]       ,  _snake_case			:					Tuple=None      )->List[int]:






   '''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 				UpperCAmelCase__						(							self			:					Any       ,  _snake_case			:					str       ,  _snake_case			:					Optional[str] = None      )->Tuple[str]:






   '''simple docstring'''

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

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

   return (out_vocab_file,)