





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


UpperCAmelCase_   				=  logging.get_logger(__name__)
class       UpperCamelCase_		(	_a       ):
					lowerCAmelCase_				=						["""input_values""", """attention_mask"""]




					def __init__(		self					,    lowerCAmelCase_=1					,    lowerCAmelCase_=1_6000					,    lowerCAmelCase_=128					,    lowerCAmelCase_=1024					,    lowerCAmelCase_=0.0					,    lowerCAmelCase_=True					,    lowerCAmelCase_=-4.2_67_73_93					,    lowerCAmelCase_=4.5_68_99_74					,    lowerCAmelCase_=False					,    **lowerCAmelCase_					,    )		->				Optional[int]:
										super().__init__(feature_size=__UpperCAmelCase					,    sampling_rate=__UpperCAmelCase					,    padding_value=__UpperCAmelCase					,    **__UpperCAmelCase	)
										_snake_case						=					num_mel_bins
										_snake_case						=					max_length
										_snake_case						=					do_normalize
										_snake_case						=					mean
										_snake_case						=					std
										_snake_case						=					return_attention_mask




					def 						lowerCAmelCase      (		self					,    lowerCAmelCase_					,    lowerCAmelCase_					,    )		->				List[str]:
										_snake_case						=					torch.from_numpy(__UpperCAmelCase	).unsqueeze(0	)
										_snake_case						=					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					,    )

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

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

										_snake_case						=					fbank.numpy()

										return fbank




					def 						lowerCAmelCase      (		self					,    lowerCAmelCase_	)		->				Tuple:
										return (input_values - (self.mean)) / (self.std * 2)






					def __call__(		self					,    lowerCAmelCase_					,    lowerCAmelCase_ = None					,    lowerCAmelCase_ = None					,    **lowerCAmelCase_					,    )		->				Optional[int]:

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

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

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

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

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

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

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

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

										if return_tensors is not None:
															_snake_case						=					padded_inputs.convert_to_tensors(__UpperCAmelCase	)

										return padded_inputs







"""simple docstring"""







from ..utils import DummyObject, requires_backends






class 		_A    (    metaclass=_a    ):







    """simple docstring"""




    UpperCAmelCase			:				int      =			["""torch""", """transformers""", """invisible_watermark"""]






    def __init__(   self							:       int							,					*__UpperCAmelCase							:       List[Any]							,					**__UpperCAmelCase							:       Optional[int]):
      requires_backends(self							,					["torch", "transformers", "invisible_watermark"])






    @classmethod
    def       __snake_case				(   cls							:       Optional[Any]							,					*__UpperCAmelCase							:       Any							,					**__UpperCAmelCase							:       int):
      requires_backends(cls							,					["torch", "transformers", "invisible_watermark"])






    @classmethod
    def       __snake_case				(   cls							:       List[Any]							,					*__UpperCAmelCase							:       Optional[Any]							,					**__UpperCAmelCase							:       Tuple):
      requires_backends(cls							,					["torch", "transformers", "invisible_watermark"])






class 		_A    (    metaclass=_a    ):







    """simple docstring"""




    UpperCAmelCase			:				Optional[int]      =			["""torch""", """transformers""", """invisible_watermark"""]






    def __init__(   self							:       int							,					*__UpperCAmelCase							:       Tuple							,					**__UpperCAmelCase							:       Tuple):
      requires_backends(self							,					["torch", "transformers", "invisible_watermark"])






    @classmethod
    def       __snake_case				(   cls							:       int							,					*__UpperCAmelCase							:       str							,					**__UpperCAmelCase							:       Optional[int]):
      requires_backends(cls							,					["torch", "transformers", "invisible_watermark"])






    @classmethod
    def       __snake_case				(   cls							:       Any							,					*__UpperCAmelCase							:       Optional[int]							,					**__UpperCAmelCase							:       Union[str, Any]):
      requires_backends(cls							,					["torch", "transformers", "invisible_watermark"])






