



"""simple docstring"""
import os
from shutil import copyfile
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple

import sentencepiece as spm

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


if TYPE_CHECKING:
			from transformers.pipelines.conversational import Conversation

__SCREAMING_SNAKE_CASE     :  Any										=					logging.get_logger(__name__)

__SCREAMING_SNAKE_CASE     :  Optional[Any]										=					{'''vocab_file''': '''tokenizer.model'''}

__SCREAMING_SNAKE_CASE     :  int										=					{
    '''vocab_file''': {
        '''hf-internal-testing/llama-tokenizer''': '''https://huggingface.co/hf-internal-testing/llama-tokenizer/resolve/main/tokenizer.model''',
    },
    '''tokenizer_file''': {
        '''hf-internal-testing/llama-tokenizer''': '''https://huggingface.co/hf-internal-testing/llama-tokenizer/resolve/main/tokenizer_config.json''',
    },
}
__SCREAMING_SNAKE_CASE     :  Tuple										=					{
    '''hf-internal-testing/llama-tokenizer''': 2_0_4_8,
}
__SCREAMING_SNAKE_CASE     :  Optional[int]										=					'''▁'''

__SCREAMING_SNAKE_CASE							,    __SCREAMING_SNAKE_CASE     :  int										=					'''[INST]''', '''[/INST]'''
__SCREAMING_SNAKE_CASE							,    __SCREAMING_SNAKE_CASE     :  Optional[Any]										=					'''<<SYS>>\n''', '''\n<</SYS>>\n\n'''

# fmt: off
__SCREAMING_SNAKE_CASE     :  Optional[Any]										=					'''You are a helpful, respectful and honest assistant. Always answer as helpfully as possible, while being safe. Your\
answers should not include any harmful, unethical, racist, sexist, toxic, dangerous, or illegal content. Please ensure\
that your responses are socially unbiased and positive in nature.

If a question does not make any sense, or is not factually coherent, explain why instead of answering something not\
correct. If you don\'t know the answer to a question, please don\'t share false information.'''


class 							lowerCamelCase_(      A__			):





					'''simple docstring'''



					lowercase__      :				Optional[Any]        =       VOCAB_FILES_NAMES
					lowercase__      :				List[str]        =       PRETRAINED_VOCAB_FILES_MAP
					lowercase__      :				Optional[Any]        =       PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
					lowercase__      :				Optional[int]        =       ['input_ids', 'attention_mask']

					def __init__(			self ,				lowerCamelCase__ ,				lowerCamelCase__="<unk>" ,				lowerCamelCase__="<s>" ,				lowerCamelCase__="</s>" ,				lowerCamelCase__=None ,				lowerCamelCase__ = None ,				lowerCamelCase__=True ,				lowerCamelCase__=False ,				lowerCamelCase__=False ,				lowerCamelCase__=True ,				**lowerCamelCase__ ,				):
									_lowerCamelCase					     = {} if sp_model_kwargs is None else sp_model_kwargs
									_lowerCamelCase					     = AddedToken(lowerCamelCase__ ,				lstrip=lowerCamelCase__ ,				rstrip=lowerCamelCase__      ) if isinstance(lowerCamelCase__ ,				lowerCamelCase__      ) else bos_token
									_lowerCamelCase					     = AddedToken(lowerCamelCase__ ,				lstrip=lowerCamelCase__ ,				rstrip=lowerCamelCase__      ) if isinstance(lowerCamelCase__ ,				lowerCamelCase__      ) else eos_token
									_lowerCamelCase					     = AddedToken(lowerCamelCase__ ,				lstrip=lowerCamelCase__ ,				rstrip=lowerCamelCase__      ) if isinstance(lowerCamelCase__ ,				lowerCamelCase__      ) else unk_token
									_lowerCamelCase					     = AddedToken(lowerCamelCase__ ,				lstrip=lowerCamelCase__ ,				rstrip=lowerCamelCase__      ) if isinstance(lowerCamelCase__ ,				lowerCamelCase__      ) else pad_token
									super().__init__(
									    bos_token=lowerCamelCase__ ,				eos_token=lowerCamelCase__ ,				unk_token=lowerCamelCase__ ,				pad_token=lowerCamelCase__ ,				add_bos_token=lowerCamelCase__ ,				add_eos_token=lowerCamelCase__ ,				sp_model_kwargs=self.sp_model_kwargs ,				clean_up_tokenization_spaces=lowerCamelCase__ ,				legacy=lowerCamelCase__ ,				**lowerCamelCase__ ,				)
									if legacy:
													logger.warning_once(
													    F"""You are using the legacy behaviour of the {self.__class__}. This means that tokens that come after special tokens will not be properly handled. We recommend you to"""
													    ''' read the related pull request available at https://github.com/huggingface/transformers/pull/24565'''      )
									_lowerCamelCase					     = legacy
									_lowerCamelCase					     = vocab_file
									_lowerCamelCase					     = add_bos_token
									_lowerCamelCase					     = add_eos_token
									_lowerCamelCase					     = spm.SentencePieceProcessor(**self.sp_model_kwargs      )
									self.sp_model.Load(lowerCamelCase__      )

					def __getstate__(			self      ):
									_lowerCamelCase					     = self.__dict__.copy()
									_lowerCamelCase					     = None
									_lowerCamelCase					     = self.sp_model.serialized_model_proto()
									return state

					def __setstate__(			self ,				lowerCamelCase__      ):
									_lowerCamelCase					     = d
									_lowerCamelCase					     = spm.SentencePieceProcessor(**self.sp_model_kwargs      )
									self.sp_model.LoadFromSerializedProto(self.sp_model_proto      )

					@property
					def 	snake_case__					(			self      ):
									return self.sp_model.get_piece_size()

					def 	snake_case__					(			self      ):
									_lowerCamelCase					     = {self.convert_ids_to_tokens(lowerCamelCase__      ): i for i in range(self.vocab_size      )}
									vocab.update(self.added_tokens_encoder      )
									return vocab

					def 	snake_case__					(			self ,				lowerCamelCase__ ,				**lowerCamelCase__      ):
									# Replace the SPIECE_UNDERLINE with a space to make sure SPIECE_UNDERLINE is only used at
									# the beginning of the text
									if not self.legacy:
													_lowerCamelCase					     = SPIECE_UNDERLINE + text.replace(lowerCamelCase__ ,				''' '''      )
									return super().tokenize(lowerCamelCase__ ,				**lowerCamelCase__      )

					def 	snake_case__					(			self ,				lowerCamelCase__      ):
									if not self.legacy:
													_lowerCamelCase					     = text.startswith(lowerCamelCase__      )
													if is_first:
																	_lowerCamelCase					     = text[1:]

									_lowerCamelCase					     = self.sp_model.encode(lowerCamelCase__ ,				out_type=lowerCamelCase__      )

									if not self.legacy and not is_first and not text.startswith(''' '''      ) and tokens[0].startswith(lowerCamelCase__      ):
													_lowerCamelCase					     = ([tokens[0][1:]] if len(tokens[0]      ) > 1 else []) + tokens[1:]
									return tokens

					def 	snake_case__					(			self ,				lowerCamelCase__      ):
									return self.sp_model.piece_to_id(lowerCamelCase__      )

					def 	snake_case__					(			self ,				lowerCamelCase__      ):
									_lowerCamelCase					     = self.sp_model.IdToPiece(lowerCamelCase__      )
									return token

					def 	snake_case__					(			self ,				lowerCamelCase__      ):
									_lowerCamelCase					     = []
									_lowerCamelCase					     = ''''''
									_lowerCamelCase					     = False
									for i, token in enumerate(lowerCamelCase__      ):
													# make sure that special tokens are not decoded using sentencepiece model
													if token in self.all_special_tokens:
																	if not prev_is_special and i != 0:
																					out_string += " "
																	out_string += self.sp_model.decode(lowerCamelCase__      ) + token
																	_lowerCamelCase					     = True
																	_lowerCamelCase					     = []
													else:
																	current_sub_tokens.append(lowerCamelCase__      )
																	_lowerCamelCase					     = False
									out_string += self.sp_model.decode(lowerCamelCase__      )
									return out_string

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

									return (out_vocab_file,)

					def 	snake_case__					(			self ,				lowerCamelCase__ ,				lowerCamelCase__=None      ):
									_lowerCamelCase					     = [self.bos_token_id] if self.add_bos_token else []
									_lowerCamelCase					     = [self.eos_token_id] if self.add_eos_token else []

									_lowerCamelCase					     = bos_token_id + token_ids_a + eos_token_id

									if token_ids_a is not None:
													_lowerCamelCase					     = output + bos_token_id + token_ids_a + eos_token_id

									return output

					def 	snake_case__					(			self ,				lowerCamelCase__ ,				lowerCamelCase__ = None ,				lowerCamelCase__ = False      ):
									if already_has_special_tokens:
													return super().get_special_tokens_mask(
													    token_ids_a=lowerCamelCase__ ,				token_ids_a=lowerCamelCase__ ,				already_has_special_tokens=lowerCamelCase__      )

									_lowerCamelCase					     = [1] if self.add_bos_token else []
									_lowerCamelCase					     = [1] if self.add_eos_token else []

									if token_ids_a is None:
													return bos_token_id + ([0] * len(lowerCamelCase__      )) + eos_token_id
									return (
									    bos_token_id
									    + ([0] * len(lowerCamelCase__      ))
									    + eos_token_id
									    + bos_token_id
									    + ([0] * len(lowerCamelCase__      ))
									    + eos_token_id
									)

					def 	snake_case__					(			self ,				lowerCamelCase__ ,				lowerCamelCase__ = None      ):
									_lowerCamelCase					     = [self.bos_token_id] if self.add_bos_token else []
									_lowerCamelCase					     = [self.eos_token_id] if self.add_eos_token else []

									_lowerCamelCase					     = [0] * len(bos_token_id + token_ids_a + eos_token_id      )

									if token_ids_a is not None:
													output += [1] * len(bos_token_id + token_ids_a + eos_token_id      )

									return output





					def 	snake_case__					(			self ,				lowerCamelCase__      ):
									_lowerCamelCase					     = list(conversation.iter_texts()      )
									if not all([is_user for is_user, msg in dialogue[::2]]      ) or not all(
									    [not is_user for is_user, msg in dialogue[1::2]]      ):
													raise ValueError(
													    '''The model only supports \'user\' and \'assistant\' roles, starting with user and alternating (u/a/u/a/u...)'''      )

									_lowerCamelCase					     = []
									if len(conversation.past_user_inputs      ) > 0:
													if not conversation.past_user_inputs[0].startswith(lowerCamelCase__      ) or E_SYS not in conversation.past_user_inputs[0]:
																	_lowerCamelCase					     = (
																	    B_SYS + DEFAULT_SYSTEM_PROMPT + E_SYS + conversation.past_user_inputs[0]
																	)
									elif not dialogue[0][1].startswith(lowerCamelCase__      ) or E_SYS not in dialogue[0][1]:
													_lowerCamelCase					     = (dialogue[0][0], B_SYS + DEFAULT_SYSTEM_PROMPT + E_SYS + dialogue[0][1])

									dialog_tokens += sum(
									    [
									        [self.bos_token_id]
									        + self.encode(
									            F"""{B_INST} {(prompt[1]).strip()} {E_INST} {(answer[1]).strip()} """ ,				add_special_tokens=lowerCamelCase__      )
									        + [self.eos_token_id]
									        for prompt, answer in zip(dialogue[::2] ,				dialogue[1::2]      )
									    ] ,				[] ,				)
									if not (dialogue[-1][0]):
													raise ValueError(F"""Last message must be from user, got {dialogue[-1]["role"]}"""      )
									dialog_tokens += [self.bos_token_id] + self.encode(
									    F"""{B_INST} {(dialogue[-1][1]).strip()} {E_INST}""" ,				add_special_tokens=lowerCamelCase__      )
									return dialog_tokens





"""simple docstring"""
import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin


if is_torch_available():
			import torch

if is_vision_available():
			from PIL import Image

			from transformers import BlipImageProcessor


class 							lowerCamelCase_(      unittest.TestCase			):





					'''simple docstring'''



					def __init__(			self ,				lowerCamelCase__ ,				lowerCamelCase__=7 ,				lowerCamelCase__=3 ,				lowerCamelCase__=1_8 ,				lowerCamelCase__=3_0 ,				lowerCamelCase__=4_0_0 ,				lowerCamelCase__=True ,				lowerCamelCase__=None ,				lowerCamelCase__=True ,				lowerCamelCase__=False ,				lowerCamelCase__=[0.4_8_1_4_5_4_6_6, 0.4_5_7_8_2_7_5, 0.4_0_8_2_1_0_7_3] ,				lowerCamelCase__=[0.2_6_8_6_2_9_5_4, 0.2_6_1_3_0_2_5_8, 0.2_7_5_7_7_7_1_1] ,				lowerCamelCase__=True ,				):
									_lowerCamelCase					     = size if size is not None else {'''height''': 2_0, '''width''': 2_0}
									_lowerCamelCase					     = parent
									_lowerCamelCase					     = batch_size
									_lowerCamelCase					     = num_channels
									_lowerCamelCase					     = image_size
									_lowerCamelCase					     = min_resolution
									_lowerCamelCase					     = max_resolution
									_lowerCamelCase					     = do_resize
									_lowerCamelCase					     = size
									_lowerCamelCase					     = do_normalize
									_lowerCamelCase					     = image_mean
									_lowerCamelCase					     = image_std
									_lowerCamelCase					     = do_pad
									_lowerCamelCase					     = do_convert_rgb

					def 	snake_case__					(			self      ):
									return {
									    "do_resize": self.do_resize,
									    "size": self.size,
									    "do_normalize": self.do_normalize,
									    "image_mean": self.image_mean,
									    "image_std": self.image_std,
									    "do_convert_rgb": self.do_convert_rgb,
									    "do_pad": self.do_pad,
									}





					def 	snake_case__					(			self ,				lowerCamelCase__=False ,				lowerCamelCase__=False ,				lowerCamelCase__=False      ):
									assert not (numpify and torchify), "You cannot specify both numpy and PyTorch tensors at the same time"

									if equal_resolution:
													_lowerCamelCase					     = []
													for i in range(self.batch_size      ):
																	image_inputs.append(
																	    np.random.randint(
																	        2_5_5 ,				size=(self.num_channels, self.max_resolution, self.max_resolution) ,				dtype=np.uinta      )      )
									else:
													_lowerCamelCase					     = []
													for i in range(self.batch_size      ):
																	_lowerCamelCase     ,   _lowerCamelCase					     = np.random.choice(np.arange(self.min_resolution ,				self.max_resolution      ) ,				2      )
																	image_inputs.append(np.random.randint(2_5_5 ,				size=(self.num_channels, width, height) ,				dtype=np.uinta      )      )

									if not numpify and not torchify:
													# PIL expects the channel dimension as last dimension
													_lowerCamelCase					     = [Image.fromarray(np.moveaxis(lowerCamelCase__ ,				0 ,				-1      )      ) for x in image_inputs]

									if torchify:
													_lowerCamelCase					     = [torch.from_numpy(lowerCamelCase__      ) for x in image_inputs]

									return image_inputs


@require_torch
@require_vision
class 							lowerCamelCase_(      A__, unittest.TestCase			):





					'''simple docstring'''



					lowercase__      :				Any        =       BlipImageProcessor if is_vision_available() else None

					def 	snake_case__					(			self      ):
									_lowerCamelCase					     = BlipImageProcessingTester(self      )

					@property
					def 	snake_case__					(			self      ):
									return self.image_processor_tester.prepare_image_processor_dict()

					def 	snake_case__					(			self      ):
									_lowerCamelCase					     = self.image_processing_class(**self.image_processor_dict      )
									self.assertTrue(hasattr(lowerCamelCase__ ,				'''do_resize'''      )      )
									self.assertTrue(hasattr(lowerCamelCase__ ,				'''size'''      )      )
									self.assertTrue(hasattr(lowerCamelCase__ ,				'''do_normalize'''      )      )
									self.assertTrue(hasattr(lowerCamelCase__ ,				'''image_mean'''      )      )
									self.assertTrue(hasattr(lowerCamelCase__ ,				'''image_std'''      )      )
									self.assertTrue(hasattr(lowerCamelCase__ ,				'''do_convert_rgb'''      )      )

					def 	snake_case__					(			self      ):
									pass

					def 	snake_case__					(			self      ):
									# Initialize image_processor
									_lowerCamelCase					     = self.image_processing_class(**self.image_processor_dict      )
									# create random PIL images
									_lowerCamelCase					     = self.image_processor_tester.prepare_inputs(equal_resolution=lowerCamelCase__      )
									for image in image_inputs:
													self.assertIsInstance(lowerCamelCase__ ,				Image.Image      )

									# Test not batched input
									_lowerCamelCase					     = image_processor(image_inputs[0] ,				return_tensors='''pt'''      ).pixel_values
									self.assertEqual(
									    encoded_images.shape ,				(
									        1,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.size['''height'''],
									        self.image_processor_tester.size['''width'''],
									    ) ,				)

									# Test batched
									_lowerCamelCase					     = image_processor(lowerCamelCase__ ,				return_tensors='''pt'''      ).pixel_values
									self.assertEqual(
									    encoded_images.shape ,				(
									        self.image_processor_tester.batch_size,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.size['''height'''],
									        self.image_processor_tester.size['''width'''],
									    ) ,				)

					def 	snake_case__					(			self      ):
									# Initialize image_processor
									_lowerCamelCase					     = self.image_processing_class(**self.image_processor_dict      )
									# create random numpy tensors
									_lowerCamelCase					     = self.image_processor_tester.prepare_inputs(equal_resolution=lowerCamelCase__ ,				numpify=lowerCamelCase__      )
									for image in image_inputs:
													self.assertIsInstance(lowerCamelCase__ ,				np.ndarray      )

									# Test not batched input
									_lowerCamelCase					     = image_processor(image_inputs[0] ,				return_tensors='''pt'''      ).pixel_values
									self.assertEqual(
									    encoded_images.shape ,				(
									        1,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.size['''height'''],
									        self.image_processor_tester.size['''width'''],
									    ) ,				)

									# Test batched
									_lowerCamelCase					     = image_processor(lowerCamelCase__ ,				return_tensors='''pt'''      ).pixel_values
									self.assertEqual(
									    encoded_images.shape ,				(
									        self.image_processor_tester.batch_size,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.size['''height'''],
									        self.image_processor_tester.size['''width'''],
									    ) ,				)





					def 	snake_case__					(			self      ):
									# Initialize image_processor
									_lowerCamelCase					     = self.image_processing_class(**self.image_processor_dict      )
									# create random PyTorch tensors
									_lowerCamelCase					     = self.image_processor_tester.prepare_inputs(equal_resolution=lowerCamelCase__ ,				torchify=lowerCamelCase__      )
									for image in image_inputs:
													self.assertIsInstance(lowerCamelCase__ ,				torch.Tensor      )

									# Test not batched input
									_lowerCamelCase					     = image_processor(image_inputs[0] ,				return_tensors='''pt'''      ).pixel_values
									self.assertEqual(
									    encoded_images.shape ,				(
									        1,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.size['''height'''],
									        self.image_processor_tester.size['''width'''],
									    ) ,				)

									# Test batched
									_lowerCamelCase					     = image_processor(lowerCamelCase__ ,				return_tensors='''pt'''      ).pixel_values
									self.assertEqual(
									    encoded_images.shape ,				(
									        self.image_processor_tester.batch_size,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.size['''height'''],
									        self.image_processor_tester.size['''width'''],
									    ) ,				)


@require_torch
@require_vision
class 							lowerCamelCase_(      A__, unittest.TestCase			):





					'''simple docstring'''



					lowercase__      :				Tuple        =       BlipImageProcessor if is_vision_available() else None

					def 	snake_case__					(			self      ):
									_lowerCamelCase					     = BlipImageProcessingTester(self ,				num_channels=4      )
									_lowerCamelCase					     = 3

					@property
					def 	snake_case__					(			self      ):
									return self.image_processor_tester.prepare_image_processor_dict()

					def 	snake_case__					(			self      ):
									_lowerCamelCase					     = self.image_processing_class(**self.image_processor_dict      )
									self.assertTrue(hasattr(lowerCamelCase__ ,				'''do_resize'''      )      )
									self.assertTrue(hasattr(lowerCamelCase__ ,				'''size'''      )      )
									self.assertTrue(hasattr(lowerCamelCase__ ,				'''do_normalize'''      )      )
									self.assertTrue(hasattr(lowerCamelCase__ ,				'''image_mean'''      )      )
									self.assertTrue(hasattr(lowerCamelCase__ ,				'''image_std'''      )      )
									self.assertTrue(hasattr(lowerCamelCase__ ,				'''do_convert_rgb'''      )      )

					def 	snake_case__					(			self      ):
									pass





					def 	snake_case__					(			self      ):
									# Initialize image_processor
									_lowerCamelCase					     = self.image_processing_class(**self.image_processor_dict      )
									# create random PIL images
									_lowerCamelCase					     = self.image_processor_tester.prepare_inputs(equal_resolution=lowerCamelCase__      )
									for image in image_inputs:
													self.assertIsInstance(lowerCamelCase__ ,				Image.Image      )

									# Test not batched input
									_lowerCamelCase					     = image_processor(image_inputs[0] ,				return_tensors='''pt'''      ).pixel_values
									self.assertEqual(
									    encoded_images.shape ,				(
									        1,
									        self.expected_encoded_image_num_channels,
									        self.image_processor_tester.size['''height'''],
									        self.image_processor_tester.size['''width'''],
									    ) ,				)

									# Test batched
									_lowerCamelCase					     = image_processor(lowerCamelCase__ ,				return_tensors='''pt'''      ).pixel_values
									self.assertEqual(
									    encoded_images.shape ,				(
									        self.image_processor_tester.batch_size,
									        self.expected_encoded_image_num_channels,
									        self.image_processor_tester.size['''height'''],
									        self.image_processor_tester.size['''width'''],
									    ) ,				)
