


"""simple docstring"""




import json
import os
from functools import lru_cache
from typing import TYPE_CHECKING, List, Optional, Tuple

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


if TYPE_CHECKING:
			from transformers.pipelines.conversational import Conversation

__SCREAMING_SNAKE_CASE					=						logging.get_logger(__name__)

__SCREAMING_SNAKE_CASE					=						{
    """vocab_file""": """vocab.json""",
    """merges_file""": """merges.txt""",
}

__SCREAMING_SNAKE_CASE					=						{
    """vocab_file""": {
        """gpt2""": """https://huggingface.co/gpt2/resolve/main/vocab.json""",
        """gpt2-medium""": """https://huggingface.co/gpt2-medium/resolve/main/vocab.json""",
        """gpt2-large""": """https://huggingface.co/gpt2-large/resolve/main/vocab.json""",
        """gpt2-xl""": """https://huggingface.co/gpt2-xl/resolve/main/vocab.json""",
        """distilgpt2""": """https://huggingface.co/distilgpt2/resolve/main/vocab.json""",
    },
    """merges_file""": {
        """gpt2""": """https://huggingface.co/gpt2/resolve/main/merges.txt""",
        """gpt2-medium""": """https://huggingface.co/gpt2-medium/resolve/main/merges.txt""",
        """gpt2-large""": """https://huggingface.co/gpt2-large/resolve/main/merges.txt""",
        """gpt2-xl""": """https://huggingface.co/gpt2-xl/resolve/main/merges.txt""",
        """distilgpt2""": """https://huggingface.co/distilgpt2/resolve/main/merges.txt""",
    },
}

__SCREAMING_SNAKE_CASE					=						{
    """gpt2""": 10_24,
    """gpt2-medium""": 10_24,
    """gpt2-large""": 10_24,
    """gpt2-xl""": 10_24,
    """distilgpt2""": 10_24,
}



@lru_cache()
def 						__a      (  ):





							"""simple docstring"""







							_a          =				(
							    list(range(ord("!"					),		ord("~"					) + 1					)					) + list(range(ord("¡"					),		ord("¬"					) + 1					)					) + list(range(ord("®"					),		ord("ÿ"					) + 1					)					)
							)
							_a          =				bs[:]
							_a          =				0
							for b in range(2**8					):
														if b not in bs:
																					bs.append(a					)
																					cs.append(2**8 + n					)
																					n += 1
							_a          =				[chr(a					) for n in cs]
							return dict(zip(a,		a					)					)



def 						__a      (  a					):





							"""simple docstring"""







							_a          =				set()
							_a          =				word[0]
							for char in word[1:]:
														pairs.add((prev_char, char)					)
														_a          =				char
							return pairs





class __snake_case		(					lowerCAmelCase__      ):







			"""simple docstring"""




			lowerCAmelCase_				:   Optional[int]     	=			VOCAB_FILES_NAMES
			lowerCAmelCase_				:   int     	=			PRETRAINED_VOCAB_FILES_MAP
			lowerCAmelCase_				:   Optional[int]     	=			PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
			lowerCAmelCase_				:   Any     	=			["""input_ids""", """attention_mask"""]




			def __init__(		self							:Optional[int]				,  UpperCamelCase__							:Union[str, Any]				,  UpperCamelCase__							:Optional[Any]				,  UpperCamelCase__							:Any="replace"				,  UpperCamelCase__							:List[Any]="<|endoftext|>"				,  UpperCamelCase__							:int="<|endoftext|>"				,  UpperCamelCase__							:Optional[int]="<|endoftext|>"				,  UpperCamelCase__							:int=None				,  UpperCamelCase__							:Union[str, Any]=False				,  UpperCamelCase__							:Any=False				,  **UpperCamelCase__							:Union[str, Any]				,  ):
										_a          =				AddedToken(lowerCAmelCase_				,  lstrip=lowerCAmelCase_				,  rstrip=lowerCAmelCase_       ) if isinstance(lowerCAmelCase_				,  lowerCAmelCase_       ) else bos_token
										_a          =				AddedToken(lowerCAmelCase_				,  lstrip=lowerCAmelCase_				,  rstrip=lowerCAmelCase_       ) if isinstance(lowerCAmelCase_				,  lowerCAmelCase_       ) else eos_token
										_a          =				AddedToken(lowerCAmelCase_				,  lstrip=lowerCAmelCase_				,  rstrip=lowerCAmelCase_       ) if isinstance(lowerCAmelCase_				,  lowerCAmelCase_       ) else unk_token
										_a          =				AddedToken(lowerCAmelCase_				,  lstrip=lowerCAmelCase_				,  rstrip=lowerCAmelCase_       ) if isinstance(lowerCAmelCase_				,  lowerCAmelCase_       ) else pad_token
										super().__init__(
										    errors=lowerCAmelCase_				,  unk_token=lowerCAmelCase_				,  bos_token=lowerCAmelCase_				,  eos_token=lowerCAmelCase_				,  pad_token=lowerCAmelCase_				,  add_prefix_space=lowerCAmelCase_				,  add_bos_token=lowerCAmelCase_				,  **lowerCAmelCase_				,  )
										_a          =				add_bos_token

										with open(lowerCAmelCase_				,  encoding="utf-8"       ) as vocab_handle:
																	_a          =				json.load(lowerCAmelCase_       )
										_a          =				{v: k for k, v in self.encoder.items()}
										_a          =				errors  # how to handle errors in decoding
										_a          =				bytes_to_unicode()
										_a          =				{v: k for k, v in self.byte_encoder.items()}
										with open(lowerCAmelCase_				,  encoding="utf-8"       ) as merges_handle:
																	_a          =				merges_handle.read().split("\n"       )[1:-1]
										_a          =				[tuple(merge.split()       ) for merge in bpe_merges]
										_a          =				dict(zip(lowerCAmelCase_				,  range(len(lowerCAmelCase_       )       )       )       )
										_a          =				{}
										_a          =				add_prefix_space

										# Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
										_a          =				re.compile(R"\'s|\'t|\'re|\'ve|\'m|\'ll|\'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+"       )




			@property
			def SCREAMING_SNAKE_CASE_      (		self							:Tuple       ):
										return len(self.encoder       )




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




			def SCREAMING_SNAKE_CASE_      (		self							:Tuple				,  UpperCamelCase__							:Union[str, Any]       ):






										if token in self.cache:
																	return self.cache[token]
										_a          =				tuple(lowerCAmelCase_       )
										_a          =				get_pairs(lowerCAmelCase_       )

										if not pairs:
																	return token

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

																								if word[i] == first and i < len(lowerCAmelCase_       ) - 1 and word[i + 1] == second:
																															new_word.append(first + second       )
																															i += 2
																								else:
																															new_word.append(word[i]       )
																															i += 1
																	_a          =				tuple(lowerCAmelCase_       )
																	_a          =				new_word
																	if len(lowerCAmelCase_       ) == 1:
																								break
																	else:
																								_a          =				get_pairs(lowerCAmelCase_       )
										_a          =				" ".join(lowerCAmelCase_       )
										_a          =				word
										return word




			def SCREAMING_SNAKE_CASE_      (		self							:Union[str, Any]				,  UpperCamelCase__							:Any				,  UpperCamelCase__							:Dict=None       ):






										if self.add_bos_token:
																	_a          =				[self.bos_token_id]
										else:
																	_a          =				[]

										_a          =				bos_token_ids + token_ids_a

										if token_ids_a is None:
																	return output

										return output + bos_token_ids + token_ids_a




			def SCREAMING_SNAKE_CASE_      (		self							:Optional[Any]				,  UpperCamelCase__							:List[int]				,  UpperCamelCase__							:Optional[List[int]] = None				,  UpperCamelCase__							:bool = False       ):






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

										if not self.add_bos_token:
																	return super().get_special_tokens_mask(
																	    token_ids_a=lowerCAmelCase_				,  token_ids_a=lowerCAmelCase_				,  already_has_special_tokens=lowerCAmelCase_       )

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




			def SCREAMING_SNAKE_CASE_      (		self							:Tuple				,  UpperCamelCase__							:Optional[int]       ):
										_a          =				[]
										for token in re.findall(self.pat				,  lowerCAmelCase_       ):
																	_a          =				"".join(
																	    self.byte_encoder[b] for b in token.encode("utf-8"       )       )  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
																	bpe_tokens.extend(bpe_token for bpe_token in self.bpe(lowerCAmelCase_       ).split(" "       )       )
										return bpe_tokens




			def SCREAMING_SNAKE_CASE_      (		self							:Union[str, Any]				,  UpperCamelCase__							:Tuple       ):
										return self.encoder.get(lowerCAmelCase_				,  self.encoder.get(self.unk_token       )       )




			def SCREAMING_SNAKE_CASE_      (		self							:str				,  UpperCamelCase__							:Optional[Any]       ):
										return self.decoder.get(lowerCAmelCase_       )




			def SCREAMING_SNAKE_CASE_      (		self							:Optional[Any]				,  UpperCamelCase__							:Dict       ):
										_a          =				"".join(lowerCAmelCase_       )
										_a          =				bytearray([self.byte_decoder[c] for c in text]       ).decode("utf-8"				,  errors=self.errors       )
										return text




			def SCREAMING_SNAKE_CASE_      (		self							:Dict				,  UpperCamelCase__							:str				,  UpperCamelCase__							:Optional[str] = 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"]       )
										_a          =				os.path.join(
										    lowerCAmelCase_				,  (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["merges_file"]       )

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

										_a          =				0
										with open(lowerCAmelCase_				,  "w"				,  encoding="utf-8"       ) as writer:
																	writer.write("#version: 0.2\n"       )
																	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 {merge_file}: BPE merge indices are not consecutive.'
																															    " Please check that the tokenizer is not corrupted!"       )
																															_a          =				token_index
																								writer.write(" ".join(lowerCAmelCase_       ) + "\n"       )
																								index += 1

										return vocab_file, merge_file




			def SCREAMING_SNAKE_CASE_      (		self							:int				,  UpperCamelCase__							:Tuple				,  UpperCamelCase__							:str=False				,  **UpperCamelCase__							:Optional[int]       ):
										_a          =				kwargs.pop("add_prefix_space"				,  self.add_prefix_space       )
										if is_split_into_words or add_prefix_space:
																	_a          =				" " + text
										return (text, kwargs)




			def SCREAMING_SNAKE_CASE_      (		self							:Optional[Any]				,  UpperCamelCase__							:"Conversation"       ):
										_a          =				[]
										for is_user, text in conversation.iter_texts():
																	input_ids.extend(self.encode(lowerCAmelCase_				,  add_special_tokens=lowerCAmelCase_       ) + [self.eos_token_id]       )
										if len(lowerCAmelCase_       ) > self.model_max_length:
																	_a          =				input_ids[-self.model_max_length :]
										return input_ids




"""simple docstring"""




import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin





class __snake_case		(					_SCREAMING_SNAKE_CASE      ):







   """simple docstring"""




   lowerCAmelCase_				:   str     	=			'AutoFeatureExtractor'
   lowerCAmelCase_				:   List[Any]     	=			'Speech2Text2Tokenizer'




   def __init__(		self							:int				,  UpperCamelCase__							:List[str]				,  UpperCamelCase__							:Union[str, Any]       ):
          super().__init__(UpperCamelCase__				,  UpperCamelCase__       )
          _a          =				self.feature_extractor
          _a          =				False




   def __call__(		self							:List[str]				,  *UpperCamelCase__							:List[str]				,  **UpperCamelCase__							:Dict       ):
          # For backward compatibility
          if self._in_target_context_manager:
                 return self.current_processor(*UpperCamelCase__				,  **UpperCamelCase__       )

          if "raw_speech" in kwargs:
                 warnings.warn("Using `raw_speech` as a keyword argument is deprecated. Use `audio` instead."       )
                 _a          =				kwargs.pop("raw_speech"       )
          else:
                 _a          =				kwargs.pop("audio"				,  UpperCamelCase__       )
          _a          =				kwargs.pop("sampling_rate"				,  UpperCamelCase__       )
          _a          =				kwargs.pop("text"				,  UpperCamelCase__       )
          if len(UpperCamelCase__       ) > 0:
                 _a          =				args[0]
                 _a          =				args[1:]

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

          if audio is not None:
                 _a          =				self.feature_extractor(UpperCamelCase__				,  *UpperCamelCase__				,  sampling_rate=UpperCamelCase__				,  **UpperCamelCase__       )
          if text is not None:
                 _a          =				self.tokenizer(UpperCamelCase__				,  **UpperCamelCase__       )

          if text is None:
                 return inputs
          elif audio is None:
                 return encodings
          else:
                 _a          =				encodings["input_ids"]
                 return inputs




   def SCREAMING_SNAKE_CASE_      (		self							:Union[str, Any]				,  *UpperCamelCase__							:int				,  **UpperCamelCase__							:Optional[Any]       ):
          return self.tokenizer.batch_decode(*UpperCamelCase__				,  **UpperCamelCase__       )




   def SCREAMING_SNAKE_CASE_      (		self							:Optional[Any]				,  *UpperCamelCase__							:List[Any]				,  **UpperCamelCase__							:List[Any]       ):
          return self.tokenizer.decode(*UpperCamelCase__				,  **UpperCamelCase__       )




   @contextmanager
   def SCREAMING_SNAKE_CASE_      (		self							:Dict       ):
          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 audio inputs, or in a separate call."       )
          _a          =				True
          _a          =				self.tokenizer
          yield
          _a          =				self.feature_extractor
          _a          =				False
