


"""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      =      logging.get_logger(__name__)




class 	__UpperCamelCase    (    UpperCamelCase	):
					lowercase_						:							Optional[int]			=     ["""input_values""", """attention_mask"""]
					def __init__(       self       :  Dict    ,   UpperCAmelCase       :  List[Any]=1    ,   UpperCAmelCase       :  Tuple=1_6000    ,   UpperCAmelCase       :  Optional[int]=128    ,   UpperCAmelCase       :  Any=1024    ,   UpperCAmelCase       :  Tuple=0.0    ,   UpperCAmelCase       :  Dict=True    ,   UpperCAmelCase       :  Tuple=-4.2_6_7_7_3_9_3    ,   UpperCAmelCase       :  str=4.5_6_8_9_9_7_4    ,   UpperCAmelCase       :  List[str]=False    ,   **UpperCAmelCase       :  Dict    ,   )  ->					Tuple:
											super().__init__(feature_size=UpperCAmelCase    ,   sampling_rate=UpperCAmelCase    ,   padding_value=UpperCAmelCase    ,   **UpperCAmelCase      )
											lowerCAmelCase      :Any				  = num_mel_bins
											lowerCAmelCase      :Optional[int]				  = max_length
											lowerCAmelCase      :Dict				  = do_normalize
											lowerCAmelCase      :Any				  = mean
											lowerCAmelCase      :Dict				  = std
											lowerCAmelCase      :Any				  = return_attention_mask
					def 					UpperCAmelCase__   (       self       :  List[str]    ,   UpperCAmelCase       :  np.ndarray    ,   UpperCAmelCase       :  int    ,   )  ->					np.ndarray:
											lowerCAmelCase      :Tuple				  = torch.from_numpy(UpperCAmelCase      ).unsqueeze(0      )
											lowerCAmelCase      :Union[str, Any]				  = ta_kaldi.fbank(
											    UpperCAmelCase    ,   htk_compat=UpperCAmelCase    ,   sample_frequency=self.sampling_rate    ,   use_energy=UpperCAmelCase    ,   window_type='hanning'    ,   num_mel_bins=self.num_mel_bins    ,   dither=0.0    ,   frame_shift=10    ,   )

											lowerCAmelCase      :Tuple				  = fbank.shape[0]
											lowerCAmelCase      :str				  = max_length - n_frames

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

											lowerCAmelCase      :Any				  = fbank.numpy()

											return fbank
					def 					UpperCAmelCase__   (       self       :  Dict    ,   UpperCAmelCase       :  np.ndarray      )  ->					np.ndarray:
											return (input_values - (self.mean)) / (self.std * 2)
					def __call__(       self       :  Dict    ,   UpperCAmelCase       :  Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]    ,   UpperCAmelCase       :  Optional[int] = None    ,   UpperCAmelCase       :  Optional[Union[str, TensorType]] = None    ,   **UpperCAmelCase       :  int    ,   )  ->					BatchFeature:

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

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

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

											# always return batch
											if not is_batched:
																	lowerCAmelCase      :Optional[int]				  = [raw_speech]

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

											# convert into BatchFeature
											lowerCAmelCase      :Tuple				  = BatchFeature({'input_values': features}      )

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

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

											if return_tensors is not None:
																	lowerCAmelCase      :Optional[int]				  = padded_inputs.convert_to_tensors(UpperCAmelCase      )

											return padded_inputs







"""simple docstring"""


import unittest
import warnings
from dataclasses import dataclass

from transformers.convert_slow_tokenizer import SpmConverter
from transformers.testing_utils import get_tests_dir




@dataclass
class 	__UpperCamelCase    :
					lowercase_						:							str




class 	__UpperCamelCase    (    unittest.TestCase	):
					def 					UpperCAmelCase__   (       self       :  Union[str, Any]      )  ->					Optional[Any]:
											lowerCAmelCase      :List[str]				  = get_tests_dir('fixtures/test_sentencepiece.model'      )
											lowerCAmelCase      :Tuple				  = get_tests_dir('fixtures/test_sentencepiece_with_bytefallback.model'      )

											lowerCAmelCase      :List[str]				  = FakeOriginalTokenizer(vocab_file=UpperCAmelCase      )

											with warnings.catch_warnings(record=UpperCAmelCase      ) as w:
																	lowerCAmelCase      :Dict				  = SpmConverter(UpperCAmelCase      )
											self.assertEqual(len(UpperCAmelCase      )    ,   0      )

											lowerCAmelCase      :int				  = FakeOriginalTokenizer(vocab_file=UpperCAmelCase      )

											with warnings.catch_warnings(record=UpperCAmelCase      ) as w:
																	lowerCAmelCase      :List[str]				  = SpmConverter(UpperCAmelCase      )
											self.assertEqual(len(UpperCAmelCase      )    ,   1      )

											self.assertIn(
											    'The sentencepiece tokenizer that you are converting to a fast tokenizer uses the byte fallback option'
											    ' which is not implemented in the fast tokenizers.'    ,   str(w[0].message      )    ,   )