

'''simple docstring'''




import collections
import os
from typing import List, Optional, Tuple

from transformers.utils import is_jieba_available, requires_backends


if is_jieba_available():
				import jieba

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


lowerCAmelCase_							=	logging.get_logger(__name__)

lowerCAmelCase_							=	{'vocab_file': 'vocab.txt'}

lowerCAmelCase_							=	{
    'vocab_file': {
        'openbmb/cpm-ant-10b': 'https://huggingface.co/openbmb/cpm-ant-10b/blob/main/vocab.txt',
    },
}

lowerCAmelCase_							=	{
    'openbmb/cpm-ant-10b': 1024,
}
def       A__	(							A  :     List[str]):


			'''simple docstring'''


			UpperCamelCase				:	Union[str, Any]    						=					collections.OrderedDict()
			with open(_UpperCamelCase							,       "r"							,       encoding="utf-8") as reader:
						UpperCamelCase				:	List[Any]    						=					reader.readlines()
			for index, token in enumerate(_UpperCamelCase):
						UpperCamelCase				:	Tuple    						=					token.rstrip("\n")
						UpperCamelCase				:	Union[str, Any]    						=					index
			return vocab

class     UpperCAmelCase_ (	SCREAMING_SNAKE_CASE__    ):



	"""simple docstring"""





	def __init__(    self				,			lowerCamelCase				,			lowerCamelCase="<unk>"				,			lowerCamelCase=2_00   )							->     Dict:




				'''simple docstring'''






				UpperCamelCase				:	Tuple    						=					vocab
				UpperCamelCase				:	Union[str, Any]    						=					unk_token
				UpperCamelCase				:	Optional[int]    						=					max_input_chars_per_word



	def 	SCREAMING_SNAKE_CASE__							(    self				,			lowerCamelCase   )							->     Optional[int]:




				'''simple docstring'''






				UpperCamelCase				:	str    						=					list(A_   )
				if len(A_   ) > self.max_input_chars_per_word:
							return [self.unk_token]

				UpperCamelCase				:	Union[str, Any]    						=					0
				UpperCamelCase				:	Any    						=					[]
				while start < len(A_   ):
							UpperCamelCase				:	str    						=					len(A_   )
							UpperCamelCase				:	Any    						=					None
							while start < end:
										UpperCamelCase				:	int    						=					''''''.join(chars[start:end]   )
										if substr in self.vocab:
													UpperCamelCase				:	List[str]    						=					substr
													break
										end -= 1
							if cur_substr is None:
										sub_tokens.append(self.unk_token   )
										start += 1
							else:
										sub_tokens.append(A_   )
										UpperCamelCase				:	Optional[int]    						=					end

				return sub_tokens

class     UpperCAmelCase_ (	SCREAMING_SNAKE_CASE__    ):



	"""simple docstring"""





	__SCREAMING_SNAKE_CASE					=       VOCAB_FILES_NAMES
	__SCREAMING_SNAKE_CASE					=       PRETRAINED_VOCAB_FILES_MAP
	__SCREAMING_SNAKE_CASE					=       PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
	__SCREAMING_SNAKE_CASE					=       ['''input_ids''', '''attention_mask''']
	__SCREAMING_SNAKE_CASE					=       False

	def __init__(    self				,			lowerCamelCase				,			lowerCamelCase="<d>"				,			lowerCamelCase="</d>"				,			lowerCamelCase="<s>"				,			lowerCamelCase="</s>"				,			lowerCamelCase="<pad>"				,			lowerCamelCase="<unk>"				,			lowerCamelCase="</n>"				,			lowerCamelCase="</_>"				,			lowerCamelCase="left"				,			**lowerCamelCase				,			)							->     Optional[int]:




				'''simple docstring'''






				requires_backends(self				,			["jieba"]   )
				super().__init__(
				    bod_token=A_				,			eod_token=A_				,			bos_token=A_				,			eos_token=A_				,			pad_token=A_				,			unk_token=A_				,			line_token=A_				,			space_token=A_				,			padding_side=A_				,			**A_				,			)
				UpperCamelCase				:	Optional[int]    						=					bod_token
				UpperCamelCase				:	Optional[Any]    						=					eod_token
				UpperCamelCase				:	Optional[Any]    						=					load_vocab(A_   )
				UpperCamelCase				:	Union[str, Any]    						=					self.encoder[space_token]
				UpperCamelCase				:	List[Any]    						=					self.encoder[line_token]

				del self.encoder[space_token]
				del self.encoder[line_token]

				UpperCamelCase				:	Any    						=					collections.OrderedDict(sorted(self.encoder.items()				,			key=lambda lowerCamelCase   : x[1]   )   )
				UpperCamelCase				:	Any    						=					{v: k for k, v in self.encoder.items()}

				UpperCamelCase				:	Union[str, Any]    						=					WordpieceTokenizer(vocab=self.encoder				,			unk_token=self.unk_token   )

	@property
	def 	SCREAMING_SNAKE_CASE__							(    self   )							->     Any:




				'''simple docstring'''






				return self.encoder[self.bod_token]

	@property
	def 	SCREAMING_SNAKE_CASE__							(    self   )							->     Optional[Any]:




				'''simple docstring'''






				return self.encoder[self.eod_token]

	@property
	def 	SCREAMING_SNAKE_CASE__							(    self   )							->     Any:




				'''simple docstring'''






				return self.encoder["\n"]

	@property
	def 	SCREAMING_SNAKE_CASE__							(    self   )							->     int:




				'''simple docstring'''






				return len(self.encoder   )

	def 	SCREAMING_SNAKE_CASE__							(    self   )							->     Any:




				'''simple docstring'''






				return dict(self.encoder				,			**self.added_tokens_encoder   )

	def 	SCREAMING_SNAKE_CASE__							(    self				,			lowerCamelCase   )							->     Union[str, Any]:




				'''simple docstring'''






				UpperCamelCase				:	Tuple    						=					[]
				for x in jieba.cut(A_				,			cut_all=A_   ):
							output_tokens.extend(self.wordpiece_tokenizer.tokenize(A_   )   )
				return output_tokens

	def 	SCREAMING_SNAKE_CASE__							(    self				,			lowerCamelCase				,			**lowerCamelCase   )							->     Tuple:




				'''simple docstring'''






				UpperCamelCase				:	Optional[int]    						=					[i for i in token_ids if i >= 0]
				UpperCamelCase				:	Any    						=					[
				    x for x in token_ids if x != self.pad_token_id and x != self.eos_token_id and x != self.bos_token_id
				]
				return super()._decode(A_				,			**A_   )

	def 	SCREAMING_SNAKE_CASE__							(    self				,			lowerCamelCase   )							->     Union[str, Any]:




				'''simple docstring'''






				return token in self.encoder

	def 	SCREAMING_SNAKE_CASE__							(    self				,			lowerCamelCase   )							->     str:




				'''simple docstring'''






				return "".join(A_   )

	def 	SCREAMING_SNAKE_CASE__							(    self				,			lowerCamelCase   )							->     Any:




				'''simple docstring'''






				return self.encoder.get(A_				,			self.encoder.get(self.unk_token   )   )

	def 	SCREAMING_SNAKE_CASE__							(    self				,			lowerCamelCase   )							->     List[Any]:




				'''simple docstring'''






				return self.decoder.get(A_				,			self.unk_token   )

	def 	SCREAMING_SNAKE_CASE__							(    self				,			lowerCamelCase				,			lowerCamelCase = None   )							->     Tuple[str]:




				'''simple docstring'''






				if os.path.isdir(A_   ):
							UpperCamelCase				:	Optional[int]    						=					os.path.join(
							    A_				,			(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]   )
				else:
							UpperCamelCase				:	str    						=					(filename_prefix + '''-''' if filename_prefix else '''''') + save_directory
				UpperCamelCase				:	int    						=					0
				if " " in self.encoder:
							UpperCamelCase				:	str    						=					self.encoder[''' ''']
							del self.encoder[" "]
				if "\n" in self.encoder:
							UpperCamelCase				:	Any    						=					self.encoder['''\n''']
							del self.encoder["\n"]
				UpperCamelCase				:	Union[str, Any]    						=					collections.OrderedDict(sorted(self.encoder.items()				,			key=lambda lowerCamelCase   : x[1]   )   )
				with open(A_				,			"w"				,			encoding="utf-8"   ) as writer:
							for token, token_index in self.encoder.items():
										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				:	Union[str, Any]    						=					token_index
										writer.write(token + "\n"   )
										index += 1
				return (vocab_file,)

	def 	SCREAMING_SNAKE_CASE__							(    self				,			lowerCamelCase				,			lowerCamelCase = None   )							->     List[int]:




				'''simple docstring'''






				if token_ids_a is None:
							return [self.bos_token_id] + token_ids_a
				return [self.bos_token_id] + token_ids_a + [self.bos_token_id] + token_ids_a



	def 	SCREAMING_SNAKE_CASE__							(    self				,			lowerCamelCase				,			lowerCamelCase = None				,			lowerCamelCase = False   )							->     List[int]:




				'''simple docstring'''






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

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


"""simple docstring"""




import itertools
import os
import random
import tempfile
import unittest

import numpy as np
from datasets import load_dataset

from transformers import is_speech_available
from transformers.testing_utils import check_json_file_has_correct_format, require_torch, require_torchaudio
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_speech_available():
  from transformers import WhisperFeatureExtractor

if is_torch_available():
  import torch

A__ :       List[Any]				= random.Random()




def 			_lowerCAmelCase		(		_UpperCamelCase			,					_UpperCamelCase=1.0			,					_UpperCamelCase=None			,					_UpperCamelCase=None    ):






  """simple docstring"""



  if rng is None:
    _lowercase:      Any   			=  global_rng

  _lowercase:      List[str]   			=  []
  for batch_idx in range(shape[0]    ):
    values.append([]    )
    for _ in range(shape[1]    ):
      values[-1].append(rng.random() * scale    )

  return values


@require_torch
@require_torchaudio
class       __magic_name__							(							unittest.TestCase			):
       def __init__(  self			,					A_			,					A_=7			,					A_=400			,					A_=2000			,					A_=10			,					A_=160			,					A_=8			,					A_=0.0			,					A_=4000			,					A_=False			,					A_=True			,					)  ->						List[Any]:





         """simple docstring"""




         _lowercase:      str   			=  parent
         _lowercase:      int   			=  batch_size
         _lowercase:      Tuple   			=  min_seq_length
         _lowercase:      Any   			=  max_seq_length
         _lowercase:      List[str]   			=  (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
         _lowercase:      Any   			=  padding_value
         _lowercase:      List[str]   			=  sampling_rate
         _lowercase:      Union[str, Any]   			=  return_attention_mask
         _lowercase:      Optional[Any]   			=  do_normalize
         _lowercase:      List[str]   			=  feature_size
         _lowercase:      Optional[Any]   			=  chunk_length
         _lowercase:      str   			=  hop_length
       def 				lowercase_				(  self							)  ->						Optional[Any]:





         """simple docstring"""




         return {
             "feature_size": self.feature_size,
             "hop_length": self.hop_length,
             "chunk_length": self.chunk_length,
             "padding_value": self.padding_value,
             "sampling_rate": self.sampling_rate,
             "return_attention_mask": self.return_attention_mask,
             "do_normalize": self.do_normalize,
         }



       def 				lowercase_				(  self			,					A_=False			,					A_=False							)  ->						List[Any]:





         """simple docstring"""




         def _flatten(A_							):
           return list(itertools.chain(*A_							)							)

         if equal_length:
           _lowercase:      List[str]   			=  [floats_list((self.max_seq_length, self.feature_size)							) for _ in range(self.batch_size							)]
         else:
           # make sure that inputs increase in size
           _lowercase:      Union[str, Any]   			=  [
               floats_list((x, self.feature_size)							)
               for x in range(self.min_seq_length			,					self.max_seq_length			,					self.seq_length_diff							)
           ]
         if numpify:
           _lowercase:      str   			=  [np.asarray(A_							) for x in speech_inputs]
         return speech_inputs


@require_torch
@require_torchaudio
class       __magic_name__							(							SCREAMING_SNAKE_CASE__ ,       unittest.TestCase			):
       UpperCamelCase_ =		WhisperFeatureExtractor if is_speech_available() else None
       def 				lowercase_				(  self							)  ->						Dict:





         """simple docstring"""




         _lowercase:      int   			=  WhisperFeatureExtractionTester(self							)
       def 				lowercase_				(  self							)  ->						Tuple:





         """simple docstring"""




         _lowercase:      Union[str, Any]   			=  self.feature_extraction_class(**self.feat_extract_dict							)

         with tempfile.TemporaryDirectory() as tmpdirname:
           _lowercase:      Any   			=  feat_extract_first.save_pretrained(A_							)[0]
           check_json_file_has_correct_format(A_							)
           _lowercase:      str   			=  self.feature_extraction_class.from_pretrained(A_							)

         _lowercase:      Dict   			=  feat_extract_first.to_dict()
         _lowercase:      List[str]   			=  feat_extract_second.to_dict()
         _lowercase:      List[str]   			=  feat_extract_first.mel_filters
         _lowercase:      str   			=  feat_extract_second.mel_filters
         self.assertTrue(np.allclose(A_			,					A_							)							)
         self.assertEqual(A_			,					A_							)
       def 				lowercase_				(  self							)  ->						Optional[int]:





         """simple docstring"""




         _lowercase:      Dict   			=  self.feature_extraction_class(**self.feat_extract_dict							)

         with tempfile.TemporaryDirectory() as tmpdirname:
           _lowercase:      Union[str, Any]   			=  os.path.join(A_			,					'''feat_extract.json'''							)
           feat_extract_first.to_json_file(A_							)
           _lowercase:      Tuple   			=  self.feature_extraction_class.from_json_file(A_							)

         _lowercase:      int   			=  feat_extract_first.to_dict()
         _lowercase:      Optional[int]   			=  feat_extract_second.to_dict()
         _lowercase:      Tuple   			=  feat_extract_first.mel_filters
         _lowercase:      Optional[int]   			=  feat_extract_second.mel_filters
         self.assertTrue(np.allclose(A_			,					A_							)							)
         self.assertEqual(A_			,					A_							)
       def 				lowercase_				(  self							)  ->						Tuple:





         """simple docstring"""




         _lowercase:      Any   			=  self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()							)
         # create three inputs of length 800, 1000, and 1200
         _lowercase:      Tuple   			=  [floats_list((1, x)							)[0] for x in range(800			,					1400			,					200							)]
         _lowercase:      List[str]   			=  [np.asarray(A_							) for speech_input in speech_inputs]

         # Test feature size
         _lowercase:      Dict   			=  feature_extractor(A_			,					padding='''max_length'''			,					return_tensors='''np'''							).input_features
         self.assertTrue(input_features.ndim == 3							)
         self.assertTrue(input_features.shape[-1] == feature_extractor.nb_max_frames							)
         self.assertTrue(input_features.shape[-2] == feature_extractor.feature_size							)

         # Test not batched input
         _lowercase:      Dict   			=  feature_extractor(speech_inputs[0]			,					return_tensors='''np'''							).input_features
         _lowercase:      Dict   			=  feature_extractor(np_speech_inputs[0]			,					return_tensors='''np'''							).input_features
         self.assertTrue(np.allclose(A_			,					A_			,					atol=1E-3							)							)

         # Test batched
         _lowercase:      Union[str, Any]   			=  feature_extractor(A_			,					return_tensors='''np'''							).input_features
         _lowercase:      List[Any]   			=  feature_extractor(A_			,					return_tensors='''np'''							).input_features
         for enc_seq_a, enc_seq_a in zip(A_			,					A_							):
           self.assertTrue(np.allclose(A_			,					A_			,					atol=1E-3							)							)

         # Test 2-D numpy arrays are batched.
         _lowercase:      Optional[Any]   			=  [floats_list((1, x)							)[0] for x in (800, 800, 800)]
         _lowercase:      Any   			=  np.asarray(A_							)
         _lowercase:      Tuple   			=  feature_extractor(A_			,					return_tensors='''np'''							).input_features
         _lowercase:      Dict   			=  feature_extractor(A_			,					return_tensors='''np'''							).input_features
         for enc_seq_a, enc_seq_a in zip(A_			,					A_							):
           self.assertTrue(np.allclose(A_			,					A_			,					atol=1E-3							)							)

         # Test truncation required
         _lowercase:      List[str]   			=  [floats_list((1, x)							)[0] for x in range(200			,					(feature_extractor.n_samples + 500)			,					200							)]
         _lowercase:      List[Any]   			=  [np.asarray(A_							) for speech_input in speech_inputs]

         _lowercase:      Any   			=  [x[: feature_extractor.n_samples] for x in speech_inputs]
         _lowercase:      str   			=  [np.asarray(A_							) for speech_input in speech_inputs_truncated]

         _lowercase:      Optional[Any]   			=  feature_extractor(A_			,					return_tensors='''np'''							).input_features
         _lowercase:      List[str]   			=  feature_extractor(A_			,					return_tensors='''np'''							).input_features
         for enc_seq_a, enc_seq_a in zip(A_			,					A_							):
           self.assertTrue(np.allclose(A_			,					A_			,					atol=1E-3							)							)
       def 				lowercase_				(  self							)  ->						Union[str, Any]:





         """simple docstring"""




         import torch

         _lowercase:      List[str]   			=  self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()							)
         _lowercase:      List[Any]   			=  np.random.rand(100			,					32							).astype(np.floataa							)
         _lowercase:      Tuple   			=  np_speech_inputs.tolist()

         for inputs in [py_speech_inputs, np_speech_inputs]:
           _lowercase:      Optional[Any]   			=  feature_extractor.pad([{'''input_features''': inputs}]			,					return_tensors='''np'''							)
           self.assertTrue(np_processed.input_features.dtype == np.floataa							)
           _lowercase:      int   			=  feature_extractor.pad([{'''input_features''': inputs}]			,					return_tensors='''pt'''							)
           self.assertTrue(pt_processed.input_features.dtype == torch.floataa							)
       def 				lowercase_				(  self			,					A_							)  ->						List[Any]:





         """simple docstring"""




         _lowercase:      Optional[Any]   			=  load_dataset('''hf-internal-testing/librispeech_asr_dummy'''			,					'''clean'''			,					split='''validation'''							)
         # automatic decoding with librispeech
         _lowercase:      List[str]   			=  ds.sort('''id'''							).select(range(A_							)							)[:num_samples]['''audio''']

         return [x["array"] for x in speech_samples]
       def 				lowercase_				(  self							)  ->						Optional[int]:





         """simple docstring"""




         _lowercase:      List[str]   			=  torch.tensor(
             [
                 0.11_93, -0.09_46, -0.10_98, -0.01_96, 0.02_25, -0.06_90, -0.17_36, 0.09_51,
                 0.09_71, -0.08_17, -0.07_02, 0.01_62, 0.02_60, 0.00_17, -0.01_92, -0.16_78,
                 0.07_09, -0.18_67, -0.06_55, -0.02_74, -0.02_34, -0.18_84, -0.05_16, -0.05_54,
                 -0.02_74, -0.14_25, -0.14_23, 0.08_37, 0.03_77, -0.08_54
             ]							)
         # fmt: on

         _lowercase:      Optional[Any]   			=  self._load_datasamples(1							)
         _lowercase:      Optional[int]   			=  WhisperFeatureExtractor()
         _lowercase:      Dict   			=  feature_extractor(A_			,					return_tensors='''pt'''							).input_features
         self.assertEqual(input_features.shape			,					(1, 80, 3000)							)
         self.assertTrue(torch.allclose(input_features[0, 0, :30]			,					A_			,					atol=1E-4							)							)



       def 				lowercase_				(  self							)  ->						Union[str, Any]:





         """simple docstring"""




         _lowercase:      Union[str, Any]   			=  self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()							)
         _lowercase:      List[str]   			=  self._load_datasamples(1							)[0]
         _lowercase:      Union[str, Any]   			=  ((audio - audio.min()) / (audio.max() - audio.min())) * 6_5535  # Rescale to [0, 65535] to show issue
         _lowercase:      Any   			=  feat_extract.zero_mean_unit_var_norm([audio]			,					attention_mask=A_							)[0]

         self.assertTrue(np.all(np.mean(A_							) < 1E-3							)							)
         self.assertTrue(np.all(np.abs(np.var(A_							) - 1							) < 1E-3							)							)
