



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

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


snake_case    :						Any  =						logging.get_logger(__name__)


snake_case    :						List[str]  =						{
    'vocab_file': 'vocab.json',
    'tokenizer_config_file': 'tokenizer_config.json',
    'merges_file': 'merges.txt',
}

snake_case    :						Dict  =						{
    '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'
        ),
    },
}

snake_case    :						Optional[int]  =						'</w>'
snake_case    :						Any  =						'@@ '
def 	snake_case__       (					__lowercase       )		->		List[Any]:







						"""simple docstring"""

						A__	:							Dict						    = set()
						A__	:							Union[str, Any]						    = word[0]
						for char in word[1:]:
												pairs.add((prev_char, char)       )
												A__	:							List[Any]						    = char
						return pairs


# Speech2Text2 has no max input length
snake_case    :						Optional[Any]  =						{'facebook/s2t-wav2vec2-large-en-de': 1_0_2_4}




class 							lowerCAmelCase__  (	UpperCamelCase						):
					__A     :    List[str]           =						VOCAB_FILES_NAMES
					__A     :    str           =						PRETRAINED_VOCAB_FILES_MAP
					__A     :    List[Any]           =						PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
					__A     :    str           =						['input_ids', 'attention_mask']







					def __init__(       self      :     List[str]			,		_A      :     Union[str, Any]			,		_A      :     List[str]="<s>"			,		_A      :     Union[str, Any]="<pad>"			,		_A      :     List[str]="</s>"			,		_A      :     Optional[Any]="<unk>"			,		_A      :     List[Any]=False			,		_A      :     Dict=None			,		**_A      :     List[Any]			,		):
											super().__init__(
											    unk_token=_A			,		bos_token=_A			,		eos_token=_A			,		pad_token=_A			,		do_lower_case=_A			,		**_A			,		)

											A__	:							int						    = do_lower_case

											with open(_A			,		encoding="utf-8") as vocab_handle:
																	A__	:							List[Any]						    = json.load(_A)
											A__	:							Tuple						    = {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.')

																	A__	:							List[str]						    = None
																	A__	:							List[Any]						    = None
											else:
																	with open(_A			,		encoding="utf-8") as merges_handle:
																							A__	:							Tuple						    = merges_handle.read().split("\n")[:-1]

																	A__	:							Optional[Any]						    = [tuple(merge.split()[:2]) for merge in merges]
																	A__	:							Tuple						    = dict(zip(_A			,		range(len(_A))))
																	A__	:							int						    = {}







					@property
					def  _lowercase			(       self      :     Optional[int]):
											return len(self.decoder)







					def  _lowercase			(       self      :     str):
											return dict(self.encoder			,		**self.added_tokens_encoder)







					def  _lowercase			(       self      :     Union[str, Any]			,		_A      :     int):
											A__	:							Any						    = tuple(token[:-1]) + (token[-1] + BPE_TOKEN_MERGES,)
											if token in self.cache:
																	return self.cache[token]
											A__	:							int						    = get_pairs(_A)

											if not pairs:
																	return token

											while True:
																	A__	:							Union[str, Any]						    = min(_A			,		key=lambda _A: self.bpe_ranks.get(_A			,		float("inf")))
																	if bigram not in self.bpe_ranks:
																							break
																	A__    ,       A__	:							Any						    = bigram
																	A__	:							int						    = []
																	A__	:							List[str]						    = 0
																	while i < len(_A):
																							try:
																													A__	:							List[str]						    = word.index(_A			,		_A)
																							except ValueError:
																													new_word.extend(word[i:])
																													break
																							else:
																													new_word.extend(word[i:j])
																													A__	:							Dict						    = j

																							if word[i] == first and i < len(_A) - 1 and word[i + 1] == second:
																													new_word.append(first + second)
																													i += 2
																							else:
																													new_word.append(word[i])
																													i += 1
																	A__	:							Any						    = tuple(_A)
																	A__	:							Union[str, Any]						    = new_word
																	if len(_A) == 1:
																							break
																	else:
																							A__	:							Dict						    = get_pairs(_A)
											A__	:							Any						    = " ".join(_A)
											if word == "\n  " + BPE_TOKEN_MERGES:
																	A__	:							Tuple						    = "\n" + BPE_TOKEN_MERGES

											if word.endswith(_A):
																	A__	:							List[Any]						    = word.replace(_A			,		"")

											A__	:							Any						    = word.replace(" "			,		_A)
											A__	:							Optional[Any]						    = word
											return word







					def  _lowercase			(       self      :     str			,		_A      :     Tuple):

											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:
																	A__	:							Optional[Any]						    = text.lower()

											A__	:							List[Any]						    = text.split()

											A__	:							int						    = []
											for token in text:
																	if token:
																							split_tokens.extend(list(self.bpe(_A).split(" ")))

											return split_tokens







					def  _lowercase			(       self      :     List[Any]			,		_A      :     str):
											return self.encoder.get(_A			,		self.encoder.get(self.unk_token))







					def  _lowercase			(       self      :     int			,		_A      :     int):
											A__	:							Dict						    = self.decoder.get(_A			,		self.unk_token)
											return result







					def  _lowercase			(       self      :     Optional[Any]			,		_A      :     List[str]):
											A__	:							List[Any]						    = " ".join(_A)

											# make sure @@ tokens are concatenated
											A__	:							str						    = "".join(string.split(_A))

											return string







					def  _lowercase			(       self      :     Any			,		_A      :     str			,		_A      :     Optional[str] = None):
											if not os.path.isdir(_A):
																	logger.error(F'Vocabulary path ({save_directory}) should be a directory')
																	return
											A__	:							int						    = os.path.join(
											    _A			,		(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"])
											A__	:							List[str]						    = os.path.join(
											    _A			,		(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["merges_file"])

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

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

											with open(_A			,		"w"			,		encoding="utf-8") as writer:
																	for bpe_tokens, token_index in sorted(self.bpe_ranks.items()			,		key=lambda _A: 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!")
																													A__	:							List[str]						    = token_index
																							writer.write(" ".join(_A) + "\n")
																							index += 1

											return (vocab_file, merges_file)




import os
import sys
from contextlib import contextmanager


# Windows only
if os.name == "nt":
		import ctypes
		import msvcrt  # noqa




		class 							lowerCAmelCase__  (	ctypes.Structure						):
							# _fields is a specific attr expected by ctypes
							__A     :    Optional[Any]           =						[('size', ctypes.c_int), ('visible', ctypes.c_byte)]
def 	snake_case__       (					)		->		List[Any]:







						"""simple docstring"""

						if os.name == "nt":
												A__	:							Optional[Any]						    = CursorInfo()
												A__	:							Tuple						    = ctypes.windll.kernelaa.GetStdHandle(-1_1       )
												ctypes.windll.kernelaa.GetConsoleCursorInfo(__lowercase							,     ctypes.byref(__lowercase       )       )
												A__	:							Any						    = False
												ctypes.windll.kernelaa.SetConsoleCursorInfo(__lowercase							,     ctypes.byref(__lowercase       )       )
						elif os.name == "posix":
												sys.stdout.write("\033[?25l"       )
												sys.stdout.flush()
def 	snake_case__       (					)		->		Dict:







						"""simple docstring"""

						if os.name == "nt":
												A__	:							List[str]						    = CursorInfo()
												A__	:							Optional[int]						    = ctypes.windll.kernelaa.GetStdHandle(-1_1       )
												ctypes.windll.kernelaa.GetConsoleCursorInfo(__lowercase							,     ctypes.byref(__lowercase       )       )
												A__	:							int						    = True
												ctypes.windll.kernelaa.SetConsoleCursorInfo(__lowercase							,     ctypes.byref(__lowercase       )       )
						elif os.name == "posix":
												sys.stdout.write("\033[?25h"       )
												sys.stdout.flush()




@contextmanager
def 	snake_case__       (					)		->		Optional[int]:







						"""simple docstring"""

						try:
												hide_cursor()
												yield
						finally:
												show_cursor()