


'''simple docstring'''



import copy
from typing import Any, Dict, List, Optional, Union

import numpy as np

from ...audio_utils import mel_filter_bank, spectrogram, window_function
from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import TensorType, logging


__A	=logging.get_logger(__name__)


class 		_snake_case					(     a__	):
					lowerCAmelCase      :Any   =					['''input_features''']




					def __init__(      self							,							_lowerCamelCase=80							,							_lowerCamelCase=1_6000							,							_lowerCamelCase=160							,							_lowerCamelCase=30							,							_lowerCamelCase=400							,							_lowerCamelCase=0.0							,							_lowerCamelCase=False							,							**_lowerCamelCase							,							):
								super().__init__(
								    feature_size=_lowerCamelCase							,							sampling_rate=_lowerCamelCase							,							padding_value=_lowerCamelCase							,							return_attention_mask=_lowerCamelCase							,							**_lowerCamelCase							,							)
								UpperCAmelCase__  :					Optional[Any]        =     n_fft
								UpperCAmelCase__  :					List[str]        =     hop_length
								UpperCAmelCase__  :					Any        =     chunk_length
								UpperCAmelCase__  :					str        =     chunk_length * sampling_rate
								UpperCAmelCase__  :					Optional[Any]        =     self.n_samples // hop_length
								UpperCAmelCase__  :					Union[str, Any]        =     sampling_rate
								UpperCAmelCase__  :					Dict        =     mel_filter_bank(
								    num_frequency_bins=1 + n_fft // 2							,							num_mel_filters=_lowerCamelCase							,							min_frequency=0.0							,							max_frequency=8000.0							,							sampling_rate=_lowerCamelCase							,							norm="""slaney"""							,							mel_scale="""slaney"""							,							)




					def 				snake_case__						(      self							,							_lowerCamelCase):
								UpperCAmelCase__  :					int        =     spectrogram(
								    _lowerCamelCase							,							window_function(self.n_fft							,							"""hann""")							,							frame_length=self.n_fft							,							hop_length=self.hop_length							,							power=2.0							,							mel_filters=self.mel_filters							,							log_mel="""log10"""							,							)
								UpperCAmelCase__  :					Optional[Any]        =     log_spec[:, :-1]
								UpperCAmelCase__  :					Dict        =     np.maximum(_lowerCamelCase							,							log_spec.max() - 8.0)
								UpperCAmelCase__  :					Optional[Any]        =     (log_spec + 4.0) / 4.0
								return log_spec




					@staticmethod
					# Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
					def 				snake_case__						(      _lowerCamelCase							,							_lowerCamelCase							,							_lowerCamelCase = 0.0):
								if attention_mask is not None:
											UpperCAmelCase__  :					Tuple        =     np.array(_lowerCamelCase							,							np.intaa)
											UpperCAmelCase__  :					Optional[Any]        =     []

											for vector, length in zip(_lowerCamelCase							,							attention_mask.sum(-1)):
														UpperCAmelCase__  :					Dict        =     (vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1e-7)
														if length < normed_slice.shape[0]:
																	UpperCAmelCase__  :					Any        =     padding_value

														normed_input_values.append(_lowerCamelCase)
								else:
											UpperCAmelCase__  :					List[Any]        =     [(x - x.mean()) / np.sqrt(x.var() + 1e-7) for x in input_values]

								return normed_input_values




					def __call__(      self							,							_lowerCamelCase							,							_lowerCamelCase = True							,							_lowerCamelCase = None							,							_lowerCamelCase = None							,							_lowerCamelCase = None							,							_lowerCamelCase = "max_length"							,							_lowerCamelCase = None							,							_lowerCamelCase = None							,							_lowerCamelCase = None							,							**_lowerCamelCase							,							):

								if sampling_rate is not None:
											if sampling_rate != self.sampling_rate:
														raise ValueError(
														    f'''The model corresponding to this feature extractor: {self.__class__.__name__} was trained using a'''
														    f''' sampling rate of {self.sampling_rate}. Please make sure that the provided `raw_speech` input'''
														    f''' was sampled with {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.""")

								UpperCAmelCase__  :					Optional[Any]        =     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}''')
								UpperCAmelCase__  :					List[str]        =     is_batched_numpy or (
								    isinstance(_lowerCamelCase							,							(list, tuple)) and (isinstance(raw_speech[0]							,							(np.ndarray, tuple, list)))
								)

								if is_batched:
											UpperCAmelCase__  :					Optional[int]        =     [np.asarray([speech]							,							dtype=np.floataa).T for speech in raw_speech]
								elif not is_batched and not isinstance(_lowerCamelCase							,							np.ndarray):
											UpperCAmelCase__  :					Tuple        =     np.asarray(_lowerCamelCase							,							dtype=np.floataa)
								elif isinstance(_lowerCamelCase							,							np.ndarray) and raw_speech.dtype is np.dtype(np.floataa):
											UpperCAmelCase__  :					int        =     raw_speech.astype(np.floataa)

								# always return batch
								if not is_batched:
											UpperCAmelCase__  :					str        =     [np.asarray([raw_speech]).T]

								UpperCAmelCase__  :					Any        =     BatchFeature({"""input_features""": raw_speech})

								# convert into correct format for padding

								UpperCAmelCase__  :					Tuple        =     self.pad(
								    _lowerCamelCase							,							padding=_lowerCamelCase							,							max_length=max_length if max_length else self.n_samples							,							truncation=_lowerCamelCase							,							pad_to_multiple_of=_lowerCamelCase							,							return_attention_mask=return_attention_mask or do_normalize							,							)

								# zero-mean and unit-variance normalization
								if do_normalize:
											UpperCAmelCase__  :					Optional[Any]        =     self.zero_mean_unit_var_norm(
											    padded_inputs["""input_features"""]							,							attention_mask=padded_inputs["""attention_mask"""]							,							padding_value=self.padding_value							,							)
											UpperCAmelCase__  :					Tuple        =     np.stack(padded_inputs["""input_features"""]							,							axis=0)

								# make sure list is in array format
								UpperCAmelCase__  :					str        =     padded_inputs.get("""input_features""").transpose(2							,							0							,							1)

								UpperCAmelCase__  :					Tuple        =     [self._np_extract_fbank_features(_lowerCamelCase) for waveform in input_features[0]]

								if isinstance(input_features[0]							,							_lowerCamelCase):
											UpperCAmelCase__  :					List[Any]        =     [np.asarray(_lowerCamelCase							,							dtype=np.floataa) for feature in input_features]
								else:
											UpperCAmelCase__  :					Any        =     input_features

								if return_attention_mask:
											# rescale from sample (48000) to feature (3000)
											UpperCAmelCase__  :					str        =     padded_inputs["""attention_mask"""][:, :: self.hop_length]

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

								return padded_inputs




					def 				snake_case__						(      self):
								UpperCAmelCase__  :					List[str]        =     copy.deepcopy(self.__dict__)
								UpperCAmelCase__  :					int        =     self.__class__.__name__
								if "mel_filters" in output:
											del output["mel_filters"]
								return output



'''simple docstring'''



import importlib
import os
from dataclasses import dataclass
from enum import Enum
from typing import Any, Dict, Optional, Union

import torch

from ..utils import BaseOutput


__A	='scheduler_config.json'


class 		_snake_case					(     a__	):
					lowerCAmelCase      :Union[str, Any]   =					1
					lowerCAmelCase      :Union[str, Any]   =					2
					lowerCAmelCase      :Optional[int]   =					3
					lowerCAmelCase      :Optional[int]   =					4
					lowerCAmelCase      :Any   =					5
					lowerCAmelCase      :Tuple   =					6
					lowerCAmelCase      :List[Any]   =					7
					lowerCAmelCase      :str   =					8
					lowerCAmelCase      :List[Any]   =					9
					lowerCAmelCase      :List[str]   =					10
					lowerCAmelCase      :Union[str, Any]   =					11
					lowerCAmelCase      :Optional[int]   =					12
					lowerCAmelCase      :str   =					13
					lowerCAmelCase      :Dict   =					14


@dataclass
class 		_snake_case					(     a__	):
					lowerCAmelCase      :torch.FloatTensor


class 		_snake_case					:
					lowerCAmelCase      :str   =					SCHEDULER_CONFIG_NAME
					lowerCAmelCase      :Union[str, Any]   =					[]
					lowerCAmelCase      :List[str]   =					True




					@classmethod
					def 				snake_case__						(      cls							,							_lowerCamelCase = None							,							_lowerCamelCase = None							,							_lowerCamelCase=False							,							**_lowerCamelCase							,							):
								UpperCAmelCase__   ,       UpperCAmelCase__   ,       UpperCAmelCase__  :					Optional[int]        =     cls.load_config(
								    pretrained_model_name_or_path=_lowerCamelCase							,							subfolder=_lowerCamelCase							,							return_unused_kwargs=_lowerCamelCase							,							return_commit_hash=_lowerCamelCase							,							**_lowerCamelCase							,							)
								return cls.from_config(_lowerCamelCase							,							return_unused_kwargs=_lowerCamelCase							,							**_lowerCamelCase)




					def 				snake_case__						(      self							,							_lowerCamelCase							,							_lowerCamelCase = False							,							**_lowerCamelCase):
								self.save_config(save_directory=_lowerCamelCase							,							push_to_hub=_lowerCamelCase							,							**_lowerCamelCase)




					@property
					def 				snake_case__						(      self):
								return self._get_compatibles()




					@classmethod
					def 				snake_case__						(      cls):
								UpperCAmelCase__  :					List[Any]        =     list(set([cls.__name__] + cls._compatibles))
								UpperCAmelCase__  :					int        =     importlib.import_module(__name__.split(""".""")[0])
								UpperCAmelCase__  :					Dict        =     [
								    getattr(_lowerCamelCase							,							_lowerCamelCase) for c in compatible_classes_str if hasattr(_lowerCamelCase							,							_lowerCamelCase)
								]
								return compatible_classes