
'''simple docstring'''





from typing import List, Optional, Union

import numpy as np

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


UpperCAmelCase            = logging.get_logger(__name__)

class     a (					__magic_name__     ):
				_snake_case		     =			['''input_values''', '''attention_mask''']




				def __init__(				self						:					List[str],     SCREAMING_SNAKE_CASE_						:					Union[str, Any]=1,     SCREAMING_SNAKE_CASE_						:					Optional[int]=1_60_00,     SCREAMING_SNAKE_CASE_						:					List[Any]=0.0,     SCREAMING_SNAKE_CASE_						:					List[Any]=False,     SCREAMING_SNAKE_CASE_						:					Union[str, Any]=True,     **SCREAMING_SNAKE_CASE_						:					Any,     ):
										super().__init__(feature_size=SCREAMING_SNAKE_CASE_,     sampling_rate=SCREAMING_SNAKE_CASE_,     padding_value=SCREAMING_SNAKE_CASE_,     **SCREAMING_SNAKE_CASE_   )
										snake_case							:  Any			=     return_attention_mask
										snake_case							:  List[str]			=     do_normalize




				@staticmethod
				def 						__snake_case				(				SCREAMING_SNAKE_CASE_						:					List[np.ndarray],     SCREAMING_SNAKE_CASE_						:					List[np.ndarray],     SCREAMING_SNAKE_CASE_						:					float = 0.0   ):
										if attention_mask is not None:
																snake_case							:  str			=     np.array(SCREAMING_SNAKE_CASE_,     np.intaa   )
																snake_case							:  int			=     []

																for vector, length in zip(SCREAMING_SNAKE_CASE_,     attention_mask.sum(-1   )   ):
																						snake_case							:  Dict			=     (vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1e-7   )
																						if length < normed_slice.shape[0]:
																												snake_case							:  Any			=     padding_value

																						normed_input_values.append(SCREAMING_SNAKE_CASE_   )
										else:
																snake_case							:  Any			=     [(x - x.mean()) / np.sqrt(x.var() + 1e-7   ) for x in input_values]

										return normed_input_values




				def __call__(				self						:					List[str],     SCREAMING_SNAKE_CASE_						:					Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]],     SCREAMING_SNAKE_CASE_						:					Union[bool, str, PaddingStrategy] = False,     SCREAMING_SNAKE_CASE_						:					Optional[int] = None,     SCREAMING_SNAKE_CASE_						:					bool = False,     SCREAMING_SNAKE_CASE_						:					Optional[int] = None,     SCREAMING_SNAKE_CASE_						:					Optional[bool] = None,     SCREAMING_SNAKE_CASE_						:					Optional[Union[str, TensorType]] = None,     SCREAMING_SNAKE_CASE_						:					Optional[int] = None,     **SCREAMING_SNAKE_CASE_						:					int,     ):

										if sampling_rate is not None:
																if sampling_rate != self.sampling_rate:
																						raise ValueError(
																						    F"""The model corresponding to this feature extractor: {self} was trained using a sampling rate of"""
																						    F""" {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled with"""
																						    F""" {self.sampling_rate} and not {sampling_rate}."""   )
										else:
																logger.warning(
																    '''It is strongly recommended to pass the ``sampling_rate`` argument to this function. '''
																    '''Failing to do so can result in silent errors that might be hard to debug.'''   )

										snake_case							:  Optional[int]			=     isinstance(SCREAMING_SNAKE_CASE_,     np.ndarray   ) and len(raw_speech.shape   ) > 1
										if is_batched_numpy and len(raw_speech.shape   ) > 2:
																raise ValueError(F"""Only mono-channel audio is supported for input to {self}"""   )
										snake_case							:  Any			=     is_batched_numpy or (
										    isinstance(SCREAMING_SNAKE_CASE_,     (list, tuple)   ) and (isinstance(raw_speech[0],     (np.ndarray, tuple, list)   ))
										)

										# always return batch
										if not is_batched:
																snake_case							:  Optional[int]			=     [raw_speech]

										# convert into correct format for padding
										snake_case							:  Tuple			=     BatchFeature({'''input_values''': raw_speech}   )

										snake_case							:  Dict			=     self.pad(
										    SCREAMING_SNAKE_CASE_,     padding=SCREAMING_SNAKE_CASE_,     max_length=SCREAMING_SNAKE_CASE_,     truncation=SCREAMING_SNAKE_CASE_,     pad_to_multiple_of=SCREAMING_SNAKE_CASE_,     return_attention_mask=SCREAMING_SNAKE_CASE_,     )

										# convert input values to correct format
										snake_case							:  Tuple			=     padded_inputs['''input_values''']
										if not isinstance(input_values[0],     np.ndarray   ):
																snake_case							:  List[Any]			=     [np.asarray(SCREAMING_SNAKE_CASE_,     dtype=np.floataa   ) for array in input_values]
										elif (
										    not isinstance(SCREAMING_SNAKE_CASE_,     np.ndarray   )
										    and isinstance(input_values[0],     np.ndarray   )
										    and input_values[0].dtype is np.dtype(np.floataa   )
										):
																snake_case							:  int			=     [array.astype(np.floataa   ) for array in input_values]
										elif isinstance(SCREAMING_SNAKE_CASE_,     np.ndarray   ) and input_values.dtype is np.dtype(np.floataa   ):
																snake_case							:  Union[str, Any]			=     input_values.astype(np.floataa   )

										# convert attention_mask to correct format
										snake_case							:  List[str]			=     padded_inputs.get('''attention_mask'''   )
										if attention_mask is not None:
																snake_case							:  Tuple			=     [np.asarray(SCREAMING_SNAKE_CASE_,     dtype=np.intaa   ) for array in attention_mask]

										# zero-mean and unit-variance normalization
										if self.do_normalize:
																snake_case							:  int			=     (
																    attention_mask
																    if self._get_padding_strategies(SCREAMING_SNAKE_CASE_,     max_length=SCREAMING_SNAKE_CASE_   ) is not PaddingStrategy.DO_NOT_PAD
																    else None
																)
																snake_case							:  Optional[int]			=     self.zero_mean_unit_var_norm(
																    padded_inputs['''input_values'''],     attention_mask=SCREAMING_SNAKE_CASE_,     padding_value=self.padding_value   )

										if return_tensors is not None:
																snake_case							:  List[str]			=     padded_inputs.convert_to_tensors(SCREAMING_SNAKE_CASE_   )

										return padded_inputs





'''simple docstring'''




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

import regex as re

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


_lowerCAmelCase	:		Tuple		  =						logging.get_logger(__name__)

_lowerCAmelCase	:		Dict		  =						{
    'vocab_file': 'vocab.json',
    'merges_file': 'merges.txt',
}

_lowerCAmelCase	:		Optional[int]		  =						{
    'vocab_file': {
        'roberta-base': 'https://huggingface.co/roberta-base/resolve/main/vocab.json',
        'roberta-large': 'https://huggingface.co/roberta-large/resolve/main/vocab.json',
        'roberta-large-mnli': 'https://huggingface.co/roberta-large-mnli/resolve/main/vocab.json',
        'distilroberta-base': 'https://huggingface.co/distilroberta-base/resolve/main/vocab.json',
        'roberta-base-openai-detector': 'https://huggingface.co/roberta-base-openai-detector/resolve/main/vocab.json',
        'roberta-large-openai-detector': (
            'https://huggingface.co/roberta-large-openai-detector/resolve/main/vocab.json'
        ),
    },
    'merges_file': {
        'roberta-base': 'https://huggingface.co/roberta-base/resolve/main/merges.txt',
        'roberta-large': 'https://huggingface.co/roberta-large/resolve/main/merges.txt',
        'roberta-large-mnli': 'https://huggingface.co/roberta-large-mnli/resolve/main/merges.txt',
        'distilroberta-base': 'https://huggingface.co/distilroberta-base/resolve/main/merges.txt',
        'roberta-base-openai-detector': 'https://huggingface.co/roberta-base-openai-detector/resolve/main/merges.txt',
        'roberta-large-openai-detector': (
            'https://huggingface.co/roberta-large-openai-detector/resolve/main/merges.txt'
        ),
    },
}

_lowerCAmelCase	:		Tuple		  =						{
    'roberta-base': 512,
    'roberta-large': 512,
    'roberta-large-mnli': 512,
    'distilroberta-base': 512,
    'roberta-base-openai-detector': 512,
    'roberta-large-openai-detector': 512,
}

@lru_cache()
def       __UpperCamelCase     (     )     ->   Any:




	"""simple docstring"""





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

def       __UpperCamelCase     (     _A     :					Union[str, Any]					)     ->   Any:




	"""simple docstring"""





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






class 			lowerCAmelCase					(					a				):
	_lowerCamelCase :       int         =			VOCAB_FILES_NAMES
	_lowerCamelCase :       Any         =			PRETRAINED_VOCAB_FILES_MAP
	_lowerCamelCase :       int         =			PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
	_lowerCamelCase :       Optional[Any]         =			["""input_ids""", """attention_mask"""]



	def __init__(       self			,       snake_case__			,       snake_case__			,       snake_case__="replace"			,       snake_case__="<s>"			,       snake_case__="</s>"			,       snake_case__="</s>"			,       snake_case__="<s>"			,       snake_case__="<unk>"			,       snake_case__="<pad>"			,       snake_case__="<mask>"			,       snake_case__=False			,       **snake_case__			,       ):
		lowerCAmelCase       :     Optional[Any]   				=			AddedToken(snake_case__			,       lstrip=snake_case__			,       rstrip=snake_case__     ) if isinstance(snake_case__			,       snake_case__     ) else bos_token
		lowerCAmelCase       :     List[Any]   				=			AddedToken(snake_case__			,       lstrip=snake_case__			,       rstrip=snake_case__     ) if isinstance(snake_case__			,       snake_case__     ) else eos_token
		lowerCAmelCase       :     Any   				=			AddedToken(snake_case__			,       lstrip=snake_case__			,       rstrip=snake_case__     ) if isinstance(snake_case__			,       snake_case__     ) else sep_token
		lowerCAmelCase       :     str   				=			AddedToken(snake_case__			,       lstrip=snake_case__			,       rstrip=snake_case__     ) if isinstance(snake_case__			,       snake_case__     ) else cls_token
		lowerCAmelCase       :     Tuple   				=			AddedToken(snake_case__			,       lstrip=snake_case__			,       rstrip=snake_case__     ) if isinstance(snake_case__			,       snake_case__     ) else unk_token
		lowerCAmelCase       :     List[Any]   				=			AddedToken(snake_case__			,       lstrip=snake_case__			,       rstrip=snake_case__     ) if isinstance(snake_case__			,       snake_case__     ) else pad_token

		# Mask token behave like a normal word, i.e. include the space before it
		lowerCAmelCase       :     List[str]   				=			AddedToken(snake_case__			,       lstrip=snake_case__			,       rstrip=snake_case__     ) if isinstance(snake_case__			,       snake_case__     ) else mask_token

		super().__init__(
		    errors=snake_case__			,       bos_token=snake_case__			,       eos_token=snake_case__			,       unk_token=snake_case__			,       sep_token=snake_case__			,       cls_token=snake_case__			,       pad_token=snake_case__			,       mask_token=snake_case__			,       add_prefix_space=snake_case__			,       **snake_case__			,       )

		with open(snake_case__			,       encoding='utf-8'     ) as vocab_handle:
			lowerCAmelCase       :     Tuple   				=			json.load(snake_case__     )
		lowerCAmelCase       :     Union[str, Any]   				=			{v: k for k, v in self.encoder.items()}
		lowerCAmelCase       :     List[str]   				=			errors  # how to handle errors in decoding
		lowerCAmelCase       :     Tuple   				=			bytes_to_unicode()
		lowerCAmelCase       :     Dict   				=			{v: k for k, v in self.byte_encoder.items()}
		with open(snake_case__			,       encoding='utf-8'     ) as merges_handle:
			lowerCAmelCase       :     List[str]   				=			merges_handle.read().split('\n'     )[1:-1]
		lowerCAmelCase       :     Tuple   				=			[tuple(merge.split()     ) for merge in bpe_merges]
		lowerCAmelCase       :     Optional[Any]   				=			dict(zip(snake_case__			,       range(len(snake_case__     )     )     )     )
		lowerCAmelCase       :     List[str]   				=			{}
		lowerCAmelCase       :     Dict   				=			add_prefix_space

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



	@property
	def 					lowercase  (       self     ):
		return len(self.encoder     )



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



	def 					lowercase  (       self			,       snake_case__     ):
		if token in self.cache:
			return self.cache[token]
		lowerCAmelCase       :     List[str]   				=			tuple(snake_case__     )
		lowerCAmelCase       :     Union[str, Any]   				=			get_pairs(snake_case__     )

		if not pairs:
			return token

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

				if word[i] == first and i < len(snake_case__     ) - 1 and word[i + 1] == second:
					new_word.append(first + second     )
					i += 2
				else:
					new_word.append(word[i]     )
					i += 1
			lowerCAmelCase       :     int   				=			tuple(snake_case__     )
			lowerCAmelCase       :     List[Any]   				=			new_word
			if len(snake_case__     ) == 1:
				break
			else:
				lowerCAmelCase       :     Union[str, Any]   				=			get_pairs(snake_case__     )
		lowerCAmelCase       :     Optional[Any]   				=			' '.join(snake_case__     )
		lowerCAmelCase       :     Any   				=			word
		return word



	def 					lowercase  (       self			,       snake_case__     ):
		lowerCAmelCase       :     List[Any]   				=			[]
		for token in re.findall(self.pat			,       snake_case__     ):
			lowerCAmelCase       :     Dict   				=			''.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(snake_case__     ).split(' '     )     )
		return bpe_tokens



	def 					lowercase  (       self			,       snake_case__     ):
		return self.encoder.get(snake_case__			,       self.encoder.get(self.unk_token     )     )



	def 					lowercase  (       self			,       snake_case__     ):
		return self.decoder.get(snake_case__     )



	def 					lowercase  (       self			,       snake_case__     ):
		lowerCAmelCase       :     Union[str, Any]   				=			''.join(snake_case__     )
		lowerCAmelCase       :     Any   				=			bytearray([self.byte_decoder[c] for c in text]     ).decode('utf-8'			,       errors=self.errors     )
		return text



	def 					lowercase  (       self			,       snake_case__			,       snake_case__ = None     ):
		if not os.path.isdir(snake_case__     ):
			logger.error(f"Vocabulary path ({save_directory}) should be a directory"     )
			return
		lowerCAmelCase       :     Union[str, Any]   				=			os.path.join(
		    snake_case__			,       (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']     )
		lowerCAmelCase       :     int   				=			os.path.join(
		    snake_case__			,       (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['merges_file']     )

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

		lowerCAmelCase       :     List[Any]   				=			0
		with open(snake_case__			,       '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 snake_case__     : 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!'     )
					lowerCAmelCase       :     int   				=			token_index
				writer.write(' '.join(snake_case__     ) + '\n'     )
				index += 1

		return vocab_file, merge_file



	def 					lowercase  (       self			,       snake_case__			,       snake_case__ = None     ):
		if token_ids_a is None:
			return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
		lowerCAmelCase       :     Optional[Any]   				=			[self.cls_token_id]
		lowerCAmelCase       :     List[str]   				=			[self.sep_token_id]
		return cls + token_ids_a + sep + sep + token_ids_a + sep



	def 					lowercase  (       self			,       snake_case__			,       snake_case__ = None			,       snake_case__ = False     ):
		if already_has_special_tokens:
			return super().get_special_tokens_mask(
			    token_ids_a=snake_case__			,       token_ids_a=snake_case__			,       already_has_special_tokens=snake_case__     )

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



	def 					lowercase  (       self			,       snake_case__			,       snake_case__ = None     ):
		lowerCAmelCase       :     Union[str, Any]   				=			[self.sep_token_id]
		lowerCAmelCase       :     Tuple   				=			[self.cls_token_id]

		if token_ids_a is None:
			return len(cls + token_ids_a + sep     ) * [0]
		return len(cls + token_ids_a + sep + sep + token_ids_a + sep     ) * [0]



	def 					lowercase  (       self			,       snake_case__			,       snake_case__=False			,       **snake_case__     ):
		lowerCAmelCase       :     List[str]   				=			kwargs.pop('add_prefix_space'			,       self.add_prefix_space     )
		if (is_split_into_words or add_prefix_space) and (len(snake_case__     ) > 0 and not text[0].isspace()):
			lowerCAmelCase       :     Optional[Any]   				=			' ' + text
		return (text, kwargs)
