




'''simple docstring'''







import os
from shutil import copyfile
from typing import TYPE_CHECKING, Optional, Tuple

from tokenizers import processors

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import is_sentencepiece_available, logging
from ...utils.versions import require_version


if TYPE_CHECKING:
			from transformers.pipelines.conversational import Conversation

require_version('tokenizers>=0.13.3')

if is_sentencepiece_available():
			from .tokenization_llama import LlamaTokenizer
else:
			_lowerCamelCase							:						Dict							     =  None

_lowerCamelCase							:						Tuple							     =  logging.get_logger(__name__)
_lowerCamelCase							:						Optional[Any]							     =  {'vocab_file': 'tokenizer.model', 'tokenizer_file': 'tokenizer.json'}

_lowerCamelCase							:						Dict							     =  '[INST]', '[/INST]'
_lowerCamelCase							:						int							     =  '<<SYS>>\n', '\n<</SYS>>\n\n'

# fmt: off
_lowerCamelCase							:						Optional[int]							     =  'You are a helpful, respectful and honest assistant. Always answer as helpfully as possible, while being safe. Your\\nanswers should not include any harmful, unethical, racist, sexist, toxic, dangerous, or illegal content. Please ensure\\nthat your responses are socially unbiased and positive in nature.\n\nIf a question does not make any sense, or is not factually coherent, explain why instead of answering something not\\ncorrect. If you don\'t know the answer to a question, please don\'t share false information.'





class 					lowerCamelCase_							(	UpperCAmelCase_							):




					'''simple docstring'''






					lowercase__   =			VOCAB_FILES_NAMES
					lowercase__   =			LlamaTokenizer
					lowercase__   =			"""left"""
					lowercase__   =			["""input_ids""", """attention_mask"""]






					def __init__(   self,  __UpperCamelCase=None,  __UpperCamelCase=None,  __UpperCamelCase=False,  __UpperCamelCase="<unk>",  __UpperCamelCase="<s>",  __UpperCamelCase="</s>",  __UpperCamelCase=True,  __UpperCamelCase=False,  **__UpperCamelCase,  ):
							super().__init__(
							    vocab_file=__UpperCamelCase,  tokenizer_file=__UpperCamelCase,  clean_up_tokenization_spaces=__UpperCamelCase,  unk_token=__UpperCamelCase,  bos_token=__UpperCamelCase,  eos_token=__UpperCamelCase,  **__UpperCamelCase,  )
							_snake_case     	=	add_bos_token
							_snake_case     	=	add_eos_token
							self.update_post_processor()

							_snake_case     	=	vocab_file
							_snake_case     	=	False if not self.vocab_file else True






					def     __SCREAMING_SNAKE_CASE     (   self			):
							_snake_case     	=	self.bos_token
							_snake_case     	=	self.bos_token_id

							_snake_case     	=	self.eos_token
							_snake_case     	=	self.eos_token_id

							_snake_case     	=	f'''{(bos+':0 ') * self.add_bos_token}$A:0{(' '+eos+':0') * self.add_eos_token}'''
							_snake_case     	=	f'''{single}{(' '+bos+':1') * self.add_bos_token} $B:1{(' '+eos+':1') * self.add_eos_token}'''

							_snake_case     	=	[]
							if self.add_bos_token:
									special_tokens.append((bos, bos_token_id)			)
							if self.add_eos_token:
									special_tokens.append((eos, eos_token_id)			)
							_snake_case     	=	processors.TemplateProcessing(
							    single=__UpperCamelCase,  pair=__UpperCamelCase,  special_tokens=__UpperCamelCase			)






					@property
					def     __SCREAMING_SNAKE_CASE     (   self			):
							return self._add_eos_token






					@property
					def     __SCREAMING_SNAKE_CASE     (   self			):
							return self._add_bos_token






					@add_eos_token.setter
					def     __SCREAMING_SNAKE_CASE     (   self,  __UpperCamelCase			):
							_snake_case     	=	value
							self.update_post_processor()






					@add_bos_token.setter
					def     __SCREAMING_SNAKE_CASE     (   self,  __UpperCamelCase			):
							_snake_case     	=	value
							self.update_post_processor()






					def     __SCREAMING_SNAKE_CASE     (   self,  __UpperCamelCase,  __UpperCamelCase = None			):
							if not self.can_save_slow_tokenizer:
									raise ValueError(
									    "Your fast tokenizer does not have the necessary information to save the vocabulary for a slow "
									    "tokenizer."			)

							if not os.path.isdir(__UpperCamelCase			):
									logger.error(f'''Vocabulary path ({save_directory}) should be a directory'''			)
									return
							_snake_case     	=	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			):
									copyfile(self.vocab_file,  __UpperCamelCase			)

							return (out_vocab_file,)






					def     __SCREAMING_SNAKE_CASE     (   self,  __UpperCamelCase			):
							_snake_case     	=	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...)"			)

							_snake_case     	=	[]
							if len(conversation.past_user_inputs			) > 0:
									if not conversation.past_user_inputs[0].startswith(__UpperCamelCase			) or E_SYS not in conversation.past_user_inputs[0]:
											_snake_case     	=	(
											    B_SYS + DEFAULT_SYSTEM_PROMPT + E_SYS + conversation.past_user_inputs[0]
											)
							elif not dialogue[0][1].startswith(__UpperCamelCase			) or E_SYS not in dialogue[0][1]:
									_snake_case     	=	(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=__UpperCamelCase			)
							        + [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=__UpperCamelCase			)
							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 CLIPImageProcessor





class 					lowerCamelCase_							(	unittest.TestCase							):




					'''simple docstring'''






					def __init__(   self,  __UpperCamelCase,  __UpperCamelCase=7,  __UpperCamelCase=3,  __UpperCamelCase=18,  __UpperCamelCase=30,  __UpperCamelCase=400,  __UpperCamelCase=True,  __UpperCamelCase=None,  __UpperCamelCase=True,  __UpperCamelCase=None,  __UpperCamelCase=True,  __UpperCamelCase=[0.48_14_54_66, 0.4_57_82_75, 0.40_82_10_73],  __UpperCamelCase=[0.26_86_29_54, 0.26_13_02_58, 0.27_57_77_11],  __UpperCamelCase=True,  ):
							_snake_case     	=	size if size is not None else {"shortest_edge": 20}
							_snake_case     	=	crop_size if crop_size is not None else {"height": 18, "width": 18}
							_snake_case     	=	parent
							_snake_case     	=	batch_size
							_snake_case     	=	num_channels
							_snake_case     	=	image_size
							_snake_case     	=	min_resolution
							_snake_case     	=	max_resolution
							_snake_case     	=	do_resize
							_snake_case     	=	size
							_snake_case     	=	do_center_crop
							_snake_case     	=	crop_size
							_snake_case     	=	do_normalize
							_snake_case     	=	image_mean
							_snake_case     	=	image_std
							_snake_case     	=	do_convert_rgb






					def     __SCREAMING_SNAKE_CASE     (   self			):
							return {
							    "do_resize": self.do_resize,
							    "size": self.size,
							    "do_center_crop": self.do_center_crop,
							    "crop_size": self.crop_size,
							    "do_normalize": self.do_normalize,
							    "image_mean": self.image_mean,
							    "image_std": self.image_std,
							    "do_convert_rgb": self.do_convert_rgb,
							}






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

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

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

							if torchify:
									_snake_case     	=	[torch.from_numpy(__UpperCamelCase			) for x in image_inputs]

							return image_inputs





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




					'''simple docstring'''






					lowercase__   =			CLIPImageProcessor if is_vision_available() else None






					def     __SCREAMING_SNAKE_CASE     (   self			):
							_snake_case     	=	CLIPImageProcessingTester(self			)






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






					def     __SCREAMING_SNAKE_CASE     (   self			):
							_snake_case     	=	self.image_processing_class(**self.image_processor_dict			)
							self.assertTrue(hasattr(__UpperCamelCase,  "do_resize"			)			)
							self.assertTrue(hasattr(__UpperCamelCase,  "size"			)			)
							self.assertTrue(hasattr(__UpperCamelCase,  "do_center_crop"			)			)
							self.assertTrue(hasattr(__UpperCamelCase,  "center_crop"			)			)
							self.assertTrue(hasattr(__UpperCamelCase,  "do_normalize"			)			)
							self.assertTrue(hasattr(__UpperCamelCase,  "image_mean"			)			)
							self.assertTrue(hasattr(__UpperCamelCase,  "image_std"			)			)
							self.assertTrue(hasattr(__UpperCamelCase,  "do_convert_rgb"			)			)






					def     __SCREAMING_SNAKE_CASE     (   self			):
							_snake_case     	=	self.image_processing_class.from_dict(self.image_processor_dict			)
							self.assertEqual(image_processor.size,  {"shortest_edge": 20}			)
							self.assertEqual(image_processor.crop_size,  {"height": 18, "width": 18}			)

							_snake_case     	=	self.image_processing_class.from_dict(self.image_processor_dict,  size=42,  crop_size=84			)
							self.assertEqual(image_processor.size,  {"shortest_edge": 42}			)
							self.assertEqual(image_processor.crop_size,  {"height": 84, "width": 84}			)






					def     __SCREAMING_SNAKE_CASE     (   self			):
							pass






					def     __SCREAMING_SNAKE_CASE     (   self			):
							# Initialize image_processing
							_snake_case     	=	self.image_processing_class(**self.image_processor_dict			)
							# create random PIL images
							_snake_case     	=	self.image_processor_tester.prepare_inputs(equal_resolution=__UpperCamelCase			)
							for image in image_inputs:
									self.assertIsInstance(__UpperCamelCase,  Image.Image			)

							# Test not batched input
							_snake_case     	=	image_processing(image_inputs[0],  return_tensors="pt"			).pixel_values
							self.assertEqual(
							    encoded_images.shape,  (
							        1,
							        self.image_processor_tester.num_channels,
							        self.image_processor_tester.crop_size["height"],
							        self.image_processor_tester.crop_size["width"],
							    ),  )

							# Test batched
							_snake_case     	=	image_processing(__UpperCamelCase,  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.crop_size["height"],
							        self.image_processor_tester.crop_size["width"],
							    ),  )






					def     __SCREAMING_SNAKE_CASE     (   self			):
							# Initialize image_processing
							_snake_case     	=	self.image_processing_class(**self.image_processor_dict			)
							# create random numpy tensors
							_snake_case     	=	self.image_processor_tester.prepare_inputs(equal_resolution=__UpperCamelCase,  numpify=__UpperCamelCase			)
							for image in image_inputs:
									self.assertIsInstance(__UpperCamelCase,  np.ndarray			)

							# Test not batched input
							_snake_case     	=	image_processing(image_inputs[0],  return_tensors="pt"			).pixel_values
							self.assertEqual(
							    encoded_images.shape,  (
							        1,
							        self.image_processor_tester.num_channels,
							        self.image_processor_tester.crop_size["height"],
							        self.image_processor_tester.crop_size["width"],
							    ),  )

							# Test batched
							_snake_case     	=	image_processing(__UpperCamelCase,  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.crop_size["height"],
							        self.image_processor_tester.crop_size["width"],
							    ),  )






					def     __SCREAMING_SNAKE_CASE     (   self			):
							# Initialize image_processing
							_snake_case     	=	self.image_processing_class(**self.image_processor_dict			)
							# create random PyTorch tensors
							_snake_case     	=	self.image_processor_tester.prepare_inputs(equal_resolution=__UpperCamelCase,  torchify=__UpperCamelCase			)
							for image in image_inputs:
									self.assertIsInstance(__UpperCamelCase,  torch.Tensor			)

							# Test not batched input
							_snake_case     	=	image_processing(image_inputs[0],  return_tensors="pt"			).pixel_values
							self.assertEqual(
							    encoded_images.shape,  (
							        1,
							        self.image_processor_tester.num_channels,
							        self.image_processor_tester.crop_size["height"],
							        self.image_processor_tester.crop_size["width"],
							    ),  )

							# Test batched
							_snake_case     	=	image_processing(__UpperCamelCase,  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.crop_size["height"],
							        self.image_processor_tester.crop_size["width"],
							    ),  )





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




					'''simple docstring'''






					lowercase__   =			CLIPImageProcessor if is_vision_available() else None






					def     __SCREAMING_SNAKE_CASE     (   self			):
							_snake_case     	=	CLIPImageProcessingTester(self,  num_channels=4			)
							_snake_case     	=	3






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






					def     __SCREAMING_SNAKE_CASE     (   self			):
							_snake_case     	=	self.image_processing_class(**self.image_processor_dict			)
							self.assertTrue(hasattr(__UpperCamelCase,  "do_resize"			)			)
							self.assertTrue(hasattr(__UpperCamelCase,  "size"			)			)
							self.assertTrue(hasattr(__UpperCamelCase,  "do_center_crop"			)			)
							self.assertTrue(hasattr(__UpperCamelCase,  "center_crop"			)			)
							self.assertTrue(hasattr(__UpperCamelCase,  "do_normalize"			)			)
							self.assertTrue(hasattr(__UpperCamelCase,  "image_mean"			)			)
							self.assertTrue(hasattr(__UpperCamelCase,  "image_std"			)			)
							self.assertTrue(hasattr(__UpperCamelCase,  "do_convert_rgb"			)			)






					def     __SCREAMING_SNAKE_CASE     (   self			):
							pass






					def     __SCREAMING_SNAKE_CASE     (   self			):
							# Initialize image_processing
							_snake_case     	=	self.image_processing_class(**self.image_processor_dict			)
							# create random PIL images
							_snake_case     	=	self.image_processor_tester.prepare_inputs(equal_resolution=__UpperCamelCase			)
							for image in image_inputs:
									self.assertIsInstance(__UpperCamelCase,  Image.Image			)

							# Test not batched input
							_snake_case     	=	image_processing(image_inputs[0],  return_tensors="pt"			).pixel_values
							self.assertEqual(
							    encoded_images.shape,  (
							        1,
							        self.expected_encoded_image_num_channels,
							        self.image_processor_tester.crop_size["height"],
							        self.image_processor_tester.crop_size["width"],
							    ),  )

							# Test batched
							_snake_case     	=	image_processing(__UpperCamelCase,  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.crop_size["height"],
							        self.image_processor_tester.crop_size["width"],
							    ),  )





