






'''simple docstring'''
import os
from typing import Optional

import fsspec
from fsspec.archive import AbstractArchiveFileSystem
from fsspec.utils import DEFAULT_BLOCK_SIZE


class  _snake_case (					snake_case       ):

					"""simple docstring"""





					_UpperCamelCase					=   ""
					_UpperCamelCase					=   (
					    None  # protocol passed in prefix to the url. ex: "gzip", for gzip://file.txt::http://foo.bar/file.txt.gz
					)
					_UpperCamelCase					=   None  # compression type in fsspec. ex: "gzip"
					_UpperCamelCase					=   None  # extension of the filename to strip. ex: "".gz" to get file.txt from file.txt.gz




					def __init__(  self ,			UpperCAmelCase__ = "" ,			UpperCAmelCase__ = None ,			UpperCAmelCase__ = None ,			**UpperCAmelCase__  )   ->						int:
										super().__init__(self ,			**UpperCAmelCase__  )
										# always open as "rb" since fsspec can then use the TextIOWrapper to make it work for "r" mode
										a_						= fsspec.open(
										    UpperCAmelCase__ ,			mode='rb' ,			protocol=UpperCAmelCase__ ,			compression=self.compression ,			client_kwargs={
										        'requote_redirect_url': False,  # see https://github.com/huggingface/datasets/pull/5459
										        'trust_env': True,  # Enable reading proxy env variables.
										        **(target_options or {}).pop('client_kwargs' ,			{}  ),  # To avoid issues if it was already passed.
										    } ,			**(target_options or {}) ,			)
										a_						= os.path.basename(self.file.path.split('::'  )[0]  )
										a_						= (
										    self.compressed_name[: self.compressed_name.rindex('.'  )]
										    if '.' in self.compressed_name
										    else self.compressed_name
										)
										a_						= None




					@classmethod
					def 			__SCREAMING_SNAKE_CASE			(  cls ,			UpperCAmelCase__  )   ->						Optional[int]:
										return super()._strip_protocol(UpperCAmelCase__  ).lstrip('/'  )




					def 			__SCREAMING_SNAKE_CASE			(  self  )   ->						int:
										if self.dir_cache is None:
															a_						= {**self.file.fs.info(self.file.path  ), 'name': self.uncompressed_name}
															a_						= {f['name']: f}




					def 			__SCREAMING_SNAKE_CASE			(  self ,			UpperCAmelCase__  )   ->						Union[str, Any]:
										return self.file.open().read()




					def 			__SCREAMING_SNAKE_CASE			(  self ,			UpperCAmelCase__ ,			UpperCAmelCase__ = "rb" ,			UpperCAmelCase__=None ,			UpperCAmelCase__=True ,			UpperCAmelCase__=None ,			**UpperCAmelCase__ ,			)   ->						Any:
										a_						= self._strip_protocol(UpperCAmelCase__  )
										if mode != "rb":
															raise ValueError(F'''Tried to read with mode {mode} on file {self.file.path} opened with mode \'rb\''''  )
										return self.file.open()



class  _snake_case (					snake_case       ):

					"""simple docstring"""





					_UpperCamelCase					=   "bz2"
					_UpperCamelCase					=   "bz2"
					_UpperCamelCase					=   ".bz2"



class  _snake_case (					snake_case       ):

					"""simple docstring"""





					_UpperCamelCase					=   "gzip"
					_UpperCamelCase					=   "gzip"
					_UpperCamelCase					=   ".gz"



class  _snake_case (					snake_case       ):

					"""simple docstring"""





					_UpperCamelCase					=   "lz4"
					_UpperCamelCase					=   "lz4"
					_UpperCamelCase					=   ".lz4"



class  _snake_case (					snake_case       ):

					"""simple docstring"""





					_UpperCamelCase					=   "xz"
					_UpperCamelCase					=   "xz"
					_UpperCamelCase					=   ".xz"



class  _snake_case (					snake_case       ):

					"""simple docstring"""





					_UpperCamelCase					=   "zstd"
					_UpperCamelCase					=   "zstd"
					_UpperCamelCase					=   ".zst"




					def __init__(  self ,			UpperCAmelCase__ ,			UpperCAmelCase__ = "rb" ,			UpperCAmelCase__ = None ,			UpperCAmelCase__ = None ,			UpperCAmelCase__ = DEFAULT_BLOCK_SIZE ,			**UpperCAmelCase__ ,			)   ->						Tuple:
										super().__init__(
										    fo=UpperCAmelCase__ ,			mode=UpperCAmelCase__ ,			target_protocol=UpperCAmelCase__ ,			target_options=UpperCAmelCase__ ,			block_size=UpperCAmelCase__ ,			**UpperCAmelCase__ ,			)
										# We need to wrap the zstd decompressor to avoid this error in fsspec==2021.7.0 and zstandard==0.15.2:
										#
										# File "/Users/user/.virtualenvs/hf-datasets/lib/python3.7/site-packages/fsspec/core.py", line 145, in open
										#     out.close = close
										# AttributeError: 'zstd.ZstdDecompressionReader' object attribute 'close' is read-only
										#
										# see https://github.com/intake/filesystem_spec/issues/725
										a_						= self.file.__enter__


										class  _snake_case :

															"""simple docstring"""





															def __init__(  self ,			UpperCAmelCase__  )   ->						Any:
																				a_						= file_




															def __enter__(  self  )   ->						Optional[int]:
																				self._file.__enter__()
																				return self




															def __exit__(  self ,			*UpperCAmelCase__ ,			**UpperCAmelCase__  )   ->						Any:
																				self._file.__exit__(*UpperCAmelCase__ ,			**UpperCAmelCase__  )




															def __iter__(  self  )   ->						Dict:
																				return iter(self._file  )




															def 			__SCREAMING_SNAKE_CASE			(  self  )   ->						Tuple:
																				return next(self._file  )




															def __getattr__(  self ,			UpperCAmelCase__  )   ->						List[Any]:
																				return getattr(self._file ,			UpperCAmelCase__  )

										def fixed_enter(*UpperCAmelCase__ ,			**UpperCAmelCase__  ):
															return WrappedFile(_enter(*UpperCAmelCase__ ,			**UpperCAmelCase__  )  )

										a_						= fixed_enter





'''simple docstring'''






from math import ceil
from typing import List, Optional, Union

import numpy as np

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


lowercase				=logging.get_logger(__name__)






class    __magic_name__				(						lowerCAmelCase							):
       UpperCAmelCase		 =["audio_values", "audio_mask"]






       def __init__(     self    ,				snake_case=2_0_4_8    ,				snake_case=1    ,				snake_case=[1_6, 1_6]    ,				snake_case=1_2_8    ,				snake_case=4_4_1_0_0    ,				snake_case=8_6    ,				snake_case=2_0_4_8    ,				snake_case=0.0    ,				**snake_case    ,				)		->    Optional[int]:




          '''simple docstring'''
          super().__init__(
              feature_size=snake_case    ,				sampling_rate=snake_case    ,				padding_value=snake_case    ,				**snake_case    ,				)

          _UpperCAmelCase  :  Dict				       =spectrogram_length
          _UpperCAmelCase  :  Any				       =num_channels
          _UpperCAmelCase  :  Optional[int]				       =patch_size
          _UpperCAmelCase  :  List[str]				       =feature_size // self.patch_size[1]
          _UpperCAmelCase  :  Optional[int]				       =n_fft
          _UpperCAmelCase  :  int				       =sampling_rate // hop_length_to_sampling_rate
          _UpperCAmelCase  :  int				       =sampling_rate
          _UpperCAmelCase  :  List[Any]				       =padding_value
          _UpperCAmelCase  :  Optional[int]				       =mel_filter_bank(
              num_frequency_bins=1 + n_fft // 2    ,				num_mel_filters=snake_case    ,				min_frequency=0.0    ,				max_frequency=2_20_50.0    ,				sampling_rate=snake_case    ,				norm='slaney'    ,				mel_scale='slaney'    ,				).T






       def      lowerCAmelCase						(     self    ,				snake_case)		->    np.ndarray:




          '''simple docstring'''
          _UpperCAmelCase  :  Any				       =spectrogram(
              snake_case    ,				window_function(self.n_fft    ,				'hann')    ,				frame_length=self.n_fft    ,				hop_length=self.hop_length    ,				power=2.0    ,				mel_filters=self.mel_filters.T    ,				log_mel='dB'    ,				db_range=80.0    ,				)
          _UpperCAmelCase  :  Any				       =log_spec[:, :-1]
          _UpperCAmelCase  :  List[str]				       =log_spec - 20.0
          _UpperCAmelCase  :  Union[str, Any]				       =np.clip(log_spec / 40.0    ,				-2.0    ,				0.0) + 1.0
          return log_spec






       def __call__(     self    ,				snake_case    ,				snake_case = None    ,				snake_case = True    ,				snake_case = None    ,				snake_case = False    ,				snake_case = False    ,				**snake_case    ,				)		->    BatchFeature:




          '''simple docstring'''

          if sampling_rate is not None:
             if sampling_rate != self.sampling_rate:
                raise ValueError(
                    'This feature extractor is set to support sampling rate'
                    f" of {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled"
                    f" 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  :  Tuple				       =isinstance(snake_case    ,				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  :  str				       =is_batched_numpy or (
              isinstance(snake_case    ,				(list, tuple)) and (isinstance(raw_speech[0]    ,				(np.ndarray, tuple, list)))
          )
          if is_batched:
             _UpperCAmelCase  :  List[str]				       =[np.asarray([speech]    ,				dtype=np.floataa).T for speech in raw_speech]
          elif not is_batched and not isinstance(snake_case    ,				np.ndarray):
             _UpperCAmelCase  :  List[Any]				       =np.asarray(snake_case    ,				dtype=np.floataa)
          elif isinstance(snake_case    ,				np.ndarray) and raw_speech.dtype is np.dtype(np.floataa):
             _UpperCAmelCase  :  List[Any]				       =raw_speech.astype(np.floataa)
          # always return batch
          if not is_batched:
             _UpperCAmelCase  :  Dict				       =[np.asarray([raw_speech]).T]

          # Convert audio signals to log mel spectrograms, truncate by time axis
          _UpperCAmelCase  :  List[Any]				       =[
              self._np_extract_fbank_features(waveform.squeeze()).T[: self.spectrogram_length] for waveform in raw_speech
          ]
          if isinstance(audio_features[0]    ,				snake_case):
             _UpperCAmelCase  :  Dict				       =[np.asarray(snake_case    ,				dtype=np.floataa) for feature in audio_features]

          # Create audio attention mask
          _UpperCAmelCase  :  Tuple				       =max(
              [ceil(feature.shape[0] / self.patch_size[0]) * self.freq_len for feature in audio_features])  # The maximum number of audio patches in a batch
          if return_attention_mask:
             _UpperCAmelCase  :  Any				       =[
                 (ceil(feature.shape[0] / self.patch_size[0]) * self.freq_len) * [1]
                 + (max_patch_len - ceil(feature.shape[0] / self.patch_size[0]) * self.freq_len) * [0]
                 for feature in audio_features
             ]
             _UpperCAmelCase  :  int				       =np.array(snake_case).astype(np.floataa)

          # convert into correct format for padding
          _UpperCAmelCase  :  List[Any]				       =max_patch_len // self.freq_len * self.patch_size[0]  # The maximum audio size in a batch
          _UpperCAmelCase  :  List[str]				       =np.ones([len(snake_case), 1, max_time_len, self.feature_size]).astype(np.floataa)
          _UpperCAmelCase  :  Any				       =padded_audio_features * self.padding_value
          for i in range(len(snake_case)):
             _UpperCAmelCase  :  List[str]				       =audio_features[i]
             _UpperCAmelCase  :  Dict				       =feature

          # return as BatchFeature
          if return_attention_mask:
             _UpperCAmelCase  :  List[str]				       ={'audio_values': padded_audio_features, 'audio_mask': audio_mask}
          else:
             _UpperCAmelCase  :  Optional[int]				       ={'audio_values': padded_audio_features}

          _UpperCAmelCase  :  Any				       =BatchFeature(data=snake_case    ,				tensor_type=snake_case)
          return encoded_inputs

