





import json
import os
from pathlib import Path
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple, Union

import sentencepiece

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


_lowerCamelCase 					=      logging.get_logger(__name__)

_lowerCamelCase 					=      '▁'

_lowerCamelCase 					=      {
    'vocab_file': 'vocab.json',
    'spm_file': 'sentencepiece.bpe.model',
}

_lowerCamelCase 					=      {
    'vocab_file': {
        'facebook/s2t-small-librispeech-asr': (
            'https://huggingface.co/facebook/s2t-small-librispeech-asr/resolve/main/vocab.json'
        ),
    },
    'spm_file': {
        'facebook/s2t-small-librispeech-asr': (
            'https://huggingface.co/facebook/s2t-small-librispeech-asr/resolve/main/sentencepiece.bpe.model'
        )
    },
}

_lowerCamelCase 					=      {
    'facebook/s2t-small-librispeech-asr': 1024,
}

_lowerCamelCase 					=      ['pt', 'fr', 'ru', 'nl', 'ro', 'it', 'es', 'de']

_lowerCamelCase 					=      {'mustc': MUSTC_LANGS}




class 	UpperCamelCase_							(  UpperCamelCase__     ):
							lowerCamelCase_							     =       VOCAB_FILES_NAMES
							lowerCamelCase_							     =       PRETRAINED_VOCAB_FILES_MAP
							lowerCamelCase_							     =       MAX_MODEL_INPUT_SIZES
							lowerCamelCase_							     =       ["input_ids", "attention_mask"]

							lowerCamelCase_							     =       []


							def __init__(				self     :Union[str, Any]				,					__A     :Dict				,					__A     :Tuple				,					__A     :List[str]="<s>"				,					__A     :str="</s>"				,					__A     :List[str]="<pad>"				,					__A     :Union[str, Any]="<unk>"				,					__A     :List[Any]=False				,					__A     :Tuple=False				,					__A     :Optional[int]=None				,					__A     :Dict=None				,					__A     :Optional[Dict[str, Any]] = None				,					**__A     :int				,					) -> None:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   {} if sp_model_kwargs is None else sp_model_kwargs

									super().__init__(
									    bos_token=__A				,					eos_token=__A				,					unk_token=__A				,					pad_token=__A				,					do_upper_case=__A				,					do_lower_case=__A				,					tgt_lang=__A				,					lang_codes=__A				,					sp_model_kwargs=self.sp_model_kwargs				,					**__A				,					)
									SCREAMING_SNAKE_CASE__								=   do_upper_case
									SCREAMING_SNAKE_CASE__								=   do_lower_case

									SCREAMING_SNAKE_CASE__								=   load_json(__A		)
									SCREAMING_SNAKE_CASE__								=   {v: k for k, v in self.encoder.items()}
									SCREAMING_SNAKE_CASE__								=   spm_file
									SCREAMING_SNAKE_CASE__								=   load_spm(__A				,					self.sp_model_kwargs		)

									if lang_codes is not None:
											SCREAMING_SNAKE_CASE__								=   lang_codes
											SCREAMING_SNAKE_CASE__								=   LANGUAGES[lang_codes]
											SCREAMING_SNAKE_CASE__								=   [f'''<lang:{lang}>''' for lang in self.langs]
											SCREAMING_SNAKE_CASE__								=   {lang: self.sp_model.PieceToId(f'''<lang:{lang}>'''		) for lang in self.langs}

											SCREAMING_SNAKE_CASE__								=   self.lang_tokens
											SCREAMING_SNAKE_CASE__								=   tgt_lang if tgt_lang is not None else self.langs[0]

											self.set_tgt_lang_special_tokens(self._tgt_lang		)
									else:
											SCREAMING_SNAKE_CASE__								=   {}


							@property
							def   _snake_case	(				self     :Dict		) -> int:



									"""simple docstring"""



									return len(self.encoder		)


							@property
							def   _snake_case	(				self     :str		) -> str:



									"""simple docstring"""



									return self._tgt_lang


							@tgt_lang.setter
							def   _snake_case	(				self     :List[Any]				,					__A     :Optional[int]		) -> None:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   new_tgt_lang
									self.set_tgt_lang_special_tokens(__A		)


							def   _snake_case	(				self     :Union[str, Any]				,					__A     :str		) -> None:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   self.lang_code_to_id[tgt_lang]
									SCREAMING_SNAKE_CASE__								=   [lang_code_id]


							def   _snake_case	(				self     :Optional[int]				,					__A     :str		) -> List[str]:



									"""simple docstring"""



									return self.sp_model.encode(__A				,					out_type=__A		)


							def   _snake_case	(				self     :List[Any]				,					__A     :Optional[Any]		) -> Any:



									"""simple docstring"""



									return self.encoder.get(__A				,					self.encoder[self.unk_token]		)


							def   _snake_case	(				self     :List[Any]				,					__A     :int		) -> str:



									"""simple docstring"""



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


							def   _snake_case	(				self     :str				,					__A     :List[str]		) -> str:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   []
									SCREAMING_SNAKE_CASE__								=   """"""
									for token in tokens:
											# make sure that special tokens are not decoded using sentencepiece model
											if token in self.all_special_tokens:
													SCREAMING_SNAKE_CASE__								=   self.sp_model.decode(__A		)
													out_string += (decoded.upper() if self.do_upper_case else decoded) + token + " "
													SCREAMING_SNAKE_CASE__								=   []
											else:
													current_sub_tokens.append(__A		)
									SCREAMING_SNAKE_CASE__								=   self.sp_model.decode(__A		)
									out_string += decoded.upper() if self.do_upper_case else decoded
									return out_string.strip()


							def   _snake_case	(				self     :Any				,					__A     :Optional[int]				,					__A     :Dict=None		) -> List[int]:



									"""simple docstring"""



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


							def   _snake_case	(				self     :int				,					__A     :List[int]				,					__A     :Optional[List[int]] = None				,					__A     :bool = False		) -> List[int]:



									"""simple docstring"""



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

									SCREAMING_SNAKE_CASE__								=   [1] * len(self.prefix_tokens		)
									SCREAMING_SNAKE_CASE__								=   [1]
									if token_ids_a is None:
											return prefix_ones + ([0] * len(__A		)) + suffix_ones
									return prefix_ones + ([0] * len(__A		)) + ([0] * len(__A		)) + suffix_ones


							def   _snake_case	(				self     :str		) -> Dict:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   self.encoder.copy()
									vocab.update(self.added_tokens_encoder		)
									return vocab


							def __getstate__(				self     :Dict		) -> Dict:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   self.__dict__.copy()
									SCREAMING_SNAKE_CASE__								=   None
									return state


							def __setstate__(				self     :Tuple				,					__A     :Dict		) -> None:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   d

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

									SCREAMING_SNAKE_CASE__								=   load_spm(self.spm_file				,					self.sp_model_kwargs		)


							def   _snake_case	(				self     :Dict				,					__A     :str				,					__A     :Optional[str] = None		) -> Tuple[str]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   Path(__A		)
									assert save_dir.is_dir(), f'''{save_directory} should be a directory'''
									SCREAMING_SNAKE_CASE__								=   save_dir / (
									    (filename_prefix + """-""" if filename_prefix else """""") + self.vocab_files_names["""vocab_file"""]
									)
									SCREAMING_SNAKE_CASE__								=   save_dir / (
									    (filename_prefix + """-""" if filename_prefix else """""") + self.vocab_files_names["""spm_file"""]
									)

									save_json(self.encoder				,					__A		)

									if os.path.abspath(self.spm_file		) != os.path.abspath(__A		) and os.path.isfile(self.spm_file		):
											copyfile(self.spm_file				,					__A		)
									elif not os.path.isfile(self.spm_file		):
											with open(__A				,					"""wb"""		) as fi:
													SCREAMING_SNAKE_CASE__								=   self.sp_model.serialized_model_proto()
													fi.write(__A		)

									return (str(__A		), str(__A		))






def 					SCREAMING_SNAKE_CASE__				(							UpperCamelCase__:				str       , UpperCamelCase__:				Dict[str, Any]     ):
		SCREAMING_SNAKE_CASE__								=   sentencepiece.SentencePieceProcessor(**UpperCamelCase__     )
		spm.Load(str(UpperCamelCase__     )     )
		return spm






def 					SCREAMING_SNAKE_CASE__				(							UpperCamelCase__:				str     ):
		with open(UpperCamelCase__       , """r"""     ) as f:
				return json.load(UpperCamelCase__     )






def 					SCREAMING_SNAKE_CASE__				(							UpperCamelCase__:				Union[str, Any]       , UpperCamelCase__:				str     ):
		with open(UpperCamelCase__       , """w"""     ) as f:
				json.dump(UpperCamelCase__       , UpperCamelCase__       , indent=2     )






import warnings
from typing import List

import numpy as np

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding
from ...utils import is_flax_available, is_tf_available, is_torch_available




class 	UpperCamelCase_							(  UpperCamelCase__     ):
							lowerCamelCase_							     =       ["image_processor", "tokenizer"]
							lowerCamelCase_							     =       "OwlViTImageProcessor"
							lowerCamelCase_							     =       ("CLIPTokenizer", "CLIPTokenizerFast")


							def __init__(				self     :Optional[Any]				,					__A     :int=None				,					__A     :Optional[int]=None				,					**__A     :str		) -> Optional[int]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__								=   None
									if "feature_extractor" in kwargs:
											warnings.warn(
											    """The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"""
											    """ instead."""				,					__A				,					)
											SCREAMING_SNAKE_CASE__								=   kwargs.pop("""feature_extractor"""		)

									SCREAMING_SNAKE_CASE__								=   image_processor if image_processor is not None else feature_extractor
									if image_processor is None:
											raise ValueError("""You need to specify an `image_processor`."""		)
									if tokenizer is None:
											raise ValueError("""You need to specify a `tokenizer`."""		)

									super().__init__(__A				,					__A		)


							def __call__(				self     :str				,					__A     :Dict=None				,					__A     :List[str]=None				,					__A     :str=None				,					__A     :Optional[int]="max_length"				,					__A     :Tuple="np"				,					**__A     :int		) -> Tuple:



									"""simple docstring"""



									if text is None and query_images is None and images is None:
											raise ValueError(
											    """You have to specify at least one text or query image or image. All three cannot be none."""		)

									if text is not None:
											if isinstance(__A				,					__A		) or (isinstance(__A				,					__A		) and not isinstance(text[0]				,					__A		)):
													SCREAMING_SNAKE_CASE__								=   [self.tokenizer(__A				,					padding=__A				,					return_tensors=__A				,					**__A		)]

											elif isinstance(__A				,					__A		) and isinstance(text[0]				,					__A		):
													SCREAMING_SNAKE_CASE__								=   []

													# Maximum number of queries across batch
													SCREAMING_SNAKE_CASE__								=   max([len(__A		) for t in text]		)

													# Pad all batch samples to max number of text queries
													for t in text:
															if len(__A		) != max_num_queries:
																	SCREAMING_SNAKE_CASE__								=   t + [""" """] * (max_num_queries - len(__A		))

															SCREAMING_SNAKE_CASE__								=   self.tokenizer(__A				,					padding=__A				,					return_tensors=__A				,					**__A		)
															encodings.append(__A		)
											else:
													raise TypeError("""Input text should be a string, a list of strings or a nested list of strings"""		)

											if return_tensors == "np":
													SCREAMING_SNAKE_CASE__								=   np.concatenate([encoding["""input_ids"""] for encoding in encodings]				,					axis=0		)
													SCREAMING_SNAKE_CASE__								=   np.concatenate([encoding["""attention_mask"""] for encoding in encodings]				,					axis=0		)

											elif return_tensors == "jax" and is_flax_available():
													import jax.numpy as jnp

													SCREAMING_SNAKE_CASE__								=   jnp.concatenate([encoding["""input_ids"""] for encoding in encodings]				,					axis=0		)
													SCREAMING_SNAKE_CASE__								=   jnp.concatenate([encoding["""attention_mask"""] for encoding in encodings]				,					axis=0		)

											elif return_tensors == "pt" and is_torch_available():
													import torch

													SCREAMING_SNAKE_CASE__								=   torch.cat([encoding["""input_ids"""] for encoding in encodings]				,					dim=0		)
													SCREAMING_SNAKE_CASE__								=   torch.cat([encoding["""attention_mask"""] for encoding in encodings]				,					dim=0		)

											elif return_tensors == "tf" and is_tf_available():
													import tensorflow as tf

													SCREAMING_SNAKE_CASE__								=   tf.stack([encoding["""input_ids"""] for encoding in encodings]				,					axis=0		)
													SCREAMING_SNAKE_CASE__								=   tf.stack([encoding["""attention_mask"""] for encoding in encodings]				,					axis=0		)

											else:
													raise ValueError("""Target return tensor type could not be returned"""		)

											SCREAMING_SNAKE_CASE__								=   BatchEncoding()
											SCREAMING_SNAKE_CASE__								=   input_ids
											SCREAMING_SNAKE_CASE__								=   attention_mask

									if query_images is not None:
											SCREAMING_SNAKE_CASE__								=   BatchEncoding()
											SCREAMING_SNAKE_CASE__								=   self.image_processor(
											    __A				,					return_tensors=__A				,					**__A		).pixel_values
											SCREAMING_SNAKE_CASE__								=   query_pixel_values

									if images is not None:
											SCREAMING_SNAKE_CASE__								=   self.image_processor(__A				,					return_tensors=__A				,					**__A		)

									if text is not None and images is not None:
											SCREAMING_SNAKE_CASE__								=   image_features.pixel_values
											return encoding
									elif query_images is not None and images is not None:
											SCREAMING_SNAKE_CASE__								=   image_features.pixel_values
											return encoding
									elif text is not None or query_images is not None:
											return encoding
									else:
											return BatchEncoding(data=dict(**__A		)				,					tensor_type=__A		)


							def   _snake_case	(				self     :List[Any]				,					*__A     :Dict				,					**__A     :Dict		) -> Optional[int]:



									"""simple docstring"""



									return self.image_processor.post_process(*__A				,					**__A		)


							def   _snake_case	(				self     :Optional[int]				,					*__A     :Dict				,					**__A     :List[str]		) -> Optional[Any]:



									"""simple docstring"""



									return self.image_processor.post_process_object_detection(*__A				,					**__A		)


							def   _snake_case	(				self     :str				,					*__A     :List[str]				,					**__A     :Union[str, Any]		) -> Any:



									"""simple docstring"""



									return self.image_processor.post_process_image_guided_detection(*__A				,					**__A		)


							def   _snake_case	(				self     :Dict				,					*__A     :List[str]				,					**__A     :List[str]		) -> int:



									"""simple docstring"""



									return self.tokenizer.batch_decode(*__A				,					**__A		)


							def   _snake_case	(				self     :Dict				,					*__A     :Dict				,					**__A     :List[str]		) -> str:



									"""simple docstring"""



									return self.tokenizer.decode(*__A				,					**__A		)


							@property
							def   _snake_case	(				self     :List[Any]		) -> Optional[int]:



									"""simple docstring"""



									warnings.warn(
									    """`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."""				,					__A				,					)
									return self.image_processor_class


							@property
							def   _snake_case	(				self     :Any		) -> Optional[Any]:



									"""simple docstring"""



									warnings.warn(
									    """`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead."""				,					__A				,					)
									return self.image_processor