
"""simple docstring"""



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

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


a					:	Optional[int]							=		logging.get_logger(__name__)


a					:	Dict							=		{
    "vocab_file": "vocab.json",
    "tokenizer_config_file": "tokenizer_config.json",
    "merges_file": "merges.txt",
}

a					:	List[Any]							=		{
    "vocab_file": {
        "facebook/s2t-wav2vec2-large-en-de": (
            "https://huggingface.co/facebook/s2t-wav2vec2-large-en-de/resolve/main/vocab.json"
        ),
    },
    "tokenizer_config_file": {
        "facebook/s2t-wav2vec2-large-en-de": (
            "https://huggingface.co/facebook/s2t-wav2vec2-large-en-de/resolve/main/tokenizer_config.json"
        ),
    },
    "merges_file": {
        "facebook/s2t-wav2vec2-large-en-de": (
            "https://huggingface.co/facebook/s2t-wav2vec2-large-en-de/resolve/main/merges.txt"
        ),
    },
}

a					:	Dict							=		"</w>"
a					:	Optional[int]							=		"@@ "


def  _UpperCamelCase	(						_A      )    ->  int:
    _UpperCAmelCase          =   set()
    _UpperCAmelCase          =   word[0]
    for char in word[1:]:
        pairs.add((prev_char, char)      )
        _UpperCAmelCase          =   char
    return pairs


# Speech2Text2 has no max input length
a					:	Dict							=		{"facebook/s2t-wav2vec2-large-en-de": 1_0_2_4}







class 		a_						( UpperCAmelCase_	):
    a      :				int     =				VOCAB_FILES_NAMES
    a      :				Tuple     =				PRETRAINED_VOCAB_FILES_MAP
    a      :				Any     =				PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
    a      :				Tuple     =				['input_ids', 'attention_mask']







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






        '''simple docstring'''






        super().__init__(
            unk_token=_lowercase     ,					bos_token=_lowercase     ,					eos_token=_lowercase     ,					pad_token=_lowercase     ,					do_lower_case=_lowercase     ,					**_lowercase     ,					)

        _UpperCAmelCase          =   do_lower_case

        with open(_lowercase     ,					encoding="""utf-8"""     ) as vocab_handle:
            _UpperCAmelCase          =   json.load(_lowercase     )
        _UpperCAmelCase          =   {v: k for k, v in self.encoder.items()}

        if merges_file is None:
            logger.info(f"""No merges files provided. {self.__class__.__name__} can only be used for decoding."""     )

            _UpperCAmelCase          =   None
            _UpperCAmelCase          =   None
        else:
            with open(_lowercase     ,					encoding="""utf-8"""     ) as merges_handle:
                _UpperCAmelCase          =   merges_handle.read().split("""\n"""     )[:-1]

            _UpperCAmelCase          =   [tuple(merge.split()[:2]     ) for merge in merges]
            _UpperCAmelCase          =   dict(zip(_lowercase     ,					range(len(_lowercase     )     )     )     )
            _UpperCAmelCase          =   {}







    @property
    def      _snake_case					(			self			:       List[str]     )		->Dict:






        '''simple docstring'''






        return len(self.decoder     )







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






        '''simple docstring'''






        return dict(self.encoder     ,					**self.added_tokens_encoder     )







    def      _snake_case					(			self			:       int     ,					__UpperCamelCase			:       Tuple     )		->Optional[int]:






        '''simple docstring'''






        _UpperCAmelCase          =   tuple(token[:-1]     ) + (token[-1] + BPE_TOKEN_MERGES,)
        if token in self.cache:
            return self.cache[token]
        _UpperCAmelCase          =   get_pairs(_lowercase     )

        if not pairs:
            return token

        while True:
            _UpperCAmelCase          =   min(_lowercase     ,					key=lambda __UpperCamelCase     : self.bpe_ranks.get(_lowercase     ,					float("""inf"""     )     )     )
            if bigram not in self.bpe_ranks:
                break
            _UpperCAmelCase    ,_UpperCAmelCase          =   bigram
            _UpperCAmelCase          =   []
            _UpperCAmelCase          =   0
            while i < len(_lowercase     ):
                try:
                    _UpperCAmelCase          =   word.index(_lowercase     ,					_lowercase     )
                except ValueError:
                    new_word.extend(word[i:]     )
                    break
                else:
                    new_word.extend(word[i:j]     )
                    _UpperCAmelCase          =   j

                if word[i] == first and i < len(_lowercase     ) - 1 and word[i + 1] == second:
                    new_word.append(first + second     )
                    i += 2
                else:
                    new_word.append(word[i]     )
                    i += 1
            _UpperCAmelCase          =   tuple(_lowercase     )
            _UpperCAmelCase          =   new_word
            if len(_lowercase     ) == 1:
                break
            else:
                _UpperCAmelCase          =   get_pairs(_lowercase     )
        _UpperCAmelCase          =   """ """.join(_lowercase     )
        if word == "\n  " + BPE_TOKEN_MERGES:
            _UpperCAmelCase          =   """\n""" + BPE_TOKEN_MERGES

        if word.endswith(_lowercase     ):
            _UpperCAmelCase          =   word.replace(_lowercase     ,					""""""     )

        _UpperCAmelCase          =   word.replace(""" """     ,					_lowercase     )
        _UpperCAmelCase          =   word
        return word







    def      _snake_case					(			self			:       str     ,					__UpperCamelCase			:       int     )		->Optional[Any]:






        '''simple docstring'''






        if self.bpe_ranks is None:
            raise ValueError(
                """This tokenizer was instantiated without a `merges.txt` file, so"""
                """ that it can only be used for decoding, not for encoding."""
                """Make sure to provide `merges.txt` file at instantiation to enable """
                """encoding."""     )

        if self.do_lower_case:
            _UpperCAmelCase          =   text.lower()

        _UpperCAmelCase          =   text.split()

        _UpperCAmelCase          =   []
        for token in text:
            if token:
                split_tokens.extend(list(self.bpe(_lowercase     ).split(""" """     )     )     )

        return split_tokens







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






        '''simple docstring'''






        return self.encoder.get(_lowercase     ,					self.encoder.get(self.unk_token     )     )







    def      _snake_case					(			self			:       Union[str, Any]     ,					__UpperCamelCase			:       int     )		->str:






        '''simple docstring'''






        _UpperCAmelCase          =   self.decoder.get(_lowercase     ,					self.unk_token     )
        return result







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






        '''simple docstring'''






        _UpperCAmelCase          =   """ """.join(_lowercase     )

        # make sure @@ tokens are concatenated
        _UpperCAmelCase          =   """""".join(string.split(_lowercase     )     )

        return string







    def      _snake_case					(			self			:       List[str]     ,					__UpperCamelCase			:       str     ,					__UpperCamelCase			:       Optional[str] = None     )		->str:






        '''simple docstring'''






        if not os.path.isdir(_lowercase     ):
            logger.error(f"""Vocabulary path ({save_directory}) should be a directory"""     )
            return
        _UpperCAmelCase          =   os.path.join(
            _lowercase     ,					(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]     )
        _UpperCAmelCase          =   os.path.join(
            _lowercase     ,					(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""merges_file"""]     )

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

        _UpperCAmelCase          =   0
        if self.bpe_ranks is None:
            return (vocab_file,)

        with open(_lowercase     ,					"""w"""     ,					encoding="""utf-8"""     ) as writer:
            for bpe_tokens, token_index in sorted(self.bpe_ranks.items()     ,					key=lambda __UpperCamelCase     : kv[1]     ):
                if index != token_index:
                    logger.warning(
                        f"""Saving vocabulary to {merges_file}: BPE merge indices are not consecutive."""
                        """ Please check that the tokenizer is not corrupted!"""     )
                    _UpperCAmelCase          =   token_index
                writer.write(""" """.join(_lowercase     ) + """\n"""     )
                index += 1

        return (vocab_file, merges_file)

"""simple docstring"""



import re
import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin




class 		a_						( _UpperCAmelCase	):
    a      :				Any     =				['image_processor', 'tokenizer']
    a      :				Optional[int]     =				'AutoImageProcessor'
    a      :				Any     =				'AutoTokenizer'







    def __init__(			self			:       List[str]     ,					__UpperCamelCase			:       List[Any]=None     ,					__UpperCamelCase			:       List[str]=None     ,					**__UpperCamelCase			:       List[Any]     )		->Dict:






        '''simple docstring'''






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

        _UpperCAmelCase          =   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__(__UpperCamelCase     ,					__UpperCamelCase     )
        _UpperCAmelCase          =   self.image_processor
        _UpperCAmelCase          =   False







    def __call__(			self			:       Union[str, Any]     ,					*__UpperCamelCase			:       str     ,					**__UpperCamelCase			:       Optional[Any]     )		->List[str]:






        '''simple docstring'''






        if self._in_target_context_manager:
            return self.current_processor(*__UpperCamelCase     ,					**__UpperCamelCase     )

        _UpperCAmelCase          =   kwargs.pop("""images"""     ,					__UpperCamelCase     )
        _UpperCAmelCase          =   kwargs.pop("""text"""     ,					__UpperCamelCase     )
        if len(__UpperCamelCase     ) > 0:
            _UpperCAmelCase          =   args[0]
            _UpperCAmelCase          =   args[1:]

        if images is None and text is None:
            raise ValueError("""You need to specify either an `images` or `text` input to process."""     )

        if images is not None:
            _UpperCAmelCase          =   self.image_processor(__UpperCamelCase     ,					*__UpperCamelCase     ,					**__UpperCamelCase     )
        if text is not None:
            _UpperCAmelCase          =   self.tokenizer(__UpperCamelCase     ,					**__UpperCamelCase     )

        if text is None:
            return inputs
        elif images is None:
            return encodings
        else:
            _UpperCAmelCase          =   encodings["""input_ids"""]
            return inputs







    def      _snake_case					(			self			:       Union[str, Any]     ,					*__UpperCamelCase			:       int     ,					**__UpperCamelCase			:       Tuple     )		->Tuple:






        '''simple docstring'''






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







    def      _snake_case					(			self			:       Tuple     ,					*__UpperCamelCase			:       int     ,					**__UpperCamelCase			:       Union[str, Any]     )		->int:






        '''simple docstring'''






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







    @contextmanager
    def      _snake_case					(			self			:       Tuple     )		->Union[str, Any]:






        '''simple docstring'''






        warnings.warn(
            """`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your """
            """labels by using the argument `text` of the regular `__call__` method (either in the same call as """
            """your images inputs, or in a separate call."""     )
        _UpperCAmelCase          =   True
        _UpperCAmelCase          =   self.tokenizer
        yield
        _UpperCAmelCase          =   self.image_processor
        _UpperCAmelCase          =   False







    def      _snake_case					(			self			:       str     ,					__UpperCamelCase			:       Dict     ,					__UpperCamelCase			:       Optional[Any]=False     ,					__UpperCamelCase			:       Union[str, Any]=None     )		->List[str]:






        '''simple docstring'''






        if added_vocab is None:
            _UpperCAmelCase          =   self.tokenizer.get_added_vocab()

        _UpperCAmelCase          =   {}

        while tokens:
            _UpperCAmelCase          =   re.search(r"""<s_(.*?)>"""     ,					__UpperCamelCase     ,					re.IGNORECASE     )
            if start_token is None:
                break
            _UpperCAmelCase          =   start_token.group(1     )
            _UpperCAmelCase          =   re.search(rf"""</s_{key}>"""     ,					__UpperCamelCase     ,					re.IGNORECASE     )
            _UpperCAmelCase          =   start_token.group()
            if end_token is None:
                _UpperCAmelCase          =   tokens.replace(__UpperCamelCase     ,					""""""     )
            else:
                _UpperCAmelCase          =   end_token.group()
                _UpperCAmelCase          =   re.escape(__UpperCamelCase     )
                _UpperCAmelCase          =   re.escape(__UpperCamelCase     )
                _UpperCAmelCase          =   re.search(f"""{start_token_escaped}(.*?){end_token_escaped}"""     ,					__UpperCamelCase     ,					re.IGNORECASE     )
                if content is not None:
                    _UpperCAmelCase          =   content.group(1     ).strip()
                    if r"<s_" in content and r"</s_" in content:  # non-leaf node
                        _UpperCAmelCase          =   self.tokenajson(__UpperCamelCase     ,					is_inner_value=__UpperCamelCase     ,					added_vocab=__UpperCamelCase     )
                        if value:
                            if len(__UpperCamelCase     ) == 1:
                                _UpperCAmelCase          =   value[0]
                            _UpperCAmelCase          =   value
                    else:  # leaf nodes
                        _UpperCAmelCase          =   []
                        for leaf in content.split(r"""<sep/>"""     ):
                            _UpperCAmelCase          =   leaf.strip()
                            if leaf in added_vocab and leaf[0] == "<" and leaf[-2:] == "/>":
                                _UpperCAmelCase          =   leaf[1:-2]  # for categorical special tokens
                            output[key].append(__UpperCamelCase     )
                        if len(output[key]     ) == 1:
                            _UpperCAmelCase          =   output[key][0]

                _UpperCAmelCase          =   tokens[tokens.find(__UpperCamelCase     ) + len(__UpperCamelCase     ) :].strip()
                if tokens[:6] == r"<sep/>":  # non-leaf nodes
                    return [output] + self.tokenajson(tokens[6:]     ,					is_inner_value=__UpperCamelCase     ,					added_vocab=__UpperCamelCase     )

        if len(__UpperCamelCase     ):
            return [output] if is_inner_value else output
        else:
            return [] if is_inner_value else {"text_sequence": tokens}







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






        '''simple docstring'''






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







    @property
    def      _snake_case					(			self			:       List[str]     )		->Dict:






        '''simple docstring'''






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