


'''simple docstring'''
from typing import List, Optional, Union

import numpy as np
import torch
import torchaudio.compliance.kaldi as ta_kaldi

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


__SCREAMING_SNAKE_CASE				:							Optional[Any]		       =		logging.get_logger(__name__)






class 				__lowerCAmelCase      (      lowercase	):

						"""simple docstring"""






						_UpperCAmelCase		:					str      	=["input_values", "attention_mask"]






						def __init__(     self       :			Dict   ,     lowerCAmelCase       :			Optional[int]=1   ,     lowerCAmelCase       :			Optional[Any]=1_60_00   ,     lowerCAmelCase       :			Union[str, Any]=1_28   ,     lowerCAmelCase       :			List[Any]=10_24   ,     lowerCAmelCase       :			Tuple=0.0   ,     lowerCAmelCase       :			List[Any]=True   ,     lowerCAmelCase       :			Optional[int]=-4.2_6_7_7_3_9_3   ,     lowerCAmelCase       :			List[str]=4.5_6_8_9_9_7_4   ,     lowerCAmelCase       :			Optional[Any]=False   ,     **lowerCAmelCase       :			Dict   ,     ):
													super().__init__(feature_size=lowerCAmelCase   ,     sampling_rate=lowerCAmelCase   ,     padding_value=lowerCAmelCase   ,     **lowerCAmelCase  )
													A_ 						=		num_mel_bins
													A_ 						=		max_length
													A_ 						=		do_normalize
													A_ 						=		mean
													A_ 						=		std
													A_ 						=		return_attention_mask






						def 							_UpperCAmelCase					(     self       :			Union[str, Any]   ,     lowerCAmelCase       :			np.ndarray   ,     lowerCAmelCase       :			int   ,     ):
													A_ 						=		torch.from_numpy(lowerCAmelCase  ).unsqueeze(0  )
													A_ 						=		ta_kaldi.fbank(
													    lowerCAmelCase   ,     htk_compat=lowerCAmelCase   ,     sample_frequency=self.sampling_rate   ,     use_energy=lowerCAmelCase   ,     window_type="hanning"   ,     num_mel_bins=self.num_mel_bins   ,     dither=0.0   ,     frame_shift=10   ,     )

													A_ 						=		fbank.shape[0]
													A_ 						=		max_length - n_frames

													# pad or truncate, depending on difference
													if difference > 0:
																				A_ 						=		torch.nn.ZeroPadad((0, 0, 0, difference)  )
																				A_ 						=		pad_module(lowerCAmelCase  )
													elif difference < 0:
																				A_ 						=		fbank[0:max_length, :]

													A_ 						=		fbank.numpy()

													return fbank






						def 							_UpperCAmelCase					(     self       :			Any   ,     lowerCAmelCase       :			np.ndarray  ):
													return (input_values - (self.mean)) / (self.std * 2)






						def __call__(     self       :			List[Any]   ,     lowerCAmelCase       :			Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]   ,     lowerCAmelCase       :			Optional[int] = None   ,     lowerCAmelCase       :			Optional[Union[str, TensorType]] = None   ,     **lowerCAmelCase       :			List[Any]   ,     ):

													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."  )

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

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

													# always return batch
													if not is_batched:
																				A_ 						=		[raw_speech]

													# extract fbank features and pad/truncate to max_length
													A_ 						=		[self._extract_fbank_features(lowerCAmelCase   ,     max_length=self.max_length  ) for waveform in raw_speech]

													# convert into BatchFeature
													A_ 						=		BatchFeature({"input_values": features}  )

													# make sure list is in array format
													A_ 						=		padded_inputs.get("input_values"  )
													if isinstance(input_values[0]   ,     lowerCAmelCase  ):
																				A_ 						=		[np.asarray(lowerCAmelCase   ,     dtype=np.floataa  ) for feature in input_values]

													# normalization
													if self.do_normalize:
																				A_ 						=		[self.normalize(lowerCAmelCase  ) for feature in input_values]

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

													return padded_inputs









'''simple docstring'''
import shutil
import tempfile
import unittest
from pathlib import Path
from shutil import copyfile

from transformers import SpeechaTextTokenizer, is_speech_available
from transformers.models.speech_to_text.tokenization_speech_to_text import VOCAB_FILES_NAMES, save_json
from transformers.testing_utils import get_tests_dir, require_sentencepiece, require_torch, require_torchaudio
from transformers.utils import FEATURE_EXTRACTOR_NAME

from .test_feature_extraction_speech_to_text import floats_list


if is_speech_available():
							from transformers import SpeechaTextFeatureExtractor, SpeechaTextProcessor


__SCREAMING_SNAKE_CASE				:							int		       =		get_tests_dir('''fixtures/test_sentencepiece.model''')






@require_torch
@require_torchaudio
@require_sentencepiece
class 				__lowerCAmelCase      (      unittest.TestCase	):

						"""simple docstring"""






						def 							_UpperCAmelCase					(     self       :			List[str]  ):
													A_ 						=		tempfile.mkdtemp()

													A_ 						=		["<s>", "<pad>", "</s>", "<unk>", "▁This", "▁is", "▁a", "▁t", "est"]
													A_ 						=		dict(zip(lowerCAmelCase   ,     range(len(lowerCAmelCase  )  )  )  )
													A_ 						=		Path(self.tmpdirname  )
													save_json(lowerCAmelCase   ,     save_dir / VOCAB_FILES_NAMES["vocab_file"]  )
													if not (save_dir / VOCAB_FILES_NAMES["spm_file"]).exists():
																				copyfile(lowerCAmelCase   ,     save_dir / VOCAB_FILES_NAMES["spm_file"]  )

													A_ 						=		SpeechaTextTokenizer.from_pretrained(self.tmpdirname  )
													tokenizer.save_pretrained(self.tmpdirname  )

													A_ 						=		{
													    "feature_size": 24,
													    "num_mel_bins": 24,
													    "padding_value": 0.0,
													    "sampling_rate": 1_60_00,
													    "return_attention_mask": False,
													    "do_normalize": True,
													}
													save_json(lowerCAmelCase   ,     save_dir / FEATURE_EXTRACTOR_NAME  )






						def 							_UpperCAmelCase					(     self       :			Union[str, Any]   ,     **lowerCAmelCase       :			str  ):
													return SpeechaTextTokenizer.from_pretrained(self.tmpdirname   ,     **lowerCAmelCase  )






						def 							_UpperCAmelCase					(     self       :			List[Any]   ,     **lowerCAmelCase       :			Optional[Any]  ):
													return SpeechaTextFeatureExtractor.from_pretrained(self.tmpdirname   ,     **lowerCAmelCase  )






						def 							_UpperCAmelCase					(     self       :			Any  ):
													shutil.rmtree(self.tmpdirname  )






						def 							_UpperCAmelCase					(     self       :			Union[str, Any]  ):
													A_ 						=		self.get_tokenizer()
													A_ 						=		self.get_feature_extractor()

													A_ 						=		SpeechaTextProcessor(tokenizer=lowerCAmelCase   ,     feature_extractor=lowerCAmelCase  )

													processor.save_pretrained(self.tmpdirname  )
													A_ 						=		SpeechaTextProcessor.from_pretrained(self.tmpdirname  )

													self.assertEqual(processor.tokenizer.get_vocab()   ,     tokenizer.get_vocab()  )
													self.assertIsInstance(processor.tokenizer   ,     lowerCAmelCase  )

													self.assertEqual(processor.feature_extractor.to_json_string()   ,     feature_extractor.to_json_string()  )
													self.assertIsInstance(processor.feature_extractor   ,     lowerCAmelCase  )






						def 							_UpperCAmelCase					(     self       :			Dict  ):
													A_ 						=		SpeechaTextProcessor(
													    tokenizer=self.get_tokenizer()   ,     feature_extractor=self.get_feature_extractor()  )
													processor.save_pretrained(self.tmpdirname  )

													A_ 						=		self.get_tokenizer(bos_token="(BOS)"   ,     eos_token="(EOS)"  )
													A_ 						=		self.get_feature_extractor(do_normalize=lowerCAmelCase   ,     padding_value=1.0  )

													A_ 						=		SpeechaTextProcessor.from_pretrained(
													    self.tmpdirname   ,     bos_token="(BOS)"   ,     eos_token="(EOS)"   ,     do_normalize=lowerCAmelCase   ,     padding_value=1.0  )

													self.assertEqual(processor.tokenizer.get_vocab()   ,     tokenizer_add_kwargs.get_vocab()  )
													self.assertIsInstance(processor.tokenizer   ,     lowerCAmelCase  )

													self.assertEqual(processor.feature_extractor.to_json_string()   ,     feature_extractor_add_kwargs.to_json_string()  )
													self.assertIsInstance(processor.feature_extractor   ,     lowerCAmelCase  )






						def 							_UpperCAmelCase					(     self       :			str  ):
													A_ 						=		self.get_feature_extractor()
													A_ 						=		self.get_tokenizer()

													A_ 						=		SpeechaTextProcessor(tokenizer=lowerCAmelCase   ,     feature_extractor=lowerCAmelCase  )

													A_ 						=		floats_list((3, 10_00)  )

													A_ 						=		feature_extractor(lowerCAmelCase   ,     return_tensors="np"  )
													A_ 						=		processor(lowerCAmelCase   ,     return_tensors="np"  )

													for key in input_feat_extract.keys():
																				self.assertAlmostEqual(input_feat_extract[key].sum()   ,     input_processor[key].sum()   ,     delta=1e-2  )






						def 							_UpperCAmelCase					(     self       :			Optional[int]  ):
													A_ 						=		self.get_feature_extractor()
													A_ 						=		self.get_tokenizer()

													A_ 						=		SpeechaTextProcessor(tokenizer=lowerCAmelCase   ,     feature_extractor=lowerCAmelCase  )

													A_ 						=		"This is a test string"

													A_ 						=		processor(text=lowerCAmelCase  )

													A_ 						=		tokenizer(lowerCAmelCase  )

													for key in encoded_tok.keys():
																				self.assertListEqual(encoded_tok[key]   ,     encoded_processor[key]  )






						def 							_UpperCAmelCase					(     self       :			int  ):
													A_ 						=		self.get_feature_extractor()
													A_ 						=		self.get_tokenizer()

													A_ 						=		SpeechaTextProcessor(tokenizer=lowerCAmelCase   ,     feature_extractor=lowerCAmelCase  )

													A_ 						=		[[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

													A_ 						=		processor.batch_decode(lowerCAmelCase  )
													A_ 						=		tokenizer.batch_decode(lowerCAmelCase  )

													self.assertListEqual(lowerCAmelCase   ,     lowerCAmelCase  )






						def 							_UpperCAmelCase					(     self       :			Tuple  ):
													A_ 						=		self.get_feature_extractor()
													A_ 						=		self.get_tokenizer()

													A_ 						=		SpeechaTextProcessor(tokenizer=lowerCAmelCase   ,     feature_extractor=lowerCAmelCase  )

													self.assertListEqual(
													    processor.model_input_names   ,     feature_extractor.model_input_names   ,     msg="`processor` and `feature_extractor` model input names do not match"   ,     )





