






import os
import unicodedata
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

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


lowerCAmelCase_           = logging.get_logger(__name__)

lowerCAmelCase_           = {'vocab_file': 'spiece.model'}

lowerCAmelCase_           = {
    'vocab_file': {
        'xlnet-base-cased': 'https://huggingface.co/xlnet-base-cased/resolve/main/spiece.model',
        'xlnet-large-cased': 'https://huggingface.co/xlnet-large-cased/resolve/main/spiece.model',
    }
}

lowerCAmelCase_           = {
    'xlnet-base-cased': None,
    'xlnet-large-cased': None,
}

# Segments (not really needed)
lowerCAmelCase_           = 0
lowerCAmelCase_           = 1
lowerCAmelCase_           = 2
lowerCAmelCase_           = 3
lowerCAmelCase_           = 4
class      UpperCamelCase (				snake_case__ ):







		"""simple docstring"""



		snake_case    =   VOCAB_FILES_NAMES
		snake_case    =   PRETRAINED_VOCAB_FILES_MAP
		snake_case    =   PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
		snake_case    =   "left"
		def __init__(  self		:       Union[str, Any]						,_SCREAMING_SNAKE_CASE		:       str						,_SCREAMING_SNAKE_CASE		:       List[Any]=False						,_SCREAMING_SNAKE_CASE		:       Tuple=True						,_SCREAMING_SNAKE_CASE		:       str=False						,_SCREAMING_SNAKE_CASE		:       Union[str, Any]="<s>"						,_SCREAMING_SNAKE_CASE		:       Any="</s>"						,_SCREAMING_SNAKE_CASE		:       Any="<unk>"						,_SCREAMING_SNAKE_CASE		:       str="<sep>"						,_SCREAMING_SNAKE_CASE		:       Optional[int]="<pad>"						,_SCREAMING_SNAKE_CASE		:       Tuple="<cls>"						,_SCREAMING_SNAKE_CASE		:       Union[str, Any]="<mask>"						,_SCREAMING_SNAKE_CASE		:       List[str]=["<eop>", "<eod>"]						,_SCREAMING_SNAKE_CASE		:       Optional[Dict[str, Any]] = None						,**_SCREAMING_SNAKE_CASE		:       str						,)    ->   None:


						'''simple docstring'''
						# Mask token behave like a normal word, i.e. include the space before it
						A	 =      AddedToken(_SCREAMING_SNAKE_CASE						,lstrip=_SCREAMING_SNAKE_CASE						,rstrip=_SCREAMING_SNAKE_CASE		) if isinstance(_SCREAMING_SNAKE_CASE						,_SCREAMING_SNAKE_CASE		) else mask_token

						A	 =      {} if sp_model_kwargs is None else sp_model_kwargs

						super().__init__(
						    do_lower_case=_SCREAMING_SNAKE_CASE						,remove_space=_SCREAMING_SNAKE_CASE						,keep_accents=_SCREAMING_SNAKE_CASE						,bos_token=_SCREAMING_SNAKE_CASE						,eos_token=_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						,additional_special_tokens=_SCREAMING_SNAKE_CASE						,sp_model_kwargs=self.sp_model_kwargs						,**_SCREAMING_SNAKE_CASE						,)

						A	 =      3

						A	 =      do_lower_case
						A	 =      remove_space
						A	 =      keep_accents
						A	 =      vocab_file

						A	 =      spm.SentencePieceProcessor(**self.sp_model_kwargs		)
						self.sp_model.Load(_SCREAMING_SNAKE_CASE		)
		@property
		def       A(  self		:       Tuple		)    ->   str:


						'''simple docstring'''
						return len(self.sp_model		)
		def       A(  self		:       int		)    ->   Optional[int]:


						'''simple docstring'''
						A	 =      {self.convert_ids_to_tokens(_SCREAMING_SNAKE_CASE		): i for i in range(self.vocab_size		)}
						vocab.update(self.added_tokens_encoder		)
						return vocab
		def __getstate__(  self		:       List[Any]		)    ->   Optional[int]:


						'''simple docstring'''
						A	 =      self.__dict__.copy()
						A	 =      None
						return state
		def __setstate__(  self		:       List[str]						,_SCREAMING_SNAKE_CASE		:       Optional[Any]		)    ->   List[str]:


						'''simple docstring'''
						A	 =      d

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

						A	 =      spm.SentencePieceProcessor(**self.sp_model_kwargs		)
						self.sp_model.Load(self.vocab_file		)
		def       A(  self		:       int						,_SCREAMING_SNAKE_CASE		:       Optional[Any]		)    ->   str:


						'''simple docstring'''
						if self.remove_space:
										A	 =      ' '.join(inputs.strip().split()		)
						else:
										A	 =      inputs
						A	 =      outputs.replace('``'						,'"'		).replace('\'\''						,'"'		)

						if not self.keep_accents:
										A	 =      unicodedata.normalize('NFKD'						,_SCREAMING_SNAKE_CASE		)
										A	 =      ''.join([c for c in outputs if not unicodedata.combining(_SCREAMING_SNAKE_CASE		)]		)
						if self.do_lower_case:
										A	 =      outputs.lower()

						return outputs
		def       A(  self		:       int						,_SCREAMING_SNAKE_CASE		:       str		)    ->   List[str]:


						'''simple docstring'''
						A	 =      self.preprocess_text(_SCREAMING_SNAKE_CASE		)
						A	 =      self.sp_model.encode(_SCREAMING_SNAKE_CASE						,out_type=_SCREAMING_SNAKE_CASE		)
						A	 =      []
						for piece in pieces:
										if len(_SCREAMING_SNAKE_CASE		) > 1 and piece[-1] == str(','		) and piece[-2].isdigit():
														A	 =      self.sp_model.EncodeAsPieces(piece[:-1].replace(_SCREAMING_SNAKE_CASE						,''		)		)
														if piece[0] != SPIECE_UNDERLINE and cur_pieces[0][0] == SPIECE_UNDERLINE:
																		if len(cur_pieces[0]		) == 1:
																						A	 =      cur_pieces[1:]
																		else:
																						A	 =      cur_pieces[0][1:]
														cur_pieces.append(piece[-1]		)
														new_pieces.extend(_SCREAMING_SNAKE_CASE		)
										else:
														new_pieces.append(_SCREAMING_SNAKE_CASE		)

						return new_pieces
		def       A(  self		:       Tuple						,_SCREAMING_SNAKE_CASE		:       Tuple		)    ->   Any:


						'''simple docstring'''
						return self.sp_model.PieceToId(_SCREAMING_SNAKE_CASE		)
		def       A(  self		:       Any						,_SCREAMING_SNAKE_CASE		:       Any		)    ->   Optional[int]:


						'''simple docstring'''
						return self.sp_model.IdToPiece(_SCREAMING_SNAKE_CASE		)
		def       A(  self		:       Union[str, Any]						,_SCREAMING_SNAKE_CASE		:       Optional[int]		)    ->   Any:


						'''simple docstring'''
						A	 =      ''.join(_SCREAMING_SNAKE_CASE		).replace(_SCREAMING_SNAKE_CASE						,' '		).strip()
						return out_string
		def       A(  self		:       Tuple						,_SCREAMING_SNAKE_CASE		:       List[int]						,_SCREAMING_SNAKE_CASE		:       bool = False						,_SCREAMING_SNAKE_CASE		:       bool = None						,_SCREAMING_SNAKE_CASE		:       bool = True						,**_SCREAMING_SNAKE_CASE		:       Optional[int]						,)    ->   str:


						'''simple docstring'''
						A	 =      kwargs.pop('use_source_tokenizer'						,_SCREAMING_SNAKE_CASE		)

						A	 =      self.convert_ids_to_tokens(_SCREAMING_SNAKE_CASE						,skip_special_tokens=_SCREAMING_SNAKE_CASE		)

						# To avoid mixing byte-level and unicode for byte-level BPT
						# we need to build string separately for added tokens and byte-level tokens
						# cf. https://github.com/huggingface/transformers/issues/1133
						A	 =      []
						A	 =      []
						for token in filtered_tokens:
										if skip_special_tokens and token in self.all_special_ids:
														continue
										if token in self.added_tokens_encoder:
														if current_sub_text:
																		sub_texts.append(self.convert_tokens_to_string(_SCREAMING_SNAKE_CASE		)		)
																		A	 =      []
														sub_texts.append(_SCREAMING_SNAKE_CASE		)
										else:
														current_sub_text.append(_SCREAMING_SNAKE_CASE		)
						if current_sub_text:
										sub_texts.append(self.convert_tokens_to_string(_SCREAMING_SNAKE_CASE		)		)

						# Mimic the behavior of the Rust tokenizer:
						# By default, there are no spaces between special tokens
						A	 =      ''.join(_SCREAMING_SNAKE_CASE		)

						A	 =      (
						    clean_up_tokenization_spaces
						    if clean_up_tokenization_spaces is not None
						    else self.clean_up_tokenization_spaces
						)
						if clean_up_tokenization_spaces:
										A	 =      self.clean_up_tokenization(_SCREAMING_SNAKE_CASE		)
										return clean_text
						else:
										return text
		def       A(  self		:       Optional[Any]						,_SCREAMING_SNAKE_CASE		:       List[int]						,_SCREAMING_SNAKE_CASE		:       Optional[List[int]] = None		)    ->   List[int]:


						'''simple docstring'''
						A	 =      [self.sep_token_id]
						A	 =      [self.cls_token_id]
						if token_ids_a is None:
										return token_ids_a + sep + cls
						return token_ids_a + sep + token_ids_a + sep + cls
		def       A(  self		:       List[Any]						,_SCREAMING_SNAKE_CASE		:       List[int]						,_SCREAMING_SNAKE_CASE		:       Optional[List[int]] = None						,_SCREAMING_SNAKE_CASE		:       bool = False		)    ->   List[int]:


						'''simple docstring'''

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

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


						'''simple docstring'''
						A	 =      [self.sep_token_id]
						A	 =      [2]

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







		def       A(  self		:       List[str]						,_SCREAMING_SNAKE_CASE		:       str						,_SCREAMING_SNAKE_CASE		:       Optional[str] = None		)    ->   Tuple[str]:


						'''simple docstring'''
						if not os.path.isdir(_SCREAMING_SNAKE_CASE		):
										logger.error(f'Vocabulary path ({save_directory}) should be a directory'		)
										return
						A	 =      os.path.join(
						    _SCREAMING_SNAKE_CASE						,(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']		)

						if os.path.abspath(self.vocab_file		) != os.path.abspath(_SCREAMING_SNAKE_CASE		) and os.path.isfile(self.vocab_file		):
										copyfile(self.vocab_file						,_SCREAMING_SNAKE_CASE		)
						elif not os.path.isfile(self.vocab_file		):
										with open(_SCREAMING_SNAKE_CASE						,'wb'		) as fi:
														A	 =      self.sp_model.serialized_model_proto()
														fi.write(_SCREAMING_SNAKE_CASE		)

						return (out_vocab_file,)













import unittest

import numpy as np

from transformers import MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING, TF_MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING
from transformers.pipelines import AudioClassificationPipeline, pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_tf,
    require_torch,
    require_torchaudio,
    slow,
)

from .test_pipelines_common import ANY
@is_pipeline_test
class      UpperCamelCase (				unittest.TestCase ):







		"""simple docstring"""



		snake_case    =   MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING
		snake_case    =   TF_MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING
		def       A(  self		:       int						,_SCREAMING_SNAKE_CASE		:       Any						,_SCREAMING_SNAKE_CASE		:       List[Any]						,_SCREAMING_SNAKE_CASE		:       Tuple		)    ->   Dict:


						'''simple docstring'''
						A	 =      AudioClassificationPipeline(model=_SCREAMING_SNAKE_CASE						,feature_extractor=_SCREAMING_SNAKE_CASE		)

						# test with a raw waveform
						A	 =      np.zeros((3_4_0_0_0,)		)
						A	 =      np.zeros((1_4_0_0_0,)		)
						return audio_classifier, [audioa, audio]
		def       A(  self		:       int						,_SCREAMING_SNAKE_CASE		:       Tuple						,_SCREAMING_SNAKE_CASE		:       List[Any]		)    ->   Union[str, Any]:


						'''simple docstring'''
						A ,		A	 =      examples
						A	 =      audio_classifier(_SCREAMING_SNAKE_CASE		)
						# by default a model is initialized with num_labels=2
						self.assertEqual(
						    _SCREAMING_SNAKE_CASE						,[
						        {'score': ANY(_SCREAMING_SNAKE_CASE		), 'label': ANY(_SCREAMING_SNAKE_CASE		)},
						        {'score': ANY(_SCREAMING_SNAKE_CASE		), 'label': ANY(_SCREAMING_SNAKE_CASE		)},
						    ]						,)
						A	 =      audio_classifier(_SCREAMING_SNAKE_CASE						,top_k=1		)
						self.assertEqual(
						    _SCREAMING_SNAKE_CASE						,[
						        {'score': ANY(_SCREAMING_SNAKE_CASE		), 'label': ANY(_SCREAMING_SNAKE_CASE		)},
						    ]						,)

						self.run_torchaudio(_SCREAMING_SNAKE_CASE		)
		@require_torchaudio
		def       A(  self		:       Optional[int]						,_SCREAMING_SNAKE_CASE		:       int		)    ->   Optional[Any]:


						'''simple docstring'''
						import datasets

						# test with a local file
						A	 =      datasets.load_dataset('hf-internal-testing/librispeech_asr_dummy'						,'clean'						,split='validation'		)
						A	 =      dataset[0]['audio']['array']
						A	 =      audio_classifier(_SCREAMING_SNAKE_CASE		)
						self.assertEqual(
						    _SCREAMING_SNAKE_CASE						,[
						        {'score': ANY(_SCREAMING_SNAKE_CASE		), 'label': ANY(_SCREAMING_SNAKE_CASE		)},
						        {'score': ANY(_SCREAMING_SNAKE_CASE		), 'label': ANY(_SCREAMING_SNAKE_CASE		)},
						    ]						,)
		@require_torch
		def       A(  self		:       int		)    ->   str:


						'''simple docstring'''
						A	 =      'anton-l/wav2vec2-random-tiny-classifier'

						A	 =      pipeline('audio-classification'						,model=_SCREAMING_SNAKE_CASE		)

						A	 =      np.ones((8_0_0_0,)		)
						A	 =      audio_classifier(_SCREAMING_SNAKE_CASE						,top_k=4		)

						A	 =      [
						    {'score': 0.0842, 'label': 'no'},
						    {'score': 0.0838, 'label': 'up'},
						    {'score': 0.0837, 'label': 'go'},
						    {'score': 0.0834, 'label': 'right'},
						]
						A	 =      [
						    {'score': 0.0845, 'label': 'stop'},
						    {'score': 0.0844, 'label': 'on'},
						    {'score': 0.0841, 'label': 'right'},
						    {'score': 0.0834, 'label': 'left'},
						]
						self.assertIn(nested_simplify(_SCREAMING_SNAKE_CASE						,decimals=4		)						,[EXPECTED_OUTPUT, EXPECTED_OUTPUT_PT_2]		)

						A	 =      {'array': np.ones((8_0_0_0,)		), 'sampling_rate': audio_classifier.feature_extractor.sampling_rate}
						A	 =      audio_classifier(_SCREAMING_SNAKE_CASE						,top_k=4		)
						self.assertIn(nested_simplify(_SCREAMING_SNAKE_CASE						,decimals=4		)						,[EXPECTED_OUTPUT, EXPECTED_OUTPUT_PT_2]		)
		@require_torch
		@slow
		def       A(  self		:       Optional[int]		)    ->   Union[str, Any]:


						'''simple docstring'''
						import datasets

						A	 =      'superb/wav2vec2-base-superb-ks'

						A	 =      pipeline('audio-classification'						,model=_SCREAMING_SNAKE_CASE		)
						A	 =      datasets.load_dataset('anton-l/superb_dummy'						,'ks'						,split='test'		)

						A	 =      np.array(dataset[3]['speech']						,dtype=np.floataa		)
						A	 =      audio_classifier(_SCREAMING_SNAKE_CASE						,top_k=4		)
						self.assertEqual(
						    nested_simplify(_SCREAMING_SNAKE_CASE						,decimals=3		)						,[
						        {'score': 0.981, 'label': 'go'},
						        {'score': 0.007, 'label': 'up'},
						        {'score': 0.006, 'label': '_unknown_'},
						        {'score': 0.001, 'label': 'down'},
						    ]						,)







		@require_tf
		@unittest.skip('Audio classification is not implemented for TF'		)
		def       A(  self		:       Any		)    ->   int:


						'''simple docstring'''
						pass





