
'''simple docstring'''







import json
import os
import re
import unicodedata
from typing import Optional, Tuple

from ...tokenization_utils import PreTrainedTokenizer, _is_control, _is_punctuation, _is_whitespace
from ...utils import logging


__SCREAMING_SNAKE_CASE		:  int        =      logging.get_logger(__name__)

__SCREAMING_SNAKE_CASE		:  Tuple        =      {
    'vocab_file': 'vocab.json',
    'merges_file': 'merges.txt',
}

__SCREAMING_SNAKE_CASE		:  Any        =      {
    'vocab_file': {'openai-gpt': 'https://huggingface.co/openai-gpt/resolve/main/vocab.json'},
    'merges_file': {'openai-gpt': 'https://huggingface.co/openai-gpt/resolve/main/merges.txt'},
}

__SCREAMING_SNAKE_CASE		:  Optional[int]        =      {
    'openai-gpt': 512,
}

def 				_snake_case (		lowercase						)       ->       str:
				__a       :     int								=	text.strip()
				if not text:
								return []
				__a       :     Dict								=	text.split()
				return tokens
class  SCREAMING_SNAKE_CASE__       (  __UpperCamelCase   ):
	def __init__(				self  ,    __UpperCamelCase=True  ,    __UpperCamelCase=None  ,    __UpperCamelCase=True  ,    __UpperCamelCase=None  ,    __UpperCamelCase=True  ,    ):



					'''simple docstring'''






					if never_split is None:
									__a       :     List[str]								=	[]
					__a       :     Dict								=	do_lower_case
					__a       :     Any								=	set(__UpperCamelCase			)
					__a       :     Dict								=	tokenize_chinese_chars
					__a       :     Dict								=	strip_accents
					__a       :     str								=	do_split_on_punc
	def __lowerCamelCase    (				self  ,    __UpperCamelCase  ,    __UpperCamelCase=None			):



					'''simple docstring'''






					__a       :     List[str]								=	self.never_split.union(set(__UpperCamelCase			)			) if never_split else self.never_split
					__a       :     int								=	self._clean_text(__UpperCamelCase			)

					# This was added on November 1st, 2018 for the multilingual and Chinese
					# models. This is also applied to the English models now, but it doesn't
					# matter since the English models were not trained on any Chinese data
					# and generally don't have any Chinese data in them (there are Chinese
					# characters in the vocabulary because Wikipedia does have some Chinese
					# words in the English Wikipedia.).
					if self.tokenize_chinese_chars:
									__a       :     Union[str, Any]								=	self._tokenize_chinese_chars(__UpperCamelCase			)
					# prevents treating the same character with different unicode codepoints as different characters
					__a       :     Dict								=	unicodedata.normalize("""NFC"""  ,    __UpperCamelCase			)
					__a       :     List[str]								=	whitespace_tokenize(__UpperCamelCase			)
					__a       :     Tuple								=	[]
					for token in orig_tokens:
									if token not in never_split:
													if self.do_lower_case:
																	__a       :     Tuple								=	token.lower()
																	if self.strip_accents is not False:
																					__a       :     int								=	self._run_strip_accents(__UpperCamelCase			)
													elif self.strip_accents:
																	__a       :     Tuple								=	self._run_strip_accents(__UpperCamelCase			)
									split_tokens.extend(self._run_split_on_punc(__UpperCamelCase  ,    __UpperCamelCase			)			)

					__a       :     List[str]								=	whitespace_tokenize(""" """.join(__UpperCamelCase			)			)
					return output_tokens
	def __lowerCamelCase    (				self  ,    __UpperCamelCase			):



					'''simple docstring'''






					__a       :     Tuple								=	unicodedata.normalize("""NFD"""  ,    __UpperCamelCase			)
					__a       :     Union[str, Any]								=	[]
					for char in text:
									__a       :     List[Any]								=	unicodedata.category(__UpperCamelCase			)
									if cat == "Mn":
													continue
									output.append(__UpperCamelCase			)
					return "".join(__UpperCamelCase			)
	def __lowerCamelCase    (				self  ,    __UpperCamelCase  ,    __UpperCamelCase=None			):



					'''simple docstring'''






					if not self.do_split_on_punc or (never_split is not None and text in never_split):
									return [text]
					__a       :     List[Any]								=	list(__UpperCamelCase			)
					__a       :     str								=	0
					__a       :     Tuple								=	True
					__a       :     str								=	[]
					while i < len(__UpperCamelCase			):
									__a       :     Any								=	chars[i]
									if _is_punctuation(__UpperCamelCase			):
													output.append([char]			)
													__a       :     Optional[int]								=	True
									else:
													if start_new_word:
																	output.append([]			)
													__a       :     Any								=	False
													output[-1].append(__UpperCamelCase			)
									i += 1

					return ["".join(__UpperCamelCase			) for x in output]
	def __lowerCamelCase    (				self  ,    __UpperCamelCase			):



					'''simple docstring'''






					__a       :     List[str]								=	[]
					for char in text:
									__a       :     int								=	ord(__UpperCamelCase			)
									if self._is_chinese_char(__UpperCamelCase			):
													output.append(""" """			)
													output.append(__UpperCamelCase			)
													output.append(""" """			)
									else:
													output.append(__UpperCamelCase			)
					return "".join(__UpperCamelCase			)
	def __lowerCamelCase    (				self  ,    __UpperCamelCase			):



					'''simple docstring'''






					if (
					    (cp >= 0X4E_00 and cp <= 0X9F_FF)
					    or (cp >= 0X34_00 and cp <= 0X4D_BF)  #
					    or (cp >= 0X2_00_00 and cp <= 0X2_A6_DF)  #
					    or (cp >= 0X2_A7_00 and cp <= 0X2_B7_3F)  #
					    or (cp >= 0X2_B7_40 and cp <= 0X2_B8_1F)  #
					    or (cp >= 0X2_B8_20 and cp <= 0X2_CE_AF)  #
					    or (cp >= 0XF9_00 and cp <= 0XFA_FF)
					    or (cp >= 0X2_F8_00 and cp <= 0X2_FA_1F)  #
					):  #
									return True

					return False

	def __lowerCamelCase    (				self  ,    __UpperCamelCase			):



					'''simple docstring'''






					__a       :     int								=	[]
					for char in text:
									__a       :     Tuple								=	ord(__UpperCamelCase			)
									if cp == 0 or cp == 0XFF_FD or _is_control(__UpperCamelCase			):
													continue
									if _is_whitespace(__UpperCamelCase			):
													output.append(""" """			)
									else:
													output.append(__UpperCamelCase			)
					return "".join(__UpperCamelCase			)







def 				_snake_case (		lowercase						)       ->       List[str]:
				__a       :     Dict								=	set()
				__a       :     List[str]								=	word[0]
				for char in word[1:]:
								pairs.add((prev_char, char)						)
								__a       :     Optional[Any]								=	char
				return pairs

def 				_snake_case (		lowercase						)       ->       Optional[Any]:
				__a       :     List[str]								=	text.replace("""—"""			,    """-"""						)
				__a       :     Union[str, Any]								=	text.replace("""–"""			,    """-"""						)
				__a       :     Tuple								=	text.replace("""―"""			,    """-"""						)
				__a       :     Union[str, Any]								=	text.replace("""…"""			,    """..."""						)
				__a       :     Dict								=	text.replace("""´"""			,    """'"""						)
				__a       :     str								=	re.sub(r"""(-+|~+|!+|\"+|;+|\?+|\++|,+|\)+|\(+|\\+|\/+|\*+|\[+|\]+|}+|{+|\|+|_+)"""			,    r""" \1 """			,    lowercase						)
				__a       :     int								=	re.sub(r"""\s*\n\s*"""			,    """ \n """			,    lowercase						)
				__a       :     Union[str, Any]								=	re.sub(r"""[^\S\n]+"""			,    """ """			,    lowercase						)
				return text.strip()
class  SCREAMING_SNAKE_CASE__       (  __UpperCamelCase   ):
	lowercase__	    =							VOCAB_FILES_NAMES
	lowercase__	    =							PRETRAINED_VOCAB_FILES_MAP
	lowercase__	    =							PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
	lowercase__	    =							["input_ids", "attention_mask"]
	def __init__(				self  ,    __UpperCamelCase  ,    __UpperCamelCase  ,    __UpperCamelCase="<unk>"  ,    **__UpperCamelCase			):



					'''simple docstring'''






					super().__init__(unk_token=__UpperCamelCase  ,    **__UpperCamelCase			)

					try:
									import ftfy
									from spacy.lang.en import English

									__a       :     List[str]								=	English()
									__a       :     Optional[int]								=	_nlp.tokenizer
									__a       :     List[str]								=	ftfy.fix_text
					except ImportError:
									logger.warning("""ftfy or spacy is not installed using BERT BasicTokenizer instead of SpaCy & ftfy."""			)
									__a       :     int								=	BasicTokenizer(do_lower_case=__UpperCamelCase			)
									__a       :     Optional[Any]								=	None

					with open(__UpperCamelCase  ,    encoding="""utf-8"""			) as vocab_handle:
									__a       :     List[str]								=	json.load(__UpperCamelCase			)
					__a       :     List[Any]								=	{v: k for k, v in self.encoder.items()}
					with open(__UpperCamelCase  ,    encoding="""utf-8"""			) as merges_handle:
									__a       :     str								=	merges_handle.read().split("""\n"""			)[1:-1]
					__a       :     Tuple								=	[tuple(merge.split()			) for merge in merges]
					__a       :     Union[str, Any]								=	dict(zip(__UpperCamelCase  ,    range(len(__UpperCamelCase			)			)			)			)
					__a       :     Any								=	{}
	@property
	def __lowerCamelCase    (				self			):



					'''simple docstring'''






					return True
	@property
	def __lowerCamelCase    (				self			):



					'''simple docstring'''






					return len(self.encoder			)
	def __lowerCamelCase    (				self			):



					'''simple docstring'''






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



					'''simple docstring'''






					__a       :     List[Any]								=	tuple(token[:-1]			) + (token[-1] + """</w>""",)
					if token in self.cache:
									return self.cache[token]
					__a       :     Tuple								=	get_pairs(__UpperCamelCase			)

					if not pairs:
									return token + "</w>"

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

													if word[i] == first and i < len(__UpperCamelCase			) - 1 and word[i + 1] == second:
																	new_word.append(first + second			)
																	i += 2
													else:
																	new_word.append(word[i]			)
																	i += 1
									__a       :     Dict								=	tuple(__UpperCamelCase			)
									__a       :     str								=	new_word
									if len(__UpperCamelCase			) == 1:
													break
									else:
													__a       :     Optional[int]								=	get_pairs(__UpperCamelCase			)
					__a       :     List[Any]								=	""" """.join(__UpperCamelCase			)
					if word == "\n  </w>":
									__a       :     Dict								=	"""\n</w>"""
					__a       :     int								=	word
					return word
	def __lowerCamelCase    (				self  ,    __UpperCamelCase			):



					'''simple docstring'''






					__a       :     List[str]								=	[]
					if self.fix_text is None:
									# Using BERT's BasicTokenizer
									__a       :     Optional[Any]								=	self.nlp.tokenize(__UpperCamelCase			)
									for token in text:
													split_tokens.extend(list(self.bpe(__UpperCamelCase			).split(""" """			)			)			)
					else:
									# Using SpaCy & ftfy (original tokenization process of OpenAI GPT)
									__a       :     List[Any]								=	self.nlp(text_standardize(self.fix_text(__UpperCamelCase			)			)			)
									for token in text:
													split_tokens.extend(list(self.bpe(token.text.lower()			).split(""" """			)			)			)
					return split_tokens
	def __lowerCamelCase    (				self  ,    __UpperCamelCase			):



					'''simple docstring'''






					return self.encoder.get(__UpperCamelCase  ,    self.encoder.get(self.unk_token			)			)
	def __lowerCamelCase    (				self  ,    __UpperCamelCase			):



					'''simple docstring'''






					return self.decoder.get(__UpperCamelCase  ,    self.unk_token			)
	def __lowerCamelCase    (				self  ,    __UpperCamelCase			):



					'''simple docstring'''






					__a       :     str								=	"""""".join(__UpperCamelCase			).replace("""</w>"""  ,    """ """			).strip()
					return out_string

	def __lowerCamelCase    (				self  ,    __UpperCamelCase  ,    __UpperCamelCase = None			):



					'''simple docstring'''






					if not os.path.isdir(__UpperCamelCase			):
									logger.error(f"""Vocabulary path ({save_directory}) should be a directory"""			)
									return
					__a       :     str								=	os.path.join(
					    __UpperCamelCase  ,    (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]			)
					__a       :     Union[str, Any]								=	os.path.join(
					    __UpperCamelCase  ,    (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""merges_file"""]			)

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

					__a       :     Tuple								=	0
					with open(__UpperCamelCase  ,    """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 __UpperCamelCase			: 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       :     Optional[Any]								=	token_index
													writer.write(""" """.join(__UpperCamelCase			) + """\n"""			)
													index += 1

					return vocab_file, merge_file

'''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 TvltImageProcessor

def 				_snake_case (		lowercase			,    lowercase=1_0			,    lowercase=1_0			,    lowercase=False			,    lowercase=False						)       ->       Any:
				__a       :     Union[str, Any]								=	[]
				for i in range(image_processor_tester.num_frames						):
								video.append(np.random.randint(2_5_5			,    size=(image_processor_tester.num_channels, width, height)			,    dtype=np.uinta						)						)

				if not numpify and not torchify:
								# PIL expects the channel dimension as last dimension
								__a       :     Optional[Any]								=	[Image.fromarray(np.moveaxis(lowercase			,    0			,    -1						)						) for frame in video]

				if torchify:
								__a       :     List[str]								=	[torch.from_numpy(lowercase						) for frame in video]

				return video

def 				_snake_case (		lowercase			,    lowercase=False			,    lowercase=False			,    lowercase=False						)       ->       Optional[Any]:
				assert not (numpify and torchify), "You cannot specify both numpy and PyTorch tensors at the same time"

				__a       :     str								=	[]
				for i in range(image_processor_tester.batch_size						):
								if equal_resolution:
												__a       :     int								=	image_processor_tester.max_resolution
								else:
												__a							,	__a       :     Any								=	np.random.choice(
												    np.arange(image_processor_tester.min_resolution			,    image_processor_tester.max_resolution						)			,    2						)
												__a       :     Any								=	prepare_video(
												    image_processor_tester=lowercase			,    width=lowercase			,    height=lowercase			,    numpify=lowercase			,    torchify=lowercase			,    )
								video_inputs.append(lowercase						)

				return video_inputs
class  SCREAMING_SNAKE_CASE__       (  unittest.TestCase   ):
	def __init__(				self  ,    __UpperCamelCase  ,    __UpperCamelCase=7  ,    __UpperCamelCase=3  ,    __UpperCamelCase=4  ,    __UpperCamelCase=18  ,    __UpperCamelCase=30  ,    __UpperCamelCase=400  ,    __UpperCamelCase=True  ,    __UpperCamelCase=None  ,    __UpperCamelCase=True  ,    __UpperCamelCase=[0.5, 0.5, 0.5]  ,    __UpperCamelCase=[0.5, 0.5, 0.5]  ,    __UpperCamelCase=True  ,    __UpperCamelCase=None  ,    ):



					'''simple docstring'''






					__a       :     int								=	size if size is not None else {"""shortest_edge""": 18}
					__a       :     Optional[int]								=	crop_size if crop_size is not None else {"""height""": 18, """width""": 18}

					__a       :     Optional[Any]								=	parent
					__a       :     Dict								=	batch_size
					__a       :     Optional[Any]								=	num_channels
					__a       :     Dict								=	num_frames
					__a       :     Dict								=	image_size
					__a       :     int								=	min_resolution
					__a       :     Tuple								=	max_resolution
					__a       :     Optional[Any]								=	do_resize
					__a       :     str								=	size
					__a       :     int								=	do_normalize
					__a       :     Any								=	image_mean
					__a       :     Optional[Any]								=	image_std
					__a       :     List[Any]								=	do_center_crop
					__a       :     List[Any]								=	crop_size

	def __lowerCamelCase    (				self			):



					'''simple docstring'''






					return {
					    "image_mean": self.image_mean,
					    "image_std": self.image_std,
					    "do_normalize": self.do_normalize,
					    "do_resize": self.do_resize,
					    "size": self.size,
					    "do_center_crop": self.do_center_crop,
					    "crop_size": self.crop_size,
					}







@require_torch
@require_vision
class  SCREAMING_SNAKE_CASE__       (  __UpperCamelCase			,					unittest.TestCase   ):
	lowercase__	    =							TvltImageProcessor if is_vision_available() else None
	def __lowerCamelCase    (				self			):



					'''simple docstring'''






					__a       :     Optional[int]								=	TvltImageProcessorTester(self			)
	@property
	def __lowerCamelCase    (				self			):



					'''simple docstring'''






					return self.image_processor_tester.prepare_image_processor_dict()
	def __lowerCamelCase    (				self			):



					'''simple docstring'''






					__a       :     Optional[Any]								=	self.image_processing_class(**self.image_processor_dict			)
					self.assertTrue(hasattr(__UpperCamelCase  ,    """image_mean"""			)			)
					self.assertTrue(hasattr(__UpperCamelCase  ,    """image_std"""			)			)
					self.assertTrue(hasattr(__UpperCamelCase  ,    """do_normalize"""			)			)
					self.assertTrue(hasattr(__UpperCamelCase  ,    """do_resize"""			)			)
					self.assertTrue(hasattr(__UpperCamelCase  ,    """do_center_crop"""			)			)
					self.assertTrue(hasattr(__UpperCamelCase  ,    """size"""			)			)
	def __lowerCamelCase    (				self			):



					'''simple docstring'''






					__a       :     Union[str, Any]								=	self.image_processing_class(**self.image_processor_dict			)
					# create random PIL videos
					__a       :     List[str]								=	prepare_video_inputs(self.image_processor_tester  ,    equal_resolution=__UpperCamelCase			)
					for video in video_inputs:
									self.assertIsInstance(__UpperCamelCase  ,    __UpperCamelCase			)
									self.assertIsInstance(video[0]  ,    Image.Image			)

					# Test not batched input
					__a       :     int								=	image_processor(video_inputs[0]  ,    return_tensors="""pt"""			).pixel_values
					self.assertEqual(
					    encoded_videos.shape  ,    (
					        1,
					        self.image_processor_tester.num_frames,
					        self.image_processor_tester.num_channels,
					        self.image_processor_tester.crop_size["""height"""],
					        self.image_processor_tester.crop_size["""width"""],
					    )  ,    )

					# Test batched
					__a       :     int								=	image_processor(__UpperCamelCase  ,    return_tensors="""pt"""			).pixel_values
					self.assertEqual(
					    encoded_videos.shape  ,    (
					        self.image_processor_tester.batch_size,
					        self.image_processor_tester.num_frames,
					        self.image_processor_tester.num_channels,
					        self.image_processor_tester.crop_size["""height"""],
					        self.image_processor_tester.crop_size["""width"""],
					    )  ,    )
	def __lowerCamelCase    (				self			):



					'''simple docstring'''






					__a       :     List[str]								=	self.image_processing_class(**self.image_processor_dict			)
					# create random numpy tensors
					__a       :     Union[str, Any]								=	prepare_video_inputs(self.image_processor_tester  ,    equal_resolution=__UpperCamelCase  ,    numpify=__UpperCamelCase			)
					for video in video_inputs:
									self.assertIsInstance(__UpperCamelCase  ,    __UpperCamelCase			)
									self.assertIsInstance(video[0]  ,    np.ndarray			)

					# Test not batched input
					__a       :     Optional[int]								=	image_processor(video_inputs[0]  ,    return_tensors="""pt"""			).pixel_values
					self.assertEqual(
					    encoded_videos.shape  ,    (
					        1,
					        self.image_processor_tester.num_frames,
					        self.image_processor_tester.num_channels,
					        self.image_processor_tester.crop_size["""height"""],
					        self.image_processor_tester.crop_size["""width"""],
					    )  ,    )

					# Test batched
					__a       :     Dict								=	image_processor(__UpperCamelCase  ,    return_tensors="""pt"""			).pixel_values
					self.assertEqual(
					    encoded_videos.shape  ,    (
					        self.image_processor_tester.batch_size,
					        self.image_processor_tester.num_frames,
					        self.image_processor_tester.num_channels,
					        self.image_processor_tester.crop_size["""height"""],
					        self.image_processor_tester.crop_size["""width"""],
					    )  ,    )

	def __lowerCamelCase    (				self			):



					'''simple docstring'''






					__a       :     Optional[Any]								=	self.image_processing_class(**self.image_processor_dict			)
					# create random PyTorch tensors
					__a       :     Optional[int]								=	prepare_video_inputs(self.image_processor_tester  ,    equal_resolution=__UpperCamelCase  ,    torchify=__UpperCamelCase			)
					for video in video_inputs:
									self.assertIsInstance(__UpperCamelCase  ,    __UpperCamelCase			)
									self.assertIsInstance(video[0]  ,    torch.Tensor			)

					# Test not batched input
					__a       :     Optional[int]								=	image_processor(video_inputs[0]  ,    return_tensors="""pt"""			).pixel_values
					self.assertEqual(
					    encoded_videos.shape  ,    (
					        1,
					        self.image_processor_tester.num_frames,
					        self.image_processor_tester.num_channels,
					        self.image_processor_tester.crop_size["""height"""],
					        self.image_processor_tester.crop_size["""width"""],
					    )  ,    )

					# Test batched
					__a       :     int								=	image_processor(__UpperCamelCase  ,    return_tensors="""pt"""			).pixel_values
					self.assertEqual(
					    encoded_videos.shape  ,    (
					        self.image_processor_tester.batch_size,
					        self.image_processor_tester.num_frames,
					        self.image_processor_tester.num_channels,
					        self.image_processor_tester.crop_size["""height"""],
					        self.image_processor_tester.crop_size["""width"""],
					    )  ,    )