




'''simple docstring'''





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

import numpy as np

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


__A						:							Optional[int]							=  logging.get_logger(__name__)




class 			UpperCAmelCase_      (   snake_case__       ):




							'''simple docstring'''



							a__						  =     ["""input_values""", """attention_mask"""]





							def __init__(					self       :	List[Any]       ,     a       :	int = 1       ,     a       :	int = 16_000       ,     a       :	float = 0.0       ,     a       :	bool = False       ,     a       :	int = 80       ,     a       :	int = 16       ,     a       :	int = 64       ,     a       :	str = "hann_window"       ,     a       :	float = 1.0       ,     a       :	float = 80       ,     a       :	float = 7_600       ,     a       :	float = 1E-1_0       ,     a       :	int = 2       ,     a       :	bool = True       ,     **a       :	str       ,     )			->					str:
										super().__init__(feature_size=UpperCAmelCase_       ,     sampling_rate=UpperCAmelCase_       ,     padding_value=UpperCAmelCase_       ,     **UpperCAmelCase_			)
										SCREAMING_SNAKE_CASE           =					do_normalize
										SCREAMING_SNAKE_CASE           =					return_attention_mask

										SCREAMING_SNAKE_CASE           =					num_mel_bins
										SCREAMING_SNAKE_CASE           =					hop_length
										SCREAMING_SNAKE_CASE           =					win_length
										SCREAMING_SNAKE_CASE           =					win_function
										SCREAMING_SNAKE_CASE           =					frame_signal_scale
										SCREAMING_SNAKE_CASE           =					fmin
										SCREAMING_SNAKE_CASE           =					fmax
										SCREAMING_SNAKE_CASE           =					mel_floor
										SCREAMING_SNAKE_CASE           =					reduction_factor

										SCREAMING_SNAKE_CASE           =					win_length * sampling_rate // 1_000
										SCREAMING_SNAKE_CASE           =					hop_length * sampling_rate // 1_000
										SCREAMING_SNAKE_CASE           =					optimal_fft_length(self.sample_size			)
										SCREAMING_SNAKE_CASE           =					(self.n_fft // 2) + 1

										SCREAMING_SNAKE_CASE           =					window_function(window_length=self.sample_size       ,     name=self.win_function       ,     periodic=UpperCAmelCase_			)

										SCREAMING_SNAKE_CASE           =					mel_filter_bank(
										    num_frequency_bins=self.n_freqs       ,     num_mel_filters=self.num_mel_bins       ,     min_frequency=self.fmin       ,     max_frequency=self.fmax       ,     sampling_rate=self.sampling_rate       ,     norm="""slaney"""       ,     mel_scale="""slaney"""       ,     )

										if frame_signal_scale != 1.0:
													warnings.warn(
													    """The argument `frame_signal_scale` is deprecated and will be removed in version 4.30.0 of Transformers"""       ,     UpperCAmelCase_       ,     )
										if reduction_factor != 2.0:
													warnings.warn(
													    """The argument `reduction_factor` is deprecated and will be removed in version 4.30.0 of Transformers"""       ,     UpperCAmelCase_       ,     )





							@staticmethod
							# Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
							def    _UpperCAmelCase     (					a       :	List[np.ndarray]       ,     a       :	List[np.ndarray]       ,     a       :	float = 0.0			)			->					List[np.ndarray]:



										if attention_mask is not None:
													SCREAMING_SNAKE_CASE           =					np.array(UpperCAmelCase_       ,     np.intaa			)
													SCREAMING_SNAKE_CASE           =					[]

													for vector, length in zip(UpperCAmelCase_       ,     attention_mask.sum(-1			)			):
																SCREAMING_SNAKE_CASE           =					(vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1E-7			)
																if length < normed_slice.shape[0]:
																			SCREAMING_SNAKE_CASE           =					padding_value

																normed_input_values.append(UpperCAmelCase_			)
										else:
													SCREAMING_SNAKE_CASE           =					[(x - x.mean()) / np.sqrt(x.var() + 1E-7			) for x in input_values]

										return normed_input_values





							def    _UpperCAmelCase     (					self       :	Tuple       ,     a       :	np.ndarray       ,     )			->					np.ndarray:
										SCREAMING_SNAKE_CASE           =					spectrogram(
										    UpperCAmelCase_       ,     window=self.window       ,     frame_length=self.sample_size       ,     hop_length=self.sample_stride       ,     fft_length=self.n_fft       ,     mel_filters=self.mel_filters       ,     mel_floor=self.mel_floor       ,     log_mel="""log10"""       ,     )
										return log_mel_spec.T





							def __call__(					self       :	int       ,     a       :	Optional[Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]] = None       ,     a       :	Optional[Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]] = None       ,     a       :	Union[bool, str, PaddingStrategy] = False       ,     a       :	Optional[int] = None       ,     a       :	bool = False       ,     a       :	Optional[int] = None       ,     a       :	Optional[bool] = None       ,     a       :	Optional[Union[str, TensorType]] = None       ,     a       :	Optional[int] = None       ,     **a       :	Optional[Any]       ,     )			->					BatchFeature:



										if audio is None and audio_target is None:
													raise ValueError("""You must provide either `audio` or `audio_target` values."""			)

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

										if audio is not None:
													SCREAMING_SNAKE_CASE           =					self._process_audio(
													    UpperCAmelCase_       ,     UpperCAmelCase_       ,     UpperCAmelCase_       ,     UpperCAmelCase_       ,     UpperCAmelCase_       ,     UpperCAmelCase_       ,     UpperCAmelCase_       ,     UpperCAmelCase_       ,     **UpperCAmelCase_       ,     )
										else:
													SCREAMING_SNAKE_CASE           =					None

										if audio_target is not None:
													SCREAMING_SNAKE_CASE           =					self._process_audio(
													    UpperCAmelCase_       ,     UpperCAmelCase_       ,     UpperCAmelCase_       ,     UpperCAmelCase_       ,     UpperCAmelCase_       ,     UpperCAmelCase_       ,     UpperCAmelCase_       ,     UpperCAmelCase_       ,     **UpperCAmelCase_       ,     )

													if inputs is None:
																return inputs_target
													else:
																SCREAMING_SNAKE_CASE           =					inputs_target["""input_values"""]
																SCREAMING_SNAKE_CASE           =					inputs_target.get("""attention_mask"""			)
																if decoder_attention_mask is not None:
																			SCREAMING_SNAKE_CASE           =					decoder_attention_mask

										return inputs





							def    _UpperCAmelCase     (					self       :	Optional[int]       ,     a       :	Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]       ,     a       :	bool = False       ,     a       :	Union[bool, str, PaddingStrategy] = False       ,     a       :	Optional[int] = None       ,     a       :	bool = False       ,     a       :	Optional[int] = None       ,     a       :	Optional[bool] = None       ,     a       :	Optional[Union[str, TensorType]] = None       ,     **a       :	Optional[int]       ,     )			->					BatchFeature:
										SCREAMING_SNAKE_CASE           =					isinstance(UpperCAmelCase_       ,     np.ndarray			) and len(speech.shape			) > 1
										if is_batched_numpy and len(speech.shape			) > 2:
													raise ValueError(f"""Only mono-channel audio is supported for input to {self}"""			)
										SCREAMING_SNAKE_CASE           =					is_batched_numpy or (
										    isinstance(UpperCAmelCase_       ,     (list, tuple)			) and (isinstance(speech[0]       ,     (np.ndarray, tuple, list)			))
										)

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

										# always return batch
										if not is_batched:
													SCREAMING_SNAKE_CASE           =					[speech]

										# needed to make pad() work on spectrogram inputs
										SCREAMING_SNAKE_CASE           =					self.feature_size

										# convert into correct format for padding
										if is_target:
													SCREAMING_SNAKE_CASE           =					[self._extract_mel_features(UpperCAmelCase_			) for waveform in speech]
													SCREAMING_SNAKE_CASE           =					BatchFeature({"""input_values""": features}			)
													SCREAMING_SNAKE_CASE           =					self.num_mel_bins
										else:
													SCREAMING_SNAKE_CASE           =					BatchFeature({"""input_values""": speech}			)

										SCREAMING_SNAKE_CASE           =					self.pad(
										    UpperCAmelCase_       ,     padding=UpperCAmelCase_       ,     max_length=UpperCAmelCase_       ,     truncation=UpperCAmelCase_       ,     pad_to_multiple_of=UpperCAmelCase_       ,     return_attention_mask=UpperCAmelCase_       ,     **UpperCAmelCase_       ,     )

										SCREAMING_SNAKE_CASE           =					feature_size_hack

										# convert input values to correct format
										SCREAMING_SNAKE_CASE           =					padded_inputs["""input_values"""]
										if not isinstance(input_values[0]       ,     np.ndarray			):
													SCREAMING_SNAKE_CASE           =					[np.asarray(UpperCAmelCase_       ,     dtype=np.floataa			) for array in input_values]
										elif (
										    not isinstance(UpperCAmelCase_       ,     np.ndarray			)
										    and isinstance(input_values[0]       ,     np.ndarray			)
										    and input_values[0].dtype is np.dtype(np.floataa			)
										):
													SCREAMING_SNAKE_CASE           =					[array.astype(np.floataa			) for array in input_values]
										elif isinstance(UpperCAmelCase_       ,     np.ndarray			) and input_values.dtype is np.dtype(np.floataa			):
													SCREAMING_SNAKE_CASE           =					input_values.astype(np.floataa			)

										# convert attention_mask to correct format
										SCREAMING_SNAKE_CASE           =					padded_inputs.get("""attention_mask"""			)
										if attention_mask is not None:
													SCREAMING_SNAKE_CASE           =					[np.asarray(UpperCAmelCase_       ,     dtype=np.intaa			) for array in attention_mask]

										# zero-mean and unit-variance normalization
										if not is_target and self.do_normalize:
													SCREAMING_SNAKE_CASE           =					(
													    attention_mask
													    if self._get_padding_strategies(UpperCAmelCase_       ,     max_length=UpperCAmelCase_			) is not PaddingStrategy.DO_NOT_PAD
													    else None
													)
													SCREAMING_SNAKE_CASE           =					self.zero_mean_unit_var_norm(
													    padded_inputs["""input_values"""]       ,     attention_mask=UpperCAmelCase_       ,     padding_value=self.padding_value			)

										if return_tensors is not None:
													SCREAMING_SNAKE_CASE           =					padded_inputs.convert_to_tensors(UpperCAmelCase_			)

										return padded_inputs







							def    _UpperCAmelCase     (					self       :	Any			)			->					Dict[str, Any]:
										SCREAMING_SNAKE_CASE           =					super().to_dict()

										# Don't serialize these as they are derived from the other properties.
										SCREAMING_SNAKE_CASE           =					["""window""", """mel_filters""", """sample_size""", """sample_stride""", """n_fft""", """n_freqs"""]
										for name in names:
													if name in output:
																del output[name]

										return output












import importlib
import shutil
import threading
import warnings
from typing import List

import fsspec
import fsspec.asyn

from . import compression
from .hffilesystem import HfFileSystem


__A						:							Tuple							=  importlib.util.find_spec("""s3fs""") is not None

if _has_safs:
		from .safilesystem import SaFileSystem  # noqa: F401

__A						:							List[compression.BaseCompressedFileFileSystem]							=  [
    compression.BzaFileSystem,
    compression.GzipFileSystem,
    compression.LzaFileSystem,
    compression.XzFileSystem,
    compression.ZstdFileSystem,
]

# Register custom filesystems
for fs_class in COMPRESSION_FILESYSTEMS + [HfFileSystem]:
		if fs_class.protocol in fsspec.registry and fsspec.registry[fs_class.protocol] is not fs_class:
				warnings.warn(f'''A filesystem protocol was already set for {fs_class.protocol} and will be overwritten.''')
		fsspec.register_implementation(fs_class.protocol, fs_class, clobber=True)
def 					lowerCamelCase_    (    SCREAMING_SNAKE_CASE   ):
			'''simple docstring'''
			if "://" in dataset_path:
						SCREAMING_SNAKE_CASE           =					dataset_path.split("""://"""   )[1]
			return dataset_path
def 					lowerCamelCase_    (    SCREAMING_SNAKE_CASE   ):
			'''simple docstring'''
			if fs is not None and fs.protocol != "file":
						return True
			else:
						return False
def 					lowerCamelCase_    (    SCREAMING_SNAKE_CASE					,	SCREAMING_SNAKE_CASE					,	SCREAMING_SNAKE_CASE   ):
			'''simple docstring'''
			SCREAMING_SNAKE_CASE           =					not is_remote_filesystem(SCREAMING_SNAKE_CASE   )
			if is_local:
						# LocalFileSystem.mv does copy + rm, it is more efficient to simply move a local directory
						shutil.move(fs._strip_protocol(SCREAMING_SNAKE_CASE   )					,	fs._strip_protocol(SCREAMING_SNAKE_CASE   )   )
			else:
						fs.mv(SCREAMING_SNAKE_CASE					,	SCREAMING_SNAKE_CASE					,	recursive=SCREAMING_SNAKE_CASE   )




def 					lowerCamelCase_    (    ):
			'''simple docstring'''
			if hasattr(fsspec.asyn					,	"""reset_lock"""   ):
						# for future fsspec>2022.05.0
						fsspec.asyn.reset_lock()
			else:
						SCREAMING_SNAKE_CASE           =					None
						SCREAMING_SNAKE_CASE           =					None
						SCREAMING_SNAKE_CASE           =					threading.Lock()






