


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


lowerCAmelCase__						=			logging.get_logger(__name__)

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

lowerCAmelCase__						=			{
    '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',
    },
}

lowerCAmelCase__						=			{
    'Salesforce/codegen-350M-mono': 20_48,
}
@lru_cache()
def 							_UpperCAmelCase    ():
       _A				:	List[str]       =			(
           list(range(ord("!"						)    ,							ord("~"						) + 1						)						) + list(range(ord("¡"						)    ,							ord("¬"						) + 1						)						) + list(range(ord("®"						)    ,							ord("ÿ"						) + 1						)						)
       )
       _A				:	Dict       =			bs[:]
       _A				:	List[str]       =			0
       for b in range(2**8						):
              if b not in bs:
                     bs.append(UpperCamelCase__						)
                     cs.append(2**8 + n						)
                     n += 1
       _A				:	str       =			[chr(UpperCamelCase__						) for n in cs]
       return dict(zip(UpperCamelCase__    ,							UpperCamelCase__						)						)
def 							_UpperCAmelCase    (UpperCamelCase__					:  str						):
       _A				:	Optional[Any]       =			set()
       _A				:	Any       =			word[0]
       for char in word[1:]:
              pairs.add((prev_char, char)						)
              _A				:	Dict       =			char
       return pairs







class 					lowerCAmelCase__	(    a):




     '''simple docstring'''
     __SCREAMING_SNAKE_CASE        =			VOCAB_FILES_NAMES
     __SCREAMING_SNAKE_CASE        =			PRETRAINED_VOCAB_FILES_MAP
     __SCREAMING_SNAKE_CASE        =			PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     __SCREAMING_SNAKE_CASE        =			["input_ids", "attention_mask"]


     def __init__(						self			,						__lowerCamelCase			,						__lowerCamelCase			,						__lowerCamelCase="replace"			,						__lowerCamelCase="<|endoftext|>"			,						__lowerCamelCase="<|endoftext|>"			,						__lowerCamelCase="<|endoftext|>"			,						__lowerCamelCase=None			,						__lowerCamelCase=False			,						__lowerCamelCase=False			,						**__lowerCamelCase			,						)      ->    Union[str, Any]:
            _A				:	Optional[int]       =			AddedToken(__lowerCamelCase			,						lstrip=__lowerCamelCase			,						rstrip=__lowerCamelCase) if isinstance(__lowerCamelCase			,						__lowerCamelCase) else bos_token
            _A				:	List[Any]       =			AddedToken(__lowerCamelCase			,						lstrip=__lowerCamelCase			,						rstrip=__lowerCamelCase) if isinstance(__lowerCamelCase			,						__lowerCamelCase) else eos_token
            _A				:	List[str]       =			AddedToken(__lowerCamelCase			,						lstrip=__lowerCamelCase			,						rstrip=__lowerCamelCase) if isinstance(__lowerCamelCase			,						__lowerCamelCase) else unk_token
            _A				:	Any       =			AddedToken(__lowerCamelCase			,						lstrip=__lowerCamelCase			,						rstrip=__lowerCamelCase) if isinstance(__lowerCamelCase			,						__lowerCamelCase) else pad_token
            super().__init__(
                errors=__lowerCamelCase			,						unk_token=__lowerCamelCase			,						bos_token=__lowerCamelCase			,						eos_token=__lowerCamelCase			,						pad_token=__lowerCamelCase			,						add_prefix_space=__lowerCamelCase			,						add_bos_token=__lowerCamelCase			,						**__lowerCamelCase			,						)
            _A				:	int       =			add_bos_token

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

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


     @property
     def 			_lowerCamelCase			(						self)      ->    Any:
            return len(self.encoder)


     def 			_lowerCamelCase			(						self)      ->    Any:
            return dict(self.encoder			,						**self.added_tokens_encoder)


     def 			_lowerCamelCase			(						self			,						__lowerCamelCase)      ->    Optional[Any]:
            if token in self.cache:
                   return self.cache[token]
            _A				:	str       =			tuple(__lowerCamelCase)
            _A				:	Union[str, Any]       =			get_pairs(__lowerCamelCase)

            if not pairs:
                   return token

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

                          if word[i] == first and i < len(__lowerCamelCase) - 1 and word[i + 1] == second:
                                 new_word.append(first + second)
                                 i += 2
                          else:
                                 new_word.append(word[i])
                                 i += 1
                   _A				:	int       =			tuple(__lowerCamelCase)
                   _A				:	Optional[Any]       =			new_word
                   if len(__lowerCamelCase) == 1:
                          break
                   else:
                          _A				:	List[str]       =			get_pairs(__lowerCamelCase)
            _A				:	Any       =			" ".join(__lowerCamelCase)
            _A				:	Dict       =			word
            return word


     def 			_lowerCamelCase			(						self			,						__lowerCamelCase			,						__lowerCamelCase=None)      ->    Optional[int]:
            if self.add_bos_token:
                   _A				:	Optional[int]       =			[self.bos_token_id]
            else:
                   _A				:	Optional[int]       =			[]

            _A				:	Optional[int]       =			bos_token_ids + token_ids_a

            if token_ids_a is None:
                   return output

            return output + bos_token_ids + token_ids_a


     def 			_lowerCamelCase			(						self			,						__lowerCamelCase)      ->    List[str]:
            _A				:	Any       =			[]
            for token in re.findall(self.pat			,						__lowerCamelCase):
                   _A				:	int       =			"".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(__lowerCamelCase).split(" "))
            return bpe_tokens


     def 			_lowerCamelCase			(						self			,						__lowerCamelCase)      ->    Optional[int]:
            return self.encoder.get(__lowerCamelCase			,						self.encoder.get(self.unk_token))


     def 			_lowerCamelCase			(						self			,						__lowerCamelCase)      ->    Union[str, Any]:
            return self.decoder.get(__lowerCamelCase)


     def 			_lowerCamelCase			(						self			,						__lowerCamelCase)      ->    Union[str, Any]:
            _A				:	Tuple       =			"".join(__lowerCamelCase)
            _A				:	str       =			bytearray([self.byte_decoder[c] for c in text]).decode("utf-8"			,						errors=self.errors)
            return text


     def 			_lowerCamelCase			(						self			,						__lowerCamelCase			,						__lowerCamelCase = None)      ->    Tuple[str]:
            if not os.path.isdir(__lowerCamelCase):
                   logger.error(F"Vocabulary path ({save_directory}) should be a directory")
                   return
            _A				:	List[str]       =			os.path.join(
                __lowerCamelCase			,						(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"])
            _A				:	List[Any]       =			os.path.join(
                __lowerCamelCase			,						(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["merges_file"])

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

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

            return vocab_file, merge_file


     def 			_lowerCamelCase			(						self			,						__lowerCamelCase			,						__lowerCamelCase=False			,						**__lowerCamelCase)      ->    Optional[int]:
            _A				:	int       =			kwargs.pop("add_prefix_space"			,						self.add_prefix_space)
            if is_split_into_words or add_prefix_space:
                   _A				:	int       =			" " + text
            return (text, kwargs)


     def 			_lowerCamelCase			(						self			,						__lowerCamelCase			,						__lowerCamelCase = False			,						__lowerCamelCase = None			,						__lowerCamelCase = None			,						**__lowerCamelCase			,						)      ->    str:
            _A				:	List[Any]       =			super()._decode(
                token_ids=__lowerCamelCase			,						skip_special_tokens=__lowerCamelCase			,						clean_up_tokenization_spaces=__lowerCamelCase			,						**__lowerCamelCase			,						)

            if truncate_before_pattern is not None and len(__lowerCamelCase) > 0:
                   _A				:	Any       =			self.truncate(__lowerCamelCase			,						__lowerCamelCase)

            return decoded_text


     def 			_lowerCamelCase			(						self			,						__lowerCamelCase			,						__lowerCamelCase)      ->    Optional[int]:
            def find_re(__lowerCamelCase			,						__lowerCamelCase			,						__lowerCamelCase):
                   _A				:	str       =			pattern.search(__lowerCamelCase			,						__lowerCamelCase)
                   return m.start() if m else -1

            _A				:	int       =			[re.compile(__lowerCamelCase			,						re.MULTILINE) for pattern in truncate_before_pattern]

            _A				:	Optional[int]       =			list(re.finditer("^print"			,						__lowerCamelCase			,						re.MULTILINE))

            if len(__lowerCamelCase) > 1:
                   _A				:	List[Any]       =			completion[: prints[1].start()]

            _A				:	Dict       =			list(re.finditer("^def"			,						__lowerCamelCase			,						re.MULTILINE))

            if len(__lowerCamelCase) > 1:
                   _A				:	Optional[Any]       =			completion[: defs[1].start()]

            _A				:	Union[str, Any]       =			0

            _A				:	Any       =			[
                pos for pos in [find_re(__lowerCamelCase			,						__lowerCamelCase			,						__lowerCamelCase) for terminal in terminals] if pos != -1
            ]

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




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 SwinaSRImageProcessor
    from transformers.image_transforms import get_image_size







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=1 / 2_5_5			,						__lowerCamelCase=True			,						__lowerCamelCase=8			,						)      ->    Optional[Any]:
            _A				:	List[str]       =			parent
            _A				:	str       =			batch_size
            _A				:	Any       =			num_channels
            _A				:	Any       =			image_size
            _A				:	Optional[int]       =			min_resolution
            _A				:	Union[str, Any]       =			max_resolution
            _A				:	Optional[Any]       =			do_rescale
            _A				:	str       =			rescale_factor
            _A				:	str       =			do_pad
            _A				:	Dict       =			pad_size


     def 			_lowerCamelCase			(						self)      ->    str:
            return {
                "do_rescale": self.do_rescale,
                "rescale_factor": self.rescale_factor,
                "do_pad": self.do_pad,
                "pad_size": self.pad_size,
            }


     def 			_lowerCamelCase			(						self			,						__lowerCamelCase=False			,						__lowerCamelCase=False			,						__lowerCamelCase=False)      ->    str:
            assert not (numpify and torchify), "You cannot specify both numpy and PyTorch tensors at the same time"

            if equal_resolution:
                   _A				:	List[str]       =			[]
                   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:
                   _A				:	List[str]       =			[]
                   for i in range(self.batch_size):
                          _A   ,      _A				:	int       =			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
                   _A				:	Optional[Any]       =			[Image.fromarray(np.moveaxis(__lowerCamelCase			,						0			,						-1)) for x in image_inputs]

            if torchify:
                   _A				:	Union[str, Any]       =			[torch.from_numpy(__lowerCamelCase) for x in image_inputs]

            return image_inputs







@require_torch
@require_vision
class 					lowerCAmelCase__	(    a					,   unittest.TestCase):




     '''simple docstring'''
     __SCREAMING_SNAKE_CASE        =			SwinaSRImageProcessor if is_vision_available() else None


     def 			_lowerCamelCase			(						self)      ->    Any:
            _A				:	int       =			SwinaSRImageProcessingTester(self)


     @property
     def 			_lowerCamelCase			(						self)      ->    Any:
            return self.image_processor_tester.prepare_image_processor_dict()


     def 			_lowerCamelCase			(						self)      ->    Union[str, Any]:
            _A				:	Tuple       =			self.image_processing_class(**self.image_processor_dict)
            self.assertTrue(hasattr(__lowerCamelCase			,						"do_rescale"))
            self.assertTrue(hasattr(__lowerCamelCase			,						"rescale_factor"))
            self.assertTrue(hasattr(__lowerCamelCase			,						"do_pad"))
            self.assertTrue(hasattr(__lowerCamelCase			,						"pad_size"))


     def 			_lowerCamelCase			(						self)      ->    Dict:
            pass


     def 			_lowerCamelCase			(						self			,						__lowerCamelCase)      ->    Tuple:
            _A   ,      _A				:	Dict       =			get_image_size(__lowerCamelCase)
            _A				:	Tuple       =			self.image_processor_tester.pad_size

            _A				:	Dict       =			(old_height // size + 1) * size - old_height
            _A				:	Any       =			(old_width // size + 1) * size - old_width
            return old_height + pad_height, old_width + pad_width


     def 			_lowerCamelCase			(						self)      ->    Dict:
            # Initialize image_processor
            _A				:	Optional[int]       =			self.image_processing_class(**self.image_processor_dict)
            # create random PIL images
            _A				:	Union[str, Any]       =			self.image_processor_tester.prepare_inputs(equal_resolution=__lowerCamelCase)
            for image in image_inputs:
                   self.assertIsInstance(__lowerCamelCase			,						Image.Image)

            # Test not batched input
            _A				:	str       =			image_processor(image_inputs[0]			,						return_tensors="pt").pixel_values
            _A   ,      _A				:	int       =			self.calculate_expected_size(np.array(image_inputs[0]))
            self.assertEqual(
                encoded_images.shape			,						(
                    1,
                    self.image_processor_tester.num_channels,
                    expected_height,
                    expected_width,
                )			,						)


     def 			_lowerCamelCase			(						self)      ->    int:
            # Initialize image_processor
            _A				:	Any       =			self.image_processing_class(**self.image_processor_dict)
            # create random numpy tensors
            _A				:	int       =			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
            _A				:	Union[str, Any]       =			image_processor(image_inputs[0]			,						return_tensors="pt").pixel_values
            _A   ,      _A				:	Union[str, Any]       =			self.calculate_expected_size(image_inputs[0])
            self.assertEqual(
                encoded_images.shape			,						(
                    1,
                    self.image_processor_tester.num_channels,
                    expected_height,
                    expected_width,
                )			,						)


     def 			_lowerCamelCase			(						self)      ->    Optional[Any]:
            # Initialize image_processor
            _A				:	str       =			self.image_processing_class(**self.image_processor_dict)
            # create random PyTorch tensors
            _A				:	Tuple       =			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
            _A				:	List[str]       =			image_processor(image_inputs[0]			,						return_tensors="pt").pixel_values
            _A   ,      _A				:	List[str]       =			self.calculate_expected_size(image_inputs[0])
            self.assertEqual(
                encoded_images.shape			,						(
                    1,
                    self.image_processor_tester.num_channels,
                    expected_height,
                    expected_width,
                )			,						)
