


import json
import os
from typing import TYPE_CHECKING, List, Optional, Tuple

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


if TYPE_CHECKING:
    from transformers.pipelines.conversational import Conversation

lowerCAmelCase__						=			logging.get_logger(__name__)

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

lowerCAmelCase__						=			{
    'vocab_file': {
        'microsoft/deberta-base': 'https://huggingface.co/microsoft/deberta-base/resolve/main/vocab.json',
        'microsoft/deberta-large': 'https://huggingface.co/microsoft/deberta-large/resolve/main/vocab.json',
        'microsoft/deberta-xlarge': 'https://huggingface.co/microsoft/deberta-xlarge/resolve/main/vocab.json',
        'microsoft/deberta-base-mnli': 'https://huggingface.co/microsoft/deberta-base-mnli/resolve/main/vocab.json',
        'microsoft/deberta-large-mnli': 'https://huggingface.co/microsoft/deberta-large-mnli/resolve/main/vocab.json',
        'microsoft/deberta-xlarge-mnli': (
            'https://huggingface.co/microsoft/deberta-xlarge-mnli/resolve/main/vocab.json'
        ),
    },
    'merges_file': {
        'microsoft/deberta-base': 'https://huggingface.co/microsoft/deberta-base/resolve/main/merges.txt',
        'microsoft/deberta-large': 'https://huggingface.co/microsoft/deberta-large/resolve/main/merges.txt',
        'microsoft/deberta-xlarge': 'https://huggingface.co/microsoft/deberta-xlarge/resolve/main/merges.txt',
        'microsoft/deberta-base-mnli': 'https://huggingface.co/microsoft/deberta-base-mnli/resolve/main/merges.txt',
        'microsoft/deberta-large-mnli': 'https://huggingface.co/microsoft/deberta-large-mnli/resolve/main/merges.txt',
        'microsoft/deberta-xlarge-mnli': (
            'https://huggingface.co/microsoft/deberta-xlarge-mnli/resolve/main/merges.txt'
        ),
    },
}

lowerCAmelCase__						=			{
    'microsoft/deberta-base': 5_12,
    'microsoft/deberta-large': 5_12,
    'microsoft/deberta-xlarge': 5_12,
    'microsoft/deberta-base-mnli': 5_12,
    'microsoft/deberta-large-mnli': 5_12,
    'microsoft/deberta-xlarge-mnli': 5_12,
}

lowerCAmelCase__						=			{
    'microsoft/deberta-base': {'do_lower_case': False},
    'microsoft/deberta-large': {'do_lower_case': False},
}
def 							_UpperCAmelCase    ():
       _A				:	List[Any]       =			(
           list(range(ord("!"						)    ,							ord("~"						) + 1						)						) + list(range(ord("¡"						)    ,							ord("¬"						) + 1						)						) + list(range(ord("®"						)    ,							ord("ÿ"						) + 1						)						)
       )
       _A				:	Optional[Any]       =			bs[:]
       _A				:	Tuple       =			0
       for b in range(2**8						):
              if b not in bs:
                     bs.append(UpperCamelCase__						)
                     cs.append(2**8 + n						)
                     n += 1
       _A				:	Tuple       =			[chr(UpperCamelCase__						) for n in cs]
       return dict(zip(UpperCamelCase__    ,							UpperCamelCase__						)						)
def 							_UpperCAmelCase    (UpperCamelCase__					:  Any						):
       _A				:	List[Any]       =			set()
       _A				:	int       =			word[0]
       for char in word[1:]:
              pairs.add((prev_char, char)						)
              _A				:	int       =			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", "token_type_ids"]


     def __init__(						self			,						__lowerCamelCase			,						__lowerCamelCase			,						__lowerCamelCase="replace"			,						__lowerCamelCase="[CLS]"			,						__lowerCamelCase="[SEP]"			,						__lowerCamelCase="[SEP]"			,						__lowerCamelCase="[CLS]"			,						__lowerCamelCase="[UNK]"			,						__lowerCamelCase="[PAD]"			,						__lowerCamelCase="[MASK]"			,						__lowerCamelCase=False			,						__lowerCamelCase=False			,						**__lowerCamelCase			,						)      ->    Union[str, Any]:
            _A				:	str       =			AddedToken(__lowerCamelCase			,						lstrip=__lowerCamelCase			,						rstrip=__lowerCamelCase) if isinstance(__lowerCamelCase			,						__lowerCamelCase) else bos_token
            _A				:	Optional[int]       =			AddedToken(__lowerCamelCase			,						lstrip=__lowerCamelCase			,						rstrip=__lowerCamelCase) if isinstance(__lowerCamelCase			,						__lowerCamelCase) else eos_token
            _A				:	Union[str, Any]       =			AddedToken(__lowerCamelCase			,						lstrip=__lowerCamelCase			,						rstrip=__lowerCamelCase) if isinstance(__lowerCamelCase			,						__lowerCamelCase) else sep_token
            _A				:	str       =			AddedToken(__lowerCamelCase			,						lstrip=__lowerCamelCase			,						rstrip=__lowerCamelCase) if isinstance(__lowerCamelCase			,						__lowerCamelCase) else cls_token
            _A				:	List[Any]       =			AddedToken(__lowerCamelCase			,						lstrip=__lowerCamelCase			,						rstrip=__lowerCamelCase) if isinstance(__lowerCamelCase			,						__lowerCamelCase) else unk_token
            _A				:	Tuple       =			AddedToken(__lowerCamelCase			,						lstrip=__lowerCamelCase			,						rstrip=__lowerCamelCase) if isinstance(__lowerCamelCase			,						__lowerCamelCase) else pad_token

            # Mask token behave like a normal word, i.e. include the space before it
            _A				:	Any       =			AddedToken(__lowerCamelCase			,						lstrip=__lowerCamelCase			,						rstrip=__lowerCamelCase) if isinstance(__lowerCamelCase			,						__lowerCamelCase) else mask_token

            super().__init__(
                errors=__lowerCamelCase			,						bos_token=__lowerCamelCase			,						eos_token=__lowerCamelCase			,						unk_token=__lowerCamelCase			,						sep_token=__lowerCamelCase			,						cls_token=__lowerCamelCase			,						pad_token=__lowerCamelCase			,						mask_token=__lowerCamelCase			,						add_prefix_space=__lowerCamelCase			,						add_bos_token=__lowerCamelCase			,						**__lowerCamelCase			,						)
            _A				:	Union[str, Any]       =			add_bos_token

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

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


     @property
     # Copied from transformers.models.gpt2.tokenization_gpt2.GPT2Tokenizer.vocab_size
     def 			_lowerCamelCase			(						self)      ->    List[Any]:
            return len(self.encoder)


     def 			_lowerCamelCase			(						self)      ->    Union[str, 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				:	List[Any]       =			tuple(__lowerCamelCase)
            _A				:	int       =			get_pairs(__lowerCamelCase)

            if not pairs:
                   return token

            while True:
                   _A				:	Union[str, Any]       =			min(__lowerCamelCase			,						key=lambda __lowerCamelCase: self.bpe_ranks.get(__lowerCamelCase			,						float("inf")))
                   if bigram not in self.bpe_ranks:
                          break
                   _A   ,      _A				:	Union[str, Any]       =			bigram
                   _A				:	Union[str, Any]       =			[]
                   _A				:	Optional[int]       =			0
                   while i < len(__lowerCamelCase):
                          try:
                                 _A				:	Union[str, Any]       =			word.index(__lowerCamelCase			,						__lowerCamelCase)
                          except ValueError:
                                 new_word.extend(word[i:])
                                 break
                          else:
                                 new_word.extend(word[i:j])
                                 _A				:	int       =			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				:	Any       =			tuple(__lowerCamelCase)
                   _A				:	Tuple       =			new_word
                   if len(__lowerCamelCase) == 1:
                          break
                   else:
                          _A				:	Dict       =			get_pairs(__lowerCamelCase)
            _A				:	int       =			" ".join(__lowerCamelCase)
            _A				:	Optional[Any]       =			word
            return word


     def 			_lowerCamelCase			(						self			,						__lowerCamelCase			,						__lowerCamelCase = None)      ->    List[int]:
            if token_ids_a is None:
                   return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
            _A				:	Tuple       =			[self.cls_token_id]
            _A				:	int       =			[self.sep_token_id]
            return cls + token_ids_a + sep + token_ids_a + sep


     def 			_lowerCamelCase			(						self			,						__lowerCamelCase			,						__lowerCamelCase = None			,						__lowerCamelCase = False)      ->    List[int]:
            if already_has_special_tokens:
                   return super().get_special_tokens_mask(
                       token_ids_a=__lowerCamelCase			,						token_ids_a=__lowerCamelCase			,						already_has_special_tokens=__lowerCamelCase)

            if token_ids_a is None:
                   return [1] + ([0] * len(__lowerCamelCase)) + [1]
            return [1] + ([0] * len(__lowerCamelCase)) + [1] + ([0] * len(__lowerCamelCase)) + [1]


     def 			_lowerCamelCase			(						self			,						__lowerCamelCase			,						__lowerCamelCase = None)      ->    List[int]:
            _A				:	List[str]       =			[self.sep_token_id]
            _A				:	Optional[int]       =			[self.cls_token_id]

            if token_ids_a is None:
                   return len(cls + token_ids_a + sep) * [0]
            return len(cls + token_ids_a + sep) * [0] + len(token_ids_a + sep) * [1]


     def 			_lowerCamelCase			(						self			,						__lowerCamelCase)      ->    List[str]:
            _A				:	int       =			[]
            for token in re.findall(self.pat			,						__lowerCamelCase):
                   _A				:	List[Any]       =			"".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)      ->    List[Any]:
            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)      ->    Optional[Any]:
            _A				:	List[str]       =			"".join(__lowerCamelCase)
            _A				:	Optional[Any]       =			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				:	Optional[int]       =			os.path.join(
                __lowerCamelCase			,						(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"])
            _A				:	Union[str, 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				:	Tuple       =			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)      ->    Tuple:
            _A				:	List[str]       =			kwargs.pop("add_prefix_space"			,						self.add_prefix_space)
            if (is_split_into_words or add_prefix_space) and (len(__lowerCamelCase) > 0 and not text[0].isspace()):
                   _A				:	Optional[int]       =			" " + text
            return (text, kwargs)


     def 			_lowerCamelCase			(						self			,						__lowerCamelCase)      ->    List[int]:
            _A				:	str       =			[]
            for is_user, text in conversation.iter_texts():
                   input_ids.extend(self.encode(__lowerCamelCase			,						add_special_tokens=__lowerCamelCase) + [self.eos_token_id])
            if len(__lowerCamelCase) > self.model_max_length:
                   _A				:	Optional[int]       =			input_ids[-self.model_max_length :]
            return input_ids




import unittest

import numpy as np

from transformers import BlipTextConfig
from transformers.testing_utils import require_torch, slow, torch_device
from transformers.utils import is_torch_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor, random_attention_mask


if is_torch_available():
    import torch

    from transformers import BlipTextModel
    from transformers.models.blip.modeling_blip import BLIP_PRETRAINED_MODEL_ARCHIVE_LIST







class 					lowerCAmelCase__	:




     '''simple docstring'''


     def __init__(						self			,						__lowerCamelCase			,						__lowerCamelCase=1_2			,						__lowerCamelCase=7			,						__lowerCamelCase=True			,						__lowerCamelCase=True			,						__lowerCamelCase=True			,						__lowerCamelCase=9_9			,						__lowerCamelCase=3_2			,						__lowerCamelCase=3_2			,						__lowerCamelCase=5			,						__lowerCamelCase=4			,						__lowerCamelCase=3_7			,						__lowerCamelCase=0.1			,						__lowerCamelCase=0.1			,						__lowerCamelCase=5_1_2			,						__lowerCamelCase=0.0_2			,						__lowerCamelCase=0			,						__lowerCamelCase=None			,						)      ->    str:
            _A				:	Optional[int]       =			parent
            _A				:	Optional[int]       =			batch_size
            _A				:	Optional[int]       =			seq_length
            _A				:	Dict       =			is_training
            _A				:	Union[str, Any]       =			use_input_mask
            _A				:	Optional[int]       =			use_labels
            _A				:	Any       =			vocab_size
            _A				:	Dict       =			hidden_size
            _A				:	Optional[Any]       =			projection_dim
            _A				:	Union[str, Any]       =			num_hidden_layers
            _A				:	Tuple       =			num_attention_heads
            _A				:	List[str]       =			intermediate_size
            _A				:	Any       =			dropout
            _A				:	List[str]       =			attention_dropout
            _A				:	str       =			max_position_embeddings
            _A				:	List[Any]       =			initializer_range
            _A				:	int       =			scope
            _A				:	List[str]       =			bos_token_id


     def 			_lowerCamelCase			(						self)      ->    List[str]:
            _A				:	Tuple       =			ids_tensor([self.batch_size, self.seq_length]			,						self.vocab_size)

            _A				:	str       =			None
            if self.use_input_mask:
                   _A				:	Union[str, Any]       =			random_attention_mask([self.batch_size, self.seq_length])

            if input_mask is not None:
                   _A   ,      _A				:	int       =			input_mask.shape
                   _A				:	Optional[Any]       =			np.random.randint(1			,						seq_length - 1			,						size=(batch_size,))
                   for batch_idx, start_index in enumerate(__lowerCamelCase):
                          _A				:	Tuple       =			1
                          _A				:	List[str]       =			0

            _A				:	str       =			self.get_config()

            return config, input_ids, input_mask


     def 			_lowerCamelCase			(						self)      ->    Optional[int]:
            return BlipTextConfig(
                vocab_size=self.vocab_size			,						hidden_size=self.hidden_size			,						projection_dim=self.projection_dim			,						num_hidden_layers=self.num_hidden_layers			,						num_attention_heads=self.num_attention_heads			,						intermediate_size=self.intermediate_size			,						dropout=self.dropout			,						attention_dropout=self.attention_dropout			,						max_position_embeddings=self.max_position_embeddings			,						initializer_range=self.initializer_range			,						bos_token_id=self.bos_token_id			,						)


     def 			_lowerCamelCase			(						self			,						__lowerCamelCase			,						__lowerCamelCase			,						__lowerCamelCase)      ->    List[str]:
            _A				:	int       =			BlipTextModel(config=__lowerCamelCase)
            model.to(__lowerCamelCase)
            model.eval()
            with torch.no_grad():
                   _A				:	int       =			model(__lowerCamelCase			,						attention_mask=__lowerCamelCase)
                   _A				:	List[str]       =			model(__lowerCamelCase)
            self.parent.assertEqual(result.last_hidden_state.shape			,						(self.batch_size, self.seq_length, self.hidden_size))
            self.parent.assertEqual(result.pooler_output.shape			,						(self.batch_size, self.hidden_size))


     def 			_lowerCamelCase			(						self)      ->    Any:
            _A				:	Dict       =			self.prepare_config_and_inputs()
            _A   ,      _A   ,      _A				:	str       =			config_and_inputs
            _A				:	List[str]       =			{"input_ids": input_ids, "attention_mask": input_mask}
            return config, inputs_dict







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




     '''simple docstring'''
     __SCREAMING_SNAKE_CASE        =			(BlipTextModel,) if is_torch_available() else ()
     __SCREAMING_SNAKE_CASE        =			False
     __SCREAMING_SNAKE_CASE        =			False
     __SCREAMING_SNAKE_CASE        =			False


     def 			_lowerCamelCase			(						self)      ->    Dict:
            _A				:	Optional[Any]       =			BlipTextModelTester(self)
            _A				:	List[str]       =			ConfigTester(self			,						config_class=__lowerCamelCase			,						hidden_size=3_7)


     def 			_lowerCamelCase			(						self)      ->    List[Any]:
            self.config_tester.run_common_tests()


     def 			_lowerCamelCase			(						self)      ->    Union[str, Any]:
            _A				:	Optional[Any]       =			self.model_tester.prepare_config_and_inputs()
            self.model_tester.create_and_check_model(*__lowerCamelCase)


     def 			_lowerCamelCase			(						self)      ->    Optional[Any]:
            pass


     def 			_lowerCamelCase			(						self)      ->    Dict:
            pass


     @unittest.skip(reason="Blip does not use inputs_embeds")
     def 			_lowerCamelCase			(						self)      ->    str:
            pass


     @unittest.skip(reason="BlipTextModel has no base class and is not available in MODEL_MAPPING")
     def 			_lowerCamelCase			(						self)      ->    Optional[Any]:
            pass


     @unittest.skip(reason="BlipTextModel has no base class and is not available in MODEL_MAPPING")
     def 			_lowerCamelCase			(						self)      ->    str:
            pass


     @slow
     def 			_lowerCamelCase			(						self)      ->    Optional[Any]:
            for model_name in BLIP_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
                   _A				:	Union[str, Any]       =			BlipTextModel.from_pretrained(__lowerCamelCase)
                   self.assertIsNotNone(__lowerCamelCase)


     def 			_lowerCamelCase			(						self)      ->    Optional[int]:
            super().test_pt_tf_model_equivalence(allow_missing_keys=__lowerCamelCase)
