



import json
import os
from functools import lru_cache
from typing import TYPE_CHECKING, List, Optional, Tuple, Union

import numpy as np
import regex as re

from ...utils import is_tf_available, is_torch_available, logging


if TYPE_CHECKING:
	if is_torch_available():
		import torch
	if is_tf_available():
		import tensorflow as tf

from ...tokenization_utils import AddedToken, PreTrainedTokenizer


UpperCAmelCase								=   logging.get_logger(__name__)

UpperCAmelCase								=   {
    """vocab_file""": """vocab.json""",
    """merges_file""": """merges.txt""",
}

UpperCAmelCase								=   {
    """vocab_file""": {
        """Salesforce/codegen-350M-mono""": """https://huggingface.co/Salesforce/codegen-350M-mono/resolve/main/vocab.json""",
    },
    """merges_file""": {
        """Salesforce/codegen-350M-mono""": """https://huggingface.co/Salesforce/codegen-350M-mono/resolve/main/merges.txt""",
    },
}

UpperCAmelCase								=   {
    """Salesforce/codegen-350M-mono""": 2_048,
}

@lru_cache()
def      A_ (		):

				"""simple docstring"""
				a__			 =						(
				    list(range(ord("""!"""						)	,						ord("""~"""						) + 1						)						) + list(range(ord("""¡"""						)	,						ord("""¬"""						) + 1						)						) + list(range(ord("""®"""						)	,						ord("""ÿ"""						) + 1						)						)
				)
				a__			 =						bs[:]
				a__			 =						0
				for b in range(2**8						):
								if b not in bs:
												bs.append(__a						)
												cs.append(2**8 + n						)
												n += 1
				a__			 =						[chr(__a						) for n in cs]
				return dict(zip(__a	,						__a						)						)

def      A_ (		__a       :			Union[str, Any]						):

				"""simple docstring"""
				a__			 =						set()
				a__			 =						word[0]
				for char in word[1:]:
								pairs.add((prev_char, char)						)
								a__			 =						char
				return pairs




class 	__snake_case (				SCREAMING_SNAKE_CASE):







							'''simple docstring'''






							UpperCamelCase__			: str             =   VOCAB_FILES_NAMES
							UpperCamelCase__			: int             =   PRETRAINED_VOCAB_FILES_MAP
							UpperCamelCase__			: Dict             =   PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							UpperCamelCase__			: Optional[int]             =   ["""input_ids""", """attention_mask"""]


							def __init__(							self					,					a_					,					a_					,					a_="replace"					,					a_="<|endoftext|>"					,					a_="<|endoftext|>"					,					a_="<|endoftext|>"					,					a_=None					,					a_=False					,					a_=False					,					**a_					,					):
											a__			 =						AddedToken(a_					,					lstrip=a_					,					rstrip=a_						) if isinstance(a_					,					a_						) else bos_token
											a__			 =						AddedToken(a_					,					lstrip=a_					,					rstrip=a_						) if isinstance(a_					,					a_						) else eos_token
											a__			 =						AddedToken(a_					,					lstrip=a_					,					rstrip=a_						) if isinstance(a_					,					a_						) else unk_token
											a__			 =						AddedToken(a_					,					lstrip=a_					,					rstrip=a_						) if isinstance(a_					,					a_						) else pad_token
											super().__init__(
											    errors=a_					,					unk_token=a_					,					bos_token=a_					,					eos_token=a_					,					pad_token=a_					,					add_prefix_space=a_					,					add_bos_token=a_					,					**a_					,					)
											a__			 =						add_bos_token

											with open(a_					,					encoding="""utf-8"""						) as vocab_handle:
															a__			 =						json.load(a_						)
											a__			 =						{v: k for k, v in self.encoder.items()}
											a__			 =						errors  # how to handle errors in decoding
											a__			 =						bytes_to_unicode()
											a__			 =						{v: k for k, v in self.byte_encoder.items()}
											with open(a_					,					encoding="""utf-8"""						) as merges_handle:
															a__			 =						merges_handle.read().split("""\n"""						)[1:-1]
											a__			 =						[tuple(merge.split()						) for merge in bpe_merges]
											a__			 =						dict(zip(a_					,					range(len(a_						)						)						)						)
											a__			 =						{}
											a__			 =						add_prefix_space

											# Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
											a__			 =						re.compile(r"""'s|'t|'re|'ve|'m|'ll|'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+"""						)


							@property
							def _a (							self						):
											return len(self.encoder						)


							def _a (							self						):
											return dict(self.encoder					,					**self.added_tokens_encoder						)


							def _a (							self					,					a_						):
											if token in self.cache:
															return self.cache[token]
											a__			 =						tuple(a_						)
											a__			 =						get_pairs(a_						)

											if not pairs:
															return token

											while True:
															a__			 =						min(a_					,					key=lambda a_						: self.bpe_ranks.get(a_					,					float("""inf"""						)						)						)
															if bigram not in self.bpe_ranks:
																			break
															a__ ,     a__			 =						bigram
															a__			 =						[]
															a__			 =						0
															while i < len(a_						):
																			try:
																							a__			 =						word.index(a_					,					a_						)
																			except ValueError:
																							new_word.extend(word[i:]						)
																							break
																			else:
																							new_word.extend(word[i:j]						)
																							a__			 =						j

																			if word[i] == first and i < len(a_						) - 1 and word[i + 1] == second:
																							new_word.append(first + second						)
																							i += 2
																			else:
																							new_word.append(word[i]						)
																							i += 1
															a__			 =						tuple(a_						)
															a__			 =						new_word
															if len(a_						) == 1:
																			break
															else:
																			a__			 =						get_pairs(a_						)
											a__			 =						""" """.join(a_						)
											a__			 =						word
											return word


							def _a (							self					,					a_					,					a_=None						):
											if self.add_bos_token:
															a__			 =						[self.bos_token_id]
											else:
															a__			 =						[]

											a__			 =						bos_token_ids + token_ids_a

											if token_ids_a is None:
															return output

											return output + bos_token_ids + token_ids_a


							def _a (							self					,					a_						):
											a__			 =						[]
											for token in re.findall(self.pat					,					a_						):
															a__			 =						"""""".join(
															    self.byte_encoder[b] for b in token.encode("""utf-8"""						)						)  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
															bpe_tokens.extend(bpe_token for bpe_token in self.bpe(a_						).split(""" """						)						)
											return bpe_tokens


							def _a (							self					,					a_						):
											return self.encoder.get(a_					,					self.encoder.get(self.unk_token						)						)


							def _a (							self					,					a_						):
											return self.decoder.get(a_						)


							def _a (							self					,					a_						):
											a__			 =						"""""".join(a_						)
											a__			 =						bytearray([self.byte_decoder[c] for c in text]						).decode("""utf-8"""					,					errors=self.errors						)
											return text


							def _a (							self					,					a_					,					a_ = None						):
											if not os.path.isdir(a_						):
															logger.error(F'''Vocabulary path ({save_directory}) should be a directory'''						)
															return
											a__			 =						os.path.join(
											    a_					,					(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]						)
											a__			 =						os.path.join(
											    a_					,					(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""merges_file"""]						)

											with open(a_					,					"""w"""					,					encoding="""utf-8"""						) as f:
															f.write(json.dumps(self.encoder					,					indent=2					,					sort_keys=a_					,					ensure_ascii=a_						) + """\n"""						)

											a__			 =						0
											with open(a_					,					"""w"""					,					encoding="""utf-8"""						) as writer:
															writer.write("""#version: 0.2\n"""						)
															for bpe_tokens, token_index in sorted(self.bpe_ranks.items()					,					key=lambda a_						: kv[1]						):
																			if index != token_index:
																							logger.warning(
																							    F'''Saving vocabulary to {merge_file}: BPE merge indices are not consecutive.'''
																							    """ Please check that the tokenizer is not corrupted!"""						)
																							a__			 =						token_index
																			writer.write(""" """.join(a_						) + """\n"""						)
																			index += 1

											return vocab_file, merge_file


							def _a (							self					,					a_					,					a_=False					,					**a_						):
											a__			 =						kwargs.pop("""add_prefix_space"""					,					self.add_prefix_space						)
											if is_split_into_words or add_prefix_space:
															a__			 =						""" """ + text
											return (text, kwargs)


							def _a (							self					,					a_					,					a_ = False					,					a_ = None					,					a_ = None					,					**a_					,					):
											a__			 =						super()._decode(
											    token_ids=a_					,					skip_special_tokens=a_					,					clean_up_tokenization_spaces=a_					,					**a_					,					)

											if truncate_before_pattern is not None and len(a_						) > 0:
															a__			 =						self.truncate(a_					,					a_						)

											return decoded_text







							def _a (							self					,					a_					,					a_						):
											def find_re(a_					,					a_					,					a_						):
															a__			 =						pattern.search(a_					,					a_						)
															return m.start() if m else -1

											a__			 =						[re.compile(a_					,					re.MULTILINE						) for pattern in truncate_before_pattern]

											a__			 =						list(re.finditer("""^print"""					,					a_					,					re.MULTILINE						)						)

											if len(a_						) > 1:
															a__			 =						completion[: prints[1].start()]

											a__			 =						list(re.finditer("""^def"""					,					a_					,					re.MULTILINE						)						)

											if len(a_						) > 1:
															a__			 =						completion[: defs[1].start()]

											a__			 =						0

											a__			 =						[
											    pos for pos in [find_re(a_					,					a_					,					a_						) for terminal in terminals] if pos != -1
											]

											if len(a_						) > 0:
															return completion[: min(a_						)]
											else:
															return completion






import shutil
import tempfile
import unittest

import numpy as np
import pytest

from transformers.testing_utils import require_vision
from transformers.utils import is_vision_available


if is_vision_available():
	from PIL import Image

	from transformers import AutoProcessor, BertTokenizer, CLIPImageProcessor, GitProcessor, PreTrainedTokenizerFast

@require_vision
class 	__snake_case (				unittest.TestCase):







							'''simple docstring'''






							def _a (							self						):
											a__			 =						tempfile.mkdtemp()

											a__			 =						CLIPImageProcessor()
											a__			 =						BertTokenizer.from_pretrained(
											    """hf-internal-testing/tiny-random-BertModel"""					,					model_input_names=["""input_ids""", """attention_mask"""]						)

											a__			 =						GitProcessor(a_					,					a_						)

											processor.save_pretrained(self.tmpdirname						)


							def _a (							self					,					**a_						):
											return AutoProcessor.from_pretrained(self.tmpdirname					,					**a_						).tokenizer


							def _a (							self					,					**a_						):
											return AutoProcessor.from_pretrained(self.tmpdirname					,					**a_						).image_processor


							def _a (							self						):
											shutil.rmtree(self.tmpdirname						)


							def _a (							self						):
											a__			 =						[np.random.randint(255					,					size=(3, 30, 400)					,					dtype=np.uinta						)]

											a__			 =						[Image.fromarray(np.moveaxis(a_					,					0					,					-1						)						) for x in image_inputs]

											return image_inputs


							def _a (							self						):
											a__			 =						GitProcessor(tokenizer=self.get_tokenizer()					,					image_processor=self.get_image_processor()						)
											processor.save_pretrained(self.tmpdirname						)

											a__			 =						self.get_tokenizer(bos_token="""(BOS)"""					,					eos_token="""(EOS)"""						)
											a__			 =						self.get_image_processor(do_normalize=a_					,					padding_value=1.0						)

											a__			 =						GitProcessor.from_pretrained(
											    self.tmpdirname					,					bos_token="""(BOS)"""					,					eos_token="""(EOS)"""					,					do_normalize=a_					,					padding_value=1.0						)

											self.assertEqual(processor.tokenizer.get_vocab()					,					tokenizer_add_kwargs.get_vocab()						)
											self.assertIsInstance(processor.tokenizer					,					a_						)

											self.assertEqual(processor.image_processor.to_json_string()					,					image_processor_add_kwargs.to_json_string()						)
											self.assertIsInstance(processor.image_processor					,					a_						)


							def _a (							self						):
											a__			 =						self.get_image_processor()
											a__			 =						self.get_tokenizer()

											a__			 =						GitProcessor(tokenizer=a_					,					image_processor=a_						)

											a__			 =						self.prepare_image_inputs()

											a__			 =						image_processor(a_					,					return_tensors="""np"""						)
											a__			 =						processor(images=a_					,					return_tensors="""np"""						)

											for key in input_feat_extract.keys():
															self.assertAlmostEqual(input_feat_extract[key].sum()					,					input_processor[key].sum()					,					delta=1E-2						)


							def _a (							self						):
											a__			 =						self.get_image_processor()
											a__			 =						self.get_tokenizer()

											a__			 =						GitProcessor(tokenizer=a_					,					image_processor=a_						)

											a__			 =						"""lower newer"""

											a__			 =						processor(text=a_						)

											a__			 =						tokenizer(a_					,					return_token_type_ids=a_						)

											for key in encoded_tok.keys():
															self.assertListEqual(encoded_tok[key]					,					encoded_processor[key]						)


							def _a (							self						):
											a__			 =						self.get_image_processor()
											a__			 =						self.get_tokenizer()

											a__			 =						GitProcessor(tokenizer=a_					,					image_processor=a_						)

											a__			 =						"""lower newer"""
											a__			 =						self.prepare_image_inputs()

											a__			 =						processor(text=a_					,					images=a_						)

											self.assertListEqual(list(inputs.keys()						)					,					["""input_ids""", """attention_mask""", """pixel_values"""]						)

											# test if it raises when no input is passed
											with pytest.raises(a_						):
															processor()


							def _a (							self						):
											a__			 =						self.get_image_processor()
											a__			 =						self.get_tokenizer()

											a__			 =						GitProcessor(tokenizer=a_					,					image_processor=a_						)

											a__			 =						[[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

											a__			 =						processor.batch_decode(a_						)
											a__			 =						tokenizer.batch_decode(a_						)

											self.assertListEqual(a_					,					a_						)







							def _a (							self						):
											a__			 =						self.get_image_processor()
											a__			 =						self.get_tokenizer()

											a__			 =						GitProcessor(tokenizer=a_					,					image_processor=a_						)

											a__			 =						"""lower newer"""
											a__			 =						self.prepare_image_inputs()

											a__			 =						processor(text=a_					,					images=a_						)

											# For now the processor supports only ['input_ids', 'attention_mask', 'pixel_values']
											self.assertListEqual(list(inputs.keys()						)					,					["""input_ids""", """attention_mask""", """pixel_values"""]						)

