






"""simple docstring"""






import json
import os
from typing import Optional, Tuple

import regex as re

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


SCREAMING_SNAKE_CASE						:						Optional[Any]					   =       logging.get_logger(__name__)

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

SCREAMING_SNAKE_CASE						:						Tuple					   =       {
    """vocab_file""": {"""ctrl""": """https://raw.githubusercontent.com/salesforce/ctrl/master/ctrl-vocab.json"""},
    """merges_file""": {"""ctrl""": """https://raw.githubusercontent.com/salesforce/ctrl/master/ctrl-merges.txt"""},
}

SCREAMING_SNAKE_CASE						:						Optional[int]					   =       {
    """ctrl""": 256,
}

SCREAMING_SNAKE_CASE						:						Any					   =       {
    """Pregnancy""": 168_629,
    """Christianity""": 7_675,
    """Explain""": 106_423,
    """Fitness""": 63_440,
    """Saving""": 63_163,
    """Ask""": 27_171,
    """Ass""": 95_985,
    """Joke""": 163_509,
    """Questions""": 45_622,
    """Thoughts""": 49_605,
    """Retail""": 52_342,
    """Feminism""": 164_338,
    """Writing""": 11_992,
    """Atheism""": 192_263,
    """Netflix""": 48_616,
    """Computing""": 39_639,
    """Opinion""": 43_213,
    """Alone""": 44_967,
    """Funny""": 58_917,
    """Gaming""": 40_358,
    """Human""": 4_088,
    """India""": 1_331,
    """Joker""": 77_138,
    """Diet""": 36_206,
    """Legal""": 11_859,
    """Norman""": 4_939,
    """Tip""": 72_689,
    """Weight""": 52_343,
    """Movies""": 46_273,
    """Running""": 23_425,
    """Science""": 2_090,
    """Horror""": 37_793,
    """Confession""": 60_572,
    """Finance""": 12_250,
    """Politics""": 16_360,
    """Scary""": 191_985,
    """Support""": 12_654,
    """Technologies""": 32_516,
    """Teenage""": 66_160,
    """Event""": 32_769,
    """Learned""": 67_460,
    """Notion""": 182_770,
    """Wikipedia""": 37_583,
    """Books""": 6_665,
    """Extract""": 76_050,
    """Confessions""": 102_701,
    """Conspiracy""": 75_932,
    """Links""": 63_674,
    """Narcissus""": 150_425,
    """Relationship""": 54_766,
    """Relationships""": 134_796,
    """Reviews""": 41_671,
    """News""": 4_256,
    """Translation""": 26_820,
    """multilingual""": 128_406,
}



def  __lowerCamelCase					(					lowerCAmelCase__				):
				A__							 =       set()
				A__							 =       word[0]
				for char in word[1:]:
								pairs.add((prev_char, char)				)
								A__							 =       char

				A__							 =       set(lowerCAmelCase__				)
				return pairs







class       snake_case_       (    _snake_case       ):



	"""simple docstring"""

	SCREAMING_SNAKE_CASE_:							int   			=					VOCAB_FILES_NAMES
	SCREAMING_SNAKE_CASE_:							str   			=					PRETRAINED_VOCAB_FILES_MAP
	SCREAMING_SNAKE_CASE_:							List[str]   			=					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
	SCREAMING_SNAKE_CASE_:							List[Any]   			=					CONTROL_CODES


	def __init__(      self  ,					__a  ,					__a  ,					__a="<unk>"  ,					**__a						):


					"""simple docstring"""

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

					with open(snake_case_  ,					encoding='utf-8'						) as vocab_handle:
									A__							 =       json.load(snake_case_						)
					A__							 =       {v: k for k, v in self.encoder.items()}
					with open(snake_case_  ,					encoding='utf-8'						) as merges_handle:
									A__							 =       merges_handle.read().split('\n'						)[1:-1]
					A__							 =       [tuple(merge.split()						) for merge in merges]
					A__							 =       dict(zip(snake_case_  ,					range(len(snake_case_						)						)						)						)
					A__							 =       {}


	@property
	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					return len(self.encoder						)


	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					return dict(self.encoder  ,					**self.added_tokens_encoder						)


	def       _UpperCAmelCase						(      self  ,					__a						):


					"""simple docstring"""

					if token in self.cache:
									return self.cache[token]
					A__							 =       tuple(snake_case_						)
					A__							 =       tuple(list(word[:-1]						) + [word[-1] + '</w>']						)
					A__							 =       get_pairs(snake_case_						)

					if not pairs:
									return token

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

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


	def       _UpperCAmelCase						(      self  ,					__a						):


					"""simple docstring"""

					A__							 =       []

					A__							 =       re.findall(r'\S+\n?'  ,					snake_case_						)

					for token in words:
									split_tokens.extend(list(self.bpe(snake_case_						).split(' '						)						)						)
					return split_tokens


	def       _UpperCAmelCase						(      self  ,					__a						):


					"""simple docstring"""

					return self.encoder.get(snake_case_  ,					self.encoder.get(self.unk_token						)						)


	def       _UpperCAmelCase						(      self  ,					__a						):


					"""simple docstring"""

					return self.decoder.get(snake_case_  ,					self.unk_token						)


	def       _UpperCAmelCase						(      self  ,					__a						):


					"""simple docstring"""

					A__							 =       " ".join(snake_case_						).replace('@@ '  ,					''						).strip()
					return out_string






	def       _UpperCAmelCase						(      self  ,					__a  ,					__a = None						):


					"""simple docstring"""

					if not os.path.isdir(snake_case_						):
									logger.error(f'''Vocabulary path ({save_directory}) should be a directory'''						)
									return
					A__							 =       os.path.join(
					    snake_case_  ,					(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']						)
					A__							 =       os.path.join(
					    snake_case_  ,					(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['merges_file']						)

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

					A__							 =       0
					with open(snake_case_  ,					'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(snake_case_						) + '\n'						)
													index += 1

					return vocab_file, merge_file

	# def decode(self, token_ids, skip_special_tokens=False, clean_up_tokenization_spaces=True):
	#     filtered_tokens = ' '.join(self.convert_ids_to_tokens(token_ids, skip_special_tokens=skip_special_tokens))
	#     tokens_generated_so_far = re.sub('(@@ )', '', string=filtered_tokens)
	#     tokens_generated_so_far = re.sub('(@@ ?$)', '', string=tokens_generated_so_far)
	#     return ''.join(tokens_generated_so_far)







"""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, prepare_image_inputs


if is_torch_available():
					import torch

if is_vision_available():
					from PIL import Image

					from transformers import PoolFormerImageProcessor





class       snake_case_       (    unittest.TestCase       ):



	"""simple docstring"""

	def __init__(      self  ,					__a  ,					__a=7  ,					__a=3  ,					__a=30  ,					__a=400  ,					__a=True  ,					__a=None  ,					__a=0.9  ,					__a=None  ,					__a=True  ,					__a=[0.5, 0.5, 0.5]  ,					__a=[0.5, 0.5, 0.5]  ,					):


					"""simple docstring"""

					A__							 =       size if size is not None else {'shortest_edge': 30}
					A__							 =       crop_size if crop_size is not None else {'height': 30, 'width': 30}
					A__							 =       parent
					A__							 =       batch_size
					A__							 =       num_channels
					A__							 =       min_resolution
					A__							 =       max_resolution
					A__							 =       do_resize_and_center_crop
					A__							 =       size
					A__							 =       crop_pct
					A__							 =       crop_size
					A__							 =       do_normalize
					A__							 =       image_mean
					A__							 =       image_std






	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					return {
					    "size": self.size,
					    "do_resize_and_center_crop": self.do_resize_and_center_crop,
					    "crop_pct": self.crop_pct,
					    "crop_size": self.crop_size,
					    "do_normalize": self.do_normalize,
					    "image_mean": self.image_mean,
					    "image_std": self.image_std,
					}





@require_torch
@require_vision
class       snake_case_       (    _lowerCamelCase    ,							unittest.TestCase       ):



	"""simple docstring"""

	SCREAMING_SNAKE_CASE_:							List[Any]   			=					PoolFormerImageProcessor if is_vision_available() else None


	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					A__							 =       PoolFormerImageProcessingTester(self						)


	@property
	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					return self.image_processor_tester.prepare_image_processor_dict()


	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					A__							 =       self.image_processing_class(**self.image_processor_dict						)
					self.assertTrue(hasattr(__a  ,					'do_resize_and_center_crop'						)						)
					self.assertTrue(hasattr(__a  ,					'size'						)						)
					self.assertTrue(hasattr(__a  ,					'crop_pct'						)						)
					self.assertTrue(hasattr(__a  ,					'do_normalize'						)						)
					self.assertTrue(hasattr(__a  ,					'image_mean'						)						)
					self.assertTrue(hasattr(__a  ,					'image_std'						)						)


	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					A__							 =       self.image_processing_class.from_dict(self.image_processor_dict						)
					self.assertEqual(image_processor.size  ,					{'shortest_edge': 30}						)
					self.assertEqual(image_processor.crop_size  ,					{'height': 30, 'width': 30}						)

					A__							 =       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       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					pass


	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					A__							 =       self.image_processing_class(**self.image_processor_dict						)
					# create random PIL images
					A__							 =       prepare_image_inputs(self.image_processor_tester  ,					equal_resolution=__a						)
					for image in image_inputs:
									self.assertIsInstance(__a  ,					Image.Image						)

					# Test not batched input
					A__							 =       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
					A__							 =       image_processing(__a  ,					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       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					A__							 =       self.image_processing_class(**self.image_processor_dict						)
					# create random numpy tensors
					A__							 =       prepare_image_inputs(self.image_processor_tester  ,					equal_resolution=__a  ,					numpify=__a						)
					for image in image_inputs:
									self.assertIsInstance(__a  ,					np.ndarray						)

					# Test not batched input
					A__							 =       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
					A__							 =       image_processing(__a  ,					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       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					A__							 =       self.image_processing_class(**self.image_processor_dict						)
					# create random PyTorch tensors
					A__							 =       prepare_image_inputs(self.image_processor_tester  ,					equal_resolution=__a  ,					torchify=__a						)
					for image in image_inputs:
									self.assertIsInstance(__a  ,					torch.Tensor						)

					# Test not batched input
					A__							 =       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
					A__							 =       image_processing(__a  ,					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'],
					    )  ,					)


