


import io
import os
import unicodedata
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

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


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

lowerCAmelCase__  :str   			=							'''▁'''

lowerCAmelCase__  :Union[str, Any]   			=							{'''vocab_file''': '''vocab.txt''', '''sentencepiece_model_ckpt''': '''sentencepiece.bpe.model'''}

lowerCAmelCase__  :List[str]   			=							{
    '''sentencepiece_model_file''': '''sentencepiece.bpe.model''',
    '''vocab_file''': '''vocab.txt''',
}

lowerCAmelCase__  :List[Any]   			=							{
    '''vocab_file''': {
        '''ernie-m-base''': '''https://huggingface.co/susnato/ernie-m-base_pytorch/blob/main/vocab.txt''',
        '''ernie-m-large''': '''https://huggingface.co/susnato/ernie-m-base_pytorch/blob/main/vocab.txt''',
    },
    '''sentencepiece_model_file''': {
        '''ernie-m-base''': '''https://huggingface.co/susnato/ernie-m-base_pytorch/blob/main/sentencepiece.bpe.model''',
        '''ernie-m-large''': '''https://huggingface.co/susnato/ernie-m-base_pytorch/blob/main/sentencepiece.bpe.model''',
    },
}

lowerCAmelCase__  :List[str]   			=							{
    '''ernie-m-base''': 5_1_4,
    '''ernie-m-large''': 5_1_4,
}

lowerCAmelCase__  :Tuple   			=							{
    '''ernie-m-base''': {'''do_lower_case''': False},
    '''ernie-m-large''': {'''do_lower_case''': False},
}

class 		__a      (				UpperCAmelCase			):
		_a	:  str				=	['input_ids']

		_a	:  Union[str, Any]				=	VOCAB_FILES_NAMES
		_a	:  Tuple				=	PRETRAINED_INIT_CONFIGURATION
		_a	:  Optional[Any]				=	PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
		_a	:  Optional[Any]				=	PRETRAINED_VOCAB_FILES_MAP
		_a	:  Optional[Any]				=	RESOURCE_FILES_NAMES
		def __init__(			self			,    _SCREAMING_SNAKE_CASE			,    _SCREAMING_SNAKE_CASE=None			,    _SCREAMING_SNAKE_CASE=False			,    _SCREAMING_SNAKE_CASE="utf8"			,    _SCREAMING_SNAKE_CASE="[UNK]"			,    _SCREAMING_SNAKE_CASE="[SEP]"			,    _SCREAMING_SNAKE_CASE="[PAD]"			,    _SCREAMING_SNAKE_CASE="[CLS]"			,    _SCREAMING_SNAKE_CASE="[MASK]"			,    _SCREAMING_SNAKE_CASE = None			,    **_SCREAMING_SNAKE_CASE			,    ) ->       None:




									"""simple docstring"""





									_UpperCAmelCase 		=			{} if sp_model_kwargs is None else sp_model_kwargs
									super().__init__(
									    do_lower_case=_SCREAMING_SNAKE_CASE			,    unk_token=_SCREAMING_SNAKE_CASE			,    sep_token=_SCREAMING_SNAKE_CASE			,    pad_token=_SCREAMING_SNAKE_CASE			,    cls_token=_SCREAMING_SNAKE_CASE			,    mask_token=_SCREAMING_SNAKE_CASE			,    vocab_file=_SCREAMING_SNAKE_CASE			,    encoding=_SCREAMING_SNAKE_CASE			,    sp_model_kwargs=self.sp_model_kwargs			,    **_SCREAMING_SNAKE_CASE			,    )
									_UpperCAmelCase 		=			do_lower_case
									_UpperCAmelCase 		=			sentencepiece_model_ckpt
									_UpperCAmelCase 		=			spm.SentencePieceProcessor(**self.sp_model_kwargs						)
									self.sp_model.Load(_SCREAMING_SNAKE_CASE						)

									# to mimic paddlenlp.transformers.ernie_m.tokenizer.ErnieMTokenizer functioning
									if vocab_file is not None:
																_UpperCAmelCase 		=			self.load_vocab(filepath=_SCREAMING_SNAKE_CASE						)
									else:
																_UpperCAmelCase 		=			{self.sp_model.id_to_piece(_SCREAMING_SNAKE_CASE						): id for id in range(self.sp_model.get_piece_size()						)}
									_UpperCAmelCase 		=			{v: k for k, v in self.vocab.items()}
		def 	UpperCAmelCase__				(			self			,    _SCREAMING_SNAKE_CASE						) ->       List[str]:




									"""simple docstring"""





									if text is None:
																return None

									_UpperCAmelCase 		=			self.tokenize(_SCREAMING_SNAKE_CASE						)
									_UpperCAmelCase		,       _UpperCAmelCase 		=			'', []

									for i, ch in enumerate(_SCREAMING_SNAKE_CASE						):
																if ch in self.SP_CHAR_MAPPING:
																							_UpperCAmelCase 		=			self.SP_CHAR_MAPPING.get(_SCREAMING_SNAKE_CASE						)
																else:
																							_UpperCAmelCase 		=			unicodedata.normalize('NFKC'			,    _SCREAMING_SNAKE_CASE						)
																if self.is_whitespace(_SCREAMING_SNAKE_CASE						):
																							continue
																normalized_text += ch
																char_mapping.extend([i] * len(_SCREAMING_SNAKE_CASE						)						)

									_UpperCAmelCase		,       _UpperCAmelCase		,       _UpperCAmelCase 		=			normalized_text, [], 0

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

									for token in split_tokens:
																if token[:1] == "▁":
																							_UpperCAmelCase 		=			token[1:]
																_UpperCAmelCase 		=			text[offset:].index(_SCREAMING_SNAKE_CASE						) + offset
																_UpperCAmelCase 		=			start + len(_SCREAMING_SNAKE_CASE						)

																token_mapping.append((char_mapping[start], char_mapping[end - 1] + 1)						)
																_UpperCAmelCase 		=			end
									return token_mapping
		@property
		def 	UpperCAmelCase__				(			self						) ->       Any:




									"""simple docstring"""





									return len(self.vocab						)
		def 	UpperCAmelCase__				(			self						) ->       Dict:




									"""simple docstring"""





									return dict(self.vocab			,    **self.added_tokens_encoder						)
		def __getstate__(			self						) ->       str:




									"""simple docstring"""





									_UpperCAmelCase 		=			self.__dict__.copy()
									_UpperCAmelCase 		=			None
									return state
		def __setstate__(			self			,    _SCREAMING_SNAKE_CASE						) ->       List[str]:




									"""simple docstring"""





									_UpperCAmelCase 		=			d

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

									_UpperCAmelCase 		=			spm.SentencePieceProcessor(**self.sp_model_kwargs						)
									self.sp_model.Load(self.sentencepiece_model_ckpt						)
		def 	UpperCAmelCase__				(			self			,    _SCREAMING_SNAKE_CASE						) ->       str:




									"""simple docstring"""





									return "".join((self.SP_CHAR_MAPPING.get(_SCREAMING_SNAKE_CASE			,    _SCREAMING_SNAKE_CASE						) for c in text)						)
		def 	UpperCAmelCase__				(			self			,    _SCREAMING_SNAKE_CASE			,    _SCREAMING_SNAKE_CASE=False			,    _SCREAMING_SNAKE_CASE=64			,    _SCREAMING_SNAKE_CASE=0.1						) ->       Union[str, Any]:




									"""simple docstring"""





									if self.sp_model_kwargs.get('enable_sampling'						) is True:
																_UpperCAmelCase 		=			True
									if self.sp_model_kwargs.get('alpha'						) is not None:
																_UpperCAmelCase 		=			self.sp_model_kwargs.get('alpha'						)
									if self.sp_model_kwargs.get('nbest_size'						) is not None:
																_UpperCAmelCase 		=			self.sp_model_kwargs.get('nbest_size'						)

									if not enable_sampling:
																_UpperCAmelCase 		=			self.sp_model.EncodeAsPieces(_SCREAMING_SNAKE_CASE						)
									else:
																_UpperCAmelCase 		=			self.sp_model.SampleEncodeAsPieces(_SCREAMING_SNAKE_CASE			,    _SCREAMING_SNAKE_CASE			,    _SCREAMING_SNAKE_CASE						)
									_UpperCAmelCase 		=			[]
									for pi, piece in enumerate(_SCREAMING_SNAKE_CASE						):
																if piece == SPIECE_UNDERLINE:
																							if not pieces[pi + 1].startswith(_SCREAMING_SNAKE_CASE						) and pi != 0:
																														new_pieces.append(_SCREAMING_SNAKE_CASE						)
																														continue
																							else:
																														continue
																_UpperCAmelCase 		=			0
																for i, chunk in enumerate(_SCREAMING_SNAKE_CASE						):
																							if chunk == SPIECE_UNDERLINE:
																														continue
																							if self.is_ch_char(_SCREAMING_SNAKE_CASE						) or self.is_punct(_SCREAMING_SNAKE_CASE						):
																														if i > lst_i and piece[lst_i:i] != SPIECE_UNDERLINE:
																																					new_pieces.append(piece[lst_i:i]						)
																														new_pieces.append(_SCREAMING_SNAKE_CASE						)
																														_UpperCAmelCase 		=			i + 1
																							elif chunk.isdigit() and i > 0 and not piece[i - 1].isdigit():
																														if i > lst_i and piece[lst_i:i] != SPIECE_UNDERLINE:
																																					new_pieces.append(piece[lst_i:i]						)
																														_UpperCAmelCase 		=			i
																							elif not chunk.isdigit() and i > 0 and piece[i - 1].isdigit():
																														if i > lst_i and piece[lst_i:i] != SPIECE_UNDERLINE:
																																					new_pieces.append(piece[lst_i:i]						)
																														_UpperCAmelCase 		=			i
																if len(_SCREAMING_SNAKE_CASE						) > lst_i:
																							new_pieces.append(piece[lst_i:]						)
									return new_pieces
		def 	UpperCAmelCase__				(			self			,    _SCREAMING_SNAKE_CASE						) ->       Any:




									"""simple docstring"""





									_UpperCAmelCase 		=			''.join(_SCREAMING_SNAKE_CASE						).replace(_SCREAMING_SNAKE_CASE			,    ' '						).strip()
									return out_string
		def 	UpperCAmelCase__				(			self			,    _SCREAMING_SNAKE_CASE						) ->       Union[str, Any]:




									"""simple docstring"""





									_UpperCAmelCase 		=			self.convert_ids_to_tokens(_SCREAMING_SNAKE_CASE						)
									_UpperCAmelCase 		=			''.join(_SCREAMING_SNAKE_CASE						).replace(_SCREAMING_SNAKE_CASE			,    ' '						).strip()
									return out_string
		def 	UpperCAmelCase__				(			self			,    _SCREAMING_SNAKE_CASE						) ->       Optional[int]:




									"""simple docstring"""





									return self.vocab.get(_SCREAMING_SNAKE_CASE			,    self.vocab.get(self.unk_token						)						)
		def 	UpperCAmelCase__				(			self			,    _SCREAMING_SNAKE_CASE						) ->       Dict:




									"""simple docstring"""





									return self.reverse_vocab.get(_SCREAMING_SNAKE_CASE			,    self.unk_token						)
		def 	UpperCAmelCase__				(			self			,    _SCREAMING_SNAKE_CASE			,    _SCREAMING_SNAKE_CASE=None						) ->       Dict:




									"""simple docstring"""





									if token_ids_a is None:
																return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
									_UpperCAmelCase 		=			[self.cls_token_id]
									_UpperCAmelCase 		=			[self.sep_token_id]
									return _cls + token_ids_a + _sep + _sep + token_ids_a + _sep
		def 	UpperCAmelCase__				(			self			,    _SCREAMING_SNAKE_CASE			,    _SCREAMING_SNAKE_CASE=None						) ->       int:




									"""simple docstring"""





									if offset_mapping_a is None:
																return [(0, 0)] + offset_mapping_a + [(0, 0)]

									return [(0, 0)] + offset_mapping_a + [(0, 0), (0, 0)] + offset_mapping_a + [(0, 0)]
		def 	UpperCAmelCase__				(			self			,    _SCREAMING_SNAKE_CASE			,    _SCREAMING_SNAKE_CASE=None			,    _SCREAMING_SNAKE_CASE=False						) ->       Optional[int]:




									"""simple docstring"""





									if already_has_special_tokens:
																if token_ids_a is not None:
																							raise ValueError(
																							    'You should not supply a second sequence if the provided sequence of '
																							    'ids is already formatted with special tokens for the model.'						)
																return [1 if x in [self.sep_token_id, self.cls_token_id] else 0 for x in token_ids_a]

									if token_ids_a is not None:
																return [1] + ([0] * len(_SCREAMING_SNAKE_CASE						)) + [1, 1] + ([0] * len(_SCREAMING_SNAKE_CASE						)) + [1]
									return [1] + ([0] * len(_SCREAMING_SNAKE_CASE						)) + [1]
		def 	UpperCAmelCase__				(			self			,    _SCREAMING_SNAKE_CASE			,    _SCREAMING_SNAKE_CASE = None						) ->       List[int]:




									"""simple docstring"""





									if token_ids_a is None:
																# [CLS] X [SEP]
																return (len(_SCREAMING_SNAKE_CASE						) + 2) * [0]

									# [CLS] A [SEP] [SEP] B [SEP]
									return [0] * (len(_SCREAMING_SNAKE_CASE						) + 1) + [1] * (len(_SCREAMING_SNAKE_CASE						) + 3)
		def 	UpperCAmelCase__				(			self			,    _SCREAMING_SNAKE_CASE						) ->       Optional[Any]:




									"""simple docstring"""





									if "\u4e00" <= char <= "\u9fff":
																return True
									return False
		def 	UpperCAmelCase__				(			self			,    _SCREAMING_SNAKE_CASE						) ->       Optional[Any]:




									"""simple docstring"""





									if ("a" <= char <= "z") or ("A" <= char <= "Z"):
																return True
									return False
		def 	UpperCAmelCase__				(			self			,    _SCREAMING_SNAKE_CASE						) ->       str:




									"""simple docstring"""





									if char in ",;:.?!~，；：。？！《》【】":
																return True
									return False
		def 	UpperCAmelCase__				(			self			,    _SCREAMING_SNAKE_CASE						) ->       Union[str, Any]:




									"""simple docstring"""





									if char == " " or char == "\t" or char == "\n" or char == "\r":
																return True
									if len(_SCREAMING_SNAKE_CASE						) == 1:
																_UpperCAmelCase 		=			unicodedata.category(_SCREAMING_SNAKE_CASE						)
																if cat == "Zs":
																							return True
									return False
		def 	UpperCAmelCase__				(			self			,    _SCREAMING_SNAKE_CASE						) ->       Any:




									"""simple docstring"""





									_UpperCAmelCase 		=			{}
									with io.open(_SCREAMING_SNAKE_CASE			,    'r'			,    encoding='utf-8'						) as f:
																for index, line in enumerate(_SCREAMING_SNAKE_CASE						):
																							_UpperCAmelCase 		=			line.rstrip('\n'						)
																							_UpperCAmelCase 		=			int(_SCREAMING_SNAKE_CASE						)

									return token_to_idx






		def 	UpperCAmelCase__				(			self			,    _SCREAMING_SNAKE_CASE			,    _SCREAMING_SNAKE_CASE = None						) ->       Tuple[str]:




									"""simple docstring"""





									_UpperCAmelCase 		=			0
									if os.path.isdir(_SCREAMING_SNAKE_CASE						):
																_UpperCAmelCase 		=			os.path.join(
																    _SCREAMING_SNAKE_CASE			,    (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']						)
									else:
																_UpperCAmelCase 		=			(filename_prefix + '-' if filename_prefix else '') + save_directory
									with open(_SCREAMING_SNAKE_CASE			,    'w'			,    encoding='utf-8'						) as writer:
																for token, token_index in sorted(self.vocab.items()			,    key=lambda _SCREAMING_SNAKE_CASE						: kv[1]						):
																							if index != token_index:
																														logger.warning(
																														    f'''Saving vocabulary to {vocab_file}: vocabulary indices are not consecutive.'''
																														    ' Please check that the vocabulary is not corrupted!'						)
																														_UpperCAmelCase 		=			token_index
																							writer.write(token + '\n'						)
																							index += 1

									_UpperCAmelCase 		=			os.path.join(_SCREAMING_SNAKE_CASE			,    'sentencepiece.bpe.model'						)
									with open(_SCREAMING_SNAKE_CASE			,    'wb'						) as fi:
																_UpperCAmelCase 		=			self.sp_model.serialized_model_proto()
																fi.write(_SCREAMING_SNAKE_CASE						)

									return (vocab_file,)



import copy
from typing import Any, Dict, List, Optional, Union

import numpy as np

from ...audio_utils import mel_filter_bank, spectrogram, window_function
from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import TensorType, logging


lowercase_									=   logging.get_logger(__name__)





class  __lowerCAmelCase     (					SCREAMING_SNAKE_CASE							):
  _a									=  ["""input_features"""]




  def __init__(				self							,   lowerCAmelCase=80							,   lowerCAmelCase=16_000							,   lowerCAmelCase=160							,   lowerCAmelCase=30							,   lowerCAmelCase=400							,   lowerCAmelCase=0.0							,   lowerCAmelCase=False							,   **lowerCAmelCase							,   )  -> Any:
        '''simple docstring'''






        super().__init__(
            feature_size=lowerCAmelCase							,   sampling_rate=lowerCAmelCase							,   padding_value=lowerCAmelCase							,   return_attention_mask=lowerCAmelCase							,   **lowerCAmelCase							,   )
        _lowercase	       =n_fft
        _lowercase	       =hop_length
        _lowercase	       =chunk_length
        _lowercase	       =chunk_length * sampling_rate
        _lowercase	       =self.n_samples // hop_length
        _lowercase	       =sampling_rate
        _lowercase	       =mel_filter_bank(
            num_frequency_bins=1 + n_fft // 2							,   num_mel_filters=lowerCAmelCase							,   min_frequency=0.0							,   max_frequency=8000.0							,   sampling_rate=lowerCAmelCase							,   norm='slaney'							,   mel_scale='slaney'							,   )




  def   A__     (				self							,   lowerCAmelCase					)  -> np.ndarray:
        '''simple docstring'''






        _lowercase	       =spectrogram(
            lowerCAmelCase							,   window_function(self.n_fft							,   'hann'					)							,   frame_length=self.n_fft							,   hop_length=self.hop_length							,   power=2.0							,   mel_filters=self.mel_filters							,   log_mel='log10'							,   )
        _lowercase	       =log_spec[:, :-1]
        _lowercase	       =np.maximum(lowerCAmelCase							,   log_spec.max() - 8.0					)
        _lowercase	       =(log_spec + 4.0) / 4.0
        return log_spec




  @staticmethod
  # Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
  def   A__     (				lowerCAmelCase							,   lowerCAmelCase							,   lowerCAmelCase = 0.0					)  -> List[np.ndarray]:
        '''simple docstring'''






        if attention_mask is not None:
              _lowercase	       =np.array(lowerCAmelCase							,   np.intaa					)
              _lowercase	       =[]

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

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

        return normed_input_values




  def __call__(				self							,   lowerCAmelCase							,   lowerCAmelCase = True							,   lowerCAmelCase = None							,   lowerCAmelCase = None							,   lowerCAmelCase = None							,   lowerCAmelCase = "max_length"							,   lowerCAmelCase = None							,   lowerCAmelCase = None							,   lowerCAmelCase = None							,   **lowerCAmelCase							,   )  -> BatchFeature:
        '''simple docstring'''






        if sampling_rate is not None:
              if sampling_rate != self.sampling_rate:
                    raise ValueError(
                        F'''The model corresponding to this feature extractor: {self.__class__.__name__} was trained using a'''
                        F''' sampling rate of {self.sampling_rate}. Please make sure that the provided `raw_speech` input'''
                        F''' was sampled with {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.'					)

        _lowercase	       =isinstance(lowerCAmelCase							,   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}'''					)
        _lowercase	       =is_batched_numpy or (
            isinstance(lowerCAmelCase							,   (list, tuple)					) and (isinstance(raw_speech[0]							,   (np.ndarray, tuple, list)					))
        )

        if is_batched:
              _lowercase	       =[np.asarray([speech]							,   dtype=np.floataa					).T for speech in raw_speech]
        elif not is_batched and not isinstance(lowerCAmelCase							,   np.ndarray					):
              _lowercase	       =np.asarray(lowerCAmelCase							,   dtype=np.floataa					)
        elif isinstance(lowerCAmelCase							,   np.ndarray					) and raw_speech.dtype is np.dtype(np.floataa					):
              _lowercase	       =raw_speech.astype(np.floataa					)

        # always return batch
        if not is_batched:
              _lowercase	       =[np.asarray([raw_speech]					).T]

        _lowercase	       =BatchFeature({'input_features': raw_speech}					)

        # convert into correct format for padding

        _lowercase	       =self.pad(
            lowerCAmelCase							,   padding=lowerCAmelCase							,   max_length=max_length if max_length else self.n_samples							,   truncation=lowerCAmelCase							,   pad_to_multiple_of=lowerCAmelCase							,   return_attention_mask=return_attention_mask or do_normalize							,   )

        # zero-mean and unit-variance normalization
        if do_normalize:
              _lowercase	       =self.zero_mean_unit_var_norm(
                  padded_inputs['input_features']							,   attention_mask=padded_inputs['attention_mask']							,   padding_value=self.padding_value							,   )
              _lowercase	       =np.stack(padded_inputs['input_features']							,   axis=0					)

        # make sure list is in array format
        _lowercase	       =padded_inputs.get('input_features'					).transpose(2							,   0							,   1					)

        _lowercase	       =[self._np_extract_fbank_features(lowerCAmelCase					) for waveform in input_features[0]]

        if isinstance(input_features[0]							,   lowerCAmelCase					):
              _lowercase	       =[np.asarray(lowerCAmelCase							,   dtype=np.floataa					) for feature in input_features]
        else:
              _lowercase	       =input_features

        if return_attention_mask:
              # rescale from sample (48000) to feature (3000)
              _lowercase	       =padded_inputs['attention_mask'][:, :: self.hop_length]

        if return_tensors is not None:
              _lowercase	       =padded_inputs.convert_to_tensors(lowerCAmelCase					)

        return padded_inputs




  def   A__     (				self					)  -> Dict[str, Any]:
        '''simple docstring'''






        _lowercase	       =copy.deepcopy(self.__dict__					)
        _lowercase	       =self.__class__.__name__
        if "mel_filters" in output:
              del output["mel_filters"]
        return output


