
'''simple docstring'''






from typing import 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 ....file_utils import PaddingStrategy, TensorType
from ....utils import logging


UpperCAmelCase_							: str =							logging.get_logger(__name__)

class 			UpperCAmelCase__    (							A			):
    lowerCAmelCase_					     =		['input_features', 'attention_mask']



    def __init__(						self   :							Any,__A   :							List[Any]=8_0,__A   :							Dict=1_6_0_0_0,__A   :							Tuple=0.0,__A   :							Dict=1_0,__A   :							int=2_5,__A   :							Union[str, Any]="hamming_window",__A   :							List[str]=32768.0,__A   :							Union[str, Any]=0.97,__A   :							str=1.0,__A   :							Union[str, Any]=True,__A   :							Tuple=True,__A   :							Optional[Any]=False,**__A   :							Optional[Any],):
          super().__init__(feature_size=__A,sampling_rate=__A,padding_value=__A,**__A  )

          _lowerCamelCase					: Dict		    = feature_size
          _lowerCamelCase					: List[str]		    = sampling_rate
          _lowerCamelCase					: Any		    = padding_value
          _lowerCamelCase					: Dict		    = hop_length
          _lowerCamelCase					: Tuple		    = win_length
          _lowerCamelCase					: str		    = frame_signal_scale
          _lowerCamelCase					: List[str]		    = preemphasis_coeff
          _lowerCamelCase					: List[str]		    = mel_floor
          _lowerCamelCase					: str		    = normalize_means
          _lowerCamelCase					: Any		    = normalize_vars
          _lowerCamelCase					: List[str]		    = win_function
          _lowerCamelCase					: Tuple		    = return_attention_mask

          _lowerCamelCase					: List[Any]		    = win_length * sampling_rate // 1_0_0_0
          _lowerCamelCase					: List[Any]		    = hop_length * sampling_rate // 1_0_0_0

          _lowerCamelCase					: Any		    = optimal_fft_length(self.sample_size  )
          _lowerCamelCase					: Dict		    = (self.n_fft // 2) + 1



    def 	lowerCamelCase_	(						self   :							Any,__A   :							np.array  ):
          if self.win_function == "hamming_window":
                _lowerCamelCase					: Any		    = window_function(window_length=self.sample_size,name=self.win_function,periodic=__A  )
          else:
                _lowerCamelCase					: Optional[int]		    = window_function(window_length=self.sample_size,name=self.win_function  )

          _lowerCamelCase					: int		    = mel_filter_bank(
              num_frequency_bins=self.n_freqs,num_mel_filters=self.feature_size,min_frequency=0.0,max_frequency=self.sampling_rate / 2.0,sampling_rate=self.sampling_rate,)

          _lowerCamelCase					: List[str]		    = spectrogram(
              one_waveform * self.frame_signal_scale,window=__A,frame_length=self.sample_size,hop_length=self.sample_stride,fft_length=self.n_fft,center=__A,preemphasis=self.preemphasis_coeff,mel_filters=__A,mel_floor=self.mel_floor,log_mel="log",)
          return msfc_features.T



    def 	lowerCamelCase_	(						self   :							Optional[int],__A   :							List[str],__A   :							Dict,__A   :							int  ):
          # make sure we normalize float32 arrays
          if self.normalize_means:
                _lowerCamelCase					: Optional[Any]		    = x[:input_length].mean(axis=0  )
                _lowerCamelCase					: Optional[int]		    = np.subtract(__A,__A  )
          if self.normalize_vars:
                _lowerCamelCase					: int		    = x[:input_length].std(axis=0  )
                _lowerCamelCase					: Any		    = np.divide(__A,__A  )

          if input_length < x.shape[0]:
                _lowerCamelCase					: Tuple		    = padding_value

          # make sure array is in float32
          _lowerCamelCase					: Optional[int]		    = x.astype(np.floataa  )

          return x



    def 	lowerCamelCase_	(						self   :							Any,__A   :							List[np.ndarray],__A   :							Optional[np.ndarray] = None  ):
          _lowerCamelCase					: Optional[int]		    = attention_mask.sum(-1  ) if attention_mask is not None else [x.shape[0] for x in input_features]
          return [self._normalize_one(__A,__A,self.padding_value  ) for x, n in zip(__A,__A  )]



    def __call__(						self   :							Optional[Any],__A   :							Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]],__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],):

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

          if is_batched:
                _lowerCamelCase					: List[Any]		    = [np.asarray(__A,dtype=np.floataa  ) for speech in raw_speech]
          elif not is_batched and not isinstance(__A,np.ndarray  ):
                _lowerCamelCase					: Dict		    = np.asarray(__A,dtype=np.floataa  )
          elif isinstance(__A,np.ndarray  ) and raw_speech.dtype is np.dtype(np.floataa  ):
                _lowerCamelCase					: Tuple		    = raw_speech.astype(np.floataa  )

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

          # extract fbank features
          _lowerCamelCase					: str		    = [self._extract_mfsc_features(__A  ) for one_waveform in raw_speech]

          # convert into correct format for padding
          _lowerCamelCase					: Union[str, Any]		    = BatchFeature({"input_features": features}  )

          _lowerCamelCase					: List[Any]		    = self.pad(
              __A,padding=__A,max_length=__A,truncation=__A,pad_to_multiple_of=__A,return_attention_mask=__A,**__A,)
          # make sure list is in array format
          _lowerCamelCase					: Optional[Any]		    = padded_inputs.get("input_features"  )
          if isinstance(input_features[0],__A  ):
                _lowerCamelCase					: int		    = [np.asarray(__A,dtype=np.floataa  ) for feature in input_features]

          _lowerCamelCase					: Dict		    = padded_inputs.get("attention_mask"  )
          if attention_mask is not None:
                _lowerCamelCase					: Dict		    = [np.asarray(__A,dtype=np.intaa  ) for array in attention_mask]

          if self.normalize_means or self.normalize_vars:
                _lowerCamelCase					: Dict		    = (
                    np.array(__A,dtype=np.intaa  )
                    if self._get_padding_strategies(__A,max_length=__A  ) is not PaddingStrategy.DO_NOT_PAD
                    and padding
                    else None
                )
                _lowerCamelCase					: Tuple		    = self.normalize(
                    padded_inputs["input_features"],attention_mask=__A  )

          if return_tensors is not None:
                _lowerCamelCase					: Dict		    = padded_inputs.convert_to_tensors(__A  )

          return padded_inputs






import os
from pathlib import Path
from unittest.mock import patch

import pytest
import zstandard as zstd

from datasets.download.download_config import DownloadConfig
from datasets.utils.file_utils import (
    OfflineModeIsEnabled,
    cached_path,
    fsspec_get,
    fsspec_head,
    ftp_get,
    ftp_head,
    get_from_cache,
    http_get,
    http_head,
)


lowerCAmelCase							:       Dict					 =	"""\
    Text data.
    Second line of data."""

lowerCAmelCase							:       str					 =	"""file"""






@pytest.fixture(scope='''session'''					)
def 		lowerCAmelCase  (	UpperCamelCase__   :   Union[str, Any]					)      ->    Any:

						"""simple docstring"""





						__SCREAMING_SNAKE_CASE:				Any           =   tmp_path_factory.mktemp('''data'''					) / (FILE_PATH + '''.zstd''')
						__SCREAMING_SNAKE_CASE:				Optional[Any]           =   bytes(UpperCamelCase__       ,			'''utf-8'''					)
						with zstd.open(UpperCamelCase__       ,			'''wb'''					) as f:
												f.write(UpperCamelCase__					)
						return path






@pytest.fixture
def 		lowerCAmelCase  (	UpperCamelCase__   :   Optional[Any]					)      ->    int:

						"""simple docstring"""





						with open(os.path.join(tmpfs.local_root_dir       ,			UpperCamelCase__					)       ,			'''w'''					) as f:
												f.write(UpperCamelCase__					)
						return FILE_PATH






@pytest.mark.parametrize('''compression_format'''       ,			['''gzip''', '''xz''', '''zstd''']					)
def 		lowerCAmelCase  (	UpperCamelCase__   :   Optional[int]       ,			UpperCamelCase__   :   List[str]       ,			UpperCamelCase__   :   Dict       ,			UpperCamelCase__   :   Optional[Any]       ,			UpperCamelCase__   :   Dict       ,			UpperCamelCase__   :   List[Any]					)      ->    Any:

						"""simple docstring"""





						__SCREAMING_SNAKE_CASE:				Dict           =   {'''gzip''': gz_file, '''xz''': xz_file, '''zstd''': zstd_path}
						__SCREAMING_SNAKE_CASE:				List[str]           =   input_paths[compression_format]
						__SCREAMING_SNAKE_CASE:				List[Any]           =   tmp_path / '''cache'''
						__SCREAMING_SNAKE_CASE:				List[str]           =   DownloadConfig(cache_dir=UpperCamelCase__       ,			extract_compressed_file=UpperCamelCase__					)
						__SCREAMING_SNAKE_CASE:				Tuple           =   cached_path(UpperCamelCase__       ,			download_config=UpperCamelCase__					)
						with open(UpperCamelCase__					) as f:
												__SCREAMING_SNAKE_CASE:				Optional[Any]           =   f.read()
						with open(UpperCamelCase__					) as f:
												__SCREAMING_SNAKE_CASE:				Dict           =   f.read()
						assert extracted_file_content == expected_file_content






@pytest.mark.parametrize('''default_extracted'''       ,			[True, False]					)
@pytest.mark.parametrize('''default_cache_dir'''       ,			[True, False]					)
def 		lowerCAmelCase  (	UpperCamelCase__   :   Tuple       ,			UpperCamelCase__   :   List[str]       ,			UpperCamelCase__   :   str       ,			UpperCamelCase__   :   Optional[int]       ,			UpperCamelCase__   :   Optional[Any]					)      ->    Optional[Any]:

						"""simple docstring"""





						__SCREAMING_SNAKE_CASE:				Tuple           =   '''custom_cache'''
						__SCREAMING_SNAKE_CASE:				Optional[int]           =   '''custom_extracted_dir'''
						__SCREAMING_SNAKE_CASE:				List[Any]           =   tmp_path / '''custom_extracted_path'''
						if default_extracted:
												__SCREAMING_SNAKE_CASE:				int           =   ('''downloads''' if default_cache_dir else custom_cache_dir, '''extracted''')
						else:
												monkeypatch.setattr('''datasets.config.EXTRACTED_DATASETS_DIR'''       ,			UpperCamelCase__					)
												monkeypatch.setattr('''datasets.config.EXTRACTED_DATASETS_PATH'''       ,			str(UpperCamelCase__					)					)
												__SCREAMING_SNAKE_CASE:				Union[str, Any]           =   custom_extracted_path.parts[-2:] if default_cache_dir else (custom_cache_dir, custom_extracted_dir)

						__SCREAMING_SNAKE_CASE:				Dict           =   xz_file
						__SCREAMING_SNAKE_CASE:				Optional[Any]           =   (
						    DownloadConfig(extract_compressed_file=UpperCamelCase__					)
						    if default_cache_dir
						    else DownloadConfig(cache_dir=tmp_path / custom_cache_dir       ,			extract_compressed_file=UpperCamelCase__					)
						)
						__SCREAMING_SNAKE_CASE:				int           =   cached_path(UpperCamelCase__       ,			download_config=UpperCamelCase__					)
						assert Path(UpperCamelCase__					).parent.parts[-2:] == expected






def 		lowerCAmelCase  (	UpperCamelCase__   :   Dict					)      ->    Union[str, Any]:

						"""simple docstring"""





						__SCREAMING_SNAKE_CASE:				Optional[Any]           =   str(Path(UpperCamelCase__					).resolve()					)
						assert cached_path(UpperCamelCase__					) == text_file
						# relative path
						__SCREAMING_SNAKE_CASE:				Optional[int]           =   str(Path(UpperCamelCase__					).resolve().relative_to(Path(os.getcwd()					)					)					)
						assert cached_path(UpperCamelCase__					) == text_file






def 		lowerCAmelCase  (	UpperCamelCase__   :   Optional[Any]					)      ->    Tuple:

						"""simple docstring"""





						__SCREAMING_SNAKE_CASE:				List[Any]           =   str(tmp_path.resolve() / '''__missing_file__.txt'''					)
						with pytest.raises(UpperCamelCase__					):
												cached_path(UpperCamelCase__					)
						# relative path
						__SCREAMING_SNAKE_CASE:				int           =   '''./__missing_file__.txt'''
						with pytest.raises(UpperCamelCase__					):
												cached_path(UpperCamelCase__					)






def 		lowerCAmelCase  (	UpperCamelCase__   :   Tuple					)      ->    Union[str, Any]:

						"""simple docstring"""





						__SCREAMING_SNAKE_CASE:				Tuple           =   get_from_cache(F"""tmp://{tmpfs_file}"""					)
						with open(UpperCamelCase__					) as f:
												__SCREAMING_SNAKE_CASE:				Tuple           =   f.read()
						assert output_file_content == FILE_CONTENT






@patch('''datasets.config.HF_DATASETS_OFFLINE'''       ,			UpperCamelCase__					)
def 		lowerCAmelCase  (	)      ->    Any:

						"""simple docstring"""





						with pytest.raises(UpperCamelCase__					):
												cached_path('''https://huggingface.co'''					)






@patch('''datasets.config.HF_DATASETS_OFFLINE'''       ,			UpperCamelCase__					)
def 		lowerCAmelCase  (	UpperCamelCase__   :   int					)      ->    Union[str, Any]:

						"""simple docstring"""





						__SCREAMING_SNAKE_CASE:				Optional[int]           =   tmp_path_factory.mktemp('''data'''					) / '''file.html'''
						with pytest.raises(UpperCamelCase__					):
												http_get('''https://huggingface.co'''       ,			temp_file=UpperCamelCase__					)
						with pytest.raises(UpperCamelCase__					):
												http_head('''https://huggingface.co'''					)






@patch('''datasets.config.HF_DATASETS_OFFLINE'''       ,			UpperCamelCase__					)
def 		lowerCAmelCase  (	UpperCamelCase__   :   Dict					)      ->    List[str]:

						"""simple docstring"""





						__SCREAMING_SNAKE_CASE:				List[Any]           =   tmp_path_factory.mktemp('''data'''					) / '''file.html'''
						with pytest.raises(UpperCamelCase__					):
												ftp_get('''ftp://huggingface.co'''       ,			temp_file=UpperCamelCase__					)
						with pytest.raises(UpperCamelCase__					):
												ftp_head('''ftp://huggingface.co'''					)






@patch('''datasets.config.HF_DATASETS_OFFLINE'''       ,			UpperCamelCase__					)
def 		lowerCAmelCase  (	UpperCamelCase__   :   Optional[int]					)      ->    Tuple:

						"""simple docstring"""





						__SCREAMING_SNAKE_CASE:				Tuple           =   tmp_path_factory.mktemp('''data'''					) / '''file.html'''
						with pytest.raises(UpperCamelCase__					):
												fsspec_get('''s3://huggingface.co'''       ,			temp_file=UpperCamelCase__					)
						with pytest.raises(UpperCamelCase__					):
												fsspec_head('''s3://huggingface.co'''					)



