





import collections
import os
from typing import List, Optional, Tuple

from transformers.utils import is_jieba_available, requires_backends


if is_jieba_available():
       import jieba

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


UpperCamelCase__      =logging.get_logger(__name__)

UpperCamelCase__      ={'vocab_file': 'vocab.txt'}

UpperCamelCase__      ={
    'vocab_file': {
        'openbmb/cpm-ant-10b': 'https://huggingface.co/openbmb/cpm-ant-10b/blob/main/vocab.txt',
    },
}

UpperCamelCase__      ={
    'openbmb/cpm-ant-10b': 1024,
}






def     lowerCamelCase__			(__lowerCamelCase	):
     _SCREAMING_SNAKE_CASE       :	int						    =							collections.OrderedDict()
     with open(__lowerCamelCase,       "r",       encoding="utf-8"	) as reader:
          _SCREAMING_SNAKE_CASE       :	str						    =							reader.readlines()
     for index, token in enumerate(__lowerCamelCase	):
          _SCREAMING_SNAKE_CASE       :	str						    =							token.rstrip("\n"	)
          _SCREAMING_SNAKE_CASE       :	Optional[Any]						    =							index
     return vocab





class 			lowerCAmelCase__( __a    ):






       '''simple docstring'''


       def __init__(      self    ,						__lowerCamelCase    ,						__lowerCamelCase="<unk>"    ,						__lowerCamelCase=2_0_0							)			->     Any:
            _SCREAMING_SNAKE_CASE       :	int						    =							vocab
            _SCREAMING_SNAKE_CASE       :	Tuple						    =							unk_token
            _SCREAMING_SNAKE_CASE       :	Any						    =							max_input_chars_per_word







       def     UpperCamelCase_			(      self    ,						__lowerCamelCase							)			->     str:
            _SCREAMING_SNAKE_CASE       :	Any						    =							list(A__							)
            if len(A__							) > self.max_input_chars_per_word:
                 return [self.unk_token]

            _SCREAMING_SNAKE_CASE       :	List[Any]						    =							0
            _SCREAMING_SNAKE_CASE       :	Optional[Any]						    =							[]
            while start < len(A__							):
                 _SCREAMING_SNAKE_CASE       :	Optional[Any]						    =							len(A__							)
                 _SCREAMING_SNAKE_CASE       :	Tuple						    =							None
                 while start < end:
                      _SCREAMING_SNAKE_CASE       :	str						    =							"".join(chars[start:end]							)
                      if substr in self.vocab:
                           _SCREAMING_SNAKE_CASE       :	int						    =							substr
                           break
                      end -= 1
                 if cur_substr is None:
                      sub_tokens.append(self.unk_token							)
                      start += 1
                 else:
                      sub_tokens.append(A__							)
                      _SCREAMING_SNAKE_CASE       :	List[Any]						    =							end

            return sub_tokens





class 			lowerCAmelCase__( __a    ):






       '''simple docstring'''


       __snake_case 			=    VOCAB_FILES_NAMES
       __snake_case 			=    PRETRAINED_VOCAB_FILES_MAP
       __snake_case 			=    PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
       __snake_case 			=    ['input_ids', 'attention_mask']
       __snake_case 			=    False




       def __init__(      self    ,						__lowerCamelCase    ,						__lowerCamelCase="<d>"    ,						__lowerCamelCase="</d>"    ,						__lowerCamelCase="<s>"    ,						__lowerCamelCase="</s>"    ,						__lowerCamelCase="<pad>"    ,						__lowerCamelCase="<unk>"    ,						__lowerCamelCase="</n>"    ,						__lowerCamelCase="</_>"    ,						__lowerCamelCase="left"    ,						**__lowerCamelCase    ,						)			->     Dict:
            requires_backends(self    ,						["jieba"]							)
            super().__init__(
                bod_token=A__    ,						eod_token=A__    ,						bos_token=A__    ,						eos_token=A__    ,						pad_token=A__    ,						unk_token=A__    ,						line_token=A__    ,						space_token=A__    ,						padding_side=A__    ,						**A__    ,						)
            _SCREAMING_SNAKE_CASE       :	Dict						    =							bod_token
            _SCREAMING_SNAKE_CASE       :	str						    =							eod_token
            _SCREAMING_SNAKE_CASE       :	Optional[Any]						    =							load_vocab(A__							)
            _SCREAMING_SNAKE_CASE       :	Tuple						    =							self.encoder[space_token]
            _SCREAMING_SNAKE_CASE       :	Any						    =							self.encoder[line_token]

            del self.encoder[space_token]
            del self.encoder[line_token]

            _SCREAMING_SNAKE_CASE       :	Optional[Any]						    =							collections.OrderedDict(sorted(self.encoder.items()    ,						key=lambda __lowerCamelCase							: x[1]							)							)
            _SCREAMING_SNAKE_CASE       :	Optional[int]						    =							{v: k for k, v in self.encoder.items()}

            _SCREAMING_SNAKE_CASE       :	List[str]						    =							WordpieceTokenizer(vocab=self.encoder    ,						unk_token=self.unk_token							)




       @property
       def     UpperCamelCase_			(      self							)			->     Any:
            return self.encoder[self.bod_token]




       @property
       def     UpperCamelCase_			(      self							)			->     int:
            return self.encoder[self.eod_token]




       @property
       def     UpperCamelCase_			(      self							)			->     List[str]:
            return self.encoder["\n"]




       @property
       def     UpperCamelCase_			(      self							)			->     int:
            return len(self.encoder							)




       def     UpperCamelCase_			(      self							)			->     int:
            return dict(self.encoder    ,						**self.added_tokens_encoder							)




       def     UpperCamelCase_			(      self    ,						__lowerCamelCase							)			->     List[Any]:
            _SCREAMING_SNAKE_CASE       :	str						    =							[]
            for x in jieba.cut(A__    ,						cut_all=A__							):
                 output_tokens.extend(self.wordpiece_tokenizer.tokenize(A__							)							)
            return output_tokens




       def     UpperCamelCase_			(      self    ,						__lowerCamelCase    ,						**__lowerCamelCase							)			->     Optional[Any]:
            _SCREAMING_SNAKE_CASE       :	Optional[int]						    =							[i for i in token_ids if i >= 0]
            _SCREAMING_SNAKE_CASE       :	Tuple						    =							[
                x for x in token_ids if x != self.pad_token_id and x != self.eos_token_id and x != self.bos_token_id
            ]
            return super()._decode(A__    ,						**A__							)




       def     UpperCamelCase_			(      self    ,						__lowerCamelCase							)			->     Dict:
            return token in self.encoder




       def     UpperCamelCase_			(      self    ,						__lowerCamelCase							)			->     str:
            return "".join(A__							)




       def     UpperCamelCase_			(      self    ,						__lowerCamelCase							)			->     int:
            return self.encoder.get(A__    ,						self.encoder.get(self.unk_token							)							)




       def     UpperCamelCase_			(      self    ,						__lowerCamelCase							)			->     int:
            return self.decoder.get(A__    ,						self.unk_token							)




       def     UpperCamelCase_			(      self    ,						__lowerCamelCase    ,						__lowerCamelCase = None							)			->     Tuple[str]:
            if os.path.isdir(A__							):
                 _SCREAMING_SNAKE_CASE       :	Union[str, Any]						    =							os.path.join(
                     A__    ,						(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]							)
            else:
                 _SCREAMING_SNAKE_CASE       :	int						    =							(filename_prefix + "-" if filename_prefix else "") + save_directory
            _SCREAMING_SNAKE_CASE       :	Optional[Any]						    =							0
            if " " in self.encoder:
                 _SCREAMING_SNAKE_CASE       :	int						    =							self.encoder[" "]
                 del self.encoder[" "]
            if "\n" in self.encoder:
                 _SCREAMING_SNAKE_CASE       :	Any						    =							self.encoder["\n"]
                 del self.encoder["\n"]
            _SCREAMING_SNAKE_CASE       :	Optional[Any]						    =							collections.OrderedDict(sorted(self.encoder.items()    ,						key=lambda __lowerCamelCase							: x[1]							)							)
            with open(A__    ,						"w"    ,						encoding="utf-8"							) as writer:
                 for token, token_index in self.encoder.items():
                      if index != token_index:
                           logger.warning(
                               F"""Saving vocabulary to {vocab_file}: vocabulary indices are not consecutive."""
                               " Please check that the vocabulary is not corrupted!"							)
                           _SCREAMING_SNAKE_CASE       :	Tuple						    =							token_index
                      writer.write(token + "\n"							)
                      index += 1
            return (vocab_file,)




       def     UpperCamelCase_			(      self    ,						__lowerCamelCase    ,						__lowerCamelCase = None							)			->     List[int]:
            if token_ids_a is None:
                 return [self.bos_token_id] + token_ids_a
            return [self.bos_token_id] + token_ids_a + [self.bos_token_id] + token_ids_a







       def     UpperCamelCase_			(      self    ,						__lowerCamelCase    ,						__lowerCamelCase = None    ,						__lowerCamelCase = False							)			->     List[int]:

            if already_has_special_tokens:
                 return super().get_special_tokens_mask(
                     token_ids_a=A__    ,						token_ids_a=A__    ,						already_has_special_tokens=A__							)

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







from __future__ import annotations

import unittest

import numpy as np

from transformers import BlipTextConfig
from transformers.testing_utils import require_tf, slow
from transformers.utils import is_tf_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor, random_attention_mask


if is_tf_available():
       import tensorflow as tf

       from transformers import TFBlipTextModel
       from transformers.models.blip.modeling_tf_blip import TF_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=2    ,						__lowerCamelCase=4    ,						__lowerCamelCase=3_7    ,						__lowerCamelCase=0.1    ,						__lowerCamelCase=0.1    ,						__lowerCamelCase=5_1_2    ,						__lowerCamelCase=0.02    ,						__lowerCamelCase=0    ,						__lowerCamelCase=None    ,						)			->     Any:
            _SCREAMING_SNAKE_CASE       :	int						    =							parent
            _SCREAMING_SNAKE_CASE       :	Union[str, Any]						    =							batch_size
            _SCREAMING_SNAKE_CASE       :	str						    =							seq_length
            _SCREAMING_SNAKE_CASE       :	str						    =							is_training
            _SCREAMING_SNAKE_CASE       :	str						    =							use_input_mask
            _SCREAMING_SNAKE_CASE       :	int						    =							use_labels
            _SCREAMING_SNAKE_CASE       :	List[Any]						    =							vocab_size
            _SCREAMING_SNAKE_CASE       :	Dict						    =							hidden_size
            _SCREAMING_SNAKE_CASE       :	int						    =							projection_dim
            _SCREAMING_SNAKE_CASE       :	Optional[Any]						    =							num_hidden_layers
            _SCREAMING_SNAKE_CASE       :	int						    =							num_attention_heads
            _SCREAMING_SNAKE_CASE       :	Optional[int]						    =							intermediate_size
            _SCREAMING_SNAKE_CASE       :	int						    =							dropout
            _SCREAMING_SNAKE_CASE       :	Union[str, Any]						    =							attention_dropout
            _SCREAMING_SNAKE_CASE       :	Optional[int]						    =							max_position_embeddings
            _SCREAMING_SNAKE_CASE       :	Dict						    =							initializer_range
            _SCREAMING_SNAKE_CASE       :	Tuple						    =							scope
            _SCREAMING_SNAKE_CASE       :	Any						    =							bos_token_id




       def     UpperCamelCase_			(      self							)			->     Optional[Any]:
            _SCREAMING_SNAKE_CASE       :	Any						    =							ids_tensor([self.batch_size, self.seq_length]    ,						self.vocab_size							)

            _SCREAMING_SNAKE_CASE       :	Optional[Any]						    =							None
            if self.use_input_mask:
                 _SCREAMING_SNAKE_CASE       :	List[str]						    =							random_attention_mask([self.batch_size, self.seq_length]							)

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

            _SCREAMING_SNAKE_CASE       :	Optional[int]						    =							self.get_config()

            return config, input_ids, tf.convert_to_tensor(__lowerCamelCase							)




       def     UpperCamelCase_			(      self							)			->     List[str]:
            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     UpperCamelCase_			(      self    ,						__lowerCamelCase    ,						__lowerCamelCase    ,						__lowerCamelCase							)			->     List[str]:
            _SCREAMING_SNAKE_CASE       :	Optional[int]						    =							TFBlipTextModel(config=__lowerCamelCase							)
            _SCREAMING_SNAKE_CASE       :	Union[str, Any]						    =							model(__lowerCamelCase    ,						attention_mask=__lowerCamelCase    ,						training=__lowerCamelCase							)
            _SCREAMING_SNAKE_CASE       :	Dict						    =							model(__lowerCamelCase    ,						training=__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     UpperCamelCase_			(      self							)			->     str:
            _SCREAMING_SNAKE_CASE       :	Optional[int]						    =							self.prepare_config_and_inputs()
            _SCREAMING_SNAKE_CASE    ,			_SCREAMING_SNAKE_CASE    ,			_SCREAMING_SNAKE_CASE       :	int						    =							config_and_inputs
            _SCREAMING_SNAKE_CASE       :	List[str]						    =							{"input_ids": input_ids, "attention_mask": input_mask}
            return config, inputs_dict





@require_tf
class 			lowerCAmelCase__( __lowercase  ,     unittest.TestCase    ):






       '''simple docstring'''


       __snake_case 			=    (TFBlipTextModel,) if is_tf_available() else ()
       __snake_case 			=    False
       __snake_case 			=    False
       __snake_case 			=    False




       def     UpperCamelCase_			(      self							)			->     Any:
            _SCREAMING_SNAKE_CASE       :	Optional[Any]						    =							BlipTextModelTester(self							)
            _SCREAMING_SNAKE_CASE       :	Tuple						    =							ConfigTester(self    ,						config_class=__lowerCamelCase    ,						hidden_size=3_7							)




       def     UpperCamelCase_			(      self							)			->     Union[str, Any]:
            self.config_tester.run_common_tests()




       def     UpperCamelCase_			(      self							)			->     int:
            _SCREAMING_SNAKE_CASE       :	Optional[int]						    =							self.model_tester.prepare_config_and_inputs()
            self.model_tester.create_and_check_model(*__lowerCamelCase							)




       def     UpperCamelCase_			(      self							)			->     int:
            pass




       def     UpperCamelCase_			(      self							)			->     Tuple:
            pass




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




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




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




       @slow
       def     UpperCamelCase_			(      self							)			->     int:
            for model_name in TF_BLIP_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
                 _SCREAMING_SNAKE_CASE       :	Union[str, Any]						    =							TFBlipTextModel.from_pretrained(__lowerCamelCase							)
                 self.assertIsNotNone(__lowerCamelCase							)







       def     UpperCamelCase_			(      self    ,						__lowerCamelCase=True							)			->     int:
            super().test_pt_tf_model_equivalence(allow_missing_keys=__lowerCamelCase							)