import os
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

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


UpperCAmelCase_								=							'''▁'''

UpperCAmelCase_								=							{'''vocab_file''': '''spiece.model'''}

UpperCAmelCase_								=							{
    '''vocab_file''': {'''google/pegasus-xsum''': '''https://huggingface.co/google/pegasus-xsum/resolve/main/spiece.model'''}
}

UpperCAmelCase_								=							{
    '''google/pegasus-xsum''': 512,
}


UpperCAmelCase_								=							logging.get_logger(__name__)





class      lowerCAmelCase						(					__UpperCAmelCase			):
     _SCREAMING_SNAKE_CASE     :     List[Any]		=VOCAB_FILES_NAMES

     _SCREAMING_SNAKE_CASE     :     List[str]		=VOCAB_FILES_NAMES
     _SCREAMING_SNAKE_CASE     :     int		=PRETRAINED_VOCAB_FILES_MAP
     _SCREAMING_SNAKE_CASE     :     List[Any]		=PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     _SCREAMING_SNAKE_CASE     :     List[Any]		=["input_ids", "attention_mask"]



     def __init__(					self				,      lowerCAmelCase__				,      lowerCAmelCase__="<pad>"				,      lowerCAmelCase__="</s>"				,      lowerCAmelCase__="<unk>"				,      lowerCAmelCase__="<mask_2>"				,      lowerCAmelCase__="<mask_1>"				,      lowerCAmelCase__=None				,      lowerCAmelCase__=103				,      lowerCAmelCase__ = None				,      **lowerCAmelCase__				,      ):
            _A=	offset
            if additional_special_tokens is not None:
                   if not isinstance(_lowerCamelCase				,      _lowerCamelCase       ):
                          raise TypeError(
                              f"additional_special_tokens should be of type {type(_lowerCamelCase       )}, but is"
                              f" {type(_lowerCamelCase       )}"       )

                   _A=	(
                       ([mask_token_sent] + additional_special_tokens)
                       if mask_token_sent not in additional_special_tokens and mask_token_sent is not None
                       else additional_special_tokens
                   )
                   # fill additional tokens with ..., <unk_token_102> in case not all additional tokens are already taken
                   additional_special_tokens_extended += [
                       f"<unk_{i}>" for i in range(len(_lowerCamelCase       )				,      self.offset - 1       )
                   ]

                   if len(set(_lowerCamelCase       )       ) != len(_lowerCamelCase       ):
                          raise ValueError(
                              'Please make sure that the provided additional_special_tokens do not contain an incorrectly'
                              f" shifted list of <unk_x> tokens. Found {additional_special_tokens_extended}."       )
                   _A=	additional_special_tokens_extended
            else:
                   _A=	[mask_token_sent] if mask_token_sent is not None else []
                   additional_special_tokens += [f"<unk_{i}>" for i in range(2				,      self.offset       )]

            _A=	{} if sp_model_kwargs is None else sp_model_kwargs

            super().__init__(
                eos_token=_lowerCamelCase				,      unk_token=_lowerCamelCase				,      mask_token=_lowerCamelCase				,      pad_token=_lowerCamelCase				,      mask_token_sent=_lowerCamelCase				,      offset=_lowerCamelCase				,      additional_special_tokens=_lowerCamelCase				,      sp_model_kwargs=self.sp_model_kwargs				,      **_lowerCamelCase				,      )
            _A=	mask_token_sent
            _A=	vocab_file
            _A=	spm.SentencePieceProcessor(**self.sp_model_kwargs       )
            self.sp_model.Load(_lowerCamelCase       )

            # add special tokens to encoder dict
            _A=	{
                0: self.pad_token,
                1: self.eos_token,
            }

            if self.mask_token_sent is not None:
                   self.encoder.update(
                       {
                           2: self.mask_token_sent,
                           3: self.mask_token,
                       }       )

            if self.offset > 0:
                   # entries 2-104 are only used for pretraining and called <mask_1>, <mask_2>, unk_2, ...unk_102
                   # mask_token_sent is already added to list -> so start at 1
                   self.encoder.update({i + 3: additional_special_tokens[i] for i in range(1				,      self.offset - 1       )}       )

            _A=	{v: k for k, v in self.encoder.items()}



     @property
     def 			a__  (					self       ):
            return len(self.sp_model       ) + self.offset



     def 			a__  (					self       ):
            _A=	{self.convert_ids_to_tokens(_lowerCamelCase       ): i for i in range(self.vocab_size       )}
            vocab.update(self.added_tokens_encoder       )
            return vocab



     def __getstate__(					self       ):
            _A=	self.__dict__.copy()
            _A=	None
            return state



     def __setstate__(					self				,      lowerCAmelCase__       ):
            _A=	d

            # for backward compatibility
            if not hasattr(self				,      'sp_model_kwargs'       ):
                   _A=	{}

            _A=	spm.SentencePieceProcessor(**self.sp_model_kwargs       )
            self.sp_model.Load(self.vocab_file       )



     def 			a__  (					self				,      lowerCAmelCase__       ):
            return self.sp_model.encode(_lowerCamelCase				,      out_type=_lowerCamelCase       )



     def 			a__  (					self				,      lowerCAmelCase__       ):
            if token in self.decoder:
                   return self.decoder[token]
            elif token in self.added_tokens_decoder:
                   return self.added_tokens_decoder[token]
            _A=	self.sp_model.piece_to_id(_lowerCamelCase       )
            return sp_id + self.offset



     def 			a__  (					self				,      lowerCAmelCase__       ):
            if index in self.encoder:
                   return self.encoder[index]
            elif index in self.added_tokens_encoder:
                   return self.added_tokens_encoder[index]
            else:
                   _A=	self.sp_model.IdToPiece(index - self.offset       )
            return token



     def 			a__  (					self				,      lowerCAmelCase__       ):
            _A=	[]
            _A=	''
            for token in tokens:
                   # make sure that special tokens are not decoded using sentencepiece model
                   if token in self.all_special_tokens:
                          out_string += self.sp_model.decode(_lowerCamelCase       ) + token
                          _A=	[]
                   else:
                          current_sub_tokens.append(_lowerCamelCase       )
            out_string += self.sp_model.decode(_lowerCamelCase       )
            return out_string.strip()



     def 			a__  (					self				,      lowerCAmelCase__=False       ):
            return 1



     def 			a__  (					self				,      lowerCAmelCase__       ):
            _A=	set(self.all_special_ids       )  # call it once instead of inside list comp
            all_special_ids.remove(self.unk_token_id       )  # <unk> is only sometimes special

            return [1 if x in all_special_ids else 0 for x in seq]



     def 			a__  (					self				,      lowerCAmelCase__				,      lowerCAmelCase__ = None				,      lowerCAmelCase__ = False       ):
            if already_has_special_tokens:
                   return self._special_token_mask(_lowerCamelCase       )
            elif token_ids_a is None:
                   return self._special_token_mask(_lowerCamelCase       ) + [1]
            else:
                   return self._special_token_mask(token_ids_a + token_ids_a       ) + [1]



     def 			a__  (					self				,      lowerCAmelCase__				,      lowerCAmelCase__=None       ):
            if token_ids_a is None:
                   return token_ids_a + [self.eos_token_id]
            # We don't expect to process pairs, but leave the pair logic for API consistency
            return token_ids_a + token_ids_a + [self.eos_token_id]





     def 			a__  (					self				,      lowerCAmelCase__				,      lowerCAmelCase__ = None       ):
            if not os.path.isdir(_lowerCamelCase       ):
                   logger.error(f"Vocabulary path ({save_directory}) should be a directory"       )
                   return
            _A=	os.path.join(
                _lowerCamelCase				,      (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']       )

            if os.path.abspath(self.vocab_file       ) != os.path.abspath(_lowerCamelCase       ) and os.path.isfile(self.vocab_file       ):
                   copyfile(self.vocab_file				,      _lowerCamelCase       )
            elif not os.path.isfile(self.vocab_file       ):
                   with open(_lowerCamelCase				,      'wb'       ) as fi:
                          _A=	self.sp_model.serialized_model_proto()
                          fi.write(_lowerCamelCase       )

            return (out_vocab_file,)
import copy
import os
from typing import Union

from ...configuration_utils import PretrainedConfig
from ...utils import logging


UpperCAmelCase_								=							logging.get_logger(__name__)

UpperCAmelCase_								=							{
    '''BridgeTower/bridgetower-base''': '''https://huggingface.co/BridgeTower/bridgetower-base/blob/main/config.json''',
    '''BridgeTower/bridgetower-base-itm-mlm''': (
        '''https://huggingface.co/BridgeTower/bridgetower-base-itm-mlm/blob/main/config.json'''
    ),
}





class      lowerCAmelCase						(					_a			):
     _SCREAMING_SNAKE_CASE     :     List[Any]		="""bridgetower_vision_model"""



     def __init__(					self				,      lowerCAmelCase__=768				,      lowerCAmelCase__=12				,      lowerCAmelCase__=3				,      lowerCAmelCase__=16				,      lowerCAmelCase__=288				,      lowerCAmelCase__=1				,      lowerCAmelCase__=1E-05				,      lowerCAmelCase__=False				,      lowerCAmelCase__=True				,      lowerCAmelCase__=False				,      **lowerCAmelCase__				,      ):
            super().__init__(**lowerCAmelCase__       )
            _A=	hidden_size
            _A=	num_hidden_layers
            _A=	num_channels
            _A=	patch_size
            _A=	image_size
            _A=	initializer_factor
            _A=	layer_norm_eps
            _A=	stop_gradient
            _A=	share_layernorm
            _A=	remove_last_layer





     @classmethod
     def 			a__  (					cls				,      lowerCAmelCase__				,      **lowerCAmelCase__       ):
            _A,			_A=	cls.get_config_dict(lowerCAmelCase__				,      **lowerCAmelCase__       )

            if config_dict.get('model_type'       ) == "bridgetower":
                   _A=	config_dict['text_config']

            if "model_type" in config_dict and hasattr(cls				,      'model_type'       ) and config_dict["model_type"] != cls.model_type:
                   logger.warning(
                       f"You are using a model of type {config_dict['model_type']} to instantiate a model of type "
                       f"{cls.model_type}. This is not supported for all configurations of models and can yield errors."       )

            return cls.from_dict(lowerCAmelCase__				,      **lowerCAmelCase__       )





class      lowerCAmelCase						(					_a			):
     _SCREAMING_SNAKE_CASE     :     List[Any]		="""bridgetower_text_model"""



     def __init__(					self				,      lowerCAmelCase__=50265				,      lowerCAmelCase__=768				,      lowerCAmelCase__=12				,      lowerCAmelCase__=12				,      lowerCAmelCase__=1				,      lowerCAmelCase__=3072				,      lowerCAmelCase__="gelu"				,      lowerCAmelCase__=0.1				,      lowerCAmelCase__=0.1				,      lowerCAmelCase__=514				,      lowerCAmelCase__=1				,      lowerCAmelCase__=1E-05				,      lowerCAmelCase__=1				,      lowerCAmelCase__=0				,      lowerCAmelCase__=2				,      lowerCAmelCase__="absolute"				,      lowerCAmelCase__=True				,      **lowerCAmelCase__				,      ):
            super().__init__(**lowerCAmelCase__       )

            _A=	vocab_size
            _A=	hidden_size
            _A=	num_hidden_layers
            _A=	num_attention_heads
            _A=	hidden_act
            _A=	initializer_factor
            _A=	intermediate_size
            _A=	hidden_dropout_prob
            _A=	attention_probs_dropout_prob
            _A=	max_position_embeddings
            _A=	type_vocab_size
            _A=	layer_norm_eps
            _A=	position_embedding_type
            _A=	use_cache
            _A=	pad_token_id
            _A=	bos_token_id
            _A=	eos_token_id





     @classmethod
     def 			a__  (					cls				,      lowerCAmelCase__				,      **lowerCAmelCase__       ):
            _A,			_A=	cls.get_config_dict(lowerCAmelCase__				,      **lowerCAmelCase__       )

            if config_dict.get('model_type'       ) == "bridgetower":
                   _A=	config_dict['text_config']

            if "model_type" in config_dict and hasattr(cls				,      'model_type'       ) and config_dict["model_type"] != cls.model_type:
                   logger.warning(
                       f"You are using a model of type {config_dict['model_type']} to instantiate a model of type "
                       f"{cls.model_type}. This is not supported for all configurations of models and can yield errors."       )

            return cls.from_dict(lowerCAmelCase__				,      **lowerCAmelCase__       )





class      lowerCAmelCase						(					_a			):
     _SCREAMING_SNAKE_CASE     :     Dict		="""bridgetower"""



     def __init__(					self				,      lowerCAmelCase__=True				,      lowerCAmelCase__="gelu"				,      lowerCAmelCase__=768				,      lowerCAmelCase__=1				,      lowerCAmelCase__=1E-05				,      lowerCAmelCase__=False				,      lowerCAmelCase__="add"				,      lowerCAmelCase__=12				,      lowerCAmelCase__=6				,      lowerCAmelCase__=False				,      lowerCAmelCase__=False				,      lowerCAmelCase__=None				,      lowerCAmelCase__=None				,      **lowerCAmelCase__				,      ):
            # TODO: remove this once the Hub files are updated.
            _A=	kwargs.pop('text_config_dict'				,      lowerCAmelCase__       )
            _A=	kwargs.pop('vision_config_dict'				,      lowerCAmelCase__       )

            super().__init__(**lowerCAmelCase__       )
            _A=	share_cross_modal_transformer_layers
            _A=	hidden_act
            _A=	hidden_size
            _A=	initializer_factor
            _A=	layer_norm_eps
            _A=	share_link_tower_layers
            _A=	link_tower_type
            _A=	num_attention_heads
            _A=	num_hidden_layers
            _A=	tie_word_embeddings
            _A=	init_layernorm_from_vision_encoder

            if text_config is None:
                   _A=	{}
                   logger.info('`text_config` is `None`. Initializing the `BridgeTowerTextConfig` with default values.'       )

            if vision_config is None:
                   _A=	{}
                   logger.info('`vision_config` is `None`. Initializing the `BridgeTowerVisionConfig` with default values.'       )

            _A=	BridgeTowerTextConfig(**lowerCAmelCase__       )
            _A=	BridgeTowerVisionConfig(**lowerCAmelCase__       )



     @classmethod
     def 			a__  (					cls				,      lowerCAmelCase__				,      lowerCAmelCase__				,      **lowerCAmelCase__       ):
            return cls(text_config=text_config.to_dict()				,      vision_config=vision_config.to_dict()				,      **lowerCAmelCase__       )





     def 			a__  (					self       ):
            _A=	copy.deepcopy(self.__dict__       )
            _A=	self.text_config.to_dict()
            _A=	self.vision_config.to_dict()
            _A=	self.__class__.model_type
            return output