




import json
import os
from pathlib import Path
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple, Union

import sentencepiece

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


__A        =    logging.get_logger(__name__)

__A        =    "▁"

__A        =    {
    "vocab_file": "vocab.json",
    "spm_file": "sentencepiece.bpe.model",
}

__A        =    {
    "vocab_file": {
        "facebook/s2t-small-librispeech-asr": (
            "https://huggingface.co/facebook/s2t-small-librispeech-asr/resolve/main/vocab.json"
        ),
    },
    "spm_file": {
        "facebook/s2t-small-librispeech-asr": (
            "https://huggingface.co/facebook/s2t-small-librispeech-asr/resolve/main/sentencepiece.bpe.model"
        )
    },
}

__A        =    {
    "facebook/s2t-small-librispeech-asr": 10_24,
}

__A        =    ["pt", "fr", "ru", "nl", "ro", "it", "es", "de"]

__A        =    {"mustc": MUSTC_LANGS}







class     __lowerCAmelCase (       __magic_name__ ):

			"""simple docstring"""







			snake_case_   		=       VOCAB_FILES_NAMES
			snake_case_   		=       PRETRAINED_VOCAB_FILES_MAP
			snake_case_   		=       MAX_MODEL_INPUT_SIZES
			snake_case_   		=       ['''input_ids''', '''attention_mask''']

			snake_case_   		=       []


			def __init__(      self			,						lowerCamelCase__			,						lowerCamelCase__			,						lowerCamelCase__="<s>"			,						lowerCamelCase__="</s>"			,						lowerCamelCase__="<pad>"			,						lowerCamelCase__="<unk>"			,						lowerCamelCase__=False			,						lowerCamelCase__=False			,						lowerCamelCase__=None			,						lowerCamelCase__=None			,						lowerCamelCase__ = None			,						**lowerCamelCase__			,						)    ->      None:



							'''simple docstring'''
							__lowerCamelCase		       =    {} if sp_model_kwargs is None else sp_model_kwargs

							super().__init__(
							    bos_token=lowerCamelCase__			,						eos_token=lowerCamelCase__			,						unk_token=lowerCamelCase__			,						pad_token=lowerCamelCase__			,						do_upper_case=lowerCamelCase__			,						do_lower_case=lowerCamelCase__			,						tgt_lang=lowerCamelCase__			,						lang_codes=lowerCamelCase__			,						sp_model_kwargs=self.sp_model_kwargs			,						**lowerCamelCase__			,						)
							__lowerCamelCase		       =    do_upper_case
							__lowerCamelCase		       =    do_lower_case

							__lowerCamelCase		       =    load_json(lowerCamelCase__							)
							__lowerCamelCase		       =    {v: k for k, v in self.encoder.items()}
							__lowerCamelCase		       =    spm_file
							__lowerCamelCase		       =    load_spm(lowerCamelCase__			,						self.sp_model_kwargs							)

							if lang_codes is not None:
											__lowerCamelCase		       =    lang_codes
											__lowerCamelCase		       =    LANGUAGES[lang_codes]
											__lowerCamelCase		       =    [f"""<lang:{lang}>""" for lang in self.langs]
											__lowerCamelCase		       =    {lang: self.sp_model.PieceToId(f"""<lang:{lang}>"""							) for lang in self.langs}

											__lowerCamelCase		       =    self.lang_tokens
											__lowerCamelCase		       =    tgt_lang if tgt_lang is not None else self.langs[0]

											self.set_tgt_lang_special_tokens(self._tgt_lang							)
							else:
											__lowerCamelCase		       =    {}


			@property
			def       lowercase_    (      self							)    ->      int:



							'''simple docstring'''
							return len(self.encoder							)


			@property
			def       lowercase_    (      self							)    ->      str:



							'''simple docstring'''
							return self._tgt_lang


			@tgt_lang.setter
			def       lowercase_    (      self			,						lowerCamelCase__							)    ->      None:



							'''simple docstring'''
							__lowerCamelCase		       =    new_tgt_lang
							self.set_tgt_lang_special_tokens(lowerCamelCase__							)


			def       lowercase_    (      self			,						lowerCamelCase__							)    ->      None:



							'''simple docstring'''
							__lowerCamelCase		       =    self.lang_code_to_id[tgt_lang]
							__lowerCamelCase		       =    [lang_code_id]


			def       lowercase_    (      self			,						lowerCamelCase__							)    ->      List[str]:



							'''simple docstring'''
							return self.sp_model.encode(lowerCamelCase__			,						out_type=lowerCamelCase__							)


			def       lowercase_    (      self			,						lowerCamelCase__							)    ->      Tuple:



							'''simple docstring'''
							return self.encoder.get(lowerCamelCase__			,						self.encoder[self.unk_token]							)


			def       lowercase_    (      self			,						lowerCamelCase__							)    ->      str:



							'''simple docstring'''
							return self.decoder.get(lowerCamelCase__			,						self.unk_token							)


			def       lowercase_    (      self			,						lowerCamelCase__							)    ->      str:



							'''simple docstring'''
							__lowerCamelCase		       =    []
							__lowerCamelCase		       =    ''
							for token in tokens:
											# make sure that special tokens are not decoded using sentencepiece model
											if token in self.all_special_tokens:
															__lowerCamelCase		       =    self.sp_model.decode(lowerCamelCase__							)
															out_string += (decoded.upper() if self.do_upper_case else decoded) + token + " "
															__lowerCamelCase		       =    []
											else:
															current_sub_tokens.append(lowerCamelCase__							)
							__lowerCamelCase		       =    self.sp_model.decode(lowerCamelCase__							)
							out_string += decoded.upper() if self.do_upper_case else decoded
							return out_string.strip()


			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__=None							)    ->      List[int]:



							'''simple docstring'''
							if token_ids_a is None:
											return self.prefix_tokens + token_ids_a + [self.eos_token_id]
							# We don't expect to process pairs, but leave the pair logic for API consistency
							return self.prefix_tokens + token_ids_a + token_ids_a + [self.eos_token_id]


			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__ = None			,						lowerCamelCase__ = False							)    ->      List[int]:



							'''simple docstring'''

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

							__lowerCamelCase		       =    [1] * len(self.prefix_tokens							)
							__lowerCamelCase		       =    [1]
							if token_ids_a is None:
											return prefix_ones + ([0] * len(lowerCamelCase__							)) + suffix_ones
							return prefix_ones + ([0] * len(lowerCamelCase__							)) + ([0] * len(lowerCamelCase__							)) + suffix_ones


			def       lowercase_    (      self							)    ->      Dict:



							'''simple docstring'''
							__lowerCamelCase		       =    self.encoder.copy()
							vocab.update(self.added_tokens_encoder							)
							return vocab


			def __getstate__(      self							)    ->      Dict:



							'''simple docstring'''
							__lowerCamelCase		       =    self.__dict__.copy()
							__lowerCamelCase		       =    None
							return state


			def __setstate__(      self			,						lowerCamelCase__							)    ->      None:



							'''simple docstring'''
							__lowerCamelCase		       =    d

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

							__lowerCamelCase		       =    load_spm(self.spm_file			,						self.sp_model_kwargs							)






			def       lowercase_    (      self			,						lowerCamelCase__			,						lowerCamelCase__ = None							)    ->      Tuple[str]:



							'''simple docstring'''
							__lowerCamelCase		       =    Path(lowerCamelCase__							)
							assert save_dir.is_dir(), f"""{save_directory} should be a directory"""
							__lowerCamelCase		       =    save_dir / (
							    (filename_prefix + '-' if filename_prefix else '') + self.vocab_files_names['vocab_file']
							)
							__lowerCamelCase		       =    save_dir / (
							    (filename_prefix + '-' if filename_prefix else '') + self.vocab_files_names['spm_file']
							)

							save_json(self.encoder			,						lowerCamelCase__							)

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

							return (str(lowerCamelCase__							), str(lowerCamelCase__							))







def 			lowerCamelCase_		(       UpperCamelCase__				:      str ,     UpperCamelCase__				:      Dict[str, Any]      )     ->       sentencepiece.SentencePieceProcessor:





				"""simple docstring"""


				__lowerCamelCase		       =    sentencepiece.SentencePieceProcessor(**UpperCamelCase__      )
				spm.Load(str(UpperCamelCase__      )      )
				return spm







def 			lowerCamelCase_		(       UpperCamelCase__				:      str      )     ->       Union[Dict, List]:





				"""simple docstring"""


				with open(UpperCamelCase__ ,     'r'      ) as f:
								return json.load(UpperCamelCase__      )







def 			lowerCamelCase_		(       UpperCamelCase__				:      Tuple ,     UpperCamelCase__				:      str      )     ->       None:





				"""simple docstring"""


				with open(UpperCamelCase__ ,     'w'      ) as f:
								json.dump(UpperCamelCase__ ,     UpperCamelCase__ ,     indent=2      )








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        =    random.Random()







def 			lowerCamelCase_		(       UpperCamelCase__				:      Optional[int] ,     UpperCamelCase__				:      List[str]=1.0 ,     UpperCamelCase__				:      Union[str, Any]=None ,     UpperCamelCase__				:      Optional[int]=None      )     ->       Optional[Any]:





				"""simple docstring"""


				if rng is None:
								__lowerCamelCase		       =    global_rng

				__lowerCamelCase		       =    []
				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     __lowerCAmelCase (       unittest.TestCase ):

			"""simple docstring"""







			def __init__(      self			,						lowerCamelCase__			,						lowerCamelCase__=7			,						lowerCamelCase__=400			,						lowerCamelCase__=2_000			,						lowerCamelCase__=10			,						lowerCamelCase__=160			,						lowerCamelCase__=8			,						lowerCamelCase__=0.0			,						lowerCamelCase__=4_000			,						lowerCamelCase__=False			,						lowerCamelCase__=True			,						)    ->      List[str]:



							'''simple docstring'''
							__lowerCamelCase		       =    parent
							__lowerCamelCase		       =    batch_size
							__lowerCamelCase		       =    min_seq_length
							__lowerCamelCase		       =    max_seq_length
							__lowerCamelCase		       =    (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
							__lowerCamelCase		       =    padding_value
							__lowerCamelCase		       =    sampling_rate
							__lowerCamelCase		       =    return_attention_mask
							__lowerCamelCase		       =    do_normalize
							__lowerCamelCase		       =    feature_size
							__lowerCamelCase		       =    chunk_length
							__lowerCamelCase		       =    hop_length


			def       lowercase_    (      self							)    ->      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			,						lowerCamelCase__=False			,						lowerCamelCase__=False							)    ->      Optional[int]:



							'''simple docstring'''
							def _flatten(lowerCamelCase__							):
											return list(itertools.chain(*lowerCamelCase__							)							)

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







@require_torch
@require_torchaudio
class     __lowerCAmelCase (       __magic_name__			,				unittest.TestCase ):

			"""simple docstring"""







			snake_case_   		=       WhisperFeatureExtractor if is_speech_available() else None


			def       lowercase_    (      self							)    ->      Any:



							'''simple docstring'''
							__lowerCamelCase		       =    WhisperFeatureExtractionTester(self							)


			def       lowercase_    (      self							)    ->      List[str]:



							'''simple docstring'''
							__lowerCamelCase		       =    self.feature_extraction_class(**self.feat_extract_dict							)

							with tempfile.TemporaryDirectory() as tmpdirname:
											__lowerCamelCase		       =    feat_extract_first.save_pretrained(lowerCamelCase__							)[0]
											check_json_file_has_correct_format(lowerCamelCase__							)
											__lowerCamelCase		       =    self.feature_extraction_class.from_pretrained(lowerCamelCase__							)

							__lowerCamelCase		       =    feat_extract_first.to_dict()
							__lowerCamelCase		       =    feat_extract_second.to_dict()
							__lowerCamelCase		       =    feat_extract_first.mel_filters
							__lowerCamelCase		       =    feat_extract_second.mel_filters
							self.assertTrue(np.allclose(lowerCamelCase__			,						lowerCamelCase__							)							)
							self.assertEqual(lowerCamelCase__			,						lowerCamelCase__							)


			def       lowercase_    (      self							)    ->      Dict:



							'''simple docstring'''
							__lowerCamelCase		       =    self.feature_extraction_class(**self.feat_extract_dict							)

							with tempfile.TemporaryDirectory() as tmpdirname:
											__lowerCamelCase		       =    os.path.join(lowerCamelCase__			,						'feat_extract.json'							)
											feat_extract_first.to_json_file(lowerCamelCase__							)
											__lowerCamelCase		       =    self.feature_extraction_class.from_json_file(lowerCamelCase__							)

							__lowerCamelCase		       =    feat_extract_first.to_dict()
							__lowerCamelCase		       =    feat_extract_second.to_dict()
							__lowerCamelCase		       =    feat_extract_first.mel_filters
							__lowerCamelCase		       =    feat_extract_second.mel_filters
							self.assertTrue(np.allclose(lowerCamelCase__			,						lowerCamelCase__							)							)
							self.assertEqual(lowerCamelCase__			,						lowerCamelCase__							)


			def       lowercase_    (      self							)    ->      Dict:



							'''simple docstring'''
							# Tests that all call wrap to encode_plus and batch_encode_plus
							__lowerCamelCase		       =    self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()							)
							# create three inputs of length 800, 1000, and 1200
							__lowerCamelCase		       =    [floats_list((1, x)							)[0] for x in range(800			,						1_400			,						200							)]
							__lowerCamelCase		       =    [np.asarray(lowerCamelCase__							) for speech_input in speech_inputs]

							# Test feature size
							__lowerCamelCase		       =    feature_extractor(lowerCamelCase__			,						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
							__lowerCamelCase		       =    feature_extractor(speech_inputs[0]			,						return_tensors='np'							).input_features
							__lowerCamelCase		       =    feature_extractor(np_speech_inputs[0]			,						return_tensors='np'							).input_features
							self.assertTrue(np.allclose(lowerCamelCase__			,						lowerCamelCase__			,						atol=1e-3							)							)

							# Test batched
							__lowerCamelCase		       =    feature_extractor(lowerCamelCase__			,						return_tensors='np'							).input_features
							__lowerCamelCase		       =    feature_extractor(lowerCamelCase__			,						return_tensors='np'							).input_features
							for enc_seq_a, enc_seq_a in zip(lowerCamelCase__			,						lowerCamelCase__							):
											self.assertTrue(np.allclose(lowerCamelCase__			,						lowerCamelCase__			,						atol=1e-3							)							)

							# Test 2-D numpy arrays are batched.
							__lowerCamelCase		       =    [floats_list((1, x)							)[0] for x in (800, 800, 800)]
							__lowerCamelCase		       =    np.asarray(lowerCamelCase__							)
							__lowerCamelCase		       =    feature_extractor(lowerCamelCase__			,						return_tensors='np'							).input_features
							__lowerCamelCase		       =    feature_extractor(lowerCamelCase__			,						return_tensors='np'							).input_features
							for enc_seq_a, enc_seq_a in zip(lowerCamelCase__			,						lowerCamelCase__							):
											self.assertTrue(np.allclose(lowerCamelCase__			,						lowerCamelCase__			,						atol=1e-3							)							)

							# Test truncation required
							__lowerCamelCase		       =    [floats_list((1, x)							)[0] for x in range(200			,						(feature_extractor.n_samples + 500)			,						200							)]
							__lowerCamelCase		       =    [np.asarray(lowerCamelCase__							) for speech_input in speech_inputs]

							__lowerCamelCase		       =    [x[: feature_extractor.n_samples] for x in speech_inputs]
							__lowerCamelCase		       =    [np.asarray(lowerCamelCase__							) for speech_input in speech_inputs_truncated]

							__lowerCamelCase		       =    feature_extractor(lowerCamelCase__			,						return_tensors='np'							).input_features
							__lowerCamelCase		       =    feature_extractor(lowerCamelCase__			,						return_tensors='np'							).input_features
							for enc_seq_a, enc_seq_a in zip(lowerCamelCase__			,						lowerCamelCase__							):
											self.assertTrue(np.allclose(lowerCamelCase__			,						lowerCamelCase__			,						atol=1e-3							)							)


			def       lowercase_    (      self							)    ->      List[str]:



							'''simple docstring'''
							import torch

							__lowerCamelCase		       =    self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()							)
							__lowerCamelCase		       =    np.random.rand(100			,						32							).astype(np.floataa							)
							__lowerCamelCase		       =    np_speech_inputs.tolist()

							for inputs in [py_speech_inputs, np_speech_inputs]:
											__lowerCamelCase		       =    feature_extractor.pad([{'input_features': inputs}]			,						return_tensors='np'							)
											self.assertTrue(np_processed.input_features.dtype == np.floataa							)
											__lowerCamelCase		       =    feature_extractor.pad([{'input_features': inputs}]			,						return_tensors='pt'							)
											self.assertTrue(pt_processed.input_features.dtype == torch.floataa							)


			def       lowercase_    (      self			,						lowerCamelCase__							)    ->      List[str]:



							'''simple docstring'''
							__lowerCamelCase		       =    load_dataset('hf-internal-testing/librispeech_asr_dummy'			,						'clean'			,						split='validation'							)
							# automatic decoding with librispeech
							__lowerCamelCase		       =    ds.sort('id'							).select(range(lowerCamelCase__							)							)[:num_samples]['audio']

							return [x["array"] for x in speech_samples]


			def       lowercase_    (      self							)    ->      Tuple:



							'''simple docstring'''
							# fmt: off
							__lowerCamelCase		       =    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

							__lowerCamelCase		       =    self._load_datasamples(1							)
							__lowerCamelCase		       =    WhisperFeatureExtractor()
							__lowerCamelCase		       =    feature_extractor(lowerCamelCase__			,						return_tensors='pt'							).input_features
							self.assertEqual(input_features.shape			,						(1, 80, 3_000)							)
							self.assertTrue(torch.allclose(input_features[0, 0, :30]			,						lowerCamelCase__			,						atol=1e-4							)							)






			def       lowercase_    (      self							)    ->      List[Any]:



							'''simple docstring'''
							__lowerCamelCase		       =    self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()							)
							__lowerCamelCase		       =    self._load_datasamples(1							)[0]
							__lowerCamelCase		       =    ((audio - audio.min()) / (audio.max() - audio.min())) * 65_535  # Rescale to [0, 65535] to show issue
							__lowerCamelCase		       =    feat_extract.zero_mean_unit_var_norm([audio]			,						attention_mask=lowerCamelCase__							)[0]

							self.assertTrue(np.all(np.mean(lowerCamelCase__							) < 1e-3							)							)
							self.assertTrue(np.all(np.abs(np.var(lowerCamelCase__							) - 1							) < 1e-3							)							)


